nbd.c

   1 /*
   2  *  Copyright (C) 2005  Anthony Liguori <anthony@codemonkey.ws>
   3  *
   4  *  Network Block Device
   5  *
   6  *  This program is free software; you can redistribute it and/or modify
   7  *  it under the terms of the GNU General Public License as published by
   8  *  the Free Software Foundation; under version 2 of the License.
   9  *
  10  *  This program is distributed in the hope that it will be useful,
  11  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  *  GNU General Public License for more details.
  14  *
  15  *  You should have received a copy of the GNU General Public License
  16  *  along with this program; if not, see <http://www.gnu.org/licenses/>.
  17  */
  18
  19 #include "nbd.h"
  20 #include "block.h"
  21 #include "block_int.h"
  22
  23 #include "qemu-coroutine.h"
  24
  25 #include <errno.h>
  26 #include <string.h>
  27 #ifndef _WIN32
  28 #include <sys/ioctl.h>
  29 #endif
  30 #if defined(__sun__) || defined(__HAIKU__)
  31 #include <sys/ioccom.h>
  32 #endif
  33 #include <ctype.h>
  34 #include <inttypes.h>
  35
  36 #ifdef __linux__
  37 #include <linux/fs.h>
  38 #endif
  39
  40 #include "qemu_socket.h"
  41 #include "qemu-queue.h"
  42
  43 //#define DEBUG_NBD
  44
  45 #ifdef DEBUG_NBD
  46 #define TRACE(msg, ...) do { \
  47     LOG(msg, ## __VA_ARGS__); \
  48 } while(0)
  49 #else
  50 #define TRACE(msg, ...) \
  51     do { } while (0)
  52 #endif
  53
  54 #define LOG(msg, ...) do { \
  55     fprintf(stderr, "%s:%s():L%d: " msg "\n", \
  56             __FILE__, __FUNCTION__, __LINE__, ## __VA_ARGS__); \
  57 } while(0)
  58
  59 /* This is all part of the "official" NBD API */
  60
  61 #define NBD_REPLY_SIZE          (4 + 4 + 8)
  62 #define NBD_REQUEST_MAGIC       0x25609513
  63 #define NBD_REPLY_MAGIC         0x67446698
  64
  65 #define NBD_SET_SOCK            _IO(0xab, 0)
  66 #define NBD_SET_BLKSIZE         _IO(0xab, 1)
  67 #define NBD_SET_SIZE            _IO(0xab, 2)
  68 #define NBD_DO_IT               _IO(0xab, 3)
  69 #define NBD_CLEAR_SOCK          _IO(0xab, 4)
  70 #define NBD_CLEAR_QUE           _IO(0xab, 5)
  71 #define NBD_PRINT_DEBUG         _IO(0xab, 6)
  72 #define NBD_SET_SIZE_BLOCKS     _IO(0xab, 7)
  73 #define NBD_DISCONNECT          _IO(0xab, 8)
  74 #define NBD_SET_TIMEOUT         _IO(0xab, 9)
  75 #define NBD_SET_FLAGS           _IO(0xab, 10)
  76
  77 #define NBD_OPT_EXPORT_NAME     (1 << 0)
  78
  79 /* That's all folks */
  80
  81 ssize_t nbd_wr_sync(int fd, void *buffer, size_t size, bool do_read)
  82 {
  83     size_t offset = 0;
  84     int err;
  85
  86     if (qemu_in_coroutine()) {
  87         if (do_read) {
  88             return qemu_co_recv(fd, buffer, size);
  89         } else {
  90             return qemu_co_send(fd, buffer, size);
  91         }
  92     }
  93
  94     while (offset < size) {
  95         ssize_t len;
  96
  97         if (do_read) {
  98             len = qemu_recv(fd, buffer + offset, size - offset, 0);
  99         } else {
 100             len = send(fd, buffer + offset, size - offset, 0);
 101         }
 102
 103         if (len < 0) {
 104             err = socket_error();
 105
 106             /* recoverable error */
 107             if (err == EINTR || (offset > 0 && err == EAGAIN)) {
 108                 continue;
 109             }
 110
 111             /* unrecoverable error */
 112             return -err;
 113         }
 114
 115         /* eof */
 116         if (len == 0) {
 117             break;
 118         }
 119
 120         offset += len;
 121     }
 122
 123     return offset;
 124 }
 125
 126 static ssize_t read_sync(int fd, void *buffer, size_t size)
 127 {
 128     /* Sockets are kept in blocking mode in the negotiation phase.  After
 129      * that, a non-readable socket simply means that another thread stole
 130      * our request/reply.  Synchronization is done with recv_coroutine, so
 131      * that this is coroutine-safe.
 132      */
 133     return nbd_wr_sync(fd, buffer, size, true);
 134 }
 135
 136 static ssize_t write_sync(int fd, void *buffer, size_t size)
 137 {
 138     int ret;
 139     do {
 140         /* For writes, we do expect the socket to be writable.  */
 141         ret = nbd_wr_sync(fd, buffer, size, false);
 142     } while (ret == -EAGAIN);
 143     return ret;
 144 }
 145
 146 static void combine_addr(char *buf, size_t len, const char* address,
 147                          uint16_t port)
 148 {
 149     /* If the address-part contains a colon, it's an IPv6 IP so needs [] */
 150     if (strstr(address, ":")) {
 151         snprintf(buf, len, "[%s]:%u", address, port);
 152     } else {
 153         snprintf(buf, len, "%s:%u", address, port);
 154     }
 155 }
 156
 157 int tcp_socket_outgoing(const char *address, uint16_t port)
 158 {
 159     char address_and_port[128];
 160     combine_addr(address_and_port, 128, address, port);
 161     return tcp_socket_outgoing_spec(address_and_port);
 162 }
 163
 164 int tcp_socket_outgoing_spec(const char *address_and_port)
 165 {
 166     return inet_connect(address_and_port, SOCK_STREAM);
 167 }
 168
 169 int tcp_socket_incoming(const char *address, uint16_t port)
 170 {
 171     char address_and_port[128];
 172     combine_addr(address_and_port, 128, address, port);
 173     return tcp_socket_incoming_spec(address_and_port);
 174 }
 175
 176 int tcp_socket_incoming_spec(const char *address_and_port)
 177 {
 178     char *ostr  = NULL;
 179     int olen = 0;
 180     return inet_listen(address_and_port, ostr, olen, SOCK_STREAM, 0);
 181 }
 182
 183 int unix_socket_incoming(const char *path)
 184 {
 185     char *ostr = NULL;
 186     int olen = 0;
 187
 188     return unix_listen(path, ostr, olen);
 189 }
 190
 191 int unix_socket_outgoing(const char *path)
 192 {
 193     return unix_connect(path);
 194 }
 195
 196 /* Basic flow
 197
 198    Server         Client
 199
 200    Negotiate
 201                   Request
 202    Response
 203                   Request
 204    Response
 205                   ...
 206    ...
 207                   Request (type == 2)
 208 */
 209
 210 static int nbd_send_negotiate(int csock, off_t size, uint32_t flags)
 211 {
 212     char buf[8 + 8 + 8 + 128];
 213     int rc;
 214
 215     /* Negotiate
 216         [ 0 ..   7]   passwd   ("NBDMAGIC")
 217         [ 8 ..  15]   magic    (0x00420281861253)
 218         [16 ..  23]   size
 219         [24 ..  27]   flags
 220         [28 .. 151]   reserved (0)
 221      */
 222
 223     socket_set_block(csock);
 224     rc = -EINVAL;
 225
 226     TRACE("Beginning negotiation.");
 227     memcpy(buf, "NBDMAGIC", 8);
 228     cpu_to_be64w((uint64_t*)(buf + 8), 0x00420281861253LL);
 229     cpu_to_be64w((uint64_t*)(buf + 16), size);
 230     cpu_to_be32w((uint32_t*)(buf + 24),
 231                  flags | NBD_FLAG_HAS_FLAGS | NBD_FLAG_SEND_TRIM |
 232                  NBD_FLAG_SEND_FLUSH | NBD_FLAG_SEND_FUA);
 233     memset(buf + 28, 0, 124);
 234
 235     if (write_sync(csock, buf, sizeof(buf)) != sizeof(buf)) {
 236         LOG("write failed");
 237         goto fail;
 238     }
 239
 240     TRACE("Negotiation succeeded.");
 241     rc = 0;
 242 fail:
 243     socket_set_nonblock(csock);
 244     return rc;
 245 }
 246
 247 int nbd_receive_negotiate(int csock, const char *name, uint32_t *flags,
 248                           off_t *size, size_t *blocksize)
 249 {
 250     char buf[256];
 251     uint64_t magic, s;
 252     uint16_t tmp;
 253     int rc;
 254
 255     TRACE("Receiving negotiation.");
 256
 257     socket_set_block(csock);
 258     rc = -EINVAL;
 259
 260     if (read_sync(csock, buf, 8) != 8) {
 261         LOG("read failed");
 262         goto fail;
 263     }
 264
 265     buf[8] = '\0';
 266     if (strlen(buf) == 0) {
 267         LOG("server connection closed");
 268         goto fail;
 269     }
 270
 271     TRACE("Magic is %c%c%c%c%c%c%c%c",
 272           qemu_isprint(buf[0]) ? buf[0] : '.',
 273           qemu_isprint(buf[1]) ? buf[1] : '.',
 274           qemu_isprint(buf[2]) ? buf[2] : '.',
 275           qemu_isprint(buf[3]) ? buf[3] : '.',
 276           qemu_isprint(buf[4]) ? buf[4] : '.',
 277           qemu_isprint(buf[5]) ? buf[5] : '.',
 278           qemu_isprint(buf[6]) ? buf[6] : '.',
 279           qemu_isprint(buf[7]) ? buf[7] : '.');
 280
 281     if (memcmp(buf, "NBDMAGIC", 8) != 0) {
 282         LOG("Invalid magic received");
 283         goto fail;
 284     }
 285
 286     if (read_sync(csock, &magic, sizeof(magic)) != sizeof(magic)) {
 287         LOG("read failed");
 288         goto fail;
 289     }
 290     magic = be64_to_cpu(magic);
 291     TRACE("Magic is 0x%" PRIx64, magic);
 292
 293     if (name) {
 294         uint32_t reserved = 0;
 295         uint32_t opt;
 296         uint32_t namesize;
 297
 298         TRACE("Checking magic (opts_magic)");
 299         if (magic != 0x49484156454F5054LL) {
 300             LOG("Bad magic received");
 301             goto fail;
 302         }
 303         if (read_sync(csock, &tmp, sizeof(tmp)) != sizeof(tmp)) {
 304             LOG("flags read failed");
 305             goto fail;
 306         }
 307         *flags = be16_to_cpu(tmp) << 16;
 308         /* reserved for future use */
 309         if (write_sync(csock, &reserved, sizeof(reserved)) !=
 310             sizeof(reserved)) {
 311             LOG("write failed (reserved)");
 312             goto fail;
 313         }
 314         /* write the export name */
 315         magic = cpu_to_be64(magic);
 316         if (write_sync(csock, &magic, sizeof(magic)) != sizeof(magic)) {
 317             LOG("write failed (magic)");
 318             goto fail;
 319         }
 320         opt = cpu_to_be32(NBD_OPT_EXPORT_NAME);
 321         if (write_sync(csock, &opt, sizeof(opt)) != sizeof(opt)) {
 322             LOG("write failed (opt)");
 323             goto fail;
 324         }
 325         namesize = cpu_to_be32(strlen(name));
 326         if (write_sync(csock, &namesize, sizeof(namesize)) !=
 327             sizeof(namesize)) {
 328             LOG("write failed (namesize)");
 329             goto fail;
 330         }
 331         if (write_sync(csock, (char*)name, strlen(name)) != strlen(name)) {
 332             LOG("write failed (name)");
 333             goto fail;
 334         }
 335     } else {
 336         TRACE("Checking magic (cli_magic)");
 337
 338         if (magic != 0x00420281861253LL) {
 339             LOG("Bad magic received");
 340             goto fail;
 341         }
 342     }
 343
 344     if (read_sync(csock, &s, sizeof(s)) != sizeof(s)) {
 345         LOG("read failed");
 346         goto fail;
 347     }
 348     *size = be64_to_cpu(s);
 349     *blocksize = 1024;
 350     TRACE("Size is %" PRIu64, *size);
 351
 352     if (!name) {
 353         if (read_sync(csock, flags, sizeof(*flags)) != sizeof(*flags)) {
 354             LOG("read failed (flags)");
 355             goto fail;
 356         }
 357         *flags = be32_to_cpup(flags);
 358     } else {
 359         if (read_sync(csock, &tmp, sizeof(tmp)) != sizeof(tmp)) {
 360             LOG("read failed (tmp)");
 361             goto fail;
 362         }
 363         *flags |= be32_to_cpu(tmp);
 364     }
 365     if (read_sync(csock, &buf, 124) != 124) {
 366         LOG("read failed (buf)");
 367         goto fail;
 368     }
 369     rc = 0;
 370
 371 fail:
 372     socket_set_nonblock(csock);
 373     return rc;
 374 }
 375
 376 #ifdef __linux__
 377 int nbd_init(int fd, int csock, uint32_t flags, off_t size, size_t blocksize)
 378 {
 379     TRACE("Setting NBD socket");
 380
 381     if (ioctl(fd, NBD_SET_SOCK, csock) < 0) {
 382         int serrno = errno;
 383         LOG("Failed to set NBD socket");
 384         return -serrno;
 385     }
 386
 387     TRACE("Setting block size to %lu", (unsigned long)blocksize);
 388
 389     if (ioctl(fd, NBD_SET_BLKSIZE, blocksize) < 0) {
 390         int serrno = errno;
 391         LOG("Failed setting NBD block size");
 392         return -serrno;
 393     }
 394
 395         TRACE("Setting size to %zd block(s)", (size_t)(size / blocksize));
 396
 397     if (ioctl(fd, NBD_SET_SIZE_BLOCKS, size / blocksize) < 0) {
 398         int serrno = errno;
 399         LOG("Failed setting size (in blocks)");
 400         return -serrno;
 401     }
 402
 403     if (flags & NBD_FLAG_READ_ONLY) {
 404         int read_only = 1;
 405         TRACE("Setting readonly attribute");
 406
 407         if (ioctl(fd, BLKROSET, (unsigned long) &read_only) < 0) {
 408             int serrno = errno;
 409             LOG("Failed setting read-only attribute");
 410             return -serrno;
 411         }
 412     }
 413
 414     if (ioctl(fd, NBD_SET_FLAGS, flags) < 0
 415         && errno != ENOTTY) {
 416         int serrno = errno;
 417         LOG("Failed setting flags");
 418         return -serrno;
 419     }
 420
 421     TRACE("Negotiation ended");
 422
 423     return 0;
 424 }
 425
 426 int nbd_disconnect(int fd)
 427 {
 428     ioctl(fd, NBD_CLEAR_QUE);
 429     ioctl(fd, NBD_DISCONNECT);
 430     ioctl(fd, NBD_CLEAR_SOCK);
 431     return 0;
 432 }
 433
 434 int nbd_client(int fd)
 435 {
 436     int ret;
 437     int serrno;
 438
 439     TRACE("Doing NBD loop");
 440
 441     ret = ioctl(fd, NBD_DO_IT);
 442     if (ret < 0 && errno == EPIPE) {
 443         /* NBD_DO_IT normally returns EPIPE when someone has disconnected
 444          * the socket via NBD_DISCONNECT.  We do not want to return 1 in
 445          * that case.
 446          */
 447         ret = 0;
 448     }
 449     serrno = errno;
 450
 451     TRACE("NBD loop returned %d: %s", ret, strerror(serrno));
 452
 453     TRACE("Clearing NBD queue");
 454     ioctl(fd, NBD_CLEAR_QUE);
 455
 456     TRACE("Clearing NBD socket");
 457     ioctl(fd, NBD_CLEAR_SOCK);
 458
 459     errno = serrno;
 460     return ret;
 461 }
 462 #else
 463 int nbd_init(int fd, int csock, uint32_t flags, off_t size, size_t blocksize)
 464 {
 465     return -ENOTSUP;
 466 }
 467
 468 int nbd_disconnect(int fd)
 469 {
 470     return -ENOTSUP;
 471 }
 472
 473 int nbd_client(int fd)
 474 {
 475     return -ENOTSUP;
 476 }
 477 #endif
 478
 479 ssize_t nbd_send_request(int csock, struct nbd_request *request)
 480 {
 481     uint8_t buf[4 + 4 + 8 + 8 + 4];
 482     ssize_t ret;
 483
 484     cpu_to_be32w((uint32_t*)buf, NBD_REQUEST_MAGIC);
 485     cpu_to_be32w((uint32_t*)(buf + 4), request->type);
 486     cpu_to_be64w((uint64_t*)(buf + 8), request->handle);
 487     cpu_to_be64w((uint64_t*)(buf + 16), request->from);
 488     cpu_to_be32w((uint32_t*)(buf + 24), request->len);
 489
 490     TRACE("Sending request to client: "
 491           "{ .from = %" PRIu64", .len = %u, .handle = %" PRIu64", .type=%i}",
 492           request->from, request->len, request->handle, request->type);
 493
 494     ret = write_sync(csock, buf, sizeof(buf));
 495     if (ret < 0) {
 496         return ret;
 497     }
 498
 499     if (ret != sizeof(buf)) {
 500         LOG("writing to socket failed");
 501         return -EINVAL;
 502     }
 503     return 0;
 504 }
 505
 506 static ssize_t nbd_receive_request(int csock, struct nbd_request *request)
 507 {
 508     uint8_t buf[4 + 4 + 8 + 8 + 4];
 509     uint32_t magic;
 510     ssize_t ret;
 511
 512     ret = read_sync(csock, buf, sizeof(buf));
 513     if (ret < 0) {
 514         return ret;
 515     }
 516
 517     if (ret != sizeof(buf)) {
 518         LOG("read failed");
 519         return -EINVAL;
 520     }
 521
 522     /* Request
 523        [ 0 ..  3]   magic   (NBD_REQUEST_MAGIC)
 524        [ 4 ..  7]   type    (0 == READ, 1 == WRITE)
 525        [ 8 .. 15]   handle
 526        [16 .. 23]   from
 527        [24 .. 27]   len
 528      */
 529
 530     magic = be32_to_cpup((uint32_t*)buf);
 531     request->type  = be32_to_cpup((uint32_t*)(buf + 4));
 532     request->handle = be64_to_cpup((uint64_t*)(buf + 8));
 533     request->from  = be64_to_cpup((uint64_t*)(buf + 16));
 534     request->len   = be32_to_cpup((uint32_t*)(buf + 24));
 535
 536     TRACE("Got request: "
 537           "{ magic = 0x%x, .type = %d, from = %" PRIu64" , len = %u }",
 538           magic, request->type, request->from, request->len);
 539
 540     if (magic != NBD_REQUEST_MAGIC) {
 541         LOG("invalid magic (got 0x%x)", magic);
 542         return -EINVAL;
 543     }
 544     return 0;
 545 }
 546
 547 ssize_t nbd_receive_reply(int csock, struct nbd_reply *reply)
 548 {
 549     uint8_t buf[NBD_REPLY_SIZE];
 550     uint32_t magic;
 551     ssize_t ret;
 552
 553     ret = read_sync(csock, buf, sizeof(buf));
 554     if (ret < 0) {
 555         return ret;
 556     }
 557
 558     if (ret != sizeof(buf)) {
 559         LOG("read failed");
 560         return -EINVAL;
 561     }
 562
 563     /* Reply
 564        [ 0 ..  3]    magic   (NBD_REPLY_MAGIC)
 565        [ 4 ..  7]    error   (0 == no error)
 566        [ 7 .. 15]    handle
 567      */
 568
 569     magic = be32_to_cpup((uint32_t*)buf);
 570     reply->error  = be32_to_cpup((uint32_t*)(buf + 4));
 571     reply->handle = be64_to_cpup((uint64_t*)(buf + 8));
 572
 573     TRACE("Got reply: "
 574           "{ magic = 0x%x, .error = %d, handle = %" PRIu64" }",
 575           magic, reply->error, reply->handle);
 576
 577     if (magic != NBD_REPLY_MAGIC) {
 578         LOG("invalid magic (got 0x%x)", magic);
 579         return -EINVAL;
 580     }
 581     return 0;
 582 }
 583
 584 static ssize_t nbd_send_reply(int csock, struct nbd_reply *reply)
 585 {
 586     uint8_t buf[4 + 4 + 8];
 587     ssize_t ret;
 588
 589     /* Reply
 590        [ 0 ..  3]    magic   (NBD_REPLY_MAGIC)
 591        [ 4 ..  7]    error   (0 == no error)
 592        [ 7 .. 15]    handle
 593      */
 594     cpu_to_be32w((uint32_t*)buf, NBD_REPLY_MAGIC);
 595     cpu_to_be32w((uint32_t*)(buf + 4), reply->error);
 596     cpu_to_be64w((uint64_t*)(buf + 8), reply->handle);
 597
 598     TRACE("Sending response to client");
 599
 600     ret = write_sync(csock, buf, sizeof(buf));
 601     if (ret < 0) {
 602         return ret;
 603     }
 604
 605     if (ret != sizeof(buf)) {
 606         LOG("writing to socket failed");
 607         return -EINVAL;
 608     }
 609     return 0;
 610 }
 611
 612 #define MAX_NBD_REQUESTS 16
 613
 614 typedef struct NBDRequest NBDRequest;
 615
 616 struct NBDRequest {
 617     QSIMPLEQ_ENTRY(NBDRequest) entry;
 618     NBDClient *client;
 619     uint8_t *data;
 620 };
 621
 622 struct NBDExport {
 623     BlockDriverState *bs;
 624     off_t dev_offset;
 625     off_t size;
 626     uint32_t nbdflags;
 627     QSIMPLEQ_HEAD(, NBDRequest) requests;
 628 };
 629
 630 struct NBDClient {
 631     int refcount;
 632     void (*close)(NBDClient *client);
 633
 634     NBDExport *exp;
 635     int sock;
 636
 637     Coroutine *recv_coroutine;
 638
 639     CoMutex send_lock;
 640     Coroutine *send_coroutine;
 641
 642     int nb_requests;
 643 };
 644
 645 static void nbd_client_get(NBDClient *client)
 646 {
 647     client->refcount++;
 648 }
 649
 650 static void nbd_client_put(NBDClient *client)
 651 {
 652     if (--client->refcount == 0) {
 653         g_free(client);
 654     }
 655 }
 656
 657 static void nbd_client_close(NBDClient *client)
 658 {
 659     qemu_set_fd_handler2(client->sock, NULL, NULL, NULL, NULL);
 660     close(client->sock);
 661     client->sock = -1;
 662     if (client->close) {
 663         client->close(client);
 664     }
 665     nbd_client_put(client);
 666 }
 667
 668 static NBDRequest *nbd_request_get(NBDClient *client)
 669 {
 670     NBDRequest *req;
 671     NBDExport *exp = client->exp;
 672
 673     assert(client->nb_requests <= MAX_NBD_REQUESTS - 1);
 674     client->nb_requests++;
 675
 676     if (QSIMPLEQ_EMPTY(&exp->requests)) {
 677         req = g_malloc0(sizeof(NBDRequest));
 678         req->data = qemu_blockalign(exp->bs, NBD_BUFFER_SIZE);
 679     } else {
 680         req = QSIMPLEQ_FIRST(&exp->requests);
 681         QSIMPLEQ_REMOVE_HEAD(&exp->requests, entry);
 682     }
 683     nbd_client_get(client);
 684     req->client = client;
 685     return req;
 686 }
 687
 688 static void nbd_request_put(NBDRequest *req)
 689 {
 690     NBDClient *client = req->client;
 691     QSIMPLEQ_INSERT_HEAD(&client->exp->requests, req, entry);
 692     if (client->nb_requests-- == MAX_NBD_REQUESTS) {
 693         qemu_notify_event();
 694     }
 695     nbd_client_put(client);
 696 }
 697
 698 NBDExport *nbd_export_new(BlockDriverState *bs, off_t dev_offset,
 699                           off_t size, uint32_t nbdflags)
 700 {
 701     NBDExport *exp = g_malloc0(sizeof(NBDExport));
 702     QSIMPLEQ_INIT(&exp->requests);
 703     exp->bs = bs;
 704     exp->dev_offset = dev_offset;
 705     exp->nbdflags = nbdflags;
 706     exp->size = size == -1 ? exp->bs->total_sectors * 512 : size;
 707     return exp;
 708 }
 709
 710 void nbd_export_close(NBDExport *exp)
 711 {
 712     while (!QSIMPLEQ_EMPTY(&exp->requests)) {
 713         NBDRequest *first = QSIMPLEQ_FIRST(&exp->requests);
 714         QSIMPLEQ_REMOVE_HEAD(&exp->requests, entry);
 715         qemu_vfree(first->data);
 716         g_free(first);
 717     }
 718
 719     bdrv_close(exp->bs);
 720     g_free(exp);
 721 }
 722
 723 static int nbd_can_read(void *opaque);
 724 static void nbd_read(void *opaque);
 725 static void nbd_restart_write(void *opaque);
 726
 727 static ssize_t nbd_co_send_reply(NBDRequest *req, struct nbd_reply *reply,
 728                                  int len)
 729 {
 730     NBDClient *client = req->client;
 731     int csock = client->sock;
 732     ssize_t rc, ret;
 733
 734     qemu_co_mutex_lock(&client->send_lock);
 735     qemu_set_fd_handler2(csock, nbd_can_read, nbd_read,
 736                          nbd_restart_write, client);
 737     client->send_coroutine = qemu_coroutine_self();
 738
 739     if (!len) {
 740         rc = nbd_send_reply(csock, reply);
 741     } else {
 742         socket_set_cork(csock, 1);
 743         rc = nbd_send_reply(csock, reply);
 744         if (rc >= 0) {
 745             ret = qemu_co_send(csock, req->data, len);
 746             if (ret != len) {
 747                 rc = -EIO;
 748             }
 749         }
 750         socket_set_cork(csock, 0);
 751     }
 752
 753     client->send_coroutine = NULL;
 754     qemu_set_fd_handler2(csock, nbd_can_read, nbd_read, NULL, client);
 755     qemu_co_mutex_unlock(&client->send_lock);
 756     return rc;
 757 }
 758
 759 static ssize_t nbd_co_receive_request(NBDRequest *req, struct nbd_request *request)
 760 {
 761     NBDClient *client = req->client;
 762     int csock = client->sock;
 763     ssize_t rc;
 764
 765     client->recv_coroutine = qemu_coroutine_self();
 766     rc = nbd_receive_request(csock, request);
 767     if (rc < 0) {
 768         if (rc != -EAGAIN) {
 769             rc = -EIO;
 770         }
 771         goto out;
 772     }
 773
 774     if (request->len > NBD_BUFFER_SIZE) {
 775         LOG("len (%u) is larger than max len (%u)",
 776             request->len, NBD_BUFFER_SIZE);
 777         rc = -EINVAL;
 778         goto out;
 779     }
 780
 781     if ((request->from + request->len) < request->from) {
 782         LOG("integer overflow detected! "
 783             "you're probably being attacked");
 784         rc = -EINVAL;
 785         goto out;
 786     }
 787
 788     TRACE("Decoding type");
 789
 790     if ((request->type & NBD_CMD_MASK_COMMAND) == NBD_CMD_WRITE) {
 791         TRACE("Reading %u byte(s)", request->len);
 792
 793         if (qemu_co_recv(csock, req->data, request->len) != request->len) {
 794             LOG("reading from socket failed");
 795             rc = -EIO;
 796             goto out;
 797         }
 798     }
 799     rc = 0;
 800
 801 out:
 802     client->recv_coroutine = NULL;
 803     return rc;
 804 }
 805
 806 static void nbd_trip(void *opaque)
 807 {
 808     NBDClient *client = opaque;
 809     NBDRequest *req = nbd_request_get(client);
 810     NBDExport *exp = client->exp;
 811     struct nbd_request request;
 812     struct nbd_reply reply;
 813     ssize_t ret;
 814
 815     TRACE("Reading request.");
 816
 817     ret = nbd_co_receive_request(req, &request);
 818     if (ret == -EAGAIN) {
 819         goto done;
 820     }
 821     if (ret == -EIO) {
 822         goto out;
 823     }
 824
 825     reply.handle = request.handle;
 826     reply.error = 0;
 827
 828     if (ret < 0) {
 829         reply.error = -ret;
 830         goto error_reply;
 831     }
 832
 833     if ((request.from + request.len) > exp->size) {
 834             LOG("From: %" PRIu64 ", Len: %u, Size: %" PRIu64
 835             ", Offset: %" PRIu64 "\n",
 836                     request.from, request.len,
 837                     (uint64_t)exp->size, (uint64_t)exp->dev_offset);
 838         LOG("requested operation past EOF--bad client?");
 839         goto invalid_request;
 840     }
 841
 842     switch (request.type & NBD_CMD_MASK_COMMAND) {
 843     case NBD_CMD_READ:
 844         TRACE("Request type is READ");
 845
 846         ret = bdrv_read(exp->bs, (request.from + exp->dev_offset) / 512,
 847                         req->data, request.len / 512);
 848         if (ret < 0) {
 849             LOG("reading from file failed");
 850             reply.error = -ret;
 851             goto error_reply;
 852         }
 853
 854         TRACE("Read %u byte(s)", request.len);
 855         if (nbd_co_send_reply(req, &reply, request.len) < 0)
 856             goto out;
 857         break;
 858     case NBD_CMD_WRITE:
 859         TRACE("Request type is WRITE");
 860
 861         if (exp->nbdflags & NBD_FLAG_READ_ONLY) {
 862             TRACE("Server is read-only, return error");
 863             reply.error = EROFS;
 864             goto error_reply;
 865         }
 866
 867         TRACE("Writing to device");
 868
 869         ret = bdrv_write(exp->bs, (request.from + exp->dev_offset) / 512,
 870                          req->data, request.len / 512);
 871         if (ret < 0) {
 872             LOG("writing to file failed");
 873             reply.error = -ret;
 874             goto error_reply;
 875         }
 876
 877         if (request.type & NBD_CMD_FLAG_FUA) {
 878             ret = bdrv_co_flush(exp->bs);
 879             if (ret < 0) {
 880                 LOG("flush failed");
 881                 reply.error = -ret;
 882                 goto error_reply;
 883             }
 884         }
 885
 886         if (nbd_co_send_reply(req, &reply, 0) < 0) {
 887             goto out;
 888         }
 889         break;
 890     case NBD_CMD_DISC:
 891         TRACE("Request type is DISCONNECT");
 892         errno = 0;
 893         goto out;
 894     case NBD_CMD_FLUSH:
 895         TRACE("Request type is FLUSH");
 896
 897         ret = bdrv_co_flush(exp->bs);
 898         if (ret < 0) {
 899             LOG("flush failed");
 900             reply.error = -ret;
 901         }
 902         if (nbd_co_send_reply(req, &reply, 0) < 0) {
 903             goto out;
 904         }
 905         break;
 906     case NBD_CMD_TRIM:
 907         TRACE("Request type is TRIM");
 908         ret = bdrv_co_discard(exp->bs, (request.from + exp->dev_offset) / 512,
 909                               request.len / 512);
 910         if (ret < 0) {
 911             LOG("discard failed");
 912             reply.error = -ret;
 913         }
 914         if (nbd_co_send_reply(req, &reply, 0) < 0) {
 915             goto out;
 916         }
 917         break;
 918     default:
 919         LOG("invalid request type (%u) received", request.type);
 920     invalid_request:
 921         reply.error = -EINVAL;
 922     error_reply:
 923         if (nbd_co_send_reply(req, &reply, 0) < 0) {
 924             goto out;
 925         }
 926         break;
 927     }
 928
 929     TRACE("Request/Reply complete");
 930
 931 done:
 932     nbd_request_put(req);
 933     return;
 934
 935 out:
 936     nbd_request_put(req);
 937     nbd_client_close(client);
 938 }
 939
 940 static int nbd_can_read(void *opaque)
 941 {
 942     NBDClient *client = opaque;
 943
 944     return client->recv_coroutine || client->nb_requests < MAX_NBD_REQUESTS;
 945 }
 946
 947 static void nbd_read(void *opaque)
 948 {
 949     NBDClient *client = opaque;
 950
 951     if (client->recv_coroutine) {
 952         qemu_coroutine_enter(client->recv_coroutine, NULL);
 953     } else {
 954         qemu_coroutine_enter(qemu_coroutine_create(nbd_trip), client);
 955     }
 956 }
 957
 958 static void nbd_restart_write(void *opaque)
 959 {
 960     NBDClient *client = opaque;
 961
 962     qemu_coroutine_enter(client->send_coroutine, NULL);
 963 }
 964
 965 NBDClient *nbd_client_new(NBDExport *exp, int csock,
 966                           void (*close)(NBDClient *))
 967 {
 968     NBDClient *client;
 969     if (nbd_send_negotiate(csock, exp->size, exp->nbdflags) < 0) {
 970         return NULL;
 971     }
 972     client = g_malloc0(sizeof(NBDClient));
 973     client->refcount = 1;
 974     client->exp = exp;
 975     client->sock = csock;
 976     client->close = close;
 977     qemu_co_mutex_init(&client->send_lock);
 978     qemu_set_fd_handler2(csock, nbd_can_read, nbd_read, NULL, client);
 979     return client;
 980 }