diff --git a/hw/dataplane/Makefile.objs b/hw/dataplane/Makefile.objs
index abd408f..682aa9e 100644
--- a/hw/dataplane/Makefile.objs
+++ b/hw/dataplane/Makefile.objs
@@ -1,3 +1,3 @@
 ifeq ($(CONFIG_VIRTIO), y)
-common-obj-$(CONFIG_VIRTIO_BLK_DATA_PLANE) += hostmem.o vring.o event-poll.o ioq.o
+common-obj-$(CONFIG_VIRTIO_BLK_DATA_PLANE) += hostmem.o vring.o event-poll.o ioq.o virtio-blk.o
 endif
diff --git a/hw/dataplane/virtio-blk.c b/hw/dataplane/virtio-blk.c
new file mode 100644
index 0000000..e201245
--- /dev/null
+++ b/hw/dataplane/virtio-blk.c
@@ -0,0 +1,465 @@
+/*
+ * 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 "iov.h"
+#include "event-poll.h"
+#include "qemu-thread.h"
+#include "vring.h"
+#include "ioq.h"
+#include "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 */
+} VirtIOBlockRequest;
+
+struct VirtIOBlockDataPlane {
+    bool started;
+    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);
+
+    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 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;
+    struct iocb *iocb;
+
+    /* 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:
+        iocb = ioq_rdwr(ioq, true, in_iov, in_num, outhdr.sector * 512);
+        break;
+
+    case VIRTIO_BLK_T_OUT:
+        iocb = ioq_rdwr(ioq, false, iov, out_num, outhdr.sector * 512);
+        break;
+
+    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, iov, out_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;
+    }
+
+    /* Fill in virtio block metadata needed for completion */
+    VirtIOBlockRequest *req = container_of(iocb, VirtIOBlockRequest, iocb);
+    req->head = head;
+    req->inhdr = inhdr;
+    return 0;
+}
+
+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->started || 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,
+                                              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) {
+        return;
+    }
+    s->started = false;
+    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, false);
+
+    vring_teardown(&s->vring);
+}
diff --git a/hw/dataplane/virtio-blk.h b/hw/dataplane/virtio-blk.h
new file mode 100644
index 0000000..1e8fdfe
--- /dev/null
+++ b/hw/dataplane/virtio-blk.h
@@ -0,0 +1,29 @@
+/*
+ * 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.
+ *
+ */
+
+#ifndef HW_DATAPLANE_VIRTIO_BLK_H
+#define HW_DATAPLANE_VIRTIO_BLK_H
+
+#include "hw/virtio.h"
+
+typedef struct VirtIOBlockDataPlane VirtIOBlockDataPlane;
+
+bool virtio_blk_data_plane_create(VirtIODevice *vdev, VirtIOBlkConf *blk,
+                                  VirtIOBlockDataPlane **dataplane);
+void virtio_blk_data_plane_destroy(VirtIOBlockDataPlane *s);
+void virtio_blk_data_plane_start(VirtIOBlockDataPlane *s);
+void virtio_blk_data_plane_stop(VirtIOBlockDataPlane *s);
+void virtio_blk_data_plane_drain(VirtIOBlockDataPlane *s);
+
+#endif /* HW_DATAPLANE_VIRTIO_BLK_H */
diff --git a/hw/virtio-blk.h b/hw/virtio-blk.h
index 454f445..43ca492 100644
--- a/hw/virtio-blk.h
+++ b/hw/virtio-blk.h
@@ -105,6 +105,7 @@ struct VirtIOBlkConf
     char *serial;
     uint32_t scsi;
     uint32_t config_wce;
+    uint32_t data_plane;
 };
 
 #define DEFINE_VIRTIO_BLK_FEATURES(_state, _field) \
diff --git a/trace-events b/trace-events
index a9a791b..1edc2ae 100644
--- a/trace-events
+++ b/trace-events
@@ -98,6 +98,12 @@ virtio_blk_rw_complete(void *req, int ret) "req %p ret %d"
 virtio_blk_handle_write(void *req, uint64_t sector, size_t nsectors) "req %p sector %"PRIu64" nsectors %zu"
 virtio_blk_handle_read(void *req, uint64_t sector, size_t nsectors) "req %p sector %"PRIu64" nsectors %zu"
 
+# hw/dataplane/virtio-blk.c
+virtio_blk_data_plane_start(void *s) "dataplane %p"
+virtio_blk_data_plane_stop(void *s) "dataplane %p"
+virtio_blk_data_plane_process_request(void *s, unsigned int out_num, unsigned int in_num, unsigned int head) "dataplane %p out_num %u in_num %u head %u"
+virtio_blk_data_plane_complete_request(void *s, unsigned int head, int ret) "dataplane %p head %u ret %d"
+
 # hw/dataplane/vring.c
 vring_setup(uint64_t physical, void *desc, void *avail, void *used) "vring physical %#"PRIx64" desc %p avail %p used %p"
 
