Since Eric's improvements to the command queue in commit 6af72b8 (and following) there's now a queue of commands waiting to be issued stored in the handle, and there's no need to issue commands only from the ready state. We can therefore remove the want_to_send logic, queue as many commands as possible, and don't need to test if the socket is ready for POLLOUT. This commit also removes some misleading comments, improves the documentation, and increases the limit on commands in flight (since this limit is effectively in place to stop memory exhaustion, not because of server limits). --- docs/libnbd.pod | 9 ++-- examples/threaded-reads-and-writes.c | 65 +++++++++++----------------- tests/aio-parallel-load.c | 56 ++++++++++-------------- tests/aio-parallel-tls.sh | 2 +- tests/aio-parallel.c | 58 +++++++++++-------------- tests/aio-parallel.sh | 2 +- 6 files changed, 83 insertions(+), 109 deletions(-) diff --git a/docs/libnbd.pod b/docs/libnbd.pod index a5dfb99..ede2539 100644 --- a/docs/libnbd.pod +++ b/docs/libnbd.pod @@ -362,9 +362,12 @@ Replies may arrive out of order. Although in theory you can have an indefinite number of requests in flight at the same time, in practice it's a good idea to limit them to -some number. It is suggested to start with a limit of 16 requests in -flight (per NBD connection), and measure how adjusting the limit up -and down affects performance for your local configuration. +some number. Libnbd will queue commands in the handle even if it +cannot write them to the server, so this limit is largely to prevent a +backlog of commands from consuming too much memory. It is suggested +to start with a limit of 64 requests in flight (per NBD connection), +and measure how adjusting the limit up and down affects performance +for your local configuration. There is a full example using multiple in-flight requests available at L<https://github.com/libguestfs/libnbd/blob/master/examples/threaded-r... diff --git a/examples/threaded-reads-and-writes.c b/examples/threaded-reads-and-writes.c index 5d0d2bd..bb82641 100644 --- a/examples/threaded-reads-and-writes.c +++ b/examples/threaded-reads-and-writes.c @@ -41,20 +41,16 @@ static int64_t exportsize; /* Number of commands that can be "in flight" at the same time on each * connection. (Therefore the total number of requests in flight may - * be up to NR_MULTI_CONN * MAX_IN_FLIGHT). qemu's NBD client can - * have up to 16 requests in flight. - * - * Some servers do not support multiple requests in flight and may - * deadlock or even crash if this is larger than 1, but common NBD - * servers should be OK. + * be up to NR_MULTI_CONN * MAX_IN_FLIGHT). See libnbd(3) section + * "Issuing multiple in-flight requests". */ -#define MAX_IN_FLIGHT 16 +#define MAX_IN_FLIGHT 64 /* The size of reads and writes. */ #define BUFFER_SIZE (1024*1024) /* Number of commands we issue (per thread). */ -#define NR_CYCLES 100000 +#define NR_CYCLES 1000000 struct thread_status { size_t i; /* Thread index, 0 .. NR_MULTI_CONN-1 */ @@ -192,7 +188,6 @@ start_thread (void *arg) uint64_t handles[MAX_IN_FLIGHT]; size_t in_flight; /* counts number of requests in flight */ int dir, r, cmd; - bool want_to_send; buf = malloc (BUFFER_SIZE); if (buf == NULL) { @@ -238,15 +233,31 @@ start_thread (void *arg) goto error; } - /* Do we want to send another request and there's room to issue it - * and the connection is in the READY state so it can be used to - * issue a request. + /* If we want to issue another request, do so. Note that we reuse + * the same buffer for multiple in-flight requests. It doesn't + * matter here because we're just trying to write random stuff, + * but that would be Very Bad in a real application. */ - want_to_send = - i > 0 && in_flight < MAX_IN_FLIGHT && nbd_aio_is_ready (nbd); + while (i > 0 && in_flight < MAX_IN_FLIGHT) { + offset = rand () % (exportsize - sizeof buf); + cmd = rand () & 1; + if (cmd == 0) + handle = nbd_aio_pwrite (nbd, buf, sizeof buf, offset, 0); + else + handle = nbd_aio_pread (nbd, buf, sizeof buf, offset, 0); + if (handle == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + goto error; + } + handles[in_flight] = handle; + i--; + in_flight++; + if (in_flight > status->most_in_flight) + status->most_in_flight = in_flight; + } fds[0].fd = nbd_aio_get_fd (nbd); - fds[0].events = want_to_send ? POLLOUT : 0; + fds[0].events = 0; fds[0].revents = 0; dir = nbd_aio_get_direction (nbd); if ((dir & LIBNBD_AIO_DIRECTION_READ) != 0) @@ -266,30 +277,6 @@ start_thread (void *arg) (fds[0].revents & POLLOUT) != 0) nbd_aio_notify_write (nbd); - /* If we can issue another request, do so. Note that we reuse the - * same buffer for multiple in-flight requests. It doesn't matter - * here because we're just trying to write random stuff, but that - * would be Very Bad in a real application. - */ - if (want_to_send && (fds[0].revents & POLLOUT) != 0 && - nbd_aio_is_ready (nbd)) { - offset = rand () % (exportsize - sizeof buf); - cmd = rand () & 1; - if (cmd == 0) - handle = nbd_aio_pwrite (nbd, buf, sizeof buf, offset, 0); - else - handle = nbd_aio_pread (nbd, buf, sizeof buf, offset, 0); - if (handle == -1) { - fprintf (stderr, "%s\n", nbd_get_error ()); - goto error; - } - handles[in_flight] = handle; - i--; - in_flight++; - if (in_flight > status->most_in_flight) - status->most_in_flight = in_flight; - } - /* If a command is ready to retire, retire it. */ for (j = 0; j < in_flight; ++j) { r = nbd_aio_command_completed (nbd, handles[j]); diff --git a/tests/aio-parallel-load.c b/tests/aio-parallel-load.c index 20fa358..398312f 100644 --- a/tests/aio-parallel-load.c +++ b/tests/aio-parallel-load.c @@ -44,7 +44,7 @@ #define NR_MULTI_CONN 8 /* Number of commands in flight per connection. */ -#define MAX_IN_FLIGHT 16 +#define MAX_IN_FLIGHT 64 /* Unix socket. */ static const char *unixsocket; @@ -169,7 +169,7 @@ start_thread (void *arg) size_t in_flight; /* counts number of requests in flight */ int dir, r, cmd; time_t t; - bool expired = false, want_to_send; + bool expired = false; nbd = nbd_create (); if (nbd == NULL) { @@ -219,15 +219,30 @@ start_thread (void *arg) break; } - /* Do we want to send another request and there's room to issue it - * and the connection is in the READY state so it can be used to - * issue a request. - */ - want_to_send = - !expired && in_flight < MAX_IN_FLIGHT && nbd_aio_is_ready (nbd); + /* If we can issue another request, do so. */ + while (!expired && in_flight < MAX_IN_FLIGHT) { + offset = rand () % (EXPORTSIZE - sizeof buf); + cmd = rand () & 1; + if (cmd == 0) { + handle = nbd_aio_pwrite (nbd, buf, sizeof buf, offset, 0); + status->bytes_sent += sizeof buf; + } + else { + handle = nbd_aio_pread (nbd, buf, sizeof buf, offset, 0); + status->bytes_received += sizeof buf; + } + if (handle == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + goto error; + } + handles[in_flight] = handle; + in_flight++; + if (in_flight > status->most_in_flight) + status->most_in_flight = in_flight; + } fds[0].fd = nbd_aio_get_fd (nbd); - fds[0].events = want_to_send ? POLLOUT : 0; + fds[0].events = 0; fds[0].revents = 0; dir = nbd_aio_get_direction (nbd); if ((dir & LIBNBD_AIO_DIRECTION_READ) != 0) @@ -247,29 +262,6 @@ start_thread (void *arg) (fds[0].revents & POLLOUT) != 0) nbd_aio_notify_write (nbd); - /* If we can issue another request, do so. */ - if (want_to_send && (fds[0].revents & POLLOUT) != 0 && - nbd_aio_is_ready (nbd)) { - offset = rand () % (EXPORTSIZE - sizeof buf); - cmd = rand () & 1; - if (cmd == 0) { - handle = nbd_aio_pwrite (nbd, buf, sizeof buf, offset, 0); - status->bytes_sent += sizeof buf; - } - else { - handle = nbd_aio_pread (nbd, buf, sizeof buf, offset, 0); - status->bytes_received += sizeof buf; - } - if (handle == -1) { - fprintf (stderr, "%s\n", nbd_get_error ()); - goto error; - } - handles[in_flight] = handle; - in_flight++; - if (in_flight > status->most_in_flight) - status->most_in_flight = in_flight; - } - /* If a command is ready to retire, retire it. */ for (i = 0; i < in_flight; ++i) { r = nbd_aio_command_completed (nbd, handles[i]); diff --git a/tests/aio-parallel-tls.sh b/tests/aio-parallel-tls.sh index d60d3fa..c0e5e91 100755 --- a/tests/aio-parallel-tls.sh +++ b/tests/aio-parallel-tls.sh @@ -20,5 +20,5 @@ nbdkit -U - --tls=require --tls-verify-peer --tls-psk=keys.psk \ --filter=cow \ - pattern size=8M \ + pattern size=64M \ --run '$VG ./aio-parallel-tls $unixsocket' diff --git a/tests/aio-parallel.c b/tests/aio-parallel.c index d3217a8..a9b0fd9 100644 --- a/tests/aio-parallel.c +++ b/tests/aio-parallel.c @@ -41,7 +41,7 @@ static char *ramdisk; #define BUFFERSIZE 16384 /* This is also defined in aio-parallel.sh and checked here. */ -#define EXPORTSIZE (8*1024*1024) +#define EXPORTSIZE (64*1024*1024) /* How long (seconds) that the test will run for. */ #define RUN_TIME 10 @@ -50,7 +50,7 @@ static char *ramdisk; #define NR_MULTI_CONN 8 /* Number of commands in flight per connection. */ -#define MAX_IN_FLIGHT 16 +#define MAX_IN_FLIGHT 64 #if BUFFERSIZE >= EXPORTSIZE / NR_MULTI_CONN / MAX_IN_FLIGHT #error "EXPORTSIZE too small" @@ -201,7 +201,7 @@ start_thread (void *arg) size_t in_flight; /* counts number of requests in flight */ int dir, r, cmd; time_t t; - bool expired = false, want_to_send; + bool expired = false; for (i = 0; i < MAX_IN_FLIGHT; ++i) commands[status->i][i].offset = -1; @@ -254,37 +254,8 @@ start_thread (void *arg) break; } - /* Do we want to send another request and there's room to issue it - * and the connection is in the READY state so it can be used to - * issue a request. - */ - want_to_send = - !expired && in_flight < MAX_IN_FLIGHT && nbd_aio_is_ready (nbd); - - fds[0].fd = nbd_aio_get_fd (nbd); - fds[0].events = want_to_send ? POLLOUT : 0; - fds[0].revents = 0; - dir = nbd_aio_get_direction (nbd); - if ((dir & LIBNBD_AIO_DIRECTION_READ) != 0) - fds[0].events |= POLLIN; - if ((dir & LIBNBD_AIO_DIRECTION_WRITE) != 0) - fds[0].events |= POLLOUT; - - if (poll (fds, 1, -1) == -1) { - perror ("poll"); - goto error; - } - - if ((dir & LIBNBD_AIO_DIRECTION_READ) != 0 && - (fds[0].revents & POLLIN) != 0) - nbd_aio_notify_read (nbd); - else if ((dir & LIBNBD_AIO_DIRECTION_WRITE) != 0 && - (fds[0].revents & POLLOUT) != 0) - nbd_aio_notify_write (nbd); - /* If we can issue another request, do so. */ - if (want_to_send && (fds[0].revents & POLLOUT) != 0 && - nbd_aio_is_ready (nbd)) { + while (!expired && in_flight < MAX_IN_FLIGHT) { /* Find a free command slot. */ for (i = 0; i < MAX_IN_FLIGHT; ++i) if (commands[status->i][i].offset == -1) @@ -316,6 +287,27 @@ start_thread (void *arg) status->most_in_flight = in_flight; } + fds[0].fd = nbd_aio_get_fd (nbd); + fds[0].events = 0; + fds[0].revents = 0; + dir = nbd_aio_get_direction (nbd); + if ((dir & LIBNBD_AIO_DIRECTION_READ) != 0) + fds[0].events |= POLLIN; + if ((dir & LIBNBD_AIO_DIRECTION_WRITE) != 0) + fds[0].events |= POLLOUT; + + if (poll (fds, 1, -1) == -1) { + perror ("poll"); + goto error; + } + + if ((dir & LIBNBD_AIO_DIRECTION_READ) != 0 && + (fds[0].revents & POLLIN) != 0) + nbd_aio_notify_read (nbd); + else if ((dir & LIBNBD_AIO_DIRECTION_WRITE) != 0 && + (fds[0].revents & POLLOUT) != 0) + nbd_aio_notify_write (nbd); + /* If a command is ready to retire, retire it. */ for (i = 0; i < MAX_IN_FLIGHT; ++i) { if (commands[status->i][i].offset >= 0) { diff --git a/tests/aio-parallel.sh b/tests/aio-parallel.sh index 40cf794..fd0c5ce 100755 --- a/tests/aio-parallel.sh +++ b/tests/aio-parallel.sh @@ -20,5 +20,5 @@ nbdkit -U - \ --filter=cow \ - pattern size=8M \ + pattern size=64M \ --run '$VG ./aio-parallel $unixsocket' -- 2.21.0

Callback functions now return an int instead of a void. This allows in some cases for the callback to indicate that there was an error. This is a small change to the API: For nbd_set_debug_callback the signature has changed, but the return value is ignored. For nbd_block_status and nbd_aio_block_status the extent function can return an error, which causes the block status command to return an error. Unfortunately this causes us to set the state to dead, although with more effort we could recover from this. Because of this behaviour I didn't document what happens in the error case as we may want to change it in future. For Python and OCaml bindings, raising any kind of exception from the callback is equivalent to returning -1 from the C callback. --- generator/generator | 36 +++++++++++++++++++++-------- generator/states-reply-structured.c | 12 +++++++--- interop/dirty-bitmap.c | 4 +++- lib/debug.c | 3 ++- lib/internal.h | 4 ++-- tests/meta-base-allocation.c | 10 ++++---- 6 files changed, 48 insertions(+), 21 deletions(-) diff --git a/generator/generator b/generator/generator index 4c895e8..ff6075d 100755 --- a/generator/generator +++ b/generator/generator @@ -830,7 +830,7 @@ and arg = written by the function *) | BytesPersistIn of string * string (* same as above, but buffer persists *) | BytesPersistOut of string * string -| Callback of string * arg list (* callback function returning void *) +| Callback of string * arg list (* callback function returning int *) | CallbackPersist of string * arg list (* as above, but callback persists *) | Flags of string (* NBD_CMD_FLAG_* flags *) | Int of string (* small int *) @@ -2680,7 +2680,7 @@ let rec print_c_arg_list ?(handle = false) args = | BytesPersistOut (n, len) -> pr "void *%s, size_t %s" n len | Callback (n, args) | CallbackPersist (n, args) -> - pr "void (*%s) " n; print_c_arg_list args + pr "int (*%s) " n; print_c_arg_list args | Flags n -> pr "uint32_t %s" n | Int n -> pr "int %s" n | Int64 n -> pr "int64_t %s" n @@ -3146,11 +3146,12 @@ let print_python_binding name { args; ret } = pr " PyObject *data;\n"; pr "};\n"; pr "\n"; - pr "static void\n"; + pr "static int\n"; pr "%s_%s_wrapper " name cb_name; print_c_arg_list args; pr "\n"; pr "{\n"; + pr " int ret;\n"; pr " PyGILState_STATE py_save = PyGILState_UNLOCKED;\n"; pr " PyObject *py_args, *py_ret;\n"; List.iter ( @@ -3213,10 +3214,14 @@ let print_python_binding name { args; ret } = pr "\n"; pr " Py_DECREF (py_args);\n"; pr "\n"; - pr " if (py_ret != NULL)\n"; + pr " if (py_ret != NULL) {\n"; + pr " ret = 0;\n"; pr " Py_DECREF (py_ret); /* return value is discarded */\n"; - pr " else\n"; + pr " }\n"; + pr " else {\n"; + pr " ret = -1;\n"; pr " PyErr_PrintEx (0); /* print exception */\n"; + pr " };\n"; pr "\n"; List.iter ( function @@ -3232,6 +3237,7 @@ let print_python_binding name { args; ret } = | Flags _ | Int _ | Int64 _ | Path _ | SockAddrAndLen _ | StringList _ | UInt _ | UInt32 _ -> assert false ) args; + pr " return ret;\n"; pr "}\n"; pr "\n" | _ -> () @@ -3917,7 +3923,7 @@ let print_ocaml_binding (name, { args; ret }) = | UInt _ | UInt32 _ -> assert false ) args in - pr "static void\n"; + pr "static int\n"; pr "%s_%s_wrapper_locked " name cb_name; print_c_arg_list args; pr "\n"; @@ -3957,24 +3963,34 @@ let print_ocaml_binding (name, { args; ret }) = pr " rv = caml_callbackN_exn (fnv, %d, args);\n" (List.length argnames); - pr " if (Is_exception_result (rv))\n"; + pr " if (Is_exception_result (rv)) {\n"; + pr " /* XXX This is not really an error as callbacks can return\n"; + pr " * an error indication. But perhaps we should direct this\n"; + pr " * to a more suitable place or formalize what exception.\n"; + pr " * means error versus unexpected failure.\n"; + pr " */\n"; pr " fprintf (stderr,\n"; pr " \"libnbd: uncaught OCaml exception: %%s\",\n"; pr " caml_format_exception (Extract_exception (rv)));\n"; + pr " CAMLreturnT (int, -1);\n"; + pr " }\n"; pr "\n"; - pr " CAMLreturn0;\n"; + pr " CAMLreturnT (int, 0);\n"; pr "}\n"; pr "\n"; - pr "static void\n"; + pr "static int\n"; pr "%s_%s_wrapper " name cb_name; print_c_arg_list args; pr "\n"; pr "{\n"; + pr " int ret;\n"; + pr "\n"; pr " caml_leave_blocking_section ();\n"; let c_argnames = List.flatten (List.map name_of_arg args) in - pr " %s_%s_wrapper_locked (%s);\n" name cb_name + pr " ret = %s_%s_wrapper_locked (%s);\n" name cb_name (String.concat ", " c_argnames); pr " caml_enter_blocking_section ();\n"; + pr " return ret;\n"; pr "}\n" | _ -> () ) args; diff --git a/generator/states-reply-structured.c b/generator/states-reply-structured.c index e6c1a8a..c835713 100644 --- a/generator/states-reply-structured.c +++ b/generator/states-reply-structured.c @@ -369,10 +369,16 @@ if (context_id == meta_context->context_id) break; - if (meta_context) + if (meta_context) { /* Call the caller's extent function. */ - cmd->extent_fn (cmd->data, meta_context->name, cmd->offset, - &h->bs_entries[1], (length-4) / 4); + if (cmd->extent_fn (cmd->data, meta_context->name, cmd->offset, + &h->bs_entries[1], (length-4) / 4) == -1) { + SET_NEXT_STATE (%.DEAD); /* XXX We should be able to recover. */ + if (errno == 0) errno = EPROTO; + set_error (errno, "extent function failed"); + return -1; + } + } else /* Emit a debug message, but ignore it. */ debug (h, "server sent unexpected meta context ID %" PRIu32, diff --git a/interop/dirty-bitmap.c b/interop/dirty-bitmap.c index b3a89d0..8d34173 100644 --- a/interop/dirty-bitmap.c +++ b/interop/dirty-bitmap.c @@ -32,7 +32,7 @@ static const char *base_allocation = "base:allocation"; static int calls; /* Track which contexts passed through callback */ -static void +static int cb (void *data, const char *metacontext, uint64_t offset, uint32_t *entries, size_t len) { @@ -71,6 +71,8 @@ cb (void *data, const char *metacontext, uint64_t offset, fprintf (stderr, "unexpected context %s\n", metacontext); exit (EXIT_FAILURE); } + + return 0; } int diff --git a/lib/debug.c b/lib/debug.c index 8e2ff76..02d49f7 100644 --- a/lib/debug.c +++ b/lib/debug.c @@ -41,7 +41,7 @@ nbd_unlocked_get_debug (struct nbd_handle *h) int nbd_unlocked_set_debug_callback (struct nbd_handle *h, void *data, - void (*debug_fn) (void *, const char *, const char *)) + int (*debug_fn) (void *, const char *, const char *)) { h->debug_fn = debug_fn; h->debug_data = data; @@ -75,6 +75,7 @@ nbd_internal_debug (struct nbd_handle *h, const char *fs, ...) goto out; if (h->debug_fn) + /* ignore return value */ h->debug_fn (h->debug_data, context, msg); else fprintf (stderr, "libnbd: debug: %s: %s\n", context ? : "unknown", msg); diff --git a/lib/internal.h b/lib/internal.h index c8e5094..c1a57ac 100644 --- a/lib/internal.h +++ b/lib/internal.h @@ -75,7 +75,7 @@ struct nbd_handle { /* For debugging. */ bool debug; void *debug_data; - void (*debug_fn) (void *, const char *, const char *); + int (*debug_fn) (void *, const char *, const char *); /* Linked list of close callbacks. */ struct close_callback *close_callbacks; @@ -212,7 +212,7 @@ struct socket { const struct socket_ops *ops; }; -typedef void (*extent_fn) (void *data, const char *metacontext, uint64_t offset, uint32_t *entries, size_t nr_entries); +typedef int (*extent_fn) (void *data, const char *metacontext, uint64_t offset, uint32_t *entries, size_t nr_entries); struct command_in_flight { struct command_in_flight *next; diff --git a/tests/meta-base-allocation.c b/tests/meta-base-allocation.c index 71f2afc..b00aa50 100644 --- a/tests/meta-base-allocation.c +++ b/tests/meta-base-allocation.c @@ -28,9 +28,9 @@ #include <libnbd.h> -static void check_extent (void *data, const char *metacontext, - uint64_t offset, - uint32_t *entries, size_t nr_entries); +static int check_extent (void *data, const char *metacontext, + uint64_t offset, + uint32_t *entries, size_t nr_entries); int main (int argc, char *argv[]) @@ -125,7 +125,7 @@ main (int argc, char *argv[]) exit (EXIT_SUCCESS); } -static void +static int check_extent (void *data, const char *metacontext, uint64_t offset, uint32_t *entries, size_t nr_entries) @@ -173,4 +173,6 @@ check_extent (void *data, const char *metacontext, else fprintf (stderr, "warning: ignored unexpected meta context %s\n", metacontext); + + return 0; } -- 2.21.0

On Tue, Jun 04, 2019 at 06:35:13AM -0500, Eric Blake wrote: TBH I'd just assume latest version, and I'll try to get out 0.1.3 soon (although not in the next few days). This sounds like the right approach, thanks. I'll push this with the s/\.$// fix you also mentioned below. Rich. -- Richard Jones, Virtualization Group, Red Hat http://people.redhat.com/~rjones Read my programming and virtualization blog: http://rwmj.wordpress.com virt-df lists disk usage of guests without needing to install any software inside the virtual machine. Supports Linux and Windows. http://people.redhat.com/~rjones/virt-df/

--- .gitignore | 1 + examples/Makefile.am | 12 + examples/concurrent-writer.c | 450 +++++++++++++++++++++++++++++++++++ 3 files changed, 463 insertions(+) diff --git a/.gitignore b/.gitignore index 30438c1..e4dad91 100644 --- a/.gitignore +++ b/.gitignore @@ -41,6 +41,7 @@ Makefile.in /docs/libnbd-api.3 /docs/libnbd-api.pod /examples/batched-read-write +/examples/concurrent-writer /examples/threaded-reads-and-writes /examples/simple-fetch-first-sector /examples/simple-reads-and-writes diff --git a/examples/Makefile.am b/examples/Makefile.am index b933873..b5f7e44 100644 --- a/examples/Makefile.am +++ b/examples/Makefile.am @@ -21,6 +21,7 @@ EXTRA_DIST = LICENSE-FOR-EXAMPLES noinst_PROGRAMS = \ batched-read-write \ + concurrent-writer \ simple-fetch-first-sector \ simple-reads-and-writes \ threaded-reads-and-writes @@ -54,6 +55,17 @@ threaded_reads_and_writes_LDADD = \ $(top_builddir)/lib/libnbd.la \ $(PTHREAD_LIBS) +concurrent_writer_SOURCES = \ + concurrent-writer.c +concurrent_writer_CPPFLAGS = \ + -I$(top_srcdir)/include +concurrent_writer_CFLAGS = \ + $(WARNINGS_CFLAGS) \ + $(PTHREAD_CFLAGS) +concurrent_writer_LDADD = \ + $(top_builddir)/lib/libnbd.la \ + $(PTHREAD_LIBS) + batched_read_write_SOURCES = \ batched-read-write.c batched_read_write_CPPFLAGS = \ diff --git a/examples/concurrent-writer.c b/examples/concurrent-writer.c new file mode 100644 index 0000000..06c9720 --- /dev/null +++ b/examples/concurrent-writer.c @@ -0,0 +1,450 @@ +/* Example usage with nbdkit: + * + * nbdkit -U - memory 100M --run './concurrent-writer $unixsocket' + * + * This will read and write randomly over the first megabyte of the + * plugin using multi-conn, multiple threads, multiple requests in + * flight on each connection, and concurrent writer threads. + * + * To run it against a remote server over TCP: + * + * ./concurrent-writer hostname port + * or + * ./concurrent-writer nbd://hostname:port + */ + +#include <stdio.h> +#include <stdlib.h> +#include <stdbool.h> +#include <stdint.h> +#include <string.h> +#include <unistd.h> +#include <errno.h> +#include <signal.h> +#include <poll.h> +#include <time.h> +#include <assert.h> + +#include <pthread.h> + +#include <libnbd.h> + +static int64_t exportsize; + +/* Number of simultaneous connections to the NBD server. The number + * of threads is NR_MULTI_CONN * 2 because there is one thread reading + * plus a concurrent writer thread. Note that some servers only + * support a limited number of simultaneous connections, and/or have a + * configurable thread pool internally, and if you exceed those limits + * then something will break. + */ +#define NR_MULTI_CONN 8 + +/* Number of commands that can be "in flight" at the same time on each + * connection. (Therefore the total number of requests in flight may + * be up to NR_MULTI_CONN * MAX_IN_FLIGHT). See libnbd(3) section + * "Issuing multiple in-flight requests". + */ +#define MAX_IN_FLIGHT 64 + +/* The size of reads and writes. */ +#define BUFFER_SIZE (1024*1024) + +/* Number of commands we issue (per thread). */ +#define NR_CYCLES 1000000 + +/* Reader thread. */ +struct thread_status { + size_t i; /* Thread index, 0 .. NR_MULTI_CONN-1 */ + int argc; /* Command line parameters. */ + char **argv; + int status; /* Return status. */ + unsigned requests; /* Total number of requests made. */ + unsigned most_in_flight; /* Most requests seen in flight. */ +}; + +static void *start_reader_thread (void *arg); + +int +main (int argc, char *argv[]) +{ + struct nbd_handle *nbd; + pthread_t threads[NR_MULTI_CONN]; + struct thread_status status[NR_MULTI_CONN]; + size_t i; + int err; + unsigned requests, most_in_flight, errors; + + srand (time (NULL)); + + if (argc < 2 || argc > 3) { + fprintf (stderr, "%s uri | socket | hostname port\n", argv[0]); + exit (EXIT_FAILURE); + } + + nbd = nbd_create (); + if (nbd == NULL) { + fprintf (stderr, "%s\n", nbd_get_error ()); + exit (EXIT_FAILURE); + } + + /* Connect first to check if the server supports writes and multi-conn. */ + if (argc == 2) { + if (strstr (argv[1], "://")) { + if (nbd_connect_uri (nbd, argv[1]) == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + exit (EXIT_FAILURE); + } + } + else if (nbd_connect_unix (nbd, argv[1]) == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + exit (EXIT_FAILURE); + } + } + else { + if (nbd_connect_tcp (nbd, argv[1], argv[2]) == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + exit (EXIT_FAILURE); + } + } + + exportsize = nbd_get_size (nbd); + if (exportsize == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + exit (EXIT_FAILURE); + } + + if (nbd_read_only (nbd) == 1) { + fprintf (stderr, "%s: error: this NBD export is read-only\n", argv[0]); + exit (EXIT_FAILURE); + } + +#if NR_MULTI_CONN > 1 + if (nbd_can_multi_conn (nbd) == 0) { + fprintf (stderr, "%s: error: " + "this NBD export does not support multi-conn\n", argv[0]); + exit (EXIT_FAILURE); + } +#endif + + nbd_close (nbd); + + /* Start the reader threads. */ + for (i = 0; i < NR_MULTI_CONN; ++i) { + status[i].i = i; + status[i].argc = argc; + status[i].argv = argv; + status[i].status = 0; + status[i].requests = 0; + status[i].most_in_flight = 0; + err = pthread_create (&threads[i], NULL, start_reader_thread, &status[i]); + if (err != 0) { + errno = err; + perror ("pthread_create"); + exit (EXIT_FAILURE); + } + } + + /* Wait for the threads to exit. */ + errors = 0; + requests = 0; + most_in_flight = 0; + for (i = 0; i < NR_MULTI_CONN; ++i) { + err = pthread_join (threads[i], NULL); + if (err != 0) { + errno = err; + perror ("pthread_join"); + exit (EXIT_FAILURE); + } + if (status[i].status != 0) { + fprintf (stderr, "thread %zu failed with status %d\n", + i, status[i].status); + errors++; + } + requests += status[i].requests; + if (status[i].most_in_flight > most_in_flight) + most_in_flight = status[i].most_in_flight; + } + + /* Make sure the number of requests that were required matches what + * we expect. + */ + assert (requests == NR_MULTI_CONN * NR_CYCLES); + + printf ("most requests seen in flight = %u (per thread) " + "vs MAX_IN_FLIGHT = %d\n", + most_in_flight, MAX_IN_FLIGHT); + + exit (errors == 0 ? EXIT_SUCCESS : EXIT_FAILURE); +} + +struct queue { + struct queue *next; + void *buf; + size_t len; +}; + +/* Concurrent writer thread (one per libnbd handle). */ +struct writer_data { + size_t i; /* Thread index, 0 .. NR_MULTI_CONN-1 */ + struct nbd_handle *nbd; /* NBD handle. */ + struct queue *q, *q_end; /* Queue of items to write. */ + pthread_mutex_t q_lock; /* Lock on queue. */ + pthread_cond_t q_cond; /* Condition on queue. */ +}; + +static void *start_writer_thread (void *arg); +static int writer (void *data, const void *buf, size_t len); + +static void * +start_reader_thread (void *arg) +{ + struct nbd_handle *nbd; + struct pollfd fds[1]; + struct thread_status *status = arg; + struct writer_data writer_data; + pthread_t writer_thread; + int err; + char *buf; + size_t i, j; + uint64_t offset, handle; + uint64_t handles[MAX_IN_FLIGHT]; + size_t in_flight; /* counts number of requests in flight */ + int dir, r, cmd; + + buf = malloc (BUFFER_SIZE); + if (buf == NULL) { + perror ("malloc"); + exit (EXIT_FAILURE); + } + + nbd = nbd_create (); + if (nbd == NULL) { + fprintf (stderr, "%s\n", nbd_get_error ()); + exit (EXIT_FAILURE); + } + + if (status->argc == 2) { + if (strstr (status->argv[1], "://")) { + if (nbd_connect_uri (nbd, status->argv[1]) == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + exit (EXIT_FAILURE); + } + } + else if (nbd_connect_unix (nbd, status->argv[1]) == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + exit (EXIT_FAILURE); + } + } + else { + if (nbd_connect_tcp (nbd, status->argv[1], status->argv[2]) == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + exit (EXIT_FAILURE); + } + } + + for (i = 0; i < sizeof buf; ++i) + buf[i] = rand (); + + /* Start the concurrent writer thread, one per handle. */ + writer_data.i = status->i; + writer_data.nbd = nbd; + writer_data.q = writer_data.q_end = NULL; + pthread_mutex_init (&writer_data.q_lock, NULL); + + err = pthread_create (&writer_thread, NULL, + start_writer_thread, &writer_data); + if (err != 0) { + errno = err; + perror ("pthread_create"); + exit (EXIT_FAILURE); + } + + if (nbd_set_concurrent_writer (nbd, &writer_data, writer) == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + exit (EXIT_FAILURE); + } + + /* Issue commands. */ + in_flight = 0; + i = NR_CYCLES; + while (i > 0 || in_flight > 0) { + if (nbd_aio_is_dead (nbd) || nbd_aio_is_closed (nbd)) { + fprintf (stderr, "thread %zu: connection is dead or closed\n", + status->i); + goto error; + } + + /* If we can issue another request, do so. Note that we reuse the + * same buffer for multiple in-flight requests. It doesn't matter + * here because we're just trying to write random stuff, but that + * would be Very Bad in a real application. + */ + while (i > 0 && in_flight < MAX_IN_FLIGHT) { + offset = rand () % (exportsize - sizeof buf); + cmd = rand () & 1; + if (cmd == 0) + handle = nbd_aio_pwrite (nbd, buf, sizeof buf, offset, 0); + else + handle = nbd_aio_pread (nbd, buf, sizeof buf, offset, 0); + if (handle == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + goto error; + } + handles[in_flight] = handle; + i--; + in_flight++; + if (in_flight > status->most_in_flight) + status->most_in_flight = in_flight; + } + + fds[0].fd = nbd_aio_get_fd (nbd); + fds[0].events = 0; + fds[0].revents = 0; + + dir = nbd_aio_get_direction (nbd); + if ((dir & LIBNBD_AIO_DIRECTION_WRITE) != 0) { + /* The concurrent writer is always writable, we don't have to + * test the socket in poll. Since calling nbd_aio_notify_write + * can change the state, after doing it we must restart the + * loop. + */ + nbd_aio_notify_write (nbd); + continue; + } + + if ((dir & LIBNBD_AIO_DIRECTION_READ) != 0) + fds[0].events |= POLLIN; + + if (poll (fds, 1, -1) == -1) { + perror ("poll"); + goto error; + } + + if ((dir & LIBNBD_AIO_DIRECTION_READ) != 0 && + (fds[0].revents & POLLIN) != 0) + nbd_aio_notify_read (nbd); + + /* If a command is ready to retire, retire it. */ + for (j = 0; j < in_flight; ++j) { + r = nbd_aio_command_completed (nbd, handles[j]); + if (r == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + goto error; + } + if (r) { + memmove (&handles[j], &handles[j+1], + sizeof (handles[0]) * (in_flight - j - 1)); + j--; + in_flight--; + status->requests++; + } + } + } + + if (nbd_shutdown (nbd) == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + exit (EXIT_FAILURE); + } + + nbd_close (nbd); + + printf ("thread %zu: finished OK\n", status->i); + + free (buf); + status->status = 0; + pthread_exit (status); + + error: + free (buf); + fprintf (stderr, "thread %zu: failed\n", status->i); + status->status = -1; + pthread_exit (status); +} + +/* This runs in the reader thread and enqueues the data which will be + * picked up by the writer thread. + */ +static int +writer (void *data, const void *buf, size_t len) +{ + struct writer_data *writer_data = data; + struct queue *item; + + item = malloc (sizeof *item); + if (!item) return -1; + item->next = NULL; + item->buf = malloc (len); + if (item->buf == NULL) { + free (item); + return -1; + } + memcpy (item->buf, buf, len); + item->len = len; + + /* Enqueue the item and signal the writer thread. */ + pthread_mutex_lock (&writer_data->q_lock); + if (writer_data->q_end == NULL) + writer_data->q = writer_data->q_end = item; + else { + writer_data->q_end->next = item; + writer_data->q_end = item; + } + pthread_cond_signal (&writer_data->q_cond); + pthread_mutex_unlock (&writer_data->q_lock); + + return 0; +} + +static void * +start_writer_thread (void *arg) +{ + struct writer_data *writer_data = arg; + struct nbd_handle *nbd = writer_data->nbd; + struct queue *item; + int fd; + struct pollfd fds[1]; + ssize_t r; + void *p; + + fd = nbd_aio_get_fd (nbd); + if (fd == -1) { + fprintf (stderr, "%s\n", nbd_get_error ()); + exit (EXIT_FAILURE); + } + + for (;;) { + /* Pick next job off the queue. */ + pthread_mutex_lock (&writer_data->q_lock); + while (writer_data->q == NULL) + pthread_cond_wait (&writer_data->q_cond, &writer_data->q_lock); + item = writer_data->q; + writer_data->q = item->next; + if (writer_data->q == NULL) + writer_data->q_end = NULL; + pthread_mutex_unlock (&writer_data->q_lock); + + p = item->buf; + while (item->len > 0) { + /* Wait for the socket to become ready to write. */ + fds[0].fd = fd; + fds[0].events = POLLOUT; + fds[0].revents = 0; + + if (poll (fds, 1, -1) == -1) goto error; + + r = send (fd, p, item->len, 0); + if (r == -1) goto error; + + p += r; + item->len -= r; + } + + free (item->buf); + free (item); + } + + error: + nbd_concurrent_writer_error (nbd, errno); + return NULL; +} -- 2.21.0