2 * QEMU Block driver for NBD
4 * Copyright (C) 2008 Bull S.A.S.
5 * Author: Laurent Vivier <Laurent.Vivier@bull.net>
8 * Copyright (C) 2007 Anthony Liguori <anthony@codemonkey.ws>
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:
17 * The above copyright notice and this permission notice shall be included in
18 * all copies or substantial portions of the Software.
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
29 #include "qemu/osdep.h"
30 #include "nbd-client.h"
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))
35 static void nbd_recv_coroutines_enter_all(NbdClientSession *s)
39 for (i = 0; i < MAX_NBD_REQUESTS; i++) {
40 if (s->recv_coroutine[i]) {
41 qemu_coroutine_enter(s->recv_coroutine[i]);
46 static void nbd_teardown_connection(BlockDriverState *bs)
48 NbdClientSession *client = nbd_get_client_session(bs);
50 if (!client->ioc) { /* Already closed */
54 /* finish any pending coroutines */
55 qio_channel_shutdown(client->ioc,
56 QIO_CHANNEL_SHUTDOWN_BOTH,
58 nbd_recv_coroutines_enter_all(client);
60 nbd_client_detach_aio_context(bs);
61 object_unref(OBJECT(client->sioc));
63 object_unref(OBJECT(client->ioc));
67 static void nbd_reply_ready(void *opaque)
69 BlockDriverState *bs = opaque;
70 NbdClientSession *s = nbd_get_client_session(bs);
74 if (!s->ioc) { /* Already closed */
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.
83 ret = nbd_receive_reply(s->ioc, &s->reply);
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) {
101 if (s->recv_coroutine[i]) {
102 qemu_coroutine_enter(s->recv_coroutine[i]);
107 nbd_teardown_connection(bs);
110 static void nbd_restart_write(void *opaque)
112 BlockDriverState *bs = opaque;
114 qemu_coroutine_enter(nbd_get_client_session(bs)->send_coroutine);
117 static int nbd_co_send_request(BlockDriverState *bs,
118 struct nbd_request *request,
121 NbdClientSession *s = nbd_get_client_session(bs);
122 AioContext *aio_context;
125 qemu_co_mutex_lock(&s->send_mutex);
127 for (i = 0; i < MAX_NBD_REQUESTS; i++) {
128 if (s->recv_coroutine[i] == NULL) {
129 s->recv_coroutine[i] = qemu_coroutine_self();
134 g_assert(qemu_in_coroutine());
135 assert(i < MAX_NBD_REQUESTS);
136 request->handle = INDEX_TO_HANDLE(s, i);
139 qemu_co_mutex_unlock(&s->send_mutex);
143 s->send_coroutine = qemu_coroutine_self();
144 aio_context = bdrv_get_aio_context(bs);
146 aio_set_fd_handler(aio_context, s->sioc->fd, false,
147 nbd_reply_ready, nbd_restart_write, bs);
149 qio_channel_set_cork(s->ioc, true);
150 rc = nbd_send_request(s->ioc, request);
152 ret = nbd_wr_syncv(s->ioc, qiov->iov, qiov->niov, request->len,
154 if (ret != request->len) {
158 qio_channel_set_cork(s->ioc, false);
160 rc = nbd_send_request(s->ioc, request);
162 aio_set_fd_handler(aio_context, s->sioc->fd, false,
163 nbd_reply_ready, NULL, bs);
164 s->send_coroutine = NULL;
165 qemu_co_mutex_unlock(&s->send_mutex);
169 static void nbd_co_receive_reply(NbdClientSession *s,
170 struct nbd_request *request,
171 struct nbd_reply *reply,
176 /* Wait until we're woken up by the read handler. TODO: perhaps
177 * peek at the next reply and avoid yielding if it's ours? */
178 qemu_coroutine_yield();
180 if (reply->handle != request->handle ||
184 if (qiov && reply->error == 0) {
185 ret = nbd_wr_syncv(s->ioc, qiov->iov, qiov->niov, request->len,
187 if (ret != request->len) {
192 /* Tell the read handler to read another header. */
197 static void nbd_coroutine_start(NbdClientSession *s,
198 struct nbd_request *request)
200 /* Poor man semaphore. The free_sema is locked when no other request
201 * can be accepted, and unlocked after receiving one reply. */
202 if (s->in_flight >= MAX_NBD_REQUESTS - 1) {
203 qemu_co_mutex_lock(&s->free_sema);
204 assert(s->in_flight < MAX_NBD_REQUESTS);
208 /* s->recv_coroutine[i] is set as soon as we get the send_lock. */
211 static void nbd_coroutine_end(NbdClientSession *s,
212 struct nbd_request *request)
214 int i = HANDLE_TO_INDEX(s, request->handle);
215 s->recv_coroutine[i] = NULL;
216 if (s->in_flight-- == MAX_NBD_REQUESTS) {
217 qemu_co_mutex_unlock(&s->free_sema);
221 int nbd_client_co_preadv(BlockDriverState *bs, uint64_t offset,
222 uint64_t bytes, QEMUIOVector *qiov, int flags)
224 NbdClientSession *client = nbd_get_client_session(bs);
225 struct nbd_request request = {
226 .type = NBD_CMD_READ,
230 struct nbd_reply reply;
233 assert(bytes <= NBD_MAX_BUFFER_SIZE);
236 nbd_coroutine_start(client, &request);
237 ret = nbd_co_send_request(bs, &request, NULL);
241 nbd_co_receive_reply(client, &request, &reply, qiov);
243 nbd_coroutine_end(client, &request);
247 int nbd_client_co_pwritev(BlockDriverState *bs, uint64_t offset,
248 uint64_t bytes, QEMUIOVector *qiov, int flags)
250 NbdClientSession *client = nbd_get_client_session(bs);
251 struct nbd_request request = {
252 .type = NBD_CMD_WRITE,
256 struct nbd_reply reply;
259 if (flags & BDRV_REQ_FUA) {
260 assert(client->nbdflags & NBD_FLAG_SEND_FUA);
261 request.type |= NBD_CMD_FLAG_FUA;
264 assert(bytes <= NBD_MAX_BUFFER_SIZE);
266 nbd_coroutine_start(client, &request);
267 ret = nbd_co_send_request(bs, &request, qiov);
271 nbd_co_receive_reply(client, &request, &reply, NULL);
273 nbd_coroutine_end(client, &request);
277 int nbd_client_co_flush(BlockDriverState *bs)
279 NbdClientSession *client = nbd_get_client_session(bs);
280 struct nbd_request request = { .type = NBD_CMD_FLUSH };
281 struct nbd_reply reply;
284 if (!(client->nbdflags & NBD_FLAG_SEND_FLUSH)) {
291 nbd_coroutine_start(client, &request);
292 ret = nbd_co_send_request(bs, &request, NULL);
296 nbd_co_receive_reply(client, &request, &reply, NULL);
298 nbd_coroutine_end(client, &request);
302 int nbd_client_co_pdiscard(BlockDriverState *bs, int64_t offset, int count)
304 NbdClientSession *client = nbd_get_client_session(bs);
305 struct nbd_request request = {
306 .type = NBD_CMD_TRIM,
310 struct nbd_reply reply;
313 if (!(client->nbdflags & NBD_FLAG_SEND_TRIM)) {
317 nbd_coroutine_start(client, &request);
318 ret = nbd_co_send_request(bs, &request, NULL);
322 nbd_co_receive_reply(client, &request, &reply, NULL);
324 nbd_coroutine_end(client, &request);
329 void nbd_client_detach_aio_context(BlockDriverState *bs)
331 aio_set_fd_handler(bdrv_get_aio_context(bs),
332 nbd_get_client_session(bs)->sioc->fd,
333 false, NULL, NULL, NULL);
336 void nbd_client_attach_aio_context(BlockDriverState *bs,
337 AioContext *new_context)
339 aio_set_fd_handler(new_context, nbd_get_client_session(bs)->sioc->fd,
340 false, nbd_reply_ready, NULL, bs);
343 void nbd_client_close(BlockDriverState *bs)
345 NbdClientSession *client = nbd_get_client_session(bs);
346 struct nbd_request request = {
347 .type = NBD_CMD_DISC,
352 if (client->ioc == NULL) {
356 nbd_send_request(client->ioc, &request);
358 nbd_teardown_connection(bs);
361 int nbd_client_init(BlockDriverState *bs,
362 QIOChannelSocket *sioc,
364 QCryptoTLSCreds *tlscreds,
365 const char *hostname,
368 NbdClientSession *client = nbd_get_client_session(bs);
372 logout("session init %s\n", export);
373 qio_channel_set_blocking(QIO_CHANNEL(sioc), true, NULL);
375 ret = nbd_receive_negotiate(QIO_CHANNEL(sioc), export,
379 &client->size, errp);
381 logout("Failed to negotiate with the NBD server\n");
384 if (client->nbdflags & NBD_FLAG_SEND_FUA) {
385 bs->supported_write_flags = BDRV_REQ_FUA;
388 qemu_co_mutex_init(&client->send_mutex);
389 qemu_co_mutex_init(&client->free_sema);
391 object_ref(OBJECT(client->sioc));
394 client->ioc = QIO_CHANNEL(sioc);
395 object_ref(OBJECT(client->ioc));
398 /* Now that we're connected, set the socket to be non-blocking and
399 * kick the reply mechanism. */
400 qio_channel_set_blocking(QIO_CHANNEL(sioc), false, NULL);
402 nbd_client_attach_aio_context(bs, bdrv_get_aio_context(bs));
404 logout("Established connection with NBD server\n");