block/nbd.c

   1 /*
   2  * QEMU Block driver for  NBD
   3  *
   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 <[email protected]>
   8  *
   9  * Some parts:
  10  *    Copyright (C) 2007 Anthony Liguori <[email protected]>
  11  *
  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:
  18  *
  19  * The above copyright notice and this permission notice shall be included in
  20  * all copies or substantial portions of the Software.
  21  *
  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
  28  * THE SOFTWARE.
  29  */
  30
  31 #include "qemu/osdep.h"
  32
  33 #include "trace.h"
  34 #include "qemu/uri.h"
  35 #include "qemu/option.h"
  36 #include "qemu/cutils.h"
  37 #include "qemu/main-loop.h"
  38
  39 #include "qapi/qapi-visit-sockets.h"
  40 #include "qapi/qmp/qstring.h"
  41
  42 #include "block/qdict.h"
  43 #include "block/nbd.h"
  44 #include "block/block_int.h"
  45
  46 #define EN_OPTSTR ":exportname="
  47 #define MAX_NBD_REQUESTS    16
  48
  49 #define HANDLE_TO_INDEX(bs, handle) ((handle) ^ (uint64_t)(intptr_t)(bs))
  50 #define INDEX_TO_HANDLE(bs, index)  ((index)  ^ (uint64_t)(intptr_t)(bs))
  51
  52 typedef struct {
  53     Coroutine *coroutine;
  54     uint64_t offset;        /* original offset of the request */
  55     bool receiving;         /* waiting for connection_co? */
  56 } NBDClientRequest;
  57
  58 typedef enum NBDClientState {
  59     NBD_CLIENT_CONNECTING_WAIT,
  60     NBD_CLIENT_CONNECTING_NOWAIT,
  61     NBD_CLIENT_CONNECTED,
  62     NBD_CLIENT_QUIT
  63 } NBDClientState;
  64
  65 typedef struct BDRVNBDState {
  66     QIOChannelSocket *sioc; /* The master data channel */
  67     QIOChannel *ioc; /* The current I/O channel which may differ (eg TLS) */
  68     NBDExportInfo info;
  69
  70     CoMutex send_mutex;
  71     CoQueue free_sema;
  72     Coroutine *connection_co;
  73     Coroutine *teardown_co;
  74     QemuCoSleepState *connection_co_sleep_ns_state;
  75     bool drained;
  76     bool wait_drained_end;
  77     int in_flight;
  78     NBDClientState state;
  79     int connect_status;
  80     Error *connect_err;
  81     bool wait_in_flight;
  82
  83     NBDClientRequest requests[MAX_NBD_REQUESTS];
  84     NBDReply reply;
  85     BlockDriverState *bs;
  86
  87     /* Connection parameters */
  88     uint32_t reconnect_delay;
  89     SocketAddress *saddr;
  90     char *export, *tlscredsid;
  91     QCryptoTLSCreds *tlscreds;
  92     const char *hostname;
  93     char *x_dirty_bitmap;
  94 } BDRVNBDState;
  95
  96 static int nbd_client_connect(BlockDriverState *bs, Error **errp);
  97
  98 static void nbd_clear_bdrvstate(BDRVNBDState *s)
  99 {
 100     object_unref(OBJECT(s->tlscreds));
 101     qapi_free_SocketAddress(s->saddr);
 102     s->saddr = NULL;
 103     g_free(s->export);
 104     s->export = NULL;
 105     g_free(s->tlscredsid);
 106     s->tlscredsid = NULL;
 107     g_free(s->x_dirty_bitmap);
 108     s->x_dirty_bitmap = NULL;
 109 }
 110
 111 static void nbd_channel_error(BDRVNBDState *s, int ret)
 112 {
 113     if (ret == -EIO) {
 114         if (s->state == NBD_CLIENT_CONNECTED) {
 115             s->state = s->reconnect_delay ? NBD_CLIENT_CONNECTING_WAIT :
 116                                             NBD_CLIENT_CONNECTING_NOWAIT;
 117         }
 118     } else {
 119         if (s->state == NBD_CLIENT_CONNECTED) {
 120             qio_channel_shutdown(s->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
 121         }
 122         s->state = NBD_CLIENT_QUIT;
 123     }
 124 }
 125
 126 static void nbd_recv_coroutines_wake_all(BDRVNBDState *s)
 127 {
 128     int i;
 129
 130     for (i = 0; i < MAX_NBD_REQUESTS; i++) {
 131         NBDClientRequest *req = &s->requests[i];
 132
 133         if (req->coroutine && req->receiving) {
 134             aio_co_wake(req->coroutine);
 135         }
 136     }
 137 }
 138
 139 static void nbd_client_detach_aio_context(BlockDriverState *bs)
 140 {
 141     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
 142
 143     qio_channel_detach_aio_context(QIO_CHANNEL(s->ioc));
 144 }
 145
 146 static void nbd_client_attach_aio_context_bh(void *opaque)
 147 {
 148     BlockDriverState *bs = opaque;
 149     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
 150
 151     /*
 152      * The node is still drained, so we know the coroutine has yielded in
 153      * nbd_read_eof(), the only place where bs->in_flight can reach 0, or it is
 154      * entered for the first time. Both places are safe for entering the
 155      * coroutine.
 156      */
 157     qemu_aio_coroutine_enter(bs->aio_context, s->connection_co);
 158     bdrv_dec_in_flight(bs);
 159 }
 160
 161 static void nbd_client_attach_aio_context(BlockDriverState *bs,
 162                                           AioContext *new_context)
 163 {
 164     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
 165
 166     /*
 167      * s->connection_co is either yielded from nbd_receive_reply or from
 168      * nbd_co_reconnect_loop()
 169      */
 170     if (s->state == NBD_CLIENT_CONNECTED) {
 171         qio_channel_attach_aio_context(QIO_CHANNEL(s->ioc), new_context);
 172     }
 173
 174     bdrv_inc_in_flight(bs);
 175
 176     /*
 177      * Need to wait here for the BH to run because the BH must run while the
 178      * node is still drained.
 179      */
 180     aio_wait_bh_oneshot(new_context, nbd_client_attach_aio_context_bh, bs);
 181 }
 182
 183 static void coroutine_fn nbd_client_co_drain_begin(BlockDriverState *bs)
 184 {
 185     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
 186
 187     s->drained = true;
 188     if (s->connection_co_sleep_ns_state) {
 189         qemu_co_sleep_wake(s->connection_co_sleep_ns_state);
 190     }
 191 }
 192
 193 static void coroutine_fn nbd_client_co_drain_end(BlockDriverState *bs)
 194 {
 195     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
 196
 197     s->drained = false;
 198     if (s->wait_drained_end) {
 199         s->wait_drained_end = false;
 200         aio_co_wake(s->connection_co);
 201     }
 202 }
 203
 204
 205 static void nbd_teardown_connection(BlockDriverState *bs)
 206 {
 207     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
 208
 209     if (s->state == NBD_CLIENT_CONNECTED) {
 210         /* finish any pending coroutines */
 211         assert(s->ioc);
 212         qio_channel_shutdown(s->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
 213     }
 214     s->state = NBD_CLIENT_QUIT;
 215     if (s->connection_co) {
 216         if (s->connection_co_sleep_ns_state) {
 217             qemu_co_sleep_wake(s->connection_co_sleep_ns_state);
 218         }
 219     }
 220     if (qemu_in_coroutine()) {
 221         s->teardown_co = qemu_coroutine_self();
 222         /* connection_co resumes us when it terminates */
 223         qemu_coroutine_yield();
 224         s->teardown_co = NULL;
 225     } else {
 226         BDRV_POLL_WHILE(bs, s->connection_co);
 227     }
 228     assert(!s->connection_co);
 229 }
 230
 231 static bool nbd_client_connecting(BDRVNBDState *s)
 232 {
 233     return s->state == NBD_CLIENT_CONNECTING_WAIT ||
 234         s->state == NBD_CLIENT_CONNECTING_NOWAIT;
 235 }
 236
 237 static bool nbd_client_connecting_wait(BDRVNBDState *s)
 238 {
 239     return s->state == NBD_CLIENT_CONNECTING_WAIT;
 240 }
 241
 242 static coroutine_fn void nbd_reconnect_attempt(BDRVNBDState *s)
 243 {
 244     Error *local_err = NULL;
 245
 246     if (!nbd_client_connecting(s)) {
 247         return;
 248     }
 249
 250     /* Wait for completion of all in-flight requests */
 251
 252     qemu_co_mutex_lock(&s->send_mutex);
 253
 254     while (s->in_flight > 0) {
 255         qemu_co_mutex_unlock(&s->send_mutex);
 256         nbd_recv_coroutines_wake_all(s);
 257         s->wait_in_flight = true;
 258         qemu_coroutine_yield();
 259         s->wait_in_flight = false;
 260         qemu_co_mutex_lock(&s->send_mutex);
 261     }
 262
 263     qemu_co_mutex_unlock(&s->send_mutex);
 264
 265     if (!nbd_client_connecting(s)) {
 266         return;
 267     }
 268
 269     /*
 270      * Now we are sure that nobody is accessing the channel, and no one will
 271      * try until we set the state to CONNECTED.
 272      */
 273
 274     /* Finalize previous connection if any */
 275     if (s->ioc) {
 276         nbd_client_detach_aio_context(s->bs);
 277         object_unref(OBJECT(s->sioc));
 278         s->sioc = NULL;
 279         object_unref(OBJECT(s->ioc));
 280         s->ioc = NULL;
 281     }
 282
 283     s->connect_status = nbd_client_connect(s->bs, &local_err);
 284     error_free(s->connect_err);
 285     s->connect_err = NULL;
 286     error_propagate(&s->connect_err, local_err);
 287
 288     if (s->connect_status < 0) {
 289         /* failed attempt */
 290         return;
 291     }
 292
 293     /* successfully connected */
 294     s->state = NBD_CLIENT_CONNECTED;
 295     qemu_co_queue_restart_all(&s->free_sema);
 296 }
 297
 298 static coroutine_fn void nbd_co_reconnect_loop(BDRVNBDState *s)
 299 {
 300     uint64_t start_time_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
 301     uint64_t delay_ns = s->reconnect_delay * NANOSECONDS_PER_SECOND;
 302     uint64_t timeout = 1 * NANOSECONDS_PER_SECOND;
 303     uint64_t max_timeout = 16 * NANOSECONDS_PER_SECOND;
 304
 305     nbd_reconnect_attempt(s);
 306
 307     while (nbd_client_connecting(s)) {
 308         if (s->state == NBD_CLIENT_CONNECTING_WAIT &&
 309             qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start_time_ns > delay_ns)
 310         {
 311             s->state = NBD_CLIENT_CONNECTING_NOWAIT;
 312             qemu_co_queue_restart_all(&s->free_sema);
 313         }
 314
 315         qemu_co_sleep_ns_wakeable(QEMU_CLOCK_REALTIME, timeout,
 316                                   &s->connection_co_sleep_ns_state);
 317         if (s->drained) {
 318             bdrv_dec_in_flight(s->bs);
 319             s->wait_drained_end = true;
 320             while (s->drained) {
 321                 /*
 322                  * We may be entered once from nbd_client_attach_aio_context_bh
 323                  * and then from nbd_client_co_drain_end. So here is a loop.
 324                  */
 325                 qemu_coroutine_yield();
 326             }
 327             bdrv_inc_in_flight(s->bs);
 328         }
 329         if (timeout < max_timeout) {
 330             timeout *= 2;
 331         }
 332
 333         nbd_reconnect_attempt(s);
 334     }
 335 }
 336
 337 static coroutine_fn void nbd_connection_entry(void *opaque)
 338 {
 339     BDRVNBDState *s = opaque;
 340     uint64_t i;
 341     int ret = 0;
 342     Error *local_err = NULL;
 343
 344     while (s->state != NBD_CLIENT_QUIT) {
 345         /*
 346          * The NBD client can only really be considered idle when it has
 347          * yielded from qio_channel_readv_all_eof(), waiting for data. This is
 348          * the point where the additional scheduled coroutine entry happens
 349          * after nbd_client_attach_aio_context().
 350          *
 351          * Therefore we keep an additional in_flight reference all the time and
 352          * only drop it temporarily here.
 353          */
 354
 355         if (nbd_client_connecting(s)) {
 356             nbd_co_reconnect_loop(s);
 357         }
 358
 359         if (s->state != NBD_CLIENT_CONNECTED) {
 360             continue;
 361         }
 362
 363         assert(s->reply.handle == 0);
 364         ret = nbd_receive_reply(s->bs, s->ioc, &s->reply, &local_err);
 365
 366         if (local_err) {
 367             trace_nbd_read_reply_entry_fail(ret, error_get_pretty(local_err));
 368             error_free(local_err);
 369             local_err = NULL;
 370         }
 371         if (ret <= 0) {
 372             nbd_channel_error(s, ret ? ret : -EIO);
 373             continue;
 374         }
 375
 376         /*
 377          * There's no need for a mutex on the receive side, because the
 378          * handler acts as a synchronization point and ensures that only
 379          * one coroutine is called until the reply finishes.
 380          */
 381         i = HANDLE_TO_INDEX(s, s->reply.handle);
 382         if (i >= MAX_NBD_REQUESTS ||
 383             !s->requests[i].coroutine ||
 384             !s->requests[i].receiving ||
 385             (nbd_reply_is_structured(&s->reply) && !s->info.structured_reply))
 386         {
 387             nbd_channel_error(s, -EINVAL);
 388             continue;
 389         }
 390
 391         /*
 392          * We're woken up again by the request itself.  Note that there
 393          * is no race between yielding and reentering connection_co.  This
 394          * is because:
 395          *
 396          * - if the request runs on the same AioContext, it is only
 397          *   entered after we yield
 398          *
 399          * - if the request runs on a different AioContext, reentering
 400          *   connection_co happens through a bottom half, which can only
 401          *   run after we yield.
 402          */
 403         aio_co_wake(s->requests[i].coroutine);
 404         qemu_coroutine_yield();
 405     }
 406
 407     qemu_co_queue_restart_all(&s->free_sema);
 408     nbd_recv_coroutines_wake_all(s);
 409     bdrv_dec_in_flight(s->bs);
 410
 411     s->connection_co = NULL;
 412     if (s->ioc) {
 413         nbd_client_detach_aio_context(s->bs);
 414         object_unref(OBJECT(s->sioc));
 415         s->sioc = NULL;
 416         object_unref(OBJECT(s->ioc));
 417         s->ioc = NULL;
 418     }
 419
 420     if (s->teardown_co) {
 421         aio_co_wake(s->teardown_co);
 422     }
 423     aio_wait_kick();
 424 }
 425
 426 static int nbd_co_send_request(BlockDriverState *bs,
 427                                NBDRequest *request,
 428                                QEMUIOVector *qiov)
 429 {
 430     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
 431     int rc, i = -1;
 432
 433     qemu_co_mutex_lock(&s->send_mutex);
 434     while (s->in_flight == MAX_NBD_REQUESTS || nbd_client_connecting_wait(s)) {
 435         qemu_co_queue_wait(&s->free_sema, &s->send_mutex);
 436     }
 437
 438     if (s->state != NBD_CLIENT_CONNECTED) {
 439         rc = -EIO;
 440         goto err;
 441     }
 442
 443     s->in_flight++;
 444
 445     for (i = 0; i < MAX_NBD_REQUESTS; i++) {
 446         if (s->requests[i].coroutine == NULL) {
 447             break;
 448         }
 449     }
 450
 451     g_assert(qemu_in_coroutine());
 452     assert(i < MAX_NBD_REQUESTS);
 453
 454     s->requests[i].coroutine = qemu_coroutine_self();
 455     s->requests[i].offset = request->from;
 456     s->requests[i].receiving = false;
 457
 458     request->handle = INDEX_TO_HANDLE(s, i);
 459
 460     assert(s->ioc);
 461
 462     if (qiov) {
 463         qio_channel_set_cork(s->ioc, true);
 464         rc = nbd_send_request(s->ioc, request);
 465         if (rc >= 0 && s->state == NBD_CLIENT_CONNECTED) {
 466             if (qio_channel_writev_all(s->ioc, qiov->iov, qiov->niov,
 467                                        NULL) < 0) {
 468                 rc = -EIO;
 469             }
 470         } else if (rc >= 0) {
 471             rc = -EIO;
 472         }
 473         qio_channel_set_cork(s->ioc, false);
 474     } else {
 475         rc = nbd_send_request(s->ioc, request);
 476     }
 477
 478 err:
 479     if (rc < 0) {
 480         nbd_channel_error(s, rc);
 481         if (i != -1) {
 482             s->requests[i].coroutine = NULL;
 483             s->in_flight--;
 484         }
 485         if (s->in_flight == 0 && s->wait_in_flight) {
 486             aio_co_wake(s->connection_co);
 487         } else {
 488             qemu_co_queue_next(&s->free_sema);
 489         }
 490     }
 491     qemu_co_mutex_unlock(&s->send_mutex);
 492     return rc;
 493 }
 494
 495 static inline uint16_t payload_advance16(uint8_t **payload)
 496 {
 497     *payload += 2;
 498     return lduw_be_p(*payload - 2);
 499 }
 500
 501 static inline uint32_t payload_advance32(uint8_t **payload)
 502 {
 503     *payload += 4;
 504     return ldl_be_p(*payload - 4);
 505 }
 506
 507 static inline uint64_t payload_advance64(uint8_t **payload)
 508 {
 509     *payload += 8;
 510     return ldq_be_p(*payload - 8);
 511 }
 512
 513 static int nbd_parse_offset_hole_payload(BDRVNBDState *s,
 514                                          NBDStructuredReplyChunk *chunk,
 515                                          uint8_t *payload, uint64_t orig_offset,
 516                                          QEMUIOVector *qiov, Error **errp)
 517 {
 518     uint64_t offset;
 519     uint32_t hole_size;
 520
 521     if (chunk->length != sizeof(offset) + sizeof(hole_size)) {
 522         error_setg(errp, "Protocol error: invalid payload for "
 523                          "NBD_REPLY_TYPE_OFFSET_HOLE");
 524         return -EINVAL;
 525     }
 526
 527     offset = payload_advance64(&payload);
 528     hole_size = payload_advance32(&payload);
 529
 530     if (!hole_size || offset < orig_offset || hole_size > qiov->size ||
 531         offset > orig_offset + qiov->size - hole_size) {
 532         error_setg(errp, "Protocol error: server sent chunk exceeding requested"
 533                          " region");
 534         return -EINVAL;
 535     }
 536     if (s->info.min_block &&
 537         !QEMU_IS_ALIGNED(hole_size, s->info.min_block)) {
 538         trace_nbd_structured_read_compliance("hole");
 539     }
 540
 541     qemu_iovec_memset(qiov, offset - orig_offset, 0, hole_size);
 542
 543     return 0;
 544 }
 545
 546 /*
 547  * nbd_parse_blockstatus_payload
 548  * Based on our request, we expect only one extent in reply, for the
 549  * base:allocation context.
 550  */
 551 static int nbd_parse_blockstatus_payload(BDRVNBDState *s,
 552                                          NBDStructuredReplyChunk *chunk,
 553                                          uint8_t *payload, uint64_t orig_length,
 554                                          NBDExtent *extent, Error **errp)
 555 {
 556     uint32_t context_id;
 557
 558     /* The server succeeded, so it must have sent [at least] one extent */
 559     if (chunk->length < sizeof(context_id) + sizeof(*extent)) {
 560         error_setg(errp, "Protocol error: invalid payload for "
 561                          "NBD_REPLY_TYPE_BLOCK_STATUS");
 562         return -EINVAL;
 563     }
 564
 565     context_id = payload_advance32(&payload);
 566     if (s->info.context_id != context_id) {
 567         error_setg(errp, "Protocol error: unexpected context id %d for "
 568                          "NBD_REPLY_TYPE_BLOCK_STATUS, when negotiated context "
 569                          "id is %d", context_id,
 570                          s->info.context_id);
 571         return -EINVAL;
 572     }
 573
 574     extent->length = payload_advance32(&payload);
 575     extent->flags = payload_advance32(&payload);
 576
 577     if (extent->length == 0) {
 578         error_setg(errp, "Protocol error: server sent status chunk with "
 579                    "zero length");
 580         return -EINVAL;
 581     }
 582
 583     /*
 584      * A server sending unaligned block status is in violation of the
 585      * protocol, but as qemu-nbd 3.1 is such a server (at least for
 586      * POSIX files that are not a multiple of 512 bytes, since qemu
 587      * rounds files up to 512-byte multiples but lseek(SEEK_HOLE)
 588      * still sees an implicit hole beyond the real EOF), it's nicer to
 589      * work around the misbehaving server. If the request included
 590      * more than the final unaligned block, truncate it back to an
 591      * aligned result; if the request was only the final block, round
 592      * up to the full block and change the status to fully-allocated
 593      * (always a safe status, even if it loses information).
 594      */
 595     if (s->info.min_block && !QEMU_IS_ALIGNED(extent->length,
 596                                                    s->info.min_block)) {
 597         trace_nbd_parse_blockstatus_compliance("extent length is unaligned");
 598         if (extent->length > s->info.min_block) {
 599             extent->length = QEMU_ALIGN_DOWN(extent->length,
 600                                              s->info.min_block);
 601         } else {
 602             extent->length = s->info.min_block;
 603             extent->flags = 0;
 604         }
 605     }
 606
 607     /*
 608      * We used NBD_CMD_FLAG_REQ_ONE, so the server should not have
 609      * sent us any more than one extent, nor should it have included
 610      * status beyond our request in that extent. However, it's easy
 611      * enough to ignore the server's noncompliance without killing the
 612      * connection; just ignore trailing extents, and clamp things to
 613      * the length of our request.
 614      */
 615     if (chunk->length > sizeof(context_id) + sizeof(*extent)) {
 616         trace_nbd_parse_blockstatus_compliance("more than one extent");
 617     }
 618     if (extent->length > orig_length) {
 619         extent->length = orig_length;
 620         trace_nbd_parse_blockstatus_compliance("extent length too large");
 621     }
 622
 623     return 0;
 624 }
 625
 626 /*
 627  * nbd_parse_error_payload
 628  * on success @errp contains message describing nbd error reply
 629  */
 630 static int nbd_parse_error_payload(NBDStructuredReplyChunk *chunk,
 631                                    uint8_t *payload, int *request_ret,
 632                                    Error **errp)
 633 {
 634     uint32_t error;
 635     uint16_t message_size;
 636
 637     assert(chunk->type & (1 << 15));
 638
 639     if (chunk->length < sizeof(error) + sizeof(message_size)) {
 640         error_setg(errp,
 641                    "Protocol error: invalid payload for structured error");
 642         return -EINVAL;
 643     }
 644
 645     error = nbd_errno_to_system_errno(payload_advance32(&payload));
 646     if (error == 0) {
 647         error_setg(errp, "Protocol error: server sent structured error chunk "
 648                          "with error = 0");
 649         return -EINVAL;
 650     }
 651
 652     *request_ret = -error;
 653     message_size = payload_advance16(&payload);
 654
 655     if (message_size > chunk->length - sizeof(error) - sizeof(message_size)) {
 656         error_setg(errp, "Protocol error: server sent structured error chunk "
 657                          "with incorrect message size");
 658         return -EINVAL;
 659     }
 660
 661     /* TODO: Add a trace point to mention the server complaint */
 662
 663     /* TODO handle ERROR_OFFSET */
 664
 665     return 0;
 666 }
 667
 668 static int nbd_co_receive_offset_data_payload(BDRVNBDState *s,
 669                                               uint64_t orig_offset,
 670                                               QEMUIOVector *qiov, Error **errp)
 671 {
 672     QEMUIOVector sub_qiov;
 673     uint64_t offset;
 674     size_t data_size;
 675     int ret;
 676     NBDStructuredReplyChunk *chunk = &s->reply.structured;
 677
 678     assert(nbd_reply_is_structured(&s->reply));
 679
 680     /* The NBD spec requires at least one byte of payload */
 681     if (chunk->length <= sizeof(offset)) {
 682         error_setg(errp, "Protocol error: invalid payload for "
 683                          "NBD_REPLY_TYPE_OFFSET_DATA");
 684         return -EINVAL;
 685     }
 686
 687     if (nbd_read64(s->ioc, &offset, "OFFSET_DATA offset", errp) < 0) {
 688         return -EIO;
 689     }
 690
 691     data_size = chunk->length - sizeof(offset);
 692     assert(data_size);
 693     if (offset < orig_offset || data_size > qiov->size ||
 694         offset > orig_offset + qiov->size - data_size) {
 695         error_setg(errp, "Protocol error: server sent chunk exceeding requested"
 696                          " region");
 697         return -EINVAL;
 698     }
 699     if (s->info.min_block && !QEMU_IS_ALIGNED(data_size, s->info.min_block)) {
 700         trace_nbd_structured_read_compliance("data");
 701     }
 702
 703     qemu_iovec_init(&sub_qiov, qiov->niov);
 704     qemu_iovec_concat(&sub_qiov, qiov, offset - orig_offset, data_size);
 705     ret = qio_channel_readv_all(s->ioc, sub_qiov.iov, sub_qiov.niov, errp);
 706     qemu_iovec_destroy(&sub_qiov);
 707
 708     return ret < 0 ? -EIO : 0;
 709 }
 710
 711 #define NBD_MAX_MALLOC_PAYLOAD 1000
 712 static coroutine_fn int nbd_co_receive_structured_payload(
 713         BDRVNBDState *s, void **payload, Error **errp)
 714 {
 715     int ret;
 716     uint32_t len;
 717
 718     assert(nbd_reply_is_structured(&s->reply));
 719
 720     len = s->reply.structured.length;
 721
 722     if (len == 0) {
 723         return 0;
 724     }
 725
 726     if (payload == NULL) {
 727         error_setg(errp, "Unexpected structured payload");
 728         return -EINVAL;
 729     }
 730
 731     if (len > NBD_MAX_MALLOC_PAYLOAD) {
 732         error_setg(errp, "Payload too large");
 733         return -EINVAL;
 734     }
 735
 736     *payload = g_new(char, len);
 737     ret = nbd_read(s->ioc, *payload, len, "structured payload", errp);
 738     if (ret < 0) {
 739         g_free(*payload);
 740         *payload = NULL;
 741         return ret;
 742     }
 743
 744     return 0;
 745 }
 746
 747 /*
 748  * nbd_co_do_receive_one_chunk
 749  * for simple reply:
 750  *   set request_ret to received reply error
 751  *   if qiov is not NULL: read payload to @qiov
 752  * for structured reply chunk:
 753  *   if error chunk: read payload, set @request_ret, do not set @payload
 754  *   else if offset_data chunk: read payload data to @qiov, do not set @payload
 755  *   else: read payload to @payload
 756  *
 757  * If function fails, @errp contains corresponding error message, and the
 758  * connection with the server is suspect.  If it returns 0, then the
 759  * transaction succeeded (although @request_ret may be a negative errno
 760  * corresponding to the server's error reply), and errp is unchanged.
 761  */
 762 static coroutine_fn int nbd_co_do_receive_one_chunk(
 763         BDRVNBDState *s, uint64_t handle, bool only_structured,
 764         int *request_ret, QEMUIOVector *qiov, void **payload, Error **errp)
 765 {
 766     int ret;
 767     int i = HANDLE_TO_INDEX(s, handle);
 768     void *local_payload = NULL;
 769     NBDStructuredReplyChunk *chunk;
 770
 771     if (payload) {
 772         *payload = NULL;
 773     }
 774     *request_ret = 0;
 775
 776     /* Wait until we're woken up by nbd_connection_entry.  */
 777     s->requests[i].receiving = true;
 778     qemu_coroutine_yield();
 779     s->requests[i].receiving = false;
 780     if (s->state != NBD_CLIENT_CONNECTED) {
 781         error_setg(errp, "Connection closed");
 782         return -EIO;
 783     }
 784     assert(s->ioc);
 785
 786     assert(s->reply.handle == handle);
 787
 788     if (nbd_reply_is_simple(&s->reply)) {
 789         if (only_structured) {
 790             error_setg(errp, "Protocol error: simple reply when structured "
 791                              "reply chunk was expected");
 792             return -EINVAL;
 793         }
 794
 795         *request_ret = -nbd_errno_to_system_errno(s->reply.simple.error);
 796         if (*request_ret < 0 || !qiov) {
 797             return 0;
 798         }
 799
 800         return qio_channel_readv_all(s->ioc, qiov->iov, qiov->niov,
 801                                      errp) < 0 ? -EIO : 0;
 802     }
 803
 804     /* handle structured reply chunk */
 805     assert(s->info.structured_reply);
 806     chunk = &s->reply.structured;
 807
 808     if (chunk->type == NBD_REPLY_TYPE_NONE) {
 809         if (!(chunk->flags & NBD_REPLY_FLAG_DONE)) {
 810             error_setg(errp, "Protocol error: NBD_REPLY_TYPE_NONE chunk without"
 811                        " NBD_REPLY_FLAG_DONE flag set");
 812             return -EINVAL;
 813         }
 814         if (chunk->length) {
 815             error_setg(errp, "Protocol error: NBD_REPLY_TYPE_NONE chunk with"
 816                        " nonzero length");
 817             return -EINVAL;
 818         }
 819         return 0;
 820     }
 821
 822     if (chunk->type == NBD_REPLY_TYPE_OFFSET_DATA) {
 823         if (!qiov) {
 824             error_setg(errp, "Unexpected NBD_REPLY_TYPE_OFFSET_DATA chunk");
 825             return -EINVAL;
 826         }
 827
 828         return nbd_co_receive_offset_data_payload(s, s->requests[i].offset,
 829                                                   qiov, errp);
 830     }
 831
 832     if (nbd_reply_type_is_error(chunk->type)) {
 833         payload = &local_payload;
 834     }
 835
 836     ret = nbd_co_receive_structured_payload(s, payload, errp);
 837     if (ret < 0) {
 838         return ret;
 839     }
 840
 841     if (nbd_reply_type_is_error(chunk->type)) {
 842         ret = nbd_parse_error_payload(chunk, local_payload, request_ret, errp);
 843         g_free(local_payload);
 844         return ret;
 845     }
 846
 847     return 0;
 848 }
 849
 850 /*
 851  * nbd_co_receive_one_chunk
 852  * Read reply, wake up connection_co and set s->quit if needed.
 853  * Return value is a fatal error code or normal nbd reply error code
 854  */
 855 static coroutine_fn int nbd_co_receive_one_chunk(
 856         BDRVNBDState *s, uint64_t handle, bool only_structured,
 857         int *request_ret, QEMUIOVector *qiov, NBDReply *reply, void **payload,
 858         Error **errp)
 859 {
 860     int ret = nbd_co_do_receive_one_chunk(s, handle, only_structured,
 861                                           request_ret, qiov, payload, errp);
 862
 863     if (ret < 0) {
 864         memset(reply, 0, sizeof(*reply));
 865         nbd_channel_error(s, ret);
 866     } else {
 867         /* For assert at loop start in nbd_connection_entry */
 868         *reply = s->reply;
 869     }
 870     s->reply.handle = 0;
 871
 872     if (s->connection_co && !s->wait_in_flight) {
 873         /*
 874          * We must check s->wait_in_flight, because we may entered by
 875          * nbd_recv_coroutines_wake_all(), in this case we should not
 876          * wake connection_co here, it will woken by last request.
 877          */
 878         aio_co_wake(s->connection_co);
 879     }
 880
 881     return ret;
 882 }
 883
 884 typedef struct NBDReplyChunkIter {
 885     int ret;
 886     int request_ret;
 887     Error *err;
 888     bool done, only_structured;
 889 } NBDReplyChunkIter;
 890
 891 static void nbd_iter_channel_error(NBDReplyChunkIter *iter,
 892                                    int ret, Error **local_err)
 893 {
 894     assert(local_err && *local_err);
 895     assert(ret < 0);
 896
 897     if (!iter->ret) {
 898         iter->ret = ret;
 899         error_propagate(&iter->err, *local_err);
 900     } else {
 901         error_free(*local_err);
 902     }
 903
 904     *local_err = NULL;
 905 }
 906
 907 static void nbd_iter_request_error(NBDReplyChunkIter *iter, int ret)
 908 {
 909     assert(ret < 0);
 910
 911     if (!iter->request_ret) {
 912         iter->request_ret = ret;
 913     }
 914 }
 915
 916 /*
 917  * NBD_FOREACH_REPLY_CHUNK
 918  * The pointer stored in @payload requires g_free() to free it.
 919  */
 920 #define NBD_FOREACH_REPLY_CHUNK(s, iter, handle, structured, \
 921                                 qiov, reply, payload) \
 922     for (iter = (NBDReplyChunkIter) { .only_structured = structured }; \
 923          nbd_reply_chunk_iter_receive(s, &iter, handle, qiov, reply, payload);)
 924
 925 /*
 926  * nbd_reply_chunk_iter_receive
 927  * The pointer stored in @payload requires g_free() to free it.
 928  */
 929 static bool nbd_reply_chunk_iter_receive(BDRVNBDState *s,
 930                                          NBDReplyChunkIter *iter,
 931                                          uint64_t handle,
 932                                          QEMUIOVector *qiov, NBDReply *reply,
 933                                          void **payload)
 934 {
 935     int ret, request_ret;
 936     NBDReply local_reply;
 937     NBDStructuredReplyChunk *chunk;
 938     Error *local_err = NULL;
 939     if (s->state != NBD_CLIENT_CONNECTED) {
 940         error_setg(&local_err, "Connection closed");
 941         nbd_iter_channel_error(iter, -EIO, &local_err);
 942         goto break_loop;
 943     }
 944
 945     if (iter->done) {
 946         /* Previous iteration was last. */
 947         goto break_loop;
 948     }
 949
 950     if (reply == NULL) {
 951         reply = &local_reply;
 952     }
 953
 954     ret = nbd_co_receive_one_chunk(s, handle, iter->only_structured,
 955                                    &request_ret, qiov, reply, payload,
 956                                    &local_err);
 957     if (ret < 0) {
 958         nbd_iter_channel_error(iter, ret, &local_err);
 959     } else if (request_ret < 0) {
 960         nbd_iter_request_error(iter, request_ret);
 961     }
 962
 963     /* Do not execute the body of NBD_FOREACH_REPLY_CHUNK for simple reply. */
 964     if (nbd_reply_is_simple(reply) || s->state != NBD_CLIENT_CONNECTED) {
 965         goto break_loop;
 966     }
 967
 968     chunk = &reply->structured;
 969     iter->only_structured = true;
 970
 971     if (chunk->type == NBD_REPLY_TYPE_NONE) {
 972         /* NBD_REPLY_FLAG_DONE is already checked in nbd_co_receive_one_chunk */
 973         assert(chunk->flags & NBD_REPLY_FLAG_DONE);
 974         goto break_loop;
 975     }
 976
 977     if (chunk->flags & NBD_REPLY_FLAG_DONE) {
 978         /* This iteration is last. */
 979         iter->done = true;
 980     }
 981
 982     /* Execute the loop body */
 983     return true;
 984
 985 break_loop:
 986     s->requests[HANDLE_TO_INDEX(s, handle)].coroutine = NULL;
 987
 988     qemu_co_mutex_lock(&s->send_mutex);
 989     s->in_flight--;
 990     if (s->in_flight == 0 && s->wait_in_flight) {
 991         aio_co_wake(s->connection_co);
 992     } else {
 993         qemu_co_queue_next(&s->free_sema);
 994     }
 995     qemu_co_mutex_unlock(&s->send_mutex);
 996
 997     return false;
 998 }
 999
1000 static int nbd_co_receive_return_code(BDRVNBDState *s, uint64_t handle,
1001                                       int *request_ret, Error **errp)
1002 {
1003     NBDReplyChunkIter iter;
1004
1005     NBD_FOREACH_REPLY_CHUNK(s, iter, handle, false, NULL, NULL, NULL) {
1006         /* nbd_reply_chunk_iter_receive does all the work */
1007     }
1008
1009     error_propagate(errp, iter.err);
1010     *request_ret = iter.request_ret;
1011     return iter.ret;
1012 }
1013
1014 static int nbd_co_receive_cmdread_reply(BDRVNBDState *s, uint64_t handle,
1015                                         uint64_t offset, QEMUIOVector *qiov,
1016                                         int *request_ret, Error **errp)
1017 {
1018     NBDReplyChunkIter iter;
1019     NBDReply reply;
1020     void *payload = NULL;
1021     Error *local_err = NULL;
1022
1023     NBD_FOREACH_REPLY_CHUNK(s, iter, handle, s->info.structured_reply,
1024                             qiov, &reply, &payload)
1025     {
1026         int ret;
1027         NBDStructuredReplyChunk *chunk = &reply.structured;
1028
1029         assert(nbd_reply_is_structured(&reply));
1030
1031         switch (chunk->type) {
1032         case NBD_REPLY_TYPE_OFFSET_DATA:
1033             /*
1034              * special cased in nbd_co_receive_one_chunk, data is already
1035              * in qiov
1036              */
1037             break;
1038         case NBD_REPLY_TYPE_OFFSET_HOLE:
1039             ret = nbd_parse_offset_hole_payload(s, &reply.structured, payload,
1040                                                 offset, qiov, &local_err);
1041             if (ret < 0) {
1042                 nbd_channel_error(s, ret);
1043                 nbd_iter_channel_error(&iter, ret, &local_err);
1044             }
1045             break;
1046         default:
1047             if (!nbd_reply_type_is_error(chunk->type)) {
1048                 /* not allowed reply type */
1049                 nbd_channel_error(s, -EINVAL);
1050                 error_setg(&local_err,
1051                            "Unexpected reply type: %d (%s) for CMD_READ",
1052                            chunk->type, nbd_reply_type_lookup(chunk->type));
1053                 nbd_iter_channel_error(&iter, -EINVAL, &local_err);
1054             }
1055         }
1056
1057         g_free(payload);
1058         payload = NULL;
1059     }
1060
1061     error_propagate(errp, iter.err);
1062     *request_ret = iter.request_ret;
1063     return iter.ret;
1064 }
1065
1066 static int nbd_co_receive_blockstatus_reply(BDRVNBDState *s,
1067                                             uint64_t handle, uint64_t length,
1068                                             NBDExtent *extent,
1069                                             int *request_ret, Error **errp)
1070 {
1071     NBDReplyChunkIter iter;
1072     NBDReply reply;
1073     void *payload = NULL;
1074     Error *local_err = NULL;
1075     bool received = false;
1076
1077     assert(!extent->length);
1078     NBD_FOREACH_REPLY_CHUNK(s, iter, handle, false, NULL, &reply, &payload) {
1079         int ret;
1080         NBDStructuredReplyChunk *chunk = &reply.structured;
1081
1082         assert(nbd_reply_is_structured(&reply));
1083
1084         switch (chunk->type) {
1085         case NBD_REPLY_TYPE_BLOCK_STATUS:
1086             if (received) {
1087                 nbd_channel_error(s, -EINVAL);
1088                 error_setg(&local_err, "Several BLOCK_STATUS chunks in reply");
1089                 nbd_iter_channel_error(&iter, -EINVAL, &local_err);
1090             }
1091             received = true;
1092
1093             ret = nbd_parse_blockstatus_payload(s, &reply.structured,
1094                                                 payload, length, extent,
1095                                                 &local_err);
1096             if (ret < 0) {
1097                 nbd_channel_error(s, ret);
1098                 nbd_iter_channel_error(&iter, ret, &local_err);
1099             }
1100             break;
1101         default:
1102             if (!nbd_reply_type_is_error(chunk->type)) {
1103                 nbd_channel_error(s, -EINVAL);
1104                 error_setg(&local_err,
1105                            "Unexpected reply type: %d (%s) "
1106                            "for CMD_BLOCK_STATUS",
1107                            chunk->type, nbd_reply_type_lookup(chunk->type));
1108                 nbd_iter_channel_error(&iter, -EINVAL, &local_err);
1109             }
1110         }
1111
1112         g_free(payload);
1113         payload = NULL;
1114     }
1115
1116     if (!extent->length && !iter.request_ret) {
1117         error_setg(&local_err, "Server did not reply with any status extents");
1118         nbd_iter_channel_error(&iter, -EIO, &local_err);
1119     }
1120
1121     error_propagate(errp, iter.err);
1122     *request_ret = iter.request_ret;
1123     return iter.ret;
1124 }
1125
1126 static int nbd_co_request(BlockDriverState *bs, NBDRequest *request,
1127                           QEMUIOVector *write_qiov)
1128 {
1129     int ret, request_ret;
1130     Error *local_err = NULL;
1131     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
1132
1133     assert(request->type != NBD_CMD_READ);
1134     if (write_qiov) {
1135         assert(request->type == NBD_CMD_WRITE);
1136         assert(request->len == iov_size(write_qiov->iov, write_qiov->niov));
1137     } else {
1138         assert(request->type != NBD_CMD_WRITE);
1139     }
1140
1141     do {
1142         ret = nbd_co_send_request(bs, request, write_qiov);
1143         if (ret < 0) {
1144             continue;
1145         }
1146
1147         ret = nbd_co_receive_return_code(s, request->handle,
1148                                          &request_ret, &local_err);
1149         if (local_err) {
1150             trace_nbd_co_request_fail(request->from, request->len,
1151                                       request->handle, request->flags,
1152                                       request->type,
1153                                       nbd_cmd_lookup(request->type),
1154                                       ret, error_get_pretty(local_err));
1155             error_free(local_err);
1156             local_err = NULL;
1157         }
1158     } while (ret < 0 && nbd_client_connecting_wait(s));
1159
1160     return ret ? ret : request_ret;
1161 }
1162
1163 static int nbd_client_co_preadv(BlockDriverState *bs, uint64_t offset,
1164                                 uint64_t bytes, QEMUIOVector *qiov, int flags)
1165 {
1166     int ret, request_ret;
1167     Error *local_err = NULL;
1168     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
1169     NBDRequest request = {
1170         .type = NBD_CMD_READ,
1171         .from = offset,
1172         .len = bytes,
1173     };
1174
1175     assert(bytes <= NBD_MAX_BUFFER_SIZE);
1176     assert(!flags);
1177
1178     if (!bytes) {
1179         return 0;
1180     }
1181     /*
1182      * Work around the fact that the block layer doesn't do
1183      * byte-accurate sizing yet - if the read exceeds the server's
1184      * advertised size because the block layer rounded size up, then
1185      * truncate the request to the server and tail-pad with zero.
1186      */
1187     if (offset >= s->info.size) {
1188         assert(bytes < BDRV_SECTOR_SIZE);
1189         qemu_iovec_memset(qiov, 0, 0, bytes);
1190         return 0;
1191     }
1192     if (offset + bytes > s->info.size) {
1193         uint64_t slop = offset + bytes - s->info.size;
1194
1195         assert(slop < BDRV_SECTOR_SIZE);
1196         qemu_iovec_memset(qiov, bytes - slop, 0, slop);
1197         request.len -= slop;
1198     }
1199
1200     do {
1201         ret = nbd_co_send_request(bs, &request, NULL);
1202         if (ret < 0) {
1203             continue;
1204         }
1205
1206         ret = nbd_co_receive_cmdread_reply(s, request.handle, offset, qiov,
1207                                            &request_ret, &local_err);
1208         if (local_err) {
1209             trace_nbd_co_request_fail(request.from, request.len, request.handle,
1210                                       request.flags, request.type,
1211                                       nbd_cmd_lookup(request.type),
1212                                       ret, error_get_pretty(local_err));
1213             error_free(local_err);
1214             local_err = NULL;
1215         }
1216     } while (ret < 0 && nbd_client_connecting_wait(s));
1217
1218     return ret ? ret : request_ret;
1219 }
1220
1221 static int nbd_client_co_pwritev(BlockDriverState *bs, uint64_t offset,
1222                                  uint64_t bytes, QEMUIOVector *qiov, int flags)
1223 {
1224     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
1225     NBDRequest request = {
1226         .type = NBD_CMD_WRITE,
1227         .from = offset,
1228         .len = bytes,
1229     };
1230
1231     assert(!(s->info.flags & NBD_FLAG_READ_ONLY));
1232     if (flags & BDRV_REQ_FUA) {
1233         assert(s->info.flags & NBD_FLAG_SEND_FUA);
1234         request.flags |= NBD_CMD_FLAG_FUA;
1235     }
1236
1237     assert(bytes <= NBD_MAX_BUFFER_SIZE);
1238
1239     if (!bytes) {
1240         return 0;
1241     }
1242     return nbd_co_request(bs, &request, qiov);
1243 }
1244
1245 static int nbd_client_co_pwrite_zeroes(BlockDriverState *bs, int64_t offset,
1246                                        int bytes, BdrvRequestFlags flags)
1247 {
1248     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
1249     NBDRequest request = {
1250         .type = NBD_CMD_WRITE_ZEROES,
1251         .from = offset,
1252         .len = bytes,
1253     };
1254
1255     assert(!(s->info.flags & NBD_FLAG_READ_ONLY));
1256     if (!(s->info.flags & NBD_FLAG_SEND_WRITE_ZEROES)) {
1257         return -ENOTSUP;
1258     }
1259
1260     if (flags & BDRV_REQ_FUA) {
1261         assert(s->info.flags & NBD_FLAG_SEND_FUA);
1262         request.flags |= NBD_CMD_FLAG_FUA;
1263     }
1264     if (!(flags & BDRV_REQ_MAY_UNMAP)) {
1265         request.flags |= NBD_CMD_FLAG_NO_HOLE;
1266     }
1267     if (flags & BDRV_REQ_NO_FALLBACK) {
1268         assert(s->info.flags & NBD_FLAG_SEND_FAST_ZERO);
1269         request.flags |= NBD_CMD_FLAG_FAST_ZERO;
1270     }
1271
1272     if (!bytes) {
1273         return 0;
1274     }
1275     return nbd_co_request(bs, &request, NULL);
1276 }
1277
1278 static int nbd_client_co_flush(BlockDriverState *bs)
1279 {
1280     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
1281     NBDRequest request = { .type = NBD_CMD_FLUSH };
1282
1283     if (!(s->info.flags & NBD_FLAG_SEND_FLUSH)) {
1284         return 0;
1285     }
1286
1287     request.from = 0;
1288     request.len = 0;
1289
1290     return nbd_co_request(bs, &request, NULL);
1291 }
1292
1293 static int nbd_client_co_pdiscard(BlockDriverState *bs, int64_t offset,
1294                                   int bytes)
1295 {
1296     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
1297     NBDRequest request = {
1298         .type = NBD_CMD_TRIM,
1299         .from = offset,
1300         .len = bytes,
1301     };
1302
1303     assert(!(s->info.flags & NBD_FLAG_READ_ONLY));
1304     if (!(s->info.flags & NBD_FLAG_SEND_TRIM) || !bytes) {
1305         return 0;
1306     }
1307
1308     return nbd_co_request(bs, &request, NULL);
1309 }
1310
1311 static int coroutine_fn nbd_client_co_block_status(
1312         BlockDriverState *bs, bool want_zero, int64_t offset, int64_t bytes,
1313         int64_t *pnum, int64_t *map, BlockDriverState **file)
1314 {
1315     int ret, request_ret;
1316     NBDExtent extent = { 0 };
1317     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
1318     Error *local_err = NULL;
1319
1320     NBDRequest request = {
1321         .type = NBD_CMD_BLOCK_STATUS,
1322         .from = offset,
1323         .len = MIN(MIN_NON_ZERO(QEMU_ALIGN_DOWN(INT_MAX,
1324                                                 bs->bl.request_alignment),
1325                                 s->info.max_block),
1326                    MIN(bytes, s->info.size - offset)),
1327         .flags = NBD_CMD_FLAG_REQ_ONE,
1328     };
1329
1330     if (!s->info.base_allocation) {
1331         *pnum = bytes;
1332         *map = offset;
1333         *file = bs;
1334         return BDRV_BLOCK_DATA | BDRV_BLOCK_OFFSET_VALID;
1335     }
1336
1337     /*
1338      * Work around the fact that the block layer doesn't do
1339      * byte-accurate sizing yet - if the status request exceeds the
1340      * server's advertised size because the block layer rounded size
1341      * up, we truncated the request to the server (above), or are
1342      * called on just the hole.
1343      */
1344     if (offset >= s->info.size) {
1345         *pnum = bytes;
1346         assert(bytes < BDRV_SECTOR_SIZE);
1347         /* Intentionally don't report offset_valid for the hole */
1348         return BDRV_BLOCK_ZERO;
1349     }
1350
1351     if (s->info.min_block) {
1352         assert(QEMU_IS_ALIGNED(request.len, s->info.min_block));
1353     }
1354     do {
1355         ret = nbd_co_send_request(bs, &request, NULL);
1356         if (ret < 0) {
1357             continue;
1358         }
1359
1360         ret = nbd_co_receive_blockstatus_reply(s, request.handle, bytes,
1361                                                &extent, &request_ret,
1362                                                &local_err);
1363         if (local_err) {
1364             trace_nbd_co_request_fail(request.from, request.len, request.handle,
1365                                       request.flags, request.type,
1366                                       nbd_cmd_lookup(request.type),
1367                                       ret, error_get_pretty(local_err));
1368             error_free(local_err);
1369             local_err = NULL;
1370         }
1371     } while (ret < 0 && nbd_client_connecting_wait(s));
1372
1373     if (ret < 0 || request_ret < 0) {
1374         return ret ? ret : request_ret;
1375     }
1376
1377     assert(extent.length);
1378     *pnum = extent.length;
1379     *map = offset;
1380     *file = bs;
1381     return (extent.flags & NBD_STATE_HOLE ? 0 : BDRV_BLOCK_DATA) |
1382         (extent.flags & NBD_STATE_ZERO ? BDRV_BLOCK_ZERO : 0) |
1383         BDRV_BLOCK_OFFSET_VALID;
1384 }
1385
1386 static int nbd_client_reopen_prepare(BDRVReopenState *state,
1387                                      BlockReopenQueue *queue, Error **errp)
1388 {
1389     BDRVNBDState *s = (BDRVNBDState *)state->bs->opaque;
1390
1391     if ((state->flags & BDRV_O_RDWR) && (s->info.flags & NBD_FLAG_READ_ONLY)) {
1392         error_setg(errp, "Can't reopen read-only NBD mount as read/write");
1393         return -EACCES;
1394     }
1395     return 0;
1396 }
1397
1398 static void nbd_client_close(BlockDriverState *bs)
1399 {
1400     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
1401     NBDRequest request = { .type = NBD_CMD_DISC };
1402
1403     if (s->ioc) {
1404         nbd_send_request(s->ioc, &request);
1405     }
1406
1407     nbd_teardown_connection(bs);
1408 }
1409
1410 static QIOChannelSocket *nbd_establish_connection(SocketAddress *saddr,
1411                                                   Error **errp)
1412 {
1413     QIOChannelSocket *sioc;
1414     Error *local_err = NULL;
1415
1416     sioc = qio_channel_socket_new();
1417     qio_channel_set_name(QIO_CHANNEL(sioc), "nbd-client");
1418
1419     qio_channel_socket_connect_sync(sioc, saddr, &local_err);
1420     if (local_err) {
1421         object_unref(OBJECT(sioc));
1422         error_propagate(errp, local_err);
1423         return NULL;
1424     }
1425
1426     qio_channel_set_delay(QIO_CHANNEL(sioc), false);
1427
1428     return sioc;
1429 }
1430
1431 static int nbd_client_connect(BlockDriverState *bs, Error **errp)
1432 {
1433     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
1434     AioContext *aio_context = bdrv_get_aio_context(bs);
1435     int ret;
1436
1437     /*
1438      * establish TCP connection, return error if it fails
1439      * TODO: Configurable retry-until-timeout behaviour.
1440      */
1441     QIOChannelSocket *sioc = nbd_establish_connection(s->saddr, errp);
1442
1443     if (!sioc) {
1444         return -ECONNREFUSED;
1445     }
1446
1447     /* NBD handshake */
1448     trace_nbd_client_connect(s->export);
1449     qio_channel_set_blocking(QIO_CHANNEL(sioc), false, NULL);
1450     qio_channel_attach_aio_context(QIO_CHANNEL(sioc), aio_context);
1451
1452     s->info.request_sizes = true;
1453     s->info.structured_reply = true;
1454     s->info.base_allocation = true;
1455     s->info.x_dirty_bitmap = g_strdup(s->x_dirty_bitmap);
1456     s->info.name = g_strdup(s->export ?: "");
1457     ret = nbd_receive_negotiate(aio_context, QIO_CHANNEL(sioc), s->tlscreds,
1458                                 s->hostname, &s->ioc, &s->info, errp);
1459     g_free(s->info.x_dirty_bitmap);
1460     g_free(s->info.name);
1461     if (ret < 0) {
1462         object_unref(OBJECT(sioc));
1463         return ret;
1464     }
1465     if (s->x_dirty_bitmap && !s->info.base_allocation) {
1466         error_setg(errp, "requested x-dirty-bitmap %s not found",
1467                    s->x_dirty_bitmap);
1468         ret = -EINVAL;
1469         goto fail;
1470     }
1471     if (s->info.flags & NBD_FLAG_READ_ONLY) {
1472         ret = bdrv_apply_auto_read_only(bs, "NBD export is read-only", errp);
1473         if (ret < 0) {
1474             goto fail;
1475         }
1476     }
1477     if (s->info.flags & NBD_FLAG_SEND_FUA) {
1478         bs->supported_write_flags = BDRV_REQ_FUA;
1479         bs->supported_zero_flags |= BDRV_REQ_FUA;
1480     }
1481     if (s->info.flags & NBD_FLAG_SEND_WRITE_ZEROES) {
1482         bs->supported_zero_flags |= BDRV_REQ_MAY_UNMAP;
1483         if (s->info.flags & NBD_FLAG_SEND_FAST_ZERO) {
1484             bs->supported_zero_flags |= BDRV_REQ_NO_FALLBACK;
1485         }
1486     }
1487
1488     s->sioc = sioc;
1489
1490     if (!s->ioc) {
1491         s->ioc = QIO_CHANNEL(sioc);
1492         object_ref(OBJECT(s->ioc));
1493     }
1494
1495     trace_nbd_client_connect_success(s->export);
1496
1497     return 0;
1498
1499  fail:
1500     /*
1501      * We have connected, but must fail for other reasons.
1502      * Send NBD_CMD_DISC as a courtesy to the server.
1503      */
1504     {
1505         NBDRequest request = { .type = NBD_CMD_DISC };
1506
1507         nbd_send_request(s->ioc ?: QIO_CHANNEL(sioc), &request);
1508
1509         object_unref(OBJECT(sioc));
1510
1511         return ret;
1512     }
1513 }
1514
1515 /*
1516  * Parse nbd_open options
1517  */
1518
1519 static int nbd_parse_uri(const char *filename, QDict *options)
1520 {
1521     URI *uri;
1522     const char *p;
1523     QueryParams *qp = NULL;
1524     int ret = 0;
1525     bool is_unix;
1526
1527     uri = uri_parse(filename);
1528     if (!uri) {
1529         return -EINVAL;
1530     }
1531
1532     /* transport */
1533     if (!g_strcmp0(uri->scheme, "nbd")) {
1534         is_unix = false;
1535     } else if (!g_strcmp0(uri->scheme, "nbd+tcp")) {
1536         is_unix = false;
1537     } else if (!g_strcmp0(uri->scheme, "nbd+unix")) {
1538         is_unix = true;
1539     } else {
1540         ret = -EINVAL;
1541         goto out;
1542     }
1543
1544     p = uri->path ? uri->path : "";
1545     if (p[0] == '/') {
1546         p++;
1547     }
1548     if (p[0]) {
1549         qdict_put_str(options, "export", p);
1550     }
1551
1552     qp = query_params_parse(uri->query);
1553     if (qp->n > 1 || (is_unix && !qp->n) || (!is_unix && qp->n)) {
1554         ret = -EINVAL;
1555         goto out;
1556     }
1557
1558     if (is_unix) {
1559         /* nbd+unix:///export?socket=path */
1560         if (uri->server || uri->port || strcmp(qp->p[0].name, "socket")) {
1561             ret = -EINVAL;
1562             goto out;
1563         }
1564         qdict_put_str(options, "server.type", "unix");
1565         qdict_put_str(options, "server.path", qp->p[0].value);
1566     } else {
1567         QString *host;
1568         char *port_str;
1569
1570         /* nbd[+tcp]://host[:port]/export */
1571         if (!uri->server) {
1572             ret = -EINVAL;
1573             goto out;
1574         }
1575
1576         /* strip braces from literal IPv6 address */
1577         if (uri->server[0] == '[') {
1578             host = qstring_from_substr(uri->server, 1,
1579                                        strlen(uri->server) - 1);
1580         } else {
1581             host = qstring_from_str(uri->server);
1582         }
1583
1584         qdict_put_str(options, "server.type", "inet");
1585         qdict_put(options, "server.host", host);
1586
1587         port_str = g_strdup_printf("%d", uri->port ?: NBD_DEFAULT_PORT);
1588         qdict_put_str(options, "server.port", port_str);
1589         g_free(port_str);
1590     }
1591
1592 out:
1593     if (qp) {
1594         query_params_free(qp);
1595     }
1596     uri_free(uri);
1597     return ret;
1598 }
1599
1600 static bool nbd_has_filename_options_conflict(QDict *options, Error **errp)
1601 {
1602     const QDictEntry *e;
1603
1604     for (e = qdict_first(options); e; e = qdict_next(options, e)) {
1605         if (!strcmp(e->key, "host") ||
1606             !strcmp(e->key, "port") ||
1607             !strcmp(e->key, "path") ||
1608             !strcmp(e->key, "export") ||
1609             strstart(e->key, "server.", NULL))
1610         {
1611             error_setg(errp, "Option '%s' cannot be used with a file name",
1612                        e->key);
1613             return true;
1614         }
1615     }
1616
1617     return false;
1618 }
1619
1620 static void nbd_parse_filename(const char *filename, QDict *options,
1621                                Error **errp)
1622 {
1623     g_autofree char *file = NULL;
1624     char *export_name;
1625     const char *host_spec;
1626     const char *unixpath;
1627
1628     if (nbd_has_filename_options_conflict(options, errp)) {
1629         return;
1630     }
1631
1632     if (strstr(filename, "://")) {
1633         int ret = nbd_parse_uri(filename, options);
1634         if (ret < 0) {
1635             error_setg(errp, "No valid URL specified");
1636         }
1637         return;
1638     }
1639
1640     file = g_strdup(filename);
1641
1642     export_name = strstr(file, EN_OPTSTR);
1643     if (export_name) {
1644         if (export_name[strlen(EN_OPTSTR)] == 0) {
1645             return;
1646         }
1647         export_name[0] = 0; /* truncate 'file' */
1648         export_name += strlen(EN_OPTSTR);
1649
1650         qdict_put_str(options, "export", export_name);
1651     }
1652
1653     /* extract the host_spec - fail if it's not nbd:... */
1654     if (!strstart(file, "nbd:", &host_spec)) {
1655         error_setg(errp, "File name string for NBD must start with 'nbd:'");
1656         return;
1657     }
1658
1659     if (!*host_spec) {
1660         return;
1661     }
1662
1663     /* are we a UNIX or TCP socket? */
1664     if (strstart(host_spec, "unix:", &unixpath)) {
1665         qdict_put_str(options, "server.type", "unix");
1666         qdict_put_str(options, "server.path", unixpath);
1667     } else {
1668         InetSocketAddress *addr = g_new(InetSocketAddress, 1);
1669
1670         if (inet_parse(addr, host_spec, errp)) {
1671             goto out_inet;
1672         }
1673
1674         qdict_put_str(options, "server.type", "inet");
1675         qdict_put_str(options, "server.host", addr->host);
1676         qdict_put_str(options, "server.port", addr->port);
1677     out_inet:
1678         qapi_free_InetSocketAddress(addr);
1679     }
1680 }
1681
1682 static bool nbd_process_legacy_socket_options(QDict *output_options,
1683                                               QemuOpts *legacy_opts,
1684                                               Error **errp)
1685 {
1686     const char *path = qemu_opt_get(legacy_opts, "path");
1687     const char *host = qemu_opt_get(legacy_opts, "host");
1688     const char *port = qemu_opt_get(legacy_opts, "port");
1689     const QDictEntry *e;
1690
1691     if (!path && !host && !port) {
1692         return true;
1693     }
1694
1695     for (e = qdict_first(output_options); e; e = qdict_next(output_options, e))
1696     {
1697         if (strstart(e->key, "server.", NULL)) {
1698             error_setg(errp, "Cannot use 'server' and path/host/port at the "
1699                        "same time");
1700             return false;
1701         }
1702     }
1703
1704     if (path && host) {
1705         error_setg(errp, "path and host may not be used at the same time");
1706         return false;
1707     } else if (path) {
1708         if (port) {
1709             error_setg(errp, "port may not be used without host");
1710             return false;
1711         }
1712
1713         qdict_put_str(output_options, "server.type", "unix");
1714         qdict_put_str(output_options, "server.path", path);
1715     } else if (host) {
1716         qdict_put_str(output_options, "server.type", "inet");
1717         qdict_put_str(output_options, "server.host", host);
1718         qdict_put_str(output_options, "server.port",
1719                       port ?: stringify(NBD_DEFAULT_PORT));
1720     }
1721
1722     return true;
1723 }
1724
1725 static SocketAddress *nbd_config(BDRVNBDState *s, QDict *options,
1726                                  Error **errp)
1727 {
1728     SocketAddress *saddr = NULL;
1729     QDict *addr = NULL;
1730     Visitor *iv = NULL;
1731     Error *local_err = NULL;
1732
1733     qdict_extract_subqdict(options, &addr, "server.");
1734     if (!qdict_size(addr)) {
1735         error_setg(errp, "NBD server address missing");
1736         goto done;
1737     }
1738
1739     iv = qobject_input_visitor_new_flat_confused(addr, errp);
1740     if (!iv) {
1741         goto done;
1742     }
1743
1744     visit_type_SocketAddress(iv, NULL, &saddr, &local_err);
1745     if (local_err) {
1746         error_propagate(errp, local_err);
1747         goto done;
1748     }
1749
1750 done:
1751     qobject_unref(addr);
1752     visit_free(iv);
1753     return saddr;
1754 }
1755
1756 static QCryptoTLSCreds *nbd_get_tls_creds(const char *id, Error **errp)
1757 {
1758     Object *obj;
1759     QCryptoTLSCreds *creds;
1760
1761     obj = object_resolve_path_component(
1762         object_get_objects_root(), id);
1763     if (!obj) {
1764         error_setg(errp, "No TLS credentials with id '%s'",
1765                    id);
1766         return NULL;
1767     }
1768     creds = (QCryptoTLSCreds *)
1769         object_dynamic_cast(obj, TYPE_QCRYPTO_TLS_CREDS);
1770     if (!creds) {
1771         error_setg(errp, "Object with id '%s' is not TLS credentials",
1772                    id);
1773         return NULL;
1774     }
1775
1776     if (creds->endpoint != QCRYPTO_TLS_CREDS_ENDPOINT_CLIENT) {
1777         error_setg(errp,
1778                    "Expecting TLS credentials with a client endpoint");
1779         return NULL;
1780     }
1781     object_ref(obj);
1782     return creds;
1783 }
1784
1785
1786 static QemuOptsList nbd_runtime_opts = {
1787     .name = "nbd",
1788     .head = QTAILQ_HEAD_INITIALIZER(nbd_runtime_opts.head),
1789     .desc = {
1790         {
1791             .name = "host",
1792             .type = QEMU_OPT_STRING,
1793             .help = "TCP host to connect to",
1794         },
1795         {
1796             .name = "port",
1797             .type = QEMU_OPT_STRING,
1798             .help = "TCP port to connect to",
1799         },
1800         {
1801             .name = "path",
1802             .type = QEMU_OPT_STRING,
1803             .help = "Unix socket path to connect to",
1804         },
1805         {
1806             .name = "export",
1807             .type = QEMU_OPT_STRING,
1808             .help = "Name of the NBD export to open",
1809         },
1810         {
1811             .name = "tls-creds",
1812             .type = QEMU_OPT_STRING,
1813             .help = "ID of the TLS credentials to use",
1814         },
1815         {
1816             .name = "x-dirty-bitmap",
1817             .type = QEMU_OPT_STRING,
1818             .help = "experimental: expose named dirty bitmap in place of "
1819                     "block status",
1820         },
1821         {
1822             .name = "reconnect-delay",
1823             .type = QEMU_OPT_NUMBER,
1824             .help = "On an unexpected disconnect, the nbd client tries to "
1825                     "connect again until succeeding or encountering a serious "
1826                     "error.  During the first @reconnect-delay seconds, all "
1827                     "requests are paused and will be rerun on a successful "
1828                     "reconnect. After that time, any delayed requests and all "
1829                     "future requests before a successful reconnect will "
1830                     "immediately fail. Default 0",
1831         },
1832         { /* end of list */ }
1833     },
1834 };
1835
1836 static int nbd_process_options(BlockDriverState *bs, QDict *options,
1837                                Error **errp)
1838 {
1839     BDRVNBDState *s = bs->opaque;
1840     QemuOpts *opts;
1841     Error *local_err = NULL;
1842     int ret = -EINVAL;
1843
1844     opts = qemu_opts_create(&nbd_runtime_opts, NULL, 0, &error_abort);
1845     qemu_opts_absorb_qdict(opts, options, &local_err);
1846     if (local_err) {
1847         error_propagate(errp, local_err);
1848         goto error;
1849     }
1850
1851     /* Translate @host, @port, and @path to a SocketAddress */
1852     if (!nbd_process_legacy_socket_options(options, opts, errp)) {
1853         goto error;
1854     }
1855
1856     /* Pop the config into our state object. Exit if invalid. */
1857     s->saddr = nbd_config(s, options, errp);
1858     if (!s->saddr) {
1859         goto error;
1860     }
1861
1862     s->export = g_strdup(qemu_opt_get(opts, "export"));
1863     if (s->export && strlen(s->export) > NBD_MAX_STRING_SIZE) {
1864         error_setg(errp, "export name too long to send to server");
1865         goto error;
1866     }
1867
1868     s->tlscredsid = g_strdup(qemu_opt_get(opts, "tls-creds"));
1869     if (s->tlscredsid) {
1870         s->tlscreds = nbd_get_tls_creds(s->tlscredsid, errp);
1871         if (!s->tlscreds) {
1872             goto error;
1873         }
1874
1875         /* TODO SOCKET_ADDRESS_KIND_FD where fd has AF_INET or AF_INET6 */
1876         if (s->saddr->type != SOCKET_ADDRESS_TYPE_INET) {
1877             error_setg(errp, "TLS only supported over IP sockets");
1878             goto error;
1879         }
1880         s->hostname = s->saddr->u.inet.host;
1881     }
1882
1883     s->x_dirty_bitmap = g_strdup(qemu_opt_get(opts, "x-dirty-bitmap"));
1884     if (s->x_dirty_bitmap && strlen(s->x_dirty_bitmap) > NBD_MAX_STRING_SIZE) {
1885         error_setg(errp, "x-dirty-bitmap query too long to send to server");
1886         goto error;
1887     }
1888
1889     s->reconnect_delay = qemu_opt_get_number(opts, "reconnect-delay", 0);
1890
1891     ret = 0;
1892
1893  error:
1894     if (ret < 0) {
1895         nbd_clear_bdrvstate(s);
1896     }
1897     qemu_opts_del(opts);
1898     return ret;
1899 }
1900
1901 static int nbd_open(BlockDriverState *bs, QDict *options, int flags,
1902                     Error **errp)
1903 {
1904     int ret;
1905     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
1906
1907     ret = nbd_process_options(bs, options, errp);
1908     if (ret < 0) {
1909         return ret;
1910     }
1911
1912     s->bs = bs;
1913     qemu_co_mutex_init(&s->send_mutex);
1914     qemu_co_queue_init(&s->free_sema);
1915
1916     ret = nbd_client_connect(bs, errp);
1917     if (ret < 0) {
1918         nbd_clear_bdrvstate(s);
1919         return ret;
1920     }
1921     /* successfully connected */
1922     s->state = NBD_CLIENT_CONNECTED;
1923
1924     s->connection_co = qemu_coroutine_create(nbd_connection_entry, s);
1925     bdrv_inc_in_flight(bs);
1926     aio_co_schedule(bdrv_get_aio_context(bs), s->connection_co);
1927
1928     return 0;
1929 }
1930
1931 static int nbd_co_flush(BlockDriverState *bs)
1932 {
1933     return nbd_client_co_flush(bs);
1934 }
1935
1936 static void nbd_refresh_limits(BlockDriverState *bs, Error **errp)
1937 {
1938     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
1939     uint32_t min = s->info.min_block;
1940     uint32_t max = MIN_NON_ZERO(NBD_MAX_BUFFER_SIZE, s->info.max_block);
1941
1942     /*
1943      * If the server did not advertise an alignment:
1944      * - a size that is not sector-aligned implies that an alignment
1945      *   of 1 can be used to access those tail bytes
1946      * - advertisement of block status requires an alignment of 1, so
1947      *   that we don't violate block layer constraints that block
1948      *   status is always aligned (as we can't control whether the
1949      *   server will report sub-sector extents, such as a hole at EOF
1950      *   on an unaligned POSIX file)
1951      * - otherwise, assume the server is so old that we are safer avoiding
1952      *   sub-sector requests
1953      */
1954     if (!min) {
1955         min = (!QEMU_IS_ALIGNED(s->info.size, BDRV_SECTOR_SIZE) ||
1956                s->info.base_allocation) ? 1 : BDRV_SECTOR_SIZE;
1957     }
1958
1959     bs->bl.request_alignment = min;
1960     bs->bl.max_pdiscard = max;
1961     bs->bl.max_pwrite_zeroes = max;
1962     bs->bl.max_transfer = max;
1963
1964     if (s->info.opt_block &&
1965         s->info.opt_block > bs->bl.opt_transfer) {
1966         bs->bl.opt_transfer = s->info.opt_block;
1967     }
1968 }
1969
1970 static void nbd_close(BlockDriverState *bs)
1971 {
1972     BDRVNBDState *s = bs->opaque;
1973
1974     nbd_client_close(bs);
1975     nbd_clear_bdrvstate(s);
1976 }
1977
1978 static int64_t nbd_getlength(BlockDriverState *bs)
1979 {
1980     BDRVNBDState *s = bs->opaque;
1981
1982     return s->info.size;
1983 }
1984
1985 static void nbd_refresh_filename(BlockDriverState *bs)
1986 {
1987     BDRVNBDState *s = bs->opaque;
1988     const char *host = NULL, *port = NULL, *path = NULL;
1989
1990     if (s->saddr->type == SOCKET_ADDRESS_TYPE_INET) {
1991         const InetSocketAddress *inet = &s->saddr->u.inet;
1992         if (!inet->has_ipv4 && !inet->has_ipv6 && !inet->has_to) {
1993             host = inet->host;
1994             port = inet->port;
1995         }
1996     } else if (s->saddr->type == SOCKET_ADDRESS_TYPE_UNIX) {
1997         path = s->saddr->u.q_unix.path;
1998     } /* else can't represent as pseudo-filename */
1999
2000     if (path && s->export) {
2001         snprintf(bs->exact_filename, sizeof(bs->exact_filename),
2002                  "nbd+unix:///%s?socket=%s", s->export, path);
2003     } else if (path && !s->export) {
2004         snprintf(bs->exact_filename, sizeof(bs->exact_filename),
2005                  "nbd+unix://?socket=%s", path);
2006     } else if (host && s->export) {
2007         snprintf(bs->exact_filename, sizeof(bs->exact_filename),
2008                  "nbd://%s:%s/%s", host, port, s->export);
2009     } else if (host && !s->export) {
2010         snprintf(bs->exact_filename, sizeof(bs->exact_filename),
2011                  "nbd://%s:%s", host, port);
2012     }
2013 }
2014
2015 static char *nbd_dirname(BlockDriverState *bs, Error **errp)
2016 {
2017     /* The generic bdrv_dirname() implementation is able to work out some
2018      * directory name for NBD nodes, but that would be wrong. So far there is no
2019      * specification for how "export paths" would work, so NBD does not have
2020      * directory names. */
2021     error_setg(errp, "Cannot generate a base directory for NBD nodes");
2022     return NULL;
2023 }
2024
2025 static const char *const nbd_strong_runtime_opts[] = {
2026     "path",
2027     "host",
2028     "port",
2029     "export",
2030     "tls-creds",
2031     "server.",
2032
2033     NULL
2034 };
2035
2036 static BlockDriver bdrv_nbd = {
2037     .format_name                = "nbd",
2038     .protocol_name              = "nbd",
2039     .instance_size              = sizeof(BDRVNBDState),
2040     .bdrv_parse_filename        = nbd_parse_filename,
2041     .bdrv_file_open             = nbd_open,
2042     .bdrv_reopen_prepare        = nbd_client_reopen_prepare,
2043     .bdrv_co_preadv             = nbd_client_co_preadv,
2044     .bdrv_co_pwritev            = nbd_client_co_pwritev,
2045     .bdrv_co_pwrite_zeroes      = nbd_client_co_pwrite_zeroes,
2046     .bdrv_close                 = nbd_close,
2047     .bdrv_co_flush_to_os        = nbd_co_flush,
2048     .bdrv_co_pdiscard           = nbd_client_co_pdiscard,
2049     .bdrv_refresh_limits        = nbd_refresh_limits,
2050     .bdrv_getlength             = nbd_getlength,
2051     .bdrv_detach_aio_context    = nbd_client_detach_aio_context,
2052     .bdrv_attach_aio_context    = nbd_client_attach_aio_context,
2053     .bdrv_co_drain_begin        = nbd_client_co_drain_begin,
2054     .bdrv_co_drain_end          = nbd_client_co_drain_end,
2055     .bdrv_refresh_filename      = nbd_refresh_filename,
2056     .bdrv_co_block_status       = nbd_client_co_block_status,
2057     .bdrv_dirname               = nbd_dirname,
2058     .strong_runtime_opts        = nbd_strong_runtime_opts,
2059 };
2060
2061 static BlockDriver bdrv_nbd_tcp = {
2062     .format_name                = "nbd",
2063     .protocol_name              = "nbd+tcp",
2064     .instance_size              = sizeof(BDRVNBDState),
2065     .bdrv_parse_filename        = nbd_parse_filename,
2066     .bdrv_file_open             = nbd_open,
2067     .bdrv_reopen_prepare        = nbd_client_reopen_prepare,
2068     .bdrv_co_preadv             = nbd_client_co_preadv,
2069     .bdrv_co_pwritev            = nbd_client_co_pwritev,
2070     .bdrv_co_pwrite_zeroes      = nbd_client_co_pwrite_zeroes,
2071     .bdrv_close                 = nbd_close,
2072     .bdrv_co_flush_to_os        = nbd_co_flush,
2073     .bdrv_co_pdiscard           = nbd_client_co_pdiscard,
2074     .bdrv_refresh_limits        = nbd_refresh_limits,
2075     .bdrv_getlength             = nbd_getlength,
2076     .bdrv_detach_aio_context    = nbd_client_detach_aio_context,
2077     .bdrv_attach_aio_context    = nbd_client_attach_aio_context,
2078     .bdrv_co_drain_begin        = nbd_client_co_drain_begin,
2079     .bdrv_co_drain_end          = nbd_client_co_drain_end,
2080     .bdrv_refresh_filename      = nbd_refresh_filename,
2081     .bdrv_co_block_status       = nbd_client_co_block_status,
2082     .bdrv_dirname               = nbd_dirname,
2083     .strong_runtime_opts        = nbd_strong_runtime_opts,
2084 };
2085
2086 static BlockDriver bdrv_nbd_unix = {
2087     .format_name                = "nbd",
2088     .protocol_name              = "nbd+unix",
2089     .instance_size              = sizeof(BDRVNBDState),
2090     .bdrv_parse_filename        = nbd_parse_filename,
2091     .bdrv_file_open             = nbd_open,
2092     .bdrv_reopen_prepare        = nbd_client_reopen_prepare,
2093     .bdrv_co_preadv             = nbd_client_co_preadv,
2094     .bdrv_co_pwritev            = nbd_client_co_pwritev,
2095     .bdrv_co_pwrite_zeroes      = nbd_client_co_pwrite_zeroes,
2096     .bdrv_close                 = nbd_close,
2097     .bdrv_co_flush_to_os        = nbd_co_flush,
2098     .bdrv_co_pdiscard           = nbd_client_co_pdiscard,
2099     .bdrv_refresh_limits        = nbd_refresh_limits,
2100     .bdrv_getlength             = nbd_getlength,
2101     .bdrv_detach_aio_context    = nbd_client_detach_aio_context,
2102     .bdrv_attach_aio_context    = nbd_client_attach_aio_context,
2103     .bdrv_co_drain_begin        = nbd_client_co_drain_begin,
2104     .bdrv_co_drain_end          = nbd_client_co_drain_end,
2105     .bdrv_refresh_filename      = nbd_refresh_filename,
2106     .bdrv_co_block_status       = nbd_client_co_block_status,
2107     .bdrv_dirname               = nbd_dirname,
2108     .strong_runtime_opts        = nbd_strong_runtime_opts,
2109 };
2110
2111 static void bdrv_nbd_init(void)
2112 {
2113     bdrv_register(&bdrv_nbd);
2114     bdrv_register(&bdrv_nbd_tcp);
2115     bdrv_register(&bdrv_nbd_unix);
2116 }
2117
2118 block_init(bdrv_nbd_init);