]>
git.proxmox.com Git - mirror_qemu.git/blob - block/linux-aio.c
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 "block/aio.h"
12 #include "qemu/queue.h"
13 #include "block/block.h"
14 #include "block/raw-aio.h"
15 #include "qemu/event_notifier.h"
16 #include "qemu/coroutine.h"
17 #include "qemu/defer-call.h"
18 #include "qapi/error.h"
19 #include "sysemu/block-backend.h"
21 /* Only used for assertions. */
22 #include "qemu/coroutine_int.h"
27 * Queue size (per-device).
29 * XXX: eventually we need to communicate this to the guest and/or make it
30 * tunable by the guest. If we get more outstanding requests at a time
31 * than this we will get EAGAIN from io_submit which is communicated to
32 * the guest as an I/O error.
34 #define MAX_EVENTS 1024
36 /* Maximum number of requests in a batch. (default value) */
37 #define DEFAULT_MAX_BATCH 32
47 QSIMPLEQ_ENTRY(qemu_laiocb
) next
;
51 unsigned int in_queue
;
52 unsigned int in_flight
;
54 QSIMPLEQ_HEAD(, qemu_laiocb
) pending
;
57 struct LinuxAioState
{
58 AioContext
*aio_context
;
63 /* No locking required, only accessed from AioContext home thread */
65 QEMUBH
*completion_bh
;
70 static void ioq_submit(LinuxAioState
*s
);
72 static inline ssize_t
io_event_ret(struct io_event
*ev
)
74 return (ssize_t
)(((uint64_t)ev
->res2
<< 32) | ev
->res
);
78 * Completes an AIO request.
80 static void qemu_laio_process_completion(struct qemu_laiocb
*laiocb
)
85 if (ret
!= -ECANCELED
) {
86 if (ret
== laiocb
->nbytes
) {
88 } else if (ret
>= 0) {
89 /* Short reads mean EOF, pad with zeros. */
90 if (laiocb
->is_read
) {
91 qemu_iovec_memset(laiocb
->qiov
, ret
, 0,
92 laiocb
->qiov
->size
- ret
);
102 * If the coroutine is already entered it must be in ioq_submit() and
103 * will notice laio->ret has been filled in when it eventually runs
104 * later. Coroutines cannot be entered recursively so avoid doing
107 assert(laiocb
->co
->ctx
== laiocb
->ctx
->aio_context
);
108 if (!qemu_coroutine_entered(laiocb
->co
)) {
109 aio_co_wake(laiocb
->co
);
114 * aio_ring buffer which is shared between userspace and kernel.
116 * This copied from linux/fs/aio.c, common header does not exist
117 * but AIO exists for ages so we assume ABI is stable.
120 unsigned id
; /* kernel internal index number */
121 unsigned nr
; /* number of io_events */
122 unsigned head
; /* Written to by userland or by kernel. */
126 unsigned compat_features
;
127 unsigned incompat_features
;
128 unsigned header_length
; /* size of aio_ring */
130 struct io_event io_events
[];
136 * @events: pointer on events array, output value
138 * Returns the number of completed events and sets a pointer
139 * on events array. This function does not update the internal
140 * ring buffer, only reads head and tail. When @events has been
141 * processed io_getevents_commit() must be called.
143 static inline unsigned int io_getevents_peek(io_context_t ctx
,
144 struct io_event
**events
)
146 struct aio_ring
*ring
= (struct aio_ring
*)ctx
;
147 unsigned int head
= ring
->head
, tail
= ring
->tail
;
150 nr
= tail
>= head
? tail
- head
: ring
->nr
- head
;
151 *events
= ring
->io_events
+ head
;
152 /* To avoid speculative loads of s->events[i] before observing tail.
153 Paired with smp_wmb() inside linux/fs/aio.c: aio_complete(). */
160 * io_getevents_commit:
162 * @nr: the number of events on which head should be advanced
164 * Advances head of a ring buffer.
166 static inline void io_getevents_commit(io_context_t ctx
, unsigned int nr
)
168 struct aio_ring
*ring
= (struct aio_ring
*)ctx
;
171 ring
->head
= (ring
->head
+ nr
) % ring
->nr
;
176 * io_getevents_advance_and_peek:
178 * @events: pointer on events array, output value
179 * @nr: the number of events on which head should be advanced
181 * Advances head of a ring buffer and returns number of elements left.
183 static inline unsigned int
184 io_getevents_advance_and_peek(io_context_t ctx
,
185 struct io_event
**events
,
188 io_getevents_commit(ctx
, nr
);
189 return io_getevents_peek(ctx
, events
);
193 * qemu_laio_process_completions:
196 * Fetches completed I/O requests and invokes their callbacks.
198 * The function is somewhat tricky because it supports nested event loops, for
199 * example when a request callback invokes aio_poll(). In order to do this,
200 * indices are kept in LinuxAioState. Function schedules BH completion so it
201 * can be called again in a nested event loop. When there are no events left
202 * to complete the BH is being canceled.
204 static void qemu_laio_process_completions(LinuxAioState
*s
)
206 struct io_event
*events
;
208 /* Reschedule so nested event loops see currently pending completions */
209 qemu_bh_schedule(s
->completion_bh
);
211 while ((s
->event_max
= io_getevents_advance_and_peek(s
->ctx
, &events
,
213 for (s
->event_idx
= 0; s
->event_idx
< s
->event_max
; ) {
214 struct iocb
*iocb
= events
[s
->event_idx
].obj
;
215 struct qemu_laiocb
*laiocb
=
216 container_of(iocb
, struct qemu_laiocb
, iocb
);
218 laiocb
->ret
= io_event_ret(&events
[s
->event_idx
]);
220 /* Change counters one-by-one because we can be nested. */
223 qemu_laio_process_completion(laiocb
);
227 qemu_bh_cancel(s
->completion_bh
);
229 /* If we are nested we have to notify the level above that we are done
230 * by setting event_max to zero, upper level will then jump out of it's
231 * own `for` loop. If we are the last all counters dropped to zero. */
236 static void qemu_laio_process_completions_and_submit(LinuxAioState
*s
)
238 qemu_laio_process_completions(s
);
240 if (!QSIMPLEQ_EMPTY(&s
->io_q
.pending
)) {
245 static void qemu_laio_completion_bh(void *opaque
)
247 LinuxAioState
*s
= opaque
;
249 qemu_laio_process_completions_and_submit(s
);
252 static void qemu_laio_completion_cb(EventNotifier
*e
)
254 LinuxAioState
*s
= container_of(e
, LinuxAioState
, e
);
256 if (event_notifier_test_and_clear(&s
->e
)) {
257 qemu_laio_process_completions_and_submit(s
);
261 static bool qemu_laio_poll_cb(void *opaque
)
263 EventNotifier
*e
= opaque
;
264 LinuxAioState
*s
= container_of(e
, LinuxAioState
, e
);
265 struct io_event
*events
;
267 return io_getevents_peek(s
->ctx
, &events
);
270 static void qemu_laio_poll_ready(EventNotifier
*opaque
)
272 EventNotifier
*e
= opaque
;
273 LinuxAioState
*s
= container_of(e
, LinuxAioState
, e
);
275 qemu_laio_process_completions_and_submit(s
);
278 static void ioq_init(LaioQueue
*io_q
)
280 QSIMPLEQ_INIT(&io_q
->pending
);
283 io_q
->blocked
= false;
286 static void ioq_submit(LinuxAioState
*s
)
289 struct qemu_laiocb
*aiocb
;
290 struct iocb
*iocbs
[MAX_EVENTS
];
291 QSIMPLEQ_HEAD(, qemu_laiocb
) completed
;
294 if (s
->io_q
.in_flight
>= MAX_EVENTS
) {
298 QSIMPLEQ_FOREACH(aiocb
, &s
->io_q
.pending
, next
) {
299 iocbs
[len
++] = &aiocb
->iocb
;
300 if (s
->io_q
.in_flight
+ len
>= MAX_EVENTS
) {
305 ret
= io_submit(s
->ctx
, len
, iocbs
);
306 if (ret
== -EAGAIN
) {
310 /* Fail the first request, retry the rest */
311 aiocb
= QSIMPLEQ_FIRST(&s
->io_q
.pending
);
312 QSIMPLEQ_REMOVE_HEAD(&s
->io_q
.pending
, next
);
315 qemu_laio_process_completion(aiocb
);
319 s
->io_q
.in_flight
+= ret
;
320 s
->io_q
.in_queue
-= ret
;
321 aiocb
= container_of(iocbs
[ret
- 1], struct qemu_laiocb
, iocb
);
322 QSIMPLEQ_SPLIT_AFTER(&s
->io_q
.pending
, aiocb
, next
, &completed
);
323 } while (ret
== len
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
));
324 s
->io_q
.blocked
= (s
->io_q
.in_queue
> 0);
326 if (s
->io_q
.in_flight
) {
327 /* We can try to complete something just right away if there are
328 * still requests in-flight. */
329 qemu_laio_process_completions(s
);
331 * Even we have completed everything (in_flight == 0), the queue can
332 * have still pended requests (in_queue > 0). We do not attempt to
333 * repeat submission to avoid IO hang. The reason is simple: s->e is
334 * still set and completion callback will be called shortly and all
335 * pended requests will be submitted from there.
340 static uint64_t laio_max_batch(LinuxAioState
*s
, uint64_t dev_max_batch
)
342 uint64_t max_batch
= s
->aio_context
->aio_max_batch
?: DEFAULT_MAX_BATCH
;
345 * AIO context can be shared between multiple block devices, so
346 * `dev_max_batch` allows reducing the batch size for latency-sensitive
349 max_batch
= MIN_NON_ZERO(dev_max_batch
, max_batch
);
351 /* limit the batch with the number of available events */
352 max_batch
= MIN_NON_ZERO(MAX_EVENTS
- s
->io_q
.in_flight
, max_batch
);
357 static void laio_deferred_fn(void *opaque
)
359 LinuxAioState
*s
= opaque
;
361 if (!s
->io_q
.blocked
&& !QSIMPLEQ_EMPTY(&s
->io_q
.pending
)) {
366 static int laio_do_submit(int fd
, struct qemu_laiocb
*laiocb
, off_t offset
,
367 int type
, uint64_t dev_max_batch
)
369 LinuxAioState
*s
= laiocb
->ctx
;
370 struct iocb
*iocbs
= &laiocb
->iocb
;
371 QEMUIOVector
*qiov
= laiocb
->qiov
;
375 io_prep_pwritev(iocbs
, fd
, qiov
->iov
, qiov
->niov
, offset
);
377 case QEMU_AIO_ZONE_APPEND
:
378 io_prep_pwritev(iocbs
, fd
, qiov
->iov
, qiov
->niov
, offset
);
381 io_prep_preadv(iocbs
, fd
, qiov
->iov
, qiov
->niov
, offset
);
383 /* Currently Linux kernel does not support other operations */
385 fprintf(stderr
, "%s: invalid AIO request type 0x%x.\n",
389 io_set_eventfd(&laiocb
->iocb
, event_notifier_get_fd(&s
->e
));
391 QSIMPLEQ_INSERT_TAIL(&s
->io_q
.pending
, laiocb
, next
);
393 if (!s
->io_q
.blocked
) {
394 if (s
->io_q
.in_queue
>= laio_max_batch(s
, dev_max_batch
)) {
397 defer_call(laio_deferred_fn
, s
);
404 int coroutine_fn
laio_co_submit(int fd
, uint64_t offset
, QEMUIOVector
*qiov
,
405 int type
, uint64_t dev_max_batch
)
408 AioContext
*ctx
= qemu_get_current_aio_context();
409 struct qemu_laiocb laiocb
= {
410 .co
= qemu_coroutine_self(),
411 .nbytes
= qiov
->size
,
412 .ctx
= aio_get_linux_aio(ctx
),
414 .is_read
= (type
== QEMU_AIO_READ
),
418 ret
= laio_do_submit(fd
, &laiocb
, offset
, type
, dev_max_batch
);
423 if (laiocb
.ret
== -EINPROGRESS
) {
424 qemu_coroutine_yield();
429 void laio_detach_aio_context(LinuxAioState
*s
, AioContext
*old_context
)
431 aio_set_event_notifier(old_context
, &s
->e
, NULL
, NULL
, NULL
);
432 qemu_bh_delete(s
->completion_bh
);
433 s
->aio_context
= NULL
;
436 void laio_attach_aio_context(LinuxAioState
*s
, AioContext
*new_context
)
438 s
->aio_context
= new_context
;
439 s
->completion_bh
= aio_bh_new(new_context
, qemu_laio_completion_bh
, s
);
440 aio_set_event_notifier(new_context
, &s
->e
,
441 qemu_laio_completion_cb
,
443 qemu_laio_poll_ready
);
446 LinuxAioState
*laio_init(Error
**errp
)
451 s
= g_malloc0(sizeof(*s
));
452 rc
= event_notifier_init(&s
->e
, false);
454 error_setg_errno(errp
, -rc
, "failed to initialize event notifier");
458 rc
= io_setup(MAX_EVENTS
, &s
->ctx
);
460 error_setg_errno(errp
, -rc
, "failed to create linux AIO context");
469 event_notifier_cleanup(&s
->e
);
475 void laio_cleanup(LinuxAioState
*s
)
477 event_notifier_cleanup(&s
->e
);
479 if (io_destroy(s
->ctx
) != 0) {
480 fprintf(stderr
, "%s: destroy AIO context %p failed\n",