block/nbd.c

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