hw/dataplane/virtio-blk.c

   1 /*
   2  * Dedicated thread for virtio-blk I/O processing
   3  *
   4  * Copyright 2012 IBM, Corp.
   5  * Copyright 2012 Red Hat, Inc. and/or its affiliates
   6  *
   7  * Authors:
   8  *   Stefan Hajnoczi <stefanha@redhat.com>
   9  *
  10  * This work is licensed under the terms of the GNU GPL, version 2 or later.
  11  * See the COPYING file in the top-level directory.
  12  *
  13  */
  14
  15 #include "trace.h"
  16 #include "qemu/iov.h"
  17 #include "qemu/thread.h"
  18 #include "qemu/error-report.h"
  19 #include "vring.h"
  20 #include "ioq.h"
  21 #include "migration/migration.h"
  22 #include "block/block.h"
  23 #include "hw/virtio-blk.h"
  24 #include "hw/dataplane/virtio-blk.h"
  25 #include "block/aio.h"
  26
  27 enum {
  28     SEG_MAX = 126,                  /* maximum number of I/O segments */
  29     VRING_MAX = SEG_MAX + 2,        /* maximum number of vring descriptors */
  30     REQ_MAX = VRING_MAX,            /* maximum number of requests in the vring,
  31                                      * is VRING_MAX / 2 with traditional and
  32                                      * VRING_MAX with indirect descriptors */
  33 };
  34
  35 typedef struct {
  36     struct iocb iocb;               /* Linux AIO control block */
  37     QEMUIOVector *inhdr;            /* iovecs for virtio_blk_inhdr */
  38     unsigned int head;              /* vring descriptor index */
  39     struct iovec *bounce_iov;       /* used if guest buffers are unaligned */
  40     QEMUIOVector *read_qiov;        /* for read completion /w bounce buffer */
  41 } VirtIOBlockRequest;
  42
  43 struct VirtIOBlockDataPlane {
  44     bool started;
  45     bool stopping;
  46     QEMUBH *start_bh;
  47     QemuThread thread;
  48
  49     VirtIOBlkConf *blk;
  50     int fd;                         /* image file descriptor */
  51
  52     VirtIODevice *vdev;
  53     Vring vring;                    /* virtqueue vring */
  54     EventNotifier *guest_notifier;  /* irq */
  55
  56     /* Note that these EventNotifiers are assigned by value.  This is
  57      * fine as long as you do not call event_notifier_cleanup on them
  58      * (because you don't own the file descriptor or handle; you just
  59      * use it).
  60      */
  61     AioContext *ctx;
  62     EventNotifier io_notifier;      /* Linux AIO completion */
  63     EventNotifier host_notifier;    /* doorbell */
  64
  65     IOQueue ioqueue;                /* Linux AIO queue (should really be per
  66                                        dataplane thread) */
  67     VirtIOBlockRequest requests[REQ_MAX]; /* pool of requests, managed by the
  68                                              queue */
  69
  70     unsigned int num_reqs;
  71
  72     Error *migration_blocker;
  73 };
  74
  75 /* Raise an interrupt to signal guest, if necessary */
  76 static void notify_guest(VirtIOBlockDataPlane *s)
  77 {
  78     if (!vring_should_notify(s->vdev, &s->vring)) {
  79         return;
  80     }
  81
  82     event_notifier_set(s->guest_notifier);
  83 }
  84
  85 static void complete_request(struct iocb *iocb, ssize_t ret, void *opaque)
  86 {
  87     VirtIOBlockDataPlane *s = opaque;
  88     VirtIOBlockRequest *req = container_of(iocb, VirtIOBlockRequest, iocb);
  89     struct virtio_blk_inhdr hdr;
  90     int len;
  91
  92     if (likely(ret >= 0)) {
  93         hdr.status = VIRTIO_BLK_S_OK;
  94         len = ret;
  95     } else {
  96         hdr.status = VIRTIO_BLK_S_IOERR;
  97         len = 0;
  98     }
  99
 100     trace_virtio_blk_data_plane_complete_request(s, req->head, ret);
 101
 102     if (req->read_qiov) {
 103         assert(req->bounce_iov);
 104         qemu_iovec_from_buf(req->read_qiov, 0, req->bounce_iov->iov_base, len);
 105         qemu_iovec_destroy(req->read_qiov);
 106         g_slice_free(QEMUIOVector, req->read_qiov);
 107     }
 108
 109     if (req->bounce_iov) {
 110         qemu_vfree(req->bounce_iov->iov_base);
 111         g_slice_free(struct iovec, req->bounce_iov);
 112     }
 113
 114     qemu_iovec_from_buf(req->inhdr, 0, &hdr, sizeof(hdr));
 115     qemu_iovec_destroy(req->inhdr);
 116     g_slice_free(QEMUIOVector, req->inhdr);
 117
 118     /* According to the virtio specification len should be the number of bytes
 119      * written to, but for virtio-blk it seems to be the number of bytes
 120      * transferred plus the status bytes.
 121      */
 122     vring_push(&s->vring, req->head, len + sizeof(hdr));
 123
 124     s->num_reqs--;
 125 }
 126
 127 static void complete_request_early(VirtIOBlockDataPlane *s, unsigned int head,
 128                                    QEMUIOVector *inhdr, unsigned char status)
 129 {
 130     struct virtio_blk_inhdr hdr = {
 131         .status = status,
 132     };
 133
 134     qemu_iovec_from_buf(inhdr, 0, &hdr, sizeof(hdr));
 135     qemu_iovec_destroy(inhdr);
 136     g_slice_free(QEMUIOVector, inhdr);
 137
 138     vring_push(&s->vring, head, sizeof(hdr));
 139     notify_guest(s);
 140 }
 141
 142 /* Get disk serial number */
 143 static void do_get_id_cmd(VirtIOBlockDataPlane *s,
 144                           struct iovec *iov, unsigned int iov_cnt,
 145                           unsigned int head, QEMUIOVector *inhdr)
 146 {
 147     char id[VIRTIO_BLK_ID_BYTES];
 148
 149     /* Serial number not NUL-terminated when shorter than buffer */
 150     strncpy(id, s->blk->serial ? s->blk->serial : "", sizeof(id));
 151     iov_from_buf(iov, iov_cnt, 0, id, sizeof(id));
 152     complete_request_early(s, head, inhdr, VIRTIO_BLK_S_OK);
 153 }
 154
 155 static int do_rdwr_cmd(VirtIOBlockDataPlane *s, bool read,
 156                        struct iovec *iov, unsigned int iov_cnt,
 157                        long long offset, unsigned int head,
 158                        QEMUIOVector *inhdr)
 159 {
 160     struct iocb *iocb;
 161     QEMUIOVector qiov;
 162     struct iovec *bounce_iov = NULL;
 163     QEMUIOVector *read_qiov = NULL;
 164
 165     qemu_iovec_init_external(&qiov, iov, iov_cnt);
 166     if (!bdrv_qiov_is_aligned(s->blk->conf.bs, &qiov)) {
 167         void *bounce_buffer = qemu_blockalign(s->blk->conf.bs, qiov.size);
 168
 169         if (read) {
 170             /* Need to copy back from bounce buffer on completion */
 171             read_qiov = g_slice_new(QEMUIOVector);
 172             qemu_iovec_init(read_qiov, iov_cnt);
 173             qemu_iovec_concat_iov(read_qiov, iov, iov_cnt, 0, qiov.size);
 174         } else {
 175             qemu_iovec_to_buf(&qiov, 0, bounce_buffer, qiov.size);
 176         }
 177
 178         /* Redirect I/O to aligned bounce buffer */
 179         bounce_iov = g_slice_new(struct iovec);
 180         bounce_iov->iov_base = bounce_buffer;
 181         bounce_iov->iov_len = qiov.size;
 182         iov = bounce_iov;
 183         iov_cnt = 1;
 184     }
 185
 186     iocb = ioq_rdwr(&s->ioqueue, read, iov, iov_cnt, offset);
 187
 188     /* Fill in virtio block metadata needed for completion */
 189     VirtIOBlockRequest *req = container_of(iocb, VirtIOBlockRequest, iocb);
 190     req->head = head;
 191     req->inhdr = inhdr;
 192     req->bounce_iov = bounce_iov;
 193     req->read_qiov = read_qiov;
 194     return 0;
 195 }
 196
 197 static int process_request(IOQueue *ioq, struct iovec iov[],
 198                            unsigned int out_num, unsigned int in_num,
 199                            unsigned int head)
 200 {
 201     VirtIOBlockDataPlane *s = container_of(ioq, VirtIOBlockDataPlane, ioqueue);
 202     struct iovec *in_iov = &iov[out_num];
 203     struct virtio_blk_outhdr outhdr;
 204     QEMUIOVector *inhdr;
 205     size_t in_size;
 206
 207     /* Copy in outhdr */
 208     if (unlikely(iov_to_buf(iov, out_num, 0, &outhdr,
 209                             sizeof(outhdr)) != sizeof(outhdr))) {
 210         error_report("virtio-blk request outhdr too short");
 211         return -EFAULT;
 212     }
 213     iov_discard_front(&iov, &out_num, sizeof(outhdr));
 214
 215     /* Grab inhdr for later */
 216     in_size = iov_size(in_iov, in_num);
 217     if (in_size < sizeof(struct virtio_blk_inhdr)) {
 218         error_report("virtio_blk request inhdr too short");
 219         return -EFAULT;
 220     }
 221     inhdr = g_slice_new(QEMUIOVector);
 222     qemu_iovec_init(inhdr, 1);
 223     qemu_iovec_concat_iov(inhdr, in_iov, in_num,
 224             in_size - sizeof(struct virtio_blk_inhdr),
 225             sizeof(struct virtio_blk_inhdr));
 226     iov_discard_back(in_iov, &in_num, sizeof(struct virtio_blk_inhdr));
 227
 228     /* TODO Linux sets the barrier bit even when not advertised! */
 229     outhdr.type &= ~VIRTIO_BLK_T_BARRIER;
 230
 231     switch (outhdr.type) {
 232     case VIRTIO_BLK_T_IN:
 233         do_rdwr_cmd(s, true, in_iov, in_num, outhdr.sector * 512, head, inhdr);
 234         return 0;
 235
 236     case VIRTIO_BLK_T_OUT:
 237         do_rdwr_cmd(s, false, iov, out_num, outhdr.sector * 512, head, inhdr);
 238         return 0;
 239
 240     case VIRTIO_BLK_T_SCSI_CMD:
 241         /* TODO support SCSI commands */
 242         complete_request_early(s, head, inhdr, VIRTIO_BLK_S_UNSUPP);
 243         return 0;
 244
 245     case VIRTIO_BLK_T_FLUSH:
 246         /* TODO fdsync not supported by Linux AIO, do it synchronously here! */
 247         if (qemu_fdatasync(s->fd) < 0) {
 248             complete_request_early(s, head, inhdr, VIRTIO_BLK_S_IOERR);
 249         } else {
 250             complete_request_early(s, head, inhdr, VIRTIO_BLK_S_OK);
 251         }
 252         return 0;
 253
 254     case VIRTIO_BLK_T_GET_ID:
 255         do_get_id_cmd(s, in_iov, in_num, head, inhdr);
 256         return 0;
 257
 258     default:
 259         error_report("virtio-blk unsupported request type %#x", outhdr.type);
 260         qemu_iovec_destroy(inhdr);
 261         g_slice_free(QEMUIOVector, inhdr);
 262         return -EFAULT;
 263     }
 264 }
 265
 266 static int flush_true(EventNotifier *e)
 267 {
 268     return true;
 269 }
 270
 271 static void handle_notify(EventNotifier *e)
 272 {
 273     VirtIOBlockDataPlane *s = container_of(e, VirtIOBlockDataPlane,
 274                                            host_notifier);
 275
 276     /* There is one array of iovecs into which all new requests are extracted
 277      * from the vring.  Requests are read from the vring and the translated
 278      * descriptors are written to the iovecs array.  The iovecs do not have to
 279      * persist across handle_notify() calls because the kernel copies the
 280      * iovecs on io_submit().
 281      *
 282      * Handling io_submit() EAGAIN may require storing the requests across
 283      * handle_notify() calls until the kernel has sufficient resources to
 284      * accept more I/O.  This is not implemented yet.
 285      */
 286     struct iovec iovec[VRING_MAX];
 287     struct iovec *end = &iovec[VRING_MAX];
 288     struct iovec *iov = iovec;
 289
 290     /* When a request is read from the vring, the index of the first descriptor
 291      * (aka head) is returned so that the completed request can be pushed onto
 292      * the vring later.
 293      *
 294      * The number of hypervisor read-only iovecs is out_num.  The number of
 295      * hypervisor write-only iovecs is in_num.
 296      */
 297     int head;
 298     unsigned int out_num = 0, in_num = 0;
 299     unsigned int num_queued;
 300
 301     event_notifier_test_and_clear(&s->host_notifier);
 302     for (;;) {
 303         /* Disable guest->host notifies to avoid unnecessary vmexits */
 304         vring_disable_notification(s->vdev, &s->vring);
 305
 306         for (;;) {
 307             head = vring_pop(s->vdev, &s->vring, iov, end, &out_num, &in_num);
 308             if (head < 0) {
 309                 break; /* no more requests */
 310             }
 311
 312             trace_virtio_blk_data_plane_process_request(s, out_num, in_num,
 313                                                         head);
 314
 315             if (process_request(&s->ioqueue, iov, out_num, in_num, head) < 0) {
 316                 vring_set_broken(&s->vring);
 317                 break;
 318             }
 319             iov += out_num + in_num;
 320         }
 321
 322         if (likely(head == -EAGAIN)) { /* vring emptied */
 323             /* Re-enable guest->host notifies and stop processing the vring.
 324              * But if the guest has snuck in more descriptors, keep processing.
 325              */
 326             if (vring_enable_notification(s->vdev, &s->vring)) {
 327                 break;
 328             }
 329         } else { /* head == -ENOBUFS or fatal error, iovecs[] is depleted */
 330             /* Since there are no iovecs[] left, stop processing for now.  Do
 331              * not re-enable guest->host notifies since the I/O completion
 332              * handler knows to check for more vring descriptors anyway.
 333              */
 334             break;
 335         }
 336     }
 337
 338     num_queued = ioq_num_queued(&s->ioqueue);
 339     if (num_queued > 0) {
 340         s->num_reqs += num_queued;
 341
 342         int rc = ioq_submit(&s->ioqueue);
 343         if (unlikely(rc < 0)) {
 344             fprintf(stderr, "ioq_submit failed %d\n", rc);
 345             exit(1);
 346         }
 347     }
 348 }
 349
 350 static int flush_io(EventNotifier *e)
 351 {
 352     VirtIOBlockDataPlane *s = container_of(e, VirtIOBlockDataPlane,
 353                                            io_notifier);
 354
 355     return s->num_reqs > 0;
 356 }
 357
 358 static void handle_io(EventNotifier *e)
 359 {
 360     VirtIOBlockDataPlane *s = container_of(e, VirtIOBlockDataPlane,
 361                                            io_notifier);
 362
 363     event_notifier_test_and_clear(&s->io_notifier);
 364     if (ioq_run_completion(&s->ioqueue, complete_request, s) > 0) {
 365         notify_guest(s);
 366     }
 367
 368     /* If there were more requests than iovecs, the vring will not be empty yet
 369      * so check again.  There should now be enough resources to process more
 370      * requests.
 371      */
 372     if (unlikely(vring_more_avail(&s->vring))) {
 373         handle_notify(&s->host_notifier);
 374     }
 375 }
 376
 377 static void *data_plane_thread(void *opaque)
 378 {
 379     VirtIOBlockDataPlane *s = opaque;
 380
 381     do {
 382         aio_poll(s->ctx, true);
 383     } while (!s->stopping || s->num_reqs > 0);
 384     return NULL;
 385 }
 386
 387 static void start_data_plane_bh(void *opaque)
 388 {
 389     VirtIOBlockDataPlane *s = opaque;
 390
 391     qemu_bh_delete(s->start_bh);
 392     s->start_bh = NULL;
 393     qemu_thread_create(&s->thread, data_plane_thread,
 394                        s, QEMU_THREAD_JOINABLE);
 395 }
 396
 397 bool virtio_blk_data_plane_create(VirtIODevice *vdev, VirtIOBlkConf *blk,
 398                                   VirtIOBlockDataPlane **dataplane)
 399 {
 400     VirtIOBlockDataPlane *s;
 401     int fd;
 402
 403     *dataplane = NULL;
 404
 405     if (!blk->data_plane) {
 406         return true;
 407     }
 408
 409     if (blk->scsi) {
 410         error_report("device is incompatible with x-data-plane, use scsi=off");
 411         return false;
 412     }
 413
 414     if (blk->config_wce) {
 415         error_report("device is incompatible with x-data-plane, "
 416                      "use config-wce=off");
 417         return false;
 418     }
 419
 420     fd = raw_get_aio_fd(blk->conf.bs);
 421     if (fd < 0) {
 422         error_report("drive is incompatible with x-data-plane, "
 423                      "use format=raw,cache=none,aio=native");
 424         return false;
 425     }
 426
 427     s = g_new0(VirtIOBlockDataPlane, 1);
 428     s->vdev = vdev;
 429     s->fd = fd;
 430     s->blk = blk;
 431
 432     /* Prevent block operations that conflict with data plane thread */
 433     bdrv_set_in_use(blk->conf.bs, 1);
 434
 435     error_setg(&s->migration_blocker,
 436             "x-data-plane does not support migration");
 437     migrate_add_blocker(s->migration_blocker);
 438
 439     *dataplane = s;
 440     return true;
 441 }
 442
 443 void virtio_blk_data_plane_destroy(VirtIOBlockDataPlane *s)
 444 {
 445     if (!s) {
 446         return;
 447     }
 448
 449     virtio_blk_data_plane_stop(s);
 450     migrate_del_blocker(s->migration_blocker);
 451     error_free(s->migration_blocker);
 452     bdrv_set_in_use(s->blk->conf.bs, 0);
 453     g_free(s);
 454 }
 455
 456 void virtio_blk_data_plane_start(VirtIOBlockDataPlane *s)
 457 {
 458     VirtQueue *vq;
 459     int i;
 460
 461     if (s->started) {
 462         return;
 463     }
 464
 465     vq = virtio_get_queue(s->vdev, 0);
 466     if (!vring_setup(&s->vring, s->vdev, 0)) {
 467         return;
 468     }
 469
 470     s->ctx = aio_context_new();
 471
 472     /* Set up guest notifier (irq) */
 473     if (s->vdev->binding->set_guest_notifiers(s->vdev->binding_opaque, 1,
 474                                               true) != 0) {
 475         fprintf(stderr, "virtio-blk failed to set guest notifier, "
 476                 "ensure -enable-kvm is set\n");
 477         exit(1);
 478     }
 479     s->guest_notifier = virtio_queue_get_guest_notifier(vq);
 480
 481     /* Set up virtqueue notify */
 482     if (s->vdev->binding->set_host_notifier(s->vdev->binding_opaque,
 483                                             0, true) != 0) {
 484         fprintf(stderr, "virtio-blk failed to set host notifier\n");
 485         exit(1);
 486     }
 487     s->host_notifier = *virtio_queue_get_host_notifier(vq);
 488     aio_set_event_notifier(s->ctx, &s->host_notifier, handle_notify, flush_true);
 489
 490     /* Set up ioqueue */
 491     ioq_init(&s->ioqueue, s->fd, REQ_MAX);
 492     for (i = 0; i < ARRAY_SIZE(s->requests); i++) {
 493         ioq_put_iocb(&s->ioqueue, &s->requests[i].iocb);
 494     }
 495     s->io_notifier = *ioq_get_notifier(&s->ioqueue);
 496     aio_set_event_notifier(s->ctx, &s->io_notifier, handle_io, flush_io);
 497
 498     s->started = true;
 499     trace_virtio_blk_data_plane_start(s);
 500
 501     /* Kick right away to begin processing requests already in vring */
 502     event_notifier_set(virtio_queue_get_host_notifier(vq));
 503
 504     /* Spawn thread in BH so it inherits iothread cpusets */
 505     s->start_bh = qemu_bh_new(start_data_plane_bh, s);
 506     qemu_bh_schedule(s->start_bh);
 507 }
 508
 509 void virtio_blk_data_plane_stop(VirtIOBlockDataPlane *s)
 510 {
 511     if (!s->started || s->stopping) {
 512         return;
 513     }
 514     s->stopping = true;
 515     trace_virtio_blk_data_plane_stop(s);
 516
 517     /* Stop thread or cancel pending thread creation BH */
 518     if (s->start_bh) {
 519         qemu_bh_delete(s->start_bh);
 520         s->start_bh = NULL;
 521     } else {
 522         aio_notify(s->ctx);
 523         qemu_thread_join(&s->thread);
 524     }
 525
 526     aio_set_event_notifier(s->ctx, &s->io_notifier, NULL, NULL);
 527     ioq_cleanup(&s->ioqueue);
 528
 529     aio_set_event_notifier(s->ctx, &s->host_notifier, NULL, NULL);
 530     s->vdev->binding->set_host_notifier(s->vdev->binding_opaque, 0, false);
 531
 532     aio_context_unref(s->ctx);
 533
 534     /* Clean up guest notifier (irq) */
 535     s->vdev->binding->set_guest_notifiers(s->vdev->binding_opaque, 1, false);
 536
 537     vring_teardown(&s->vring);
 538     s->started = false;
 539     s->stopping = false;
 540 }