2 * QEMU Block driver for NBD
4 * Copyright (C) 2016 Red Hat, Inc.
5 * Copyright (C) 2008 Bull S.A.S.
6 * Author: Laurent Vivier <Laurent.Vivier@bull.net>
9 * Copyright (C) 2007 Anthony Liguori <anthony@codemonkey.ws>
11 * Permission is hereby granted, free of charge, to any person obtaining a copy
12 * of this software and associated documentation files (the "Software"), to deal
13 * in the Software without restriction, including without limitation the rights
14 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
15 * copies of the Software, and to permit persons to whom the Software is
16 * furnished to do so, subject to the following conditions:
18 * The above copyright notice and this permission notice shall be included in
19 * all copies or substantial portions of the Software.
21 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
22 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
23 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
24 * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
25 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
26 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
30 #include "qemu/osdep.h"
31 #include "qapi/error.h"
32 #include "nbd-client.h"
34 #define HANDLE_TO_INDEX(bs, handle) ((handle) ^ ((uint64_t)(intptr_t)bs))
35 #define INDEX_TO_HANDLE(bs, index) ((index) ^ ((uint64_t)(intptr_t)bs))
37 static void nbd_recv_coroutines_wake_all(NBDClientSession
*s
)
41 for (i
= 0; i
< MAX_NBD_REQUESTS
; i
++) {
42 NBDClientRequest
*req
= &s
->requests
[i
];
44 if (req
->coroutine
&& req
->receiving
) {
45 aio_co_wake(req
->coroutine
);
50 static void nbd_teardown_connection(BlockDriverState
*bs
)
52 NBDClientSession
*client
= nbd_get_client_session(bs
);
54 if (!client
->ioc
) { /* Already closed */
58 /* finish any pending coroutines */
59 qio_channel_shutdown(client
->ioc
,
60 QIO_CHANNEL_SHUTDOWN_BOTH
,
62 BDRV_POLL_WHILE(bs
, client
->read_reply_co
);
64 nbd_client_detach_aio_context(bs
);
65 object_unref(OBJECT(client
->sioc
));
67 object_unref(OBJECT(client
->ioc
));
71 static coroutine_fn
void nbd_read_reply_entry(void *opaque
)
73 NBDClientSession
*s
= opaque
;
76 Error
*local_err
= NULL
;
79 assert(s
->reply
.handle
== 0);
80 ret
= nbd_receive_reply(s
->ioc
, &s
->reply
, &local_err
);
82 error_report_err(local_err
);
88 /* There's no need for a mutex on the receive side, because the
89 * handler acts as a synchronization point and ensures that only
90 * one coroutine is called until the reply finishes.
92 i
= HANDLE_TO_INDEX(s
, s
->reply
.handle
);
93 if (i
>= MAX_NBD_REQUESTS
||
94 !s
->requests
[i
].coroutine
||
95 !s
->requests
[i
].receiving
) {
99 /* We're woken up again by the request itself. Note that there
100 * is no race between yielding and reentering read_reply_co. This
103 * - if the request runs on the same AioContext, it is only
104 * entered after we yield
106 * - if the request runs on a different AioContext, reentering
107 * read_reply_co happens through a bottom half, which can only
108 * run after we yield.
110 aio_co_wake(s
->requests
[i
].coroutine
);
111 qemu_coroutine_yield();
115 nbd_recv_coroutines_wake_all(s
);
116 s
->read_reply_co
= NULL
;
119 static int nbd_co_send_request(BlockDriverState
*bs
,
123 NBDClientSession
*s
= nbd_get_client_session(bs
);
126 qemu_co_mutex_lock(&s
->send_mutex
);
127 while (s
->in_flight
== MAX_NBD_REQUESTS
) {
128 qemu_co_queue_wait(&s
->free_sema
, &s
->send_mutex
);
132 for (i
= 0; i
< MAX_NBD_REQUESTS
; i
++) {
133 if (s
->requests
[i
].coroutine
== NULL
) {
138 g_assert(qemu_in_coroutine());
139 assert(i
< MAX_NBD_REQUESTS
);
141 s
->requests
[i
].coroutine
= qemu_coroutine_self();
142 s
->requests
[i
].receiving
= false;
144 request
->handle
= INDEX_TO_HANDLE(s
, i
);
156 qio_channel_set_cork(s
->ioc
, true);
157 rc
= nbd_send_request(s
->ioc
, request
);
158 if (rc
>= 0 && !s
->quit
) {
159 ret
= nbd_rwv(s
->ioc
, qiov
->iov
, qiov
->niov
, request
->len
, false,
161 if (ret
!= request
->len
) {
165 qio_channel_set_cork(s
->ioc
, false);
167 rc
= nbd_send_request(s
->ioc
, request
);
173 s
->requests
[i
].coroutine
= NULL
;
175 qemu_co_queue_next(&s
->free_sema
);
177 qemu_co_mutex_unlock(&s
->send_mutex
);
181 static void nbd_co_receive_reply(NBDClientSession
*s
,
186 int i
= HANDLE_TO_INDEX(s
, request
->handle
);
189 /* Wait until we're woken up by nbd_read_reply_entry. */
190 s
->requests
[i
].receiving
= true;
191 qemu_coroutine_yield();
192 s
->requests
[i
].receiving
= false;
194 if (reply
->handle
!= request
->handle
|| !s
->ioc
|| s
->quit
) {
197 if (qiov
&& reply
->error
== 0) {
198 ret
= nbd_rwv(s
->ioc
, qiov
->iov
, qiov
->niov
, request
->len
, true,
200 if (ret
!= request
->len
) {
206 /* Tell the read handler to read another header. */
210 s
->requests
[i
].coroutine
= NULL
;
212 /* Kick the read_reply_co to get the next reply. */
213 if (s
->read_reply_co
) {
214 aio_co_wake(s
->read_reply_co
);
217 qemu_co_mutex_lock(&s
->send_mutex
);
219 qemu_co_queue_next(&s
->free_sema
);
220 qemu_co_mutex_unlock(&s
->send_mutex
);
223 static int nbd_co_request(BlockDriverState
*bs
,
227 NBDClientSession
*client
= nbd_get_client_session(bs
);
231 assert(!qiov
|| request
->type
== NBD_CMD_WRITE
||
232 request
->type
== NBD_CMD_READ
);
233 ret
= nbd_co_send_request(bs
, request
,
234 request
->type
== NBD_CMD_WRITE
? qiov
: NULL
);
238 nbd_co_receive_reply(client
, request
, &reply
,
239 request
->type
== NBD_CMD_READ
? qiov
: NULL
);
244 int nbd_client_co_preadv(BlockDriverState
*bs
, uint64_t offset
,
245 uint64_t bytes
, QEMUIOVector
*qiov
, int flags
)
247 NBDRequest request
= {
248 .type
= NBD_CMD_READ
,
253 assert(bytes
<= NBD_MAX_BUFFER_SIZE
);
256 return nbd_co_request(bs
, &request
, qiov
);
259 int nbd_client_co_pwritev(BlockDriverState
*bs
, uint64_t offset
,
260 uint64_t bytes
, QEMUIOVector
*qiov
, int flags
)
262 NBDClientSession
*client
= nbd_get_client_session(bs
);
263 NBDRequest request
= {
264 .type
= NBD_CMD_WRITE
,
269 if (flags
& BDRV_REQ_FUA
) {
270 assert(client
->info
.flags
& NBD_FLAG_SEND_FUA
);
271 request
.flags
|= NBD_CMD_FLAG_FUA
;
274 assert(bytes
<= NBD_MAX_BUFFER_SIZE
);
276 return nbd_co_request(bs
, &request
, qiov
);
279 int nbd_client_co_pwrite_zeroes(BlockDriverState
*bs
, int64_t offset
,
280 int bytes
, BdrvRequestFlags flags
)
282 NBDClientSession
*client
= nbd_get_client_session(bs
);
283 NBDRequest request
= {
284 .type
= NBD_CMD_WRITE_ZEROES
,
289 if (!(client
->info
.flags
& NBD_FLAG_SEND_WRITE_ZEROES
)) {
293 if (flags
& BDRV_REQ_FUA
) {
294 assert(client
->info
.flags
& NBD_FLAG_SEND_FUA
);
295 request
.flags
|= NBD_CMD_FLAG_FUA
;
297 if (!(flags
& BDRV_REQ_MAY_UNMAP
)) {
298 request
.flags
|= NBD_CMD_FLAG_NO_HOLE
;
301 return nbd_co_request(bs
, &request
, NULL
);
304 int nbd_client_co_flush(BlockDriverState
*bs
)
306 NBDClientSession
*client
= nbd_get_client_session(bs
);
307 NBDRequest request
= { .type
= NBD_CMD_FLUSH
};
309 if (!(client
->info
.flags
& NBD_FLAG_SEND_FLUSH
)) {
316 return nbd_co_request(bs
, &request
, NULL
);
319 int nbd_client_co_pdiscard(BlockDriverState
*bs
, int64_t offset
, int bytes
)
321 NBDClientSession
*client
= nbd_get_client_session(bs
);
322 NBDRequest request
= {
323 .type
= NBD_CMD_TRIM
,
328 if (!(client
->info
.flags
& NBD_FLAG_SEND_TRIM
)) {
332 return nbd_co_request(bs
, &request
, NULL
);
335 void nbd_client_detach_aio_context(BlockDriverState
*bs
)
337 NBDClientSession
*client
= nbd_get_client_session(bs
);
338 qio_channel_detach_aio_context(QIO_CHANNEL(client
->ioc
));
341 void nbd_client_attach_aio_context(BlockDriverState
*bs
,
342 AioContext
*new_context
)
344 NBDClientSession
*client
= nbd_get_client_session(bs
);
345 qio_channel_attach_aio_context(QIO_CHANNEL(client
->ioc
), new_context
);
346 aio_co_schedule(new_context
, client
->read_reply_co
);
349 void nbd_client_close(BlockDriverState
*bs
)
351 NBDClientSession
*client
= nbd_get_client_session(bs
);
352 NBDRequest request
= { .type
= NBD_CMD_DISC
};
354 if (client
->ioc
== NULL
) {
358 nbd_send_request(client
->ioc
, &request
);
360 nbd_teardown_connection(bs
);
363 int nbd_client_init(BlockDriverState
*bs
,
364 QIOChannelSocket
*sioc
,
366 QCryptoTLSCreds
*tlscreds
,
367 const char *hostname
,
370 NBDClientSession
*client
= nbd_get_client_session(bs
);
374 logout("session init %s\n", export
);
375 qio_channel_set_blocking(QIO_CHANNEL(sioc
), true, NULL
);
377 client
->info
.request_sizes
= true;
378 ret
= nbd_receive_negotiate(QIO_CHANNEL(sioc
), export
,
380 &client
->ioc
, &client
->info
, errp
);
382 logout("Failed to negotiate with the NBD server\n");
385 if (client
->info
.flags
& NBD_FLAG_SEND_FUA
) {
386 bs
->supported_write_flags
= BDRV_REQ_FUA
;
387 bs
->supported_zero_flags
|= BDRV_REQ_FUA
;
389 if (client
->info
.flags
& NBD_FLAG_SEND_WRITE_ZEROES
) {
390 bs
->supported_zero_flags
|= BDRV_REQ_MAY_UNMAP
;
392 if (client
->info
.min_block
> bs
->bl
.request_alignment
) {
393 bs
->bl
.request_alignment
= client
->info
.min_block
;
396 qemu_co_mutex_init(&client
->send_mutex
);
397 qemu_co_queue_init(&client
->free_sema
);
399 object_ref(OBJECT(client
->sioc
));
402 client
->ioc
= QIO_CHANNEL(sioc
);
403 object_ref(OBJECT(client
->ioc
));
406 /* Now that we're connected, set the socket to be non-blocking and
407 * kick the reply mechanism. */
408 qio_channel_set_blocking(QIO_CHANNEL(sioc
), false, NULL
);
409 client
->read_reply_co
= qemu_coroutine_create(nbd_read_reply_entry
, client
);
410 nbd_client_attach_aio_context(bs
, bdrv_get_aio_context(bs
));
412 logout("Established connection with NBD server\n");