10:34 a.m.
----- Forwarded message -----
Date: Sat, 18 Feb 2017 22:21:19 -0500
Subject: nbdkit async
Hello,
Hope this is the right person to contact regarding nbdkit design.
I have a high latency massively parallel device that I am currently
implementing as an nbdkit plugin in c++ and have run into some design
limitations due to the synchronous callback interface nbdkit requires.
Nbdkit is currently designed to call the plugin
pread/pwrite/trim/flush/zero ops as synchronous calls and expects when the
plugin functions return that it can then send the nbd reply to the socket.
It's parallel thread model is also not implemented as of yet but the
current design still mandates a worker thread per parallel op in progress
due to the synchronous design of the plugin calls.
I would like to modify this to allow for an alternative operating mode
where nbdkit calls the plugin functions and expects the plugin to callback
to nbdkit when a request has completed rather than responding right after
the plugin call returns to nbdkit.
If you are familiar with fuse low level api design, something very similar
to that.
An example flow for a read request would be as follows:
1) nbdkit reads and validates the request from the socket
2) nbdkit calls handle_request but now also passing in the nbd request
handle value
3) nbdkit bundles the nbd request handle value, bool flush_on_update, and
read size into an opaque ptr to struct
4) nbdkit calls my_plugin.pread passing in the usual args + the opaque ptr
5) my_plugin.pread makes an asynchronous read call with a handler set on
completion to call nbdkit_reply_read(conn, opaque ptr, buf) or on error
nbdkit_reply_error(conn, opaque_ptr, error)
6) my_plugin.pread returns back to nbdkit without error after it has
started the async op but before it has completed
7) nbdkit doesn't send a response to the conn->sockout beause when the
async op has completed my_plugin will callback to nbdkit for it to send the
response
8) nbdkit loop continues right away on the next request and it reads and
validates the next request from conn->sockin without waiting for the
previous request to complete
*) Now requires an additional mutex on the conn->sockout for writing
responses
The benefit of this approach is that 2 threads (1 thread for reading
requests from the socket and kicking off requests to the plugin and 1
thread (or more) in a worker pool executing the async handler callbacks)
can service 100s of slow nbd requests in parallel overcoming high latency.
The current design with synchronous callbacks we can provide async in our
plugin layer for pwrites and implement our flush to enforce it but we can't
get around a single slow high latency read op blocking the entire pipe.
I'm willing to work on this in a branch and push this up as opensource but
first wanted to check if this functionality extension is in fact something
redhat would want for nbdkit and if so if there were suggestions to the
implementation.
Initial implementation approach was going to be similar to the
fuse_low_level approach and create an entirely separate header file for the
asynchronous plugin api because the plugin calls now need an additional
parameter (opaque ptr to handle for nbdkit_reply_). This header file
nbdkit_plugin_async.h defines the plugin struct with slightly different
function ptr prototypes that accepts the opaque ptr to nbd request handle
and some additional callback functions nbdkit_reply_error, nbdkit_reply,
and nbdkit_reply_read. The user of this plugin interface is required to
call either nbdkit_reply_error or nbdkit_reply[_read] in each of the
pread/pwrite/flush/trim/zero ops.
If you got this far thank you for the long read and please let me know if
there is any interest.
Thank you,
- Shaun McDowell
----- End forwarded message -----
--
Richard Jones, Virtualization Group, Red Hat http://people.redhat.com/~rjones
Read my programming and virtualization blog: http://rwmj.wordpress.com
virt-builder quickly builds VMs from scratch
http://libguestfs.org/virt-builder.1.html
3:03 a.m.
----- Forwarded message -----
Date: Sat, 18 Feb 2017 22:21:19 -0500
Subject: nbdkit async
Hello,
Hope this is the right person to contact regarding nbdkit design.
I have a high latency massively parallel device that I am currently
implementing as an nbdkit plugin in c++ and have run into some design
limitations due to the synchronous callback interface nbdkit requires.
Is the concern that each client requires a single thread, consuming
memory (eg for stack space), but because of the high latency plugin
these threads will be hanging around not doing very much? And/or is
it that the client is blocked while servicing each request?
Nbdkit is currently designed to call the plugin
pread/pwrite/trim/flush/zero ops as synchronous calls and expects when the
plugin functions return that it can then send the nbd reply to the socket.
It's parallel thread model is also not implemented as of yet
I think this bit can be fixed fairly easily. One change which is
especially easy to make is to send back the NBD_FLAG_CAN_MULTI_CONN
flag (under control of the plugin).
Anyway this doesn't solve your problem ...
but the
current design still mandates a worker thread per parallel op in progress
due to the synchronous design of the plugin calls.
And the synchronous / 1 thread per client design of the server.
I would like to modify this to allow for an alternative operating
mode
where nbdkit calls the plugin functions and expects the plugin to callback
to nbdkit when a request has completed rather than responding right after
the plugin call returns to nbdkit.
If you are familiar with fuse low level api design, something very similar
to that.
An example flow for a read request would be as follows:
1) nbdkit reads and validates the request from the socket
2) nbdkit calls handle_request but now also passing in the nbd request
handle value
3) nbdkit bundles the nbd request handle value, bool flush_on_update, and
read size into an opaque ptr to struct
4) nbdkit calls my_plugin.pread passing in the usual args + the opaque ptr
We can't change the existing API, so this would have to be exposed
through new plugin entry point(s).
5) my_plugin.pread makes an asynchronous read call with a handler set
on
completion to call nbdkit_reply_read(conn, opaque ptr, buf) or on error
nbdkit_reply_error(conn, opaque_ptr, error)
6) my_plugin.pread returns back to nbdkit without error after it has
started the async op but before it has completed
7) nbdkit doesn't send a response to the conn->sockout beause when the
async op has completed my_plugin will callback to nbdkit for it to send the
response
8) nbdkit loop continues right away on the next request and it reads and
validates the next request from conn->sockin without waiting for the
previous request to complete
*) Now requires an additional mutex on the conn->sockout for writing
responses
The benefit of this approach is that 2 threads (1 thread for reading
requests from the socket and kicking off requests to the plugin and 1
thread (or more) in a worker pool executing the async handler callbacks)
can service 100s of slow nbd requests in parallel overcoming high latency.
The current design with synchronous callbacks we can provide async in our
plugin layer for pwrites and implement our flush to enforce it but we can't
get around a single slow high latency read op blocking the entire pipe.
I'm willing to work on this in a branch and push this up as opensource but
first wanted to check if this functionality extension is in fact something
redhat would want for nbdkit and if so if there were suggestions to the
implementation.
It depends on how much it complicates the internals of nbdkit (which
are currently quite simple). Would need to see the patches ...
You can help by splitting into simple changes which are generally
applicable (eg. supporting NBD_FLAG_CAN_MULTI_CONN), and other changes
which are more difficult to integrate.
Initial implementation approach was going to be similar to the
fuse_low_level approach and create an entirely separate header file for the
asynchronous plugin api because the plugin calls now need an additional
parameter (opaque ptr to handle for nbdkit_reply_). This header file
nbdkit_plugin_async.h defines the plugin struct with slightly different
function ptr prototypes that accepts the opaque ptr to nbd request handle
and some additional callback functions nbdkit_reply_error, nbdkit_reply,
and nbdkit_reply_read. The user of this plugin interface is required to
call either nbdkit_reply_error or nbdkit_reply[_read] in each of the
pread/pwrite/flush/trim/zero ops.
If you got this far thank you for the long read and please let me know if
there is any interest.
Rich.
--
Richard Jones, Virtualization Group, Red Hat http://people.redhat.com/~rjones
Read my programming and virtualization blog: http://rwmj.wordpress.com
virt-top is 'top' for virtual machines. Tiny program with many
powerful monitoring features, net stats, disk stats, logging, etc.
http://people.redhat.com/~rjones/virt-top
2:32 p.m.
The concern is a client is blocked while processing a request. The nbdkit
server design requires a thread per request being processed regardless of
the number of connections or clients. We want to run 1000's of requests in
parallel without needing a thread at nbdkit layer per request in flight.
Our plugin layer is built around boost asio and a few threads in a worker
pool running an io service can be processing 1000s of requests in parallel.
(Our plugin is a gateway of sorts and requests are sent back out over the
network. While our plugin waits for the read or write data we don't block
in a thread we handle other requests that are ready).
The current nbdkit server design requires a thread per request in progress
because it is built around a synchronous callback to the plugin layer and
the main recv_request_send_reply loop holds the only copy of the request
handle that is needed to make the reply.
A more flexible design would be the recv_request_send_reply loop is instead
split into a recv_request loop and a send_reply func. The recv_request loop
forwards the request handle to the handle_request call. The existing
nbdkit_plugin struct would function identically and send_reply is called
after the plugin.pread + fua is finished.
An alternative plugin struct "nbdkit_plugin_lowlevel" could define a
different interface where an opaque ptr to the handle + connection + flags
are passed in and the plugin is required to call nbdkit_reply (opaque *ptr,
...) to send the reply to the nbd client rather than nbdkit auto sending
the reply after returning from the plugin function.
Some pseudo code example changes for what I had in mind.
struct operation {
struct connection *conn;
uint64_t handle;
uint32_t cmd;
uint32_t flags;
char *buf;
uint32_t count;
};
// unchanged
struct nbdkit_plugin {
...
int (*pread) (void *handle, void *buf, uint32_t count, uint64_t offset);
int (*pwrite) (void *handle, const void *buf, uint32_t count, uint64_t
offset);
int (*flush) (void *handle);
int (*trim) (void *handle, uint32_t count, uint64_t offset);
int (*zero) (void *handle, uint32_t count, uint64_t offset, int may_trim);
}
// new lowlevel api
struct nbdkit_plugin_lowlevel {
...
int (*pread) (void *op, void *handle, void *buf, uint32_t count, uint64_t
offset);
int (*pwrite) (void *op, void *handle, void *buf, uint32_t count,
uint64_t offset);
int (*flush) (void *op, void *handle);
int (*trim) (void *op, void *handle, uint32_t count, uint64_t offset);
int (*zero) (void *op, void *handle, uint32_t count, uint64_t offset, int
may_trim);
};
// Called by the lowlevel api to send a reply to the client
void
nbdkit_reply(void *vop)
{
int r;
bool flush_after_command;
struct operation *op = (struct operation *) vop;
flush_after_command = (op->flags & NBD_CMD_FLAG_FUA) != 0;
if (!op->conn->can_flush || op->conn->readonly)
flush_after_command = false;
if (flush_after_command) {
op->flags = 0; // clear flags
r = plugin_flush_lowlevel (op, op->conn->handle);
if (r == -1)
// do error stuff;
}
else if (op->cmd == NBD_CMD_READ)
send_reply(op, op->buf, op->count, 0);
else
send_reply(op, NULL, 0, 0);
}
int
my_plugin_pwrite_lowlevel (void *op, void *handle, void *buf,
uint32_t count, uint64_t offset)
{
if (is_readonly(handle)) {
nbdkit_reply_error (op, EROFS);
return 0;
}
if (some critically bad issue)
return -1;
// this returns right away before write has completed and when it
// does complete calls the handler lambda
my_storage.async_write(buf, count, offset,
[op, count](const boost::asio::error_code & ec) {
if (ec)
nbdkit_reply_error(op, ec.value());
else
nbdkit_reply(op);
});
return 0;
}
// connections.c
static int
_send_reply (struct operation *op, uint32_t count, void *buf, uint32_t
error)
{
int r;
struct reply reply;
reply.magic = htobe32 (NBD_REPLY_MAGIC);
reply.handle = op->handle;
reply.error = htobe32 (nbd_errno (error));
if (error != 0) {
/* Since we're about to send only the limited NBD_E* errno to the
* client, don't lose the information about what really happened
* on the server side. Make sure there is a way for the operator
* to retrieve the real error.
*/
debug ("sending error reply: %s", strerror (error));
}
r = xwrite (op->conn->sockout, &reply, sizeof reply);
if (r == -1) {
nbdkit_error ("write reply: %m");
return -1;
}
if (op->cmd == NBD_CMD_READ) { /* Send the read data buffer. */
r = xwrite (op->conn->sockout, buf, count);
if (r == -1) {
nbdkit_error ("write data: %m");
return -1;
}
}
return 0;
}
// new mutex on writes due to parallel nature of responding to the socket
int
send_reply (struct operation *op, uint32_t count, void *buf, uint32_t error)
{
int r;
plugin_lock_reply (op->conn);
r = _send_reply (op, count, buf, error);
plugin_unlock_reply (op->conn);
free (op);
return r;
}
On Mon, Feb 20, 2017 at 4:03 AM, Richard W.M. Jones <rjones(a)redhat.com>
wrote:
> ----- Forwarded message -----
>
> Date: Sat, 18 Feb 2017 22:21:19 -0500
> Subject: nbdkit async
>
> Hello,
>
> Hope this is the right person to contact regarding nbdkit design.
>
> I have a high latency massively parallel device that I am currently
> implementing as an nbdkit plugin in c++ and have run into some design
> limitations due to the synchronous callback interface nbdkit requires.
Is the concern that each client requires a single thread, consuming
memory (eg for stack space), but because of the high latency plugin
these threads will be hanging around not doing very much? And/or is
it that the client is blocked while servicing each request?
> Nbdkit is currently designed to call the plugin
> pread/pwrite/trim/flush/zero ops as synchronous calls and expects when
the
> plugin functions return that it can then send the nbd reply to the
socket.
>
> It's parallel thread model is also not implemented as of yet
I think this bit can be fixed fairly easily. One change which is
especially easy to make is to send back the NBD_FLAG_CAN_MULTI_CONN
flag (under control of the plugin).
Anyway this doesn't solve your problem ...
> but the
> current design still mandates a worker thread per parallel op in progress
> due to the synchronous design of the plugin calls.
And the synchronous / 1 thread per client design of the server.
> I would like to modify this to allow for an alternative operating mode
> where nbdkit calls the plugin functions and expects the plugin to
callback
> to nbdkit when a request has completed rather than responding right after
> the plugin call returns to nbdkit.
>
> If you are familiar with fuse low level api design, something very
similar
> to that.
>
> An example flow for a read request would be as follows:
>
> 1) nbdkit reads and validates the request from the socket
> 2) nbdkit calls handle_request but now also passing in the nbd request
> handle value
> 3) nbdkit bundles the nbd request handle value, bool flush_on_update, and
> read size into an opaque ptr to struct
> 4) nbdkit calls my_plugin.pread passing in the usual args + the opaque
ptr
We can't change the existing API, so this would have to be exposed
through new plugin entry point(s).
> 5) my_plugin.pread makes an asynchronous read call with a handler set on
> completion to call nbdkit_reply_read(conn, opaque ptr, buf) or on error
> nbdkit_reply_error(conn, opaque_ptr, error)
> 6) my_plugin.pread returns back to nbdkit without error after it has
> started the async op but before it has completed
> 7) nbdkit doesn't send a response to the conn->sockout beause when the
> async op has completed my_plugin will callback to nbdkit for it to send
the
> response
> 8) nbdkit loop continues right away on the next request and it reads and
> validates the next request from conn->sockin without waiting for the
> previous request to complete
> *) Now requires an additional mutex on the conn->sockout for writing
> responses
>
> The benefit of this approach is that 2 threads (1 thread for reading
> requests from the socket and kicking off requests to the plugin and 1
> thread (or more) in a worker pool executing the async handler callbacks)
> can service 100s of slow nbd requests in parallel overcoming high
latency.
>
> The current design with synchronous callbacks we can provide async in our
> plugin layer for pwrites and implement our flush to enforce it but we
can't
> get around a single slow high latency read op blocking the entire pipe.
>
> I'm willing to work on this in a branch and push this up as opensource
but
> first wanted to check if this functionality extension is in fact
something
> redhat would want for nbdkit and if so if there were suggestions to the
> implementation.
It depends on how much it complicates the internals of nbdkit (which
are currently quite simple). Would need to see the patches ...
You can help by splitting into simple changes which are generally
applicable (eg. supporting NBD_FLAG_CAN_MULTI_CONN), and other changes
which are more difficult to integrate.
> Initial implementation approach was going to be similar to the
> fuse_low_level approach and create an entirely separate header file for
the
> asynchronous plugin api because the plugin calls now need an additional
> parameter (opaque ptr to handle for nbdkit_reply_). This header file
> nbdkit_plugin_async.h defines the plugin struct with slightly different
> function ptr prototypes that accepts the opaque ptr to nbd request
handle
> and some additional callback functions nbdkit_reply_error, nbdkit_reply,
> and nbdkit_reply_read. The user of this plugin interface is required to
> call either nbdkit_reply_error or nbdkit_reply[_read] in each of the
> pread/pwrite/flush/trim/zero ops.
>
> If you got this far thank you for the long read and please let me know if
> there is any interest.
Rich.
--
Richard Jones, Virtualization Group, Red Hat http://people.redhat.com/~
rjones
Read my programming and virtualization blog: http://rwmj.wordpress.com
virt-top is 'top' for virtual machines. Tiny program with many
powerful monitoring features, net stats, disk stats, logging, etc.
http://people.redhat.com/~rjones/virt-top
5:50 p.m.
I went ahead and implemented 2 new hooks on the existing nbdkit_plugin
struct for async_pread and async_pwrite to get some testable numbers.
The test is set up with an all in memory block device (data being copied to
and from a std vector as simulated reading and writing).
Every read and write op has a 64ms wait before completing (simulated device
latency).
These tests are using nbd-client local to the nbd-server with a unix domain
socket for communication
The device has an ext4 filesystem on it mkfs.ext4 -b 4096 /dev/nbd0
*Baseline test with unmodified nbdkit*
dd if=/dev/zero of=./mnt/zeros bs=128k count=80 conv=fsync
80+0 records in
80+0 records out
10485760 bytes (10 MB, 10 MiB) copied, 5.40797 s, 1.9 MB/s
echo 1 | sudo tee /proc/sys/vm/drop_caches
time cat mnt/zeros > /dev/null
real 0m5.386s
user 0m0.004s
sys 0m0.000s
Read and Write performance are both around 2MB/s which is exactly a single
128k read or write every 64ms.
*With nbdkit modified to use async_pread and async_pwrite and a buffer pool
of 64 buffers (each 128k for a total of 8MB buffer memory)*
dd if=/dev/zero of=./mnt/zeros bs=128k count=8000 conv=fsync
8000+0 records in
8000+0 records out
1048576000 bytes (1.0 GB, 1000 MiB) copied, 8.7736 s, 120 MB/s
echo 1 | sudo tee /proc/sys/vm/drop_caches
time cat mnt/zeros > /dev/null
real 0m8.153s
user 0m0.000s
sys 0m0.320s
Read and Write performance are now 120MB/s (about 64x faster) because we
can process 64 ops in parallel. Our throughput scaled nearly linearly
without needing 64 threads in the nbdkit. Total memory for the buffers is
8MB.
*With async_pread/async_pwrite and a buffer pool of 1024 buffers (128MB
buffer memory) and 2 io service threads (3 threads total, 1 thread in
nbdkit pulling requests off the socket)*
dd if=/dev/zero of=./mnt/zeros bs=128k count=80000 conv=fsync
80000+0 records in
80000+0 records out
10485760000 bytes (10 GB, 9.8 GiB) copied, 5.86029 s, 1.8 GB/s
echo 1 | sudo tee /proc/sys/vm/drop_caches
time cat mnt/zeros > /dev/null
real 0m12.545s
user 0m0.012s
sys 0m2.444s
Read performance was capped at around 825 MiB/s for 1 file sequential read
but when performing 2 files in parallel the read throughput was 1.6GB/s and
for 4 files in parallel 1.9GB/s.
Below is the patch for changes made to nbdkit to support this.
diff --git a/include/nbdkit-plugin.h b/include/nbdkit-plugin.h
index 95cba8d..2e88cad 100644
--- a/include/nbdkit-plugin.h
+++ b/include/nbdkit-plugin.h
@@ -38,15 +38,13 @@
#include <stdarg.h>
#include <stdint.h>
-
-#ifdef __cplusplus
-extern "C" {
-#endif
+#include <stdbool.h>
#define NBDKIT_THREAD_MODEL_SERIALIZE_CONNECTIONS 0
#define NBDKIT_THREAD_MODEL_SERIALIZE_ALL_REQUESTS 1
#define NBDKIT_THREAD_MODEL_SERIALIZE_REQUESTS 2
#define NBDKIT_THREAD_MODEL_PARALLEL 3
+#define NBDKIT_THREAD_MODEL_ASYNC 4
#define NBDKIT_API_VERSION 1
@@ -94,28 +92,35 @@ struct nbdkit_plugin {
int errno_is_preserved;
+ int (*async_pread) (void *conn, uint64_t reqid, bool flush, void
*handle, void *buf, uint32_t count, uint64_t offset);
+ int (*async_pwrite) (void *conn, uint64_t reqid, bool flush, void
*handle, const void *buf, uint32_t count, uint64_t offset);
/* int (*set_exportname) (void *handle, const char *exportname); */
};
-extern void nbdkit_set_error (int err);
-extern void nbdkit_error (const char *msg, ...)
- __attribute__((format (printf, 1, 2)));
-extern void nbdkit_verror (const char *msg, va_list args);
-extern void nbdkit_debug (const char *msg, ...)
- __attribute__((format (printf, 1, 2)));
-extern void nbdkit_vdebug (const char *msg, va_list args);
-
-extern char *nbdkit_absolute_path (const char *path);
-extern int64_t nbdkit_parse_size (const char *str);
-
#ifdef __cplusplus
-#define NBDKIT_CXX_LANG_C extern "C"
+#define NBDKIT_CXX_LANG_C "C"
#else
#define NBDKIT_CXX_LANG_C /* nothing */
#endif
+extern NBDKIT_CXX_LANG_C void nbdkit_set_error (int err);
+extern NBDKIT_CXX_LANG_C void nbdkit_error (const char *msg, ...)
+ __attribute__((format (printf, 1, 2)));
+extern NBDKIT_CXX_LANG_C void nbdkit_verror (const char *msg, va_list
args);
+extern NBDKIT_CXX_LANG_C void nbdkit_debug (const char *msg, ...)
+ __attribute__((format (printf, 1, 2)));
+extern NBDKIT_CXX_LANG_C void nbdkit_vdebug (const char *msg, va_list
args);
+
+extern NBDKIT_CXX_LANG_C char *nbdkit_absolute_path (const char *path);
+extern NBDKIT_CXX_LANG_C int64_t nbdkit_parse_size (const char *str);
+
+extern NBDKIT_CXX_LANG_C int nbdkit_async_reply (void *conn, uint64_t
reqid);
+extern NBDKIT_CXX_LANG_C int nbdkit_async_reply_read (void *conn, uint64_t
reqid, uint32_t count, void *buf);
+extern NBDKIT_CXX_LANG_C int nbdkit_async_reply_error (void *conn,
uint64_t reqid);
+
+
#define NBDKIT_REGISTER_PLUGIN(plugin) \
- NBDKIT_CXX_LANG_C \
+ extern NBDKIT_CXX_LANG_C \
struct nbdkit_plugin * \
plugin_init (void) \
{ \
@@ -125,8 +130,4 @@ extern int64_t nbdkit_parse_size (const char *str);
return &(plugin); \
}
-#ifdef __cplusplus
-}
-#endif
-
#endif /* NBDKIT_PLUGIN_H */
diff --git a/src/connections.c b/src/connections.c
index a0d689a..e03a0f6 100644
--- a/src/connections.c
+++ b/src/connections.c
@@ -62,7 +62,8 @@
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 int recv_request (struct connection *conn);
+static int send_reply (struct connection *conn, uint64_t handle, uint32_t
count, void *buf, uint32_t error);
static int
_handle_single_connection (int sockin, int sockout)
@@ -86,7 +87,7 @@ _handle_single_connection (int sockin, int sockout)
* a thread pool.
*/
while (!quit) {
- r = recv_request_send_reply (conn);
+ r = recv_request (conn);
if (r == -1)
goto err;
if (r == 0)
@@ -127,6 +128,7 @@ new_connection (int sockin, int sockout)
conn->sockin = sockin;
conn->sockout = sockout;
pthread_mutex_init (&conn->request_lock, NULL);
+ pthread_mutex_init (&conn->reply_lock, NULL);
return conn;
}
@@ -143,6 +145,7 @@ free_connection (struct connection *conn)
close (conn->sockout);
pthread_mutex_destroy (&conn->request_lock);
+ pthread_mutex_destroy (&conn->reply_lock);
if (conn->handle)
plugin_close (conn);
@@ -626,13 +629,13 @@ get_error (struct connection *conn)
* On read/write errors, sets *error appropriately and returns 0.
*/
static int
-_handle_request (struct connection *conn,
+_handle_request (struct connection *conn, uint64_t handle,
uint32_t cmd, uint32_t flags, uint64_t offset, uint32_t
count,
- void *buf,
- uint32_t *error)
+ void *buf)
{
- bool flush_after_command;
int r;
+ uint32_t error = 0;
+ bool flush_after_command;
/* Flush after command performed? */
flush_after_command = (flags & NBD_CMD_FLAG_FUA) != 0;
@@ -645,42 +648,54 @@ _handle_request (struct connection *conn,
switch (cmd) {
case NBD_CMD_READ:
- r = plugin_pread (conn, buf, count, offset);
+ if (plugin_can_async_read (conn)) {
+ r = plugin_async_pread (conn, handle, flush_after_command, buf,
count, offset);
+ if (r == 0)
+ return 0; // plugin now has responsibility of sending response
+ }
+ else
+ r = plugin_pread (conn, buf, count, offset);
if (r == -1) {
- *error = get_error (conn);
- return 0;
+ error = get_error (conn);
+ return send_reply (conn, handle, 0, NULL, error);
}
break;
case NBD_CMD_WRITE:
- r = plugin_pwrite (conn, buf, count, offset);
+ if (plugin_can_async_write (conn)) {
+ r = plugin_async_pwrite (conn, handle, flush_after_command, buf,
count, offset);
+ if (r == 0)
+ return 0; // plugin now has responsibility of sending response
+ }
+ else
+ r = plugin_pwrite (conn, buf, count, offset);
if (r == -1) {
- *error = get_error (conn);
- return 0;
+ error = get_error (conn);
+ return send_reply (conn, handle, 0, NULL, error);
}
break;
case NBD_CMD_FLUSH:
r = plugin_flush (conn);
if (r == -1) {
- *error = get_error (conn);
- return 0;
+ error = get_error (conn);
+ return send_reply (conn, handle, 0, NULL, error);
}
break;
case NBD_CMD_TRIM:
r = plugin_trim (conn, count, offset);
if (r == -1) {
- *error = get_error (conn);
- return 0;
+ error = get_error (conn);
+ return send_reply (conn, handle, 0, NULL, error);
}
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;
+ error = get_error (conn);
+ return send_reply (conn, handle, 0, NULL, error);
}
break;
@@ -691,24 +706,28 @@ _handle_request (struct connection *conn,
if (flush_after_command) {
r = plugin_flush (conn);
if (r == -1) {
- *error = get_error (conn);
- return 0;
+ error = get_error (conn);
+ return send_reply (conn, handle, 0, NULL, error);
}
}
- return 0;
+ if (cmd == NBD_CMD_READ)
+ r = send_reply (conn, handle, count, buf, error);
+ else
+ r = send_reply (conn, handle, 0, NULL, error);
+
+ return r;
}
static int
-handle_request (struct connection *conn,
+handle_request (struct connection *conn, uint64_t handle,
uint32_t cmd, uint32_t flags, uint64_t offset, uint32_t
count,
- void *buf,
- uint32_t *error)
+ void *buf)
{
int r;
plugin_lock_request (conn);
- r = _handle_request (conn, cmd, flags, offset, count, buf, error);
+ r = _handle_request (conn, handle, cmd, flags, offset, count, buf);
plugin_unlock_request (conn);
return r;
@@ -763,11 +782,10 @@ nbd_errno (int error)
}
static int
-recv_request_send_reply (struct connection *conn)
+recv_request (struct connection *conn)
{
int r;
struct request request;
- struct reply reply;
uint32_t magic, cmd, flags, count, error = 0;
uint64_t offset;
CLEANUP_FREE char *buf = NULL;
@@ -808,7 +826,10 @@ recv_request_send_reply (struct connection *conn)
if (r == 0) { /* request not valid */
if (cmd == NBD_CMD_WRITE)
skip_over_write_buffer (conn->sockin, count);
- goto send_reply;
+ r = send_reply (conn, request.handle, 0, NULL, error);
+ if (r == -1)
+ return -1;
+ return 1;
}
/* Allocate the data buffer used for either read or write requests. */
@@ -819,7 +840,9 @@ recv_request_send_reply (struct connection *conn)
error = ENOMEM;
if (cmd == NBD_CMD_WRITE)
skip_over_write_buffer (conn->sockin, count);
- goto send_reply;
+ r = send_reply (conn, request.handle, 0, NULL, error);
+ if (r == -1)
+ return -1;
}
}
@@ -837,14 +860,20 @@ 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);
+ r = handle_request (conn, request.handle, cmd, flags, offset, count,
buf);
if (r == -1)
return -1;
- /* Send the reply packet. */
- send_reply:
+ return 1; /* command processed ok */
+}
+
+static int
+_send_reply (struct connection *conn, uint64_t handle, uint32_t count,
void *buf, uint32_t error)
+{
+ int r;
+ struct reply reply;
reply.magic = htobe32 (NBD_REPLY_MAGIC);
- reply.handle = request.handle;
+ reply.handle = handle;
reply.error = htobe32 (nbd_errno (error));
if (error != 0) {
@@ -862,8 +891,7 @@ recv_request_send_reply (struct connection *conn)
return -1;
}
- /* Send the read data buffer. */
- if (cmd == NBD_CMD_READ) {
+ if (error == 0 && buf != NULL) { /* Send the read data buffer. */
r = xwrite (conn->sockout, buf, count);
if (r == -1) {
nbdkit_error ("write data: %m");
@@ -871,5 +899,37 @@ recv_request_send_reply (struct connection *conn)
}
}
- return 1; /* command processed ok */
+ return 0;
+}
+
+static int
+send_reply (struct connection *conn, uint64_t handle, uint32_t count, void
*buf, uint32_t error)
+{
+ int r;
+
+ plugin_lock_reply (conn);
+ r = _send_reply (conn, handle, count, buf, error);
+ plugin_unlock_reply (conn);
+
+ return r;
+}
+
+int
+nbdkit_async_reply (void *conn, uint64_t reqid)
+{
+ return send_reply (conn, reqid, 0, NULL, 0);
+}
+
+int
+nbdkit_async_reply_read (void *conn, uint64_t reqid, uint32_t count, void
*buf)
+{
+
+ return send_reply (conn, reqid, count, buf, 0);
+}
+
+int
+nbdkit_async_reply_error (void *conn, uint64_t reqid)
+{
+ uint32_t error = get_error (conn);
+ return send_reply (conn, reqid, 0, NULL, error);
}
diff --git a/src/internal.h b/src/internal.h
index e73edf1..93d32e9 100644
--- a/src/internal.h
+++ b/src/internal.h
@@ -114,6 +114,7 @@ extern void cleanup_free (void *ptr);
struct connection {
int sockin, sockout;
pthread_mutex_t request_lock;
+ pthread_mutex_t reply_lock;
void *handle;
uint64_t exportsize;
int readonly;
@@ -140,6 +141,8 @@ 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 void plugin_lock_reply (struct connection *conn);
+extern void plugin_unlock_reply (struct connection *conn);
extern int plugin_errno_is_preserved (void);
extern int plugin_open (struct connection *conn, int readonly);
extern void plugin_close (struct connection *conn);
@@ -148,8 +151,12 @@ extern int plugin_can_write (struct connection *conn);
extern int plugin_can_flush (struct connection *conn);
extern int plugin_is_rotational (struct connection *conn);
extern int plugin_can_trim (struct connection *conn);
+extern int plugin_can_async_read (struct connection *conn);
extern int plugin_pread (struct connection *conn, void *buf, uint32_t
count, uint64_t offset);
+extern int plugin_async_pread (struct connection *conn, uint64_t handle,
bool flush, void *buf, uint32_t count, uint64_t offset);
+extern int plugin_can_async_write (struct connection *conn);
extern int plugin_pwrite (struct connection *conn, void *buf, uint32_t
count, uint64_t offset);
+extern int plugin_async_pwrite (struct connection *conn, uint64_t handle,
bool flush, void *buf, uint32_t count, uint64_t offset);
extern int plugin_flush (struct connection *conn);
extern int plugin_trim (struct connection *conn, uint32_t count, uint64_t
offset);
extern int plugin_zero (struct connection *conn, uint32_t count, uint64_t
offset, int may_trim);
diff --git a/src/plugins.c b/src/plugins.c
index eeed8a9..9da30db 100644
--- a/src/plugins.c
+++ b/src/plugins.c
@@ -121,8 +121,9 @@ plugin_register (const char *_filename,
program_name, filename);
exit (EXIT_FAILURE);
}
- if (plugin.pread == NULL) {
- fprintf (stderr, "%s: %s: plugin must have a .pread callback\n",
+ if (plugin.pread == NULL &&
+ (plugin._thread_model != NBDKIT_THREAD_MODEL_ASYNC ||
plugin.async_pread == NULL)) {
+ fprintf (stderr, "%s: %s: plugin must have either a .pread or
.async_pread callback\n",
program_name, filename);
exit (EXIT_FAILURE);
}
@@ -231,6 +232,9 @@ plugin_dump_fields (void)
case NBDKIT_THREAD_MODEL_PARALLEL:
printf ("parallel");
break;
+ case NBDKIT_THREAD_MODEL_ASYNC:
+ printf ("async");
+ break;
default:
printf ("%d # unknown thread model!", plugin._thread_model);
break;
@@ -258,6 +262,8 @@ plugin_dump_fields (void)
HAS (flush);
HAS (trim);
HAS (zero);
+ HAS (async_pread);
+ HAS (async_pwrite);
#undef HAS
}
@@ -350,6 +356,28 @@ plugin_unlock_request (struct connection *conn)
}
}
+void
+plugin_lock_reply (struct connection *conn)
+{
+ assert (dl);
+
+ if (plugin._thread_model >= NBDKIT_THREAD_MODEL_PARALLEL) {
+ debug ("acquire per-connection reply lock");
+ pthread_mutex_lock (&conn->reply_lock);
+ }
+}
+
+void
+plugin_unlock_reply (struct connection *conn)
+{
+ assert (dl);
+
+ if (plugin._thread_model >= NBDKIT_THREAD_MODEL_PARALLEL) {
+ debug ("release per-connection reply lock");
+ pthread_mutex_unlock (&conn->reply_lock);
+ }
+}
+
int
plugin_errno_is_preserved (void)
{
@@ -414,7 +442,8 @@ plugin_can_write (struct connection *conn)
if (plugin.can_write)
return plugin.can_write (conn->handle);
else
- return plugin.pwrite != NULL;
+ return plugin.pwrite != NULL ||
+ (plugin._thread_model == NBDKIT_THREAD_MODEL_ASYNC &&
plugin.async_pwrite != NULL);
}
int
@@ -460,6 +489,16 @@ plugin_can_trim (struct connection *conn)
}
int
+plugin_can_async_read (struct connection *conn)
+{
+ assert (dl);
+ assert (conn->handle);
+
+ return ((plugin._thread_model == NBDKIT_THREAD_MODEL_ASYNC) &&
+ (plugin.async_pread != NULL));
+}
+
+int
plugin_pread (struct connection *conn,
void *buf, uint32_t count, uint64_t offset)
{
@@ -473,6 +512,27 @@ plugin_pread (struct connection *conn,
}
int
+plugin_async_pread (struct connection *conn, uint64_t handle, bool flush,
+ void *buf, uint32_t count, uint64_t offset)
+{
+ assert (dl);
+ assert (conn->handle);
+
+ debug ("async_pread count=%" PRIu32 " offset=%" PRIu64, count,
offset);
+
+ return plugin.async_pread (conn, handle, flush, conn->handle, buf,
count, offset);
+}
+
+int
+plugin_can_async_write (struct connection *conn)
+{
+ assert (dl);
+ assert (conn->handle);
+
+ return (plugin._thread_model == NBDKIT_THREAD_MODEL_ASYNC) &&
(plugin.async_pwrite != NULL);
+}
+
+int
plugin_pwrite (struct connection *conn,
void *buf, uint32_t count, uint64_t offset)
{
@@ -490,23 +550,44 @@ plugin_pwrite (struct connection *conn,
}
int
+plugin_async_pwrite (struct connection *conn, uint64_t handle, bool flush,
+ void *buf, uint32_t count, uint64_t offset)
+{
+ assert (dl);
+ assert (conn->handle);
+
+ debug ("async_pwrite count=%" PRIu32 " offset=%" PRIu64, count,
offset);
+
+ if (plugin.async_pwrite != NULL)
+ return plugin.async_pwrite (conn, handle, flush, conn->handle, buf,
count, offset);
+ else {
+ errno = EROFS;
+ return -1;
+ }
+}
+
+int
plugin_flush (struct connection *conn)
{
+ int r;
assert (dl);
assert (conn->handle);
debug ("flush");
if (plugin.flush != NULL)
- return plugin.flush (conn->handle);
+ r = plugin.flush (conn->handle);
else {
errno = EINVAL;
- return -1;
+ r = -1;
}
+
+ return r;
}
int
-plugin_trim (struct connection *conn, uint32_t count, uint64_t offset)
+plugin_trim (struct connection *conn,
+ uint32_t count, uint64_t offset)
{
assert (dl);
assert (conn->handle);
2831
days inactive
2833
days old
3 comments
2 participants
participants (2)
-
Richard W.M. Jones -
Shaun McDowell