1
2
3
4
5
6
7
8
9
10
11#include "qemu/osdep.h"
12#include <liburing.h>
13#include "block/aio.h"
14#include "qemu/queue.h"
15#include "block/block.h"
16#include "block/raw-aio.h"
17#include "qemu/coroutine.h"
18#include "qapi/error.h"
19#include "trace.h"
20
21
22#define MAX_ENTRIES 128
23
24typedef struct LuringAIOCB {
25 Coroutine *co;
26 struct io_uring_sqe sqeq;
27 ssize_t ret;
28 QEMUIOVector *qiov;
29 bool is_read;
30 QSIMPLEQ_ENTRY(LuringAIOCB) next;
31
32
33
34
35
36 int total_read;
37 QEMUIOVector resubmit_qiov;
38} LuringAIOCB;
39
40typedef struct LuringQueue {
41 int plugged;
42 unsigned int in_queue;
43 unsigned int in_flight;
44 bool blocked;
45 QSIMPLEQ_HEAD(, LuringAIOCB) submit_queue;
46} LuringQueue;
47
48typedef struct LuringState {
49 AioContext *aio_context;
50
51 struct io_uring ring;
52
53
54 LuringQueue io_q;
55
56
57 QEMUBH *completion_bh;
58} LuringState;
59
60
61
62
63
64
65
66static void luring_resubmit(LuringState *s, LuringAIOCB *luringcb)
67{
68 QSIMPLEQ_INSERT_TAIL(&s->io_q.submit_queue, luringcb, next);
69 s->io_q.in_queue++;
70}
71
72
73
74
75
76
77
78static void luring_resubmit_short_read(LuringState *s, LuringAIOCB *luringcb,
79 int nread)
80{
81 QEMUIOVector *resubmit_qiov;
82 size_t remaining;
83
84 trace_luring_resubmit_short_read(s, luringcb, nread);
85
86
87 luringcb->total_read += nread;
88 remaining = luringcb->qiov->size - luringcb->total_read;
89
90
91 resubmit_qiov = &luringcb->resubmit_qiov;
92 if (resubmit_qiov->iov == NULL) {
93 qemu_iovec_init(resubmit_qiov, luringcb->qiov->niov);
94 } else {
95 qemu_iovec_reset(resubmit_qiov);
96 }
97 qemu_iovec_concat(resubmit_qiov, luringcb->qiov, luringcb->total_read,
98 remaining);
99
100
101 luringcb->sqeq.off += nread;
102 luringcb->sqeq.addr = (__u64)(uintptr_t)luringcb->resubmit_qiov.iov;
103 luringcb->sqeq.len = luringcb->resubmit_qiov.niov;
104
105 luring_resubmit(s, luringcb);
106}
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121static void luring_process_completions(LuringState *s)
122{
123 struct io_uring_cqe *cqes;
124 int total_bytes;
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140 qemu_bh_schedule(s->completion_bh);
141
142 while (io_uring_peek_cqe(&s->ring, &cqes) == 0) {
143 LuringAIOCB *luringcb;
144 int ret;
145
146 if (!cqes) {
147 break;
148 }
149
150 luringcb = io_uring_cqe_get_data(cqes);
151 ret = cqes->res;
152 io_uring_cqe_seen(&s->ring, cqes);
153 cqes = NULL;
154
155
156 s->io_q.in_flight--;
157 trace_luring_process_completion(s, luringcb, ret);
158
159
160 total_bytes = ret + luringcb->total_read;
161
162 if (ret < 0) {
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177 if (ret == -EINTR || ret == -EAGAIN) {
178 luring_resubmit(s, luringcb);
179 continue;
180 }
181 } else if (!luringcb->qiov) {
182 goto end;
183 } else if (total_bytes == luringcb->qiov->size) {
184 ret = 0;
185
186 } else {
187
188 if (luringcb->is_read) {
189 if (ret > 0) {
190 luring_resubmit_short_read(s, luringcb, ret);
191 continue;
192 } else {
193
194 qemu_iovec_memset(luringcb->qiov, total_bytes, 0,
195 luringcb->qiov->size - total_bytes);
196 ret = 0;
197 }
198 } else {
199 ret = -ENOSPC;
200 }
201 }
202end:
203 luringcb->ret = ret;
204 qemu_iovec_destroy(&luringcb->resubmit_qiov);
205
206
207
208
209
210
211
212 if (!qemu_coroutine_entered(luringcb->co)) {
213 aio_co_wake(luringcb->co);
214 }
215 }
216 qemu_bh_cancel(s->completion_bh);
217}
218
219static int ioq_submit(LuringState *s)
220{
221 int ret = 0;
222 LuringAIOCB *luringcb, *luringcb_next;
223
224 while (s->io_q.in_queue > 0) {
225
226
227
228
229 QSIMPLEQ_FOREACH_SAFE(luringcb, &s->io_q.submit_queue, next,
230 luringcb_next) {
231 struct io_uring_sqe *sqes = io_uring_get_sqe(&s->ring);
232 if (!sqes) {
233 break;
234 }
235
236 *sqes = luringcb->sqeq;
237 QSIMPLEQ_REMOVE_HEAD(&s->io_q.submit_queue, next);
238 }
239 ret = io_uring_submit(&s->ring);
240 trace_luring_io_uring_submit(s, ret);
241
242 if (ret <= 0) {
243 if (ret == -EAGAIN || ret == -EINTR) {
244 continue;
245 }
246 break;
247 }
248 s->io_q.in_flight += ret;
249 s->io_q.in_queue -= ret;
250 }
251 s->io_q.blocked = (s->io_q.in_queue > 0);
252
253 if (s->io_q.in_flight) {
254
255
256
257
258 luring_process_completions(s);
259 }
260 return ret;
261}
262
263static void luring_process_completions_and_submit(LuringState *s)
264{
265 aio_context_acquire(s->aio_context);
266 luring_process_completions(s);
267
268 if (!s->io_q.plugged && s->io_q.in_queue > 0) {
269 ioq_submit(s);
270 }
271 aio_context_release(s->aio_context);
272}
273
274static void qemu_luring_completion_bh(void *opaque)
275{
276 LuringState *s = opaque;
277 luring_process_completions_and_submit(s);
278}
279
280static void qemu_luring_completion_cb(void *opaque)
281{
282 LuringState *s = opaque;
283 luring_process_completions_and_submit(s);
284}
285
286static bool qemu_luring_poll_cb(void *opaque)
287{
288 LuringState *s = opaque;
289
290 return io_uring_cq_ready(&s->ring);
291}
292
293static void qemu_luring_poll_ready(void *opaque)
294{
295 LuringState *s = opaque;
296
297 luring_process_completions_and_submit(s);
298}
299
300static void ioq_init(LuringQueue *io_q)
301{
302 QSIMPLEQ_INIT(&io_q->submit_queue);
303 io_q->plugged = 0;
304 io_q->in_queue = 0;
305 io_q->in_flight = 0;
306 io_q->blocked = false;
307}
308
309void luring_io_plug(BlockDriverState *bs, LuringState *s)
310{
311 trace_luring_io_plug(s);
312 s->io_q.plugged++;
313}
314
315void luring_io_unplug(BlockDriverState *bs, LuringState *s)
316{
317 assert(s->io_q.plugged);
318 trace_luring_io_unplug(s, s->io_q.blocked, s->io_q.plugged,
319 s->io_q.in_queue, s->io_q.in_flight);
320 if (--s->io_q.plugged == 0 &&
321 !s->io_q.blocked && s->io_q.in_queue > 0) {
322 ioq_submit(s);
323 }
324}
325
326
327
328
329
330
331
332
333
334
335
336
337static int luring_do_submit(int fd, LuringAIOCB *luringcb, LuringState *s,
338 uint64_t offset, int type)
339{
340 int ret;
341 struct io_uring_sqe *sqes = &luringcb->sqeq;
342
343 switch (type) {
344 case QEMU_AIO_WRITE:
345 io_uring_prep_writev(sqes, fd, luringcb->qiov->iov,
346 luringcb->qiov->niov, offset);
347 break;
348 case QEMU_AIO_READ:
349 io_uring_prep_readv(sqes, fd, luringcb->qiov->iov,
350 luringcb->qiov->niov, offset);
351 break;
352 case QEMU_AIO_FLUSH:
353 io_uring_prep_fsync(sqes, fd, IORING_FSYNC_DATASYNC);
354 break;
355 default:
356 fprintf(stderr, "%s: invalid AIO request type, aborting 0x%x.\n",
357 __func__, type);
358 abort();
359 }
360 io_uring_sqe_set_data(sqes, luringcb);
361
362 QSIMPLEQ_INSERT_TAIL(&s->io_q.submit_queue, luringcb, next);
363 s->io_q.in_queue++;
364 trace_luring_do_submit(s, s->io_q.blocked, s->io_q.plugged,
365 s->io_q.in_queue, s->io_q.in_flight);
366 if (!s->io_q.blocked &&
367 (!s->io_q.plugged ||
368 s->io_q.in_flight + s->io_q.in_queue >= MAX_ENTRIES)) {
369 ret = ioq_submit(s);
370 trace_luring_do_submit_done(s, ret);
371 return ret;
372 }
373 return 0;
374}
375
376int coroutine_fn luring_co_submit(BlockDriverState *bs, LuringState *s, int fd,
377 uint64_t offset, QEMUIOVector *qiov, int type)
378{
379 int ret;
380 LuringAIOCB luringcb = {
381 .co = qemu_coroutine_self(),
382 .ret = -EINPROGRESS,
383 .qiov = qiov,
384 .is_read = (type == QEMU_AIO_READ),
385 };
386 trace_luring_co_submit(bs, s, &luringcb, fd, offset, qiov ? qiov->size : 0,
387 type);
388 ret = luring_do_submit(fd, &luringcb, s, offset, type);
389
390 if (ret < 0) {
391 return ret;
392 }
393
394 if (luringcb.ret == -EINPROGRESS) {
395 qemu_coroutine_yield();
396 }
397 return luringcb.ret;
398}
399
400void luring_detach_aio_context(LuringState *s, AioContext *old_context)
401{
402 aio_set_fd_handler(old_context, s->ring.ring_fd, false,
403 NULL, NULL, NULL, NULL, s);
404 qemu_bh_delete(s->completion_bh);
405 s->aio_context = NULL;
406}
407
408void luring_attach_aio_context(LuringState *s, AioContext *new_context)
409{
410 s->aio_context = new_context;
411 s->completion_bh = aio_bh_new(new_context, qemu_luring_completion_bh, s);
412 aio_set_fd_handler(s->aio_context, s->ring.ring_fd, false,
413 qemu_luring_completion_cb, NULL,
414 qemu_luring_poll_cb, qemu_luring_poll_ready, s);
415}
416
417LuringState *luring_init(Error **errp)
418{
419 int rc;
420 LuringState *s = g_new0(LuringState, 1);
421 struct io_uring *ring = &s->ring;
422
423 trace_luring_init_state(s, sizeof(*s));
424
425 rc = io_uring_queue_init(MAX_ENTRIES, ring, 0);
426 if (rc < 0) {
427 error_setg_errno(errp, errno, "failed to init linux io_uring ring");
428 g_free(s);
429 return NULL;
430 }
431
432 ioq_init(&s->io_q);
433 return s;
434
435}
436
437void luring_cleanup(LuringState *s)
438{
439 io_uring_queue_exit(&s->ring);
440 trace_luring_cleanup_state(s);
441 g_free(s);
442}
443