1
2
3
4
5
6
7
8
9
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"
18
19#include <libaio.h>
20
21
22
23
24
25
26
27
28
29#define MAX_EVENTS 128
30
31struct qemu_laiocb {
32 BlockAIOCB common;
33 Coroutine *co;
34 LinuxAioState *ctx;
35 struct iocb iocb;
36 ssize_t ret;
37 size_t nbytes;
38 QEMUIOVector *qiov;
39 bool is_read;
40 QSIMPLEQ_ENTRY(qemu_laiocb) next;
41};
42
43typedef struct {
44 int plugged;
45 unsigned int in_queue;
46 unsigned int in_flight;
47 bool blocked;
48 QSIMPLEQ_HEAD(, qemu_laiocb) pending;
49} LaioQueue;
50
51struct LinuxAioState {
52 AioContext *aio_context;
53
54 io_context_t ctx;
55 EventNotifier e;
56
57
58 LaioQueue io_q;
59
60
61 QEMUBH *completion_bh;
62 struct io_event events[MAX_EVENTS];
63 int event_idx;
64 int event_max;
65};
66
67static void ioq_submit(LinuxAioState *s);
68
69static inline ssize_t io_event_ret(struct io_event *ev)
70{
71 return (ssize_t)(((uint64_t)ev->res2 << 32) | ev->res);
72}
73
74
75
76
77static void qemu_laio_process_completion(struct qemu_laiocb *laiocb)
78{
79 int ret;
80
81 ret = laiocb->ret;
82 if (ret != -ECANCELED) {
83 if (ret == laiocb->nbytes) {
84 ret = 0;
85 } else if (ret >= 0) {
86
87 if (laiocb->is_read) {
88 qemu_iovec_memset(laiocb->qiov, ret, 0,
89 laiocb->qiov->size - ret);
90 } else {
91 ret = -ENOSPC;
92 }
93 }
94 }
95
96 laiocb->ret = ret;
97 if (laiocb->co) {
98 qemu_coroutine_enter(laiocb->co);
99 } else {
100 laiocb->common.cb(laiocb->common.opaque, ret);
101 qemu_aio_unref(laiocb);
102 }
103}
104
105
106
107
108
109
110
111
112
113
114
115
116static void qemu_laio_completion_bh(void *opaque)
117{
118 LinuxAioState *s = opaque;
119
120
121 if (s->event_idx == s->event_max) {
122 do {
123 struct timespec ts = { 0 };
124 s->event_max = io_getevents(s->ctx, MAX_EVENTS, MAX_EVENTS,
125 s->events, &ts);
126 } while (s->event_max == -EINTR);
127
128 s->event_idx = 0;
129 if (s->event_max <= 0) {
130 s->event_max = 0;
131 return;
132 }
133 s->io_q.in_flight -= s->event_max;
134 }
135
136
137 qemu_bh_schedule(s->completion_bh);
138
139
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);
144
145 laiocb->ret = io_event_ret(&s->events[s->event_idx]);
146 s->event_idx++;
147
148 qemu_laio_process_completion(laiocb);
149 }
150
151 if (!s->io_q.plugged && !QSIMPLEQ_EMPTY(&s->io_q.pending)) {
152 ioq_submit(s);
153 }
154
155 qemu_bh_cancel(s->completion_bh);
156}
157
158static void qemu_laio_completion_cb(EventNotifier *e)
159{
160 LinuxAioState *s = container_of(e, LinuxAioState, e);
161
162 if (event_notifier_test_and_clear(&s->e)) {
163 qemu_laio_completion_bh(s);
164 }
165}
166
167static void laio_cancel(BlockAIOCB *blockacb)
168{
169 struct qemu_laiocb *laiocb = (struct qemu_laiocb *)blockacb;
170 struct io_event event;
171 int ret;
172
173 if (laiocb->ret != -EINPROGRESS) {
174 return;
175 }
176 ret = io_cancel(laiocb->ctx->ctx, &laiocb->iocb, &event);
177 laiocb->ret = -ECANCELED;
178 if (ret != 0) {
179
180 return;
181 }
182
183 laiocb->common.cb(laiocb->common.opaque, laiocb->ret);
184}
185
186static const AIOCBInfo laio_aiocb_info = {
187 .aiocb_size = sizeof(struct qemu_laiocb),
188 .cancel_async = laio_cancel,
189};
190
191static void ioq_init(LaioQueue *io_q)
192{
193 QSIMPLEQ_INIT(&io_q->pending);
194 io_q->plugged = 0;
195 io_q->in_queue = 0;
196 io_q->in_flight = 0;
197 io_q->blocked = false;
198}
199
200static void ioq_submit(LinuxAioState *s)
201{
202 int ret, len;
203 struct qemu_laiocb *aiocb;
204 struct iocb *iocbs[MAX_EVENTS];
205 QSIMPLEQ_HEAD(, qemu_laiocb) completed;
206
207 do {
208 if (s->io_q.in_flight >= MAX_EVENTS) {
209 break;
210 }
211 len = 0;
212 QSIMPLEQ_FOREACH(aiocb, &s->io_q.pending, next) {
213 iocbs[len++] = &aiocb->iocb;
214 if (s->io_q.in_flight + len >= MAX_EVENTS) {
215 break;
216 }
217 }
218
219 ret = io_submit(s->ctx, len, iocbs);
220 if (ret == -EAGAIN) {
221 break;
222 }
223 if (ret < 0) {
224
225 aiocb = QSIMPLEQ_FIRST(&s->io_q.pending);
226 QSIMPLEQ_REMOVE_HEAD(&s->io_q.pending, next);
227 s->io_q.in_queue--;
228 aiocb->ret = ret;
229 qemu_laio_process_completion(aiocb);
230 continue;
231 }
232
233 s->io_q.in_flight += ret;
234 s->io_q.in_queue -= ret;
235 aiocb = container_of(iocbs[ret - 1], struct qemu_laiocb, iocb);
236 QSIMPLEQ_SPLIT_AFTER(&s->io_q.pending, aiocb, next, &completed);
237 } while (ret == len && !QSIMPLEQ_EMPTY(&s->io_q.pending));
238 s->io_q.blocked = (s->io_q.in_queue > 0);
239}
240
241void laio_io_plug(BlockDriverState *bs, LinuxAioState *s)
242{
243 s->io_q.plugged++;
244}
245
246void laio_io_unplug(BlockDriverState *bs, LinuxAioState *s)
247{
248 assert(s->io_q.plugged);
249 if (--s->io_q.plugged == 0 &&
250 !s->io_q.blocked && !QSIMPLEQ_EMPTY(&s->io_q.pending)) {
251 ioq_submit(s);
252 }
253}
254
255static int laio_do_submit(int fd, struct qemu_laiocb *laiocb, off_t offset,
256 int type)
257{
258 LinuxAioState *s = laiocb->ctx;
259 struct iocb *iocbs = &laiocb->iocb;
260 QEMUIOVector *qiov = laiocb->qiov;
261
262 switch (type) {
263 case QEMU_AIO_WRITE:
264 io_prep_pwritev(iocbs, fd, qiov->iov, qiov->niov, offset);
265 break;
266 case QEMU_AIO_READ:
267 io_prep_preadv(iocbs, fd, qiov->iov, qiov->niov, offset);
268 break;
269
270 default:
271 fprintf(stderr, "%s: invalid AIO request type 0x%x.\n",
272 __func__, type);
273 return -EIO;
274 }
275 io_set_eventfd(&laiocb->iocb, event_notifier_get_fd(&s->e));
276
277 QSIMPLEQ_INSERT_TAIL(&s->io_q.pending, laiocb, next);
278 s->io_q.in_queue++;
279 if (!s->io_q.blocked &&
280 (!s->io_q.plugged ||
281 s->io_q.in_flight + s->io_q.in_queue >= MAX_EVENTS)) {
282 ioq_submit(s);
283 }
284
285 return 0;
286}
287
288int coroutine_fn laio_co_submit(BlockDriverState *bs, LinuxAioState *s, int fd,
289 uint64_t offset, QEMUIOVector *qiov, int type)
290{
291 int ret;
292 struct qemu_laiocb laiocb = {
293 .co = qemu_coroutine_self(),
294 .nbytes = qiov->size,
295 .ctx = s,
296 .is_read = (type == QEMU_AIO_READ),
297 .qiov = qiov,
298 };
299
300 ret = laio_do_submit(fd, &laiocb, offset, type);
301 if (ret < 0) {
302 return ret;
303 }
304
305 qemu_coroutine_yield();
306 return laiocb.ret;
307}
308
309BlockAIOCB *laio_submit(BlockDriverState *bs, LinuxAioState *s, int fd,
310 int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
311 BlockCompletionFunc *cb, void *opaque, int type)
312{
313 struct qemu_laiocb *laiocb;
314 off_t offset = sector_num * BDRV_SECTOR_SIZE;
315 int ret;
316
317 laiocb = qemu_aio_get(&laio_aiocb_info, bs, cb, opaque);
318 laiocb->nbytes = nb_sectors * BDRV_SECTOR_SIZE;
319 laiocb->ctx = s;
320 laiocb->ret = -EINPROGRESS;
321 laiocb->is_read = (type == QEMU_AIO_READ);
322 laiocb->qiov = qiov;
323
324 ret = laio_do_submit(fd, laiocb, offset, type);
325 if (ret < 0) {
326 qemu_aio_unref(laiocb);
327 return NULL;
328 }
329
330 return &laiocb->common;
331}
332
333void laio_detach_aio_context(LinuxAioState *s, AioContext *old_context)
334{
335 aio_set_event_notifier(old_context, &s->e, false, NULL);
336 qemu_bh_delete(s->completion_bh);
337}
338
339void laio_attach_aio_context(LinuxAioState *s, AioContext *new_context)
340{
341 s->aio_context = new_context;
342 s->completion_bh = aio_bh_new(new_context, qemu_laio_completion_bh, s);
343 aio_set_event_notifier(new_context, &s->e, false,
344 qemu_laio_completion_cb);
345}
346
347LinuxAioState *laio_init(void)
348{
349 LinuxAioState *s;
350
351 s = g_malloc0(sizeof(*s));
352 if (event_notifier_init(&s->e, false) < 0) {
353 goto out_free_state;
354 }
355
356 if (io_setup(MAX_EVENTS, &s->ctx) != 0) {
357 goto out_close_efd;
358 }
359
360 ioq_init(&s->io_q);
361
362 return s;
363
364out_close_efd:
365 event_notifier_cleanup(&s->e);
366out_free_state:
367 g_free(s);
368 return NULL;
369}
370
371void laio_cleanup(LinuxAioState *s)
372{
373 event_notifier_cleanup(&s->e);
374
375 if (io_destroy(s->ctx) != 0) {
376 fprintf(stderr, "%s: destroy AIO context %p failed\n",
377 __func__, &s->ctx);
378 }
379 g_free(s);
380}
381