server/connections.c

   1 /* nbdkit
   2  * Copyright (C) 2013-2019 Red Hat Inc.
   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 <stdint.h>
  38 #include <inttypes.h>
  39 #include <string.h>
  40 #include <unistd.h>
  41 #include <sys/socket.h>
  42 #include <fcntl.h>
  43 #include <assert.h>
  44
  45 #include "internal.h"
  46 #include "utils.h"
  47
  48 /* Default number of parallel requests. */
  49 #define DEFAULT_PARALLEL_REQUESTS 16
  50
  51 static struct connection *new_connection (int sockin, int sockout,
  52                                           int nworkers);
  53 static void free_connection (struct connection *conn);
  54
  55 /* Don't call these raw socket functions directly.  Use conn->recv etc. */
  56 static int raw_recv (struct connection *, void *buf, size_t len);
  57 static int raw_send_socket (struct connection *, const void *buf, size_t len,
  58                             int flags);
  59 static int raw_send_other (struct connection *, const void *buf, size_t len,
  60                            int flags);
  61 static void raw_close (struct connection *);
  62
  63 void *
  64 connection_get_handle (struct connection *conn, size_t i)
  65 {
  66   assert (i < conn->nr_handles);
  67   return conn->handles[i].handle;
  68 }
  69
  70 int
  71 connection_get_status (struct connection *conn)
  72 {
  73   int r;
  74
  75   if (conn->nworkers &&
  76       pthread_mutex_lock (&conn->status_lock))
  77     abort ();
  78   r = conn->status;
  79   if (conn->nworkers &&
  80       pthread_mutex_unlock (&conn->status_lock))
  81     abort ();
  82   return r;
  83 }
  84
  85 /* Update the status if the new value is lower than the existing value.
  86  * For convenience, return the incoming value.
  87  */
  88 int
  89 connection_set_status (struct connection *conn, int value)
  90 {
  91   if (conn->nworkers &&
  92       pthread_mutex_lock (&conn->status_lock))
  93     abort ();
  94   if (value < conn->status) {
  95     if (conn->nworkers && conn->status > 0) {
  96       char c = 0;
  97
  98       assert (conn->status_pipe[1] >= 0);
  99       if (write (conn->status_pipe[1], &c, 1) != 1 && errno != EAGAIN)
 100         nbdkit_debug ("failed to notify pipe-to-self: %m");
 101     }
 102     conn->status = value;
 103   }
 104   if (conn->nworkers &&
 105       pthread_mutex_unlock (&conn->status_lock))
 106     abort ();
 107   return value;
 108 }
 109
 110 struct worker_data {
 111   struct connection *conn;
 112   char *name;
 113 };
 114
 115 static void *
 116 connection_worker (void *data)
 117 {
 118   struct worker_data *worker = data;
 119   struct connection *conn = worker->conn;
 120   char *name = worker->name;
 121
 122   debug ("starting worker thread %s", name);
 123   threadlocal_new_server_thread ();
 124   threadlocal_set_name (name);
 125   threadlocal_set_conn (conn);
 126   free (worker);
 127
 128   while (!quit && connection_get_status (conn) > 0)
 129     protocol_recv_request_send_reply (conn);
 130   debug ("exiting worker thread %s", threadlocal_get_name ());
 131   free (name);
 132   return NULL;
 133 }
 134
 135 static int
 136 _handle_single_connection (int sockin, int sockout)
 137 {
 138   const char *plugin_name;
 139   int ret = -1, r;
 140   struct connection *conn;
 141   int nworkers = threads ? threads : DEFAULT_PARALLEL_REQUESTS;
 142   pthread_t *workers = NULL;
 143
 144   if (backend->thread_model (backend) < NBDKIT_THREAD_MODEL_PARALLEL ||
 145       nworkers == 1)
 146     nworkers = 0;
 147   conn = new_connection (sockin, sockout, nworkers);
 148   if (!conn)
 149     goto done;
 150
 151   /* NB: because of an asynchronous exit backend can be set to NULL at
 152    * just about any time.
 153    */
 154   if (backend)
 155     plugin_name = backend->plugin_name (backend);
 156   else
 157     plugin_name = "(unknown)";
 158   threadlocal_set_name (plugin_name);
 159
 160   /* NBD handshake.
 161    *
 162    * Note that this calls the backend .open callback when it is safe
 163    * to do so (eg. after TLS authentication).
 164    */
 165   if (protocol_handshake (conn) == -1)
 166     goto done;
 167
 168   if (!nworkers) {
 169     /* No need for a separate thread. */
 170     debug ("handshake complete, processing requests serially");
 171     while (!quit && connection_get_status (conn) > 0)
 172       protocol_recv_request_send_reply (conn);
 173   }
 174   else {
 175     /* Create thread pool to process requests. */
 176     debug ("handshake complete, processing requests with %d threads",
 177            nworkers);
 178     workers = calloc (nworkers, sizeof *workers);
 179     if (!workers) {
 180       perror ("malloc");
 181       goto done;
 182     }
 183
 184     for (nworkers = 0; nworkers < conn->nworkers; nworkers++) {
 185       struct worker_data *worker = malloc (sizeof *worker);
 186       int err;
 187
 188       if (!worker) {
 189         perror ("malloc");
 190         connection_set_status (conn, -1);
 191         goto wait;
 192       }
 193       if (asprintf (&worker->name, "%s.%d", plugin_name, nworkers) < 0) {
 194         perror ("asprintf");
 195         connection_set_status (conn, -1);
 196         free (worker);
 197         goto wait;
 198       }
 199       worker->conn = conn;
 200       err = pthread_create (&workers[nworkers], NULL, connection_worker,
 201                             worker);
 202       if (err) {
 203         errno = err;
 204         perror ("pthread_create");
 205         connection_set_status (conn, -1);
 206         free (worker);
 207         goto wait;
 208       }
 209     }
 210
 211   wait:
 212     while (nworkers)
 213       pthread_join (workers[--nworkers], NULL);
 214     free (workers);
 215   }
 216
 217   /* Finalize (for filters), called just before close. */
 218   lock_request (conn);
 219   if (backend)
 220     r = backend->finalize (backend, conn);
 221   else
 222     r = 0;
 223   unlock_request (conn);
 224   if (r == -1)
 225     goto done;
 226
 227   ret = connection_get_status (conn);
 228  done:
 229   free_connection (conn);
 230   return ret;
 231 }
 232
 233 int
 234 handle_single_connection (int sockin, int sockout)
 235 {
 236   int r;
 237
 238   lock_connection ();
 239   r = _handle_single_connection (sockin, sockout);
 240   unlock_connection ();
 241
 242   return r;
 243 }
 244
 245 static struct connection *
 246 new_connection (int sockin, int sockout, int nworkers)
 247 {
 248   struct connection *conn;
 249   int opt;
 250   socklen_t optlen = sizeof opt;
 251   struct backend *b;
 252
 253   conn = calloc (1, sizeof *conn);
 254   if (conn == NULL) {
 255     perror ("malloc");
 256     return NULL;
 257   }
 258
 259   conn->status_pipe[0] = conn->status_pipe[1] = -1;
 260
 261   conn->exportname = strdup ("");
 262   if (conn->exportname == NULL) {
 263     perror ("strdup");
 264     goto error;
 265   }
 266   conn->handles = calloc (backend->i + 1, sizeof *conn->handles);
 267   if (conn->handles == NULL) {
 268     perror ("malloc");
 269     goto error;
 270   }
 271   conn->nr_handles = backend->i + 1;
 272   for_each_backend (b)
 273     reset_b_conn_handle (&conn->handles[b->i]);
 274
 275   conn->status = 1;
 276   conn->nworkers = nworkers;
 277   if (nworkers) {
 278 #ifdef HAVE_PIPE2
 279     if (pipe2 (conn->status_pipe, O_NONBLOCK | O_CLOEXEC)) {
 280       perror ("pipe2");
 281       goto error;
 282     }
 283 #else
 284     /* If we were fully parallel, then this function could be
 285      * accepting connections in one thread while another thread could
 286      * be in a plugin trying to fork.  But plugins.c forced
 287      * thread_model to serialize_all_requests when it detects a lack
 288      * of atomic CLOEXEC, at which point, we can use a mutex to ensure
 289      * we aren't accepting until the plugin is not running, making
 290      * non-atomicity okay.
 291      */
 292     assert (backend->thread_model (backend) <=
 293             NBDKIT_THREAD_MODEL_SERIALIZE_ALL_REQUESTS);
 294     lock_request (NULL);
 295     if (pipe (conn->status_pipe)) {
 296       perror ("pipe");
 297       unlock_request (NULL);
 298       goto error;
 299     }
 300     if (set_nonblock (set_cloexec (conn->status_pipe[0])) == -1) {
 301       perror ("fcntl");
 302       close (conn->status_pipe[1]);
 303       unlock_request (NULL);
 304       goto error;
 305     }
 306     if (set_nonblock (set_cloexec (conn->status_pipe[1])) == -1) {
 307       perror ("fcntl");
 308       close (conn->status_pipe[0]);
 309       unlock_request (NULL);
 310       goto error;
 311     }
 312     unlock_request (NULL);
 313 #endif
 314   }
 315   conn->sockin = sockin;
 316   conn->sockout = sockout;
 317   pthread_mutex_init (&conn->request_lock, NULL);
 318   pthread_mutex_init (&conn->read_lock, NULL);
 319   pthread_mutex_init (&conn->write_lock, NULL);
 320   pthread_mutex_init (&conn->status_lock, NULL);
 321
 322   conn->recv = raw_recv;
 323   if (getsockopt (sockout, SOL_SOCKET, SO_TYPE, &opt, &optlen) == 0)
 324     conn->send = raw_send_socket;
 325   else
 326     conn->send = raw_send_other;
 327   conn->close = raw_close;
 328
 329   threadlocal_set_conn (conn);
 330
 331   return conn;
 332
 333  error:
 334   if (conn->status_pipe[0] >= 0)
 335     close (conn->status_pipe[0]);
 336   if (conn->status_pipe[1] >= 0)
 337     close (conn->status_pipe[1]);
 338   free (conn->exportname);
 339   free (conn->handles);
 340   free (conn);
 341   return NULL;
 342 }
 343
 344 static void
 345 free_connection (struct connection *conn)
 346 {
 347   if (!conn)
 348     return;
 349
 350   threadlocal_set_conn (NULL);
 351   conn->close (conn);
 352   if (listen_stdin) {
 353     int fd;
 354
 355     /* Restore something to stdin/out so the rest of our code can
 356      * continue to assume that all new fds will be above stderr.
 357      * Swap directions to get EBADF on improper use of stdin/out.
 358      */
 359     fd = open ("/dev/null", O_WRONLY | O_CLOEXEC);
 360     assert (fd == 0);
 361     fd = open ("/dev/null", O_RDONLY | O_CLOEXEC);
 362     assert (fd == 1);
 363   }
 364
 365   /* Don't call the plugin again if quit has been set because the main
 366    * thread will be in the process of unloading it.  The plugin.unload
 367    * callback should always be called.
 368    */
 369   if (!quit && connection_get_handle (conn, 0)) {
 370     lock_request (conn);
 371     backend_close (backend, conn);
 372     unlock_request (conn);
 373   }
 374
 375   if (conn->status_pipe[0] >= 0) {
 376     close (conn->status_pipe[0]);
 377     close (conn->status_pipe[1]);
 378   }
 379
 380   pthread_mutex_destroy (&conn->request_lock);
 381   pthread_mutex_destroy (&conn->read_lock);
 382   pthread_mutex_destroy (&conn->write_lock);
 383   pthread_mutex_destroy (&conn->status_lock);
 384
 385   free (conn->handles);
 386   free (conn->exportname);
 387   free (conn);
 388 }
 389
 390 /* Write buffer to conn->sockout with send() and either succeed completely
 391  * (returns 0) or fail (returns -1). flags may include SEND_MORE as a hint
 392  * that this send will be followed by related data.
 393  */
 394 static int
 395 raw_send_socket (struct connection *conn, const void *vbuf, size_t len,
 396                  int flags)
 397 {
 398   int sock = conn->sockout;
 399   const char *buf = vbuf;
 400   ssize_t r;
 401   int f = 0;
 402
 403 #ifdef MSG_MORE
 404   if (flags & SEND_MORE)
 405     f |= MSG_MORE;
 406 #endif
 407   while (len > 0) {
 408     r = send (sock, buf, len, f);
 409     if (r == -1) {
 410       if (errno == EINTR || errno == EAGAIN)
 411         continue;
 412       return -1;
 413     }
 414     buf += r;
 415     len -= r;
 416   }
 417
 418   return 0;
 419 }
 420
 421 /* Write buffer to conn->sockout with write() and either succeed completely
 422  * (returns 0) or fail (returns -1). flags is ignored.
 423  */
 424 static int
 425 raw_send_other (struct connection *conn, const void *vbuf, size_t len,
 426                 int flags)
 427 {
 428   int sock = conn->sockout;
 429   const char *buf = vbuf;
 430   ssize_t r;
 431
 432   while (len > 0) {
 433     r = write (sock, buf, len);
 434     if (r == -1) {
 435       if (errno == EINTR || errno == EAGAIN)
 436         continue;
 437       return -1;
 438     }
 439     buf += r;
 440     len -= r;
 441   }
 442
 443   return 0;
 444 }
 445
 446 /* Read buffer from conn->sockin and either succeed completely
 447  * (returns > 0), read an EOF (returns 0), or fail (returns -1).
 448  */
 449 static int
 450 raw_recv (struct connection *conn, void *vbuf, size_t len)
 451 {
 452   int sock = conn->sockin;
 453   char *buf = vbuf;
 454   ssize_t r;
 455   bool first_read = true;
 456
 457   while (len > 0) {
 458     r = read (sock, buf, len);
 459     if (r == -1) {
 460       if (errno == EINTR || errno == EAGAIN)
 461         continue;
 462       return -1;
 463     }
 464     if (r == 0) {
 465       if (first_read)
 466         return 0;
 467       /* Partial record read.  This is an error. */
 468       errno = EBADMSG;
 469       return -1;
 470     }
 471     first_read = false;
 472     buf += r;
 473     len -= r;
 474   }
 475
 476   return 1;
 477 }
 478
 479 /* There's no place in the NBD protocol to send back errors from
 480  * close, so this function ignores errors.
 481  */
 482 static void
 483 raw_close (struct connection *conn)
 484 {
 485   if (conn->sockin >= 0)
 486     close (conn->sockin);
 487   if (conn->sockout >= 0 && conn->sockin != conn->sockout)
 488     close (conn->sockout);
 489 }