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