2 * Dedicated thread for virtio-blk I/O processing
4 * Copyright 2012 IBM, Corp.
5 * Copyright 2012 Red Hat, Inc. and/or its affiliates
8 * Stefan Hajnoczi <stefanha@redhat.com>
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.
17 #include "qemu/thread.h"
18 #include "qemu/error-report.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"
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 */
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 */
43 struct VirtIOBlockDataPlane
{
50 int fd
; /* image file descriptor */
53 Vring vring
; /* virtqueue vring */
54 EventNotifier
*guest_notifier
; /* irq */
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
62 EventNotifier io_notifier
; /* Linux AIO completion */
63 EventNotifier host_notifier
; /* doorbell */
65 IOQueue ioqueue
; /* Linux AIO queue (should really be per
67 VirtIOBlockRequest requests
[REQ_MAX
]; /* pool of requests, managed by the
70 unsigned int num_reqs
;
72 Error
*migration_blocker
;
75 /* Raise an interrupt to signal guest, if necessary */
76 static void notify_guest(VirtIOBlockDataPlane
*s
)
78 if (!vring_should_notify(s
->vdev
, &s
->vring
)) {
82 event_notifier_set(s
->guest_notifier
);
85 static void complete_request(struct iocb
*iocb
, ssize_t ret
, void *opaque
)
87 VirtIOBlockDataPlane
*s
= opaque
;
88 VirtIOBlockRequest
*req
= container_of(iocb
, VirtIOBlockRequest
, iocb
);
89 struct virtio_blk_inhdr hdr
;
92 if (likely(ret
>= 0)) {
93 hdr
.status
= VIRTIO_BLK_S_OK
;
96 hdr
.status
= VIRTIO_BLK_S_IOERR
;
100 trace_virtio_blk_data_plane_complete_request(s
, req
->head
, ret
);
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
);
109 if (req
->bounce_iov
) {
110 qemu_vfree(req
->bounce_iov
->iov_base
);
111 g_slice_free(struct iovec
, req
->bounce_iov
);
114 qemu_iovec_from_buf(req
->inhdr
, 0, &hdr
, sizeof(hdr
));
115 qemu_iovec_destroy(req
->inhdr
);
116 g_slice_free(QEMUIOVector
, req
->inhdr
);
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.
122 vring_push(&s
->vring
, req
->head
, len
+ sizeof(hdr
));
127 static void complete_request_early(VirtIOBlockDataPlane
*s
, unsigned int head
,
128 QEMUIOVector
*inhdr
, unsigned char status
)
130 struct virtio_blk_inhdr hdr
= {
134 qemu_iovec_from_buf(inhdr
, 0, &hdr
, sizeof(hdr
));
135 qemu_iovec_destroy(inhdr
);
136 g_slice_free(QEMUIOVector
, inhdr
);
138 vring_push(&s
->vring
, head
, sizeof(hdr
));
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
)
147 char id
[VIRTIO_BLK_ID_BYTES
];
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
);
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
,
162 struct iovec
*bounce_iov
= NULL
;
163 QEMUIOVector
*read_qiov
= NULL
;
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
);
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
);
175 qemu_iovec_to_buf(&qiov
, 0, bounce_buffer
, qiov
.size
);
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
;
186 iocb
= ioq_rdwr(&s
->ioqueue
, read
, iov
, iov_cnt
, offset
);
188 /* Fill in virtio block metadata needed for completion */
189 VirtIOBlockRequest
*req
= container_of(iocb
, VirtIOBlockRequest
, iocb
);
192 req
->bounce_iov
= bounce_iov
;
193 req
->read_qiov
= read_qiov
;
197 static int process_request(IOQueue
*ioq
, struct iovec iov
[],
198 unsigned int out_num
, unsigned int in_num
,
201 VirtIOBlockDataPlane
*s
= container_of(ioq
, VirtIOBlockDataPlane
, ioqueue
);
202 struct iovec
*in_iov
= &iov
[out_num
];
203 struct virtio_blk_outhdr 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");
213 iov_discard_front(&iov
, &out_num
, sizeof(outhdr
));
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");
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
));
228 /* TODO Linux sets the barrier bit even when not advertised! */
229 outhdr
.type
&= ~VIRTIO_BLK_T_BARRIER
;
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
);
236 case VIRTIO_BLK_T_OUT
:
237 do_rdwr_cmd(s
, false, iov
, out_num
, outhdr
.sector
* 512, head
, inhdr
);
240 case VIRTIO_BLK_T_SCSI_CMD
:
241 /* TODO support SCSI commands */
242 complete_request_early(s
, head
, inhdr
, VIRTIO_BLK_S_UNSUPP
);
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
);
250 complete_request_early(s
, head
, inhdr
, VIRTIO_BLK_S_OK
);
254 case VIRTIO_BLK_T_GET_ID
:
255 do_get_id_cmd(s
, in_iov
, in_num
, head
, inhdr
);
259 error_report("virtio-blk unsupported request type %#x", outhdr
.type
);
260 qemu_iovec_destroy(inhdr
);
261 g_slice_free(QEMUIOVector
, inhdr
);
266 static void handle_notify(EventNotifier
*e
)
268 VirtIOBlockDataPlane
*s
= container_of(e
, VirtIOBlockDataPlane
,
271 /* There is one array of iovecs into which all new requests are extracted
272 * from the vring. Requests are read from the vring and the translated
273 * descriptors are written to the iovecs array. The iovecs do not have to
274 * persist across handle_notify() calls because the kernel copies the
275 * iovecs on io_submit().
277 * Handling io_submit() EAGAIN may require storing the requests across
278 * handle_notify() calls until the kernel has sufficient resources to
279 * accept more I/O. This is not implemented yet.
281 struct iovec iovec
[VRING_MAX
];
282 struct iovec
*end
= &iovec
[VRING_MAX
];
283 struct iovec
*iov
= iovec
;
285 /* When a request is read from the vring, the index of the first descriptor
286 * (aka head) is returned so that the completed request can be pushed onto
289 * The number of hypervisor read-only iovecs is out_num. The number of
290 * hypervisor write-only iovecs is in_num.
293 unsigned int out_num
= 0, in_num
= 0;
294 unsigned int num_queued
;
296 event_notifier_test_and_clear(&s
->host_notifier
);
298 /* Disable guest->host notifies to avoid unnecessary vmexits */
299 vring_disable_notification(s
->vdev
, &s
->vring
);
302 head
= vring_pop(s
->vdev
, &s
->vring
, iov
, end
, &out_num
, &in_num
);
304 break; /* no more requests */
307 trace_virtio_blk_data_plane_process_request(s
, out_num
, in_num
,
310 if (process_request(&s
->ioqueue
, iov
, out_num
, in_num
, head
) < 0) {
311 vring_set_broken(&s
->vring
);
314 iov
+= out_num
+ in_num
;
317 if (likely(head
== -EAGAIN
)) { /* vring emptied */
318 /* Re-enable guest->host notifies and stop processing the vring.
319 * But if the guest has snuck in more descriptors, keep processing.
321 if (vring_enable_notification(s
->vdev
, &s
->vring
)) {
324 } else { /* head == -ENOBUFS or fatal error, iovecs[] is depleted */
325 /* Since there are no iovecs[] left, stop processing for now. Do
326 * not re-enable guest->host notifies since the I/O completion
327 * handler knows to check for more vring descriptors anyway.
333 num_queued
= ioq_num_queued(&s
->ioqueue
);
334 if (num_queued
> 0) {
335 s
->num_reqs
+= num_queued
;
337 int rc
= ioq_submit(&s
->ioqueue
);
338 if (unlikely(rc
< 0)) {
339 fprintf(stderr
, "ioq_submit failed %d\n", rc
);
345 static void handle_io(EventNotifier
*e
)
347 VirtIOBlockDataPlane
*s
= container_of(e
, VirtIOBlockDataPlane
,
350 event_notifier_test_and_clear(&s
->io_notifier
);
351 if (ioq_run_completion(&s
->ioqueue
, complete_request
, s
) > 0) {
355 /* If there were more requests than iovecs, the vring will not be empty yet
356 * so check again. There should now be enough resources to process more
359 if (unlikely(vring_more_avail(&s
->vring
))) {
360 handle_notify(&s
->host_notifier
);
364 static void *data_plane_thread(void *opaque
)
366 VirtIOBlockDataPlane
*s
= opaque
;
369 aio_poll(s
->ctx
, true);
370 } while (!s
->stopping
|| s
->num_reqs
> 0);
374 static void start_data_plane_bh(void *opaque
)
376 VirtIOBlockDataPlane
*s
= opaque
;
378 qemu_bh_delete(s
->start_bh
);
380 qemu_thread_create(&s
->thread
, data_plane_thread
,
381 s
, QEMU_THREAD_JOINABLE
);
384 bool virtio_blk_data_plane_create(VirtIODevice
*vdev
, VirtIOBlkConf
*blk
,
385 VirtIOBlockDataPlane
**dataplane
)
387 VirtIOBlockDataPlane
*s
;
392 if (!blk
->data_plane
) {
397 error_report("device is incompatible with x-data-plane, use scsi=off");
401 if (blk
->config_wce
) {
402 error_report("device is incompatible with x-data-plane, "
403 "use config-wce=off");
407 fd
= raw_get_aio_fd(blk
->conf
.bs
);
409 error_report("drive is incompatible with x-data-plane, "
410 "use format=raw,cache=none,aio=native");
414 s
= g_new0(VirtIOBlockDataPlane
, 1);
419 /* Prevent block operations that conflict with data plane thread */
420 bdrv_set_in_use(blk
->conf
.bs
, 1);
422 error_setg(&s
->migration_blocker
,
423 "x-data-plane does not support migration");
424 migrate_add_blocker(s
->migration_blocker
);
430 void virtio_blk_data_plane_destroy(VirtIOBlockDataPlane
*s
)
436 virtio_blk_data_plane_stop(s
);
437 migrate_del_blocker(s
->migration_blocker
);
438 error_free(s
->migration_blocker
);
439 bdrv_set_in_use(s
->blk
->conf
.bs
, 0);
443 void virtio_blk_data_plane_start(VirtIOBlockDataPlane
*s
)
452 vq
= virtio_get_queue(s
->vdev
, 0);
453 if (!vring_setup(&s
->vring
, s
->vdev
, 0)) {
457 s
->ctx
= aio_context_new();
459 /* Set up guest notifier (irq) */
460 if (s
->vdev
->binding
->set_guest_notifiers(s
->vdev
->binding_opaque
, 1,
462 fprintf(stderr
, "virtio-blk failed to set guest notifier, "
463 "ensure -enable-kvm is set\n");
466 s
->guest_notifier
= virtio_queue_get_guest_notifier(vq
);
468 /* Set up virtqueue notify */
469 if (s
->vdev
->binding
->set_host_notifier(s
->vdev
->binding_opaque
,
471 fprintf(stderr
, "virtio-blk failed to set host notifier\n");
474 s
->host_notifier
= *virtio_queue_get_host_notifier(vq
);
475 aio_set_event_notifier(s
->ctx
, &s
->host_notifier
, handle_notify
, NULL
);
478 ioq_init(&s
->ioqueue
, s
->fd
, REQ_MAX
);
479 for (i
= 0; i
< ARRAY_SIZE(s
->requests
); i
++) {
480 ioq_put_iocb(&s
->ioqueue
, &s
->requests
[i
].iocb
);
482 s
->io_notifier
= *ioq_get_notifier(&s
->ioqueue
);
483 aio_set_event_notifier(s
->ctx
, &s
->io_notifier
, handle_io
, NULL
);
486 trace_virtio_blk_data_plane_start(s
);
488 /* Kick right away to begin processing requests already in vring */
489 event_notifier_set(virtio_queue_get_host_notifier(vq
));
491 /* Spawn thread in BH so it inherits iothread cpusets */
492 s
->start_bh
= qemu_bh_new(start_data_plane_bh
, s
);
493 qemu_bh_schedule(s
->start_bh
);
496 void virtio_blk_data_plane_stop(VirtIOBlockDataPlane
*s
)
498 if (!s
->started
|| s
->stopping
) {
502 trace_virtio_blk_data_plane_stop(s
);
504 /* Stop thread or cancel pending thread creation BH */
506 qemu_bh_delete(s
->start_bh
);
510 qemu_thread_join(&s
->thread
);
513 aio_set_event_notifier(s
->ctx
, &s
->io_notifier
, NULL
, NULL
);
514 ioq_cleanup(&s
->ioqueue
);
516 aio_set_event_notifier(s
->ctx
, &s
->host_notifier
, NULL
, NULL
);
517 s
->vdev
->binding
->set_host_notifier(s
->vdev
->binding_opaque
, 0, false);
519 aio_context_unref(s
->ctx
);
521 /* Clean up guest notifier (irq) */
522 s
->vdev
->binding
->set_guest_notifiers(s
->vdev
->binding_opaque
, 1, false);
524 vring_teardown(&s
->vring
);