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