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
31 #define MAX_QUEUED_IO 128
42 QSIMPLEQ_ENTRY(qemu_laiocb
) next
;
49 QSIMPLEQ_HEAD(, qemu_laiocb
) pending
;
52 struct LinuxAioState
{
56 /* io queue for submit at batch */
59 /* I/O completion processing */
60 QEMUBH
*completion_bh
;
61 struct io_event events
[MAX_EVENTS
];
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
, NULL
);
99 laiocb
->common
.cb(laiocb
->common
.opaque
, ret
);
100 qemu_aio_unref(laiocb
);
104 /* The completion BH fetches completed I/O requests and invokes their
107 * The function is somewhat tricky because it supports nested event loops, for
108 * example when a request callback invokes aio_poll(). In order to do this,
109 * the completion events array and index are kept in LinuxAioState. The BH
110 * reschedules itself as long as there are completions pending so it will
111 * either be called again in a nested event loop or will be called after all
112 * events have been completed. When there are no events left to complete, the
113 * BH returns without rescheduling.
115 static void qemu_laio_completion_bh(void *opaque
)
117 LinuxAioState
*s
= opaque
;
119 /* Fetch more completion events when empty */
120 if (s
->event_idx
== s
->event_max
) {
122 struct timespec ts
= { 0 };
123 s
->event_max
= io_getevents(s
->ctx
, MAX_EVENTS
, MAX_EVENTS
,
125 } while (s
->event_max
== -EINTR
);
128 if (s
->event_max
<= 0) {
130 return; /* no more events */
134 /* Reschedule so nested event loops see currently pending completions */
135 qemu_bh_schedule(s
->completion_bh
);
137 /* Process completion events */
138 while (s
->event_idx
< s
->event_max
) {
139 struct iocb
*iocb
= s
->events
[s
->event_idx
].obj
;
140 struct qemu_laiocb
*laiocb
=
141 container_of(iocb
, struct qemu_laiocb
, iocb
);
143 laiocb
->ret
= io_event_ret(&s
->events
[s
->event_idx
]);
146 qemu_laio_process_completion(laiocb
);
149 if (!s
->io_q
.plugged
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
)) {
154 static void qemu_laio_completion_cb(EventNotifier
*e
)
156 LinuxAioState
*s
= container_of(e
, LinuxAioState
, e
);
158 if (event_notifier_test_and_clear(&s
->e
)) {
159 qemu_bh_schedule(s
->completion_bh
);
163 static void laio_cancel(BlockAIOCB
*blockacb
)
165 struct qemu_laiocb
*laiocb
= (struct qemu_laiocb
*)blockacb
;
166 struct io_event event
;
169 if (laiocb
->ret
!= -EINPROGRESS
) {
172 ret
= io_cancel(laiocb
->ctx
->ctx
, &laiocb
->iocb
, &event
);
173 laiocb
->ret
= -ECANCELED
;
175 /* iocb is not cancelled, cb will be called by the event loop later */
179 laiocb
->common
.cb(laiocb
->common
.opaque
, laiocb
->ret
);
182 static const AIOCBInfo laio_aiocb_info
= {
183 .aiocb_size
= sizeof(struct qemu_laiocb
),
184 .cancel_async
= laio_cancel
,
187 static void ioq_init(LaioQueue
*io_q
)
189 QSIMPLEQ_INIT(&io_q
->pending
);
192 io_q
->blocked
= false;
195 static void ioq_submit(LinuxAioState
*s
)
198 struct qemu_laiocb
*aiocb
;
199 struct iocb
*iocbs
[MAX_QUEUED_IO
];
200 QSIMPLEQ_HEAD(, qemu_laiocb
) completed
;
204 QSIMPLEQ_FOREACH(aiocb
, &s
->io_q
.pending
, next
) {
205 iocbs
[len
++] = &aiocb
->iocb
;
206 if (len
== MAX_QUEUED_IO
) {
211 ret
= io_submit(s
->ctx
, len
, iocbs
);
212 if (ret
== -EAGAIN
) {
220 aiocb
= container_of(iocbs
[ret
- 1], struct qemu_laiocb
, iocb
);
221 QSIMPLEQ_SPLIT_AFTER(&s
->io_q
.pending
, aiocb
, next
, &completed
);
222 } while (ret
== len
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
));
223 s
->io_q
.blocked
= (s
->io_q
.n
> 0);
226 void laio_io_plug(BlockDriverState
*bs
, LinuxAioState
*s
)
228 assert(!s
->io_q
.plugged
);
232 void laio_io_unplug(BlockDriverState
*bs
, LinuxAioState
*s
)
234 assert(s
->io_q
.plugged
);
236 if (!s
->io_q
.blocked
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
)) {
241 static int laio_do_submit(int fd
, struct qemu_laiocb
*laiocb
, off_t offset
,
244 LinuxAioState
*s
= laiocb
->ctx
;
245 struct iocb
*iocbs
= &laiocb
->iocb
;
246 QEMUIOVector
*qiov
= laiocb
->qiov
;
250 io_prep_pwritev(iocbs
, fd
, qiov
->iov
, qiov
->niov
, offset
);
253 io_prep_preadv(iocbs
, fd
, qiov
->iov
, qiov
->niov
, offset
);
255 /* Currently Linux kernel does not support other operations */
257 fprintf(stderr
, "%s: invalid AIO request type 0x%x.\n",
261 io_set_eventfd(&laiocb
->iocb
, event_notifier_get_fd(&s
->e
));
263 QSIMPLEQ_INSERT_TAIL(&s
->io_q
.pending
, laiocb
, next
);
265 if (!s
->io_q
.blocked
&&
266 (!s
->io_q
.plugged
|| s
->io_q
.n
>= MAX_QUEUED_IO
)) {
273 int coroutine_fn
laio_co_submit(BlockDriverState
*bs
, LinuxAioState
*s
, int fd
,
274 uint64_t offset
, QEMUIOVector
*qiov
, int type
)
277 struct qemu_laiocb laiocb
= {
278 .co
= qemu_coroutine_self(),
279 .nbytes
= qiov
->size
,
281 .is_read
= (type
== QEMU_AIO_READ
),
285 ret
= laio_do_submit(fd
, &laiocb
, offset
, type
);
290 qemu_coroutine_yield();
294 BlockAIOCB
*laio_submit(BlockDriverState
*bs
, LinuxAioState
*s
, int fd
,
295 int64_t sector_num
, QEMUIOVector
*qiov
, int nb_sectors
,
296 BlockCompletionFunc
*cb
, void *opaque
, int type
)
298 struct qemu_laiocb
*laiocb
;
299 off_t offset
= sector_num
* BDRV_SECTOR_SIZE
;
302 laiocb
= qemu_aio_get(&laio_aiocb_info
, bs
, cb
, opaque
);
303 laiocb
->nbytes
= nb_sectors
* BDRV_SECTOR_SIZE
;
305 laiocb
->ret
= -EINPROGRESS
;
306 laiocb
->is_read
= (type
== QEMU_AIO_READ
);
309 ret
= laio_do_submit(fd
, laiocb
, offset
, type
);
311 qemu_aio_unref(laiocb
);
315 return &laiocb
->common
;
318 void laio_detach_aio_context(LinuxAioState
*s
, AioContext
*old_context
)
320 aio_set_event_notifier(old_context
, &s
->e
, false, NULL
);
321 qemu_bh_delete(s
->completion_bh
);
324 void laio_attach_aio_context(LinuxAioState
*s
, AioContext
*new_context
)
326 s
->completion_bh
= aio_bh_new(new_context
, qemu_laio_completion_bh
, s
);
327 aio_set_event_notifier(new_context
, &s
->e
, false,
328 qemu_laio_completion_cb
);
331 LinuxAioState
*laio_init(void)
335 s
= g_malloc0(sizeof(*s
));
336 if (event_notifier_init(&s
->e
, false) < 0) {
340 if (io_setup(MAX_EVENTS
, &s
->ctx
) != 0) {
349 event_notifier_cleanup(&s
->e
);
355 void laio_cleanup(LinuxAioState
*s
)
357 event_notifier_cleanup(&s
->e
);
359 if (io_destroy(s
->ctx
) != 0) {
360 fprintf(stderr
, "%s: destroy AIO context %p failed\n",