2 * Linux native AIO support.
4 * Copyright (C) 2009 IBM, Corp.
5 * Copyright (C) 2009 Red Hat, Inc.
7 * This work is licensed under the terms of the GNU GPL, version 2 or later.
8 * See the COPYING file in the top-level directory.
10 #include "qemu/osdep.h"
11 #include "qemu-common.h"
12 #include "block/aio.h"
13 #include "qemu/queue.h"
14 #include "block/block.h"
15 #include "block/raw-aio.h"
16 #include "qemu/event_notifier.h"
17 #include "qemu/coroutine.h"
22 * Queue size (per-device).
24 * XXX: eventually we need to communicate this to the guest and/or make it
25 * tunable by the guest. If we get more outstanding requests at a time
26 * than this we will get EAGAIN from io_submit which is communicated to
27 * the guest as an I/O error.
29 #define MAX_EVENTS 128
40 QSIMPLEQ_ENTRY(qemu_laiocb
) next
;
45 unsigned int in_queue
;
46 unsigned int in_flight
;
48 QSIMPLEQ_HEAD(, qemu_laiocb
) pending
;
51 struct LinuxAioState
{
52 AioContext
*aio_context
;
57 /* io queue for submit at batch. Protected by AioContext lock. */
60 /* I/O completion processing. Only runs in I/O thread. */
61 QEMUBH
*completion_bh
;
66 static void ioq_submit(LinuxAioState
*s
);
68 static inline ssize_t
io_event_ret(struct io_event
*ev
)
70 return (ssize_t
)(((uint64_t)ev
->res2
<< 32) | ev
->res
);
74 * Completes an AIO request (calls the callback and frees the ACB).
76 static void qemu_laio_process_completion(struct qemu_laiocb
*laiocb
)
78 LinuxAioState
*s
= laiocb
->ctx
;
82 if (ret
!= -ECANCELED
) {
83 if (ret
== laiocb
->nbytes
) {
85 } else if (ret
>= 0) {
86 /* Short reads mean EOF, pad with zeros. */
87 if (laiocb
->is_read
) {
88 qemu_iovec_memset(laiocb
->qiov
, ret
, 0,
89 laiocb
->qiov
->size
- ret
);
97 aio_context_acquire(s
->aio_context
);
99 /* If the coroutine is already entered it must be in ioq_submit() and
100 * will notice laio->ret has been filled in when it eventually runs
101 * later. Coroutines cannot be entered recursively so avoid doing
104 if (!qemu_coroutine_entered(laiocb
->co
)) {
105 qemu_coroutine_enter(laiocb
->co
);
108 laiocb
->common
.cb(laiocb
->common
.opaque
, ret
);
109 qemu_aio_unref(laiocb
);
111 aio_context_release(s
->aio_context
);
115 * aio_ring buffer which is shared between userspace and kernel.
117 * This copied from linux/fs/aio.c, common header does not exist
118 * but AIO exists for ages so we assume ABI is stable.
121 unsigned id
; /* kernel internal index number */
122 unsigned nr
; /* number of io_events */
123 unsigned head
; /* Written to by userland or by kernel. */
127 unsigned compat_features
;
128 unsigned incompat_features
;
129 unsigned header_length
; /* size of aio_ring */
131 struct io_event io_events
[0];
137 * @events: pointer on events array, output value
139 * Returns the number of completed events and sets a pointer
140 * on events array. This function does not update the internal
141 * ring buffer, only reads head and tail. When @events has been
142 * processed io_getevents_commit() must be called.
144 static inline unsigned int io_getevents_peek(io_context_t ctx
,
145 struct io_event
**events
)
147 struct aio_ring
*ring
= (struct aio_ring
*)ctx
;
148 unsigned int head
= ring
->head
, tail
= ring
->tail
;
151 nr
= tail
>= head
? tail
- head
: ring
->nr
- head
;
152 *events
= ring
->io_events
+ head
;
153 /* To avoid speculative loads of s->events[i] before observing tail.
154 Paired with smp_wmb() inside linux/fs/aio.c: aio_complete(). */
161 * io_getevents_commit:
163 * @nr: the number of events on which head should be advanced
165 * Advances head of a ring buffer.
167 static inline void io_getevents_commit(io_context_t ctx
, unsigned int nr
)
169 struct aio_ring
*ring
= (struct aio_ring
*)ctx
;
172 ring
->head
= (ring
->head
+ nr
) % ring
->nr
;
177 * io_getevents_advance_and_peek:
179 * @events: pointer on events array, output value
180 * @nr: the number of events on which head should be advanced
182 * Advances head of a ring buffer and returns number of elements left.
184 static inline unsigned int
185 io_getevents_advance_and_peek(io_context_t ctx
,
186 struct io_event
**events
,
189 io_getevents_commit(ctx
, nr
);
190 return io_getevents_peek(ctx
, events
);
194 * qemu_laio_process_completions:
197 * Fetches completed I/O requests and invokes their callbacks.
199 * The function is somewhat tricky because it supports nested event loops, for
200 * example when a request callback invokes aio_poll(). In order to do this,
201 * indices are kept in LinuxAioState. Function schedules BH completion so it
202 * can be called again in a nested event loop. When there are no events left
203 * to complete the BH is being canceled.
205 static void qemu_laio_process_completions(LinuxAioState
*s
)
207 struct io_event
*events
;
209 /* Reschedule so nested event loops see currently pending completions */
210 qemu_bh_schedule(s
->completion_bh
);
212 while ((s
->event_max
= io_getevents_advance_and_peek(s
->ctx
, &events
,
214 for (s
->event_idx
= 0; s
->event_idx
< s
->event_max
; ) {
215 struct iocb
*iocb
= events
[s
->event_idx
].obj
;
216 struct qemu_laiocb
*laiocb
=
217 container_of(iocb
, struct qemu_laiocb
, iocb
);
219 laiocb
->ret
= io_event_ret(&events
[s
->event_idx
]);
221 /* Change counters one-by-one because we can be nested. */
224 qemu_laio_process_completion(laiocb
);
228 qemu_bh_cancel(s
->completion_bh
);
230 /* If we are nested we have to notify the level above that we are done
231 * by setting event_max to zero, upper level will then jump out of it's
232 * own `for` loop. If we are the last all counters droped to zero. */
237 static void qemu_laio_process_completions_and_submit(LinuxAioState
*s
)
239 qemu_laio_process_completions(s
);
241 aio_context_acquire(s
->aio_context
);
242 if (!s
->io_q
.plugged
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
)) {
245 aio_context_release(s
->aio_context
);
248 static void qemu_laio_completion_bh(void *opaque
)
250 LinuxAioState
*s
= opaque
;
252 qemu_laio_process_completions_and_submit(s
);
255 static void qemu_laio_completion_cb(EventNotifier
*e
)
257 LinuxAioState
*s
= container_of(e
, LinuxAioState
, e
);
259 if (event_notifier_test_and_clear(&s
->e
)) {
260 qemu_laio_process_completions_and_submit(s
);
264 static bool qemu_laio_poll_cb(void *opaque
)
266 EventNotifier
*e
= opaque
;
267 LinuxAioState
*s
= container_of(e
, LinuxAioState
, e
);
268 struct io_event
*events
;
270 if (!io_getevents_peek(s
->ctx
, &events
)) {
274 qemu_laio_process_completions_and_submit(s
);
278 static void laio_cancel(BlockAIOCB
*blockacb
)
280 struct qemu_laiocb
*laiocb
= (struct qemu_laiocb
*)blockacb
;
281 struct io_event event
;
284 if (laiocb
->ret
!= -EINPROGRESS
) {
287 ret
= io_cancel(laiocb
->ctx
->ctx
, &laiocb
->iocb
, &event
);
288 laiocb
->ret
= -ECANCELED
;
290 /* iocb is not cancelled, cb will be called by the event loop later */
294 laiocb
->common
.cb(laiocb
->common
.opaque
, laiocb
->ret
);
297 static const AIOCBInfo laio_aiocb_info
= {
298 .aiocb_size
= sizeof(struct qemu_laiocb
),
299 .cancel_async
= laio_cancel
,
302 static void ioq_init(LaioQueue
*io_q
)
304 QSIMPLEQ_INIT(&io_q
->pending
);
308 io_q
->blocked
= false;
311 static void ioq_submit(LinuxAioState
*s
)
314 struct qemu_laiocb
*aiocb
;
315 struct iocb
*iocbs
[MAX_EVENTS
];
316 QSIMPLEQ_HEAD(, qemu_laiocb
) completed
;
319 if (s
->io_q
.in_flight
>= MAX_EVENTS
) {
323 QSIMPLEQ_FOREACH(aiocb
, &s
->io_q
.pending
, next
) {
324 iocbs
[len
++] = &aiocb
->iocb
;
325 if (s
->io_q
.in_flight
+ len
>= MAX_EVENTS
) {
330 ret
= io_submit(s
->ctx
, len
, iocbs
);
331 if (ret
== -EAGAIN
) {
335 /* Fail the first request, retry the rest */
336 aiocb
= QSIMPLEQ_FIRST(&s
->io_q
.pending
);
337 QSIMPLEQ_REMOVE_HEAD(&s
->io_q
.pending
, next
);
340 qemu_laio_process_completion(aiocb
);
344 s
->io_q
.in_flight
+= ret
;
345 s
->io_q
.in_queue
-= ret
;
346 aiocb
= container_of(iocbs
[ret
- 1], struct qemu_laiocb
, iocb
);
347 QSIMPLEQ_SPLIT_AFTER(&s
->io_q
.pending
, aiocb
, next
, &completed
);
348 } while (ret
== len
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
));
349 s
->io_q
.blocked
= (s
->io_q
.in_queue
> 0);
351 if (s
->io_q
.in_flight
) {
352 /* We can try to complete something just right away if there are
353 * still requests in-flight. */
354 qemu_laio_process_completions(s
);
356 * Even we have completed everything (in_flight == 0), the queue can
357 * have still pended requests (in_queue > 0). We do not attempt to
358 * repeat submission to avoid IO hang. The reason is simple: s->e is
359 * still set and completion callback will be called shortly and all
360 * pended requests will be submitted from there.
365 void laio_io_plug(BlockDriverState
*bs
, LinuxAioState
*s
)
370 void laio_io_unplug(BlockDriverState
*bs
, LinuxAioState
*s
)
372 assert(s
->io_q
.plugged
);
373 if (--s
->io_q
.plugged
== 0 &&
374 !s
->io_q
.blocked
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
)) {
379 static int laio_do_submit(int fd
, struct qemu_laiocb
*laiocb
, off_t offset
,
382 LinuxAioState
*s
= laiocb
->ctx
;
383 struct iocb
*iocbs
= &laiocb
->iocb
;
384 QEMUIOVector
*qiov
= laiocb
->qiov
;
388 io_prep_pwritev(iocbs
, fd
, qiov
->iov
, qiov
->niov
, offset
);
391 io_prep_preadv(iocbs
, fd
, qiov
->iov
, qiov
->niov
, offset
);
393 /* Currently Linux kernel does not support other operations */
395 fprintf(stderr
, "%s: invalid AIO request type 0x%x.\n",
399 io_set_eventfd(&laiocb
->iocb
, event_notifier_get_fd(&s
->e
));
401 QSIMPLEQ_INSERT_TAIL(&s
->io_q
.pending
, laiocb
, next
);
403 if (!s
->io_q
.blocked
&&
405 s
->io_q
.in_flight
+ s
->io_q
.in_queue
>= MAX_EVENTS
)) {
412 int coroutine_fn
laio_co_submit(BlockDriverState
*bs
, LinuxAioState
*s
, int fd
,
413 uint64_t offset
, QEMUIOVector
*qiov
, int type
)
416 struct qemu_laiocb laiocb
= {
417 .co
= qemu_coroutine_self(),
418 .nbytes
= qiov
->size
,
421 .is_read
= (type
== QEMU_AIO_READ
),
425 ret
= laio_do_submit(fd
, &laiocb
, offset
, type
);
430 if (laiocb
.ret
== -EINPROGRESS
) {
431 qemu_coroutine_yield();
436 BlockAIOCB
*laio_submit(BlockDriverState
*bs
, LinuxAioState
*s
, int fd
,
437 int64_t sector_num
, QEMUIOVector
*qiov
, int nb_sectors
,
438 BlockCompletionFunc
*cb
, void *opaque
, int type
)
440 struct qemu_laiocb
*laiocb
;
441 off_t offset
= sector_num
* BDRV_SECTOR_SIZE
;
444 laiocb
= qemu_aio_get(&laio_aiocb_info
, bs
, cb
, opaque
);
445 laiocb
->nbytes
= nb_sectors
* BDRV_SECTOR_SIZE
;
447 laiocb
->ret
= -EINPROGRESS
;
448 laiocb
->is_read
= (type
== QEMU_AIO_READ
);
451 ret
= laio_do_submit(fd
, laiocb
, offset
, type
);
453 qemu_aio_unref(laiocb
);
457 return &laiocb
->common
;
460 void laio_detach_aio_context(LinuxAioState
*s
, AioContext
*old_context
)
462 aio_set_event_notifier(old_context
, &s
->e
, false, NULL
, NULL
);
463 qemu_bh_delete(s
->completion_bh
);
464 s
->aio_context
= NULL
;
467 void laio_attach_aio_context(LinuxAioState
*s
, AioContext
*new_context
)
469 s
->aio_context
= new_context
;
470 s
->completion_bh
= aio_bh_new(new_context
, qemu_laio_completion_bh
, s
);
471 aio_set_event_notifier(new_context
, &s
->e
, false,
472 qemu_laio_completion_cb
,
476 LinuxAioState
*laio_init(void)
480 s
= g_malloc0(sizeof(*s
));
481 if (event_notifier_init(&s
->e
, false) < 0) {
485 if (io_setup(MAX_EVENTS
, &s
->ctx
) != 0) {
494 event_notifier_cleanup(&s
->e
);
500 void laio_cleanup(LinuxAioState
*s
)
502 event_notifier_cleanup(&s
->e
);
504 if (io_destroy(s
->ctx
) != 0) {
505 fprintf(stderr
, "%s: destroy AIO context %p failed\n",