block/nbd-client.c

   1 /*
   2  * QEMU Block driver for  NBD
   3  *
   4  * Copyright (C) 2016 Red Hat, Inc.
   5  * Copyright (C) 2008 Bull S.A.S.
   6  *     Author: Laurent Vivier <[email protected]>
   7  *
   8  * Some parts:
   9  *    Copyright (C) 2007 Anthony Liguori <[email protected]>
  10  *
  11  * Permission is hereby granted, free of charge, to any person obtaining a copy
  12  * of this software and associated documentation files (the "Software"), to deal
  13  * in the Software without restriction, including without limitation the rights
  14  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  15  * copies of the Software, and to permit persons to whom the Software is
  16  * furnished to do so, subject to the following conditions:
  17  *
  18  * The above copyright notice and this permission notice shall be included in
  19  * all copies or substantial portions of the Software.
  20  *
  21  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  22  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  23  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
  24  * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  25  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  26  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  27  * THE SOFTWARE.
  28  */
  29
  30 #include "qemu/osdep.h"
  31
  32 #include "trace.h"
  33 #include "qapi/error.h"
  34 #include "nbd-client.h"
  35
  36 #define HANDLE_TO_INDEX(bs, handle) ((handle) ^ (uint64_t)(intptr_t)(bs))
  37 #define INDEX_TO_HANDLE(bs, index)  ((index)  ^ (uint64_t)(intptr_t)(bs))
  38
  39 static void nbd_recv_coroutines_wake_all(NBDClientSession *s)
  40 {
  41     int i;
  42
  43     for (i = 0; i < MAX_NBD_REQUESTS; i++) {
  44         NBDClientRequest *req = &s->requests[i];
  45
  46         if (req->coroutine && req->receiving) {
  47             aio_co_wake(req->coroutine);
  48         }
  49     }
  50 }
  51
  52 static void nbd_teardown_connection(BlockDriverState *bs)
  53 {
  54     NBDClientSession *client = nbd_get_client_session(bs);
  55
  56     if (!client->ioc) { /* Already closed */
  57         return;
  58     }
  59
  60     /* finish any pending coroutines */
  61     qio_channel_shutdown(client->ioc,
  62                          QIO_CHANNEL_SHUTDOWN_BOTH,
  63                          NULL);
  64     BDRV_POLL_WHILE(bs, client->read_reply_co);
  65
  66     nbd_client_detach_aio_context(bs);
  67     object_unref(OBJECT(client->sioc));
  68     client->sioc = NULL;
  69     object_unref(OBJECT(client->ioc));
  70     client->ioc = NULL;
  71 }
  72
  73 static coroutine_fn void nbd_read_reply_entry(void *opaque)
  74 {
  75     NBDClientSession *s = opaque;
  76     uint64_t i;
  77     int ret = 0;
  78     Error *local_err = NULL;
  79
  80     while (!s->quit) {
  81         assert(s->reply.handle == 0);
  82         ret = nbd_receive_reply(s->ioc, &s->reply, &local_err);
  83         if (local_err) {
  84             trace_nbd_read_reply_entry_fail(ret, error_get_pretty(local_err));
  85             error_free(local_err);
  86         }
  87         if (ret <= 0) {
  88             break;
  89         }
  90
  91         /* There's no need for a mutex on the receive side, because the
  92          * handler acts as a synchronization point and ensures that only
  93          * one coroutine is called until the reply finishes.
  94          */
  95         i = HANDLE_TO_INDEX(s, s->reply.handle);
  96         if (i >= MAX_NBD_REQUESTS ||
  97             !s->requests[i].coroutine ||
  98             !s->requests[i].receiving ||
  99             (nbd_reply_is_structured(&s->reply) && !s->info.structured_reply))
 100         {
 101             break;
 102         }
 103
 104         /* We're woken up again by the request itself.  Note that there
 105          * is no race between yielding and reentering read_reply_co.  This
 106          * is because:
 107          *
 108          * - if the request runs on the same AioContext, it is only
 109          *   entered after we yield
 110          *
 111          * - if the request runs on a different AioContext, reentering
 112          *   read_reply_co happens through a bottom half, which can only
 113          *   run after we yield.
 114          */
 115         aio_co_wake(s->requests[i].coroutine);
 116         qemu_coroutine_yield();
 117     }
 118
 119     s->quit = true;
 120     nbd_recv_coroutines_wake_all(s);
 121     s->read_reply_co = NULL;
 122 }
 123
 124 static int nbd_co_send_request(BlockDriverState *bs,
 125                                NBDRequest *request,
 126                                QEMUIOVector *qiov)
 127 {
 128     NBDClientSession *s = nbd_get_client_session(bs);
 129     int rc, i;
 130
 131     qemu_co_mutex_lock(&s->send_mutex);
 132     while (s->in_flight == MAX_NBD_REQUESTS) {
 133         qemu_co_queue_wait(&s->free_sema, &s->send_mutex);
 134     }
 135     s->in_flight++;
 136
 137     for (i = 0; i < MAX_NBD_REQUESTS; i++) {
 138         if (s->requests[i].coroutine == NULL) {
 139             break;
 140         }
 141     }
 142
 143     g_assert(qemu_in_coroutine());
 144     assert(i < MAX_NBD_REQUESTS);
 145
 146     s->requests[i].coroutine = qemu_coroutine_self();
 147     s->requests[i].offset = request->from;
 148     s->requests[i].receiving = false;
 149
 150     request->handle = INDEX_TO_HANDLE(s, i);
 151
 152     if (s->quit) {
 153         rc = -EIO;
 154         goto err;
 155     }
 156     if (!s->ioc) {
 157         rc = -EPIPE;
 158         goto err;
 159     }
 160
 161     if (qiov) {
 162         qio_channel_set_cork(s->ioc, true);
 163         rc = nbd_send_request(s->ioc, request);
 164         if (rc >= 0 && !s->quit) {
 165             if (qio_channel_writev_all(s->ioc, qiov->iov, qiov->niov,
 166                                        NULL) < 0) {
 167                 rc = -EIO;
 168             }
 169         } else if (rc >= 0) {
 170             rc = -EIO;
 171         }
 172         qio_channel_set_cork(s->ioc, false);
 173     } else {
 174         rc = nbd_send_request(s->ioc, request);
 175     }
 176
 177 err:
 178     if (rc < 0) {
 179         s->quit = true;
 180         s->requests[i].coroutine = NULL;
 181         s->in_flight--;
 182         qemu_co_queue_next(&s->free_sema);
 183     }
 184     qemu_co_mutex_unlock(&s->send_mutex);
 185     return rc;
 186 }
 187
 188 static inline uint16_t payload_advance16(uint8_t **payload)
 189 {
 190     *payload += 2;
 191     return lduw_be_p(*payload - 2);
 192 }
 193
 194 static inline uint32_t payload_advance32(uint8_t **payload)
 195 {
 196     *payload += 4;
 197     return ldl_be_p(*payload - 4);
 198 }
 199
 200 static inline uint64_t payload_advance64(uint8_t **payload)
 201 {
 202     *payload += 8;
 203     return ldq_be_p(*payload - 8);
 204 }
 205
 206 static int nbd_parse_offset_hole_payload(NBDStructuredReplyChunk *chunk,
 207                                          uint8_t *payload, uint64_t orig_offset,
 208                                          QEMUIOVector *qiov, Error **errp)
 209 {
 210     uint64_t offset;
 211     uint32_t hole_size;
 212
 213     if (chunk->length != sizeof(offset) + sizeof(hole_size)) {
 214         error_setg(errp, "Protocol error: invalid payload for "
 215                          "NBD_REPLY_TYPE_OFFSET_HOLE");
 216         return -EINVAL;
 217     }
 218
 219     offset = payload_advance64(&payload);
 220     hole_size = payload_advance32(&payload);
 221
 222     if (!hole_size || offset < orig_offset || hole_size > qiov->size ||
 223         offset > orig_offset + qiov->size - hole_size) {
 224         error_setg(errp, "Protocol error: server sent chunk exceeding requested"
 225                          " region");
 226         return -EINVAL;
 227     }
 228
 229     qemu_iovec_memset(qiov, offset - orig_offset, 0, hole_size);
 230
 231     return 0;
 232 }
 233
 234 /* nbd_parse_blockstatus_payload
 235  * support only one extent in reply and only for
 236  * base:allocation context
 237  */
 238 static int nbd_parse_blockstatus_payload(NBDClientSession *client,
 239                                          NBDStructuredReplyChunk *chunk,
 240                                          uint8_t *payload, uint64_t orig_length,
 241                                          NBDExtent *extent, Error **errp)
 242 {
 243     uint32_t context_id;
 244
 245     if (chunk->length != sizeof(context_id) + sizeof(*extent)) {
 246         error_setg(errp, "Protocol error: invalid payload for "
 247                          "NBD_REPLY_TYPE_BLOCK_STATUS");
 248         return -EINVAL;
 249     }
 250
 251     context_id = payload_advance32(&payload);
 252     if (client->info.meta_base_allocation_id != context_id) {
 253         error_setg(errp, "Protocol error: unexpected context id %d for "
 254                          "NBD_REPLY_TYPE_BLOCK_STATUS, when negotiated context "
 255                          "id is %d", context_id,
 256                          client->info.meta_base_allocation_id);
 257         return -EINVAL;
 258     }
 259
 260     extent->length = payload_advance32(&payload);
 261     extent->flags = payload_advance32(&payload);
 262
 263     if (extent->length == 0 ||
 264         (client->info.min_block && !QEMU_IS_ALIGNED(extent->length,
 265                                                     client->info.min_block))) {
 266         error_setg(errp, "Protocol error: server sent status chunk with "
 267                    "invalid length");
 268         return -EINVAL;
 269     }
 270
 271     /* The server is allowed to send us extra information on the final
 272      * extent; just clamp it to the length we requested. */
 273     if (extent->length > orig_length) {
 274         extent->length = orig_length;
 275     }
 276
 277     return 0;
 278 }
 279
 280 /* nbd_parse_error_payload
 281  * on success @errp contains message describing nbd error reply
 282  */
 283 static int nbd_parse_error_payload(NBDStructuredReplyChunk *chunk,
 284                                    uint8_t *payload, int *request_ret,
 285                                    Error **errp)
 286 {
 287     uint32_t error;
 288     uint16_t message_size;
 289
 290     assert(chunk->type & (1 << 15));
 291
 292     if (chunk->length < sizeof(error) + sizeof(message_size)) {
 293         error_setg(errp,
 294                    "Protocol error: invalid payload for structured error");
 295         return -EINVAL;
 296     }
 297
 298     error = nbd_errno_to_system_errno(payload_advance32(&payload));
 299     if (error == 0) {
 300         error_setg(errp, "Protocol error: server sent structured error chunk "
 301                          "with error = 0");
 302         return -EINVAL;
 303     }
 304
 305     *request_ret = -error;
 306     message_size = payload_advance16(&payload);
 307
 308     if (message_size > chunk->length - sizeof(error) - sizeof(message_size)) {
 309         error_setg(errp, "Protocol error: server sent structured error chunk "
 310                          "with incorrect message size");
 311         return -EINVAL;
 312     }
 313
 314     /* TODO: Add a trace point to mention the server complaint */
 315
 316     /* TODO handle ERROR_OFFSET */
 317
 318     return 0;
 319 }
 320
 321 static int nbd_co_receive_offset_data_payload(NBDClientSession *s,
 322                                               uint64_t orig_offset,
 323                                               QEMUIOVector *qiov, Error **errp)
 324 {
 325     QEMUIOVector sub_qiov;
 326     uint64_t offset;
 327     size_t data_size;
 328     int ret;
 329     NBDStructuredReplyChunk *chunk = &s->reply.structured;
 330
 331     assert(nbd_reply_is_structured(&s->reply));
 332
 333     /* The NBD spec requires at least one byte of payload */
 334     if (chunk->length <= sizeof(offset)) {
 335         error_setg(errp, "Protocol error: invalid payload for "
 336                          "NBD_REPLY_TYPE_OFFSET_DATA");
 337         return -EINVAL;
 338     }
 339
 340     if (nbd_read(s->ioc, &offset, sizeof(offset), errp) < 0) {
 341         return -EIO;
 342     }
 343     be64_to_cpus(&offset);
 344
 345     data_size = chunk->length - sizeof(offset);
 346     assert(data_size);
 347     if (offset < orig_offset || data_size > qiov->size ||
 348         offset > orig_offset + qiov->size - data_size) {
 349         error_setg(errp, "Protocol error: server sent chunk exceeding requested"
 350                          " region");
 351         return -EINVAL;
 352     }
 353
 354     qemu_iovec_init(&sub_qiov, qiov->niov);
 355     qemu_iovec_concat(&sub_qiov, qiov, offset - orig_offset, data_size);
 356     ret = qio_channel_readv_all(s->ioc, sub_qiov.iov, sub_qiov.niov, errp);
 357     qemu_iovec_destroy(&sub_qiov);
 358
 359     return ret < 0 ? -EIO : 0;
 360 }
 361
 362 #define NBD_MAX_MALLOC_PAYLOAD 1000
 363 /* nbd_co_receive_structured_payload
 364  */
 365 static coroutine_fn int nbd_co_receive_structured_payload(
 366         NBDClientSession *s, void **payload, Error **errp)
 367 {
 368     int ret;
 369     uint32_t len;
 370
 371     assert(nbd_reply_is_structured(&s->reply));
 372
 373     len = s->reply.structured.length;
 374
 375     if (len == 0) {
 376         return 0;
 377     }
 378
 379     if (payload == NULL) {
 380         error_setg(errp, "Unexpected structured payload");
 381         return -EINVAL;
 382     }
 383
 384     if (len > NBD_MAX_MALLOC_PAYLOAD) {
 385         error_setg(errp, "Payload too large");
 386         return -EINVAL;
 387     }
 388
 389     *payload = g_new(char, len);
 390     ret = nbd_read(s->ioc, *payload, len, errp);
 391     if (ret < 0) {
 392         g_free(*payload);
 393         *payload = NULL;
 394         return ret;
 395     }
 396
 397     return 0;
 398 }
 399
 400 /* nbd_co_do_receive_one_chunk
 401  * for simple reply:
 402  *   set request_ret to received reply error
 403  *   if qiov is not NULL: read payload to @qiov
 404  * for structured reply chunk:
 405  *   if error chunk: read payload, set @request_ret, do not set @payload
 406  *   else if offset_data chunk: read payload data to @qiov, do not set @payload
 407  *   else: read payload to @payload
 408  *
 409  * If function fails, @errp contains corresponding error message, and the
 410  * connection with the server is suspect.  If it returns 0, then the
 411  * transaction succeeded (although @request_ret may be a negative errno
 412  * corresponding to the server's error reply), and errp is unchanged.
 413  */
 414 static coroutine_fn int nbd_co_do_receive_one_chunk(
 415         NBDClientSession *s, uint64_t handle, bool only_structured,
 416         int *request_ret, QEMUIOVector *qiov, void **payload, Error **errp)
 417 {
 418     int ret;
 419     int i = HANDLE_TO_INDEX(s, handle);
 420     void *local_payload = NULL;
 421     NBDStructuredReplyChunk *chunk;
 422
 423     if (payload) {
 424         *payload = NULL;
 425     }
 426     *request_ret = 0;
 427
 428     /* Wait until we're woken up by nbd_read_reply_entry.  */
 429     s->requests[i].receiving = true;
 430     qemu_coroutine_yield();
 431     s->requests[i].receiving = false;
 432     if (!s->ioc || s->quit) {
 433         error_setg(errp, "Connection closed");
 434         return -EIO;
 435     }
 436
 437     assert(s->reply.handle == handle);
 438
 439     if (nbd_reply_is_simple(&s->reply)) {
 440         if (only_structured) {
 441             error_setg(errp, "Protocol error: simple reply when structured "
 442                              "reply chunk was expected");
 443             return -EINVAL;
 444         }
 445
 446         *request_ret = -nbd_errno_to_system_errno(s->reply.simple.error);
 447         if (*request_ret < 0 || !qiov) {
 448             return 0;
 449         }
 450
 451         return qio_channel_readv_all(s->ioc, qiov->iov, qiov->niov,
 452                                      errp) < 0 ? -EIO : 0;
 453     }
 454
 455     /* handle structured reply chunk */
 456     assert(s->info.structured_reply);
 457     chunk = &s->reply.structured;
 458
 459     if (chunk->type == NBD_REPLY_TYPE_NONE) {
 460         if (!(chunk->flags & NBD_REPLY_FLAG_DONE)) {
 461             error_setg(errp, "Protocol error: NBD_REPLY_TYPE_NONE chunk without"
 462                        " NBD_REPLY_FLAG_DONE flag set");
 463             return -EINVAL;
 464         }
 465         if (chunk->length) {
 466             error_setg(errp, "Protocol error: NBD_REPLY_TYPE_NONE chunk with"
 467                        " nonzero length");
 468             return -EINVAL;
 469         }
 470         return 0;
 471     }
 472
 473     if (chunk->type == NBD_REPLY_TYPE_OFFSET_DATA) {
 474         if (!qiov) {
 475             error_setg(errp, "Unexpected NBD_REPLY_TYPE_OFFSET_DATA chunk");
 476             return -EINVAL;
 477         }
 478
 479         return nbd_co_receive_offset_data_payload(s, s->requests[i].offset,
 480                                                   qiov, errp);
 481     }
 482
 483     if (nbd_reply_type_is_error(chunk->type)) {
 484         payload = &local_payload;
 485     }
 486
 487     ret = nbd_co_receive_structured_payload(s, payload, errp);
 488     if (ret < 0) {
 489         return ret;
 490     }
 491
 492     if (nbd_reply_type_is_error(chunk->type)) {
 493         ret = nbd_parse_error_payload(chunk, local_payload, request_ret, errp);
 494         g_free(local_payload);
 495         return ret;
 496     }
 497
 498     return 0;
 499 }
 500
 501 /* nbd_co_receive_one_chunk
 502  * Read reply, wake up read_reply_co and set s->quit if needed.
 503  * Return value is a fatal error code or normal nbd reply error code
 504  */
 505 static coroutine_fn int nbd_co_receive_one_chunk(
 506         NBDClientSession *s, uint64_t handle, bool only_structured,
 507         QEMUIOVector *qiov, NBDReply *reply, void **payload, Error **errp)
 508 {
 509     int request_ret;
 510     int ret = nbd_co_do_receive_one_chunk(s, handle, only_structured,
 511                                           &request_ret, qiov, payload, errp);
 512
 513     if (ret < 0) {
 514         s->quit = true;
 515     } else {
 516         /* For assert at loop start in nbd_read_reply_entry */
 517         if (reply) {
 518             *reply = s->reply;
 519         }
 520         s->reply.handle = 0;
 521         ret = request_ret;
 522     }
 523
 524     if (s->read_reply_co) {
 525         aio_co_wake(s->read_reply_co);
 526     }
 527
 528     return ret;
 529 }
 530
 531 typedef struct NBDReplyChunkIter {
 532     int ret;
 533     bool fatal;
 534     Error *err;
 535     bool done, only_structured;
 536 } NBDReplyChunkIter;
 537
 538 static void nbd_iter_error(NBDReplyChunkIter *iter, bool fatal,
 539                            int ret, Error **local_err)
 540 {
 541     assert(ret < 0);
 542
 543     if ((fatal && !iter->fatal) || iter->ret == 0) {
 544         if (iter->ret != 0) {
 545             error_free(iter->err);
 546             iter->err = NULL;
 547         }
 548         iter->fatal = fatal;
 549         iter->ret = ret;
 550         error_propagate(&iter->err, *local_err);
 551     } else {
 552         error_free(*local_err);
 553     }
 554
 555     *local_err = NULL;
 556 }
 557
 558 /* NBD_FOREACH_REPLY_CHUNK
 559  */
 560 #define NBD_FOREACH_REPLY_CHUNK(s, iter, handle, structured, \
 561                                 qiov, reply, payload) \
 562     for (iter = (NBDReplyChunkIter) { .only_structured = structured }; \
 563          nbd_reply_chunk_iter_receive(s, &iter, handle, qiov, reply, payload);)
 564
 565 /* nbd_reply_chunk_iter_receive
 566  */
 567 static bool nbd_reply_chunk_iter_receive(NBDClientSession *s,
 568                                          NBDReplyChunkIter *iter,
 569                                          uint64_t handle,
 570                                          QEMUIOVector *qiov, NBDReply *reply,
 571                                          void **payload)
 572 {
 573     int ret;
 574     NBDReply local_reply;
 575     NBDStructuredReplyChunk *chunk;
 576     Error *local_err = NULL;
 577     if (s->quit) {
 578         error_setg(&local_err, "Connection closed");
 579         nbd_iter_error(iter, true, -EIO, &local_err);
 580         goto break_loop;
 581     }
 582
 583     if (iter->done) {
 584         /* Previous iteration was last. */
 585         goto break_loop;
 586     }
 587
 588     if (reply == NULL) {
 589         reply = &local_reply;
 590     }
 591
 592     ret = nbd_co_receive_one_chunk(s, handle, iter->only_structured,
 593                                    qiov, reply, payload, &local_err);
 594     if (ret < 0) {
 595         /* If it is a fatal error s->quit is set by nbd_co_receive_one_chunk */
 596         nbd_iter_error(iter, s->quit, ret, &local_err);
 597     }
 598
 599     /* Do not execute the body of NBD_FOREACH_REPLY_CHUNK for simple reply. */
 600     if (nbd_reply_is_simple(&s->reply) || s->quit) {
 601         goto break_loop;
 602     }
 603
 604     chunk = &reply->structured;
 605     iter->only_structured = true;
 606
 607     if (chunk->type == NBD_REPLY_TYPE_NONE) {
 608         /* NBD_REPLY_FLAG_DONE is already checked in nbd_co_receive_one_chunk */
 609         assert(chunk->flags & NBD_REPLY_FLAG_DONE);
 610         goto break_loop;
 611     }
 612
 613     if (chunk->flags & NBD_REPLY_FLAG_DONE) {
 614         /* This iteration is last. */
 615         iter->done = true;
 616     }
 617
 618     /* Execute the loop body */
 619     return true;
 620
 621 break_loop:
 622     s->requests[HANDLE_TO_INDEX(s, handle)].coroutine = NULL;
 623
 624     qemu_co_mutex_lock(&s->send_mutex);
 625     s->in_flight--;
 626     qemu_co_queue_next(&s->free_sema);
 627     qemu_co_mutex_unlock(&s->send_mutex);
 628
 629     return false;
 630 }
 631
 632 static int nbd_co_receive_return_code(NBDClientSession *s, uint64_t handle,
 633                                       Error **errp)
 634 {
 635     NBDReplyChunkIter iter;
 636
 637     NBD_FOREACH_REPLY_CHUNK(s, iter, handle, false, NULL, NULL, NULL) {
 638         /* nbd_reply_chunk_iter_receive does all the work */
 639     }
 640
 641     error_propagate(errp, iter.err);
 642     return iter.ret;
 643 }
 644
 645 static int nbd_co_receive_cmdread_reply(NBDClientSession *s, uint64_t handle,
 646                                         uint64_t offset, QEMUIOVector *qiov,
 647                                         Error **errp)
 648 {
 649     NBDReplyChunkIter iter;
 650     NBDReply reply;
 651     void *payload = NULL;
 652     Error *local_err = NULL;
 653
 654     NBD_FOREACH_REPLY_CHUNK(s, iter, handle, s->info.structured_reply,
 655                             qiov, &reply, &payload)
 656     {
 657         int ret;
 658         NBDStructuredReplyChunk *chunk = &reply.structured;
 659
 660         assert(nbd_reply_is_structured(&reply));
 661
 662         switch (chunk->type) {
 663         case NBD_REPLY_TYPE_OFFSET_DATA:
 664             /* special cased in nbd_co_receive_one_chunk, data is already
 665              * in qiov */
 666             break;
 667         case NBD_REPLY_TYPE_OFFSET_HOLE:
 668             ret = nbd_parse_offset_hole_payload(&reply.structured, payload,
 669                                                 offset, qiov, &local_err);
 670             if (ret < 0) {
 671                 s->quit = true;
 672                 nbd_iter_error(&iter, true, ret, &local_err);
 673             }
 674             break;
 675         default:
 676             if (!nbd_reply_type_is_error(chunk->type)) {
 677                 /* not allowed reply type */
 678                 s->quit = true;
 679                 error_setg(&local_err,
 680                            "Unexpected reply type: %d (%s) for CMD_READ",
 681                            chunk->type, nbd_reply_type_lookup(chunk->type));
 682                 nbd_iter_error(&iter, true, -EINVAL, &local_err);
 683             }
 684         }
 685
 686         g_free(payload);
 687         payload = NULL;
 688     }
 689
 690     error_propagate(errp, iter.err);
 691     return iter.ret;
 692 }
 693
 694 static int nbd_co_receive_blockstatus_reply(NBDClientSession *s,
 695                                             uint64_t handle, uint64_t length,
 696                                             NBDExtent *extent, Error **errp)
 697 {
 698     NBDReplyChunkIter iter;
 699     NBDReply reply;
 700     void *payload = NULL;
 701     Error *local_err = NULL;
 702     bool received = false;
 703
 704     assert(!extent->length);
 705     NBD_FOREACH_REPLY_CHUNK(s, iter, handle, s->info.structured_reply,
 706                             NULL, &reply, &payload)
 707     {
 708         int ret;
 709         NBDStructuredReplyChunk *chunk = &reply.structured;
 710
 711         assert(nbd_reply_is_structured(&reply));
 712
 713         switch (chunk->type) {
 714         case NBD_REPLY_TYPE_BLOCK_STATUS:
 715             if (received) {
 716                 s->quit = true;
 717                 error_setg(&local_err, "Several BLOCK_STATUS chunks in reply");
 718                 nbd_iter_error(&iter, true, -EINVAL, &local_err);
 719             }
 720             received = true;
 721
 722             ret = nbd_parse_blockstatus_payload(s, &reply.structured,
 723                                                 payload, length, extent,
 724                                                 &local_err);
 725             if (ret < 0) {
 726                 s->quit = true;
 727                 nbd_iter_error(&iter, true, ret, &local_err);
 728             }
 729             break;
 730         default:
 731             if (!nbd_reply_type_is_error(chunk->type)) {
 732                 s->quit = true;
 733                 error_setg(&local_err,
 734                            "Unexpected reply type: %d (%s) "
 735                            "for CMD_BLOCK_STATUS",
 736                            chunk->type, nbd_reply_type_lookup(chunk->type));
 737                 nbd_iter_error(&iter, true, -EINVAL, &local_err);
 738             }
 739         }
 740
 741         g_free(payload);
 742         payload = NULL;
 743     }
 744
 745     if (!extent->length && !iter.err) {
 746         error_setg(&iter.err,
 747                    "Server did not reply with any status extents");
 748         if (!iter.ret) {
 749             iter.ret = -EIO;
 750         }
 751     }
 752     error_propagate(errp, iter.err);
 753     return iter.ret;
 754 }
 755
 756 static int nbd_co_request(BlockDriverState *bs, NBDRequest *request,
 757                           QEMUIOVector *write_qiov)
 758 {
 759     int ret;
 760     Error *local_err = NULL;
 761     NBDClientSession *client = nbd_get_client_session(bs);
 762
 763     assert(request->type != NBD_CMD_READ);
 764     if (write_qiov) {
 765         assert(request->type == NBD_CMD_WRITE);
 766         assert(request->len == iov_size(write_qiov->iov, write_qiov->niov));
 767     } else {
 768         assert(request->type != NBD_CMD_WRITE);
 769     }
 770     ret = nbd_co_send_request(bs, request, write_qiov);
 771     if (ret < 0) {
 772         return ret;
 773     }
 774
 775     ret = nbd_co_receive_return_code(client, request->handle, &local_err);
 776     if (local_err) {
 777         trace_nbd_co_request_fail(request->from, request->len, request->handle,
 778                                   request->flags, request->type,
 779                                   nbd_cmd_lookup(request->type),
 780                                   ret, error_get_pretty(local_err));
 781         error_free(local_err);
 782     }
 783     return ret;
 784 }
 785
 786 int nbd_client_co_preadv(BlockDriverState *bs, uint64_t offset,
 787                          uint64_t bytes, QEMUIOVector *qiov, int flags)
 788 {
 789     int ret;
 790     Error *local_err = NULL;
 791     NBDClientSession *client = nbd_get_client_session(bs);
 792     NBDRequest request = {
 793         .type = NBD_CMD_READ,
 794         .from = offset,
 795         .len = bytes,
 796     };
 797
 798     assert(bytes <= NBD_MAX_BUFFER_SIZE);
 799     assert(!flags);
 800
 801     if (!bytes) {
 802         return 0;
 803     }
 804     ret = nbd_co_send_request(bs, &request, NULL);
 805     if (ret < 0) {
 806         return ret;
 807     }
 808
 809     ret = nbd_co_receive_cmdread_reply(client, request.handle, offset, qiov,
 810                                        &local_err);
 811     if (local_err) {
 812         trace_nbd_co_request_fail(request.from, request.len, request.handle,
 813                                   request.flags, request.type,
 814                                   nbd_cmd_lookup(request.type),
 815                                   ret, error_get_pretty(local_err));
 816         error_free(local_err);
 817     }
 818     return ret;
 819 }
 820
 821 int nbd_client_co_pwritev(BlockDriverState *bs, uint64_t offset,
 822                           uint64_t bytes, QEMUIOVector *qiov, int flags)
 823 {
 824     NBDClientSession *client = nbd_get_client_session(bs);
 825     NBDRequest request = {
 826         .type = NBD_CMD_WRITE,
 827         .from = offset,
 828         .len = bytes,
 829     };
 830
 831     assert(!(client->info.flags & NBD_FLAG_READ_ONLY));
 832     if (flags & BDRV_REQ_FUA) {
 833         assert(client->info.flags & NBD_FLAG_SEND_FUA);
 834         request.flags |= NBD_CMD_FLAG_FUA;
 835     }
 836
 837     assert(bytes <= NBD_MAX_BUFFER_SIZE);
 838
 839     if (!bytes) {
 840         return 0;
 841     }
 842     return nbd_co_request(bs, &request, qiov);
 843 }
 844
 845 int nbd_client_co_pwrite_zeroes(BlockDriverState *bs, int64_t offset,
 846                                 int bytes, BdrvRequestFlags flags)
 847 {
 848     NBDClientSession *client = nbd_get_client_session(bs);
 849     NBDRequest request = {
 850         .type = NBD_CMD_WRITE_ZEROES,
 851         .from = offset,
 852         .len = bytes,
 853     };
 854
 855     assert(!(client->info.flags & NBD_FLAG_READ_ONLY));
 856     if (!(client->info.flags & NBD_FLAG_SEND_WRITE_ZEROES)) {
 857         return -ENOTSUP;
 858     }
 859
 860     if (flags & BDRV_REQ_FUA) {
 861         assert(client->info.flags & NBD_FLAG_SEND_FUA);
 862         request.flags |= NBD_CMD_FLAG_FUA;
 863     }
 864     if (!(flags & BDRV_REQ_MAY_UNMAP)) {
 865         request.flags |= NBD_CMD_FLAG_NO_HOLE;
 866     }
 867
 868     if (!bytes) {
 869         return 0;
 870     }
 871     return nbd_co_request(bs, &request, NULL);
 872 }
 873
 874 int nbd_client_co_flush(BlockDriverState *bs)
 875 {
 876     NBDClientSession *client = nbd_get_client_session(bs);
 877     NBDRequest request = { .type = NBD_CMD_FLUSH };
 878
 879     if (!(client->info.flags & NBD_FLAG_SEND_FLUSH)) {
 880         return 0;
 881     }
 882
 883     request.from = 0;
 884     request.len = 0;
 885
 886     return nbd_co_request(bs, &request, NULL);
 887 }
 888
 889 int nbd_client_co_pdiscard(BlockDriverState *bs, int64_t offset, int bytes)
 890 {
 891     NBDClientSession *client = nbd_get_client_session(bs);
 892     NBDRequest request = {
 893         .type = NBD_CMD_TRIM,
 894         .from = offset,
 895         .len = bytes,
 896     };
 897
 898     assert(!(client->info.flags & NBD_FLAG_READ_ONLY));
 899     if (!(client->info.flags & NBD_FLAG_SEND_TRIM) || !bytes) {
 900         return 0;
 901     }
 902
 903     return nbd_co_request(bs, &request, NULL);
 904 }
 905
 906 int coroutine_fn nbd_client_co_block_status(BlockDriverState *bs,
 907                                             bool want_zero,
 908                                             int64_t offset, int64_t bytes,
 909                                             int64_t *pnum, int64_t *map,
 910                                             BlockDriverState **file)
 911 {
 912     int64_t ret;
 913     NBDExtent extent = { 0 };
 914     NBDClientSession *client = nbd_get_client_session(bs);
 915     Error *local_err = NULL;
 916
 917     NBDRequest request = {
 918         .type = NBD_CMD_BLOCK_STATUS,
 919         .from = offset,
 920         .len = MIN(MIN_NON_ZERO(QEMU_ALIGN_DOWN(INT_MAX,
 921                                                 bs->bl.request_alignment),
 922                                 client->info.max_block), bytes),
 923         .flags = NBD_CMD_FLAG_REQ_ONE,
 924     };
 925
 926     if (!client->info.base_allocation) {
 927         *pnum = bytes;
 928         return BDRV_BLOCK_DATA;
 929     }
 930
 931     ret = nbd_co_send_request(bs, &request, NULL);
 932     if (ret < 0) {
 933         return ret;
 934     }
 935
 936     ret = nbd_co_receive_blockstatus_reply(client, request.handle, bytes,
 937                                            &extent, &local_err);
 938     if (local_err) {
 939         trace_nbd_co_request_fail(request.from, request.len, request.handle,
 940                                   request.flags, request.type,
 941                                   nbd_cmd_lookup(request.type),
 942                                   ret, error_get_pretty(local_err));
 943         error_free(local_err);
 944     }
 945     if (ret < 0) {
 946         return ret;
 947     }
 948
 949     assert(extent.length);
 950     *pnum = extent.length;
 951     return (extent.flags & NBD_STATE_HOLE ? 0 : BDRV_BLOCK_DATA) |
 952            (extent.flags & NBD_STATE_ZERO ? BDRV_BLOCK_ZERO : 0);
 953 }
 954
 955 void nbd_client_detach_aio_context(BlockDriverState *bs)
 956 {
 957     NBDClientSession *client = nbd_get_client_session(bs);
 958     qio_channel_detach_aio_context(QIO_CHANNEL(client->ioc));
 959 }
 960
 961 void nbd_client_attach_aio_context(BlockDriverState *bs,
 962                                    AioContext *new_context)
 963 {
 964     NBDClientSession *client = nbd_get_client_session(bs);
 965     qio_channel_attach_aio_context(QIO_CHANNEL(client->ioc), new_context);
 966     aio_co_schedule(new_context, client->read_reply_co);
 967 }
 968
 969 void nbd_client_close(BlockDriverState *bs)
 970 {
 971     NBDClientSession *client = nbd_get_client_session(bs);
 972     NBDRequest request = { .type = NBD_CMD_DISC };
 973
 974     if (client->ioc == NULL) {
 975         return;
 976     }
 977
 978     nbd_send_request(client->ioc, &request);
 979
 980     nbd_teardown_connection(bs);
 981 }
 982
 983 int nbd_client_init(BlockDriverState *bs,
 984                     QIOChannelSocket *sioc,
 985                     const char *export,
 986                     QCryptoTLSCreds *tlscreds,
 987                     const char *hostname,
 988                     const char *x_dirty_bitmap,
 989                     Error **errp)
 990 {
 991     NBDClientSession *client = nbd_get_client_session(bs);
 992     int ret;
 993
 994     /* NBD handshake */
 995     logout("session init %s\n", export);
 996     qio_channel_set_blocking(QIO_CHANNEL(sioc), true, NULL);
 997
 998     client->info.request_sizes = true;
 999     client->info.structured_reply = true;
1000     client->info.base_allocation = true;
1001     client->info.x_dirty_bitmap = g_strdup(x_dirty_bitmap);
1002     ret = nbd_receive_negotiate(QIO_CHANNEL(sioc), export,
1003                                 tlscreds, hostname,
1004                                 &client->ioc, &client->info, errp);
1005     g_free(client->info.x_dirty_bitmap);
1006     if (ret < 0) {
1007         logout("Failed to negotiate with the NBD server\n");
1008         return ret;
1009     }
1010     if (x_dirty_bitmap && !client->info.base_allocation) {
1011         error_setg(errp, "requested x-dirty-bitmap %s not found",
1012                    x_dirty_bitmap);
1013         ret = -EINVAL;
1014         goto fail;
1015     }
1016     if (client->info.flags & NBD_FLAG_READ_ONLY) {
1017         ret = bdrv_apply_auto_read_only(bs, "NBD export is read-only", errp);
1018         if (ret < 0) {
1019             goto fail;
1020         }
1021     }
1022     if (client->info.flags & NBD_FLAG_SEND_FUA) {
1023         bs->supported_write_flags = BDRV_REQ_FUA;
1024         bs->supported_zero_flags |= BDRV_REQ_FUA;
1025     }
1026     if (client->info.flags & NBD_FLAG_SEND_WRITE_ZEROES) {
1027         bs->supported_zero_flags |= BDRV_REQ_MAY_UNMAP;
1028     }
1029
1030     qemu_co_mutex_init(&client->send_mutex);
1031     qemu_co_queue_init(&client->free_sema);
1032     client->sioc = sioc;
1033     object_ref(OBJECT(client->sioc));
1034
1035     if (!client->ioc) {
1036         client->ioc = QIO_CHANNEL(sioc);
1037         object_ref(OBJECT(client->ioc));
1038     }
1039
1040     /* Now that we're connected, set the socket to be non-blocking and
1041      * kick the reply mechanism.  */
1042     qio_channel_set_blocking(QIO_CHANNEL(sioc), false, NULL);
1043     client->read_reply_co = qemu_coroutine_create(nbd_read_reply_entry, client);
1044     nbd_client_attach_aio_context(bs, bdrv_get_aio_context(bs));
1045
1046     logout("Established connection with NBD server\n");
1047     return 0;
1048
1049  fail:
1050     /*
1051      * We have connected, but must fail for other reasons. The
1052      * connection is still blocking; send NBD_CMD_DISC as a courtesy
1053      * to the server.
1054      */
1055     {
1056         NBDRequest request = { .type = NBD_CMD_DISC };
1057
1058         nbd_send_request(client->ioc ?: QIO_CHANNEL(sioc), &request);
1059         return ret;
1060     }
1061 }