block/nbd.c

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