xref: /openbmc/qemu/block/io_uring.c (revision 75e79f5a)
1 /*
2  * Linux io_uring support.
3  *
4  * Copyright (C) 2009 IBM, Corp.
5  * Copyright (C) 2009 Red Hat, Inc.
6  * Copyright (C) 2019 Aarushi Mehta
7  *
8  * This work is licensed under the terms of the GNU GPL, version 2 or later.
9  * See the COPYING file in the top-level directory.
10  */
11 #include "qemu/osdep.h"
12 #include <liburing.h>
13 #include "block/aio.h"
14 #include "qemu/queue.h"
15 #include "block/block.h"
16 #include "block/raw-aio.h"
17 #include "qemu/coroutine.h"
18 #include "qemu/defer-call.h"
19 #include "qapi/error.h"
20 #include "sysemu/block-backend.h"
21 #include "trace.h"
22 
23 /* Only used for assertions.  */
24 #include "qemu/coroutine_int.h"
25 
26 /* io_uring ring size */
27 #define MAX_ENTRIES 128
28 
29 typedef struct LuringAIOCB {
30     Coroutine *co;
31     struct io_uring_sqe sqeq;
32     ssize_t ret;
33     QEMUIOVector *qiov;
34     bool is_read;
35     QSIMPLEQ_ENTRY(LuringAIOCB) next;
36 
37     /*
38      * Buffered reads may require resubmission, see
39      * luring_resubmit_short_read().
40      */
41     int total_read;
42     QEMUIOVector resubmit_qiov;
43 } LuringAIOCB;
44 
45 typedef struct LuringQueue {
46     unsigned int in_queue;
47     unsigned int in_flight;
48     bool blocked;
49     QSIMPLEQ_HEAD(, LuringAIOCB) submit_queue;
50 } LuringQueue;
51 
52 struct LuringState {
53     AioContext *aio_context;
54 
55     struct io_uring ring;
56 
57     /* No locking required, only accessed from AioContext home thread */
58     LuringQueue io_q;
59 
60     QEMUBH *completion_bh;
61 };
62 
63 /**
64  * luring_resubmit:
65  *
66  * Resubmit a request by appending it to submit_queue.  The caller must ensure
67  * that ioq_submit() is called later so that submit_queue requests are started.
68  */
luring_resubmit(LuringState * s,LuringAIOCB * luringcb)69 static void luring_resubmit(LuringState *s, LuringAIOCB *luringcb)
70 {
71     QSIMPLEQ_INSERT_TAIL(&s->io_q.submit_queue, luringcb, next);
72     s->io_q.in_queue++;
73 }
74 
75 /**
76  * luring_resubmit_short_read:
77  *
78  * Short reads are rare but may occur. The remaining read request needs to be
79  * resubmitted.
80  */
luring_resubmit_short_read(LuringState * s,LuringAIOCB * luringcb,int nread)81 static void luring_resubmit_short_read(LuringState *s, LuringAIOCB *luringcb,
82                                        int nread)
83 {
84     QEMUIOVector *resubmit_qiov;
85     size_t remaining;
86 
87     trace_luring_resubmit_short_read(s, luringcb, nread);
88 
89     /* Update read position */
90     luringcb->total_read += nread;
91     remaining = luringcb->qiov->size - luringcb->total_read;
92 
93     /* Shorten qiov */
94     resubmit_qiov = &luringcb->resubmit_qiov;
95     if (resubmit_qiov->iov == NULL) {
96         qemu_iovec_init(resubmit_qiov, luringcb->qiov->niov);
97     } else {
98         qemu_iovec_reset(resubmit_qiov);
99     }
100     qemu_iovec_concat(resubmit_qiov, luringcb->qiov, luringcb->total_read,
101                       remaining);
102 
103     /* Update sqe */
104     luringcb->sqeq.off += nread;
105     luringcb->sqeq.addr = (uintptr_t)luringcb->resubmit_qiov.iov;
106     luringcb->sqeq.len = luringcb->resubmit_qiov.niov;
107 
108     luring_resubmit(s, luringcb);
109 }
110 
111 /**
112  * luring_process_completions:
113  * @s: AIO state
114  *
115  * Fetches completed I/O requests, consumes cqes and invokes their callbacks
116  * The function is somewhat tricky because it supports nested event loops, for
117  * example when a request callback invokes aio_poll().
118  *
119  * Function schedules BH completion so it  can be called again in a nested
120  * event loop.  When there are no events left  to complete the BH is being
121  * canceled.
122  *
123  */
luring_process_completions(LuringState * s)124 static void luring_process_completions(LuringState *s)
125 {
126     struct io_uring_cqe *cqes;
127     int total_bytes;
128 
129     defer_call_begin();
130 
131     /*
132      * Request completion callbacks can run the nested event loop.
133      * Schedule ourselves so the nested event loop will "see" remaining
134      * completed requests and process them.  Without this, completion
135      * callbacks that wait for other requests using a nested event loop
136      * would hang forever.
137      *
138      * This workaround is needed because io_uring uses poll_wait, which
139      * is woken up when new events are added to the uring, thus polling on
140      * the same uring fd will block unless more events are received.
141      *
142      * Other leaf block drivers (drivers that access the data themselves)
143      * are networking based, so they poll sockets for data and run the
144      * correct coroutine.
145      */
146     qemu_bh_schedule(s->completion_bh);
147 
148     while (io_uring_peek_cqe(&s->ring, &cqes) == 0) {
149         LuringAIOCB *luringcb;
150         int ret;
151 
152         if (!cqes) {
153             break;
154         }
155 
156         luringcb = io_uring_cqe_get_data(cqes);
157         ret = cqes->res;
158         io_uring_cqe_seen(&s->ring, cqes);
159         cqes = NULL;
160 
161         /* Change counters one-by-one because we can be nested. */
162         s->io_q.in_flight--;
163         trace_luring_process_completion(s, luringcb, ret);
164 
165         /* total_read is non-zero only for resubmitted read requests */
166         total_bytes = ret + luringcb->total_read;
167 
168         if (ret < 0) {
169             /*
170              * Only writev/readv/fsync requests on regular files or host block
171              * devices are submitted. Therefore -EAGAIN is not expected but it's
172              * known to happen sometimes with Linux SCSI. Submit again and hope
173              * the request completes successfully.
174              *
175              * For more information, see:
176              * https://lore.kernel.org/io-uring/20210727165811.284510-3-axboe@kernel.dk/T/#u
177              *
178              * If the code is changed to submit other types of requests in the
179              * future, then this workaround may need to be extended to deal with
180              * genuine -EAGAIN results that should not be resubmitted
181              * immediately.
182              */
183             if (ret == -EINTR || ret == -EAGAIN) {
184                 luring_resubmit(s, luringcb);
185                 continue;
186             }
187         } else if (!luringcb->qiov) {
188             goto end;
189         } else if (total_bytes == luringcb->qiov->size) {
190             ret = 0;
191         /* Only read/write */
192         } else {
193             /* Short Read/Write */
194             if (luringcb->is_read) {
195                 if (ret > 0) {
196                     luring_resubmit_short_read(s, luringcb, ret);
197                     continue;
198                 } else {
199                     /* Pad with zeroes */
200                     qemu_iovec_memset(luringcb->qiov, total_bytes, 0,
201                                       luringcb->qiov->size - total_bytes);
202                     ret = 0;
203                 }
204             } else {
205                 ret = -ENOSPC;
206             }
207         }
208 end:
209         luringcb->ret = ret;
210         qemu_iovec_destroy(&luringcb->resubmit_qiov);
211 
212         /*
213          * If the coroutine is already entered it must be in ioq_submit()
214          * and will notice luringcb->ret has been filled in when it
215          * eventually runs later. Coroutines cannot be entered recursively
216          * so avoid doing that!
217          */
218         assert(luringcb->co->ctx == s->aio_context);
219         if (!qemu_coroutine_entered(luringcb->co)) {
220             aio_co_wake(luringcb->co);
221         }
222     }
223 
224     qemu_bh_cancel(s->completion_bh);
225 
226     defer_call_end();
227 }
228 
ioq_submit(LuringState * s)229 static int ioq_submit(LuringState *s)
230 {
231     int ret = 0;
232     LuringAIOCB *luringcb, *luringcb_next;
233 
234     while (s->io_q.in_queue > 0) {
235         /*
236          * Try to fetch sqes from the ring for requests waiting in
237          * the overflow queue
238          */
239         QSIMPLEQ_FOREACH_SAFE(luringcb, &s->io_q.submit_queue, next,
240                               luringcb_next) {
241             struct io_uring_sqe *sqes = io_uring_get_sqe(&s->ring);
242             if (!sqes) {
243                 break;
244             }
245             /* Prep sqe for submission */
246             *sqes = luringcb->sqeq;
247             QSIMPLEQ_REMOVE_HEAD(&s->io_q.submit_queue, next);
248         }
249         ret = io_uring_submit(&s->ring);
250         trace_luring_io_uring_submit(s, ret);
251         /* Prevent infinite loop if submission is refused */
252         if (ret <= 0) {
253             if (ret == -EAGAIN || ret == -EINTR) {
254                 continue;
255             }
256             break;
257         }
258         s->io_q.in_flight += ret;
259         s->io_q.in_queue  -= ret;
260     }
261     s->io_q.blocked = (s->io_q.in_queue > 0);
262 
263     if (s->io_q.in_flight) {
264         /*
265          * We can try to complete something just right away if there are
266          * still requests in-flight.
267          */
268         luring_process_completions(s);
269     }
270     return ret;
271 }
272 
luring_process_completions_and_submit(LuringState * s)273 static void luring_process_completions_and_submit(LuringState *s)
274 {
275     luring_process_completions(s);
276 
277     if (s->io_q.in_queue > 0) {
278         ioq_submit(s);
279     }
280 }
281 
qemu_luring_completion_bh(void * opaque)282 static void qemu_luring_completion_bh(void *opaque)
283 {
284     LuringState *s = opaque;
285     luring_process_completions_and_submit(s);
286 }
287 
qemu_luring_completion_cb(void * opaque)288 static void qemu_luring_completion_cb(void *opaque)
289 {
290     LuringState *s = opaque;
291     luring_process_completions_and_submit(s);
292 }
293 
qemu_luring_poll_cb(void * opaque)294 static bool qemu_luring_poll_cb(void *opaque)
295 {
296     LuringState *s = opaque;
297 
298     return io_uring_cq_ready(&s->ring);
299 }
300 
qemu_luring_poll_ready(void * opaque)301 static void qemu_luring_poll_ready(void *opaque)
302 {
303     LuringState *s = opaque;
304 
305     luring_process_completions_and_submit(s);
306 }
307 
ioq_init(LuringQueue * io_q)308 static void ioq_init(LuringQueue *io_q)
309 {
310     QSIMPLEQ_INIT(&io_q->submit_queue);
311     io_q->in_queue = 0;
312     io_q->in_flight = 0;
313     io_q->blocked = false;
314 }
315 
luring_deferred_fn(void * opaque)316 static void luring_deferred_fn(void *opaque)
317 {
318     LuringState *s = opaque;
319     trace_luring_unplug_fn(s, s->io_q.blocked, s->io_q.in_queue,
320                            s->io_q.in_flight);
321     if (!s->io_q.blocked && s->io_q.in_queue > 0) {
322         ioq_submit(s);
323     }
324 }
325 
326 /**
327  * luring_do_submit:
328  * @fd: file descriptor for I/O
329  * @luringcb: AIO control block
330  * @s: AIO state
331  * @offset: offset for request
332  * @type: type of request
333  *
334  * Fetches sqes from ring, adds to pending queue and preps them
335  *
336  */
luring_do_submit(int fd,LuringAIOCB * luringcb,LuringState * s,uint64_t offset,int type)337 static int luring_do_submit(int fd, LuringAIOCB *luringcb, LuringState *s,
338                             uint64_t offset, int type)
339 {
340     int ret;
341     struct io_uring_sqe *sqes = &luringcb->sqeq;
342 
343     switch (type) {
344     case QEMU_AIO_WRITE:
345         io_uring_prep_writev(sqes, fd, luringcb->qiov->iov,
346                              luringcb->qiov->niov, offset);
347         break;
348     case QEMU_AIO_ZONE_APPEND:
349         io_uring_prep_writev(sqes, fd, luringcb->qiov->iov,
350                              luringcb->qiov->niov, offset);
351         break;
352     case QEMU_AIO_READ:
353         io_uring_prep_readv(sqes, fd, luringcb->qiov->iov,
354                             luringcb->qiov->niov, offset);
355         break;
356     case QEMU_AIO_FLUSH:
357         io_uring_prep_fsync(sqes, fd, IORING_FSYNC_DATASYNC);
358         break;
359     default:
360         fprintf(stderr, "%s: invalid AIO request type, aborting 0x%x.\n",
361                         __func__, type);
362         abort();
363     }
364     io_uring_sqe_set_data(sqes, luringcb);
365 
366     QSIMPLEQ_INSERT_TAIL(&s->io_q.submit_queue, luringcb, next);
367     s->io_q.in_queue++;
368     trace_luring_do_submit(s, s->io_q.blocked, s->io_q.in_queue,
369                            s->io_q.in_flight);
370     if (!s->io_q.blocked) {
371         if (s->io_q.in_flight + s->io_q.in_queue >= MAX_ENTRIES) {
372             ret = ioq_submit(s);
373             trace_luring_do_submit_done(s, ret);
374             return ret;
375         }
376 
377         defer_call(luring_deferred_fn, s);
378     }
379     return 0;
380 }
381 
luring_co_submit(BlockDriverState * bs,int fd,uint64_t offset,QEMUIOVector * qiov,int type)382 int coroutine_fn luring_co_submit(BlockDriverState *bs, int fd, uint64_t offset,
383                                   QEMUIOVector *qiov, int type)
384 {
385     int ret;
386     AioContext *ctx = qemu_get_current_aio_context();
387     LuringState *s = aio_get_linux_io_uring(ctx);
388     LuringAIOCB luringcb = {
389         .co         = qemu_coroutine_self(),
390         .ret        = -EINPROGRESS,
391         .qiov       = qiov,
392         .is_read    = (type == QEMU_AIO_READ),
393     };
394     trace_luring_co_submit(bs, s, &luringcb, fd, offset, qiov ? qiov->size : 0,
395                            type);
396     ret = luring_do_submit(fd, &luringcb, s, offset, type);
397 
398     if (ret < 0) {
399         return ret;
400     }
401 
402     if (luringcb.ret == -EINPROGRESS) {
403         qemu_coroutine_yield();
404     }
405     return luringcb.ret;
406 }
407 
luring_detach_aio_context(LuringState * s,AioContext * old_context)408 void luring_detach_aio_context(LuringState *s, AioContext *old_context)
409 {
410     aio_set_fd_handler(old_context, s->ring.ring_fd,
411                        NULL, NULL, NULL, NULL, s);
412     qemu_bh_delete(s->completion_bh);
413     s->aio_context = NULL;
414 }
415 
luring_attach_aio_context(LuringState * s,AioContext * new_context)416 void luring_attach_aio_context(LuringState *s, AioContext *new_context)
417 {
418     s->aio_context = new_context;
419     s->completion_bh = aio_bh_new(new_context, qemu_luring_completion_bh, s);
420     aio_set_fd_handler(s->aio_context, s->ring.ring_fd,
421                        qemu_luring_completion_cb, NULL,
422                        qemu_luring_poll_cb, qemu_luring_poll_ready, s);
423 }
424 
luring_init(Error ** errp)425 LuringState *luring_init(Error **errp)
426 {
427     int rc;
428     LuringState *s = g_new0(LuringState, 1);
429     struct io_uring *ring = &s->ring;
430 
431     trace_luring_init_state(s, sizeof(*s));
432 
433     rc = io_uring_queue_init(MAX_ENTRIES, ring, 0);
434     if (rc < 0) {
435         error_setg_errno(errp, -rc, "failed to init linux io_uring ring");
436         g_free(s);
437         return NULL;
438     }
439 
440     ioq_init(&s->io_q);
441     return s;
442 
443 }
444 
luring_cleanup(LuringState * s)445 void luring_cleanup(LuringState *s)
446 {
447     io_uring_queue_exit(&s->ring);
448     trace_luring_cleanup_state(s);
449     g_free(s);
450 }
451