2 * QEMU Block driver for NBD
4 * Copyright (c) 2019 Virtuozzo International GmbH.
5 * Copyright (C) 2016 Red Hat, Inc.
6 * Copyright (C) 2008 Bull S.A.S.
7 * Author: Laurent Vivier <Laurent.Vivier@bull.net>
10 * Copyright (C) 2007 Anthony Liguori <anthony@codemonkey.ws>
12 * Permission is hereby granted, free of charge, to any person obtaining a copy
13 * of this software and associated documentation files (the "Software"), to deal
14 * in the Software without restriction, including without limitation the rights
15 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
16 * copies of the Software, and to permit persons to whom the Software is
17 * furnished to do so, subject to the following conditions:
19 * The above copyright notice and this permission notice shall be included in
20 * all copies or substantial portions of the Software.
22 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
23 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
24 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
25 * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
26 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
27 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
31 #include "qemu/osdep.h"
35 #include "qemu/option.h"
36 #include "qemu/cutils.h"
37 #include "qemu/main-loop.h"
38 #include "qemu/atomic.h"
40 #include "qapi/qapi-visit-sockets.h"
41 #include "qapi/qmp/qstring.h"
42 #include "qapi/clone-visitor.h"
44 #include "block/qdict.h"
45 #include "block/nbd.h"
46 #include "block/block_int.h"
47 #include "block/coroutines.h"
49 #include "qemu/yank.h"
51 #define EN_OPTSTR ":exportname="
52 #define MAX_NBD_REQUESTS 16
54 #define HANDLE_TO_INDEX(bs, handle) ((handle) ^ (uint64_t)(intptr_t)(bs))
55 #define INDEX_TO_HANDLE(bs, index) ((index) ^ (uint64_t)(intptr_t)(bs))
59 uint64_t offset
; /* original offset of the request */
60 bool receiving
; /* waiting for connection_co? */
63 typedef enum NBDClientState
{
64 NBD_CLIENT_CONNECTING_WAIT
,
65 NBD_CLIENT_CONNECTING_NOWAIT
,
70 typedef struct BDRVNBDState
{
71 QIOChannel
*ioc
; /* The current I/O channel */
76 Coroutine
*connection_co
;
77 Coroutine
*teardown_co
;
78 QemuCoSleep reconnect_sleep
;
80 bool wait_drained_end
;
85 QEMUTimer
*reconnect_delay_timer
;
87 NBDClientRequest requests
[MAX_NBD_REQUESTS
];
91 /* Connection parameters */
92 uint32_t reconnect_delay
;
94 char *export
, *tlscredsid
;
95 QCryptoTLSCreds
*tlscreds
;
100 NBDClientConnection
*conn
;
103 static void nbd_yank(void *opaque
);
105 static void nbd_clear_bdrvstate(BlockDriverState
*bs
)
107 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
109 nbd_client_connection_release(s
->conn
);
112 yank_unregister_instance(BLOCKDEV_YANK_INSTANCE(bs
->node_name
));
114 object_unref(OBJECT(s
->tlscreds
));
115 qapi_free_SocketAddress(s
->saddr
);
119 g_free(s
->tlscredsid
);
120 s
->tlscredsid
= NULL
;
121 g_free(s
->x_dirty_bitmap
);
122 s
->x_dirty_bitmap
= NULL
;
125 static bool nbd_client_connected(BDRVNBDState
*s
)
127 return qatomic_load_acquire(&s
->state
) == NBD_CLIENT_CONNECTED
;
130 static bool nbd_recv_coroutine_wake_one(NBDClientRequest
*req
)
132 if (req
->receiving
) {
133 req
->receiving
= false;
134 aio_co_wake(req
->coroutine
);
141 static void nbd_recv_coroutines_wake(BDRVNBDState
*s
, bool all
)
145 for (i
= 0; i
< MAX_NBD_REQUESTS
; i
++) {
146 if (nbd_recv_coroutine_wake_one(&s
->requests
[i
]) && !all
) {
152 static void nbd_channel_error(BDRVNBDState
*s
, int ret
)
154 if (nbd_client_connected(s
)) {
155 qio_channel_shutdown(s
->ioc
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
159 if (nbd_client_connected(s
)) {
160 s
->state
= s
->reconnect_delay
? NBD_CLIENT_CONNECTING_WAIT
:
161 NBD_CLIENT_CONNECTING_NOWAIT
;
164 s
->state
= NBD_CLIENT_QUIT
;
168 static void reconnect_delay_timer_del(BDRVNBDState
*s
)
170 if (s
->reconnect_delay_timer
) {
171 timer_free(s
->reconnect_delay_timer
);
172 s
->reconnect_delay_timer
= NULL
;
176 static void reconnect_delay_timer_cb(void *opaque
)
178 BDRVNBDState
*s
= opaque
;
180 if (qatomic_load_acquire(&s
->state
) == NBD_CLIENT_CONNECTING_WAIT
) {
181 s
->state
= NBD_CLIENT_CONNECTING_NOWAIT
;
182 while (qemu_co_enter_next(&s
->free_sema
, NULL
)) {
183 /* Resume all queued requests */
187 reconnect_delay_timer_del(s
);
190 static void reconnect_delay_timer_init(BDRVNBDState
*s
, uint64_t expire_time_ns
)
192 if (qatomic_load_acquire(&s
->state
) != NBD_CLIENT_CONNECTING_WAIT
) {
196 assert(!s
->reconnect_delay_timer
);
197 s
->reconnect_delay_timer
= aio_timer_new(bdrv_get_aio_context(s
->bs
),
200 reconnect_delay_timer_cb
, s
);
201 timer_mod(s
->reconnect_delay_timer
, expire_time_ns
);
204 static void nbd_client_detach_aio_context(BlockDriverState
*bs
)
206 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
208 /* Timer is deleted in nbd_client_co_drain_begin() */
209 assert(!s
->reconnect_delay_timer
);
211 * If reconnect is in progress we may have no ->ioc. It will be
212 * re-instantiated in the proper aio context once the connection is
216 qio_channel_detach_aio_context(QIO_CHANNEL(s
->ioc
));
220 static void nbd_client_attach_aio_context_bh(void *opaque
)
222 BlockDriverState
*bs
= opaque
;
223 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
225 if (s
->connection_co
) {
227 * The node is still drained, so we know the coroutine has yielded in
228 * nbd_read_eof(), the only place where bs->in_flight can reach 0, or
229 * it is entered for the first time. Both places are safe for entering
232 qemu_aio_coroutine_enter(bs
->aio_context
, s
->connection_co
);
234 bdrv_dec_in_flight(bs
);
237 static void nbd_client_attach_aio_context(BlockDriverState
*bs
,
238 AioContext
*new_context
)
240 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
243 * s->connection_co is either yielded from nbd_receive_reply or from
244 * nbd_co_reconnect_loop()
246 if (nbd_client_connected(s
)) {
247 qio_channel_attach_aio_context(QIO_CHANNEL(s
->ioc
), new_context
);
250 bdrv_inc_in_flight(bs
);
253 * Need to wait here for the BH to run because the BH must run while the
254 * node is still drained.
256 aio_wait_bh_oneshot(new_context
, nbd_client_attach_aio_context_bh
, bs
);
259 static void coroutine_fn
nbd_client_co_drain_begin(BlockDriverState
*bs
)
261 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
264 qemu_co_sleep_wake(&s
->reconnect_sleep
);
266 nbd_co_establish_connection_cancel(s
->conn
);
268 reconnect_delay_timer_del(s
);
270 if (qatomic_load_acquire(&s
->state
) == NBD_CLIENT_CONNECTING_WAIT
) {
271 s
->state
= NBD_CLIENT_CONNECTING_NOWAIT
;
272 qemu_co_queue_restart_all(&s
->free_sema
);
276 static void coroutine_fn
nbd_client_co_drain_end(BlockDriverState
*bs
)
278 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
281 if (s
->wait_drained_end
) {
282 s
->wait_drained_end
= false;
283 aio_co_wake(s
->connection_co
);
288 static void nbd_teardown_connection(BlockDriverState
*bs
)
290 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
293 /* finish any pending coroutines */
294 qio_channel_shutdown(s
->ioc
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
297 s
->state
= NBD_CLIENT_QUIT
;
298 if (s
->connection_co
) {
299 qemu_co_sleep_wake(&s
->reconnect_sleep
);
300 nbd_co_establish_connection_cancel(s
->conn
);
302 if (qemu_in_coroutine()) {
303 s
->teardown_co
= qemu_coroutine_self();
304 /* connection_co resumes us when it terminates */
305 qemu_coroutine_yield();
306 s
->teardown_co
= NULL
;
308 BDRV_POLL_WHILE(bs
, s
->connection_co
);
310 assert(!s
->connection_co
);
313 static bool nbd_client_connecting(BDRVNBDState
*s
)
315 NBDClientState state
= qatomic_load_acquire(&s
->state
);
316 return state
== NBD_CLIENT_CONNECTING_WAIT
||
317 state
== NBD_CLIENT_CONNECTING_NOWAIT
;
320 static bool nbd_client_connecting_wait(BDRVNBDState
*s
)
322 return qatomic_load_acquire(&s
->state
) == NBD_CLIENT_CONNECTING_WAIT
;
326 * Update @bs with information learned during a completed negotiation process.
327 * Return failure if the server's advertised options are incompatible with the
330 static int nbd_handle_updated_info(BlockDriverState
*bs
, Error
**errp
)
332 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
335 if (s
->x_dirty_bitmap
) {
336 if (!s
->info
.base_allocation
) {
337 error_setg(errp
, "requested x-dirty-bitmap %s not found",
341 if (strcmp(s
->x_dirty_bitmap
, "qemu:allocation-depth") == 0) {
342 s
->alloc_depth
= true;
346 if (s
->info
.flags
& NBD_FLAG_READ_ONLY
) {
347 ret
= bdrv_apply_auto_read_only(bs
, "NBD export is read-only", errp
);
353 if (s
->info
.flags
& NBD_FLAG_SEND_FUA
) {
354 bs
->supported_write_flags
= BDRV_REQ_FUA
;
355 bs
->supported_zero_flags
|= BDRV_REQ_FUA
;
358 if (s
->info
.flags
& NBD_FLAG_SEND_WRITE_ZEROES
) {
359 bs
->supported_zero_flags
|= BDRV_REQ_MAY_UNMAP
;
360 if (s
->info
.flags
& NBD_FLAG_SEND_FAST_ZERO
) {
361 bs
->supported_zero_flags
|= BDRV_REQ_NO_FALLBACK
;
365 trace_nbd_client_handshake_success(s
->export
);
370 int coroutine_fn
nbd_co_do_establish_connection(BlockDriverState
*bs
,
373 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
378 s
->ioc
= nbd_co_establish_connection(s
->conn
, &s
->info
, true, errp
);
380 return -ECONNREFUSED
;
383 yank_register_function(BLOCKDEV_YANK_INSTANCE(s
->bs
->node_name
), nbd_yank
,
386 ret
= nbd_handle_updated_info(s
->bs
, NULL
);
389 * We have connected, but must fail for other reasons.
390 * Send NBD_CMD_DISC as a courtesy to the server.
392 NBDRequest request
= { .type
= NBD_CMD_DISC
};
394 nbd_send_request(s
->ioc
, &request
);
396 yank_unregister_function(BLOCKDEV_YANK_INSTANCE(s
->bs
->node_name
),
398 object_unref(OBJECT(s
->ioc
));
404 qio_channel_set_blocking(s
->ioc
, false, NULL
);
405 qio_channel_attach_aio_context(s
->ioc
, bdrv_get_aio_context(bs
));
407 /* successfully connected */
408 s
->state
= NBD_CLIENT_CONNECTED
;
409 qemu_co_queue_restart_all(&s
->free_sema
);
414 static coroutine_fn
void nbd_reconnect_attempt(BDRVNBDState
*s
)
416 if (!nbd_client_connecting(s
)) {
420 /* Wait for completion of all in-flight requests */
422 qemu_co_mutex_lock(&s
->send_mutex
);
424 while (s
->in_flight
> 0) {
425 qemu_co_mutex_unlock(&s
->send_mutex
);
426 nbd_recv_coroutines_wake(s
, true);
427 s
->wait_in_flight
= true;
428 qemu_coroutine_yield();
429 s
->wait_in_flight
= false;
430 qemu_co_mutex_lock(&s
->send_mutex
);
433 qemu_co_mutex_unlock(&s
->send_mutex
);
435 if (!nbd_client_connecting(s
)) {
440 * Now we are sure that nobody is accessing the channel, and no one will
441 * try until we set the state to CONNECTED.
444 /* Finalize previous connection if any */
446 qio_channel_detach_aio_context(QIO_CHANNEL(s
->ioc
));
447 yank_unregister_function(BLOCKDEV_YANK_INSTANCE(s
->bs
->node_name
),
449 object_unref(OBJECT(s
->ioc
));
453 nbd_co_do_establish_connection(s
->bs
, NULL
);
456 static coroutine_fn
void nbd_co_reconnect_loop(BDRVNBDState
*s
)
458 uint64_t timeout
= 1 * NANOSECONDS_PER_SECOND
;
459 uint64_t max_timeout
= 16 * NANOSECONDS_PER_SECOND
;
461 if (qatomic_load_acquire(&s
->state
) == NBD_CLIENT_CONNECTING_WAIT
) {
462 reconnect_delay_timer_init(s
, qemu_clock_get_ns(QEMU_CLOCK_REALTIME
) +
463 s
->reconnect_delay
* NANOSECONDS_PER_SECOND
);
466 nbd_reconnect_attempt(s
);
468 while (nbd_client_connecting(s
)) {
470 bdrv_dec_in_flight(s
->bs
);
471 s
->wait_drained_end
= true;
474 * We may be entered once from nbd_client_attach_aio_context_bh
475 * and then from nbd_client_co_drain_end. So here is a loop.
477 qemu_coroutine_yield();
479 bdrv_inc_in_flight(s
->bs
);
481 qemu_co_sleep_ns_wakeable(&s
->reconnect_sleep
,
482 QEMU_CLOCK_REALTIME
, timeout
);
486 if (timeout
< max_timeout
) {
491 nbd_reconnect_attempt(s
);
494 reconnect_delay_timer_del(s
);
497 static coroutine_fn
void nbd_connection_entry(void *opaque
)
499 BDRVNBDState
*s
= opaque
;
502 Error
*local_err
= NULL
;
504 while (qatomic_load_acquire(&s
->state
) != NBD_CLIENT_QUIT
) {
506 * The NBD client can only really be considered idle when it has
507 * yielded from qio_channel_readv_all_eof(), waiting for data. This is
508 * the point where the additional scheduled coroutine entry happens
509 * after nbd_client_attach_aio_context().
511 * Therefore we keep an additional in_flight reference all the time and
512 * only drop it temporarily here.
515 if (nbd_client_connecting(s
)) {
516 nbd_co_reconnect_loop(s
);
519 if (!nbd_client_connected(s
)) {
523 assert(s
->reply
.handle
== 0);
524 ret
= nbd_receive_reply(s
->bs
, s
->ioc
, &s
->reply
, &local_err
);
527 trace_nbd_read_reply_entry_fail(ret
, error_get_pretty(local_err
));
528 error_free(local_err
);
532 nbd_channel_error(s
, ret
? ret
: -EIO
);
537 * There's no need for a mutex on the receive side, because the
538 * handler acts as a synchronization point and ensures that only
539 * one coroutine is called until the reply finishes.
541 i
= HANDLE_TO_INDEX(s
, s
->reply
.handle
);
542 if (i
>= MAX_NBD_REQUESTS
||
543 !s
->requests
[i
].coroutine
||
544 !s
->requests
[i
].receiving
||
545 (nbd_reply_is_structured(&s
->reply
) && !s
->info
.structured_reply
))
547 nbd_channel_error(s
, -EINVAL
);
552 * We're woken up again by the request itself. Note that there
553 * is no race between yielding and reentering connection_co. This
556 * - if the request runs on the same AioContext, it is only
557 * entered after we yield
559 * - if the request runs on a different AioContext, reentering
560 * connection_co happens through a bottom half, which can only
561 * run after we yield.
563 s
->requests
[i
].receiving
= false;
564 aio_co_wake(s
->requests
[i
].coroutine
);
565 qemu_coroutine_yield();
568 qemu_co_queue_restart_all(&s
->free_sema
);
569 nbd_recv_coroutines_wake(s
, true);
570 bdrv_dec_in_flight(s
->bs
);
572 s
->connection_co
= NULL
;
574 qio_channel_detach_aio_context(QIO_CHANNEL(s
->ioc
));
575 yank_unregister_function(BLOCKDEV_YANK_INSTANCE(s
->bs
->node_name
),
577 object_unref(OBJECT(s
->ioc
));
581 if (s
->teardown_co
) {
582 aio_co_wake(s
->teardown_co
);
587 static int nbd_co_send_request(BlockDriverState
*bs
,
591 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
594 qemu_co_mutex_lock(&s
->send_mutex
);
595 while (s
->in_flight
== MAX_NBD_REQUESTS
|| nbd_client_connecting_wait(s
)) {
596 qemu_co_queue_wait(&s
->free_sema
, &s
->send_mutex
);
599 if (!nbd_client_connected(s
)) {
606 for (i
= 0; i
< MAX_NBD_REQUESTS
; i
++) {
607 if (s
->requests
[i
].coroutine
== NULL
) {
612 g_assert(qemu_in_coroutine());
613 assert(i
< MAX_NBD_REQUESTS
);
615 s
->requests
[i
].coroutine
= qemu_coroutine_self();
616 s
->requests
[i
].offset
= request
->from
;
617 s
->requests
[i
].receiving
= false;
619 request
->handle
= INDEX_TO_HANDLE(s
, i
);
624 qio_channel_set_cork(s
->ioc
, true);
625 rc
= nbd_send_request(s
->ioc
, request
);
626 if (nbd_client_connected(s
) && rc
>= 0) {
627 if (qio_channel_writev_all(s
->ioc
, qiov
->iov
, qiov
->niov
,
631 } else if (rc
>= 0) {
634 qio_channel_set_cork(s
->ioc
, false);
636 rc
= nbd_send_request(s
->ioc
, request
);
641 nbd_channel_error(s
, rc
);
643 s
->requests
[i
].coroutine
= NULL
;
646 if (s
->in_flight
== 0 && s
->wait_in_flight
) {
647 aio_co_wake(s
->connection_co
);
649 qemu_co_queue_next(&s
->free_sema
);
652 qemu_co_mutex_unlock(&s
->send_mutex
);
656 static inline uint16_t payload_advance16(uint8_t **payload
)
659 return lduw_be_p(*payload
- 2);
662 static inline uint32_t payload_advance32(uint8_t **payload
)
665 return ldl_be_p(*payload
- 4);
668 static inline uint64_t payload_advance64(uint8_t **payload
)
671 return ldq_be_p(*payload
- 8);
674 static int nbd_parse_offset_hole_payload(BDRVNBDState
*s
,
675 NBDStructuredReplyChunk
*chunk
,
676 uint8_t *payload
, uint64_t orig_offset
,
677 QEMUIOVector
*qiov
, Error
**errp
)
682 if (chunk
->length
!= sizeof(offset
) + sizeof(hole_size
)) {
683 error_setg(errp
, "Protocol error: invalid payload for "
684 "NBD_REPLY_TYPE_OFFSET_HOLE");
688 offset
= payload_advance64(&payload
);
689 hole_size
= payload_advance32(&payload
);
691 if (!hole_size
|| offset
< orig_offset
|| hole_size
> qiov
->size
||
692 offset
> orig_offset
+ qiov
->size
- hole_size
) {
693 error_setg(errp
, "Protocol error: server sent chunk exceeding requested"
697 if (s
->info
.min_block
&&
698 !QEMU_IS_ALIGNED(hole_size
, s
->info
.min_block
)) {
699 trace_nbd_structured_read_compliance("hole");
702 qemu_iovec_memset(qiov
, offset
- orig_offset
, 0, hole_size
);
708 * nbd_parse_blockstatus_payload
709 * Based on our request, we expect only one extent in reply, for the
710 * base:allocation context.
712 static int nbd_parse_blockstatus_payload(BDRVNBDState
*s
,
713 NBDStructuredReplyChunk
*chunk
,
714 uint8_t *payload
, uint64_t orig_length
,
715 NBDExtent
*extent
, Error
**errp
)
719 /* The server succeeded, so it must have sent [at least] one extent */
720 if (chunk
->length
< sizeof(context_id
) + sizeof(*extent
)) {
721 error_setg(errp
, "Protocol error: invalid payload for "
722 "NBD_REPLY_TYPE_BLOCK_STATUS");
726 context_id
= payload_advance32(&payload
);
727 if (s
->info
.context_id
!= context_id
) {
728 error_setg(errp
, "Protocol error: unexpected context id %d for "
729 "NBD_REPLY_TYPE_BLOCK_STATUS, when negotiated context "
730 "id is %d", context_id
,
735 extent
->length
= payload_advance32(&payload
);
736 extent
->flags
= payload_advance32(&payload
);
738 if (extent
->length
== 0) {
739 error_setg(errp
, "Protocol error: server sent status chunk with "
745 * A server sending unaligned block status is in violation of the
746 * protocol, but as qemu-nbd 3.1 is such a server (at least for
747 * POSIX files that are not a multiple of 512 bytes, since qemu
748 * rounds files up to 512-byte multiples but lseek(SEEK_HOLE)
749 * still sees an implicit hole beyond the real EOF), it's nicer to
750 * work around the misbehaving server. If the request included
751 * more than the final unaligned block, truncate it back to an
752 * aligned result; if the request was only the final block, round
753 * up to the full block and change the status to fully-allocated
754 * (always a safe status, even if it loses information).
756 if (s
->info
.min_block
&& !QEMU_IS_ALIGNED(extent
->length
,
757 s
->info
.min_block
)) {
758 trace_nbd_parse_blockstatus_compliance("extent length is unaligned");
759 if (extent
->length
> s
->info
.min_block
) {
760 extent
->length
= QEMU_ALIGN_DOWN(extent
->length
,
763 extent
->length
= s
->info
.min_block
;
769 * We used NBD_CMD_FLAG_REQ_ONE, so the server should not have
770 * sent us any more than one extent, nor should it have included
771 * status beyond our request in that extent. However, it's easy
772 * enough to ignore the server's noncompliance without killing the
773 * connection; just ignore trailing extents, and clamp things to
774 * the length of our request.
776 if (chunk
->length
> sizeof(context_id
) + sizeof(*extent
)) {
777 trace_nbd_parse_blockstatus_compliance("more than one extent");
779 if (extent
->length
> orig_length
) {
780 extent
->length
= orig_length
;
781 trace_nbd_parse_blockstatus_compliance("extent length too large");
785 * HACK: if we are using x-dirty-bitmaps to access
786 * qemu:allocation-depth, treat all depths > 2 the same as 2,
787 * since nbd_client_co_block_status is only expecting the low two
790 if (s
->alloc_depth
&& extent
->flags
> 2) {
798 * nbd_parse_error_payload
799 * on success @errp contains message describing nbd error reply
801 static int nbd_parse_error_payload(NBDStructuredReplyChunk
*chunk
,
802 uint8_t *payload
, int *request_ret
,
806 uint16_t message_size
;
808 assert(chunk
->type
& (1 << 15));
810 if (chunk
->length
< sizeof(error
) + sizeof(message_size
)) {
812 "Protocol error: invalid payload for structured error");
816 error
= nbd_errno_to_system_errno(payload_advance32(&payload
));
818 error_setg(errp
, "Protocol error: server sent structured error chunk "
823 *request_ret
= -error
;
824 message_size
= payload_advance16(&payload
);
826 if (message_size
> chunk
->length
- sizeof(error
) - sizeof(message_size
)) {
827 error_setg(errp
, "Protocol error: server sent structured error chunk "
828 "with incorrect message size");
832 /* TODO: Add a trace point to mention the server complaint */
834 /* TODO handle ERROR_OFFSET */
839 static int nbd_co_receive_offset_data_payload(BDRVNBDState
*s
,
840 uint64_t orig_offset
,
841 QEMUIOVector
*qiov
, Error
**errp
)
843 QEMUIOVector sub_qiov
;
847 NBDStructuredReplyChunk
*chunk
= &s
->reply
.structured
;
849 assert(nbd_reply_is_structured(&s
->reply
));
851 /* The NBD spec requires at least one byte of payload */
852 if (chunk
->length
<= sizeof(offset
)) {
853 error_setg(errp
, "Protocol error: invalid payload for "
854 "NBD_REPLY_TYPE_OFFSET_DATA");
858 if (nbd_read64(s
->ioc
, &offset
, "OFFSET_DATA offset", errp
) < 0) {
862 data_size
= chunk
->length
- sizeof(offset
);
864 if (offset
< orig_offset
|| data_size
> qiov
->size
||
865 offset
> orig_offset
+ qiov
->size
- data_size
) {
866 error_setg(errp
, "Protocol error: server sent chunk exceeding requested"
870 if (s
->info
.min_block
&& !QEMU_IS_ALIGNED(data_size
, s
->info
.min_block
)) {
871 trace_nbd_structured_read_compliance("data");
874 qemu_iovec_init(&sub_qiov
, qiov
->niov
);
875 qemu_iovec_concat(&sub_qiov
, qiov
, offset
- orig_offset
, data_size
);
876 ret
= qio_channel_readv_all(s
->ioc
, sub_qiov
.iov
, sub_qiov
.niov
, errp
);
877 qemu_iovec_destroy(&sub_qiov
);
879 return ret
< 0 ? -EIO
: 0;
882 #define NBD_MAX_MALLOC_PAYLOAD 1000
883 static coroutine_fn
int nbd_co_receive_structured_payload(
884 BDRVNBDState
*s
, void **payload
, Error
**errp
)
889 assert(nbd_reply_is_structured(&s
->reply
));
891 len
= s
->reply
.structured
.length
;
897 if (payload
== NULL
) {
898 error_setg(errp
, "Unexpected structured payload");
902 if (len
> NBD_MAX_MALLOC_PAYLOAD
) {
903 error_setg(errp
, "Payload too large");
907 *payload
= g_new(char, len
);
908 ret
= nbd_read(s
->ioc
, *payload
, len
, "structured payload", errp
);
919 * nbd_co_do_receive_one_chunk
921 * set request_ret to received reply error
922 * if qiov is not NULL: read payload to @qiov
923 * for structured reply chunk:
924 * if error chunk: read payload, set @request_ret, do not set @payload
925 * else if offset_data chunk: read payload data to @qiov, do not set @payload
926 * else: read payload to @payload
928 * If function fails, @errp contains corresponding error message, and the
929 * connection with the server is suspect. If it returns 0, then the
930 * transaction succeeded (although @request_ret may be a negative errno
931 * corresponding to the server's error reply), and errp is unchanged.
933 static coroutine_fn
int nbd_co_do_receive_one_chunk(
934 BDRVNBDState
*s
, uint64_t handle
, bool only_structured
,
935 int *request_ret
, QEMUIOVector
*qiov
, void **payload
, Error
**errp
)
938 int i
= HANDLE_TO_INDEX(s
, handle
);
939 void *local_payload
= NULL
;
940 NBDStructuredReplyChunk
*chunk
;
947 /* Wait until we're woken up by nbd_connection_entry. */
948 s
->requests
[i
].receiving
= true;
949 qemu_coroutine_yield();
950 assert(!s
->requests
[i
].receiving
);
951 if (!nbd_client_connected(s
)) {
952 error_setg(errp
, "Connection closed");
957 assert(s
->reply
.handle
== handle
);
959 if (nbd_reply_is_simple(&s
->reply
)) {
960 if (only_structured
) {
961 error_setg(errp
, "Protocol error: simple reply when structured "
962 "reply chunk was expected");
966 *request_ret
= -nbd_errno_to_system_errno(s
->reply
.simple
.error
);
967 if (*request_ret
< 0 || !qiov
) {
971 return qio_channel_readv_all(s
->ioc
, qiov
->iov
, qiov
->niov
,
972 errp
) < 0 ? -EIO
: 0;
975 /* handle structured reply chunk */
976 assert(s
->info
.structured_reply
);
977 chunk
= &s
->reply
.structured
;
979 if (chunk
->type
== NBD_REPLY_TYPE_NONE
) {
980 if (!(chunk
->flags
& NBD_REPLY_FLAG_DONE
)) {
981 error_setg(errp
, "Protocol error: NBD_REPLY_TYPE_NONE chunk without"
982 " NBD_REPLY_FLAG_DONE flag set");
986 error_setg(errp
, "Protocol error: NBD_REPLY_TYPE_NONE chunk with"
993 if (chunk
->type
== NBD_REPLY_TYPE_OFFSET_DATA
) {
995 error_setg(errp
, "Unexpected NBD_REPLY_TYPE_OFFSET_DATA chunk");
999 return nbd_co_receive_offset_data_payload(s
, s
->requests
[i
].offset
,
1003 if (nbd_reply_type_is_error(chunk
->type
)) {
1004 payload
= &local_payload
;
1007 ret
= nbd_co_receive_structured_payload(s
, payload
, errp
);
1012 if (nbd_reply_type_is_error(chunk
->type
)) {
1013 ret
= nbd_parse_error_payload(chunk
, local_payload
, request_ret
, errp
);
1014 g_free(local_payload
);
1022 * nbd_co_receive_one_chunk
1023 * Read reply, wake up connection_co and set s->quit if needed.
1024 * Return value is a fatal error code or normal nbd reply error code
1026 static coroutine_fn
int nbd_co_receive_one_chunk(
1027 BDRVNBDState
*s
, uint64_t handle
, bool only_structured
,
1028 int *request_ret
, QEMUIOVector
*qiov
, NBDReply
*reply
, void **payload
,
1031 int ret
= nbd_co_do_receive_one_chunk(s
, handle
, only_structured
,
1032 request_ret
, qiov
, payload
, errp
);
1035 memset(reply
, 0, sizeof(*reply
));
1036 nbd_channel_error(s
, ret
);
1038 /* For assert at loop start in nbd_connection_entry */
1041 s
->reply
.handle
= 0;
1043 if (s
->connection_co
&& !s
->wait_in_flight
) {
1045 * We must check s->wait_in_flight, because we may entered by
1046 * nbd_recv_coroutines_wake(), in this case we should not
1047 * wake connection_co here, it will woken by last request.
1049 aio_co_wake(s
->connection_co
);
1055 typedef struct NBDReplyChunkIter
{
1059 bool done
, only_structured
;
1060 } NBDReplyChunkIter
;
1062 static void nbd_iter_channel_error(NBDReplyChunkIter
*iter
,
1063 int ret
, Error
**local_err
)
1065 assert(local_err
&& *local_err
);
1070 error_propagate(&iter
->err
, *local_err
);
1072 error_free(*local_err
);
1078 static void nbd_iter_request_error(NBDReplyChunkIter
*iter
, int ret
)
1082 if (!iter
->request_ret
) {
1083 iter
->request_ret
= ret
;
1088 * NBD_FOREACH_REPLY_CHUNK
1089 * The pointer stored in @payload requires g_free() to free it.
1091 #define NBD_FOREACH_REPLY_CHUNK(s, iter, handle, structured, \
1092 qiov, reply, payload) \
1093 for (iter = (NBDReplyChunkIter) { .only_structured = structured }; \
1094 nbd_reply_chunk_iter_receive(s, &iter, handle, qiov, reply, payload);)
1097 * nbd_reply_chunk_iter_receive
1098 * The pointer stored in @payload requires g_free() to free it.
1100 static bool nbd_reply_chunk_iter_receive(BDRVNBDState
*s
,
1101 NBDReplyChunkIter
*iter
,
1103 QEMUIOVector
*qiov
, NBDReply
*reply
,
1106 int ret
, request_ret
;
1107 NBDReply local_reply
;
1108 NBDStructuredReplyChunk
*chunk
;
1109 Error
*local_err
= NULL
;
1110 if (!nbd_client_connected(s
)) {
1111 error_setg(&local_err
, "Connection closed");
1112 nbd_iter_channel_error(iter
, -EIO
, &local_err
);
1117 /* Previous iteration was last. */
1121 if (reply
== NULL
) {
1122 reply
= &local_reply
;
1125 ret
= nbd_co_receive_one_chunk(s
, handle
, iter
->only_structured
,
1126 &request_ret
, qiov
, reply
, payload
,
1129 nbd_iter_channel_error(iter
, ret
, &local_err
);
1130 } else if (request_ret
< 0) {
1131 nbd_iter_request_error(iter
, request_ret
);
1134 /* Do not execute the body of NBD_FOREACH_REPLY_CHUNK for simple reply. */
1135 if (nbd_reply_is_simple(reply
) || !nbd_client_connected(s
)) {
1139 chunk
= &reply
->structured
;
1140 iter
->only_structured
= true;
1142 if (chunk
->type
== NBD_REPLY_TYPE_NONE
) {
1143 /* NBD_REPLY_FLAG_DONE is already checked in nbd_co_receive_one_chunk */
1144 assert(chunk
->flags
& NBD_REPLY_FLAG_DONE
);
1148 if (chunk
->flags
& NBD_REPLY_FLAG_DONE
) {
1149 /* This iteration is last. */
1153 /* Execute the loop body */
1157 s
->requests
[HANDLE_TO_INDEX(s
, handle
)].coroutine
= NULL
;
1159 qemu_co_mutex_lock(&s
->send_mutex
);
1161 if (s
->in_flight
== 0 && s
->wait_in_flight
) {
1162 aio_co_wake(s
->connection_co
);
1164 qemu_co_queue_next(&s
->free_sema
);
1166 qemu_co_mutex_unlock(&s
->send_mutex
);
1171 static int nbd_co_receive_return_code(BDRVNBDState
*s
, uint64_t handle
,
1172 int *request_ret
, Error
**errp
)
1174 NBDReplyChunkIter iter
;
1176 NBD_FOREACH_REPLY_CHUNK(s
, iter
, handle
, false, NULL
, NULL
, NULL
) {
1177 /* nbd_reply_chunk_iter_receive does all the work */
1180 error_propagate(errp
, iter
.err
);
1181 *request_ret
= iter
.request_ret
;
1185 static int nbd_co_receive_cmdread_reply(BDRVNBDState
*s
, uint64_t handle
,
1186 uint64_t offset
, QEMUIOVector
*qiov
,
1187 int *request_ret
, Error
**errp
)
1189 NBDReplyChunkIter iter
;
1191 void *payload
= NULL
;
1192 Error
*local_err
= NULL
;
1194 NBD_FOREACH_REPLY_CHUNK(s
, iter
, handle
, s
->info
.structured_reply
,
1195 qiov
, &reply
, &payload
)
1198 NBDStructuredReplyChunk
*chunk
= &reply
.structured
;
1200 assert(nbd_reply_is_structured(&reply
));
1202 switch (chunk
->type
) {
1203 case NBD_REPLY_TYPE_OFFSET_DATA
:
1205 * special cased in nbd_co_receive_one_chunk, data is already
1209 case NBD_REPLY_TYPE_OFFSET_HOLE
:
1210 ret
= nbd_parse_offset_hole_payload(s
, &reply
.structured
, payload
,
1211 offset
, qiov
, &local_err
);
1213 nbd_channel_error(s
, ret
);
1214 nbd_iter_channel_error(&iter
, ret
, &local_err
);
1218 if (!nbd_reply_type_is_error(chunk
->type
)) {
1219 /* not allowed reply type */
1220 nbd_channel_error(s
, -EINVAL
);
1221 error_setg(&local_err
,
1222 "Unexpected reply type: %d (%s) for CMD_READ",
1223 chunk
->type
, nbd_reply_type_lookup(chunk
->type
));
1224 nbd_iter_channel_error(&iter
, -EINVAL
, &local_err
);
1232 error_propagate(errp
, iter
.err
);
1233 *request_ret
= iter
.request_ret
;
1237 static int nbd_co_receive_blockstatus_reply(BDRVNBDState
*s
,
1238 uint64_t handle
, uint64_t length
,
1240 int *request_ret
, Error
**errp
)
1242 NBDReplyChunkIter iter
;
1244 void *payload
= NULL
;
1245 Error
*local_err
= NULL
;
1246 bool received
= false;
1248 assert(!extent
->length
);
1249 NBD_FOREACH_REPLY_CHUNK(s
, iter
, handle
, false, NULL
, &reply
, &payload
) {
1251 NBDStructuredReplyChunk
*chunk
= &reply
.structured
;
1253 assert(nbd_reply_is_structured(&reply
));
1255 switch (chunk
->type
) {
1256 case NBD_REPLY_TYPE_BLOCK_STATUS
:
1258 nbd_channel_error(s
, -EINVAL
);
1259 error_setg(&local_err
, "Several BLOCK_STATUS chunks in reply");
1260 nbd_iter_channel_error(&iter
, -EINVAL
, &local_err
);
1264 ret
= nbd_parse_blockstatus_payload(s
, &reply
.structured
,
1265 payload
, length
, extent
,
1268 nbd_channel_error(s
, ret
);
1269 nbd_iter_channel_error(&iter
, ret
, &local_err
);
1273 if (!nbd_reply_type_is_error(chunk
->type
)) {
1274 nbd_channel_error(s
, -EINVAL
);
1275 error_setg(&local_err
,
1276 "Unexpected reply type: %d (%s) "
1277 "for CMD_BLOCK_STATUS",
1278 chunk
->type
, nbd_reply_type_lookup(chunk
->type
));
1279 nbd_iter_channel_error(&iter
, -EINVAL
, &local_err
);
1287 if (!extent
->length
&& !iter
.request_ret
) {
1288 error_setg(&local_err
, "Server did not reply with any status extents");
1289 nbd_iter_channel_error(&iter
, -EIO
, &local_err
);
1292 error_propagate(errp
, iter
.err
);
1293 *request_ret
= iter
.request_ret
;
1297 static int nbd_co_request(BlockDriverState
*bs
, NBDRequest
*request
,
1298 QEMUIOVector
*write_qiov
)
1300 int ret
, request_ret
;
1301 Error
*local_err
= NULL
;
1302 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
1304 assert(request
->type
!= NBD_CMD_READ
);
1306 assert(request
->type
== NBD_CMD_WRITE
);
1307 assert(request
->len
== iov_size(write_qiov
->iov
, write_qiov
->niov
));
1309 assert(request
->type
!= NBD_CMD_WRITE
);
1313 ret
= nbd_co_send_request(bs
, request
, write_qiov
);
1318 ret
= nbd_co_receive_return_code(s
, request
->handle
,
1319 &request_ret
, &local_err
);
1321 trace_nbd_co_request_fail(request
->from
, request
->len
,
1322 request
->handle
, request
->flags
,
1324 nbd_cmd_lookup(request
->type
),
1325 ret
, error_get_pretty(local_err
));
1326 error_free(local_err
);
1329 } while (ret
< 0 && nbd_client_connecting_wait(s
));
1331 return ret
? ret
: request_ret
;
1334 static int nbd_client_co_preadv(BlockDriverState
*bs
, int64_t offset
,
1335 int64_t bytes
, QEMUIOVector
*qiov
,
1336 BdrvRequestFlags flags
)
1338 int ret
, request_ret
;
1339 Error
*local_err
= NULL
;
1340 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
1341 NBDRequest request
= {
1342 .type
= NBD_CMD_READ
,
1347 assert(bytes
<= NBD_MAX_BUFFER_SIZE
);
1354 * Work around the fact that the block layer doesn't do
1355 * byte-accurate sizing yet - if the read exceeds the server's
1356 * advertised size because the block layer rounded size up, then
1357 * truncate the request to the server and tail-pad with zero.
1359 if (offset
>= s
->info
.size
) {
1360 assert(bytes
< BDRV_SECTOR_SIZE
);
1361 qemu_iovec_memset(qiov
, 0, 0, bytes
);
1364 if (offset
+ bytes
> s
->info
.size
) {
1365 uint64_t slop
= offset
+ bytes
- s
->info
.size
;
1367 assert(slop
< BDRV_SECTOR_SIZE
);
1368 qemu_iovec_memset(qiov
, bytes
- slop
, 0, slop
);
1369 request
.len
-= slop
;
1373 ret
= nbd_co_send_request(bs
, &request
, NULL
);
1378 ret
= nbd_co_receive_cmdread_reply(s
, request
.handle
, offset
, qiov
,
1379 &request_ret
, &local_err
);
1381 trace_nbd_co_request_fail(request
.from
, request
.len
, request
.handle
,
1382 request
.flags
, request
.type
,
1383 nbd_cmd_lookup(request
.type
),
1384 ret
, error_get_pretty(local_err
));
1385 error_free(local_err
);
1388 } while (ret
< 0 && nbd_client_connecting_wait(s
));
1390 return ret
? ret
: request_ret
;
1393 static int nbd_client_co_pwritev(BlockDriverState
*bs
, int64_t offset
,
1394 int64_t bytes
, QEMUIOVector
*qiov
,
1395 BdrvRequestFlags flags
)
1397 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
1398 NBDRequest request
= {
1399 .type
= NBD_CMD_WRITE
,
1404 assert(!(s
->info
.flags
& NBD_FLAG_READ_ONLY
));
1405 if (flags
& BDRV_REQ_FUA
) {
1406 assert(s
->info
.flags
& NBD_FLAG_SEND_FUA
);
1407 request
.flags
|= NBD_CMD_FLAG_FUA
;
1410 assert(bytes
<= NBD_MAX_BUFFER_SIZE
);
1415 return nbd_co_request(bs
, &request
, qiov
);
1418 static int nbd_client_co_pwrite_zeroes(BlockDriverState
*bs
, int64_t offset
,
1419 int64_t bytes
, BdrvRequestFlags flags
)
1421 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
1422 NBDRequest request
= {
1423 .type
= NBD_CMD_WRITE_ZEROES
,
1425 .len
= bytes
, /* .len is uint32_t actually */
1428 assert(bytes
<= UINT32_MAX
); /* rely on max_pwrite_zeroes */
1430 assert(!(s
->info
.flags
& NBD_FLAG_READ_ONLY
));
1431 if (!(s
->info
.flags
& NBD_FLAG_SEND_WRITE_ZEROES
)) {
1435 if (flags
& BDRV_REQ_FUA
) {
1436 assert(s
->info
.flags
& NBD_FLAG_SEND_FUA
);
1437 request
.flags
|= NBD_CMD_FLAG_FUA
;
1439 if (!(flags
& BDRV_REQ_MAY_UNMAP
)) {
1440 request
.flags
|= NBD_CMD_FLAG_NO_HOLE
;
1442 if (flags
& BDRV_REQ_NO_FALLBACK
) {
1443 assert(s
->info
.flags
& NBD_FLAG_SEND_FAST_ZERO
);
1444 request
.flags
|= NBD_CMD_FLAG_FAST_ZERO
;
1450 return nbd_co_request(bs
, &request
, NULL
);
1453 static int nbd_client_co_flush(BlockDriverState
*bs
)
1455 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
1456 NBDRequest request
= { .type
= NBD_CMD_FLUSH
};
1458 if (!(s
->info
.flags
& NBD_FLAG_SEND_FLUSH
)) {
1465 return nbd_co_request(bs
, &request
, NULL
);
1468 static int nbd_client_co_pdiscard(BlockDriverState
*bs
, int64_t offset
,
1471 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
1472 NBDRequest request
= {
1473 .type
= NBD_CMD_TRIM
,
1475 .len
= bytes
, /* len is uint32_t */
1478 assert(bytes
<= UINT32_MAX
); /* rely on max_pdiscard */
1480 assert(!(s
->info
.flags
& NBD_FLAG_READ_ONLY
));
1481 if (!(s
->info
.flags
& NBD_FLAG_SEND_TRIM
) || !bytes
) {
1485 return nbd_co_request(bs
, &request
, NULL
);
1488 static int coroutine_fn
nbd_client_co_block_status(
1489 BlockDriverState
*bs
, bool want_zero
, int64_t offset
, int64_t bytes
,
1490 int64_t *pnum
, int64_t *map
, BlockDriverState
**file
)
1492 int ret
, request_ret
;
1493 NBDExtent extent
= { 0 };
1494 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
1495 Error
*local_err
= NULL
;
1497 NBDRequest request
= {
1498 .type
= NBD_CMD_BLOCK_STATUS
,
1500 .len
= MIN(QEMU_ALIGN_DOWN(INT_MAX
, bs
->bl
.request_alignment
),
1501 MIN(bytes
, s
->info
.size
- offset
)),
1502 .flags
= NBD_CMD_FLAG_REQ_ONE
,
1505 if (!s
->info
.base_allocation
) {
1509 return BDRV_BLOCK_DATA
| BDRV_BLOCK_OFFSET_VALID
;
1513 * Work around the fact that the block layer doesn't do
1514 * byte-accurate sizing yet - if the status request exceeds the
1515 * server's advertised size because the block layer rounded size
1516 * up, we truncated the request to the server (above), or are
1517 * called on just the hole.
1519 if (offset
>= s
->info
.size
) {
1521 assert(bytes
< BDRV_SECTOR_SIZE
);
1522 /* Intentionally don't report offset_valid for the hole */
1523 return BDRV_BLOCK_ZERO
;
1526 if (s
->info
.min_block
) {
1527 assert(QEMU_IS_ALIGNED(request
.len
, s
->info
.min_block
));
1530 ret
= nbd_co_send_request(bs
, &request
, NULL
);
1535 ret
= nbd_co_receive_blockstatus_reply(s
, request
.handle
, bytes
,
1536 &extent
, &request_ret
,
1539 trace_nbd_co_request_fail(request
.from
, request
.len
, request
.handle
,
1540 request
.flags
, request
.type
,
1541 nbd_cmd_lookup(request
.type
),
1542 ret
, error_get_pretty(local_err
));
1543 error_free(local_err
);
1546 } while (ret
< 0 && nbd_client_connecting_wait(s
));
1548 if (ret
< 0 || request_ret
< 0) {
1549 return ret
? ret
: request_ret
;
1552 assert(extent
.length
);
1553 *pnum
= extent
.length
;
1556 return (extent
.flags
& NBD_STATE_HOLE
? 0 : BDRV_BLOCK_DATA
) |
1557 (extent
.flags
& NBD_STATE_ZERO
? BDRV_BLOCK_ZERO
: 0) |
1558 BDRV_BLOCK_OFFSET_VALID
;
1561 static int nbd_client_reopen_prepare(BDRVReopenState
*state
,
1562 BlockReopenQueue
*queue
, Error
**errp
)
1564 BDRVNBDState
*s
= (BDRVNBDState
*)state
->bs
->opaque
;
1566 if ((state
->flags
& BDRV_O_RDWR
) && (s
->info
.flags
& NBD_FLAG_READ_ONLY
)) {
1567 error_setg(errp
, "Can't reopen read-only NBD mount as read/write");
1573 static void nbd_yank(void *opaque
)
1575 BlockDriverState
*bs
= opaque
;
1576 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
1578 qatomic_store_release(&s
->state
, NBD_CLIENT_QUIT
);
1579 qio_channel_shutdown(QIO_CHANNEL(s
->ioc
), QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
1582 static void nbd_client_close(BlockDriverState
*bs
)
1584 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
1585 NBDRequest request
= { .type
= NBD_CMD_DISC
};
1588 nbd_send_request(s
->ioc
, &request
);
1591 nbd_teardown_connection(bs
);
1596 * Parse nbd_open options
1599 static int nbd_parse_uri(const char *filename
, QDict
*options
)
1603 QueryParams
*qp
= NULL
;
1607 uri
= uri_parse(filename
);
1613 if (!g_strcmp0(uri
->scheme
, "nbd")) {
1615 } else if (!g_strcmp0(uri
->scheme
, "nbd+tcp")) {
1617 } else if (!g_strcmp0(uri
->scheme
, "nbd+unix")) {
1624 p
= uri
->path
? uri
->path
: "";
1629 qdict_put_str(options
, "export", p
);
1632 qp
= query_params_parse(uri
->query
);
1633 if (qp
->n
> 1 || (is_unix
&& !qp
->n
) || (!is_unix
&& qp
->n
)) {
1639 /* nbd+unix:///export?socket=path */
1640 if (uri
->server
|| uri
->port
|| strcmp(qp
->p
[0].name
, "socket")) {
1644 qdict_put_str(options
, "server.type", "unix");
1645 qdict_put_str(options
, "server.path", qp
->p
[0].value
);
1650 /* nbd[+tcp]://host[:port]/export */
1656 /* strip braces from literal IPv6 address */
1657 if (uri
->server
[0] == '[') {
1658 host
= qstring_from_substr(uri
->server
, 1,
1659 strlen(uri
->server
) - 1);
1661 host
= qstring_from_str(uri
->server
);
1664 qdict_put_str(options
, "server.type", "inet");
1665 qdict_put(options
, "server.host", host
);
1667 port_str
= g_strdup_printf("%d", uri
->port
?: NBD_DEFAULT_PORT
);
1668 qdict_put_str(options
, "server.port", port_str
);
1674 query_params_free(qp
);
1680 static bool nbd_has_filename_options_conflict(QDict
*options
, Error
**errp
)
1682 const QDictEntry
*e
;
1684 for (e
= qdict_first(options
); e
; e
= qdict_next(options
, e
)) {
1685 if (!strcmp(e
->key
, "host") ||
1686 !strcmp(e
->key
, "port") ||
1687 !strcmp(e
->key
, "path") ||
1688 !strcmp(e
->key
, "export") ||
1689 strstart(e
->key
, "server.", NULL
))
1691 error_setg(errp
, "Option '%s' cannot be used with a file name",
1700 static void nbd_parse_filename(const char *filename
, QDict
*options
,
1703 g_autofree
char *file
= NULL
;
1705 const char *host_spec
;
1706 const char *unixpath
;
1708 if (nbd_has_filename_options_conflict(options
, errp
)) {
1712 if (strstr(filename
, "://")) {
1713 int ret
= nbd_parse_uri(filename
, options
);
1715 error_setg(errp
, "No valid URL specified");
1720 file
= g_strdup(filename
);
1722 export_name
= strstr(file
, EN_OPTSTR
);
1724 if (export_name
[strlen(EN_OPTSTR
)] == 0) {
1727 export_name
[0] = 0; /* truncate 'file' */
1728 export_name
+= strlen(EN_OPTSTR
);
1730 qdict_put_str(options
, "export", export_name
);
1733 /* extract the host_spec - fail if it's not nbd:... */
1734 if (!strstart(file
, "nbd:", &host_spec
)) {
1735 error_setg(errp
, "File name string for NBD must start with 'nbd:'");
1743 /* are we a UNIX or TCP socket? */
1744 if (strstart(host_spec
, "unix:", &unixpath
)) {
1745 qdict_put_str(options
, "server.type", "unix");
1746 qdict_put_str(options
, "server.path", unixpath
);
1748 InetSocketAddress
*addr
= g_new(InetSocketAddress
, 1);
1750 if (inet_parse(addr
, host_spec
, errp
)) {
1754 qdict_put_str(options
, "server.type", "inet");
1755 qdict_put_str(options
, "server.host", addr
->host
);
1756 qdict_put_str(options
, "server.port", addr
->port
);
1758 qapi_free_InetSocketAddress(addr
);
1762 static bool nbd_process_legacy_socket_options(QDict
*output_options
,
1763 QemuOpts
*legacy_opts
,
1766 const char *path
= qemu_opt_get(legacy_opts
, "path");
1767 const char *host
= qemu_opt_get(legacy_opts
, "host");
1768 const char *port
= qemu_opt_get(legacy_opts
, "port");
1769 const QDictEntry
*e
;
1771 if (!path
&& !host
&& !port
) {
1775 for (e
= qdict_first(output_options
); e
; e
= qdict_next(output_options
, e
))
1777 if (strstart(e
->key
, "server.", NULL
)) {
1778 error_setg(errp
, "Cannot use 'server' and path/host/port at the "
1785 error_setg(errp
, "path and host may not be used at the same time");
1789 error_setg(errp
, "port may not be used without host");
1793 qdict_put_str(output_options
, "server.type", "unix");
1794 qdict_put_str(output_options
, "server.path", path
);
1796 qdict_put_str(output_options
, "server.type", "inet");
1797 qdict_put_str(output_options
, "server.host", host
);
1798 qdict_put_str(output_options
, "server.port",
1799 port
?: stringify(NBD_DEFAULT_PORT
));
1805 static SocketAddress
*nbd_config(BDRVNBDState
*s
, QDict
*options
,
1808 SocketAddress
*saddr
= NULL
;
1812 qdict_extract_subqdict(options
, &addr
, "server.");
1813 if (!qdict_size(addr
)) {
1814 error_setg(errp
, "NBD server address missing");
1818 iv
= qobject_input_visitor_new_flat_confused(addr
, errp
);
1823 if (!visit_type_SocketAddress(iv
, NULL
, &saddr
, errp
)) {
1827 if (socket_address_parse_named_fd(saddr
, errp
) < 0) {
1828 qapi_free_SocketAddress(saddr
);
1834 qobject_unref(addr
);
1839 static QCryptoTLSCreds
*nbd_get_tls_creds(const char *id
, Error
**errp
)
1842 QCryptoTLSCreds
*creds
;
1844 obj
= object_resolve_path_component(
1845 object_get_objects_root(), id
);
1847 error_setg(errp
, "No TLS credentials with id '%s'",
1851 creds
= (QCryptoTLSCreds
*)
1852 object_dynamic_cast(obj
, TYPE_QCRYPTO_TLS_CREDS
);
1854 error_setg(errp
, "Object with id '%s' is not TLS credentials",
1859 if (!qcrypto_tls_creds_check_endpoint(creds
,
1860 QCRYPTO_TLS_CREDS_ENDPOINT_CLIENT
,
1869 static QemuOptsList nbd_runtime_opts
= {
1871 .head
= QTAILQ_HEAD_INITIALIZER(nbd_runtime_opts
.head
),
1875 .type
= QEMU_OPT_STRING
,
1876 .help
= "TCP host to connect to",
1880 .type
= QEMU_OPT_STRING
,
1881 .help
= "TCP port to connect to",
1885 .type
= QEMU_OPT_STRING
,
1886 .help
= "Unix socket path to connect to",
1890 .type
= QEMU_OPT_STRING
,
1891 .help
= "Name of the NBD export to open",
1894 .name
= "tls-creds",
1895 .type
= QEMU_OPT_STRING
,
1896 .help
= "ID of the TLS credentials to use",
1899 .name
= "x-dirty-bitmap",
1900 .type
= QEMU_OPT_STRING
,
1901 .help
= "experimental: expose named dirty bitmap in place of "
1905 .name
= "reconnect-delay",
1906 .type
= QEMU_OPT_NUMBER
,
1907 .help
= "On an unexpected disconnect, the nbd client tries to "
1908 "connect again until succeeding or encountering a serious "
1909 "error. During the first @reconnect-delay seconds, all "
1910 "requests are paused and will be rerun on a successful "
1911 "reconnect. After that time, any delayed requests and all "
1912 "future requests before a successful reconnect will "
1913 "immediately fail. Default 0",
1915 { /* end of list */ }
1919 static int nbd_process_options(BlockDriverState
*bs
, QDict
*options
,
1922 BDRVNBDState
*s
= bs
->opaque
;
1926 opts
= qemu_opts_create(&nbd_runtime_opts
, NULL
, 0, &error_abort
);
1927 if (!qemu_opts_absorb_qdict(opts
, options
, errp
)) {
1931 /* Translate @host, @port, and @path to a SocketAddress */
1932 if (!nbd_process_legacy_socket_options(options
, opts
, errp
)) {
1936 /* Pop the config into our state object. Exit if invalid. */
1937 s
->saddr
= nbd_config(s
, options
, errp
);
1942 s
->export
= g_strdup(qemu_opt_get(opts
, "export"));
1943 if (s
->export
&& strlen(s
->export
) > NBD_MAX_STRING_SIZE
) {
1944 error_setg(errp
, "export name too long to send to server");
1948 s
->tlscredsid
= g_strdup(qemu_opt_get(opts
, "tls-creds"));
1949 if (s
->tlscredsid
) {
1950 s
->tlscreds
= nbd_get_tls_creds(s
->tlscredsid
, errp
);
1955 /* TODO SOCKET_ADDRESS_KIND_FD where fd has AF_INET or AF_INET6 */
1956 if (s
->saddr
->type
!= SOCKET_ADDRESS_TYPE_INET
) {
1957 error_setg(errp
, "TLS only supported over IP sockets");
1960 s
->hostname
= s
->saddr
->u
.inet
.host
;
1963 s
->x_dirty_bitmap
= g_strdup(qemu_opt_get(opts
, "x-dirty-bitmap"));
1964 if (s
->x_dirty_bitmap
&& strlen(s
->x_dirty_bitmap
) > NBD_MAX_STRING_SIZE
) {
1965 error_setg(errp
, "x-dirty-bitmap query too long to send to server");
1969 s
->reconnect_delay
= qemu_opt_get_number(opts
, "reconnect-delay", 0);
1974 qemu_opts_del(opts
);
1978 static int nbd_open(BlockDriverState
*bs
, QDict
*options
, int flags
,
1982 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
1985 qemu_co_mutex_init(&s
->send_mutex
);
1986 qemu_co_queue_init(&s
->free_sema
);
1988 if (!yank_register_instance(BLOCKDEV_YANK_INSTANCE(bs
->node_name
), errp
)) {
1992 ret
= nbd_process_options(bs
, options
, errp
);
1997 s
->conn
= nbd_client_connection_new(s
->saddr
, true, s
->export
,
1998 s
->x_dirty_bitmap
, s
->tlscreds
);
2000 /* TODO: Configurable retry-until-timeout behaviour. */
2001 ret
= nbd_do_establish_connection(bs
, errp
);
2006 s
->connection_co
= qemu_coroutine_create(nbd_connection_entry
, s
);
2007 bdrv_inc_in_flight(bs
);
2008 aio_co_schedule(bdrv_get_aio_context(bs
), s
->connection_co
);
2013 nbd_clear_bdrvstate(bs
);
2017 static int nbd_co_flush(BlockDriverState
*bs
)
2019 return nbd_client_co_flush(bs
);
2022 static void nbd_refresh_limits(BlockDriverState
*bs
, Error
**errp
)
2024 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
2025 uint32_t min
= s
->info
.min_block
;
2026 uint32_t max
= MIN_NON_ZERO(NBD_MAX_BUFFER_SIZE
, s
->info
.max_block
);
2029 * If the server did not advertise an alignment:
2030 * - a size that is not sector-aligned implies that an alignment
2031 * of 1 can be used to access those tail bytes
2032 * - advertisement of block status requires an alignment of 1, so
2033 * that we don't violate block layer constraints that block
2034 * status is always aligned (as we can't control whether the
2035 * server will report sub-sector extents, such as a hole at EOF
2036 * on an unaligned POSIX file)
2037 * - otherwise, assume the server is so old that we are safer avoiding
2038 * sub-sector requests
2041 min
= (!QEMU_IS_ALIGNED(s
->info
.size
, BDRV_SECTOR_SIZE
) ||
2042 s
->info
.base_allocation
) ? 1 : BDRV_SECTOR_SIZE
;
2045 bs
->bl
.request_alignment
= min
;
2046 bs
->bl
.max_pdiscard
= QEMU_ALIGN_DOWN(INT_MAX
, min
);
2047 bs
->bl
.max_pwrite_zeroes
= max
;
2048 bs
->bl
.max_transfer
= max
;
2050 if (s
->info
.opt_block
&&
2051 s
->info
.opt_block
> bs
->bl
.opt_transfer
) {
2052 bs
->bl
.opt_transfer
= s
->info
.opt_block
;
2056 static void nbd_close(BlockDriverState
*bs
)
2058 nbd_client_close(bs
);
2059 nbd_clear_bdrvstate(bs
);
2063 * NBD cannot truncate, but if the caller asks to truncate to the same size, or
2064 * to a smaller size with exact=false, there is no reason to fail the
2067 * Preallocation mode is ignored since it does not seems useful to fail when
2068 * we never change anything.
2070 static int coroutine_fn
nbd_co_truncate(BlockDriverState
*bs
, int64_t offset
,
2071 bool exact
, PreallocMode prealloc
,
2072 BdrvRequestFlags flags
, Error
**errp
)
2074 BDRVNBDState
*s
= bs
->opaque
;
2076 if (offset
!= s
->info
.size
&& exact
) {
2077 error_setg(errp
, "Cannot resize NBD nodes");
2081 if (offset
> s
->info
.size
) {
2082 error_setg(errp
, "Cannot grow NBD nodes");
2089 static int64_t nbd_getlength(BlockDriverState
*bs
)
2091 BDRVNBDState
*s
= bs
->opaque
;
2093 return s
->info
.size
;
2096 static void nbd_refresh_filename(BlockDriverState
*bs
)
2098 BDRVNBDState
*s
= bs
->opaque
;
2099 const char *host
= NULL
, *port
= NULL
, *path
= NULL
;
2102 if (s
->saddr
->type
== SOCKET_ADDRESS_TYPE_INET
) {
2103 const InetSocketAddress
*inet
= &s
->saddr
->u
.inet
;
2104 if (!inet
->has_ipv4
&& !inet
->has_ipv6
&& !inet
->has_to
) {
2108 } else if (s
->saddr
->type
== SOCKET_ADDRESS_TYPE_UNIX
) {
2109 path
= s
->saddr
->u
.q_unix
.path
;
2110 } /* else can't represent as pseudo-filename */
2112 if (path
&& s
->export
) {
2113 len
= snprintf(bs
->exact_filename
, sizeof(bs
->exact_filename
),
2114 "nbd+unix:///%s?socket=%s", s
->export
, path
);
2115 } else if (path
&& !s
->export
) {
2116 len
= snprintf(bs
->exact_filename
, sizeof(bs
->exact_filename
),
2117 "nbd+unix://?socket=%s", path
);
2118 } else if (host
&& s
->export
) {
2119 len
= snprintf(bs
->exact_filename
, sizeof(bs
->exact_filename
),
2120 "nbd://%s:%s/%s", host
, port
, s
->export
);
2121 } else if (host
&& !s
->export
) {
2122 len
= snprintf(bs
->exact_filename
, sizeof(bs
->exact_filename
),
2123 "nbd://%s:%s", host
, port
);
2125 if (len
>= sizeof(bs
->exact_filename
)) {
2126 /* Name is too long to represent exactly, so leave it empty. */
2127 bs
->exact_filename
[0] = '\0';
2131 static char *nbd_dirname(BlockDriverState
*bs
, Error
**errp
)
2133 /* The generic bdrv_dirname() implementation is able to work out some
2134 * directory name for NBD nodes, but that would be wrong. So far there is no
2135 * specification for how "export paths" would work, so NBD does not have
2136 * directory names. */
2137 error_setg(errp
, "Cannot generate a base directory for NBD nodes");
2141 static const char *const nbd_strong_runtime_opts
[] = {
2152 static void nbd_cancel_in_flight(BlockDriverState
*bs
)
2154 BDRVNBDState
*s
= (BDRVNBDState
*)bs
->opaque
;
2156 reconnect_delay_timer_del(s
);
2158 if (s
->state
== NBD_CLIENT_CONNECTING_WAIT
) {
2159 s
->state
= NBD_CLIENT_CONNECTING_NOWAIT
;
2160 qemu_co_queue_restart_all(&s
->free_sema
);
2164 static BlockDriver bdrv_nbd
= {
2165 .format_name
= "nbd",
2166 .protocol_name
= "nbd",
2167 .instance_size
= sizeof(BDRVNBDState
),
2168 .bdrv_parse_filename
= nbd_parse_filename
,
2169 .bdrv_co_create_opts
= bdrv_co_create_opts_simple
,
2170 .create_opts
= &bdrv_create_opts_simple
,
2171 .bdrv_file_open
= nbd_open
,
2172 .bdrv_reopen_prepare
= nbd_client_reopen_prepare
,
2173 .bdrv_co_preadv
= nbd_client_co_preadv
,
2174 .bdrv_co_pwritev
= nbd_client_co_pwritev
,
2175 .bdrv_co_pwrite_zeroes
= nbd_client_co_pwrite_zeroes
,
2176 .bdrv_close
= nbd_close
,
2177 .bdrv_co_flush_to_os
= nbd_co_flush
,
2178 .bdrv_co_pdiscard
= nbd_client_co_pdiscard
,
2179 .bdrv_refresh_limits
= nbd_refresh_limits
,
2180 .bdrv_co_truncate
= nbd_co_truncate
,
2181 .bdrv_getlength
= nbd_getlength
,
2182 .bdrv_detach_aio_context
= nbd_client_detach_aio_context
,
2183 .bdrv_attach_aio_context
= nbd_client_attach_aio_context
,
2184 .bdrv_co_drain_begin
= nbd_client_co_drain_begin
,
2185 .bdrv_co_drain_end
= nbd_client_co_drain_end
,
2186 .bdrv_refresh_filename
= nbd_refresh_filename
,
2187 .bdrv_co_block_status
= nbd_client_co_block_status
,
2188 .bdrv_dirname
= nbd_dirname
,
2189 .strong_runtime_opts
= nbd_strong_runtime_opts
,
2190 .bdrv_cancel_in_flight
= nbd_cancel_in_flight
,
2193 static BlockDriver bdrv_nbd_tcp
= {
2194 .format_name
= "nbd",
2195 .protocol_name
= "nbd+tcp",
2196 .instance_size
= sizeof(BDRVNBDState
),
2197 .bdrv_parse_filename
= nbd_parse_filename
,
2198 .bdrv_co_create_opts
= bdrv_co_create_opts_simple
,
2199 .create_opts
= &bdrv_create_opts_simple
,
2200 .bdrv_file_open
= nbd_open
,
2201 .bdrv_reopen_prepare
= nbd_client_reopen_prepare
,
2202 .bdrv_co_preadv
= nbd_client_co_preadv
,
2203 .bdrv_co_pwritev
= nbd_client_co_pwritev
,
2204 .bdrv_co_pwrite_zeroes
= nbd_client_co_pwrite_zeroes
,
2205 .bdrv_close
= nbd_close
,
2206 .bdrv_co_flush_to_os
= nbd_co_flush
,
2207 .bdrv_co_pdiscard
= nbd_client_co_pdiscard
,
2208 .bdrv_refresh_limits
= nbd_refresh_limits
,
2209 .bdrv_co_truncate
= nbd_co_truncate
,
2210 .bdrv_getlength
= nbd_getlength
,
2211 .bdrv_detach_aio_context
= nbd_client_detach_aio_context
,
2212 .bdrv_attach_aio_context
= nbd_client_attach_aio_context
,
2213 .bdrv_co_drain_begin
= nbd_client_co_drain_begin
,
2214 .bdrv_co_drain_end
= nbd_client_co_drain_end
,
2215 .bdrv_refresh_filename
= nbd_refresh_filename
,
2216 .bdrv_co_block_status
= nbd_client_co_block_status
,
2217 .bdrv_dirname
= nbd_dirname
,
2218 .strong_runtime_opts
= nbd_strong_runtime_opts
,
2219 .bdrv_cancel_in_flight
= nbd_cancel_in_flight
,
2222 static BlockDriver bdrv_nbd_unix
= {
2223 .format_name
= "nbd",
2224 .protocol_name
= "nbd+unix",
2225 .instance_size
= sizeof(BDRVNBDState
),
2226 .bdrv_parse_filename
= nbd_parse_filename
,
2227 .bdrv_co_create_opts
= bdrv_co_create_opts_simple
,
2228 .create_opts
= &bdrv_create_opts_simple
,
2229 .bdrv_file_open
= nbd_open
,
2230 .bdrv_reopen_prepare
= nbd_client_reopen_prepare
,
2231 .bdrv_co_preadv
= nbd_client_co_preadv
,
2232 .bdrv_co_pwritev
= nbd_client_co_pwritev
,
2233 .bdrv_co_pwrite_zeroes
= nbd_client_co_pwrite_zeroes
,
2234 .bdrv_close
= nbd_close
,
2235 .bdrv_co_flush_to_os
= nbd_co_flush
,
2236 .bdrv_co_pdiscard
= nbd_client_co_pdiscard
,
2237 .bdrv_refresh_limits
= nbd_refresh_limits
,
2238 .bdrv_co_truncate
= nbd_co_truncate
,
2239 .bdrv_getlength
= nbd_getlength
,
2240 .bdrv_detach_aio_context
= nbd_client_detach_aio_context
,
2241 .bdrv_attach_aio_context
= nbd_client_attach_aio_context
,
2242 .bdrv_co_drain_begin
= nbd_client_co_drain_begin
,
2243 .bdrv_co_drain_end
= nbd_client_co_drain_end
,
2244 .bdrv_refresh_filename
= nbd_refresh_filename
,
2245 .bdrv_co_block_status
= nbd_client_co_block_status
,
2246 .bdrv_dirname
= nbd_dirname
,
2247 .strong_runtime_opts
= nbd_strong_runtime_opts
,
2248 .bdrv_cancel_in_flight
= nbd_cancel_in_flight
,
2251 static void bdrv_nbd_init(void)
2253 bdrv_register(&bdrv_nbd
);
2254 bdrv_register(&bdrv_nbd_tcp
);
2255 bdrv_register(&bdrv_nbd_unix
);
2258 block_init(bdrv_nbd_init
);