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         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 void
 136 handle_single_connection (int sockin, int sockout)
 137 {
 138   const char *plugin_name;
 139   int r;
 140   struct connection *conn;
 141   int nworkers = threads ? threads : DEFAULT_PARALLEL_REQUESTS;
 142   pthread_t *workers = NULL;
 143
 144   lock_connection ();
 145
 146   if (backend->thread_model (backend) < NBDKIT_THREAD_MODEL_PARALLEL ||
 147       nworkers == 1)
 148     nworkers = 0;
 149   conn = new_connection (sockin, sockout, nworkers);
 150   if (!conn)
 151     goto done;
 152
 153   /* NB: because of an asynchronous exit backend can be set to NULL at
 154    * just about any time.
 155    */
 156   if (backend)
 157     plugin_name = backend->plugin_name (backend);
 158   else
 159     plugin_name = "(unknown)";
 160   threadlocal_set_name (plugin_name);
 161
 162   if (backend && backend->preconnect (backend, conn, read_only) == -1)
 163     goto done;
 164
 165   /* NBD handshake.
 166    *
 167    * Note that this calls the backend .open callback when it is safe
 168    * to do so (eg. after TLS authentication).
 169    */
 170   if (protocol_handshake (conn) == -1)
 171     goto done;
 172
 173   if (!nworkers) {
 174     /* No need for a separate thread. */
 175     debug ("handshake complete, processing requests serially");
 176     while (!quit && connection_get_status (conn) > 0)
 177       protocol_recv_request_send_reply (conn);
 178   }
 179   else {
 180     /* Create thread pool to process requests. */
 181     debug ("handshake complete, processing requests with %d threads",
 182            nworkers);
 183     workers = calloc (nworkers, sizeof *workers);
 184     if (unlikely (!workers)) {
 185       perror ("malloc");
 186       goto done;
 187     }
 188
 189     for (nworkers = 0; nworkers < conn->nworkers; nworkers++) {
 190       struct worker_data *worker = malloc (sizeof *worker);
 191       int err;
 192
 193       if (unlikely (!worker)) {
 194         perror ("malloc");
 195         connection_set_status (conn, -1);
 196         goto wait;
 197       }
 198       if (unlikely (asprintf (&worker->name, "%s.%d", plugin_name, nworkers)
 199                     < 0)) {
 200         perror ("asprintf");
 201         connection_set_status (conn, -1);
 202         free (worker);
 203         goto wait;
 204       }
 205       worker->conn = conn;
 206       err = pthread_create (&workers[nworkers], NULL, connection_worker,
 207                             worker);
 208       if (unlikely (err)) {
 209         errno = err;
 210         perror ("pthread_create");
 211         connection_set_status (conn, -1);
 212         free (worker);
 213         goto wait;
 214       }
 215     }
 216
 217   wait:
 218     while (nworkers)
 219       pthread_join (workers[--nworkers], NULL);
 220     free (workers);
 221   }
 222
 223   /* Finalize (for filters), called just before close. */
 224   lock_request (conn);
 225   r = backend_finalize (backend, conn);
 226   unlock_request (conn);
 227   if (r == -1)
 228     goto done;
 229
 230  done:
 231   free_connection (conn);
 232   unlock_connection ();
 233 }
 234
 235 static struct connection *
 236 new_connection (int sockin, int sockout, int nworkers)
 237 {
 238   struct connection *conn;
 239   int opt;
 240   socklen_t optlen = sizeof opt;
 241   struct backend *b;
 242
 243   conn = calloc (1, sizeof *conn);
 244   if (conn == NULL) {
 245     perror ("malloc");
 246     return NULL;
 247   }
 248
 249   conn->status_pipe[0] = conn->status_pipe[1] = -1;
 250
 251   conn->handles = calloc (backend->i + 1, sizeof *conn->handles);
 252   if (conn->handles == NULL) {
 253     perror ("malloc");
 254     goto error;
 255   }
 256   conn->nr_handles = backend->i + 1;
 257   for_each_backend (b)
 258     reset_b_conn_handle (&conn->handles[b->i]);
 259
 260   conn->status = 1;
 261   conn->nworkers = nworkers;
 262   if (nworkers) {
 263 #ifdef HAVE_PIPE2
 264     if (pipe2 (conn->status_pipe, O_NONBLOCK | O_CLOEXEC)) {
 265       perror ("pipe2");
 266       goto error;
 267     }
 268 #else
 269     /* If we were fully parallel, then this function could be
 270      * accepting connections in one thread while another thread could
 271      * be in a plugin trying to fork.  But plugins.c forced
 272      * thread_model to serialize_all_requests when it detects a lack
 273      * of atomic CLOEXEC, at which point, we can use a mutex to ensure
 274      * we aren't accepting until the plugin is not running, making
 275      * non-atomicity okay.
 276      */
 277     assert (backend->thread_model (backend) <=
 278             NBDKIT_THREAD_MODEL_SERIALIZE_ALL_REQUESTS);
 279     lock_request (NULL);
 280     if (pipe (conn->status_pipe)) {
 281       perror ("pipe");
 282       unlock_request (NULL);
 283       goto error;
 284     }
 285     if (set_nonblock (set_cloexec (conn->status_pipe[0])) == -1) {
 286       perror ("fcntl");
 287       close (conn->status_pipe[1]);
 288       unlock_request (NULL);
 289       goto error;
 290     }
 291     if (set_nonblock (set_cloexec (conn->status_pipe[1])) == -1) {
 292       perror ("fcntl");
 293       close (conn->status_pipe[0]);
 294       unlock_request (NULL);
 295       goto error;
 296     }
 297     unlock_request (NULL);
 298 #endif
 299   }
 300   conn->sockin = sockin;
 301   conn->sockout = sockout;
 302   pthread_mutex_init (&conn->request_lock, NULL);
 303   pthread_mutex_init (&conn->read_lock, NULL);
 304   pthread_mutex_init (&conn->write_lock, NULL);
 305   pthread_mutex_init (&conn->status_lock, NULL);
 306
 307   conn->recv = raw_recv;
 308   if (getsockopt (sockout, SOL_SOCKET, SO_TYPE, &opt, &optlen) == 0)
 309     conn->send = raw_send_socket;
 310   else
 311     conn->send = raw_send_other;
 312   conn->close = raw_close;
 313
 314   threadlocal_set_conn (conn);
 315
 316   return conn;
 317
 318  error:
 319   if (conn->status_pipe[0] >= 0)
 320     close (conn->status_pipe[0]);
 321   if (conn->status_pipe[1] >= 0)
 322     close (conn->status_pipe[1]);
 323   free (conn->handles);
 324   free (conn);
 325   return NULL;
 326 }
 327
 328 static void
 329 free_connection (struct connection *conn)
 330 {
 331   if (!conn)
 332     return;
 333
 334   threadlocal_set_conn (NULL);
 335   conn->close (conn);
 336   if (listen_stdin) {
 337     int fd;
 338
 339     /* Restore something to stdin/out so the rest of our code can
 340      * continue to assume that all new fds will be above stderr.
 341      * Swap directions to get EBADF on improper use of stdin/out.
 342      */
 343     fd = open ("/dev/null", O_WRONLY | O_CLOEXEC);
 344     assert (fd == 0);
 345     fd = open ("/dev/null", O_RDONLY | O_CLOEXEC);
 346     assert (fd == 1);
 347   }
 348
 349   /* Don't call the plugin again if quit has been set because the main
 350    * thread will be in the process of unloading it.  The plugin.unload
 351    * callback should always be called.
 352    */
 353   if (!quit) {
 354     lock_request (conn);
 355     backend_close (backend, conn);
 356     unlock_request (conn);
 357   }
 358
 359   if (conn->status_pipe[0] >= 0) {
 360     close (conn->status_pipe[0]);
 361     close (conn->status_pipe[1]);
 362   }
 363
 364   pthread_mutex_destroy (&conn->request_lock);
 365   pthread_mutex_destroy (&conn->read_lock);
 366   pthread_mutex_destroy (&conn->write_lock);
 367   pthread_mutex_destroy (&conn->status_lock);
 368
 369   free (conn->handles);
 370   free (conn);
 371 }
 372
 373 /* Write buffer to conn->sockout with send() and either succeed completely
 374  * (returns 0) or fail (returns -1). flags may include SEND_MORE as a hint
 375  * that this send will be followed by related data.
 376  */
 377 static int
 378 raw_send_socket (struct connection *conn, const void *vbuf, size_t len,
 379                  int flags)
 380 {
 381   int sock = conn->sockout;
 382   const char *buf = vbuf;
 383   ssize_t r;
 384   int f = 0;
 385
 386 #ifdef MSG_MORE
 387   if (flags & SEND_MORE)
 388     f |= MSG_MORE;
 389 #endif
 390   while (len > 0) {
 391     r = send (sock, buf, len, f);
 392     if (r == -1) {
 393       if (errno == EINTR || errno == EAGAIN)
 394         continue;
 395       return -1;
 396     }
 397     buf += r;
 398     len -= r;
 399   }
 400
 401   return 0;
 402 }
 403
 404 /* Write buffer to conn->sockout with write() and either succeed completely
 405  * (returns 0) or fail (returns -1). flags is ignored.
 406  */
 407 static int
 408 raw_send_other (struct connection *conn, const void *vbuf, size_t len,
 409                 int flags)
 410 {
 411   int sock = conn->sockout;
 412   const char *buf = vbuf;
 413   ssize_t r;
 414
 415   while (len > 0) {
 416     r = write (sock, buf, len);
 417     if (r == -1) {
 418       if (errno == EINTR || errno == EAGAIN)
 419         continue;
 420       return -1;
 421     }
 422     buf += r;
 423     len -= r;
 424   }
 425
 426   return 0;
 427 }
 428
 429 /* Read buffer from conn->sockin and either succeed completely
 430  * (returns > 0), read an EOF (returns 0), or fail (returns -1).
 431  */
 432 static int
 433 raw_recv (struct connection *conn, void *vbuf, size_t len)
 434 {
 435   int sock = conn->sockin;
 436   char *buf = vbuf;
 437   ssize_t r;
 438   bool first_read = true;
 439
 440   while (len > 0) {
 441     r = read (sock, buf, len);
 442     if (r == -1) {
 443       if (errno == EINTR || errno == EAGAIN)
 444         continue;
 445       return -1;
 446     }
 447     if (r == 0) {
 448       if (first_read)
 449         return 0;
 450       /* Partial record read.  This is an error. */
 451       errno = EBADMSG;
 452       return -1;
 453     }
 454     first_read = false;
 455     buf += r;
 456     len -= r;
 457   }
 458
 459   return 1;
 460 }
 461
 462 /* There's no place in the NBD protocol to send back errors from
 463  * close, so this function ignores errors.
 464  */
 465 static void
 466 raw_close (struct connection *conn)
 467 {
 468   if (conn->sockin >= 0)
 469     close (conn->sockin);
 470   if (conn->sockout >= 0 && conn->sockin != conn->sockout)
 471     close (conn->sockout);
 472 }