qemu/hw/dataplane/virtio-blk.c
Stefan Hajnoczi cd7fdfe59f dataplane: avoid reentrancy during virtio_blk_data_plane_stop()
When dataplane is stopping, the s->vdev->binding->set_host_notifier(...,
false) call can invoke the virtqueue handler if an ioeventfd
notification is pending.  This causes hw/virtio-blk.c to invoke
virtio_blk_data_plane_start() before virtio_blk_data_plane_stop()
returns!

The result is that we try to restart dataplane while trying to stop it
and the following assertion is raised:

  msix_set_mask_notifier: Assertion `!dev->msix_mask_notifier' failed.

Although the code was intended to prevent this scenario, the s->started
boolean isn't enough.  Add s->stopping so that we can postpone clearing
s->started until we've completely stopped dataplane.

This way, virtqueue handler calls during virtio_blk_data_plane_stop()
are ignored.  When dataplane is legitimately started again later we
already self-kick ourselves to resume processing.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
2013-01-18 16:57:59 +01:00

518 lines
16 KiB
C

/*
* Dedicated thread for virtio-blk I/O processing
*
* Copyright 2012 IBM, Corp.
* Copyright 2012 Red Hat, Inc. and/or its affiliates
*
* Authors:
* Stefan Hajnoczi <stefanha@redhat.com>
*
* This work is licensed under the terms of the GNU GPL, version 2 or later.
* See the COPYING file in the top-level directory.
*
*/
#include "trace.h"
#include "qemu/iov.h"
#include "event-poll.h"
#include "qemu/thread.h"
#include "vring.h"
#include "ioq.h"
#include "migration/migration.h"
#include "hw/virtio-blk.h"
#include "hw/dataplane/virtio-blk.h"
enum {
SEG_MAX = 126, /* maximum number of I/O segments */
VRING_MAX = SEG_MAX + 2, /* maximum number of vring descriptors */
REQ_MAX = VRING_MAX, /* maximum number of requests in the vring,
* is VRING_MAX / 2 with traditional and
* VRING_MAX with indirect descriptors */
};
typedef struct {
struct iocb iocb; /* Linux AIO control block */
QEMUIOVector *inhdr; /* iovecs for virtio_blk_inhdr */
unsigned int head; /* vring descriptor index */
struct iovec *bounce_iov; /* used if guest buffers are unaligned */
QEMUIOVector *read_qiov; /* for read completion /w bounce buffer */
} VirtIOBlockRequest;
struct VirtIOBlockDataPlane {
bool started;
bool stopping;
QEMUBH *start_bh;
QemuThread thread;
VirtIOBlkConf *blk;
int fd; /* image file descriptor */
VirtIODevice *vdev;
Vring vring; /* virtqueue vring */
EventNotifier *guest_notifier; /* irq */
EventPoll event_poll; /* event poller */
EventHandler io_handler; /* Linux AIO completion handler */
EventHandler notify_handler; /* virtqueue notify handler */
IOQueue ioqueue; /* Linux AIO queue (should really be per
dataplane thread) */
VirtIOBlockRequest requests[REQ_MAX]; /* pool of requests, managed by the
queue */
unsigned int num_reqs;
Error *migration_blocker;
};
/* Raise an interrupt to signal guest, if necessary */
static void notify_guest(VirtIOBlockDataPlane *s)
{
if (!vring_should_notify(s->vdev, &s->vring)) {
return;
}
event_notifier_set(s->guest_notifier);
}
static void complete_request(struct iocb *iocb, ssize_t ret, void *opaque)
{
VirtIOBlockDataPlane *s = opaque;
VirtIOBlockRequest *req = container_of(iocb, VirtIOBlockRequest, iocb);
struct virtio_blk_inhdr hdr;
int len;
if (likely(ret >= 0)) {
hdr.status = VIRTIO_BLK_S_OK;
len = ret;
} else {
hdr.status = VIRTIO_BLK_S_IOERR;
len = 0;
}
trace_virtio_blk_data_plane_complete_request(s, req->head, ret);
if (req->read_qiov) {
assert(req->bounce_iov);
qemu_iovec_from_buf(req->read_qiov, 0, req->bounce_iov->iov_base, len);
qemu_iovec_destroy(req->read_qiov);
g_slice_free(QEMUIOVector, req->read_qiov);
}
if (req->bounce_iov) {
qemu_vfree(req->bounce_iov->iov_base);
g_slice_free(struct iovec, req->bounce_iov);
}
qemu_iovec_from_buf(req->inhdr, 0, &hdr, sizeof(hdr));
qemu_iovec_destroy(req->inhdr);
g_slice_free(QEMUIOVector, req->inhdr);
/* According to the virtio specification len should be the number of bytes
* written to, but for virtio-blk it seems to be the number of bytes
* transferred plus the status bytes.
*/
vring_push(&s->vring, req->head, len + sizeof(hdr));
s->num_reqs--;
}
static void complete_request_early(VirtIOBlockDataPlane *s, unsigned int head,
QEMUIOVector *inhdr, unsigned char status)
{
struct virtio_blk_inhdr hdr = {
.status = status,
};
qemu_iovec_from_buf(inhdr, 0, &hdr, sizeof(hdr));
qemu_iovec_destroy(inhdr);
g_slice_free(QEMUIOVector, inhdr);
vring_push(&s->vring, head, sizeof(hdr));
notify_guest(s);
}
/* Get disk serial number */
static void do_get_id_cmd(VirtIOBlockDataPlane *s,
struct iovec *iov, unsigned int iov_cnt,
unsigned int head, QEMUIOVector *inhdr)
{
char id[VIRTIO_BLK_ID_BYTES];
/* Serial number not NUL-terminated when shorter than buffer */
strncpy(id, s->blk->serial ? s->blk->serial : "", sizeof(id));
iov_from_buf(iov, iov_cnt, 0, id, sizeof(id));
complete_request_early(s, head, inhdr, VIRTIO_BLK_S_OK);
}
static int do_rdwr_cmd(VirtIOBlockDataPlane *s, bool read,
struct iovec *iov, unsigned int iov_cnt,
long long offset, unsigned int head,
QEMUIOVector *inhdr)
{
struct iocb *iocb;
QEMUIOVector qiov;
struct iovec *bounce_iov = NULL;
QEMUIOVector *read_qiov = NULL;
qemu_iovec_init_external(&qiov, iov, iov_cnt);
if (!bdrv_qiov_is_aligned(s->blk->conf.bs, &qiov)) {
void *bounce_buffer = qemu_blockalign(s->blk->conf.bs, qiov.size);
if (read) {
/* Need to copy back from bounce buffer on completion */
read_qiov = g_slice_new(QEMUIOVector);
qemu_iovec_init(read_qiov, iov_cnt);
qemu_iovec_concat_iov(read_qiov, iov, iov_cnt, 0, qiov.size);
} else {
qemu_iovec_to_buf(&qiov, 0, bounce_buffer, qiov.size);
}
/* Redirect I/O to aligned bounce buffer */
bounce_iov = g_slice_new(struct iovec);
bounce_iov->iov_base = bounce_buffer;
bounce_iov->iov_len = qiov.size;
iov = bounce_iov;
iov_cnt = 1;
}
iocb = ioq_rdwr(&s->ioqueue, read, iov, iov_cnt, offset);
/* Fill in virtio block metadata needed for completion */
VirtIOBlockRequest *req = container_of(iocb, VirtIOBlockRequest, iocb);
req->head = head;
req->inhdr = inhdr;
req->bounce_iov = bounce_iov;
req->read_qiov = read_qiov;
return 0;
}
static int process_request(IOQueue *ioq, struct iovec iov[],
unsigned int out_num, unsigned int in_num,
unsigned int head)
{
VirtIOBlockDataPlane *s = container_of(ioq, VirtIOBlockDataPlane, ioqueue);
struct iovec *in_iov = &iov[out_num];
struct virtio_blk_outhdr outhdr;
QEMUIOVector *inhdr;
size_t in_size;
/* Copy in outhdr */
if (unlikely(iov_to_buf(iov, out_num, 0, &outhdr,
sizeof(outhdr)) != sizeof(outhdr))) {
error_report("virtio-blk request outhdr too short");
return -EFAULT;
}
iov_discard_front(&iov, &out_num, sizeof(outhdr));
/* Grab inhdr for later */
in_size = iov_size(in_iov, in_num);
if (in_size < sizeof(struct virtio_blk_inhdr)) {
error_report("virtio_blk request inhdr too short");
return -EFAULT;
}
inhdr = g_slice_new(QEMUIOVector);
qemu_iovec_init(inhdr, 1);
qemu_iovec_concat_iov(inhdr, in_iov, in_num,
in_size - sizeof(struct virtio_blk_inhdr),
sizeof(struct virtio_blk_inhdr));
iov_discard_back(in_iov, &in_num, sizeof(struct virtio_blk_inhdr));
/* TODO Linux sets the barrier bit even when not advertised! */
outhdr.type &= ~VIRTIO_BLK_T_BARRIER;
switch (outhdr.type) {
case VIRTIO_BLK_T_IN:
do_rdwr_cmd(s, true, in_iov, in_num, outhdr.sector * 512, head, inhdr);
return 0;
case VIRTIO_BLK_T_OUT:
do_rdwr_cmd(s, false, iov, out_num, outhdr.sector * 512, head, inhdr);
return 0;
case VIRTIO_BLK_T_SCSI_CMD:
/* TODO support SCSI commands */
complete_request_early(s, head, inhdr, VIRTIO_BLK_S_UNSUPP);
return 0;
case VIRTIO_BLK_T_FLUSH:
/* TODO fdsync not supported by Linux AIO, do it synchronously here! */
if (qemu_fdatasync(s->fd) < 0) {
complete_request_early(s, head, inhdr, VIRTIO_BLK_S_IOERR);
} else {
complete_request_early(s, head, inhdr, VIRTIO_BLK_S_OK);
}
return 0;
case VIRTIO_BLK_T_GET_ID:
do_get_id_cmd(s, in_iov, in_num, head, inhdr);
return 0;
default:
error_report("virtio-blk unsupported request type %#x", outhdr.type);
qemu_iovec_destroy(inhdr);
g_slice_free(QEMUIOVector, inhdr);
return -EFAULT;
}
}
static void handle_notify(EventHandler *handler)
{
VirtIOBlockDataPlane *s = container_of(handler, VirtIOBlockDataPlane,
notify_handler);
/* There is one array of iovecs into which all new requests are extracted
* from the vring. Requests are read from the vring and the translated
* descriptors are written to the iovecs array. The iovecs do not have to
* persist across handle_notify() calls because the kernel copies the
* iovecs on io_submit().
*
* Handling io_submit() EAGAIN may require storing the requests across
* handle_notify() calls until the kernel has sufficient resources to
* accept more I/O. This is not implemented yet.
*/
struct iovec iovec[VRING_MAX];
struct iovec *end = &iovec[VRING_MAX];
struct iovec *iov = iovec;
/* When a request is read from the vring, the index of the first descriptor
* (aka head) is returned so that the completed request can be pushed onto
* the vring later.
*
* The number of hypervisor read-only iovecs is out_num. The number of
* hypervisor write-only iovecs is in_num.
*/
int head;
unsigned int out_num = 0, in_num = 0;
unsigned int num_queued;
for (;;) {
/* Disable guest->host notifies to avoid unnecessary vmexits */
vring_disable_notification(s->vdev, &s->vring);
for (;;) {
head = vring_pop(s->vdev, &s->vring, iov, end, &out_num, &in_num);
if (head < 0) {
break; /* no more requests */
}
trace_virtio_blk_data_plane_process_request(s, out_num, in_num,
head);
if (process_request(&s->ioqueue, iov, out_num, in_num, head) < 0) {
vring_set_broken(&s->vring);
break;
}
iov += out_num + in_num;
}
if (likely(head == -EAGAIN)) { /* vring emptied */
/* Re-enable guest->host notifies and stop processing the vring.
* But if the guest has snuck in more descriptors, keep processing.
*/
if (vring_enable_notification(s->vdev, &s->vring)) {
break;
}
} else { /* head == -ENOBUFS or fatal error, iovecs[] is depleted */
/* Since there are no iovecs[] left, stop processing for now. Do
* not re-enable guest->host notifies since the I/O completion
* handler knows to check for more vring descriptors anyway.
*/
break;
}
}
num_queued = ioq_num_queued(&s->ioqueue);
if (num_queued > 0) {
s->num_reqs += num_queued;
int rc = ioq_submit(&s->ioqueue);
if (unlikely(rc < 0)) {
fprintf(stderr, "ioq_submit failed %d\n", rc);
exit(1);
}
}
}
static void handle_io(EventHandler *handler)
{
VirtIOBlockDataPlane *s = container_of(handler, VirtIOBlockDataPlane,
io_handler);
if (ioq_run_completion(&s->ioqueue, complete_request, s) > 0) {
notify_guest(s);
}
/* If there were more requests than iovecs, the vring will not be empty yet
* so check again. There should now be enough resources to process more
* requests.
*/
if (unlikely(vring_more_avail(&s->vring))) {
handle_notify(&s->notify_handler);
}
}
static void *data_plane_thread(void *opaque)
{
VirtIOBlockDataPlane *s = opaque;
do {
event_poll(&s->event_poll);
} while (!s->stopping || s->num_reqs > 0);
return NULL;
}
static void start_data_plane_bh(void *opaque)
{
VirtIOBlockDataPlane *s = opaque;
qemu_bh_delete(s->start_bh);
s->start_bh = NULL;
qemu_thread_create(&s->thread, data_plane_thread,
s, QEMU_THREAD_JOINABLE);
}
bool virtio_blk_data_plane_create(VirtIODevice *vdev, VirtIOBlkConf *blk,
VirtIOBlockDataPlane **dataplane)
{
VirtIOBlockDataPlane *s;
int fd;
*dataplane = NULL;
if (!blk->data_plane) {
return true;
}
if (blk->scsi) {
error_report("device is incompatible with x-data-plane, use scsi=off");
return false;
}
if (blk->config_wce) {
error_report("device is incompatible with x-data-plane, "
"use config-wce=off");
return false;
}
fd = raw_get_aio_fd(blk->conf.bs);
if (fd < 0) {
error_report("drive is incompatible with x-data-plane, "
"use format=raw,cache=none,aio=native");
return false;
}
s = g_new0(VirtIOBlockDataPlane, 1);
s->vdev = vdev;
s->fd = fd;
s->blk = blk;
/* Prevent block operations that conflict with data plane thread */
bdrv_set_in_use(blk->conf.bs, 1);
error_setg(&s->migration_blocker,
"x-data-plane does not support migration");
migrate_add_blocker(s->migration_blocker);
*dataplane = s;
return true;
}
void virtio_blk_data_plane_destroy(VirtIOBlockDataPlane *s)
{
if (!s) {
return;
}
virtio_blk_data_plane_stop(s);
migrate_del_blocker(s->migration_blocker);
error_free(s->migration_blocker);
bdrv_set_in_use(s->blk->conf.bs, 0);
g_free(s);
}
void virtio_blk_data_plane_start(VirtIOBlockDataPlane *s)
{
VirtQueue *vq;
int i;
if (s->started) {
return;
}
vq = virtio_get_queue(s->vdev, 0);
if (!vring_setup(&s->vring, s->vdev, 0)) {
return;
}
event_poll_init(&s->event_poll);
/* Set up guest notifier (irq) */
if (s->vdev->binding->set_guest_notifiers(s->vdev->binding_opaque, 1,
true) != 0) {
fprintf(stderr, "virtio-blk failed to set guest notifier, "
"ensure -enable-kvm is set\n");
exit(1);
}
s->guest_notifier = virtio_queue_get_guest_notifier(vq);
/* Set up virtqueue notify */
if (s->vdev->binding->set_host_notifier(s->vdev->binding_opaque,
0, true) != 0) {
fprintf(stderr, "virtio-blk failed to set host notifier\n");
exit(1);
}
event_poll_add(&s->event_poll, &s->notify_handler,
virtio_queue_get_host_notifier(vq),
handle_notify);
/* Set up ioqueue */
ioq_init(&s->ioqueue, s->fd, REQ_MAX);
for (i = 0; i < ARRAY_SIZE(s->requests); i++) {
ioq_put_iocb(&s->ioqueue, &s->requests[i].iocb);
}
event_poll_add(&s->event_poll, &s->io_handler,
ioq_get_notifier(&s->ioqueue), handle_io);
s->started = true;
trace_virtio_blk_data_plane_start(s);
/* Kick right away to begin processing requests already in vring */
event_notifier_set(virtio_queue_get_host_notifier(vq));
/* Spawn thread in BH so it inherits iothread cpusets */
s->start_bh = qemu_bh_new(start_data_plane_bh, s);
qemu_bh_schedule(s->start_bh);
}
void virtio_blk_data_plane_stop(VirtIOBlockDataPlane *s)
{
if (!s->started || s->stopping) {
return;
}
s->stopping = true;
trace_virtio_blk_data_plane_stop(s);
/* Stop thread or cancel pending thread creation BH */
if (s->start_bh) {
qemu_bh_delete(s->start_bh);
s->start_bh = NULL;
} else {
event_poll_notify(&s->event_poll);
qemu_thread_join(&s->thread);
}
ioq_cleanup(&s->ioqueue);
s->vdev->binding->set_host_notifier(s->vdev->binding_opaque, 0, false);
event_poll_cleanup(&s->event_poll);
/* Clean up guest notifier (irq) */
s->vdev->binding->set_guest_notifiers(s->vdev->binding_opaque, 1, false);
vring_teardown(&s->vring);
s->started = false;
s->stopping = false;
}