diff options
| -rw-r--r-- | block/nbd-client.c | 19 | ||||
| -rw-r--r-- | blockdev-nbd.c | 6 | ||||
| -rw-r--r-- | include/block/nbd.h | 20 | ||||
| -rw-r--r-- | nbd/client.c | 40 | ||||
| -rw-r--r-- | nbd/common.c | 68 | ||||
| -rw-r--r-- | nbd/nbd-internal.h | 18 | ||||
| -rw-r--r-- | nbd/server.c | 150 | ||||
| -rw-r--r-- | qemu-nbd.c | 10 |
8 files changed, 180 insertions, 151 deletions
diff --git a/block/nbd-client.c b/block/nbd-client.c index c4379a9821..75bb2d92f6 100644 --- a/block/nbd-client.c +++ b/block/nbd-client.c @@ -80,7 +80,7 @@ static void nbd_reply_ready(void *opaque) * that another thread has done the same thing in parallel, so * the socket is not readable anymore. */ - ret = nbd_receive_reply(s->sioc->fd, &s->reply); + ret = nbd_receive_reply(s->ioc, &s->reply); if (ret == -EAGAIN) { return; } @@ -131,6 +131,7 @@ static int nbd_co_send_request(BlockDriverState *bs, } } + g_assert(qemu_in_coroutine()); assert(i < MAX_NBD_REQUESTS); request->handle = INDEX_TO_HANDLE(s, i); @@ -146,17 +147,17 @@ static int nbd_co_send_request(BlockDriverState *bs, nbd_reply_ready, nbd_restart_write, bs); if (qiov) { qio_channel_set_cork(s->ioc, true); - rc = nbd_send_request(s->sioc->fd, request); + rc = nbd_send_request(s->ioc, request); if (rc >= 0) { - ret = qemu_co_sendv(s->sioc->fd, qiov->iov, qiov->niov, - offset, request->len); + ret = nbd_wr_syncv(s->ioc, qiov->iov, qiov->niov, + offset, request->len, 0); if (ret != request->len) { rc = -EIO; } } qio_channel_set_cork(s->ioc, false); } else { - rc = nbd_send_request(s->sioc->fd, request); + rc = nbd_send_request(s->ioc, request); } aio_set_fd_handler(aio_context, s->sioc->fd, false, nbd_reply_ready, NULL, bs); @@ -180,8 +181,8 @@ static void nbd_co_receive_reply(NbdClientSession *s, reply->error = EIO; } else { if (qiov && reply->error == 0) { - ret = qemu_co_recvv(s->sioc->fd, qiov->iov, qiov->niov, - offset, request->len); + ret = nbd_wr_syncv(s->ioc, qiov->iov, qiov->niov, + offset, request->len, 1); if (ret != request->len) { reply->error = EIO; } @@ -388,7 +389,7 @@ void nbd_client_close(BlockDriverState *bs) return; } - nbd_send_request(client->sioc->fd, &request); + nbd_send_request(client->ioc, &request); nbd_teardown_connection(bs); } @@ -403,7 +404,7 @@ int nbd_client_init(BlockDriverState *bs, QIOChannelSocket *sioc, logout("session init %s\n", export); qio_channel_set_blocking(QIO_CHANNEL(sioc), true, NULL); - ret = nbd_receive_negotiate(sioc->fd, export, + ret = nbd_receive_negotiate(QIO_CHANNEL(sioc), export, &client->nbdflags, &client->size, errp); if (ret < 0) { logout("Failed to negotiate with the NBD server\n"); diff --git a/blockdev-nbd.c b/blockdev-nbd.c index 9baf88316c..2148753fff 100644 --- a/blockdev-nbd.c +++ b/blockdev-nbd.c @@ -27,7 +27,6 @@ static gboolean nbd_accept(QIOChannel *ioc, GIOCondition condition, gpointer opaque) { QIOChannelSocket *cioc; - int fd; cioc = qio_channel_socket_accept(QIO_CHANNEL_SOCKET(ioc), NULL); @@ -35,10 +34,7 @@ static gboolean nbd_accept(QIOChannel *ioc, GIOCondition condition, return TRUE; } - fd = dup(cioc->fd); - if (fd >= 0) { - nbd_client_new(NULL, fd, nbd_client_put); - } + nbd_client_new(NULL, cioc, nbd_client_put); object_unref(OBJECT(cioc)); return TRUE; } diff --git a/include/block/nbd.h b/include/block/nbd.h index 7eccb41da8..1080ef83de 100644 --- a/include/block/nbd.h +++ b/include/block/nbd.h @@ -23,6 +23,7 @@ #include "qemu-common.h" #include "qemu/option.h" +#include "io/channel-socket.h" struct nbd_request { uint32_t magic; @@ -73,12 +74,17 @@ enum { /* Maximum size of a single READ/WRITE data buffer */ #define NBD_MAX_BUFFER_SIZE (32 * 1024 * 1024) -ssize_t nbd_wr_sync(int fd, void *buffer, size_t size, bool do_read); -int nbd_receive_negotiate(int csock, const char *name, uint32_t *flags, +ssize_t nbd_wr_syncv(QIOChannel *ioc, + struct iovec *iov, + size_t niov, + size_t offset, + size_t length, + bool do_read); +int nbd_receive_negotiate(QIOChannel *ioc, const char *name, uint32_t *flags, off_t *size, Error **errp); -int nbd_init(int fd, int csock, uint32_t flags, off_t size); -ssize_t nbd_send_request(int csock, struct nbd_request *request); -ssize_t nbd_receive_reply(int csock, struct nbd_reply *reply); +int nbd_init(int fd, QIOChannelSocket *sioc, uint32_t flags, off_t size); +ssize_t nbd_send_request(QIOChannel *ioc, struct nbd_request *request); +ssize_t nbd_receive_reply(QIOChannel *ioc, struct nbd_reply *reply); int nbd_client(int fd); int nbd_disconnect(int fd); @@ -98,7 +104,9 @@ NBDExport *nbd_export_find(const char *name); void nbd_export_set_name(NBDExport *exp, const char *name); void nbd_export_close_all(void); -void nbd_client_new(NBDExport *exp, int csock, void (*close_fn)(NBDClient *)); +void nbd_client_new(NBDExport *exp, + QIOChannelSocket *sioc, + void (*close)(NBDClient *)); void nbd_client_get(NBDClient *client); void nbd_client_put(NBDClient *client); diff --git a/nbd/client.c b/nbd/client.c index f07cb4822d..f6dd0a398f 100644 --- a/nbd/client.c +++ b/nbd/client.c @@ -71,7 +71,7 @@ static QTAILQ_HEAD(, NBDExport) exports = QTAILQ_HEAD_INITIALIZER(exports); */ -int nbd_receive_negotiate(int csock, const char *name, uint32_t *flags, +int nbd_receive_negotiate(QIOChannel *ioc, const char *name, uint32_t *flags, off_t *size, Error **errp) { char buf[256]; @@ -83,7 +83,7 @@ int nbd_receive_negotiate(int csock, const char *name, uint32_t *flags, rc = -EINVAL; - if (read_sync(csock, buf, 8) != 8) { + if (read_sync(ioc, buf, 8) != 8) { error_setg(errp, "Failed to read data"); goto fail; } @@ -109,7 +109,7 @@ int nbd_receive_negotiate(int csock, const char *name, uint32_t *flags, goto fail; } - if (read_sync(csock, &magic, sizeof(magic)) != sizeof(magic)) { + if (read_sync(ioc, &magic, sizeof(magic)) != sizeof(magic)) { error_setg(errp, "Failed to read magic"); goto fail; } @@ -130,35 +130,35 @@ int nbd_receive_negotiate(int csock, const char *name, uint32_t *flags, } goto fail; } - if (read_sync(csock, &tmp, sizeof(tmp)) != sizeof(tmp)) { + if (read_sync(ioc, &tmp, sizeof(tmp)) != sizeof(tmp)) { error_setg(errp, "Failed to read server flags"); goto fail; } *flags = be16_to_cpu(tmp) << 16; /* reserved for future use */ - if (write_sync(csock, &reserved, sizeof(reserved)) != + if (write_sync(ioc, &reserved, sizeof(reserved)) != sizeof(reserved)) { error_setg(errp, "Failed to read reserved field"); goto fail; } /* write the export name */ magic = cpu_to_be64(magic); - if (write_sync(csock, &magic, sizeof(magic)) != sizeof(magic)) { + if (write_sync(ioc, &magic, sizeof(magic)) != sizeof(magic)) { error_setg(errp, "Failed to send export name magic"); goto fail; } opt = cpu_to_be32(NBD_OPT_EXPORT_NAME); - if (write_sync(csock, &opt, sizeof(opt)) != sizeof(opt)) { + if (write_sync(ioc, &opt, sizeof(opt)) != sizeof(opt)) { error_setg(errp, "Failed to send export name option number"); goto fail; } namesize = cpu_to_be32(strlen(name)); - if (write_sync(csock, &namesize, sizeof(namesize)) != + if (write_sync(ioc, &namesize, sizeof(namesize)) != sizeof(namesize)) { error_setg(errp, "Failed to send export name length"); goto fail; } - if (write_sync(csock, (char*)name, strlen(name)) != strlen(name)) { + if (write_sync(ioc, (char *)name, strlen(name)) != strlen(name)) { error_setg(errp, "Failed to send export name"); goto fail; } @@ -175,7 +175,7 @@ int nbd_receive_negotiate(int csock, const char *name, uint32_t *flags, } } - if (read_sync(csock, &s, sizeof(s)) != sizeof(s)) { + if (read_sync(ioc, &s, sizeof(s)) != sizeof(s)) { error_setg(errp, "Failed to read export length"); goto fail; } @@ -183,19 +183,19 @@ int nbd_receive_negotiate(int csock, const char *name, uint32_t *flags, TRACE("Size is %" PRIu64, *size); if (!name) { - if (read_sync(csock, flags, sizeof(*flags)) != sizeof(*flags)) { + if (read_sync(ioc, flags, sizeof(*flags)) != sizeof(*flags)) { error_setg(errp, "Failed to read export flags"); goto fail; } *flags = be32_to_cpup(flags); } else { - if (read_sync(csock, &tmp, sizeof(tmp)) != sizeof(tmp)) { + if (read_sync(ioc, &tmp, sizeof(tmp)) != sizeof(tmp)) { error_setg(errp, "Failed to read export flags"); goto fail; } *flags |= be16_to_cpu(tmp); } - if (read_sync(csock, &buf, 124) != 124) { + if (read_sync(ioc, &buf, 124) != 124) { error_setg(errp, "Failed to read reserved block"); goto fail; } @@ -206,11 +206,11 @@ fail: } #ifdef __linux__ -int nbd_init(int fd, int csock, uint32_t flags, off_t size) +int nbd_init(int fd, QIOChannelSocket *sioc, uint32_t flags, off_t size) { TRACE("Setting NBD socket"); - if (ioctl(fd, NBD_SET_SOCK, csock) < 0) { + if (ioctl(fd, NBD_SET_SOCK, sioc->fd) < 0) { int serrno = errno; LOG("Failed to set NBD socket"); return -serrno; @@ -283,7 +283,7 @@ int nbd_client(int fd) return ret; } #else -int nbd_init(int fd, int csock, uint32_t flags, off_t size) +int nbd_init(int fd, QIOChannelSocket *ioc, uint32_t flags, off_t size) { return -ENOTSUP; } @@ -294,7 +294,7 @@ int nbd_client(int fd) } #endif -ssize_t nbd_send_request(int csock, struct nbd_request *request) +ssize_t nbd_send_request(QIOChannel *ioc, struct nbd_request *request) { uint8_t buf[NBD_REQUEST_SIZE]; ssize_t ret; @@ -309,7 +309,7 @@ ssize_t nbd_send_request(int csock, struct nbd_request *request) "{ .from = %" PRIu64", .len = %u, .handle = %" PRIu64", .type=%i}", request->from, request->len, request->handle, request->type); - ret = write_sync(csock, buf, sizeof(buf)); + ret = write_sync(ioc, buf, sizeof(buf)); if (ret < 0) { return ret; } @@ -321,13 +321,13 @@ ssize_t nbd_send_request(int csock, struct nbd_request *request) return 0; } -ssize_t nbd_receive_reply(int csock, struct nbd_reply *reply) +ssize_t nbd_receive_reply(QIOChannel *ioc, struct nbd_reply *reply) { uint8_t buf[NBD_REPLY_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; } diff --git a/nbd/common.c b/nbd/common.c index 178d4a7ff2..2b19c95099 100644 --- a/nbd/common.c +++ b/nbd/common.c @@ -19,47 +19,59 @@ #include "qemu/osdep.h" #include "nbd-internal.h" -ssize_t nbd_wr_sync(int fd, void *buffer, size_t size, bool do_read) +ssize_t nbd_wr_syncv(QIOChannel *ioc, + struct iovec *iov, + size_t niov, + size_t offset, + size_t length, + bool do_read) { - size_t offset = 0; - int err; + ssize_t done = 0; + Error *local_err = NULL; + struct iovec *local_iov = g_new(struct iovec, niov); + struct iovec *local_iov_head = local_iov; + unsigned int nlocal_iov = niov; - if (qemu_in_coroutine()) { - if (do_read) { - return qemu_co_recv(fd, buffer, size); - } else { - return qemu_co_send(fd, buffer, size); - } - } + nlocal_iov = iov_copy(local_iov, nlocal_iov, + iov, niov, + offset, length); - while (offset < size) { + while (nlocal_iov > 0) { ssize_t len; - if (do_read) { - len = qemu_recv(fd, buffer + offset, size - offset, 0); + len = qio_channel_readv(ioc, local_iov, nlocal_iov, &local_err); } else { - len = send(fd, buffer + offset, size - offset, 0); + len = qio_channel_writev(ioc, local_iov, nlocal_iov, &local_err); } - - if (len < 0) { - err = socket_error(); - - /* recoverable error */ - if (err == EINTR || (offset > 0 && (err == EAGAIN || err == EWOULDBLOCK))) { - continue; + if (len == QIO_CHANNEL_ERR_BLOCK) { + if (qemu_in_coroutine()) { + /* XXX figure out if we can create a variant on + * qio_channel_yield() that works with AIO contexts + * and consider using that in this branch */ + qemu_coroutine_yield(); + } else { + qio_channel_wait(ioc, + do_read ? G_IO_IN : G_IO_OUT); } - - /* unrecoverable error */ - return -err; + continue; + } + if (len < 0) { + TRACE("I/O error: %s", error_get_pretty(local_err)); + error_free(local_err); + /* XXX handle Error objects */ + done = -EIO; + goto cleanup; } - /* eof */ - if (len == 0) { + if (do_read && len == 0) { break; } - offset += len; + iov_discard_front(&local_iov, &nlocal_iov, len); + done += len; } - return offset; + cleanup: + g_free(local_iov_head); + return done; } diff --git a/nbd/nbd-internal.h b/nbd/nbd-internal.h index c0a657575b..9960034612 100644 --- a/nbd/nbd-internal.h +++ b/nbd/nbd-internal.h @@ -13,6 +13,7 @@ #include "sysemu/block-backend.h" #include "qemu/coroutine.h" +#include "qemu/iov.h" #include <errno.h> #include <string.h> @@ -29,7 +30,6 @@ #include <linux/fs.h> #endif -#include "qemu/sockets.h" #include "qemu/queue.h" #include "qemu/main-loop.h" @@ -90,24 +90,22 @@ #define NBD_EINVAL 22 #define NBD_ENOSPC 28 -static inline ssize_t read_sync(int fd, void *buffer, size_t size) +static inline ssize_t read_sync(QIOChannel *ioc, void *buffer, size_t size) { + struct iovec iov = { .iov_base = buffer, .iov_len = 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. */ - return nbd_wr_sync(fd, buffer, size, true); + return nbd_wr_syncv(ioc, &iov, 1, 0, size, true); } -static inline ssize_t write_sync(int fd, void *buffer, size_t size) +static inline ssize_t write_sync(QIOChannel *ioc, void *buffer, size_t size) { - int ret; - do { - /* For writes, we do expect the socket to be writable. */ - ret = nbd_wr_sync(fd, buffer, size, false); - } while (ret == -EAGAIN); - return ret; + struct iovec iov = { .iov_base = buffer, .iov_len = size }; + + return nbd_wr_syncv(ioc, &iov, 1, 0, size, false); } #endif 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; diff --git a/qemu-nbd.c b/qemu-nbd.c index bc309e0231..5e54290411 100644 --- a/qemu-nbd.c +++ b/qemu-nbd.c @@ -249,7 +249,7 @@ static void *nbd_client_thread(void *arg) goto out; } - ret = nbd_receive_negotiate(sioc->fd, NULL, &nbdflags, + ret = nbd_receive_negotiate(QIO_CHANNEL(sioc), NULL, &nbdflags, &size, &local_error); if (ret < 0) { if (local_error) { @@ -265,7 +265,7 @@ static void *nbd_client_thread(void *arg) goto out_socket; } - ret = nbd_init(fd, sioc->fd, nbdflags, size); + ret = nbd_init(fd, sioc, nbdflags, size); if (ret < 0) { goto out_fd; } @@ -325,7 +325,6 @@ static void nbd_client_closed(NBDClient *client) static gboolean nbd_accept(QIOChannel *ioc, GIOCondition cond, gpointer opaque) { QIOChannelSocket *cioc; - int fd; cioc = qio_channel_socket_accept(QIO_CHANNEL_SOCKET(ioc), NULL); @@ -340,10 +339,7 @@ static gboolean nbd_accept(QIOChannel *ioc, GIOCondition cond, gpointer opaque) nb_fds++; nbd_update_server_watch(); - fd = dup(cioc->fd); - if (fd >= 0) { - nbd_client_new(exp, fd, nbd_client_closed); - } + nbd_client_new(exp, cioc, nbd_client_closed); object_unref(OBJECT(cioc)); return TRUE; |