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_readv(BlockDriverState
*bs
, int64_t sector_num
,
222 int nb_sectors
, QEMUIOVector
*qiov
)
224 NbdClientSession
*client
= nbd_get_client_session(bs
);
225 struct nbd_request request
= { .type
= NBD_CMD_READ
};
226 struct nbd_reply reply
;
229 assert(nb_sectors
<= NBD_MAX_SECTORS
);
230 request
.from
= sector_num
* 512;
231 request
.len
= nb_sectors
* 512;
233 nbd_coroutine_start(client
, &request
);
234 ret
= nbd_co_send_request(bs
, &request
, NULL
);
238 nbd_co_receive_reply(client
, &request
, &reply
, qiov
);
240 nbd_coroutine_end(client
, &request
);
245 int nbd_client_co_writev(BlockDriverState
*bs
, int64_t sector_num
,
246 int nb_sectors
, QEMUIOVector
*qiov
, int flags
)
248 NbdClientSession
*client
= nbd_get_client_session(bs
);
249 struct nbd_request request
= { .type
= NBD_CMD_WRITE
};
250 struct nbd_reply reply
;
253 if (flags
& BDRV_REQ_FUA
) {
254 assert(client
->nbdflags
& NBD_FLAG_SEND_FUA
);
255 request
.type
|= NBD_CMD_FLAG_FUA
;
258 assert(nb_sectors
<= NBD_MAX_SECTORS
);
259 request
.from
= sector_num
* 512;
260 request
.len
= nb_sectors
* 512;
262 nbd_coroutine_start(client
, &request
);
263 ret
= nbd_co_send_request(bs
, &request
, qiov
);
267 nbd_co_receive_reply(client
, &request
, &reply
, NULL
);
269 nbd_coroutine_end(client
, &request
);
273 int nbd_client_co_flush(BlockDriverState
*bs
)
275 NbdClientSession
*client
= nbd_get_client_session(bs
);
276 struct nbd_request request
= { .type
= NBD_CMD_FLUSH
};
277 struct nbd_reply reply
;
280 if (!(client
->nbdflags
& NBD_FLAG_SEND_FLUSH
)) {
287 nbd_coroutine_start(client
, &request
);
288 ret
= nbd_co_send_request(bs
, &request
, NULL
);
292 nbd_co_receive_reply(client
, &request
, &reply
, NULL
);
294 nbd_coroutine_end(client
, &request
);
298 int nbd_client_co_discard(BlockDriverState
*bs
, int64_t sector_num
,
301 NbdClientSession
*client
= nbd_get_client_session(bs
);
302 struct nbd_request request
= { .type
= NBD_CMD_TRIM
};
303 struct nbd_reply reply
;
306 if (!(client
->nbdflags
& NBD_FLAG_SEND_TRIM
)) {
309 request
.from
= sector_num
* 512;
310 request
.len
= nb_sectors
* 512;
312 nbd_coroutine_start(client
, &request
);
313 ret
= nbd_co_send_request(bs
, &request
, NULL
);
317 nbd_co_receive_reply(client
, &request
, &reply
, NULL
);
319 nbd_coroutine_end(client
, &request
);
324 void nbd_client_detach_aio_context(BlockDriverState
*bs
)
326 aio_set_fd_handler(bdrv_get_aio_context(bs
),
327 nbd_get_client_session(bs
)->sioc
->fd
,
328 false, NULL
, NULL
, NULL
);
331 void nbd_client_attach_aio_context(BlockDriverState
*bs
,
332 AioContext
*new_context
)
334 aio_set_fd_handler(new_context
, nbd_get_client_session(bs
)->sioc
->fd
,
335 false, nbd_reply_ready
, NULL
, bs
);
338 void nbd_client_close(BlockDriverState
*bs
)
340 NbdClientSession
*client
= nbd_get_client_session(bs
);
341 struct nbd_request request
= {
342 .type
= NBD_CMD_DISC
,
347 if (client
->ioc
== NULL
) {
351 nbd_send_request(client
->ioc
, &request
);
353 nbd_teardown_connection(bs
);
356 int nbd_client_init(BlockDriverState
*bs
,
357 QIOChannelSocket
*sioc
,
359 QCryptoTLSCreds
*tlscreds
,
360 const char *hostname
,
363 NbdClientSession
*client
= nbd_get_client_session(bs
);
367 logout("session init %s\n", export
);
368 qio_channel_set_blocking(QIO_CHANNEL(sioc
), true, NULL
);
370 ret
= nbd_receive_negotiate(QIO_CHANNEL(sioc
), export
,
374 &client
->size
, errp
);
376 logout("Failed to negotiate with the NBD server\n");
379 if (client
->nbdflags
& NBD_FLAG_SEND_FUA
) {
380 bs
->supported_write_flags
= BDRV_REQ_FUA
;
383 qemu_co_mutex_init(&client
->send_mutex
);
384 qemu_co_mutex_init(&client
->free_sema
);
386 object_ref(OBJECT(client
->sioc
));
389 client
->ioc
= QIO_CHANNEL(sioc
);
390 object_ref(OBJECT(client
->ioc
));
393 /* Now that we're connected, set the socket to be non-blocking and
394 * kick the reply mechanism. */
395 qio_channel_set_blocking(QIO_CHANNEL(sioc
), false, NULL
);
397 nbd_client_attach_aio_context(bs
, bdrv_get_aio_context(bs
));
399 logout("Established connection with NBD server\n");