_handle_request() was never returning anything but 0; and had to communicate the error value through a parameter by reference. As there is no return value that a plugin can use to declare the connection useless, we can consolidate this into just returning the error value. Meanwhile, we don't need a wrapper layer of handle_request() just to grab locks; so shuffle the naming a bit to reduce the lines of code. Signed-off-by: Eric Blake <eblake(a)redhat.com&gt; --- src/connections.c | 55 ++++++++++++++++--------------------------------------- 1 file changed, 16 insertions(+), 39 deletions(-) diff --git a/src/connections.c b/src/connections.c index f3be63d..0ad252c 100644 --- a/src/connections.c +++ b/src/connections.c @@ -749,16 +749,14 @@ get_error (struct connection *conn) * check them again. 'buf' is either the data to be written or the * data to be returned, and points to a buffer of size 'count' bytes. * - * Only returns -1 if there is a fatal error and the connection cannot - * continue. - * - * On read/write errors, sets *error appropriately and returns 0. + * In all cases, the return value is the system errno value that will + * later be converted to the nbd error to send back to the client (0 + * for success). */ -static int -_handle_request (struct connection *conn, - uint32_t cmd, uint32_t flags, uint64_t offset, uint32_t count, - void *buf, - uint32_t *error) +static uint32_t +handle_request (struct connection *conn, + uint32_t cmd, uint32_t flags, uint64_t offset, uint32_t count, + void *buf) { bool flush_after_command; int r; @@ -776,40 +774,35 @@ _handle_request (struct connection *conn, case NBD_CMD_READ: r = plugin_pread (conn, buf, count, offset); if (r == -1) { - *error = get_error (conn); - return 0; + return get_error (conn); } break; case NBD_CMD_WRITE: r = plugin_pwrite (conn, buf, count, offset); if (r == -1) { - *error = get_error (conn); - return 0; + return get_error (conn); } break; case NBD_CMD_FLUSH: r = plugin_flush (conn); if (r == -1) { - *error = get_error (conn); - return 0; + return get_error (conn); } break; case NBD_CMD_TRIM: r = plugin_trim (conn, count, offset); if (r == -1) { - *error = get_error (conn); - return 0; + return get_error (conn); } break; case NBD_CMD_WRITE_ZEROES: r = plugin_zero (conn, count, offset, !(flags & NBD_CMD_FLAG_NO_HOLE)); if (r == -1) { - *error = get_error (conn); - return 0; + return get_error (conn); } break; @@ -820,8 +813,7 @@ _handle_request (struct connection *conn, if (flush_after_command) { r = plugin_flush (conn); if (r == -1) { - *error = get_error (conn); - return 0; + return get_error (conn); } } @@ -829,21 +821,6 @@ _handle_request (struct connection *conn, } static int -handle_request (struct connection *conn, - uint32_t cmd, uint32_t flags, uint64_t offset, uint32_t count, - void *buf, - uint32_t *error) -{ - int r; - - plugin_lock_request (conn); - r = _handle_request (conn, cmd, flags, offset, count, buf, error); - plugin_unlock_request (conn); - - return r; -} - -static int skip_over_write_buffer (int sock, size_t count) { char buf[BUFSIZ]; @@ -974,9 +951,9 @@ recv_request_send_reply (struct connection *conn) } /* Perform the request. Only this part happens inside the request lock. */ - r = handle_request (conn, cmd, flags, offset, count, buf, &error); - if (r == -1) - return -1; + plugin_lock_request (conn); + error = handle_request (conn, cmd, flags, offset, count, buf); + plugin_unlock_request (conn); /* Send the reply packet. */ send_reply: -- 2.13.6

In preparation for parallel processing, we need to be sure that two threads belonging to the same connection cannot interleave their I/O except at message boundaries. Add a mutex around all reads and writes that must occur as a group (for now, there is no contention for either mutex). Signed-off-by: Eric Blake <eblake(a)redhat.com&gt; --- src/connections.c | 27 +++++++++++++++++++++++++-- 1 file changed, 25 insertions(+), 2 deletions(-) diff --git a/src/connections.c b/src/connections.c index dada9aa..dd43a9a 100644 --- a/src/connections.c +++ b/src/connections.c @@ -62,6 +62,8 @@ /* Connection structure. */ struct connection { pthread_mutex_t request_lock; + pthread_mutex_t read_lock; + pthread_mutex_t write_lock; void *handle; void *crypto_session; @@ -206,6 +208,8 @@ new_connection (int sockin, int sockout) conn->sockin = sockin; conn->sockout = sockout; pthread_mutex_init (&conn->request_lock, NULL); + pthread_mutex_init (&conn->read_lock, NULL); + pthread_mutex_init (&conn->write_lock, NULL); conn->recv = raw_recv; conn->send = raw_send; @@ -223,6 +227,8 @@ free_connection (struct connection *conn) conn->close (conn); pthread_mutex_destroy (&conn->request_lock); + pthread_mutex_destroy (&conn->read_lock); + pthread_mutex_destroy (&conn->write_lock); /* Don't call the plugin again if quit has been set because the main * thread will be in the process of unloading it. The plugin.unload @@ -888,19 +894,23 @@ recv_request_send_reply (struct connection *conn) CLEANUP_FREE char *buf = NULL; /* Read the request packet. */ + pthread_mutex_lock (&conn->read_lock); r = conn->recv (conn, &request, sizeof request); if (r == -1) { nbdkit_error ("read request: %m"); + pthread_mutex_unlock (&conn->read_lock); return -1; } if (r == 0) { debug ("client closed input socket, closing connection"); + pthread_mutex_unlock (&conn->read_lock); return 0; /* disconnect */ } magic = be32toh (request.magic); if (magic != NBD_REQUEST_MAGIC) { nbdkit_error ("invalid request: 'magic' field is incorrect (0x%x)", magic); + pthread_mutex_unlock (&conn->read_lock); return -1; } @@ -913,14 +923,18 @@ recv_request_send_reply (struct connection *conn) if (cmd == NBD_CMD_DISC) { debug ("client sent disconnect command, closing connection"); + pthread_mutex_unlock (&conn->read_lock); return 0; /* disconnect */ } /* Validate the request. */ if (!validate_request (conn, cmd, flags, offset, count, &error)) { if (cmd == NBD_CMD_WRITE && - skip_over_write_buffer (conn->sockin, count) < 0) + skip_over_write_buffer (conn->sockin, count) < 0) { + pthread_mutex_unlock (&conn->read_lock); return -1; + } + pthread_mutex_unlock (&conn->read_lock); goto send_reply; } @@ -931,8 +945,11 @@ recv_request_send_reply (struct connection *conn) perror ("malloc"); error = ENOMEM; if (cmd == NBD_CMD_WRITE && - skip_over_write_buffer (conn->sockin, count) < 0) + skip_over_write_buffer (conn->sockin, count) < 0) { + pthread_mutex_unlock (&conn->read_lock); return -1; + } + pthread_mutex_unlock (&conn->read_lock); goto send_reply; } } @@ -946,9 +963,11 @@ recv_request_send_reply (struct connection *conn) } if (r == -1) { nbdkit_error ("read data: %m"); + pthread_mutex_unlock (&conn->read_lock); return -1; } } + pthread_mutex_unlock (&conn->read_lock); /* Perform the request. Only this part happens inside the request lock. */ if (quit) { @@ -962,6 +981,7 @@ recv_request_send_reply (struct connection *conn) /* Send the reply packet. */ send_reply: + pthread_mutex_lock (&conn->write_lock); reply.magic = htobe32 (NBD_REPLY_MAGIC); reply.handle = request.handle; reply.error = htobe32 (nbd_errno (error)); @@ -978,6 +998,7 @@ recv_request_send_reply (struct connection *conn) r = conn->send (conn, &reply, sizeof reply); if (r == -1) { nbdkit_error ("write reply: %m"); + pthread_mutex_unlock (&conn->write_lock); return -1; } @@ -986,9 +1007,11 @@ recv_request_send_reply (struct connection *conn) r = conn->send (conn, buf, count); if (r == -1) { nbdkit_error ("write data: %m"); + pthread_mutex_unlock (&conn->write_lock); return -1; } } + pthread_mutex_unlock (&conn->write_lock); return 1; /* command processed ok */ } -- 2.13.6

Once we have multiple threads during parallel processing, we need to be sure that any I/O error flagged by one thread prevents the next thread from attempting I/O. Although we already have a separate lock for reads and writes, it's easier if status is shared by both actions, which needs yet another mutex; however we can optimize (via accessor functions) and only need to use the mutex if there are actually multiple threads at work. The next thing to notice is that because we now update status at all important points, the return value of _handle_single_connection() can just be the latest status, and recv_request_send_reply() can now return void. This will come in handy later as it will avoid trying to coordinate a value out of multiple threads. Signed-off-by: Eric Blake <eblake(a)redhat.com&gt; --- src/connections.c | 99 ++++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 69 insertions(+), 30 deletions(-) diff --git a/src/connections.c b/src/connections.c index dd43a9a..75d8884 100644 --- a/src/connections.c +++ b/src/connections.c @@ -64,8 +64,11 @@ struct connection { pthread_mutex_t request_lock; pthread_mutex_t read_lock; pthread_mutex_t write_lock; + pthread_mutex_t status_lock; + int status; /* 1 for more I/O with client, 0 for shutdown, -1 on error */ void *handle; void *crypto_session; + int nworkers; /* TODO set up a thread pool for parallel workers */ uint64_t exportsize; int readonly; @@ -83,7 +86,7 @@ struct connection { static struct connection *new_connection (int sockin, int sockout); static void free_connection (struct connection *conn); static int negotiate_handshake (struct connection *conn); -static int recv_request_send_reply (struct connection *conn); +static void recv_request_send_reply (struct connection *conn); /* Don't call these raw socket functions directly. Use conn->recv etc. */ static int raw_recv (struct connection *, void *buf, size_t len); @@ -146,40 +149,58 @@ connection_set_close (struct connection *conn, connection_close_function close) } static int +get_status (struct connection *conn) +{ + int r; + + if (conn->nworkers) + pthread_mutex_lock (&conn->status_lock); + r = conn->status; + if (conn->nworkers) + pthread_mutex_unlock (&conn->status_lock); + return r; +} + +/* Update the status if the new value is lower than the existing value. */ +static void +set_status (struct connection *conn, int value) +{ + if (conn->nworkers) + pthread_mutex_lock (&conn->status_lock); + if (value < conn->status) + conn->status = value; + if (conn->nworkers) + pthread_mutex_unlock (&conn->status_lock); +} + +static int _handle_single_connection (int sockin, int sockout) { - int r; + int r = -1; struct connection *conn = new_connection (sockin, sockout); if (!conn) - goto err; + goto done; if (plugin_open (conn, readonly) == -1) - goto err; + goto done; threadlocal_set_name (plugin_name ()); /* Handshake. */ if (negotiate_handshake (conn) == -1) - goto err; + goto done; /* Process requests. XXX Allow these to be dispatched in parallel using * a thread pool. */ - while (!quit) { - r = recv_request_send_reply (conn); - if (r == -1) - goto err; - if (r == 0) - break; - } + while (!quit && get_status (conn) > 0) + recv_request_send_reply (conn); + r = get_status (conn); + done: free_connection (conn); - return 0; - - err: - free_connection (conn); - return -1; + return r; } int @@ -205,11 +226,13 @@ new_connection (int sockin, int sockout) return NULL; } + conn->status = 1; conn->sockin = sockin; conn->sockout = sockout; pthread_mutex_init (&conn->request_lock, NULL); pthread_mutex_init (&conn->read_lock, NULL); pthread_mutex_init (&conn->write_lock, NULL); + pthread_mutex_init (&conn->status_lock, NULL); conn->recv = raw_recv; conn->send = raw_send; @@ -229,6 +252,7 @@ free_connection (struct connection *conn) pthread_mutex_destroy (&conn->request_lock); pthread_mutex_destroy (&conn->read_lock); pthread_mutex_destroy (&conn->write_lock); + pthread_mutex_destroy (&conn->status_lock); /* Don't call the plugin again if quit has been set because the main * thread will be in the process of unloading it. The plugin.unload @@ -883,7 +907,7 @@ nbd_errno (int error) } } -static int +static void recv_request_send_reply (struct connection *conn) { int r; @@ -895,23 +919,30 @@ recv_request_send_reply (struct connection *conn) /* Read the request packet. */ pthread_mutex_lock (&conn->read_lock); + if (get_status (conn) < 0) { + pthread_mutex_unlock (&conn->read_lock); + return; + } r = conn->recv (conn, &request, sizeof request); if (r == -1) { nbdkit_error ("read request: %m"); + set_status (conn, -1); pthread_mutex_unlock (&conn->read_lock); - return -1; + return; } if (r == 0) { debug ("client closed input socket, closing connection"); + set_status (conn, 0); pthread_mutex_unlock (&conn->read_lock); - return 0; /* disconnect */ + return; /* disconnect */ } magic = be32toh (request.magic); if (magic != NBD_REQUEST_MAGIC) { nbdkit_error ("invalid request: 'magic' field is incorrect (0x%x)", magic); + set_status (conn, -1); pthread_mutex_unlock (&conn->read_lock); - return -1; + return; } cmd = be32toh (request.type); @@ -923,16 +954,18 @@ recv_request_send_reply (struct connection *conn) if (cmd == NBD_CMD_DISC) { debug ("client sent disconnect command, closing connection"); + set_status (conn, 0); pthread_mutex_unlock (&conn->read_lock); - return 0; /* disconnect */ + return; /* disconnect */ } /* Validate the request. */ if (!validate_request (conn, cmd, flags, offset, count, &error)) { if (cmd == NBD_CMD_WRITE && skip_over_write_buffer (conn->sockin, count) < 0) { + set_status (conn, -1); pthread_mutex_unlock (&conn->read_lock); - return -1; + return; } pthread_mutex_unlock (&conn->read_lock); goto send_reply; @@ -946,8 +979,9 @@ recv_request_send_reply (struct connection *conn) error = ENOMEM; if (cmd == NBD_CMD_WRITE && skip_over_write_buffer (conn->sockin, count) < 0) { + set_status (conn, -1); pthread_mutex_unlock (&conn->read_lock); - return -1; + return; } pthread_mutex_unlock (&conn->read_lock); goto send_reply; @@ -963,14 +997,15 @@ recv_request_send_reply (struct connection *conn) } if (r == -1) { nbdkit_error ("read data: %m"); + set_status (conn, -1); pthread_mutex_unlock (&conn->read_lock); - return -1; + return; } } pthread_mutex_unlock (&conn->read_lock); /* Perform the request. Only this part happens inside the request lock. */ - if (quit) { + if (quit || !get_status (conn)) { error = ESHUTDOWN; } else { @@ -982,6 +1017,10 @@ recv_request_send_reply (struct connection *conn) /* Send the reply packet. */ send_reply: pthread_mutex_lock (&conn->write_lock); + if (get_status (conn) < 0) { + pthread_mutex_unlock (&conn->write_lock); + return; + } reply.magic = htobe32 (NBD_REPLY_MAGIC); reply.handle = request.handle; reply.error = htobe32 (nbd_errno (error)); @@ -999,7 +1038,8 @@ recv_request_send_reply (struct connection *conn) if (r == -1) { nbdkit_error ("write reply: %m"); pthread_mutex_unlock (&conn->write_lock); - return -1; + set_status (conn, -1); + return; } /* Send the read data buffer. */ @@ -1007,13 +1047,12 @@ recv_request_send_reply (struct connection *conn) r = conn->send (conn, buf, count); if (r == -1) { nbdkit_error ("write data: %m"); + set_status (conn, -1); pthread_mutex_unlock (&conn->write_lock); - return -1; + return; } } pthread_mutex_unlock (&conn->write_lock); - - return 1; /* command processed ok */ } /* Write buffer to conn->sockout and either succeed completely -- 2.13.6

It's finally time to implement one of the TODO items: we want to support a thread pool of parallel readers from the client, in order to allow multiple in-flight operations with potential out-of-order completion. We also need at least one plugin that supports parallel processing for testing the option; the file plugin fits the bill. Add and document a new command line option, -t/--threads=N, which controls how many threads to create per connection (although we only ever spawn multiple threads if the plugin is parallel, since otherwise, at most one thread is running at a time anyway). Setting -t 1 forces a parallel plugin to behave serialized, setting to other values allows tuning for performance; the default of 16 matches the choice of MAX_NBD_REQUESTS used in qemu. One easy way to test: term1$ echo hello > junk term1$ ./nbdkit -f -v -r file file=junk rdelay=2s wdelay=1s term2$ qemu-io -f raw nbd://localhost:10809/ --trace='nbd_*' \ -c 'aio_read 0 1' -c 'aio_write -P 0x6c 2 2' -c 'aio_flush' If the write completes before the read, then nbdkit was properly handling things in parallel with out-of-order replies. Signed-off-by: Eric Blake <eblake(a)redhat.com&gt; --- TODO | 7 ------- docs/nbdkit.pod | 12 +++++++++++- nbdkit.in | 2 +- plugins/file/file.c | 2 +- src/connections.c | 10 +++++++--- src/internal.h | 2 ++ src/main.c | 20 ++++++++++++++++++-- src/plugins.c | 8 ++++++++ 8 files changed, 48 insertions(+), 15 deletions(-) diff --git a/TODO b/TODO index 6c5bb5b..db7469b 100644 --- a/TODO +++ b/TODO @@ -12,10 +12,3 @@ * Glance and/or cinder plugins. * Performance - measure and improve it. - -* Implement true parallel request handling. Currently - NBDKIT_THREAD_MODEL_SERIALIZE_REQUESTS and - NBDKIT_THREAD_MODEL_PARALLEL are the same, because we handle - requests within each connection synchronously one at a time. We - could (and should) be able to handle them in parallel by having - another thread pool for requests. diff --git a/docs/nbdkit.pod b/docs/nbdkit.pod index e3043ba..4593391 100644 --- a/docs/nbdkit.pod +++ b/docs/nbdkit.pod @@ -9,7 +9,7 @@ nbdkit - A toolkit for creating NBD servers nbdkit [-e EXPORTNAME] [--exit-with-parent] [-f] [-g GROUP] [-i IPADDR] [--newstyle] [--oldstyle] [-P PIDFILE] [-p PORT] [-r] - [--run CMD] [-s] [--selinux-label LABEL] + [--run CMD] [-s] [--selinux-label LABEL] [-t THREADS] [--tls=off|on|require] [--tls-certificates /path/to/certificates] [--tls-verify-peer] [-U SOCKET] [-u USER] [-v] [-V] @@ -230,6 +230,16 @@ Unix domain sockets: nbdkit --selinux-label system_u:object_r:svirt_t:s0 ... +=item B<-t> THREADS + += item B<--threads> THREADS + +Set the number of threads to be used per connection, which in turn +controls the number of outstanding requests that can be processed at +once. Only matters for plugins with thread_model=parallel (where it +defaults to 16). To force serialized behavior (useful if the client +is not prepared for out-of-order responses), set this to 1. + =item B<--tls=off> =item B<--tls=on> diff --git a/nbdkit.in b/nbdkit.in index 6be89ec..9c3d625 100644 --- a/nbdkit.in +++ b/nbdkit.in @@ -65,7 +65,7 @@ verbose= while [ $# -gt 0 ]; do case "$1" in # Flags that take an argument. We must not rewrite the argument. - -e | --export* | -g | --group | -i | --ip* | -P | --pid* | -p | --port | --run | --selinux-label | --tls | --tls-certificates | -U | --unix | -u | --user) + -e | --export* | -g | --group | -i | --ip* | -P | --pid* | -p | --port | --run | --selinux-label | -t | --threads | --tls | --tls-certificates | -U | --unix | -u | --user) args[$i]="$1" ((++i)) args[$i]="$2" diff --git a/plugins/file/file.c b/plugins/file/file.c index a603be8..ef5da3d 100644 --- a/plugins/file/file.c +++ b/plugins/file/file.c @@ -200,7 +200,7 @@ file_close (void *handle) free (h); } -#define THREAD_MODEL NBDKIT_THREAD_MODEL_SERIALIZE_REQUESTS +#define THREAD_MODEL NBDKIT_THREAD_MODEL_PARALLEL /* Get the file size. */ static int64_t diff --git a/src/connections.c b/src/connections.c index 5257032..2d184b0 100644 --- a/src/connections.c +++ b/src/connections.c @@ -59,6 +59,9 @@ /* Maximum length of any option data (bytes). */ #define MAX_OPTION_LENGTH 4096 +/* Default number of parallel requests. */ +#define DEFAULT_PARALLEL_REQUESTS 16 + /* Connection structure. */ struct connection { pthread_mutex_t request_lock; @@ -203,9 +206,11 @@ _handle_single_connection (int sockin, int sockout) { int r = -1; struct connection *conn; - int nworkers = 1; /* TODO default to 16 for parallel plugins, with command-line override */ + int nworkers = threads ? threads : DEFAULT_PARALLEL_REQUESTS; pthread_t *workers = NULL; + if (!plugin_is_parallel()) + nworkers = 0; conn = new_connection (sockin, sockout, nworkers); if (!conn) goto done; @@ -219,10 +224,9 @@ _handle_single_connection (int sockin, int sockout) if (negotiate_handshake (conn) == -1) goto done; - if (nworkers == 1) { + if (!nworkers) { /* No need for a separate thread. */ debug ("handshake complete, processing requests serially"); - nworkers = conn->nworkers = 0; while (!quit && get_status (conn) > 0) recv_request_send_reply (conn); } diff --git a/src/internal.h b/src/internal.h index 1fc5d69..b79c12c 100644 --- a/src/internal.h +++ b/src/internal.h @@ -103,6 +103,7 @@ extern const char *tls_certificates_dir; extern int tls_verify_peer; extern char *unixsocket; extern int verbose; +extern int threads; extern volatile int quit; @@ -151,6 +152,7 @@ extern void plugin_lock_connection (void); extern void plugin_unlock_connection (void); extern void plugin_lock_request (struct connection *conn); extern void plugin_unlock_request (struct connection *conn); +extern bool plugin_is_parallel (void); extern int plugin_errno_is_preserved (void); extern int plugin_open (struct connection *conn, int readonly); extern void plugin_close (struct connection *conn); diff --git a/src/main.c b/src/main.c index c9f08ab..cc5e9e3 100644 --- a/src/main.c +++ b/src/main.c @@ -84,6 +84,7 @@ int readonly; /* -r */ char *run; /* --run */ int listen_stdin; /* -s */ const char *selinux_label; /* --selinux-label */ +int threads; /* -t */ int tls; /* --tls : 0=off 1=on 2=require */ const char *tls_certificates_dir; /* --tls-certificates */ int tls_verify_peer; /* --tls-verify-peer */ @@ -99,7 +100,7 @@ static char *random_fifo = NULL; enum { HELP_OPTION = CHAR_MAX + 1 }; -static const char *short_options = "e:fg:i:nop:P:rsu:U:vV"; +static const char *short_options = "e:fg:i:nop:P:rst:u:U:vV"; static const struct option long_options[] = { { "help", 0, NULL, HELP_OPTION }, { "dump-config",0, NULL, 0 }, @@ -126,6 +127,7 @@ static const struct option long_options[] = { { "selinux-label", 1, NULL, 0 }, { "single", 0, NULL, 's' }, { "stdin", 0, NULL, 's' }, + { "threads", 1, NULL, 't' }, { "tls", 1, NULL, 0 }, { "tls-certificates", 1, NULL, 0 }, { "tls-verify-peer", 0, NULL, 0 }, @@ -143,7 +145,7 @@ usage (void) " [-e EXPORTNAME] [--exit-with-parent] [-f]\n" " [-g GROUP] [-i IPADDR]\n" " [--newstyle] [--oldstyle] [-P PIDFILE] [-p PORT] [-r]\n" - " [--run CMD] [-s] [--selinux-label LABEL]\n" + " [--run CMD] [-s] [--selinux-label LABEL] [-t THREADS]\n" " [--tls=off|on|require] [--tls-certificates /path/to/certificates]\n" " [--tls-verify-peer]\n" " [-U SOCKET] [-u USER] [-v] [-V]\n" @@ -331,6 +333,20 @@ main (int argc, char *argv[]) listen_stdin = 1; break; + case 't': + { + char *end; + + errno = 0; + threads = strtoul (optarg, &end, 0); + if (errno || *end) { + fprintf (stderr, "%s: cannot parse '%s' into threads\n", + program_name, optarg); + exit (EXIT_FAILURE); + } + /* XXX Worth a maximimum limit on threads? */ + } + case 'U': if (socket_activation) { fprintf (stderr, "%s: cannot use socket activation with -U flag\n", diff --git a/src/plugins.c b/src/plugins.c index e8c6b28..47c4fa5 100644 --- a/src/plugins.c +++ b/src/plugins.c @@ -360,6 +360,14 @@ plugin_unlock_request (struct connection *conn) } } +bool +plugin_is_parallel (void) +{ + assert (dl); + + return plugin._thread_model >= NBDKIT_THREAD_MODEL_PARALLEL; +} + int plugin_errno_is_preserved (void) { -- 2.13.6