block/nbd.c

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