search for: nworkers

I want to make my nbd forwarding plugin fully parallel - but to do that, I first need to make nbdkit itself fully parallel ;) With this series, I was finally able to demonstrate out-of-order responses when using qemu-io (which is great at sending back-to-back requests prior to waiting for responses) coupled with the nbd file plugin (which has a great feature of rdelay and wdelay, to make it

...------- 2 files changed, 38 insertions(+), 24 deletions(-) diff --git a/server/connections.c b/server/connections.c index a30a541..b7d9a6a 100644 --- a/server/connections.c +++ b/server/connections.c @@ -91,11 +91,13 @@ connection_get_status (struct connection *conn) { int r; - if (conn->nworkers) - pthread_mutex_lock (&conn->status_lock); + if (conn->nworkers && + pthread_mutex_lock (&conn->status_lock)) + abort (); r = conn->status; - if (conn->nworkers) - pthread_mutex_unlock (&conn->status_lock); + if (conn->nworkers &&amp...

I've posted some of these patches or ideas before; but now I'm confident enough with the series that it should be ready to push; at any rate, I can now run test-socket-activation in a tight loop without triggering any crashes or hangs. With this in place, I'm going back to work on making the nbd forwarder wort with the parallel thread model. Eric Blake (8): sockets: Use

...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) go...

...ify pipe-to-self: %m"); + debug ("failed to notify pipe-to-self: %m"); } conn->status = value; } @@ -176,7 +176,7 @@ _handle_single_connection (int sockin, int sockout) debug ("handshake complete, processing requests with %d threads", nworkers); workers = calloc (nworkers, sizeof *workers); - if (!workers) { + if (unlikely (!workers)) { perror ("malloc"); goto done; } @@ -185,12 +185,13 @@ _handle_single_connection (int sockin, int sockout) struct worker_data *worker = malloc (sizeof *worker...

...crypto.c \ errors.c \ internal.h \ + locks.c \ main.c \ plugins.c \ protocol.h \ diff --git a/src/connections.c b/src/connections.c index 111a810..74bb8e4 100644 --- a/src/connections.c +++ b/src/connections.c @@ -211,7 +211,7 @@ _handle_single_connection (int sockin, int sockout) int nworkers = threads ? threads : DEFAULT_PARALLEL_REQUESTS; pthread_t *workers = NULL; - if (!plugin_is_parallel() || nworkers == 1) + if (plugin_thread_model () < NBDKIT_THREAD_MODEL_PARALLEL || nworkers == 1) nworkers = 0; conn = new_connection (sockin, sockout, nworkers); if (!conn) @@...

...crypto.c \ errors.c \ internal.h \ + locks.c \ main.c \ plugins.c \ protocol.h \ diff --git a/src/connections.c b/src/connections.c index 111a810..74bb8e4 100644 --- a/src/connections.c +++ b/src/connections.c @@ -211,7 +211,7 @@ _handle_single_connection (int sockin, int sockout) int nworkers = threads ? threads : DEFAULT_PARALLEL_REQUESTS; pthread_t *workers = NULL; - if (!plugin_is_parallel() || nworkers == 1) + if (plugin_thread_model () < NBDKIT_THREAD_MODEL_PARALLEL || nworkers == 1) nworkers = 0; conn = new_connection (sockin, sockout, nworkers); if (!conn) @@...

We noticed while writing various libnbd tests that when the delay filter is in use, there are scenarios where we had to resort to SIGKILL to get rid of nbdkit, because it was non-responsive to SIGINT. I'm still trying to figure out the best way to add testsuite coverage of this, but already proved to myself that it works from the command line, under two scenarios that both used to cause long

...rver/internal.h | 17 +++++++++++++++++ 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/server/connections.c b/server/connections.c index 819f7b8..ec28815 100644 --- a/server/connections.c +++ b/server/connections.c @@ -269,9 +269,8 @@ new_connection (int sockin, int sockout, int nworkers) goto error; } conn->nr_handles = backend->i + 1; - memset (conn->handles, -1, conn->nr_handles * sizeof *conn->handles); for_each_backend (b) - conn->handles[b->i].handle = NULL; + reset_b_conn_handle (&conn->handles[b->i]); conn->status...

I do have a question about whether patch 2 is right, or whether I've exposed a bigger problem in the truncate (and possibly other) filter, but the rest seem fairly straightforward. Eric Blake (4): server: Check for pthread lock failures truncate: Factor out reading real_size under mutex plugins: Check for mutex failures filters: Check for mutex failures filters/cache/cache.c

...-1) + h->exportsize = b->get_size (b, conn); + return h->exportsize; } int diff --git a/server/connections.c b/server/connections.c index 1c501002..2a5150cd 100644 --- a/server/connections.c +++ b/server/connections.c @@ -260,6 +260,7 @@ new_connection (int sockin, int sockout, int nworkers) struct connection *conn; int opt; socklen_t optlen = sizeof opt; + struct backend *b; conn = calloc (1, sizeof *conn); if (conn == NULL) { @@ -273,6 +274,8 @@ new_connection (int sockin, int sockout, int nworkers) return NULL; } conn->nr_handles = backend->i + 1;...

...t exportsize; int can_write; @@ -195,7 +203,7 @@ struct b_conn_handle { }; static inline void -reset_b_conn_handle (struct b_conn_handle *h) +reset_handle (struct handle *h) { h->handle = NULL; h->state = 0; @@ -222,7 +230,7 @@ struct connection { void *crypto_session; int nworkers; - struct b_conn_handle *handles; + struct handle *handles; /* One per plugin and filter. */ size_t nr_handles; char exportname[NBD_MAX_STRING + 1]; @@ -239,6 +247,12 @@ struct connection { connection_close_function close; }; +static inline struct handle * +get_handle (struc...

v1 -> v2: - Fixed everything mentioned in the review. Rich.

Somewhat invasive but mostly mechanical change to how plugins are called. This patch is in preparation for adding a second backend subtype for filters. Rich.

...;plugin_name (backend)); /* Handshake. */ if (negotiate_handshake (conn) == -1) @@ -253,7 +253,7 @@ _handle_single_connection (int sockin, int sockout) goto wait; } if (asprintf (&worker->name, - "%s.%d", backend->name (backend), nworkers) < 0) { + "%s.%d", backend->plugin_name (backend), nworkers) < 0) { perror ("asprintf"); set_status (conn, -1); free (worker); diff --git a/src/internal.h b/src/internal.h index c4ee51b..dc26665 100644 --- a/src/internal.h +...

...++++++++++------------------------ 5 files changed, 360 insertions(+), 291 deletions(-) diff --git a/src/connections.c b/src/connections.c index 74bb8e4..921a5b2 100644 --- a/src/connections.c +++ b/src/connections.c @@ -211,16 +211,17 @@ _handle_single_connection (int sockin, int sockout) int nworkers = threads ? threads : DEFAULT_PARALLEL_REQUESTS; pthread_t *workers = NULL; - if (plugin_thread_model () < NBDKIT_THREAD_MODEL_PARALLEL || nworkers == 1) + if (backend->thread_model (backend) < NBDKIT_THREAD_MODEL_PARALLEL || + nworkers == 1) nworkers = 0; conn = new_co...

...++++++++++------------------------ 5 files changed, 368 insertions(+), 295 deletions(-) diff --git a/src/connections.c b/src/connections.c index 74bb8e4..921a5b2 100644 --- a/src/connections.c +++ b/src/connections.c @@ -211,16 +211,17 @@ _handle_single_connection (int sockin, int sockout) int nworkers = threads ? threads : DEFAULT_PARALLEL_REQUESTS; pthread_t *workers = NULL; - if (plugin_thread_model () < NBDKIT_THREAD_MODEL_PARALLEL || nworkers == 1) + if (backend->thread_model (backend) < NBDKIT_THREAD_MODEL_PARALLEL || + nworkers == 1) nworkers = 0; conn = new_co...

...++++++++++------------------------ 5 files changed, 367 insertions(+), 294 deletions(-) diff --git a/src/connections.c b/src/connections.c index 74bb8e4..921a5b2 100644 --- a/src/connections.c +++ b/src/connections.c @@ -211,16 +211,17 @@ _handle_single_connection (int sockin, int sockout) int nworkers = threads ? threads : DEFAULT_PARALLEL_REQUESTS; pthread_t *workers = NULL; - if (plugin_thread_model () < NBDKIT_THREAD_MODEL_PARALLEL || nworkers == 1) + if (backend->thread_model (backend) < NBDKIT_THREAD_MODEL_PARALLEL || + nworkers == 1) nworkers = 0; conn = new_co...

...r/connections.c +++ b/server/connections.c @@ -146,23 +146,23 @@ handle_single_connection (int sockin, int sockout) lock_connection (); - if (backend->thread_model (backend) < NBDKIT_THREAD_MODEL_PARALLEL || + if (top->thread_model (top) < NBDKIT_THREAD_MODEL_PARALLEL || nworkers == 1) nworkers = 0; conn = new_connection (sockin, sockout, nworkers); if (!conn) goto done; - /* NB: because of an asynchronous exit backend can be set to NULL at + /* NB: because of an asynchronous exit top can be set to NULL at * just about any time. */ - if (backend...

This fixes the long-standing crash on close when nbdkit exits. I did try first to fix threads so we're using a proper thread pool, but that's difficult to implement. So this does the minimal change needed to fix the crash instead. There are still two segfaults that happen during running the test suite. One is deliberately caused (tests/test-captive.sh). The other appears to be an