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 */
60 /* I/O completion processing */
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
)
81 if (ret
!= -ECANCELED
) {
82 if (ret
== laiocb
->nbytes
) {
84 } else if (ret
>= 0) {
85 /* Short reads mean EOF, pad with zeros. */
86 if (laiocb
->is_read
) {
87 qemu_iovec_memset(laiocb
->qiov
, ret
, 0,
88 laiocb
->qiov
->size
- ret
);
97 qemu_coroutine_enter(laiocb
->co
);
99 laiocb
->common
.cb(laiocb
->common
.opaque
, ret
);
100 qemu_aio_unref(laiocb
);
105 * aio_ring buffer which is shared between userspace and kernel.
107 * This copied from linux/fs/aio.c, common header does not exist
108 * but AIO exists for ages so we assume ABI is stable.
111 unsigned id
; /* kernel internal index number */
112 unsigned nr
; /* number of io_events */
113 unsigned head
; /* Written to by userland or by kernel. */
117 unsigned compat_features
;
118 unsigned incompat_features
;
119 unsigned header_length
; /* size of aio_ring */
121 struct io_event io_events
[0];
127 * @events: pointer on events array, output value
129 * Returns the number of completed events and sets a pointer
130 * on events array. This function does not update the internal
131 * ring buffer, only reads head and tail. When @events has been
132 * processed io_getevents_commit() must be called.
134 static inline unsigned int io_getevents_peek(io_context_t ctx
,
135 struct io_event
**events
)
137 struct aio_ring
*ring
= (struct aio_ring
*)ctx
;
138 unsigned int head
= ring
->head
, tail
= ring
->tail
;
141 nr
= tail
>= head
? tail
- head
: ring
->nr
- head
;
142 *events
= ring
->io_events
+ head
;
143 /* To avoid speculative loads of s->events[i] before observing tail.
144 Paired with smp_wmb() inside linux/fs/aio.c: aio_complete(). */
151 * io_getevents_commit:
153 * @nr: the number of events on which head should be advanced
155 * Advances head of a ring buffer.
157 static inline void io_getevents_commit(io_context_t ctx
, unsigned int nr
)
159 struct aio_ring
*ring
= (struct aio_ring
*)ctx
;
162 ring
->head
= (ring
->head
+ nr
) % ring
->nr
;
167 * io_getevents_advance_and_peek:
169 * @events: pointer on events array, output value
170 * @nr: the number of events on which head should be advanced
172 * Advances head of a ring buffer and returns number of elements left.
174 static inline unsigned int
175 io_getevents_advance_and_peek(io_context_t ctx
,
176 struct io_event
**events
,
179 io_getevents_commit(ctx
, nr
);
180 return io_getevents_peek(ctx
, events
);
184 * qemu_laio_process_completions:
187 * Fetches completed I/O requests and invokes their callbacks.
189 * The function is somewhat tricky because it supports nested event loops, for
190 * example when a request callback invokes aio_poll(). In order to do this,
191 * indices are kept in LinuxAioState. Function schedules BH completion so it
192 * can be called again in a nested event loop. When there are no events left
193 * to complete the BH is being canceled.
195 static void qemu_laio_process_completions(LinuxAioState
*s
)
197 struct io_event
*events
;
199 /* Reschedule so nested event loops see currently pending completions */
200 qemu_bh_schedule(s
->completion_bh
);
202 while ((s
->event_max
= io_getevents_advance_and_peek(s
->ctx
, &events
,
204 for (s
->event_idx
= 0; s
->event_idx
< s
->event_max
; ) {
205 struct iocb
*iocb
= events
[s
->event_idx
].obj
;
206 struct qemu_laiocb
*laiocb
=
207 container_of(iocb
, struct qemu_laiocb
, iocb
);
209 laiocb
->ret
= io_event_ret(&events
[s
->event_idx
]);
211 /* Change counters one-by-one because we can be nested. */
214 qemu_laio_process_completion(laiocb
);
218 qemu_bh_cancel(s
->completion_bh
);
220 /* If we are nested we have to notify the level above that we are done
221 * by setting event_max to zero, upper level will then jump out of it's
222 * own `for` loop. If we are the last all counters droped to zero. */
227 static void qemu_laio_process_completions_and_submit(LinuxAioState
*s
)
229 qemu_laio_process_completions(s
);
230 if (!s
->io_q
.plugged
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
)) {
235 static void qemu_laio_completion_bh(void *opaque
)
237 LinuxAioState
*s
= opaque
;
239 qemu_laio_process_completions_and_submit(s
);
242 static void qemu_laio_completion_cb(EventNotifier
*e
)
244 LinuxAioState
*s
= container_of(e
, LinuxAioState
, e
);
246 if (event_notifier_test_and_clear(&s
->e
)) {
247 qemu_laio_process_completions_and_submit(s
);
251 static void laio_cancel(BlockAIOCB
*blockacb
)
253 struct qemu_laiocb
*laiocb
= (struct qemu_laiocb
*)blockacb
;
254 struct io_event event
;
257 if (laiocb
->ret
!= -EINPROGRESS
) {
260 ret
= io_cancel(laiocb
->ctx
->ctx
, &laiocb
->iocb
, &event
);
261 laiocb
->ret
= -ECANCELED
;
263 /* iocb is not cancelled, cb will be called by the event loop later */
267 laiocb
->common
.cb(laiocb
->common
.opaque
, laiocb
->ret
);
270 static const AIOCBInfo laio_aiocb_info
= {
271 .aiocb_size
= sizeof(struct qemu_laiocb
),
272 .cancel_async
= laio_cancel
,
275 static void ioq_init(LaioQueue
*io_q
)
277 QSIMPLEQ_INIT(&io_q
->pending
);
281 io_q
->blocked
= false;
284 static void ioq_submit(LinuxAioState
*s
)
287 struct qemu_laiocb
*aiocb
;
288 struct iocb
*iocbs
[MAX_EVENTS
];
289 QSIMPLEQ_HEAD(, qemu_laiocb
) completed
;
292 if (s
->io_q
.in_flight
>= MAX_EVENTS
) {
296 QSIMPLEQ_FOREACH(aiocb
, &s
->io_q
.pending
, next
) {
297 iocbs
[len
++] = &aiocb
->iocb
;
298 if (s
->io_q
.in_flight
+ len
>= MAX_EVENTS
) {
303 ret
= io_submit(s
->ctx
, len
, iocbs
);
304 if (ret
== -EAGAIN
) {
308 /* Fail the first request, retry the rest */
309 aiocb
= QSIMPLEQ_FIRST(&s
->io_q
.pending
);
310 QSIMPLEQ_REMOVE_HEAD(&s
->io_q
.pending
, next
);
313 qemu_laio_process_completion(aiocb
);
317 s
->io_q
.in_flight
+= ret
;
318 s
->io_q
.in_queue
-= ret
;
319 aiocb
= container_of(iocbs
[ret
- 1], struct qemu_laiocb
, iocb
);
320 QSIMPLEQ_SPLIT_AFTER(&s
->io_q
.pending
, aiocb
, next
, &completed
);
321 } while (ret
== len
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
));
322 s
->io_q
.blocked
= (s
->io_q
.in_queue
> 0);
325 void laio_io_plug(BlockDriverState
*bs
, LinuxAioState
*s
)
330 void laio_io_unplug(BlockDriverState
*bs
, LinuxAioState
*s
)
332 assert(s
->io_q
.plugged
);
333 if (--s
->io_q
.plugged
== 0 &&
334 !s
->io_q
.blocked
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
)) {
339 static int laio_do_submit(int fd
, struct qemu_laiocb
*laiocb
, off_t offset
,
342 LinuxAioState
*s
= laiocb
->ctx
;
343 struct iocb
*iocbs
= &laiocb
->iocb
;
344 QEMUIOVector
*qiov
= laiocb
->qiov
;
348 io_prep_pwritev(iocbs
, fd
, qiov
->iov
, qiov
->niov
, offset
);
351 io_prep_preadv(iocbs
, fd
, qiov
->iov
, qiov
->niov
, offset
);
353 /* Currently Linux kernel does not support other operations */
355 fprintf(stderr
, "%s: invalid AIO request type 0x%x.\n",
359 io_set_eventfd(&laiocb
->iocb
, event_notifier_get_fd(&s
->e
));
361 QSIMPLEQ_INSERT_TAIL(&s
->io_q
.pending
, laiocb
, next
);
363 if (!s
->io_q
.blocked
&&
365 s
->io_q
.in_flight
+ s
->io_q
.in_queue
>= MAX_EVENTS
)) {
372 int coroutine_fn
laio_co_submit(BlockDriverState
*bs
, LinuxAioState
*s
, int fd
,
373 uint64_t offset
, QEMUIOVector
*qiov
, int type
)
376 struct qemu_laiocb laiocb
= {
377 .co
= qemu_coroutine_self(),
378 .nbytes
= qiov
->size
,
380 .is_read
= (type
== QEMU_AIO_READ
),
384 ret
= laio_do_submit(fd
, &laiocb
, offset
, type
);
389 qemu_coroutine_yield();
393 BlockAIOCB
*laio_submit(BlockDriverState
*bs
, LinuxAioState
*s
, int fd
,
394 int64_t sector_num
, QEMUIOVector
*qiov
, int nb_sectors
,
395 BlockCompletionFunc
*cb
, void *opaque
, int type
)
397 struct qemu_laiocb
*laiocb
;
398 off_t offset
= sector_num
* BDRV_SECTOR_SIZE
;
401 laiocb
= qemu_aio_get(&laio_aiocb_info
, bs
, cb
, opaque
);
402 laiocb
->nbytes
= nb_sectors
* BDRV_SECTOR_SIZE
;
404 laiocb
->ret
= -EINPROGRESS
;
405 laiocb
->is_read
= (type
== QEMU_AIO_READ
);
408 ret
= laio_do_submit(fd
, laiocb
, offset
, type
);
410 qemu_aio_unref(laiocb
);
414 return &laiocb
->common
;
417 void laio_detach_aio_context(LinuxAioState
*s
, AioContext
*old_context
)
419 aio_set_event_notifier(old_context
, &s
->e
, false, NULL
);
420 qemu_bh_delete(s
->completion_bh
);
423 void laio_attach_aio_context(LinuxAioState
*s
, AioContext
*new_context
)
425 s
->aio_context
= new_context
;
426 s
->completion_bh
= aio_bh_new(new_context
, qemu_laio_completion_bh
, s
);
427 aio_set_event_notifier(new_context
, &s
->e
, false,
428 qemu_laio_completion_cb
);
431 LinuxAioState
*laio_init(void)
435 s
= g_malloc0(sizeof(*s
));
436 if (event_notifier_init(&s
->e
, false) < 0) {
440 if (io_setup(MAX_EVENTS
, &s
->ctx
) != 0) {
449 event_notifier_cleanup(&s
->e
);
455 void laio_cleanup(LinuxAioState
*s
)
457 event_notifier_cleanup(&s
->e
);
459 if (io_destroy(s
->ctx
) != 0) {
460 fprintf(stderr
, "%s: destroy AIO context %p failed\n",