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
{
53 AioContext
*aio_context
;
58 /* io queue for submit at batch */
61 /* I/O completion processing */
62 QEMUBH
*completion_bh
;
63 struct io_event events
[MAX_EVENTS
];
68 static void ioq_submit(LinuxAioState
*s
);
70 static inline ssize_t
io_event_ret(struct io_event
*ev
)
72 return (ssize_t
)(((uint64_t)ev
->res2
<< 32) | ev
->res
);
76 * Completes an AIO request (calls the callback and frees the ACB).
78 static void qemu_laio_process_completion(struct qemu_laiocb
*laiocb
)
83 if (ret
!= -ECANCELED
) {
84 if (ret
== laiocb
->nbytes
) {
86 } else if (ret
>= 0) {
87 /* Short reads mean EOF, pad with zeros. */
88 if (laiocb
->is_read
) {
89 qemu_iovec_memset(laiocb
->qiov
, ret
, 0,
90 laiocb
->qiov
->size
- ret
);
99 qemu_coroutine_enter(laiocb
->co
);
101 laiocb
->common
.cb(laiocb
->common
.opaque
, ret
);
102 qemu_aio_unref(laiocb
);
106 /* The completion BH fetches completed I/O requests and invokes their
109 * The function is somewhat tricky because it supports nested event loops, for
110 * example when a request callback invokes aio_poll(). In order to do this,
111 * the completion events array and index are kept in LinuxAioState. The BH
112 * reschedules itself as long as there are completions pending so it will
113 * either be called again in a nested event loop or will be called after all
114 * events have been completed. When there are no events left to complete, the
115 * BH returns without rescheduling.
117 static void qemu_laio_completion_bh(void *opaque
)
119 LinuxAioState
*s
= opaque
;
121 /* Fetch more completion events when empty */
122 if (s
->event_idx
== s
->event_max
) {
124 struct timespec ts
= { 0 };
125 s
->event_max
= io_getevents(s
->ctx
, MAX_EVENTS
, MAX_EVENTS
,
127 } while (s
->event_max
== -EINTR
);
130 if (s
->event_max
<= 0) {
132 return; /* no more events */
136 /* Reschedule so nested event loops see currently pending completions */
137 qemu_bh_schedule(s
->completion_bh
);
139 /* Process completion events */
140 while (s
->event_idx
< s
->event_max
) {
141 struct iocb
*iocb
= s
->events
[s
->event_idx
].obj
;
142 struct qemu_laiocb
*laiocb
=
143 container_of(iocb
, struct qemu_laiocb
, iocb
);
145 laiocb
->ret
= io_event_ret(&s
->events
[s
->event_idx
]);
148 qemu_laio_process_completion(laiocb
);
151 if (!s
->io_q
.plugged
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
)) {
155 qemu_bh_cancel(s
->completion_bh
);
158 static void qemu_laio_completion_cb(EventNotifier
*e
)
160 LinuxAioState
*s
= container_of(e
, LinuxAioState
, e
);
162 if (event_notifier_test_and_clear(&s
->e
)) {
163 qemu_laio_completion_bh(s
);
167 static void laio_cancel(BlockAIOCB
*blockacb
)
169 struct qemu_laiocb
*laiocb
= (struct qemu_laiocb
*)blockacb
;
170 struct io_event event
;
173 if (laiocb
->ret
!= -EINPROGRESS
) {
176 ret
= io_cancel(laiocb
->ctx
->ctx
, &laiocb
->iocb
, &event
);
177 laiocb
->ret
= -ECANCELED
;
179 /* iocb is not cancelled, cb will be called by the event loop later */
183 laiocb
->common
.cb(laiocb
->common
.opaque
, laiocb
->ret
);
186 static const AIOCBInfo laio_aiocb_info
= {
187 .aiocb_size
= sizeof(struct qemu_laiocb
),
188 .cancel_async
= laio_cancel
,
191 static void ioq_init(LaioQueue
*io_q
)
193 QSIMPLEQ_INIT(&io_q
->pending
);
196 io_q
->blocked
= false;
199 static void ioq_submit(LinuxAioState
*s
)
202 struct qemu_laiocb
*aiocb
;
203 struct iocb
*iocbs
[MAX_QUEUED_IO
];
204 QSIMPLEQ_HEAD(, qemu_laiocb
) completed
;
208 QSIMPLEQ_FOREACH(aiocb
, &s
->io_q
.pending
, next
) {
209 iocbs
[len
++] = &aiocb
->iocb
;
210 if (len
== MAX_QUEUED_IO
) {
215 ret
= io_submit(s
->ctx
, len
, iocbs
);
216 if (ret
== -EAGAIN
) {
224 aiocb
= container_of(iocbs
[ret
- 1], struct qemu_laiocb
, iocb
);
225 QSIMPLEQ_SPLIT_AFTER(&s
->io_q
.pending
, aiocb
, next
, &completed
);
226 } while (ret
== len
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
));
227 s
->io_q
.blocked
= (s
->io_q
.n
> 0);
230 void laio_io_plug(BlockDriverState
*bs
, LinuxAioState
*s
)
235 void laio_io_unplug(BlockDriverState
*bs
, LinuxAioState
*s
)
237 assert(s
->io_q
.plugged
);
238 if (--s
->io_q
.plugged
== 0 &&
239 !s
->io_q
.blocked
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
)) {
244 static int laio_do_submit(int fd
, struct qemu_laiocb
*laiocb
, off_t offset
,
247 LinuxAioState
*s
= laiocb
->ctx
;
248 struct iocb
*iocbs
= &laiocb
->iocb
;
249 QEMUIOVector
*qiov
= laiocb
->qiov
;
253 io_prep_pwritev(iocbs
, fd
, qiov
->iov
, qiov
->niov
, offset
);
256 io_prep_preadv(iocbs
, fd
, qiov
->iov
, qiov
->niov
, offset
);
258 /* Currently Linux kernel does not support other operations */
260 fprintf(stderr
, "%s: invalid AIO request type 0x%x.\n",
264 io_set_eventfd(&laiocb
->iocb
, event_notifier_get_fd(&s
->e
));
266 QSIMPLEQ_INSERT_TAIL(&s
->io_q
.pending
, laiocb
, next
);
268 if (!s
->io_q
.blocked
&&
269 (!s
->io_q
.plugged
|| s
->io_q
.n
>= MAX_QUEUED_IO
)) {
276 int coroutine_fn
laio_co_submit(BlockDriverState
*bs
, LinuxAioState
*s
, int fd
,
277 uint64_t offset
, QEMUIOVector
*qiov
, int type
)
280 struct qemu_laiocb laiocb
= {
281 .co
= qemu_coroutine_self(),
282 .nbytes
= qiov
->size
,
284 .is_read
= (type
== QEMU_AIO_READ
),
288 ret
= laio_do_submit(fd
, &laiocb
, offset
, type
);
293 qemu_coroutine_yield();
297 BlockAIOCB
*laio_submit(BlockDriverState
*bs
, LinuxAioState
*s
, int fd
,
298 int64_t sector_num
, QEMUIOVector
*qiov
, int nb_sectors
,
299 BlockCompletionFunc
*cb
, void *opaque
, int type
)
301 struct qemu_laiocb
*laiocb
;
302 off_t offset
= sector_num
* BDRV_SECTOR_SIZE
;
305 laiocb
= qemu_aio_get(&laio_aiocb_info
, bs
, cb
, opaque
);
306 laiocb
->nbytes
= nb_sectors
* BDRV_SECTOR_SIZE
;
308 laiocb
->ret
= -EINPROGRESS
;
309 laiocb
->is_read
= (type
== QEMU_AIO_READ
);
312 ret
= laio_do_submit(fd
, laiocb
, offset
, type
);
314 qemu_aio_unref(laiocb
);
318 return &laiocb
->common
;
321 void laio_detach_aio_context(LinuxAioState
*s
, AioContext
*old_context
)
323 aio_set_event_notifier(old_context
, &s
->e
, false, NULL
);
324 qemu_bh_delete(s
->completion_bh
);
327 void laio_attach_aio_context(LinuxAioState
*s
, AioContext
*new_context
)
329 s
->aio_context
= new_context
;
330 s
->completion_bh
= aio_bh_new(new_context
, qemu_laio_completion_bh
, s
);
331 aio_set_event_notifier(new_context
, &s
->e
, false,
332 qemu_laio_completion_cb
);
335 LinuxAioState
*laio_init(void)
339 s
= g_malloc0(sizeof(*s
));
340 if (event_notifier_init(&s
->e
, false) < 0) {
344 if (io_setup(MAX_EVENTS
, &s
->ctx
) != 0) {
353 event_notifier_cleanup(&s
->e
);
359 void laio_cleanup(LinuxAioState
*s
)
361 event_notifier_cleanup(&s
->e
);
363 if (io_destroy(s
->ctx
) != 0) {
364 fprintf(stderr
, "%s: destroy AIO context %p failed\n",