block: move logical block size check function to a common utility function
[qemu.git] / block / linux-aio.c
blob3c0527c2bf8970960bd66077d6ba81da6a7cc85c
1 /*
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.
9 */
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 "qapi/error.h"
19 #include <libaio.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 1024
31 struct qemu_laiocb {
32 Coroutine *co;
33 LinuxAioState *ctx;
34 struct iocb iocb;
35 ssize_t ret;
36 size_t nbytes;
37 QEMUIOVector *qiov;
38 bool is_read;
39 QSIMPLEQ_ENTRY(qemu_laiocb) next;
42 typedef struct {
43 int plugged;
44 unsigned int in_queue;
45 unsigned int in_flight;
46 bool blocked;
47 QSIMPLEQ_HEAD(, qemu_laiocb) pending;
48 } LaioQueue;
50 struct LinuxAioState {
51 AioContext *aio_context;
53 io_context_t ctx;
54 EventNotifier e;
56 /* io queue for submit at batch. Protected by AioContext lock. */
57 LaioQueue io_q;
59 /* I/O completion processing. Only runs in I/O thread. */
60 QEMUBH *completion_bh;
61 int event_idx;
62 int event_max;
65 static void ioq_submit(LinuxAioState *s);
67 static inline ssize_t io_event_ret(struct io_event *ev)
69 return (ssize_t)(((uint64_t)ev->res2 << 32) | ev->res);
73 * Completes an AIO request.
75 static void qemu_laio_process_completion(struct qemu_laiocb *laiocb)
77 int ret;
79 ret = laiocb->ret;
80 if (ret != -ECANCELED) {
81 if (ret == laiocb->nbytes) {
82 ret = 0;
83 } else if (ret >= 0) {
84 /* Short reads mean EOF, pad with zeros. */
85 if (laiocb->is_read) {
86 qemu_iovec_memset(laiocb->qiov, ret, 0,
87 laiocb->qiov->size - ret);
88 } else {
89 ret = -ENOSPC;
94 laiocb->ret = ret;
97 * If the coroutine is already entered it must be in ioq_submit() and
98 * will notice laio->ret has been filled in when it eventually runs
99 * later. Coroutines cannot be entered recursively so avoid doing
100 * that!
102 if (!qemu_coroutine_entered(laiocb->co)) {
103 aio_co_wake(laiocb->co);
108 * aio_ring buffer which is shared between userspace and kernel.
110 * This copied from linux/fs/aio.c, common header does not exist
111 * but AIO exists for ages so we assume ABI is stable.
113 struct aio_ring {
114 unsigned id; /* kernel internal index number */
115 unsigned nr; /* number of io_events */
116 unsigned head; /* Written to by userland or by kernel. */
117 unsigned tail;
119 unsigned magic;
120 unsigned compat_features;
121 unsigned incompat_features;
122 unsigned header_length; /* size of aio_ring */
124 struct io_event io_events[];
128 * io_getevents_peek:
129 * @ctx: AIO context
130 * @events: pointer on events array, output value
132 * Returns the number of completed events and sets a pointer
133 * on events array. This function does not update the internal
134 * ring buffer, only reads head and tail. When @events has been
135 * processed io_getevents_commit() must be called.
137 static inline unsigned int io_getevents_peek(io_context_t ctx,
138 struct io_event **events)
140 struct aio_ring *ring = (struct aio_ring *)ctx;
141 unsigned int head = ring->head, tail = ring->tail;
142 unsigned int nr;
144 nr = tail >= head ? tail - head : ring->nr - head;
145 *events = ring->io_events + head;
146 /* To avoid speculative loads of s->events[i] before observing tail.
147 Paired with smp_wmb() inside linux/fs/aio.c: aio_complete(). */
148 smp_rmb();
150 return nr;
154 * io_getevents_commit:
155 * @ctx: AIO context
156 * @nr: the number of events on which head should be advanced
158 * Advances head of a ring buffer.
160 static inline void io_getevents_commit(io_context_t ctx, unsigned int nr)
162 struct aio_ring *ring = (struct aio_ring *)ctx;
164 if (nr) {
165 ring->head = (ring->head + nr) % ring->nr;
170 * io_getevents_advance_and_peek:
171 * @ctx: AIO context
172 * @events: pointer on events array, output value
173 * @nr: the number of events on which head should be advanced
175 * Advances head of a ring buffer and returns number of elements left.
177 static inline unsigned int
178 io_getevents_advance_and_peek(io_context_t ctx,
179 struct io_event **events,
180 unsigned int nr)
182 io_getevents_commit(ctx, nr);
183 return io_getevents_peek(ctx, events);
187 * qemu_laio_process_completions:
188 * @s: AIO state
190 * Fetches completed I/O requests and invokes their callbacks.
192 * The function is somewhat tricky because it supports nested event loops, for
193 * example when a request callback invokes aio_poll(). In order to do this,
194 * indices are kept in LinuxAioState. Function schedules BH completion so it
195 * can be called again in a nested event loop. When there are no events left
196 * to complete the BH is being canceled.
198 static void qemu_laio_process_completions(LinuxAioState *s)
200 struct io_event *events;
202 /* Reschedule so nested event loops see currently pending completions */
203 qemu_bh_schedule(s->completion_bh);
205 while ((s->event_max = io_getevents_advance_and_peek(s->ctx, &events,
206 s->event_idx))) {
207 for (s->event_idx = 0; s->event_idx < s->event_max; ) {
208 struct iocb *iocb = events[s->event_idx].obj;
209 struct qemu_laiocb *laiocb =
210 container_of(iocb, struct qemu_laiocb, iocb);
212 laiocb->ret = io_event_ret(&events[s->event_idx]);
214 /* Change counters one-by-one because we can be nested. */
215 s->io_q.in_flight--;
216 s->event_idx++;
217 qemu_laio_process_completion(laiocb);
221 qemu_bh_cancel(s->completion_bh);
223 /* If we are nested we have to notify the level above that we are done
224 * by setting event_max to zero, upper level will then jump out of it's
225 * own `for` loop. If we are the last all counters droped to zero. */
226 s->event_max = 0;
227 s->event_idx = 0;
230 static void qemu_laio_process_completions_and_submit(LinuxAioState *s)
232 aio_context_acquire(s->aio_context);
233 qemu_laio_process_completions(s);
235 if (!s->io_q.plugged && !QSIMPLEQ_EMPTY(&s->io_q.pending)) {
236 ioq_submit(s);
238 aio_context_release(s->aio_context);
241 static void qemu_laio_completion_bh(void *opaque)
243 LinuxAioState *s = opaque;
245 qemu_laio_process_completions_and_submit(s);
248 static void qemu_laio_completion_cb(EventNotifier *e)
250 LinuxAioState *s = container_of(e, LinuxAioState, e);
252 if (event_notifier_test_and_clear(&s->e)) {
253 qemu_laio_process_completions_and_submit(s);
257 static bool qemu_laio_poll_cb(void *opaque)
259 EventNotifier *e = opaque;
260 LinuxAioState *s = container_of(e, LinuxAioState, e);
261 struct io_event *events;
263 if (!io_getevents_peek(s->ctx, &events)) {
264 return false;
267 qemu_laio_process_completions_and_submit(s);
268 return true;
271 static void ioq_init(LaioQueue *io_q)
273 QSIMPLEQ_INIT(&io_q->pending);
274 io_q->plugged = 0;
275 io_q->in_queue = 0;
276 io_q->in_flight = 0;
277 io_q->blocked = false;
280 static void ioq_submit(LinuxAioState *s)
282 int ret, len;
283 struct qemu_laiocb *aiocb;
284 struct iocb *iocbs[MAX_EVENTS];
285 QSIMPLEQ_HEAD(, qemu_laiocb) completed;
287 do {
288 if (s->io_q.in_flight >= MAX_EVENTS) {
289 break;
291 len = 0;
292 QSIMPLEQ_FOREACH(aiocb, &s->io_q.pending, next) {
293 iocbs[len++] = &aiocb->iocb;
294 if (s->io_q.in_flight + len >= MAX_EVENTS) {
295 break;
299 ret = io_submit(s->ctx, len, iocbs);
300 if (ret == -EAGAIN) {
301 break;
303 if (ret < 0) {
304 /* Fail the first request, retry the rest */
305 aiocb = QSIMPLEQ_FIRST(&s->io_q.pending);
306 QSIMPLEQ_REMOVE_HEAD(&s->io_q.pending, next);
307 s->io_q.in_queue--;
308 aiocb->ret = ret;
309 qemu_laio_process_completion(aiocb);
310 continue;
313 s->io_q.in_flight += ret;
314 s->io_q.in_queue -= ret;
315 aiocb = container_of(iocbs[ret - 1], struct qemu_laiocb, iocb);
316 QSIMPLEQ_SPLIT_AFTER(&s->io_q.pending, aiocb, next, &completed);
317 } while (ret == len && !QSIMPLEQ_EMPTY(&s->io_q.pending));
318 s->io_q.blocked = (s->io_q.in_queue > 0);
320 if (s->io_q.in_flight) {
321 /* We can try to complete something just right away if there are
322 * still requests in-flight. */
323 qemu_laio_process_completions(s);
325 * Even we have completed everything (in_flight == 0), the queue can
326 * have still pended requests (in_queue > 0). We do not attempt to
327 * repeat submission to avoid IO hang. The reason is simple: s->e is
328 * still set and completion callback will be called shortly and all
329 * pended requests will be submitted from there.
334 void laio_io_plug(BlockDriverState *bs, LinuxAioState *s)
336 s->io_q.plugged++;
339 void laio_io_unplug(BlockDriverState *bs, LinuxAioState *s)
341 assert(s->io_q.plugged);
342 if (--s->io_q.plugged == 0 &&
343 !s->io_q.blocked && !QSIMPLEQ_EMPTY(&s->io_q.pending)) {
344 ioq_submit(s);
348 static int laio_do_submit(int fd, struct qemu_laiocb *laiocb, off_t offset,
349 int type)
351 LinuxAioState *s = laiocb->ctx;
352 struct iocb *iocbs = &laiocb->iocb;
353 QEMUIOVector *qiov = laiocb->qiov;
355 switch (type) {
356 case QEMU_AIO_WRITE:
357 io_prep_pwritev(iocbs, fd, qiov->iov, qiov->niov, offset);
358 break;
359 case QEMU_AIO_READ:
360 io_prep_preadv(iocbs, fd, qiov->iov, qiov->niov, offset);
361 break;
362 /* Currently Linux kernel does not support other operations */
363 default:
364 fprintf(stderr, "%s: invalid AIO request type 0x%x.\n",
365 __func__, type);
366 return -EIO;
368 io_set_eventfd(&laiocb->iocb, event_notifier_get_fd(&s->e));
370 QSIMPLEQ_INSERT_TAIL(&s->io_q.pending, laiocb, next);
371 s->io_q.in_queue++;
372 if (!s->io_q.blocked &&
373 (!s->io_q.plugged ||
374 s->io_q.in_flight + s->io_q.in_queue >= MAX_EVENTS)) {
375 ioq_submit(s);
378 return 0;
381 int coroutine_fn laio_co_submit(BlockDriverState *bs, LinuxAioState *s, int fd,
382 uint64_t offset, QEMUIOVector *qiov, int type)
384 int ret;
385 struct qemu_laiocb laiocb = {
386 .co = qemu_coroutine_self(),
387 .nbytes = qiov->size,
388 .ctx = s,
389 .ret = -EINPROGRESS,
390 .is_read = (type == QEMU_AIO_READ),
391 .qiov = qiov,
394 ret = laio_do_submit(fd, &laiocb, offset, type);
395 if (ret < 0) {
396 return ret;
399 if (laiocb.ret == -EINPROGRESS) {
400 qemu_coroutine_yield();
402 return laiocb.ret;
405 void laio_detach_aio_context(LinuxAioState *s, AioContext *old_context)
407 aio_set_event_notifier(old_context, &s->e, false, NULL, NULL);
408 qemu_bh_delete(s->completion_bh);
409 s->aio_context = NULL;
412 void laio_attach_aio_context(LinuxAioState *s, AioContext *new_context)
414 s->aio_context = new_context;
415 s->completion_bh = aio_bh_new(new_context, qemu_laio_completion_bh, s);
416 aio_set_event_notifier(new_context, &s->e, false,
417 qemu_laio_completion_cb,
418 qemu_laio_poll_cb);
421 LinuxAioState *laio_init(Error **errp)
423 int rc;
424 LinuxAioState *s;
426 s = g_malloc0(sizeof(*s));
427 rc = event_notifier_init(&s->e, false);
428 if (rc < 0) {
429 error_setg_errno(errp, -rc, "failed to to initialize event notifier");
430 goto out_free_state;
433 rc = io_setup(MAX_EVENTS, &s->ctx);
434 if (rc < 0) {
435 error_setg_errno(errp, -rc, "failed to create linux AIO context");
436 goto out_close_efd;
439 ioq_init(&s->io_q);
441 return s;
443 out_close_efd:
444 event_notifier_cleanup(&s->e);
445 out_free_state:
446 g_free(s);
447 return NULL;
450 void laio_cleanup(LinuxAioState *s)
452 event_notifier_cleanup(&s->e);
454 if (io_destroy(s->ctx) != 0) {
455 fprintf(stderr, "%s: destroy AIO context %p failed\n",
456 __func__, &s->ctx);
458 g_free(s);