filters/pause/pause.c

   1 /* nbdkit
   2  * Copyright Red Hat
   3  *
   4  * Redistribution and use in source and binary forms, with or without
   5  * modification, are permitted provided that the following conditions are
   6  * met:
   7  *
   8  * * Redistributions of source code must retain the above copyright
   9  * notice, this list of conditions and the following disclaimer.
  10  *
  11  * * Redistributions in binary form must reproduce the above copyright
  12  * notice, this list of conditions and the following disclaimer in the
  13  * documentation and/or other materials provided with the distribution.
  14  *
  15  * * Neither the name of Red Hat nor the names of its contributors may be
  16  * used to endorse or promote products derived from this software without
  17  * specific prior written permission.
  18  *
  19  * THIS SOFTWARE IS PROVIDED BY RED HAT AND CONTRIBUTORS ''AS IS'' AND
  20  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
  21  * THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
  22  * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL RED HAT OR
  23  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  24  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  25  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  26  * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  27  * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  28  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
  29  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  30  * SUCH DAMAGE.
  31  */
  32
  33 #include <config.h>
  34
  35 #include <stdio.h>
  36 #include <stdlib.h>
  37 #include <stdbool.h>
  38 #include <stdint.h>
  39 #include <string.h>
  40 #include <unistd.h>
  41 #include <errno.h>
  42 #include <sys/types.h>
  43 #include <sys/socket.h>
  44 #include <sys/un.h>
  45
  46 #include <pthread.h>
  47
  48 #include <nbdkit-filter.h>
  49
  50 #include "ascii-ctype.h"
  51 #include "cleanup.h"
  52 #include "utils.h"
  53 #include "unix-path-max.h"
  54
  55 static char *sockfile;
  56 static int sock = -1;
  57
  58 static void
  59 pause_unload (void)
  60 {
  61   if (sock >= 0)
  62     close (sock);
  63   if (sockfile) {
  64     unlink (sockfile);
  65     free (sockfile);
  66   }
  67 }
  68
  69 static int
  70 pause_config (nbdkit_next_config *next, nbdkit_backend *nxdata,
  71               const char *key, const char *value)
  72 {
  73   if (strcmp (key, "pause-control") == 0) {
  74     free (sockfile);
  75     sockfile = nbdkit_absolute_path (value);
  76     if (sockfile == NULL)
  77       return -1;
  78     return 0;
  79   }
  80   else
  81     return next (nxdata, key, value);
  82 }
  83
  84 static int
  85 pause_config_complete (nbdkit_next_config_complete *next,
  86                        nbdkit_backend *nxdata)
  87 {
  88   size_t len;
  89   struct sockaddr_un addr;
  90
  91   if (sockfile == NULL) {
  92     nbdkit_error ("pause-control socket was not set");
  93     return -1;
  94   }
  95   len = strlen (sockfile);
  96   if (len >= UNIX_PATH_MAX) {
  97     nbdkit_error ("pause-control socket path too long: "
  98                   "length %zu > max %d bytes",
  99                   len, UNIX_PATH_MAX-1);
 100     return -1;
 101   }
 102
 103   /* If the socket already exists, remove it. */
 104   unlink (sockfile);
 105
 106 #ifdef SOCK_CLOEXEC
 107   sock = socket (AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0);
 108 #else
 109   /* Fortunately, this code is only run at startup, so there is no
 110    * risk of the fd leaking to a plugin's fork()
 111    */
 112   sock = set_cloexec (socket (AF_UNIX, SOCK_STREAM, 0));
 113 #endif
 114   if (sock == -1) {
 115     nbdkit_error ("socket: %m");
 116     return -1;
 117   }
 118
 119   addr.sun_family = AF_UNIX;
 120   memcpy (addr.sun_path, sockfile, len+1 /* trailing \0 */);
 121
 122   if (bind (sock, (struct sockaddr *) &addr, sizeof addr) == -1) {
 123     nbdkit_error ("%s: %m", sockfile);
 124     return -1;
 125   }
 126
 127   if (listen (sock, SOMAXCONN) == -1) {
 128     nbdkit_error ("listen: %m");
 129     return -1;
 130   }
 131
 132   return next (nxdata);
 133 }
 134
 135 #define pause_config_help \
 136   "pause-control=SOCKET           Control socket."
 137
 138 /* This is locked by the background thread when we're paused, causing
 139  * all requests in the main threads to hang.
 140  */
 141 static pthread_mutex_t paused = PTHREAD_MUTEX_INITIALIZER;
 142 static bool is_paused = false;
 143
 144 /* This keeps track of the number of NBD requests in flight. */
 145 static pthread_mutex_t count_lock = PTHREAD_MUTEX_INITIALIZER;
 146 static pthread_cond_t count_cond = PTHREAD_COND_INITIALIZER;
 147 static unsigned count_requests = 0;
 148
 149 static void
 150 do_pause (void)
 151 {
 152   if (is_paused) return;
 153
 154   /* Grabbing the paused lock is enough to stop request processing. */
 155   pthread_mutex_lock (&paused);
 156   is_paused = true;
 157
 158   /* However we must also wait until all outstanding requests have
 159    * been completed before we send the acknowledgement.
 160    */
 161   nbdkit_debug ("pause: pausing, waiting for requests to complete");
 162   ACQUIRE_LOCK_FOR_CURRENT_SCOPE (&count_lock);
 163   while (count_requests > 0)
 164     pthread_cond_wait (&count_cond, &count_lock);
 165   nbdkit_debug ("pause: paused");
 166 }
 167
 168 static void
 169 do_resume (void)
 170 {
 171   if (!is_paused) return;
 172
 173   /* Release the worker threads. */
 174   is_paused = false;
 175   pthread_mutex_unlock (&paused);
 176   nbdkit_debug ("pause: resumed");
 177 }
 178
 179 /* Background thread which monitors the control socket.  This can only
 180  * accept one connection at a time.
 181  */
 182 static void *
 183 control_socket_thread (void *arg)
 184 {
 185   int s;
 186   char c;
 187   ssize_t n;
 188
 189   for (;;) {
 190 #ifdef HAVE_ACCEPT4
 191     s = accept4 (sock, NULL, NULL, SOCK_CLOEXEC);
 192 #else
 193     /* This isn't thread-safe but there's not a lot we can do. */
 194     s = set_cloexec (accept (sock, NULL, NULL));
 195 #endif
 196     if (s == -1) goto out;
 197
 198     /* Read commands (which are single bytes) until end of file. */
 199     while ((n = read (s, &c, 1)) == 1) {
 200       switch (c) {
 201       case 'p':
 202         do_pause ();
 203         break;
 204       case 'r':
 205         do_resume ();
 206         break;
 207       case '\n':
 208       case '\t':
 209       case ' ':
 210         /* For convenience of interactive use, ignore and don't
 211          * respond to some whitespace characters.
 212          */
 213         continue;
 214       default:          /* Unknown command. */
 215         c = 'X';
 216       }
 217       /* Send the response. */
 218       c = ascii_toupper (c);
 219       n = write (s, &c, 1);
 220       if (n == -1) goto out;
 221     }
 222     if (n == 0)
 223       errno = 0;
 224   out:
 225     if (errno && (errno != EINTR && errno != EAGAIN))
 226       nbdkit_error ("accept: %m");
 227     if (s >= 0)
 228       close (s);
 229   }
 230
 231   /*NOTREACHED*/
 232   return NULL;
 233 }
 234
 235 /* Start the background thread after fork. */
 236 static int
 237 pause_after_fork (nbdkit_backend *nxdata)
 238 {
 239   int err;
 240   pthread_t thread;
 241
 242   err = pthread_create (&thread, NULL, control_socket_thread, NULL);
 243   if (err != 0) {
 244     errno = err;
 245     nbdkit_error ("pthread_create: %m");
 246     return -1;
 247   }
 248   return 0;
 249 }
 250
 251 /* This is called before processing each NBD request. */
 252 static void
 253 begin_request (void)
 254 {
 255   /* This will hang if we're paused. */
 256   ACQUIRE_LOCK_FOR_CURRENT_SCOPE (&paused);
 257
 258   /* Count the number of requests in flight. */
 259   ACQUIRE_LOCK_FOR_CURRENT_SCOPE (&count_lock);
 260   count_requests++;
 261 }
 262
 263 /* This is called after processing each NBD request. */
 264 static void
 265 end_request (void)
 266 {
 267   ACQUIRE_LOCK_FOR_CURRENT_SCOPE (&count_lock);
 268   count_requests--;
 269   pthread_cond_signal (&count_cond);
 270 }
 271
 272 /* Read data. */
 273 static int
 274 pause_pread (nbdkit_next *next,
 275              void *handle, void *buf, uint32_t count, uint64_t offset,
 276              uint32_t flags, int *err)
 277 {
 278   int r;
 279
 280   begin_request ();
 281   r = next->pread (next, buf, count, offset, flags, err);
 282   end_request ();
 283   return r;
 284 }
 285
 286 /* Write data. */
 287 static int
 288 pause_pwrite (nbdkit_next *next,
 289               void *handle,
 290               const void *buf, uint32_t count, uint64_t offset, uint32_t flags,
 291               int *err)
 292 {
 293   int r;
 294
 295   begin_request ();
 296   r = next->pwrite (next, buf, count, offset, flags, err);
 297   end_request ();
 298   return r;
 299 }
 300
 301 /* Zero data. */
 302 static int
 303 pause_zero (nbdkit_next *next,
 304             void *handle, uint32_t count, uint64_t offset, uint32_t flags,
 305             int *err)
 306 {
 307   int r;
 308
 309   begin_request ();
 310   r = next->zero (next, count, offset, flags, err);
 311   end_request ();
 312   return r;
 313 }
 314
 315 /* Trim data. */
 316 static int
 317 pause_trim (nbdkit_next *next,
 318             void *handle, uint32_t count, uint64_t offset,
 319             uint32_t flags, int *err)
 320 {
 321   int r;
 322
 323   begin_request ();
 324   r = next->trim (next, count, offset, flags, err);
 325   end_request ();
 326   return r;
 327 }
 328
 329 /* Extents. */
 330 static int
 331 pause_extents (nbdkit_next *next,
 332                void *handle, uint32_t count, uint64_t offset, uint32_t flags,
 333                struct nbdkit_extents *extents, int *err)
 334 {
 335   int r;
 336
 337   begin_request ();
 338   r = next->extents (next, count, offset, flags, extents, err);
 339   end_request ();
 340   return r;
 341 }
 342
 343 /* Cache. */
 344 static int
 345 pause_cache (nbdkit_next *next,
 346              void *handle, uint32_t count, uint64_t offset, uint32_t flags,
 347              int *err)
 348 {
 349   int r;
 350
 351   begin_request ();
 352   r = next->cache (next, count, offset, flags, err);
 353   end_request ();
 354   return r;
 355 }
 356
 357 static struct nbdkit_filter filter = {
 358   .name              = "pause",
 359   .longname          = "nbdkit pause filter",
 360   .unload            = pause_unload,
 361   .config            = pause_config,
 362   .config_complete   = pause_config_complete,
 363   .config_help       = pause_config_help,
 364   .after_fork        = pause_after_fork,
 365   .pread             = pause_pread,
 366   .pwrite            = pause_pwrite,
 367   .zero              = pause_zero,
 368   .trim              = pause_trim,
 369   .extents           = pause_extents,
 370   .cache             = pause_cache,
 371 };
 372
 373 NBDKIT_REGISTER_FILTER (filter)