Merge remote-tracking branch 'remotes/kevin/tags/for-upstream' into staging
[qemu.git] / block / linux-aio.c
1 /*
2 * Linux native AIO support.
3 *
4 * Copyright (C) 2009 IBM, Corp.
5 * Copyright (C) 2009 Red Hat, Inc.
6 *
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-common.h"
11 #include "block/aio.h"
12 #include "qemu/queue.h"
13 #include "block/raw-aio.h"
14 #include "qemu/event_notifier.h"
15
16 #include <libaio.h>
17
18 /*
19 * Queue size (per-device).
20 *
21 * XXX: eventually we need to communicate this to the guest and/or make it
22 * tunable by the guest. If we get more outstanding requests at a time
23 * than this we will get EAGAIN from io_submit which is communicated to
24 * the guest as an I/O error.
25 */
26 #define MAX_EVENTS 128
27
28 struct qemu_laiocb {
29 BlockDriverAIOCB common;
30 struct qemu_laio_state *ctx;
31 struct iocb iocb;
32 ssize_t ret;
33 size_t nbytes;
34 QEMUIOVector *qiov;
35 bool is_read;
36 QLIST_ENTRY(qemu_laiocb) node;
37 };
38
39 struct qemu_laio_state {
40 io_context_t ctx;
41 EventNotifier e;
42 };
43
44 static inline ssize_t io_event_ret(struct io_event *ev)
45 {
46 return (ssize_t)(((uint64_t)ev->res2 << 32) | ev->res);
47 }
48
49 /*
50 * Completes an AIO request (calls the callback and frees the ACB).
51 */
52 static void qemu_laio_process_completion(struct qemu_laio_state *s,
53 struct qemu_laiocb *laiocb)
54 {
55 int ret;
56
57 ret = laiocb->ret;
58 if (ret != -ECANCELED) {
59 if (ret == laiocb->nbytes) {
60 ret = 0;
61 } else if (ret >= 0) {
62 /* Short reads mean EOF, pad with zeros. */
63 if (laiocb->is_read) {
64 qemu_iovec_memset(laiocb->qiov, ret, 0,
65 laiocb->qiov->size - ret);
66 } else {
67 ret = -EINVAL;
68 }
69 }
70
71 laiocb->common.cb(laiocb->common.opaque, ret);
72 }
73
74 qemu_aio_release(laiocb);
75 }
76
77 static void qemu_laio_completion_cb(EventNotifier *e)
78 {
79 struct qemu_laio_state *s = container_of(e, struct qemu_laio_state, e);
80
81 while (event_notifier_test_and_clear(&s->e)) {
82 struct io_event events[MAX_EVENTS];
83 struct timespec ts = { 0 };
84 int nevents, i;
85
86 do {
87 nevents = io_getevents(s->ctx, MAX_EVENTS, MAX_EVENTS, events, &ts);
88 } while (nevents == -EINTR);
89
90 for (i = 0; i < nevents; i++) {
91 struct iocb *iocb = events[i].obj;
92 struct qemu_laiocb *laiocb =
93 container_of(iocb, struct qemu_laiocb, iocb);
94
95 laiocb->ret = io_event_ret(&events[i]);
96 qemu_laio_process_completion(s, laiocb);
97 }
98 }
99 }
100
101 static void laio_cancel(BlockDriverAIOCB *blockacb)
102 {
103 struct qemu_laiocb *laiocb = (struct qemu_laiocb *)blockacb;
104 struct io_event event;
105 int ret;
106
107 if (laiocb->ret != -EINPROGRESS)
108 return;
109
110 /*
111 * Note that as of Linux 2.6.31 neither the block device code nor any
112 * filesystem implements cancellation of AIO request.
113 * Thus the polling loop below is the normal code path.
114 */
115 ret = io_cancel(laiocb->ctx->ctx, &laiocb->iocb, &event);
116 if (ret == 0) {
117 laiocb->ret = -ECANCELED;
118 return;
119 }
120
121 /*
122 * We have to wait for the iocb to finish.
123 *
124 * The only way to get the iocb status update is by polling the io context.
125 * We might be able to do this slightly more optimal by removing the
126 * O_NONBLOCK flag.
127 */
128 while (laiocb->ret == -EINPROGRESS) {
129 qemu_laio_completion_cb(&laiocb->ctx->e);
130 }
131 }
132
133 static const AIOCBInfo laio_aiocb_info = {
134 .aiocb_size = sizeof(struct qemu_laiocb),
135 .cancel = laio_cancel,
136 };
137
138 BlockDriverAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
139 int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
140 BlockDriverCompletionFunc *cb, void *opaque, int type)
141 {
142 struct qemu_laio_state *s = aio_ctx;
143 struct qemu_laiocb *laiocb;
144 struct iocb *iocbs;
145 off_t offset = sector_num * 512;
146
147 laiocb = qemu_aio_get(&laio_aiocb_info, bs, cb, opaque);
148 laiocb->nbytes = nb_sectors * 512;
149 laiocb->ctx = s;
150 laiocb->ret = -EINPROGRESS;
151 laiocb->is_read = (type == QEMU_AIO_READ);
152 laiocb->qiov = qiov;
153
154 iocbs = &laiocb->iocb;
155
156 switch (type) {
157 case QEMU_AIO_WRITE:
158 io_prep_pwritev(iocbs, fd, qiov->iov, qiov->niov, offset);
159 break;
160 case QEMU_AIO_READ:
161 io_prep_preadv(iocbs, fd, qiov->iov, qiov->niov, offset);
162 break;
163 /* Currently Linux kernel does not support other operations */
164 default:
165 fprintf(stderr, "%s: invalid AIO request type 0x%x.\n",
166 __func__, type);
167 goto out_free_aiocb;
168 }
169 io_set_eventfd(&laiocb->iocb, event_notifier_get_fd(&s->e));
170
171 if (io_submit(s->ctx, 1, &iocbs) < 0)
172 goto out_free_aiocb;
173 return &laiocb->common;
174
175 out_free_aiocb:
176 qemu_aio_release(laiocb);
177 return NULL;
178 }
179
180 void *laio_init(void)
181 {
182 struct qemu_laio_state *s;
183
184 s = g_malloc0(sizeof(*s));
185 if (event_notifier_init(&s->e, false) < 0) {
186 goto out_free_state;
187 }
188
189 if (io_setup(MAX_EVENTS, &s->ctx) != 0) {
190 goto out_close_efd;
191 }
192
193 qemu_aio_set_event_notifier(&s->e, qemu_laio_completion_cb);
194
195 return s;
196
197 out_close_efd:
198 event_notifier_cleanup(&s->e);
199 out_free_state:
200 g_free(s);
201 return NULL;
202 }