aboutsummaryrefslogtreecommitdiff
path: root/nbd
diff options
context:
space:
mode:
Diffstat (limited to 'nbd')
-rw-r--r--nbd/client.c40
-rw-r--r--nbd/common.c68
-rw-r--r--nbd/nbd-internal.h18
-rw-r--r--nbd/server.c150
4 files changed, 152 insertions, 124 deletions
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;
generated by cgit v1.2.3 (git 2.26.2) at 2024-12-22 03:39:44 +0000