dataplane: replace internal thread with IOThread
[qemu.git] / hw / block / 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 "hw/virtio/dataplane/vring.h"
20 #include "ioq.h"
21 #include "block/block.h"
22 #include "hw/virtio/virtio-blk.h"
23 #include "virtio-blk.h"
24 #include "block/aio.h"
25 #include "hw/virtio/virtio-bus.h"
26 #include "monitor/monitor.h" /* for object_add() */
27
28 enum {
29 SEG_MAX = 126, /* maximum number of I/O segments */
30 VRING_MAX = SEG_MAX + 2, /* maximum number of vring descriptors */
31 REQ_MAX = VRING_MAX, /* maximum number of requests in the vring,
32 * is VRING_MAX / 2 with traditional and
33 * VRING_MAX with indirect descriptors */
34 };
35
36 typedef struct {
37 struct iocb iocb; /* Linux AIO control block */
38 QEMUIOVector *inhdr; /* iovecs for virtio_blk_inhdr */
39 VirtQueueElement *elem; /* saved data from the virtqueue */
40 struct iovec *bounce_iov; /* used if guest buffers are unaligned */
41 QEMUIOVector *read_qiov; /* for read completion /w bounce buffer */
42 } VirtIOBlockRequest;
43
44 struct VirtIOBlockDataPlane {
45 bool started;
46 bool starting;
47 bool stopping;
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 IOThread *iothread;
62 bool internal_iothread;
63 AioContext *ctx;
64 EventNotifier io_notifier; /* Linux AIO completion */
65 EventNotifier host_notifier; /* doorbell */
66
67 IOQueue ioqueue; /* Linux AIO queue (should really be per
68 IOThread) */
69 VirtIOBlockRequest requests[REQ_MAX]; /* pool of requests, managed by the
70 queue */
71
72 unsigned int num_reqs;
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->elem->index, 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->elem, len + sizeof(hdr));
123 req->elem = NULL;
124 s->num_reqs--;
125 }
126
127 static void complete_request_early(VirtIOBlockDataPlane *s, VirtQueueElement *elem,
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, elem, 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 VirtQueueElement *elem, QEMUIOVector *inhdr)
146 {
147 char id[VIRTIO_BLK_ID_BYTES];
148
149 /* Serial number not NUL-terminated when longer 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, elem, inhdr, VIRTIO_BLK_S_OK);
153 }
154
155 static int do_rdwr_cmd(VirtIOBlockDataPlane *s, bool read,
156 struct iovec *iov, unsigned iov_cnt,
157 long long offset, VirtQueueElement *elem,
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->elem = elem;
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, VirtQueueElement *elem)
198 {
199 VirtIOBlockDataPlane *s = container_of(ioq, VirtIOBlockDataPlane, ioqueue);
200 struct iovec *iov = elem->out_sg;
201 struct iovec *in_iov = elem->in_sg;
202 unsigned out_num = elem->out_num;
203 unsigned in_num = elem->in_num;
204 struct virtio_blk_outhdr outhdr;
205 QEMUIOVector *inhdr;
206 size_t in_size;
207
208 /* Copy in outhdr */
209 if (unlikely(iov_to_buf(iov, out_num, 0, &outhdr,
210 sizeof(outhdr)) != sizeof(outhdr))) {
211 error_report("virtio-blk request outhdr too short");
212 return -EFAULT;
213 }
214 iov_discard_front(&iov, &out_num, sizeof(outhdr));
215
216 /* Grab inhdr for later */
217 in_size = iov_size(in_iov, in_num);
218 if (in_size < sizeof(struct virtio_blk_inhdr)) {
219 error_report("virtio_blk request inhdr too short");
220 return -EFAULT;
221 }
222 inhdr = g_slice_new(QEMUIOVector);
223 qemu_iovec_init(inhdr, 1);
224 qemu_iovec_concat_iov(inhdr, in_iov, in_num,
225 in_size - sizeof(struct virtio_blk_inhdr),
226 sizeof(struct virtio_blk_inhdr));
227 iov_discard_back(in_iov, &in_num, sizeof(struct virtio_blk_inhdr));
228
229 /* TODO Linux sets the barrier bit even when not advertised! */
230 outhdr.type &= ~VIRTIO_BLK_T_BARRIER;
231
232 switch (outhdr.type) {
233 case VIRTIO_BLK_T_IN:
234 do_rdwr_cmd(s, true, in_iov, in_num, outhdr.sector * 512, elem, inhdr);
235 return 0;
236
237 case VIRTIO_BLK_T_OUT:
238 do_rdwr_cmd(s, false, iov, out_num, outhdr.sector * 512, elem, inhdr);
239 return 0;
240
241 case VIRTIO_BLK_T_SCSI_CMD:
242 /* TODO support SCSI commands */
243 complete_request_early(s, elem, inhdr, VIRTIO_BLK_S_UNSUPP);
244 return 0;
245
246 case VIRTIO_BLK_T_FLUSH:
247 /* TODO fdsync not supported by Linux AIO, do it synchronously here! */
248 if (qemu_fdatasync(s->fd) < 0) {
249 complete_request_early(s, elem, inhdr, VIRTIO_BLK_S_IOERR);
250 } else {
251 complete_request_early(s, elem, inhdr, VIRTIO_BLK_S_OK);
252 }
253 return 0;
254
255 case VIRTIO_BLK_T_GET_ID:
256 do_get_id_cmd(s, in_iov, in_num, elem, inhdr);
257 return 0;
258
259 default:
260 error_report("virtio-blk unsupported request type %#x", outhdr.type);
261 qemu_iovec_destroy(inhdr);
262 g_slice_free(QEMUIOVector, inhdr);
263 return -EFAULT;
264 }
265 }
266
267 static void handle_notify(EventNotifier *e)
268 {
269 VirtIOBlockDataPlane *s = container_of(e, VirtIOBlockDataPlane,
270 host_notifier);
271
272 VirtQueueElement *elem;
273 int ret;
274 unsigned int num_queued;
275
276 event_notifier_test_and_clear(&s->host_notifier);
277 for (;;) {
278 /* Disable guest->host notifies to avoid unnecessary vmexits */
279 vring_disable_notification(s->vdev, &s->vring);
280
281 for (;;) {
282 ret = vring_pop(s->vdev, &s->vring, &elem);
283 if (ret < 0) {
284 assert(elem == NULL);
285 break; /* no more requests */
286 }
287
288 trace_virtio_blk_data_plane_process_request(s, elem->out_num,
289 elem->in_num, elem->index);
290
291 if (process_request(&s->ioqueue, elem) < 0) {
292 vring_set_broken(&s->vring);
293 vring_free_element(elem);
294 ret = -EFAULT;
295 break;
296 }
297 }
298
299 if (likely(ret == -EAGAIN)) { /* vring emptied */
300 /* Re-enable guest->host notifies and stop processing the vring.
301 * But if the guest has snuck in more descriptors, keep processing.
302 */
303 if (vring_enable_notification(s->vdev, &s->vring)) {
304 break;
305 }
306 } else { /* ret == -ENOBUFS or fatal error, iovecs[] is depleted */
307 /* Since there are no iovecs[] left, stop processing for now. Do
308 * not re-enable guest->host notifies since the I/O completion
309 * handler knows to check for more vring descriptors anyway.
310 */
311 break;
312 }
313 }
314
315 num_queued = ioq_num_queued(&s->ioqueue);
316 if (num_queued > 0) {
317 s->num_reqs += num_queued;
318
319 int rc = ioq_submit(&s->ioqueue);
320 if (unlikely(rc < 0)) {
321 fprintf(stderr, "ioq_submit failed %d\n", rc);
322 exit(1);
323 }
324 }
325 }
326
327 static void handle_io(EventNotifier *e)
328 {
329 VirtIOBlockDataPlane *s = container_of(e, VirtIOBlockDataPlane,
330 io_notifier);
331
332 event_notifier_test_and_clear(&s->io_notifier);
333 if (ioq_run_completion(&s->ioqueue, complete_request, s) > 0) {
334 notify_guest(s);
335 }
336
337 /* If there were more requests than iovecs, the vring will not be empty yet
338 * so check again. There should now be enough resources to process more
339 * requests.
340 */
341 if (unlikely(vring_more_avail(&s->vring))) {
342 handle_notify(&s->host_notifier);
343 }
344 }
345
346 /* Context: QEMU global mutex held */
347 void virtio_blk_data_plane_create(VirtIODevice *vdev, VirtIOBlkConf *blk,
348 VirtIOBlockDataPlane **dataplane,
349 Error **errp)
350 {
351 VirtIOBlockDataPlane *s;
352 int fd;
353
354 *dataplane = NULL;
355
356 if (!blk->data_plane) {
357 return;
358 }
359
360 if (blk->scsi) {
361 error_setg(errp,
362 "device is incompatible with x-data-plane, use scsi=off");
363 return;
364 }
365
366 if (blk->config_wce) {
367 error_setg(errp, "device is incompatible with x-data-plane, "
368 "use config-wce=off");
369 return;
370 }
371
372 /* If dataplane is (re-)enabled while the guest is running there could be
373 * block jobs that can conflict.
374 */
375 if (bdrv_in_use(blk->conf.bs)) {
376 error_setg(errp,
377 "cannot start dataplane thread while device is in use");
378 return;
379 }
380
381 fd = raw_get_aio_fd(blk->conf.bs);
382 if (fd < 0) {
383 error_setg(errp, "drive is incompatible with x-data-plane, "
384 "use format=raw,cache=none,aio=native");
385 return;
386 }
387
388 s = g_new0(VirtIOBlockDataPlane, 1);
389 s->vdev = vdev;
390 s->fd = fd;
391 s->blk = blk;
392
393 if (blk->iothread) {
394 s->internal_iothread = false;
395 s->iothread = blk->iothread;
396 object_ref(OBJECT(s->iothread));
397 } else {
398 /* Create per-device IOThread if none specified */
399 Error *local_err = NULL;
400
401 s->internal_iothread = true;
402 object_add(TYPE_IOTHREAD, vdev->name, NULL, NULL, &local_err);
403 if (error_is_set(&local_err)) {
404 error_propagate(errp, local_err);
405 g_free(s);
406 return;
407 }
408 s->iothread = iothread_find(vdev->name);
409 assert(s->iothread);
410 }
411 s->ctx = iothread_get_aio_context(s->iothread);
412
413 /* Prevent block operations that conflict with data plane thread */
414 bdrv_set_in_use(blk->conf.bs, 1);
415
416 *dataplane = s;
417 }
418
419 /* Context: QEMU global mutex held */
420 void virtio_blk_data_plane_destroy(VirtIOBlockDataPlane *s)
421 {
422 if (!s) {
423 return;
424 }
425
426 virtio_blk_data_plane_stop(s);
427 bdrv_set_in_use(s->blk->conf.bs, 0);
428 object_unref(OBJECT(s->iothread));
429 if (s->internal_iothread) {
430 object_unparent(OBJECT(s->iothread));
431 }
432 g_free(s);
433 }
434
435 /* Context: QEMU global mutex held */
436 void virtio_blk_data_plane_start(VirtIOBlockDataPlane *s)
437 {
438 BusState *qbus = BUS(qdev_get_parent_bus(DEVICE(s->vdev)));
439 VirtioBusClass *k = VIRTIO_BUS_GET_CLASS(qbus);
440 VirtQueue *vq;
441 int i;
442
443 if (s->started) {
444 return;
445 }
446
447 if (s->starting) {
448 return;
449 }
450
451 s->starting = true;
452
453 vq = virtio_get_queue(s->vdev, 0);
454 if (!vring_setup(&s->vring, s->vdev, 0)) {
455 s->starting = false;
456 return;
457 }
458
459 /* Set up guest notifier (irq) */
460 if (k->set_guest_notifiers(qbus->parent, 1, true) != 0) {
461 fprintf(stderr, "virtio-blk failed to set guest notifier, "
462 "ensure -enable-kvm is set\n");
463 exit(1);
464 }
465 s->guest_notifier = virtio_queue_get_guest_notifier(vq);
466
467 /* Set up virtqueue notify */
468 if (k->set_host_notifier(qbus->parent, 0, true) != 0) {
469 fprintf(stderr, "virtio-blk failed to set host notifier\n");
470 exit(1);
471 }
472 s->host_notifier = *virtio_queue_get_host_notifier(vq);
473
474 /* Set up ioqueue */
475 ioq_init(&s->ioqueue, s->fd, REQ_MAX);
476 for (i = 0; i < ARRAY_SIZE(s->requests); i++) {
477 ioq_put_iocb(&s->ioqueue, &s->requests[i].iocb);
478 }
479 s->io_notifier = *ioq_get_notifier(&s->ioqueue);
480
481 s->starting = false;
482 s->started = true;
483 trace_virtio_blk_data_plane_start(s);
484
485 /* Kick right away to begin processing requests already in vring */
486 event_notifier_set(virtio_queue_get_host_notifier(vq));
487
488 /* Get this show started by hooking up our callbacks */
489 aio_context_acquire(s->ctx);
490 aio_set_event_notifier(s->ctx, &s->host_notifier, handle_notify);
491 aio_set_event_notifier(s->ctx, &s->io_notifier, handle_io);
492 aio_context_release(s->ctx);
493 }
494
495 /* Context: QEMU global mutex held */
496 void virtio_blk_data_plane_stop(VirtIOBlockDataPlane *s)
497 {
498 BusState *qbus = BUS(qdev_get_parent_bus(DEVICE(s->vdev)));
499 VirtioBusClass *k = VIRTIO_BUS_GET_CLASS(qbus);
500 if (!s->started || s->stopping) {
501 return;
502 }
503 s->stopping = true;
504 trace_virtio_blk_data_plane_stop(s);
505
506 aio_context_acquire(s->ctx);
507
508 /* Stop notifications for new requests from guest */
509 aio_set_event_notifier(s->ctx, &s->host_notifier, NULL);
510
511 /* Complete pending requests */
512 while (s->num_reqs > 0) {
513 aio_poll(s->ctx, true);
514 }
515
516 /* Stop ioq callbacks (there are no pending requests left) */
517 aio_set_event_notifier(s->ctx, &s->io_notifier, NULL);
518
519 aio_context_release(s->ctx);
520
521 /* Sync vring state back to virtqueue so that non-dataplane request
522 * processing can continue when we disable the host notifier below.
523 */
524 vring_teardown(&s->vring, s->vdev, 0);
525
526 ioq_cleanup(&s->ioqueue);
527 k->set_host_notifier(qbus->parent, 0, false);
528
529 /* Clean up guest notifier (irq) */
530 k->set_guest_notifiers(qbus->parent, 1, false);
531
532 s->started = false;
533 s->stopping = false;
534 }