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