[qemu.git] / block / nbd-client.c

/*
 * QEMU Block driver for  NBD
 *
 * Copyright (C) 2008 Bull S.A.S.
 *     Author: Laurent Vivier <[email protected]>
 *
 * Some parts:
 *    Copyright (C) 2007 Anthony Liguori <[email protected]>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
 * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */

#include "qemu/osdep.h"
#include "nbd-client.h"

#define HANDLE_TO_INDEX(bs, handle) ((handle) ^ ((uint64_t)(intptr_t)bs))
#define INDEX_TO_HANDLE(bs, index)  ((index)  ^ ((uint64_t)(intptr_t)bs))

static void nbd_recv_coroutines_enter_all(NbdClientSession *s)
{
    int i;

    for (i = 0; i < MAX_NBD_REQUESTS; i++) {
        if (s->recv_coroutine[i]) {
            qemu_coroutine_enter(s->recv_coroutine[i], NULL);
        }
    }
}

static void nbd_teardown_connection(BlockDriverState *bs)
{
    NbdClientSession *client = nbd_get_client_session(bs);

    if (!client->ioc) { /* Already closed */
        return;
    }

    /* finish any pending coroutines */
    qio_channel_shutdown(client->ioc,
                         QIO_CHANNEL_SHUTDOWN_BOTH,
                         NULL);
    nbd_recv_coroutines_enter_all(client);

    nbd_client_detach_aio_context(bs);
    object_unref(OBJECT(client->sioc));
    client->sioc = NULL;
    object_unref(OBJECT(client->ioc));
    client->ioc = NULL;
}

static void nbd_reply_ready(void *opaque)
{
    BlockDriverState *bs = opaque;
    NbdClientSession *s = nbd_get_client_session(bs);
    uint64_t i;
    int ret;

    if (!s->ioc) { /* Already closed */
        return;
    }

    if (s->reply.handle == 0) {
        /* No reply already in flight.  Fetch a header.  It is possible
         * that another thread has done the same thing in parallel, so
         * the socket is not readable anymore.
         */
        ret = nbd_receive_reply(s->ioc, &s->reply);
        if (ret == -EAGAIN) {
            return;
        }
        if (ret < 0) {
            s->reply.handle = 0;
            goto fail;
        }
    }

    /* There's no need for a mutex on the receive side, because the
     * handler acts as a synchronization point and ensures that only
     * one coroutine is called until the reply finishes.  */
    i = HANDLE_TO_INDEX(s, s->reply.handle);
    if (i >= MAX_NBD_REQUESTS) {
        goto fail;
    }

    if (s->recv_coroutine[i]) {
        qemu_coroutine_enter(s->recv_coroutine[i], NULL);
        return;
    }

fail:
    nbd_teardown_connection(bs);
}

static void nbd_restart_write(void *opaque)
{
    BlockDriverState *bs = opaque;

    qemu_coroutine_enter(nbd_get_client_session(bs)->send_coroutine, NULL);
}

static int nbd_co_send_request(BlockDriverState *bs,
                               struct nbd_request *request,
                               QEMUIOVector *qiov, int offset)
{
    NbdClientSession *s = nbd_get_client_session(bs);
    AioContext *aio_context;
    int rc, ret, i;

    qemu_co_mutex_lock(&s->send_mutex);

    for (i = 0; i < MAX_NBD_REQUESTS; i++) {
        if (s->recv_coroutine[i] == NULL) {
            s->recv_coroutine[i] = qemu_coroutine_self();
            break;
        }
    }

    g_assert(qemu_in_coroutine());
    assert(i < MAX_NBD_REQUESTS);
    request->handle = INDEX_TO_HANDLE(s, i);

    if (!s->ioc) {
        qemu_co_mutex_unlock(&s->send_mutex);
        return -EPIPE;
    }

    s->send_coroutine = qemu_coroutine_self();
    aio_context = bdrv_get_aio_context(bs);

    aio_set_fd_handler(aio_context, s->sioc->fd, false,
                       nbd_reply_ready, nbd_restart_write, bs);
    if (qiov) {
        qio_channel_set_cork(s->ioc, true);
        rc = nbd_send_request(s->ioc, request);
        if (rc >= 0) {
            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->ioc, request);
    }
    aio_set_fd_handler(aio_context, s->sioc->fd, false,
                       nbd_reply_ready, NULL, bs);
    s->send_coroutine = NULL;
    qemu_co_mutex_unlock(&s->send_mutex);
    return rc;
}

static void nbd_co_receive_reply(NbdClientSession *s,
    struct nbd_request *request, struct nbd_reply *reply,
    QEMUIOVector *qiov, int offset)
{
    int ret;

    /* Wait until we're woken up by the read handler.  TODO: perhaps
     * peek at the next reply and avoid yielding if it's ours?  */
    qemu_coroutine_yield();
    *reply = s->reply;
    if (reply->handle != request->handle ||
        !s->ioc) {
        reply->error = EIO;
    } else {
        if (qiov && reply->error == 0) {
            ret = nbd_wr_syncv(s->ioc, qiov->iov, qiov->niov,
                               offset, request->len, 1);
            if (ret != request->len) {
                reply->error = EIO;
            }
        }

        /* Tell the read handler to read another header.  */
        s->reply.handle = 0;
    }
}

static void nbd_coroutine_start(NbdClientSession *s,
   struct nbd_request *request)
{
    /* Poor man semaphore.  The free_sema is locked when no other request
     * can be accepted, and unlocked after receiving one reply.  */
    if (s->in_flight >= MAX_NBD_REQUESTS - 1) {
        qemu_co_mutex_lock(&s->free_sema);
        assert(s->in_flight < MAX_NBD_REQUESTS);
    }
    s->in_flight++;

    /* s->recv_coroutine[i] is set as soon as we get the send_lock.  */
}

static void nbd_coroutine_end(NbdClientSession *s,
    struct nbd_request *request)
{
    int i = HANDLE_TO_INDEX(s, request->handle);
    s->recv_coroutine[i] = NULL;
    if (s->in_flight-- == MAX_NBD_REQUESTS) {
        qemu_co_mutex_unlock(&s->free_sema);
    }
}

static int nbd_co_readv_1(BlockDriverState *bs, int64_t sector_num,
                          int nb_sectors, QEMUIOVector *qiov,
                          int offset)
{
    NbdClientSession *client = nbd_get_client_session(bs);
    struct nbd_request request = { .type = NBD_CMD_READ };
    struct nbd_reply reply;
    ssize_t ret;

    request.from = sector_num * 512;
    request.len = nb_sectors * 512;

    nbd_coroutine_start(client, &request);
    ret = nbd_co_send_request(bs, &request, NULL, 0);
    if (ret < 0) {
        reply.error = -ret;
    } else {
        nbd_co_receive_reply(client, &request, &reply, qiov, offset);
    }
    nbd_coroutine_end(client, &request);
    return -reply.error;

}

static int nbd_co_writev_1(BlockDriverState *bs, int64_t sector_num,
                           int nb_sectors, QEMUIOVector *qiov,
                           int offset, int flags)
{
    NbdClientSession *client = nbd_get_client_session(bs);
    struct nbd_request request = { .type = NBD_CMD_WRITE };
    struct nbd_reply reply;
    ssize_t ret;

    if (flags & BDRV_REQ_FUA) {
        assert(client->nbdflags & NBD_FLAG_SEND_FUA);
        request.type |= NBD_CMD_FLAG_FUA;
    }

    request.from = sector_num * 512;
    request.len = nb_sectors * 512;

    nbd_coroutine_start(client, &request);
    ret = nbd_co_send_request(bs, &request, qiov, offset);
    if (ret < 0) {
        reply.error = -ret;
    } else {
        nbd_co_receive_reply(client, &request, &reply, NULL, 0);
    }
    nbd_coroutine_end(client, &request);
    return -reply.error;
}

int nbd_client_co_readv(BlockDriverState *bs, int64_t sector_num,
                        int nb_sectors, QEMUIOVector *qiov)
{
    int offset = 0;
    int ret;
    while (nb_sectors > NBD_MAX_SECTORS) {
        ret = nbd_co_readv_1(bs, sector_num, NBD_MAX_SECTORS, qiov, offset);
        if (ret < 0) {
            return ret;
        }
        offset += NBD_MAX_SECTORS * 512;
        sector_num += NBD_MAX_SECTORS;
        nb_sectors -= NBD_MAX_SECTORS;
    }
    return nbd_co_readv_1(bs, sector_num, nb_sectors, qiov, offset);
}

int nbd_client_co_writev(BlockDriverState *bs, int64_t sector_num,
                         int nb_sectors, QEMUIOVector *qiov, int flags)
{
    int offset = 0;
    int ret;
    while (nb_sectors > NBD_MAX_SECTORS) {
        ret = nbd_co_writev_1(bs, sector_num, NBD_MAX_SECTORS, qiov, offset,
                              flags);
        if (ret < 0) {
            return ret;
        }
        offset += NBD_MAX_SECTORS * 512;
        sector_num += NBD_MAX_SECTORS;
        nb_sectors -= NBD_MAX_SECTORS;
    }
    return nbd_co_writev_1(bs, sector_num, nb_sectors, qiov, offset, flags);
}

int nbd_client_co_flush(BlockDriverState *bs)
{
    NbdClientSession *client = nbd_get_client_session(bs);
    struct nbd_request request = { .type = NBD_CMD_FLUSH };
    struct nbd_reply reply;
    ssize_t ret;

    if (!(client->nbdflags & NBD_FLAG_SEND_FLUSH)) {
        return 0;
    }

    request.from = 0;
    request.len = 0;

    nbd_coroutine_start(client, &request);
    ret = nbd_co_send_request(bs, &request, NULL, 0);
    if (ret < 0) {
        reply.error = -ret;
    } else {
        nbd_co_receive_reply(client, &request, &reply, NULL, 0);
    }
    nbd_coroutine_end(client, &request);
    return -reply.error;
}

int nbd_client_co_discard(BlockDriverState *bs, int64_t sector_num,
                          int nb_sectors)
{
    NbdClientSession *client = nbd_get_client_session(bs);
    struct nbd_request request = { .type = NBD_CMD_TRIM };
    struct nbd_reply reply;
    ssize_t ret;

    if (!(client->nbdflags & NBD_FLAG_SEND_TRIM)) {
        return 0;
    }
    request.from = sector_num * 512;
    request.len = nb_sectors * 512;

    nbd_coroutine_start(client, &request);
    ret = nbd_co_send_request(bs, &request, NULL, 0);
    if (ret < 0) {
        reply.error = -ret;
    } else {
        nbd_co_receive_reply(client, &request, &reply, NULL, 0);
    }
    nbd_coroutine_end(client, &request);
    return -reply.error;

}

void nbd_client_detach_aio_context(BlockDriverState *bs)
{
    aio_set_fd_handler(bdrv_get_aio_context(bs),
                       nbd_get_client_session(bs)->sioc->fd,
                       false, NULL, NULL, NULL);
}

void nbd_client_attach_aio_context(BlockDriverState *bs,
                                   AioContext *new_context)
{
    aio_set_fd_handler(new_context, nbd_get_client_session(bs)->sioc->fd,
                       false, nbd_reply_ready, NULL, bs);
}

void nbd_client_close(BlockDriverState *bs)
{
    NbdClientSession *client = nbd_get_client_session(bs);
    struct nbd_request request = {
        .type = NBD_CMD_DISC,
        .from = 0,
        .len = 0
    };

    if (client->ioc == NULL) {
        return;
    }

    nbd_send_request(client->ioc, &request);

    nbd_teardown_connection(bs);
}

int nbd_client_init(BlockDriverState *bs,
                    QIOChannelSocket *sioc,
                    const char *export,
                    QCryptoTLSCreds *tlscreds,
                    const char *hostname,
                    Error **errp)
{
    NbdClientSession *client = nbd_get_client_session(bs);
    int ret;

    /* NBD handshake */
    logout("session init %s\n", export);
    qio_channel_set_blocking(QIO_CHANNEL(sioc), true, NULL);

    ret = nbd_receive_negotiate(QIO_CHANNEL(sioc), export,
                                &client->nbdflags,
                                tlscreds, hostname,
                                &client->ioc,
                                &client->size, errp);
    if (ret < 0) {
        logout("Failed to negotiate with the NBD server\n");
        return ret;
    }
    if (client->nbdflags & NBD_FLAG_SEND_FUA) {
        bs->supported_write_flags = BDRV_REQ_FUA;
    }

    qemu_co_mutex_init(&client->send_mutex);
    qemu_co_mutex_init(&client->free_sema);
    client->sioc = sioc;
    object_ref(OBJECT(client->sioc));

    if (!client->ioc) {
        client->ioc = QIO_CHANNEL(sioc);
        object_ref(OBJECT(client->ioc));
    }

    /* Now that we're connected, set the socket to be non-blocking and
     * kick the reply mechanism.  */
    qio_channel_set_blocking(QIO_CHANNEL(sioc), false, NULL);

    nbd_client_attach_aio_context(bs, bdrv_get_aio_context(bs));

    logout("Established connection with NBD server\n");
    return 0;
}
Commit	Line	Data
2302c1ca MAL	1	/*
	2	* QEMU Block driver for NBD
	3	*
	4	* Copyright (C) 2008 Bull S.A.S.
	5	* Author: Laurent Vivier <[email protected]>
	6	*
	7	* Some parts:
	8	* Copyright (C) 2007 Anthony Liguori <[email protected]>
	9	*
	10	* Permission is hereby granted, free of charge, to any person obtaining a copy
	11	* of this software and associated documentation files (the "Software"), to deal
	12	* in the Software without restriction, including without limitation the rights
	13	* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
	14	* copies of the Software, and to permit persons to whom the Software is
	15	* furnished to do so, subject to the following conditions:
	16	*
	17	* The above copyright notice and this permission notice shall be included in
	18	* all copies or substantial portions of the Software.
	19	*
	20	* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
	21	* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
	22	* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
	23	* THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
	24	* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
	25	* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
	26	* THE SOFTWARE.
	27	*/
	28
80c71a24	29	#include "qemu/osdep.h"
2302c1ca	30	#include "nbd-client.h"
2302c1ca MAL	31
	32	#define HANDLE_TO_INDEX(bs, handle) ((handle) ^ ((uint64_t)(intptr_t)bs))
	33	#define INDEX_TO_HANDLE(bs, index) ((index) ^ ((uint64_t)(intptr_t)bs))
	34
69152c09 MAL	35	static void nbd_recv_coroutines_enter_all(NbdClientSession *s)
	36	{
	37	int i;
	38
	39	for (i = 0; i < MAX_NBD_REQUESTS; i++) {
	40	if (s->recv_coroutine[i]) {
	41	qemu_coroutine_enter(s->recv_coroutine[i], NULL);
	42	}
	43	}
	44	}
	45
f53a829b	46	static void nbd_teardown_connection(BlockDriverState *bs)
4a41a2d6	47	{
f53a829b HR	48	NbdClientSession *client = nbd_get_client_session(bs);
f53a829b HR	49
064097d9 DB	50	if (!client->ioc) { /* Already closed */
	51	return;
	52	}
	53
4a41a2d6	54	/* finish any pending coroutines */
064097d9 DB	55	qio_channel_shutdown(client->ioc,
	56	QIO_CHANNEL_SHUTDOWN_BOTH,
	57	NULL);
4a41a2d6 SH	58	nbd_recv_coroutines_enter_all(client);
4a41a2d6 SH	59
f53a829b	60	nbd_client_detach_aio_context(bs);
064097d9 DB	61	object_unref(OBJECT(client->sioc));
	62	client->sioc = NULL;
	63	object_unref(OBJECT(client->ioc));
	64	client->ioc = NULL;
4a41a2d6 SH	65	}
4a41a2d6 SH	66
2302c1ca MAL	67	static void nbd_reply_ready(void *opaque)
2302c1ca MAL	68	{
f53a829b HR	69	BlockDriverState *bs = opaque;
f53a829b HR	70	NbdClientSession *s = nbd_get_client_session(bs);
2302c1ca MAL	71	uint64_t i;
	72	int ret;
	73
064097d9 DB	74	if (!s->ioc) { /* Already closed */
	75	return;
	76	}
	77
2302c1ca MAL	78	if (s->reply.handle == 0) {
	79	/* No reply already in flight. Fetch a header. It is possible
	80	* that another thread has done the same thing in parallel, so
	81	* the socket is not readable anymore.
	82	*/
1c778ef7	83	ret = nbd_receive_reply(s->ioc, &s->reply);
2302c1ca MAL	84	if (ret == -EAGAIN) {
	85	return;
	86	}
	87	if (ret < 0) {
	88	s->reply.handle = 0;
	89	goto fail;
	90	}
	91	}
	92
	93	/* There's no need for a mutex on the receive side, because the
	94	* handler acts as a synchronization point and ensures that only
	95	* one coroutine is called until the reply finishes. */
	96	i = HANDLE_TO_INDEX(s, s->reply.handle);
	97	if (i >= MAX_NBD_REQUESTS) {
	98	goto fail;
	99	}
	100
	101	if (s->recv_coroutine[i]) {
	102	qemu_coroutine_enter(s->recv_coroutine[i], NULL);
	103	return;
	104	}
	105
	106	fail:
f53a829b	107	nbd_teardown_connection(bs);
2302c1ca MAL	108	}
	109
	110	static void nbd_restart_write(void *opaque)
	111	{
f53a829b	112	BlockDriverState *bs = opaque;
2302c1ca	113
f53a829b	114	qemu_coroutine_enter(nbd_get_client_session(bs)->send_coroutine, NULL);
2302c1ca MAL	115	}
2302c1ca MAL	116
f53a829b HR	117	static int nbd_co_send_request(BlockDriverState *bs,
	118	struct nbd_request *request,
	119	QEMUIOVector *qiov, int offset)
2302c1ca	120	{
f53a829b	121	NbdClientSession *s = nbd_get_client_session(bs);
69447cd8	122	AioContext *aio_context;
141cabe6	123	int rc, ret, i;
2302c1ca MAL	124
2302c1ca MAL	125	qemu_co_mutex_lock(&s->send_mutex);
141cabe6 BW	126
	127	for (i = 0; i < MAX_NBD_REQUESTS; i++) {
	128	if (s->recv_coroutine[i] == NULL) {
	129	s->recv_coroutine[i] = qemu_coroutine_self();
	130	break;
	131	}
	132	}
	133
1c778ef7	134	g_assert(qemu_in_coroutine());
141cabe6 BW	135	assert(i < MAX_NBD_REQUESTS);
141cabe6 BW	136	request->handle = INDEX_TO_HANDLE(s, i);
064097d9 DB	137
	138	if (!s->ioc) {
	139	qemu_co_mutex_unlock(&s->send_mutex);
	140	return -EPIPE;
	141	}
	142
2302c1ca	143	s->send_coroutine = qemu_coroutine_self();
f53a829b	144	aio_context = bdrv_get_aio_context(bs);
141cabe6	145
064097d9	146	aio_set_fd_handler(aio_context, s->sioc->fd, false,
f53a829b	147	nbd_reply_ready, nbd_restart_write, bs);
2302c1ca	148	if (qiov) {
064097d9	149	qio_channel_set_cork(s->ioc, true);
1c778ef7	150	rc = nbd_send_request(s->ioc, request);
2302c1ca	151	if (rc >= 0) {
1c778ef7 DB	152	ret = nbd_wr_syncv(s->ioc, qiov->iov, qiov->niov,
1c778ef7 DB	153	offset, request->len, 0);
2302c1ca MAL	154	if (ret != request->len) {
	155	rc = -EIO;
	156	}
	157	}
064097d9	158	qio_channel_set_cork(s->ioc, false);
2302c1ca	159	} else {
1c778ef7	160	rc = nbd_send_request(s->ioc, request);
2302c1ca	161	}
064097d9	162	aio_set_fd_handler(aio_context, s->sioc->fd, false,
dca21ef2	163	nbd_reply_ready, NULL, bs);
2302c1ca MAL	164	s->send_coroutine = NULL;
	165	qemu_co_mutex_unlock(&s->send_mutex);
	166	return rc;
	167	}
	168
	169	static void nbd_co_receive_reply(NbdClientSession *s,
	170	struct nbd_request request, struct nbd_reply reply,
	171	QEMUIOVector *qiov, int offset)
	172	{
	173	int ret;
	174
	175	/* Wait until we're woken up by the read handler. TODO: perhaps
	176	* peek at the next reply and avoid yielding if it's ours? */
	177	qemu_coroutine_yield();
	178	*reply = s->reply;
064097d9 DB	179	if (reply->handle != request->handle \|\|
064097d9 DB	180	!s->ioc) {
2302c1ca MAL	181	reply->error = EIO;
	182	} else {
	183	if (qiov && reply->error == 0) {
1c778ef7 DB	184	ret = nbd_wr_syncv(s->ioc, qiov->iov, qiov->niov,
1c778ef7 DB	185	offset, request->len, 1);
2302c1ca MAL	186	if (ret != request->len) {
	187	reply->error = EIO;
	188	}
	189	}
	190
	191	/* Tell the read handler to read another header. */
	192	s->reply.handle = 0;
	193	}
	194	}
	195
	196	static void nbd_coroutine_start(NbdClientSession *s,
	197	struct nbd_request *request)
	198	{
2302c1ca MAL	199	/* Poor man semaphore. The free_sema is locked when no other request
	200	* can be accepted, and unlocked after receiving one reply. */
	201	if (s->in_flight >= MAX_NBD_REQUESTS - 1) {
	202	qemu_co_mutex_lock(&s->free_sema);
	203	assert(s->in_flight < MAX_NBD_REQUESTS);
	204	}
	205	s->in_flight++;
	206
141cabe6	207	/* s->recv_coroutine[i] is set as soon as we get the send_lock. */
2302c1ca MAL	208	}
	209
	210	static void nbd_coroutine_end(NbdClientSession *s,
	211	struct nbd_request *request)
	212	{
	213	int i = HANDLE_TO_INDEX(s, request->handle);
	214	s->recv_coroutine[i] = NULL;
	215	if (s->in_flight-- == MAX_NBD_REQUESTS) {
	216	qemu_co_mutex_unlock(&s->free_sema);
	217	}
	218	}
	219
f53a829b	220	static int nbd_co_readv_1(BlockDriverState *bs, int64_t sector_num,
2302c1ca MAL	221	int nb_sectors, QEMUIOVector *qiov,
	222	int offset)
	223	{
f53a829b	224	NbdClientSession *client = nbd_get_client_session(bs);
b1b27b64	225	struct nbd_request request = { .type = NBD_CMD_READ };
2302c1ca MAL	226	struct nbd_reply reply;
	227	ssize_t ret;
	228
2302c1ca MAL	229	request.from = sector_num * 512;
	230	request.len = nb_sectors * 512;
	231
	232	nbd_coroutine_start(client, &request);
f53a829b	233	ret = nbd_co_send_request(bs, &request, NULL, 0);
2302c1ca MAL	234	if (ret < 0) {
	235	reply.error = -ret;
	236	} else {
	237	nbd_co_receive_reply(client, &request, &reply, qiov, offset);
	238	}
	239	nbd_coroutine_end(client, &request);
	240	return -reply.error;
	241
	242	}
	243
f53a829b	244	static int nbd_co_writev_1(BlockDriverState *bs, int64_t sector_num,
2302c1ca	245	int nb_sectors, QEMUIOVector *qiov,
52a46505	246	int offset, int flags)
2302c1ca	247	{
f53a829b	248	NbdClientSession *client = nbd_get_client_session(bs);
b1b27b64	249	struct nbd_request request = { .type = NBD_CMD_WRITE };
2302c1ca MAL	250	struct nbd_reply reply;
	251	ssize_t ret;
	252
52a46505 EB	253	if (flags & BDRV_REQ_FUA) {
52a46505 EB	254	assert(client->nbdflags & NBD_FLAG_SEND_FUA);
2302c1ca MAL	255	request.type \|= NBD_CMD_FLAG_FUA;
	256	}
	257
	258	request.from = sector_num * 512;
	259	request.len = nb_sectors * 512;
	260
	261	nbd_coroutine_start(client, &request);
f53a829b	262	ret = nbd_co_send_request(bs, &request, qiov, offset);
2302c1ca MAL	263	if (ret < 0) {
	264	reply.error = -ret;
	265	} else {
	266	nbd_co_receive_reply(client, &request, &reply, NULL, 0);
	267	}
	268	nbd_coroutine_end(client, &request);
	269	return -reply.error;
	270	}
	271
f53a829b HR	272	int nbd_client_co_readv(BlockDriverState *bs, int64_t sector_num,
f53a829b HR	273	int nb_sectors, QEMUIOVector *qiov)
2302c1ca MAL	274	{
	275	int offset = 0;
	276	int ret;
	277	while (nb_sectors > NBD_MAX_SECTORS) {
f53a829b	278	ret = nbd_co_readv_1(bs, sector_num, NBD_MAX_SECTORS, qiov, offset);
2302c1ca MAL	279	if (ret < 0) {
	280	return ret;
	281	}
	282	offset += NBD_MAX_SECTORS * 512;
	283	sector_num += NBD_MAX_SECTORS;
	284	nb_sectors -= NBD_MAX_SECTORS;
	285	}
f53a829b	286	return nbd_co_readv_1(bs, sector_num, nb_sectors, qiov, offset);
2302c1ca MAL	287	}
2302c1ca MAL	288
f53a829b	289	int nbd_client_co_writev(BlockDriverState *bs, int64_t sector_num,
52a46505	290	int nb_sectors, QEMUIOVector *qiov, int flags)
2302c1ca MAL	291	{
	292	int offset = 0;
	293	int ret;
	294	while (nb_sectors > NBD_MAX_SECTORS) {
2b556518 KW	295	ret = nbd_co_writev_1(bs, sector_num, NBD_MAX_SECTORS, qiov, offset,
2b556518 KW	296	flags);
2302c1ca MAL	297	if (ret < 0) {
	298	return ret;
	299	}
	300	offset += NBD_MAX_SECTORS * 512;
	301	sector_num += NBD_MAX_SECTORS;
	302	nb_sectors -= NBD_MAX_SECTORS;
	303	}
2b556518	304	return nbd_co_writev_1(bs, sector_num, nb_sectors, qiov, offset, flags);
2302c1ca MAL	305	}
2302c1ca MAL	306
f53a829b	307	int nbd_client_co_flush(BlockDriverState *bs)
2302c1ca	308	{
f53a829b	309	NbdClientSession *client = nbd_get_client_session(bs);
b1b27b64	310	struct nbd_request request = { .type = NBD_CMD_FLUSH };
2302c1ca MAL	311	struct nbd_reply reply;
	312	ssize_t ret;
	313
	314	if (!(client->nbdflags & NBD_FLAG_SEND_FLUSH)) {
	315	return 0;
	316	}
	317
2302c1ca MAL	318	request.from = 0;
	319	request.len = 0;
	320
	321	nbd_coroutine_start(client, &request);
f53a829b	322	ret = nbd_co_send_request(bs, &request, NULL, 0);
2302c1ca MAL	323	if (ret < 0) {
	324	reply.error = -ret;
	325	} else {
	326	nbd_co_receive_reply(client, &request, &reply, NULL, 0);
	327	}
	328	nbd_coroutine_end(client, &request);
	329	return -reply.error;
	330	}
	331
f53a829b HR	332	int nbd_client_co_discard(BlockDriverState *bs, int64_t sector_num,
f53a829b HR	333	int nb_sectors)
2302c1ca	334	{
f53a829b	335	NbdClientSession *client = nbd_get_client_session(bs);
b1b27b64	336	struct nbd_request request = { .type = NBD_CMD_TRIM };
2302c1ca MAL	337	struct nbd_reply reply;
	338	ssize_t ret;
	339
	340	if (!(client->nbdflags & NBD_FLAG_SEND_TRIM)) {
	341	return 0;
	342	}
2302c1ca MAL	343	request.from = sector_num * 512;
	344	request.len = nb_sectors * 512;
	345
	346	nbd_coroutine_start(client, &request);
f53a829b	347	ret = nbd_co_send_request(bs, &request, NULL, 0);
2302c1ca MAL	348	if (ret < 0) {
	349	reply.error = -ret;
	350	} else {
	351	nbd_co_receive_reply(client, &request, &reply, NULL, 0);
	352	}
	353	nbd_coroutine_end(client, &request);
	354	return -reply.error;
	355
	356	}
	357
f53a829b	358	void nbd_client_detach_aio_context(BlockDriverState *bs)
69447cd8	359	{
f53a829b	360	aio_set_fd_handler(bdrv_get_aio_context(bs),
064097d9	361	nbd_get_client_session(bs)->sioc->fd,
dca21ef2	362	false, NULL, NULL, NULL);
69447cd8 SH	363	}
69447cd8 SH	364
f53a829b HR	365	void nbd_client_attach_aio_context(BlockDriverState *bs,
f53a829b HR	366	AioContext *new_context)
69447cd8	367	{
064097d9	368	aio_set_fd_handler(new_context, nbd_get_client_session(bs)->sioc->fd,
dca21ef2	369	false, nbd_reply_ready, NULL, bs);
69447cd8 SH	370	}
69447cd8 SH	371
f53a829b	372	void nbd_client_close(BlockDriverState *bs)
2302c1ca	373	{
f53a829b	374	NbdClientSession *client = nbd_get_client_session(bs);
b1b27b64 MAL	375	struct nbd_request request = {
	376	.type = NBD_CMD_DISC,
	377	.from = 0,
	378	.len = 0
	379	};
2302c1ca	380
064097d9	381	if (client->ioc == NULL) {
4a41a2d6 SH	382	return;
	383	}
	384
1c778ef7	385	nbd_send_request(client->ioc, &request);
5ad283eb	386
f53a829b	387	nbd_teardown_connection(bs);
2302c1ca MAL	388	}
2302c1ca MAL	389
75822a12 DB	390	int nbd_client_init(BlockDriverState *bs,
	391	QIOChannelSocket *sioc,
	392	const char *export,
	393	QCryptoTLSCreds *tlscreds,
	394	const char *hostname,
	395	Error **errp)
2302c1ca	396	{
f53a829b	397	NbdClientSession *client = nbd_get_client_session(bs);
2302c1ca MAL	398	int ret;
	399
	400	/* NBD handshake */
e2bc625f	401	logout("session init %s\n", export);
064097d9 DB	402	qio_channel_set_blocking(QIO_CHANNEL(sioc), true, NULL);
064097d9 DB	403
1c778ef7	404	ret = nbd_receive_negotiate(QIO_CHANNEL(sioc), export,
f95910fe	405	&client->nbdflags,
75822a12	406	tlscreds, hostname,
f95910fe DB	407	&client->ioc,
f95910fe DB	408	&client->size, errp);
2302c1ca MAL	409	if (ret < 0) {
2302c1ca MAL	410	logout("Failed to negotiate with the NBD server\n");
2302c1ca MAL	411	return ret;
2302c1ca MAL	412	}
4df863f3 EB	413	if (client->nbdflags & NBD_FLAG_SEND_FUA) {
	414	bs->supported_write_flags = BDRV_REQ_FUA;
	415	}
2302c1ca MAL	416
	417	qemu_co_mutex_init(&client->send_mutex);
	418	qemu_co_mutex_init(&client->free_sema);
064097d9 DB	419	client->sioc = sioc;
064097d9 DB	420	object_ref(OBJECT(client->sioc));
f95910fe DB	421
	422	if (!client->ioc) {
	423	client->ioc = QIO_CHANNEL(sioc);
	424	object_ref(OBJECT(client->ioc));
	425	}
2302c1ca MAL	426
	427	/* Now that we're connected, set the socket to be non-blocking and
	428	* kick the reply mechanism. */
064097d9 DB	429	qio_channel_set_blocking(QIO_CHANNEL(sioc), false, NULL);
064097d9 DB	430
f53a829b	431	nbd_client_attach_aio_context(bs, bdrv_get_aio_context(bs));
2302c1ca MAL	432
	433	logout("Established connection with NBD server\n");
	434	return 0;
	435	}