diff options
Diffstat (limited to 'nbd/server.c')
| -rw-r--r-- | nbd/server.c | 150 |
1 files changed, 84 insertions, 66 deletions
diff --git a/nbd/server.c b/nbd/server.c index dc1d66fa47..15aa03da95 100644 --- a/nbd/server.c +++ b/nbd/server.c @@ -76,7 +76,8 @@ struct NBDClient { void (*close)(NBDClient *client); NBDExport *exp; - int sock; + QIOChannelSocket *sioc; /* The underlying data channel */ + QIOChannel *ioc; /* The current I/O channel which may differ (eg TLS) */ Coroutine *recv_coroutine; @@ -96,45 +97,56 @@ static void nbd_set_handlers(NBDClient *client); static void nbd_unset_handlers(NBDClient *client); static void nbd_update_can_read(NBDClient *client); -static void nbd_negotiate_continue(void *opaque) +static gboolean nbd_negotiate_continue(QIOChannel *ioc, + GIOCondition condition, + void *opaque) { qemu_coroutine_enter(opaque, NULL); + return TRUE; } -static ssize_t nbd_negotiate_read(int fd, void *buffer, size_t size) +static ssize_t nbd_negotiate_read(QIOChannel *ioc, void *buffer, size_t size) { ssize_t ret; + guint watch; assert(qemu_in_coroutine()); /* Negotiation are always in main loop. */ - qemu_set_fd_handler(fd, nbd_negotiate_continue, NULL, - qemu_coroutine_self()); - ret = read_sync(fd, buffer, size); - qemu_set_fd_handler(fd, NULL, NULL, NULL); + watch = qio_channel_add_watch(ioc, + G_IO_IN, + nbd_negotiate_continue, + qemu_coroutine_self(), + NULL); + ret = read_sync(ioc, buffer, size); + g_source_remove(watch); return ret; } -static ssize_t nbd_negotiate_write(int fd, void *buffer, size_t size) +static ssize_t nbd_negotiate_write(QIOChannel *ioc, void *buffer, size_t size) { ssize_t ret; + guint watch; assert(qemu_in_coroutine()); /* Negotiation are always in main loop. */ - qemu_set_fd_handler(fd, NULL, nbd_negotiate_continue, - qemu_coroutine_self()); - ret = write_sync(fd, buffer, size); - qemu_set_fd_handler(fd, NULL, NULL, NULL); + watch = qio_channel_add_watch(ioc, + G_IO_OUT, + nbd_negotiate_continue, + qemu_coroutine_self(), + NULL); + ret = write_sync(ioc, buffer, size); + g_source_remove(watch); return ret; } -static ssize_t nbd_negotiate_drop_sync(int fd, size_t size) +static ssize_t nbd_negotiate_drop_sync(QIOChannel *ioc, size_t size) { ssize_t ret, dropped = size; uint8_t *buffer = g_malloc(MIN(65536, size)); while (size > 0) { - ret = nbd_negotiate_read(fd, buffer, MIN(65536, size)); + ret = nbd_negotiate_read(ioc, buffer, MIN(65536, size)); if (ret < 0) { g_free(buffer); return ret; @@ -175,66 +187,66 @@ static ssize_t nbd_negotiate_drop_sync(int fd, size_t size) */ -static int nbd_negotiate_send_rep(int csock, uint32_t type, uint32_t opt) +static int nbd_negotiate_send_rep(QIOChannel *ioc, uint32_t type, uint32_t opt) { uint64_t magic; uint32_t len; magic = cpu_to_be64(NBD_REP_MAGIC); - if (nbd_negotiate_write(csock, &magic, sizeof(magic)) != sizeof(magic)) { + if (nbd_negotiate_write(ioc, &magic, sizeof(magic)) != sizeof(magic)) { LOG("write failed (rep magic)"); return -EINVAL; } opt = cpu_to_be32(opt); - if (nbd_negotiate_write(csock, &opt, sizeof(opt)) != sizeof(opt)) { + if (nbd_negotiate_write(ioc, &opt, sizeof(opt)) != sizeof(opt)) { LOG("write failed (rep opt)"); return -EINVAL; } type = cpu_to_be32(type); - if (nbd_negotiate_write(csock, &type, sizeof(type)) != sizeof(type)) { + if (nbd_negotiate_write(ioc, &type, sizeof(type)) != sizeof(type)) { LOG("write failed (rep type)"); return -EINVAL; } len = cpu_to_be32(0); - if (nbd_negotiate_write(csock, &len, sizeof(len)) != sizeof(len)) { + if (nbd_negotiate_write(ioc, &len, sizeof(len)) != sizeof(len)) { LOG("write failed (rep data length)"); return -EINVAL; } return 0; } -static int nbd_negotiate_send_rep_list(int csock, NBDExport *exp) +static int nbd_negotiate_send_rep_list(QIOChannel *ioc, NBDExport *exp) { uint64_t magic, name_len; uint32_t opt, type, len; name_len = strlen(exp->name); magic = cpu_to_be64(NBD_REP_MAGIC); - if (nbd_negotiate_write(csock, &magic, sizeof(magic)) != sizeof(magic)) { + if (nbd_negotiate_write(ioc, &magic, sizeof(magic)) != sizeof(magic)) { LOG("write failed (magic)"); return -EINVAL; } opt = cpu_to_be32(NBD_OPT_LIST); - if (nbd_negotiate_write(csock, &opt, sizeof(opt)) != sizeof(opt)) { + if (nbd_negotiate_write(ioc, &opt, sizeof(opt)) != sizeof(opt)) { LOG("write failed (opt)"); return -EINVAL; } type = cpu_to_be32(NBD_REP_SERVER); - if (nbd_negotiate_write(csock, &type, sizeof(type)) != sizeof(type)) { + if (nbd_negotiate_write(ioc, &type, sizeof(type)) != sizeof(type)) { LOG("write failed (reply type)"); return -EINVAL; } len = cpu_to_be32(name_len + sizeof(len)); - if (nbd_negotiate_write(csock, &len, sizeof(len)) != sizeof(len)) { + if (nbd_negotiate_write(ioc, &len, sizeof(len)) != sizeof(len)) { LOG("write failed (length)"); return -EINVAL; } len = cpu_to_be32(name_len); - if (nbd_negotiate_write(csock, &len, sizeof(len)) != sizeof(len)) { + if (nbd_negotiate_write(ioc, &len, sizeof(len)) != sizeof(len)) { LOG("write failed (length)"); return -EINVAL; } - if (nbd_negotiate_write(csock, exp->name, name_len) != name_len) { + if (nbd_negotiate_write(ioc, exp->name, name_len) != name_len) { LOG("write failed (buffer)"); return -EINVAL; } @@ -243,30 +255,29 @@ static int nbd_negotiate_send_rep_list(int csock, NBDExport *exp) static int nbd_negotiate_handle_list(NBDClient *client, uint32_t length) { - int csock; NBDExport *exp; - csock = client->sock; if (length) { - if (nbd_negotiate_drop_sync(csock, length) != length) { + if (nbd_negotiate_drop_sync(client->ioc, length) != length) { return -EIO; } - return nbd_negotiate_send_rep(csock, NBD_REP_ERR_INVALID, NBD_OPT_LIST); + return nbd_negotiate_send_rep(client->ioc, + NBD_REP_ERR_INVALID, NBD_OPT_LIST); } /* For each export, send a NBD_REP_SERVER reply. */ QTAILQ_FOREACH(exp, &exports, next) { - if (nbd_negotiate_send_rep_list(csock, exp)) { + if (nbd_negotiate_send_rep_list(client->ioc, exp)) { return -EINVAL; } } /* Finish with a NBD_REP_ACK. */ - return nbd_negotiate_send_rep(csock, NBD_REP_ACK, NBD_OPT_LIST); + return nbd_negotiate_send_rep(client->ioc, NBD_REP_ACK, NBD_OPT_LIST); } static int nbd_negotiate_handle_export_name(NBDClient *client, uint32_t length) { - int rc = -EINVAL, csock = client->sock; + int rc = -EINVAL; char name[256]; /* Client sends: @@ -277,7 +288,7 @@ static int nbd_negotiate_handle_export_name(NBDClient *client, uint32_t length) LOG("Bad length received"); goto fail; } - if (nbd_negotiate_read(csock, name, length) != length) { + if (nbd_negotiate_read(client->ioc, name, length) != length) { LOG("read failed"); goto fail; } @@ -298,7 +309,6 @@ fail: static int nbd_negotiate_options(NBDClient *client) { - int csock = client->sock; uint32_t flags; /* Client sends: @@ -315,7 +325,8 @@ static int nbd_negotiate_options(NBDClient *client) ... Rest of request */ - if (nbd_negotiate_read(csock, &flags, sizeof(flags)) != sizeof(flags)) { + if (nbd_negotiate_read(client->ioc, &flags, sizeof(flags)) != + sizeof(flags)) { LOG("read failed"); return -EIO; } @@ -331,7 +342,8 @@ static int nbd_negotiate_options(NBDClient *client) uint32_t tmp, length; uint64_t magic; - if (nbd_negotiate_read(csock, &magic, sizeof(magic)) != sizeof(magic)) { + if (nbd_negotiate_read(client->ioc, &magic, sizeof(magic)) != + sizeof(magic)) { LOG("read failed"); return -EINVAL; } @@ -341,13 +353,13 @@ static int nbd_negotiate_options(NBDClient *client) return -EINVAL; } - if (nbd_negotiate_read(csock, &tmp, sizeof(tmp)) != sizeof(tmp)) { + if (nbd_negotiate_read(client->ioc, &tmp, sizeof(tmp)) != sizeof(tmp)) { LOG("read failed"); return -EINVAL; } - if (nbd_negotiate_read(csock, &length, - sizeof(length)) != sizeof(length)) { + if (nbd_negotiate_read(client->ioc, &length, sizeof(length)) != + sizeof(length)) { LOG("read failed"); return -EINVAL; } @@ -371,7 +383,7 @@ static int nbd_negotiate_options(NBDClient *client) default: tmp = be32_to_cpu(tmp); LOG("Unsupported option 0x%x", tmp); - nbd_negotiate_send_rep(client->sock, NBD_REP_ERR_UNSUP, tmp); + nbd_negotiate_send_rep(client->ioc, NBD_REP_ERR_UNSUP, tmp); return -EINVAL; } } @@ -385,7 +397,6 @@ typedef struct { static coroutine_fn int nbd_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 | @@ -410,6 +421,7 @@ static coroutine_fn int nbd_negotiate(NBDClientNewData *data) [28 .. 151] reserved (0) */ + qio_channel_set_blocking(client->ioc, false, NULL); rc = -EINVAL; TRACE("Beginning negotiation."); @@ -426,12 +438,12 @@ static coroutine_fn int nbd_negotiate(NBDClientNewData *data) } if (client->exp) { - if (nbd_negotiate_write(csock, buf, sizeof(buf)) != sizeof(buf)) { + if (nbd_negotiate_write(client->ioc, buf, sizeof(buf)) != sizeof(buf)) { LOG("write failed"); goto fail; } } else { - if (nbd_negotiate_write(csock, buf, 18) != 18) { + if (nbd_negotiate_write(client->ioc, buf, 18) != 18) { LOG("write failed"); goto fail; } @@ -444,8 +456,8 @@ static coroutine_fn int nbd_negotiate(NBDClientNewData *data) assert ((client->exp->nbdflags & ~65535) == 0); stq_be_p(buf + 18, client->exp->size); stw_be_p(buf + 26, client->exp->nbdflags | myflags); - if (nbd_negotiate_write(csock, buf + 18, - sizeof(buf) - 18) != sizeof(buf) - 18) { + if (nbd_negotiate_write(client->ioc, buf + 18, sizeof(buf) - 18) != + sizeof(buf) - 18) { LOG("write failed"); goto fail; } @@ -475,13 +487,13 @@ int nbd_disconnect(int fd) } #endif -static ssize_t nbd_receive_request(int csock, struct nbd_request *request) +static ssize_t nbd_receive_request(QIOChannel *ioc, struct nbd_request *request) { uint8_t buf[NBD_REQUEST_SIZE]; uint32_t magic; ssize_t ret; - ret = read_sync(csock, buf, sizeof(buf)); + ret = read_sync(ioc, buf, sizeof(buf)); if (ret < 0) { return ret; } @@ -516,7 +528,7 @@ static ssize_t nbd_receive_request(int csock, struct nbd_request *request) return 0; } -static ssize_t nbd_send_reply(int csock, struct nbd_reply *reply) +static ssize_t nbd_send_reply(QIOChannel *ioc, struct nbd_reply *reply) { uint8_t buf[NBD_REPLY_SIZE]; ssize_t ret; @@ -534,7 +546,7 @@ static ssize_t nbd_send_reply(int csock, struct nbd_reply *reply) TRACE("Sending response to client"); - ret = write_sync(csock, buf, sizeof(buf)); + ret = write_sync(ioc, buf, sizeof(buf)); if (ret < 0) { return ret; } @@ -562,8 +574,8 @@ void nbd_client_put(NBDClient *client) assert(client->closing); nbd_unset_handlers(client); - close(client->sock); - client->sock = -1; + object_unref(OBJECT(client->sioc)); + object_unref(OBJECT(client->ioc)); if (client->exp) { QTAILQ_REMOVE(&client->exp->clients, client, next); nbd_export_put(client->exp); @@ -583,7 +595,8 @@ static void client_close(NBDClient *client) /* Force requests to finish. They will drop their own references, * then we'll close the socket and free the NBDClient. */ - shutdown(client->sock, 2); + qio_channel_shutdown(client->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, + NULL); /* Also tell the client, so that they release their reference. */ if (client->close) { @@ -789,25 +802,25 @@ static ssize_t nbd_co_send_reply(NBDRequest *req, struct nbd_reply *reply, int len) { NBDClient *client = req->client; - int csock = client->sock; ssize_t rc, ret; + g_assert(qemu_in_coroutine()); qemu_co_mutex_lock(&client->send_lock); client->send_coroutine = qemu_coroutine_self(); nbd_set_handlers(client); if (!len) { - rc = nbd_send_reply(csock, reply); + rc = nbd_send_reply(client->ioc, reply); } else { - socket_set_cork(csock, 1); - rc = nbd_send_reply(csock, reply); + qio_channel_set_cork(client->ioc, true); + rc = nbd_send_reply(client->ioc, reply); if (rc >= 0) { - ret = qemu_co_send(csock, req->data, len); + ret = write_sync(client->ioc, req->data, len); if (ret != len) { rc = -EIO; } } - socket_set_cork(csock, 0); + qio_channel_set_cork(client->ioc, false); } client->send_coroutine = NULL; @@ -819,14 +832,14 @@ static ssize_t nbd_co_send_reply(NBDRequest *req, struct nbd_reply *reply, static ssize_t nbd_co_receive_request(NBDRequest *req, struct nbd_request *request) { NBDClient *client = req->client; - int csock = client->sock; uint32_t command; ssize_t rc; + g_assert(qemu_in_coroutine()); client->recv_coroutine = qemu_coroutine_self(); nbd_update_can_read(client); - rc = nbd_receive_request(csock, request); + rc = nbd_receive_request(client->ioc, request); if (rc < 0) { if (rc != -EAGAIN) { rc = -EIO; @@ -861,7 +874,7 @@ static ssize_t nbd_co_receive_request(NBDRequest *req, struct nbd_request *reque if (command == NBD_CMD_WRITE) { TRACE("Reading %u byte(s)", request->len); - if (qemu_co_recv(csock, req->data, request->len) != request->len) { + if (read_sync(client->ioc, req->data, request->len) != request->len) { LOG("reading from socket failed"); rc = -EIO; goto out; @@ -1056,7 +1069,7 @@ static void nbd_restart_write(void *opaque) static void nbd_set_handlers(NBDClient *client) { if (client->exp && client->exp->ctx) { - aio_set_fd_handler(client->exp->ctx, client->sock, + aio_set_fd_handler(client->exp->ctx, client->sioc->fd, true, client->can_read ? nbd_read : NULL, client->send_coroutine ? nbd_restart_write : NULL, @@ -1067,7 +1080,7 @@ static void nbd_set_handlers(NBDClient *client) static void nbd_unset_handlers(NBDClient *client) { if (client->exp && client->exp->ctx) { - aio_set_fd_handler(client->exp->ctx, client->sock, + aio_set_fd_handler(client->exp->ctx, client->sioc->fd, true, NULL, NULL, NULL); } } @@ -1109,7 +1122,9 @@ out: g_free(data); } -void nbd_client_new(NBDExport *exp, int csock, void (*close_fn)(NBDClient *)) +void nbd_client_new(NBDExport *exp, + QIOChannelSocket *sioc, + void (*close_fn)(NBDClient *)) { NBDClient *client; NBDClientNewData *data = g_new(NBDClientNewData, 1); @@ -1117,7 +1132,10 @@ void nbd_client_new(NBDExport *exp, int csock, void (*close_fn)(NBDClient *)) client = g_malloc0(sizeof(NBDClient)); client->refcount = 1; client->exp = exp; - client->sock = csock; + client->sioc = sioc; + object_ref(OBJECT(client->sioc)); + client->ioc = QIO_CHANNEL(sioc); + object_ref(OBJECT(client->ioc)); client->can_read = true; client->close = close_fn; |