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->handles = calloc (backend->i + 1, sizeof *conn->handles);
 262   if (conn->handles == NULL) {
 263     perror ("malloc");
 264     goto error;
 265   }
 266   conn->nr_handles = backend->i + 1;
 267   for_each_backend (b)
 268     reset_b_conn_handle (&conn->handles[b->i]);
 269
 270   conn->status = 1;
 271   conn->nworkers = nworkers;
 272   if (nworkers) {
 273 #ifdef HAVE_PIPE2
 274     if (pipe2 (conn->status_pipe, O_NONBLOCK | O_CLOEXEC)) {
 275       perror ("pipe2");
 276       goto error;
 277     }
 278 #else
 279     /* If we were fully parallel, then this function could be
 280      * accepting connections in one thread while another thread could
 281      * be in a plugin trying to fork.  But plugins.c forced
 282      * thread_model to serialize_all_requests when it detects a lack
 283      * of atomic CLOEXEC, at which point, we can use a mutex to ensure
 284      * we aren't accepting until the plugin is not running, making
 285      * non-atomicity okay.
 286      */
 287     assert (backend->thread_model (backend) <=
 288             NBDKIT_THREAD_MODEL_SERIALIZE_ALL_REQUESTS);
 289     lock_request (NULL);
 290     if (pipe (conn->status_pipe)) {
 291       perror ("pipe");
 292       unlock_request (NULL);
 293       goto error;
 294     }
 295     if (set_nonblock (set_cloexec (conn->status_pipe[0])) == -1) {
 296       perror ("fcntl");
 297       close (conn->status_pipe[1]);
 298       unlock_request (NULL);
 299       goto error;
 300     }
 301     if (set_nonblock (set_cloexec (conn->status_pipe[1])) == -1) {
 302       perror ("fcntl");
 303       close (conn->status_pipe[0]);
 304       unlock_request (NULL);
 305       goto error;
 306     }
 307     unlock_request (NULL);
 308 #endif
 309   }
 310   conn->sockin = sockin;
 311   conn->sockout = sockout;
 312   pthread_mutex_init (&conn->request_lock, NULL);
 313   pthread_mutex_init (&conn->read_lock, NULL);
 314   pthread_mutex_init (&conn->write_lock, NULL);
 315   pthread_mutex_init (&conn->status_lock, NULL);
 316
 317   conn->recv = raw_recv;
 318   if (getsockopt (sockout, SOL_SOCKET, SO_TYPE, &opt, &optlen) == 0)
 319     conn->send = raw_send_socket;
 320   else
 321     conn->send = raw_send_other;
 322   conn->close = raw_close;
 323
 324   threadlocal_set_conn (conn);
 325
 326   return conn;
 327
 328  error:
 329   if (conn->status_pipe[0] >= 0)
 330     close (conn->status_pipe[0]);
 331   if (conn->status_pipe[1] >= 0)
 332     close (conn->status_pipe[1]);
 333   free (conn->handles);
 334   free (conn);
 335   return NULL;
 336 }
 337
 338 static void
 339 free_connection (struct connection *conn)
 340 {
 341   if (!conn)
 342     return;
 343
 344   threadlocal_set_conn (NULL);
 345   conn->close (conn);
 346   if (listen_stdin) {
 347     int fd;
 348
 349     /* Restore something to stdin/out so the rest of our code can
 350      * continue to assume that all new fds will be above stderr.
 351      * Swap directions to get EBADF on improper use of stdin/out.
 352      */
 353     fd = open ("/dev/null", O_WRONLY | O_CLOEXEC);
 354     assert (fd == 0);
 355     fd = open ("/dev/null", O_RDONLY | O_CLOEXEC);
 356     assert (fd == 1);
 357   }
 358
 359   /* Don't call the plugin again if quit has been set because the main
 360    * thread will be in the process of unloading it.  The plugin.unload
 361    * callback should always be called.
 362    */
 363   if (!quit && connection_get_handle (conn, 0)) {
 364     lock_request (conn);
 365     backend_close (backend, conn);
 366     unlock_request (conn);
 367   }
 368
 369   if (conn->status_pipe[0] >= 0) {
 370     close (conn->status_pipe[0]);
 371     close (conn->status_pipe[1]);
 372   }
 373
 374   pthread_mutex_destroy (&conn->request_lock);
 375   pthread_mutex_destroy (&conn->read_lock);
 376   pthread_mutex_destroy (&conn->write_lock);
 377   pthread_mutex_destroy (&conn->status_lock);
 378
 379   free (conn->handles);
 380   free (conn);
 381 }
 382
 383 /* Write buffer to conn->sockout with send() and either succeed completely
 384  * (returns 0) or fail (returns -1). flags may include SEND_MORE as a hint
 385  * that this send will be followed by related data.
 386  */
 387 static int
 388 raw_send_socket (struct connection *conn, const void *vbuf, size_t len,
 389                  int flags)
 390 {
 391   int sock = conn->sockout;
 392   const char *buf = vbuf;
 393   ssize_t r;
 394   int f = 0;
 395
 396 #ifdef MSG_MORE
 397   if (flags & SEND_MORE)
 398     f |= MSG_MORE;
 399 #endif
 400   while (len > 0) {
 401     r = send (sock, buf, len, f);
 402     if (r == -1) {
 403       if (errno == EINTR || errno == EAGAIN)
 404         continue;
 405       return -1;
 406     }
 407     buf += r;
 408     len -= r;
 409   }
 410
 411   return 0;
 412 }
 413
 414 /* Write buffer to conn->sockout with write() and either succeed completely
 415  * (returns 0) or fail (returns -1). flags is ignored.
 416  */
 417 static int
 418 raw_send_other (struct connection *conn, const void *vbuf, size_t len,
 419                 int flags)
 420 {
 421   int sock = conn->sockout;
 422   const char *buf = vbuf;
 423   ssize_t r;
 424
 425   while (len > 0) {
 426     r = write (sock, buf, len);
 427     if (r == -1) {
 428       if (errno == EINTR || errno == EAGAIN)
 429         continue;
 430       return -1;
 431     }
 432     buf += r;
 433     len -= r;
 434   }
 435
 436   return 0;
 437 }
 438
 439 /* Read buffer from conn->sockin and either succeed completely
 440  * (returns > 0), read an EOF (returns 0), or fail (returns -1).
 441  */
 442 static int
 443 raw_recv (struct connection *conn, void *vbuf, size_t len)
 444 {
 445   int sock = conn->sockin;
 446   char *buf = vbuf;
 447   ssize_t r;
 448   bool first_read = true;
 449
 450   while (len > 0) {
 451     r = read (sock, buf, len);
 452     if (r == -1) {
 453       if (errno == EINTR || errno == EAGAIN)
 454         continue;
 455       return -1;
 456     }
 457     if (r == 0) {
 458       if (first_read)
 459         return 0;
 460       /* Partial record read.  This is an error. */
 461       errno = EBADMSG;
 462       return -1;
 463     }
 464     first_read = false;
 465     buf += r;
 466     len -= r;
 467   }
 468
 469   return 1;
 470 }
 471
 472 /* There's no place in the NBD protocol to send back errors from
 473  * close, so this function ignores errors.
 474  */
 475 static void
 476 raw_close (struct connection *conn)
 477 {
 478   if (conn->sockin >= 0)
 479     close (conn->sockin);
 480   if (conn->sockout >= 0 && conn->sockin != conn->sockout)
 481     close (conn->sockout);
 482 }