[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[Qemu-devel] [PATCH v2 2/2] nbd: Coroutine based nbd_send_negotiate
From: |
Fam Zheng |
Subject: |
[Qemu-devel] [PATCH v2 2/2] nbd: Coroutine based nbd_send_negotiate |
Date: |
Mon, 11 Jan 2016 11:36:13 +0800 |
Create a coroutine in nbd_client_new, so that nbd_send_negotiate doesn't
need qemu_set_block().
A handler is needed for csock fd in case the coroutine yields during
I/O.
With this, if the other end disappears in the middle of the negotiation,
we don't block the whole event loop.
Signed-off-by: Fam Zheng <address@hidden>
---
nbd.c | 68 +++++++++++++++++++++++++++++++++++++++++++++++++------------------
1 file changed, 50 insertions(+), 18 deletions(-)
diff --git a/nbd.c b/nbd.c
index f8d0221..8f288a0 100644
--- a/nbd.c
+++ b/nbd.c
@@ -238,10 +238,9 @@ ssize_t nbd_wr_sync(int fd, void *buffer, size_t size,
bool do_read)
static ssize_t read_sync(int fd, void *buffer, size_t size)
{
- /* Sockets are kept in blocking mode in the negotiation phase. After
- * that, a non-readable socket simply means that another thread stole
- * our request/reply. Synchronization is done with recv_coroutine, so
- * that this is coroutine-safe.
+ /* A non-readable socket simply means that another thread stole our
+ * request/reply. Synchronization is done with recv_coroutine, so that
+ * this is coroutine-safe.
*/
return nbd_wr_sync(fd, buffer, size, true);
}
@@ -504,13 +503,25 @@ static int nbd_receive_options(NBDClient *client)
}
}
-static int nbd_send_negotiate(NBDClient *client)
+typedef struct {
+ NBDClient *client;
+ Coroutine *co;
+} NBDClientNewData;
+
+static void nbd_negotiate_continue(void *opaque)
{
+ qemu_coroutine_enter(opaque, NULL);
+}
+
+static coroutine_fn int nbd_send_negotiate(NBDClientNewData *data)
+{
+ NBDClient *client = data->client;
int csock = client->sock;
char buf[8 + 8 + 8 + 128];
int rc;
const int myflags = (NBD_FLAG_HAS_FLAGS | NBD_FLAG_SEND_TRIM |
NBD_FLAG_SEND_FLUSH | NBD_FLAG_SEND_FUA);
+ AioContext *ctx = client->exp ? client->exp->ctx : qemu_get_aio_context();
/* Negotiation header without options:
[ 0 .. 7] passwd ("NBDMAGIC")
@@ -531,9 +542,11 @@ static int nbd_send_negotiate(NBDClient *client)
[28 .. 151] reserved (0)
*/
- qemu_set_block(csock);
rc = -EINVAL;
+ aio_set_fd_handler(ctx, client->sock, true,
+ nbd_negotiate_continue,
+ nbd_negotiate_continue, data->co);
TRACE("Beginning negotiation.");
memset(buf, 0, sizeof(buf));
memcpy(buf, "NBDMAGIC", 8);
@@ -575,7 +588,8 @@ static int nbd_send_negotiate(NBDClient *client)
TRACE("Negotiation succeeded.");
rc = 0;
fail:
- qemu_set_nonblock(csock);
+ aio_set_fd_handler(ctx, client->sock, true,
+ NULL, NULL, NULL);
return rc;
}
@@ -1475,25 +1489,43 @@ static void nbd_update_can_read(NBDClient *client)
}
}
+static coroutine_fn void nbd_co_client_start(void *opaque)
+{
+ NBDClientNewData *data = opaque;
+ NBDClient *client = data->client;
+ NBDExport *exp = client->exp;
+
+ if (exp) {
+ nbd_export_get(exp);
+ }
+ if (nbd_send_negotiate(data)) {
+ shutdown(client->sock, 2);
+ client->close(client);
+ goto out;
+ }
+ qemu_co_mutex_init(&client->send_lock);
+ nbd_set_handlers(client);
+
+ if (exp) {
+ QTAILQ_INSERT_TAIL(&exp->clients, client, next);
+ }
+out:
+ g_free(data);
+}
+
void nbd_client_new(NBDExport *exp, int csock, void (*close_fn)(NBDClient *))
{
NBDClient *client;
+ NBDClientNewData *data = g_new(NBDClientNewData, 1);
+
client = g_malloc0(sizeof(NBDClient));
client->refcount = 1;
client->exp = exp;
client->sock = csock;
client->can_read = true;
- if (nbd_send_negotiate(client)) {
- shutdown(client->sock, 2);
- close_fn(client);
- return;
- }
client->close = close_fn;
- qemu_co_mutex_init(&client->send_lock);
- nbd_set_handlers(client);
- if (exp) {
- QTAILQ_INSERT_TAIL(&exp->clients, client, next);
- nbd_export_get(exp);
- }
+ data->client = client;
+ data->co = qemu_coroutine_create(nbd_co_client_start);
+ qemu_coroutine_enter(data->co, data);
}
--
2.4.3