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