9bcfec1726b05402bbc0875b2a53bf3edfc13e52
[fio.git] / engines / io_uring.c
1 /*
2  * io_uring engine
3  *
4  * IO engine using the new native Linux aio io_uring interface. See:
5  *
6  * http://git.kernel.dk/cgit/linux-block/log/?h=io_uring
7  *
8  */
9 #include <stdlib.h>
10 #include <unistd.h>
11 #include <errno.h>
12 #include <sys/time.h>
13 #include <sys/resource.h>
14
15 #include "../fio.h"
16 #include "../lib/pow2.h"
17 #include "../optgroup.h"
18 #include "../lib/memalign.h"
19 #include "../lib/fls.h"
20
21 #ifdef ARCH_HAVE_IOURING
22
23 #include "../lib/types.h"
24 #include "../os/linux/io_uring.h"
25
26 struct io_sq_ring {
27         unsigned *head;
28         unsigned *tail;
29         unsigned *ring_mask;
30         unsigned *ring_entries;
31         unsigned *flags;
32         unsigned *array;
33 };
34
35 struct io_cq_ring {
36         unsigned *head;
37         unsigned *tail;
38         unsigned *ring_mask;
39         unsigned *ring_entries;
40         struct io_uring_cqe *cqes;
41 };
42
43 struct ioring_mmap {
44         void *ptr;
45         size_t len;
46 };
47
48 struct ioring_data {
49         int ring_fd;
50
51         struct io_u **io_u_index;
52
53         struct io_sq_ring sq_ring;
54         struct io_uring_sqe *sqes;
55         struct iovec *iovecs;
56         unsigned sq_ring_mask;
57
58         struct io_cq_ring cq_ring;
59         unsigned cq_ring_mask;
60
61         int queued;
62         int cq_ring_off;
63         unsigned iodepth;
64
65         struct ioring_mmap mmap[3];
66 };
67
68 struct ioring_options {
69         void *pad;
70         unsigned int hipri;
71         unsigned int fixedbufs;
72         unsigned int sqpoll_thread;
73         unsigned int sqpoll_set;
74         unsigned int sqpoll_cpu;
75 };
76
77 static int fio_ioring_sqpoll_cb(void *data, unsigned long long *val)
78 {
79         struct ioring_options *o = data;
80
81         o->sqpoll_cpu = *val;
82         o->sqpoll_set = 1;
83         return 0;
84 }
85
86 static struct fio_option options[] = {
87         {
88                 .name   = "hipri",
89                 .lname  = "High Priority",
90                 .type   = FIO_OPT_STR_SET,
91                 .off1   = offsetof(struct ioring_options, hipri),
92                 .help   = "Use polled IO completions",
93                 .category = FIO_OPT_C_ENGINE,
94                 .group  = FIO_OPT_G_LIBAIO,
95         },
96         {
97                 .name   = "fixedbufs",
98                 .lname  = "Fixed (pre-mapped) IO buffers",
99                 .type   = FIO_OPT_STR_SET,
100                 .off1   = offsetof(struct ioring_options, fixedbufs),
101                 .help   = "Pre map IO buffers",
102                 .category = FIO_OPT_C_ENGINE,
103                 .group  = FIO_OPT_G_LIBAIO,
104         },
105         {
106                 .name   = "sqthread_poll",
107                 .lname  = "Kernel SQ thread polling",
108                 .type   = FIO_OPT_INT,
109                 .off1   = offsetof(struct ioring_options, sqpoll_thread),
110                 .help   = "Offload submission/completion to kernel thread",
111                 .category = FIO_OPT_C_ENGINE,
112                 .group  = FIO_OPT_G_LIBAIO,
113         },
114         {
115                 .name   = "sqthread_poll_cpu",
116                 .lname  = "SQ Thread Poll CPU",
117                 .type   = FIO_OPT_INT,
118                 .cb     = fio_ioring_sqpoll_cb,
119                 .help   = "What CPU to run SQ thread polling on",
120                 .category = FIO_OPT_C_ENGINE,
121                 .group  = FIO_OPT_G_LIBAIO,
122         },
123         {
124                 .name   = NULL,
125         },
126 };
127
128 static int io_uring_enter(struct ioring_data *ld, unsigned int to_submit,
129                          unsigned int min_complete, unsigned int flags)
130 {
131         return syscall(__NR_sys_io_uring_enter, ld->ring_fd, to_submit,
132                         min_complete, flags, NULL, 0);
133 }
134
135 static int fio_ioring_prep(struct thread_data *td, struct io_u *io_u)
136 {
137         struct ioring_data *ld = td->io_ops_data;
138         struct ioring_options *o = td->eo;
139         struct fio_file *f = io_u->file;
140         struct io_uring_sqe *sqe;
141
142         sqe = &ld->sqes[io_u->index];
143         sqe->fd = f->fd;
144         sqe->flags = 0;
145         sqe->ioprio = 0;
146         sqe->buf_index = 0;
147
148         if (io_u->ddir == DDIR_READ || io_u->ddir == DDIR_WRITE) {
149                 if (o->fixedbufs) {
150                         if (io_u->ddir == DDIR_READ)
151                                 sqe->opcode = IORING_OP_READ_FIXED;
152                         else
153                                 sqe->opcode = IORING_OP_WRITE_FIXED;
154                         sqe->addr = (unsigned long) io_u->xfer_buf;
155                         sqe->len = io_u->xfer_buflen;
156                         sqe->buf_index = io_u->index;
157                 } else {
158                         if (io_u->ddir == DDIR_READ)
159                                 sqe->opcode = IORING_OP_READV;
160                         else
161                                 sqe->opcode = IORING_OP_WRITEV;
162                         sqe->addr = (unsigned long) &ld->iovecs[io_u->index];
163                         sqe->len = 1;
164                 }
165                 sqe->off = io_u->offset;
166         } else if (ddir_sync(io_u->ddir)) {
167                 sqe->fsync_flags = 0;
168                 if (io_u->ddir == DDIR_DATASYNC)
169                         sqe->fsync_flags |= IORING_FSYNC_DATASYNC;
170                 sqe->opcode = IORING_OP_FSYNC;
171         }
172
173         sqe->user_data = (unsigned long) io_u;
174         return 0;
175 }
176
177 static struct io_u *fio_ioring_event(struct thread_data *td, int event)
178 {
179         struct ioring_data *ld = td->io_ops_data;
180         struct io_uring_cqe *cqe;
181         struct io_u *io_u;
182         unsigned index;
183
184         index = (event + ld->cq_ring_off) & ld->cq_ring_mask;
185
186         cqe = &ld->cq_ring.cqes[index];
187         io_u = (struct io_u *) (uintptr_t) cqe->user_data;
188
189         if (cqe->res != io_u->xfer_buflen) {
190                 if (cqe->res > io_u->xfer_buflen)
191                         io_u->error = -cqe->res;
192                 else
193                         io_u->resid = io_u->xfer_buflen - cqe->res;
194         } else
195                 io_u->error = 0;
196
197         return io_u;
198 }
199
200 static int fio_ioring_cqring_reap(struct thread_data *td, unsigned int events,
201                                    unsigned int max)
202 {
203         struct ioring_data *ld = td->io_ops_data;
204         struct io_cq_ring *ring = &ld->cq_ring;
205         unsigned head, reaped = 0;
206
207         head = *ring->head;
208         do {
209                 read_barrier();
210                 if (head == *ring->tail)
211                         break;
212                 reaped++;
213                 head++;
214         } while (reaped + events < max);
215
216         *ring->head = head;
217         write_barrier();
218         return reaped;
219 }
220
221 static int fio_ioring_getevents(struct thread_data *td, unsigned int min,
222                                 unsigned int max, const struct timespec *t)
223 {
224         struct ioring_data *ld = td->io_ops_data;
225         unsigned actual_min = td->o.iodepth_batch_complete_min == 0 ? 0 : min;
226         struct ioring_options *o = td->eo;
227         struct io_cq_ring *ring = &ld->cq_ring;
228         unsigned events = 0;
229         int r;
230
231         ld->cq_ring_off = *ring->head;
232         do {
233                 r = fio_ioring_cqring_reap(td, events, max);
234                 if (r) {
235                         events += r;
236                         if (actual_min != 0)
237                                 actual_min -= r;
238                         continue;
239                 }
240
241                 if (!o->sqpoll_thread) {
242                         r = io_uring_enter(ld, 0, actual_min,
243                                                 IORING_ENTER_GETEVENTS);
244                         if (r < 0) {
245                                 if (errno == EAGAIN)
246                                         continue;
247                                 td_verror(td, errno, "io_uring_enter");
248                                 break;
249                         }
250                 }
251         } while (events < min);
252
253         return r < 0 ? r : events;
254 }
255
256 static enum fio_q_status fio_ioring_queue(struct thread_data *td,
257                                           struct io_u *io_u)
258 {
259         struct ioring_data *ld = td->io_ops_data;
260         struct io_sq_ring *ring = &ld->sq_ring;
261         unsigned tail, next_tail;
262
263         fio_ro_check(td, io_u);
264
265         if (ld->queued == ld->iodepth)
266                 return FIO_Q_BUSY;
267
268         if (io_u->ddir == DDIR_TRIM) {
269                 if (ld->queued)
270                         return FIO_Q_BUSY;
271
272                 do_io_u_trim(td, io_u);
273                 io_u_mark_submit(td, 1);
274                 io_u_mark_complete(td, 1);
275                 return FIO_Q_COMPLETED;
276         }
277
278         tail = *ring->tail;
279         next_tail = tail + 1;
280         read_barrier();
281         if (next_tail == *ring->head)
282                 return FIO_Q_BUSY;
283
284         /* ensure sqe stores are ordered with tail update */
285         write_barrier();
286         ring->array[tail & ld->sq_ring_mask] = io_u->index;
287         *ring->tail = next_tail;
288         write_barrier();
289
290         ld->queued++;
291         return FIO_Q_QUEUED;
292 }
293
294 static void fio_ioring_queued(struct thread_data *td, int start, int nr)
295 {
296         struct ioring_data *ld = td->io_ops_data;
297         struct timespec now;
298
299         if (!fio_fill_issue_time(td))
300                 return;
301
302         fio_gettime(&now, NULL);
303
304         while (nr--) {
305                 struct io_sq_ring *ring = &ld->sq_ring;
306                 int index = ring->array[start & ld->sq_ring_mask];
307                 struct io_u *io_u = ld->io_u_index[index];
308
309                 memcpy(&io_u->issue_time, &now, sizeof(now));
310                 io_u_queued(td, io_u);
311
312                 start++;
313         }
314 }
315
316 static int fio_ioring_commit(struct thread_data *td)
317 {
318         struct ioring_data *ld = td->io_ops_data;
319         struct ioring_options *o = td->eo;
320         int ret;
321
322         if (!ld->queued)
323                 return 0;
324
325         /*
326          * Kernel side does submission. just need to check if the ring is
327          * flagged as needing a kick, if so, call io_uring_enter(). This
328          * only happens if we've been idle too long.
329          */
330         if (o->sqpoll_thread) {
331                 struct io_sq_ring *ring = &ld->sq_ring;
332
333                 read_barrier();
334                 if (*ring->flags & IORING_SQ_NEED_WAKEUP)
335                         io_uring_enter(ld, ld->queued, 0,
336                                         IORING_ENTER_SQ_WAKEUP);
337                 ld->queued = 0;
338                 return 0;
339         }
340
341         do {
342                 unsigned start = *ld->sq_ring.head;
343                 long nr = ld->queued;
344
345                 ret = io_uring_enter(ld, nr, 0, IORING_ENTER_GETEVENTS);
346                 if (ret > 0) {
347                         fio_ioring_queued(td, start, ret);
348                         io_u_mark_submit(td, ret);
349
350                         ld->queued -= ret;
351                         ret = 0;
352                 } else if (!ret) {
353                         io_u_mark_submit(td, ret);
354                         continue;
355                 } else {
356                         if (errno == EAGAIN) {
357                                 ret = fio_ioring_cqring_reap(td, 0, ld->queued);
358                                 if (ret)
359                                         continue;
360                                 /* Shouldn't happen */
361                                 usleep(1);
362                                 continue;
363                         }
364                         td_verror(td, errno, "io_uring_enter submit");
365                         break;
366                 }
367         } while (ld->queued);
368
369         return ret;
370 }
371
372 static void fio_ioring_unmap(struct ioring_data *ld)
373 {
374         int i;
375
376         for (i = 0; i < ARRAY_SIZE(ld->mmap); i++)
377                 munmap(ld->mmap[i].ptr, ld->mmap[i].len);
378         close(ld->ring_fd);
379 }
380
381 static void fio_ioring_cleanup(struct thread_data *td)
382 {
383         struct ioring_data *ld = td->io_ops_data;
384
385         if (ld) {
386                 if (!(td->flags & TD_F_CHILD))
387                         fio_ioring_unmap(ld);
388
389                 free(ld->io_u_index);
390                 free(ld->iovecs);
391                 free(ld);
392         }
393 }
394
395 static int fio_ioring_mmap(struct ioring_data *ld, struct io_uring_params *p)
396 {
397         struct io_sq_ring *sring = &ld->sq_ring;
398         struct io_cq_ring *cring = &ld->cq_ring;
399         void *ptr;
400
401         ld->mmap[0].len = p->sq_off.array + p->sq_entries * sizeof(__u32);
402         ptr = mmap(0, ld->mmap[0].len, PROT_READ | PROT_WRITE,
403                         MAP_SHARED | MAP_POPULATE, ld->ring_fd,
404                         IORING_OFF_SQ_RING);
405         ld->mmap[0].ptr = ptr;
406         sring->head = ptr + p->sq_off.head;
407         sring->tail = ptr + p->sq_off.tail;
408         sring->ring_mask = ptr + p->sq_off.ring_mask;
409         sring->ring_entries = ptr + p->sq_off.ring_entries;
410         sring->flags = ptr + p->sq_off.flags;
411         sring->array = ptr + p->sq_off.array;
412         ld->sq_ring_mask = *sring->ring_mask;
413
414         ld->mmap[1].len = p->sq_entries * sizeof(struct io_uring_sqe);
415         ld->sqes = mmap(0, ld->mmap[1].len, PROT_READ | PROT_WRITE,
416                                 MAP_SHARED | MAP_POPULATE, ld->ring_fd,
417                                 IORING_OFF_SQES);
418         ld->mmap[1].ptr = ld->sqes;
419
420         ld->mmap[2].len = p->cq_off.cqes +
421                                 p->cq_entries * sizeof(struct io_uring_cqe);
422         ptr = mmap(0, ld->mmap[2].len, PROT_READ | PROT_WRITE,
423                         MAP_SHARED | MAP_POPULATE, ld->ring_fd,
424                         IORING_OFF_CQ_RING);
425         ld->mmap[2].ptr = ptr;
426         cring->head = ptr + p->cq_off.head;
427         cring->tail = ptr + p->cq_off.tail;
428         cring->ring_mask = ptr + p->cq_off.ring_mask;
429         cring->ring_entries = ptr + p->cq_off.ring_entries;
430         cring->cqes = ptr + p->cq_off.cqes;
431         ld->cq_ring_mask = *cring->ring_mask;
432         return 0;
433 }
434
435 static int fio_ioring_queue_init(struct thread_data *td)
436 {
437         struct ioring_data *ld = td->io_ops_data;
438         struct ioring_options *o = td->eo;
439         int depth = td->o.iodepth;
440         struct io_uring_params p;
441         int ret;
442
443         memset(&p, 0, sizeof(p));
444
445         if (o->hipri)
446                 p.flags |= IORING_SETUP_IOPOLL;
447         if (o->sqpoll_thread) {
448                 p.flags |= IORING_SETUP_SQPOLL;
449                 if (o->sqpoll_set) {
450                         p.flags |= IORING_SETUP_SQ_AFF;
451                         p.sq_thread_cpu = o->sqpoll_cpu;
452                 }
453         }
454
455         ret = syscall(__NR_sys_io_uring_setup, depth, &p);
456         if (ret < 0)
457                 return ret;
458
459         ld->ring_fd = ret;
460
461         if (o->fixedbufs) {
462                 struct rlimit rlim = {
463                         .rlim_cur = RLIM_INFINITY,
464                         .rlim_max = RLIM_INFINITY,
465                 };
466
467                 if (setrlimit(RLIMIT_MEMLOCK, &rlim) < 0)
468                         return -1;
469
470                 ret = syscall(__NR_sys_io_uring_register, ld->ring_fd,
471                                 IORING_REGISTER_BUFFERS, ld->iovecs, depth);
472                 if (ret < 0)
473                         return ret;
474         }
475
476         return fio_ioring_mmap(ld, &p);
477 }
478
479 static int fio_ioring_post_init(struct thread_data *td)
480 {
481         struct ioring_data *ld = td->io_ops_data;
482         struct io_u *io_u;
483         int err, i;
484
485         for (i = 0; i < td->o.iodepth; i++) {
486                 struct iovec *iov = &ld->iovecs[i];
487
488                 io_u = ld->io_u_index[i];
489                 iov->iov_base = io_u->buf;
490                 iov->iov_len = td_max_bs(td);
491         }
492
493         err = fio_ioring_queue_init(td);
494         if (err) {
495                 td_verror(td, errno, "io_queue_init");
496                 return 1;
497         }
498
499         return 0;
500 }
501
502 static unsigned roundup_pow2(unsigned depth)
503 {
504         return 1UL << __fls(depth - 1);
505 }
506
507 static int fio_ioring_init(struct thread_data *td)
508 {
509         struct ioring_data *ld;
510
511         ld = calloc(1, sizeof(*ld));
512
513         /* ring depth must be a power-of-2 */
514         ld->iodepth = td->o.iodepth;
515         td->o.iodepth = roundup_pow2(td->o.iodepth);
516
517         /* io_u index */
518         ld->io_u_index = calloc(td->o.iodepth, sizeof(struct io_u *));
519         ld->iovecs = calloc(td->o.iodepth, sizeof(struct iovec));
520
521         td->io_ops_data = ld;
522         return 0;
523 }
524
525 static int fio_ioring_io_u_init(struct thread_data *td, struct io_u *io_u)
526 {
527         struct ioring_data *ld = td->io_ops_data;
528
529         ld->io_u_index[io_u->index] = io_u;
530         return 0;
531 }
532
533 static struct ioengine_ops ioengine = {
534         .name                   = "io_uring",
535         .version                = FIO_IOOPS_VERSION,
536         .flags                  = FIO_ASYNCIO_SYNC_TRIM,
537         .init                   = fio_ioring_init,
538         .post_init              = fio_ioring_post_init,
539         .io_u_init              = fio_ioring_io_u_init,
540         .prep                   = fio_ioring_prep,
541         .queue                  = fio_ioring_queue,
542         .commit                 = fio_ioring_commit,
543         .getevents              = fio_ioring_getevents,
544         .event                  = fio_ioring_event,
545         .cleanup                = fio_ioring_cleanup,
546         .open_file              = generic_open_file,
547         .close_file             = generic_close_file,
548         .get_file_size          = generic_get_file_size,
549         .options                = options,
550         .option_struct_size     = sizeof(struct ioring_options),
551 };
552
553 static void fio_init fio_ioring_register(void)
554 {
555         register_ioengine(&ioengine);
556 }
557
558 static void fio_exit fio_ioring_unregister(void)
559 {
560         unregister_ioengine(&ioengine);
561 }
562 #endif