--- docs/libnbd.pod | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/docs/libnbd.pod b/docs/libnbd.pod index b909833..f299ef1 100644 --- a/docs/libnbd.pod +++ b/docs/libnbd.pod @@ -228,6 +228,8 @@ For C<nbd_connect_tcp> the third parameter is the port name or number, which can either be a name from F</etc/services> or the port number as a string (eg. C<"10809">). +=head2 Connecting to a subprocess + Some NBD servers — notably L<nbdkit(1)> with the C<-s> parameter — can also accept a single NBD connection on stdin/stdout. You can run these servers as a subprocess of your main program. This example @@ -237,6 +239,23 @@ as the libnbd handle is closed: char *argv[] = { "nbdkit", "-s", "memory", "1G", NULL }; nbd_connect_command (nbd, argv); +=head2 Connecting to a URI + +libnbd supports the NBD URI specification. + +This specification is currently evolving, and discussion about it can +be found on the L<NBD mailing list|https://lists.debian.org/nbd/>;. A +final link to the specification will be added to this documentation +when it is available. + +To connect to a URI via the high level API, use: + + nbd_connect_uri (nbd, "nbd://example.com/"); + +This feature is implemented by calling other libnbd APIs to set up the +export name, TLS parameters, and finally connect over a Unix domain +socket or TCP. + =head1 EXPORTS AND FLAGS It is possible for NBD servers to serve different content on different -- 2.21.0

--- docs/libnbd.pod | 29 ++++++++++++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/docs/libnbd.pod b/docs/libnbd.pod index f299ef1..7cbb9cd 100644 --- a/docs/libnbd.pod +++ b/docs/libnbd.pod @@ -334,7 +334,30 @@ to prefetch. =back -=head1 MULTI-CONN +=head1 PERFORMANCE + +=head2 Issuing multiple in-flight requests + +NBD servers which properly implement the spec can handle multiple +requests in flight over the same connection at the same time. Libnbd +supports this when using the low level API. To use it you simply +issue more requests as needed (eg. using calls like C<nbd_aio_pread>, +C<nbd_aio_pwrite>) without waiting for previous commands to complete. + +Each request is identified by a unique 64 bit handle (assigned by +libnbd), allowing libnbd and callers to match replies to requests. +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. + +There is a full example using multiple in-flight requests available at +L<https://github.com/libguestfs/libnbd/blob/master/examples/threaded-reads-and-writes.c> + +=head2 Multi-conn Some NBD servers advertise “multi-conn” which means that it is safe to make multiple connections to the server and load-balance commands @@ -358,6 +381,10 @@ If multi-conn is supported then you can open further connections: nbd[i]_connect_tcp (nbd[i], "server", "10809"); } +If you are issuing multiple in-flight requests (see above) and +limiting the number, then the limit should be applied to each +individual NBD connection. + =head1 ENCRYPTION AND AUTHENTICATION The NBD protocol and libnbd supports TLS (sometimes incorrectly called -- 2.21.0

Just a simple refactoring in preparation for forthcoming work. --- generator/states-reply.c | 2 +- generator/states.c | 4 ++-- lib/crypto.c | 5 +++-- lib/internal.h | 6 ++++-- lib/socket.c | 5 +++-- 5 files changed, 13 insertions(+), 9 deletions(-) diff --git a/generator/states-reply.c b/generator/states-reply.c index 5be3431..f0ef47c 100644 --- a/generator/states-reply.c +++ b/generator/states-reply.c @@ -36,7 +36,7 @@ h->rbuf = &h->sbuf; h->rlen = sizeof h->sbuf.simple_reply; - r = h->sock->ops->recv (h->sock, h->rbuf, h->rlen); + r = h->sock->ops->recv (h, h->sock, h->rbuf, h->rlen); if (r == -1) { /* This should never happen because when we enter this state we * should have notification that the socket is ready to read. diff --git a/generator/states.c b/generator/states.c index 834fa44..bce4f85 100644 --- a/generator/states.c +++ b/generator/states.c @@ -61,7 +61,7 @@ recv_into_rbuf (struct nbd_handle *h) rlen = h->rlen > sizeof buf ? sizeof buf : h->rlen; } - r = h->sock->ops->recv (h->sock, rbuf, rlen); + r = h->sock->ops->recv (h, h->sock, rbuf, rlen); if (r == -1) { if (errno == EAGAIN || errno == EWOULDBLOCK) return 1; /* more data */ @@ -92,7 +92,7 @@ send_from_wbuf (struct nbd_handle *h) if (h->wlen == 0) return 0; /* move to next state */ - r = h->sock->ops->send (h->sock, h->wbuf, h->wlen); + r = h->sock->ops->send (h, h->sock, h->wbuf, h->wlen); if (r == -1) { if (errno == EAGAIN || errno == EWOULDBLOCK) return 1; /* more data */ diff --git a/lib/crypto.c b/lib/crypto.c index c437788..aba2e27 100644 --- a/lib/crypto.c +++ b/lib/crypto.c @@ -145,7 +145,7 @@ nbd_unlocked_set_tls_psk_file (struct nbd_handle *h, const char *filename) #ifdef HAVE_GNUTLS static ssize_t -tls_recv (struct socket *sock, void *buf, size_t len) +tls_recv (struct nbd_handle *h, struct socket *sock, void *buf, size_t len) { ssize_t r; @@ -163,7 +163,8 @@ tls_recv (struct socket *sock, void *buf, size_t len) } static ssize_t -tls_send (struct socket *sock, const void *buf, size_t len) +tls_send (struct nbd_handle *h, + struct socket *sock, const void *buf, size_t len) { ssize_t r; diff --git a/lib/internal.h b/lib/internal.h index 73cb3f9..c8e5094 100644 --- a/lib/internal.h +++ b/lib/internal.h @@ -188,8 +188,10 @@ struct close_callback { }; struct socket_ops { - ssize_t (*recv) (struct socket *sock, void *buf, size_t len); - ssize_t (*send) (struct socket *sock, const void *buf, size_t len); + ssize_t (*recv) (struct nbd_handle *h, + struct socket *sock, void *buf, size_t len); + ssize_t (*send) (struct nbd_handle *h, + struct socket *sock, const void *buf, size_t len); int (*get_fd) (struct socket *sock); int (*close) (struct socket *sock); }; diff --git a/lib/socket.c b/lib/socket.c index df933be..f48e455 100644 --- a/lib/socket.c +++ b/lib/socket.c @@ -30,7 +30,7 @@ #include "internal.h" static ssize_t -socket_recv (struct socket *sock, void *buf, size_t len) +socket_recv (struct nbd_handle *h, struct socket *sock, void *buf, size_t len) { ssize_t r; @@ -41,7 +41,8 @@ socket_recv (struct socket *sock, void *buf, size_t len) } static ssize_t -socket_send (struct socket *sock, const void *buf, size_t len) +socket_send (struct nbd_handle *h, + struct socket *sock, const void *buf, size_t len) { ssize_t r; -- 2.21.0

On Mon, Jun 03, 2019 at 11:19:15AM -0500, Eric Blake wrote: It would work, but only coincidentally. If we added a different "upper layer" (other than lib/crypto.c) which worked in a slightly different way then there wouldn't the relationship between h and h->sock. Thanks, I've cleaned up 1 & 2 adding your suggestions and will push them in a moment. 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/

On 6/3/19 10:29 AM, Richard W.M. Jones wrote: maybe add 'from different threads' No return value. Comments below at [1] s/there/there is/ Will we ever actually reach a state that is blocked on a write completion for aio_get_direction to ever even request a WRITE or BOTH? Or will the state machine always manage to churn through requests in their entirety and back to state READY which is just a READ state? Is it worth a get_concurrent_writer()? Not strictly necessary, though. [1] So h->writer is NOT allowed to fail directly (if it fails, it must call nbd_concurrent_writer_error instead). Stems from the fact that the generator doesn't allow callbacks with a return type, but livable. But may warrant extra wording in the documentation. -- Eric Blake, Principal Software Engineer Red Hat, Inc. +1-919-301-3226 Virtualization: qemu.org | libvirt.org

--- .gitignore | 1 + examples/Makefile.am | 12 + examples/concurrent-writer.c | 450 +++++++++++++++++++++++++++ examples/threaded-reads-and-writes.c | 2 +- 4 files changed, 464 insertions(+), 1 deletion(-) 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..11a9f22 --- /dev/null +++ b/examples/concurrent-writer.c @@ -0,0 +1,450 @@ +/* Example usage with nbdkit: + * + * nbdkit -U - memory 1M --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 (note this will destroy + * the first megabyte of the remote disk): + * + * ./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). 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. + */ +#define MAX_IN_FLIGHT 16 + +/* Number of commands we issue (per thread). */ +#define NR_CYCLES 1000000 + +/* Reader thread. */ +struct reader_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 reader_threads[NR_MULTI_CONN]; + struct reader_status reader_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 (nbd_can_multi_conn (nbd) == 0) { + fprintf (stderr, "%s: error: " + "this NBD export does not support multi-conn\n", argv[0]); + exit (EXIT_FAILURE); + } + + nbd_close (nbd); + + /* Start the reader threads. */ + for (i = 0; i < NR_MULTI_CONN; ++i) { + reader_status[i].i = i; + reader_status[i].argc = argc; + reader_status[i].argv = argv; + reader_status[i].status = 0; + reader_status[i].requests = 0; + reader_status[i].most_in_flight = 0; + err = pthread_create (&reader_threads[i], NULL, + start_reader_thread, &reader_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 (reader_threads[i], NULL); + if (err != 0) { + errno = err; + perror ("pthread_join"); + exit (EXIT_FAILURE); + } + if (reader_status[i].status != 0) { + fprintf (stderr, "thread %zu failed with status %d\n", + i, reader_status[i].status); + errors++; + } + requests += reader_status[i].requests; + if (reader_status[i].most_in_flight > most_in_flight) + most_in_flight = reader_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 void 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 reader_status *status = arg; + struct writer_data writer_data; + pthread_t writer_thread; + int err; + char buf[512]; + 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; + bool want_to_send; + + 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; + } + + /* 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 = + i > 0 && 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_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 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]); + 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); + + status->status = 0; + pthread_exit (status); + + error: + 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 void +writer (void *data, const void *buf, size_t len) +{ + struct writer_data *writer_data = data; + struct queue *item; + + item = malloc (sizeof *item); + if (!item) goto error; + item->next = NULL; + item->buf = malloc (len); + if (item->buf == NULL) { + free (item); + goto error; + } + 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; + + error: + nbd_concurrent_writer_error (writer_data->nbd, errno); +} + +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); + + 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; +} diff --git a/examples/threaded-reads-and-writes.c b/examples/threaded-reads-and-writes.c index 3e3fc32..a92e7b5 100644 --- a/examples/threaded-reads-and-writes.c +++ b/examples/threaded-reads-and-writes.c @@ -52,7 +52,7 @@ static int64_t exportsize; #define MAX_IN_FLIGHT 16 /* Number of commands we issue (per thread). */ -#define NR_CYCLES 10000 +#define NR_CYCLES 1000000 struct thread_status { size_t i; /* Thread index, 0 .. NR_MULTI_CONN-1 */ -- 2.21.0