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