5b3509a98d734bd3fe50157b81a3aa1a358b9547
[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                         continue;
237                 }
238
239                 if (!o->sqpoll_thread) {
240                         r = io_uring_enter(ld, 0, actual_min,
241                                                 IORING_ENTER_GETEVENTS);
242                         if (r < 0) {
243                                 if (errno == EAGAIN)
244                                         continue;
245                                 td_verror(td, errno, "io_uring_enter");
246                                 break;
247                         }
248                 }
249         } while (events < min);
250
251         return r < 0 ? r : events;
252 }
253
254 static enum fio_q_status fio_ioring_queue(struct thread_data *td,
255                                           struct io_u *io_u)
256 {
257         struct ioring_data *ld = td->io_ops_data;
258         struct io_sq_ring *ring = &ld->sq_ring;
259         unsigned tail, next_tail;
260
261         fio_ro_check(td, io_u);
262
263         if (ld->queued == ld->iodepth)
264                 return FIO_Q_BUSY;
265
266         if (io_u->ddir == DDIR_TRIM) {
267                 if (ld->queued)
268                         return FIO_Q_BUSY;
269
270                 do_io_u_trim(td, io_u);
271                 io_u_mark_submit(td, 1);
272                 io_u_mark_complete(td, 1);
273                 return FIO_Q_COMPLETED;
274         }
275
276         tail = *ring->tail;
277         next_tail = tail + 1;
278         read_barrier();
279         if (next_tail == *ring->head)
280                 return FIO_Q_BUSY;
281
282         /* ensure sqe stores are ordered with tail update */
283         write_barrier();
284         ring->array[tail & ld->sq_ring_mask] = io_u->index;
285         *ring->tail = next_tail;
286         write_barrier();
287
288         ld->queued++;
289         return FIO_Q_QUEUED;
290 }
291
292 static void fio_ioring_queued(struct thread_data *td, int start, int nr)
293 {
294         struct ioring_data *ld = td->io_ops_data;
295         struct timespec now;
296
297         if (!fio_fill_issue_time(td))
298                 return;
299
300         fio_gettime(&now, NULL);
301
302         while (nr--) {
303                 struct io_sq_ring *ring = &ld->sq_ring;
304                 int index = ring->array[start & ld->sq_ring_mask];
305                 struct io_u *io_u = ld->io_u_index[index];
306
307                 memcpy(&io_u->issue_time, &now, sizeof(now));
308                 io_u_queued(td, io_u);
309
310                 start++;
311         }
312 }
313
314 static int fio_ioring_commit(struct thread_data *td)
315 {
316         struct ioring_data *ld = td->io_ops_data;
317         struct ioring_options *o = td->eo;
318         int ret;
319
320         if (!ld->queued)
321                 return 0;
322
323         /*
324          * Kernel side does submission. just need to check if the ring is
325          * flagged as needing a kick, if so, call io_uring_enter(). This
326          * only happens if we've been idle too long.
327          */
328         if (o->sqpoll_thread) {
329                 struct io_sq_ring *ring = &ld->sq_ring;
330
331                 read_barrier();
332                 if (*ring->flags & IORING_SQ_NEED_WAKEUP)
333                         io_uring_enter(ld, ld->queued, 0,
334                                         IORING_ENTER_SQ_WAKEUP);
335                 ld->queued = 0;
336                 return 0;
337         }
338
339         do {
340                 unsigned start = *ld->sq_ring.head;
341                 long nr = ld->queued;
342
343                 ret = io_uring_enter(ld, nr, 0, IORING_ENTER_GETEVENTS);
344                 if (ret > 0) {
345                         fio_ioring_queued(td, start, ret);
346                         io_u_mark_submit(td, ret);
347
348                         ld->queued -= ret;
349                         ret = 0;
350                 } else if (!ret) {
351                         io_u_mark_submit(td, ret);
352                         continue;
353                 } else {
354                         if (errno == EAGAIN) {
355                                 ret = fio_ioring_cqring_reap(td, 0, ld->queued);
356                                 if (ret)
357                                         continue;
358                                 /* Shouldn't happen */
359                                 usleep(1);
360                                 continue;
361                         }
362                         td_verror(td, errno, "io_uring_enter submit");
363                         break;
364                 }
365         } while (ld->queued);
366
367         return ret;
368 }
369
370 static void fio_ioring_unmap(struct ioring_data *ld)
371 {
372         int i;
373
374         for (i = 0; i < ARRAY_SIZE(ld->mmap); i++)
375                 munmap(ld->mmap[i].ptr, ld->mmap[i].len);
376         close(ld->ring_fd);
377 }
378
379 static void fio_ioring_cleanup(struct thread_data *td)
380 {
381         struct ioring_data *ld = td->io_ops_data;
382
383         if (ld) {
384                 if (!(td->flags & TD_F_CHILD))
385                         fio_ioring_unmap(ld);
386
387                 free(ld->io_u_index);
388                 free(ld->iovecs);
389                 free(ld);
390         }
391 }
392
393 static int fio_ioring_mmap(struct ioring_data *ld, struct io_uring_params *p)
394 {
395         struct io_sq_ring *sring = &ld->sq_ring;
396         struct io_cq_ring *cring = &ld->cq_ring;
397         void *ptr;
398
399         ld->mmap[0].len = p->sq_off.array + p->sq_entries * sizeof(__u32);
400         ptr = mmap(0, ld->mmap[0].len, PROT_READ | PROT_WRITE,
401                         MAP_SHARED | MAP_POPULATE, ld->ring_fd,
402                         IORING_OFF_SQ_RING);
403         ld->mmap[0].ptr = ptr;
404         sring->head = ptr + p->sq_off.head;
405         sring->tail = ptr + p->sq_off.tail;
406         sring->ring_mask = ptr + p->sq_off.ring_mask;
407         sring->ring_entries = ptr + p->sq_off.ring_entries;
408         sring->flags = ptr + p->sq_off.flags;
409         sring->array = ptr + p->sq_off.array;
410         ld->sq_ring_mask = *sring->ring_mask;
411
412         ld->mmap[1].len = p->sq_entries * sizeof(struct io_uring_sqe);
413         ld->sqes = mmap(0, ld->mmap[1].len, PROT_READ | PROT_WRITE,
414                                 MAP_SHARED | MAP_POPULATE, ld->ring_fd,
415                                 IORING_OFF_SQES);
416         ld->mmap[1].ptr = ld->sqes;
417
418         ld->mmap[2].len = p->cq_off.cqes +
419                                 p->cq_entries * sizeof(struct io_uring_cqe);
420         ptr = mmap(0, ld->mmap[2].len, PROT_READ | PROT_WRITE,
421                         MAP_SHARED | MAP_POPULATE, ld->ring_fd,
422                         IORING_OFF_CQ_RING);
423         ld->mmap[2].ptr = ptr;
424         cring->head = ptr + p->cq_off.head;
425         cring->tail = ptr + p->cq_off.tail;
426         cring->ring_mask = ptr + p->cq_off.ring_mask;
427         cring->ring_entries = ptr + p->cq_off.ring_entries;
428         cring->cqes = ptr + p->cq_off.cqes;
429         ld->cq_ring_mask = *cring->ring_mask;
430         return 0;
431 }
432
433 static int fio_ioring_queue_init(struct thread_data *td)
434 {
435         struct ioring_data *ld = td->io_ops_data;
436         struct ioring_options *o = td->eo;
437         int depth = td->o.iodepth;
438         struct io_uring_params p;
439         int ret;
440
441         memset(&p, 0, sizeof(p));
442
443         if (o->hipri)
444                 p.flags |= IORING_SETUP_IOPOLL;
445         if (o->sqpoll_thread) {
446                 p.flags |= IORING_SETUP_SQPOLL;
447                 if (o->sqpoll_set) {
448                         p.flags |= IORING_SETUP_SQ_AFF;
449                         p.sq_thread_cpu = o->sqpoll_cpu;
450                 }
451         }
452
453         ret = syscall(__NR_sys_io_uring_setup, depth, &p);
454         if (ret < 0)
455                 return ret;
456
457         ld->ring_fd = ret;
458
459         if (o->fixedbufs) {
460                 struct rlimit rlim = {
461                         .rlim_cur = RLIM_INFINITY,
462                         .rlim_max = RLIM_INFINITY,
463                 };
464
465                 if (setrlimit(RLIMIT_MEMLOCK, &rlim) < 0)
466                         return -1;
467
468                 ret = syscall(__NR_sys_io_uring_register, ld->ring_fd,
469                                 IORING_REGISTER_BUFFERS, ld->iovecs, depth);
470                 if (ret < 0)
471                         return ret;
472         }
473
474         return fio_ioring_mmap(ld, &p);
475 }
476
477 static int fio_ioring_post_init(struct thread_data *td)
478 {
479         struct ioring_data *ld = td->io_ops_data;
480         struct io_u *io_u;
481         int err, i;
482
483         for (i = 0; i < td->o.iodepth; i++) {
484                 struct iovec *iov = &ld->iovecs[i];
485
486                 io_u = ld->io_u_index[i];
487                 iov->iov_base = io_u->buf;
488                 iov->iov_len = td_max_bs(td);
489         }
490
491         err = fio_ioring_queue_init(td);
492         if (err) {
493                 td_verror(td, errno, "io_queue_init");
494                 return 1;
495         }
496
497         return 0;
498 }
499
500 static unsigned roundup_pow2(unsigned depth)
501 {
502         return 1UL << __fls(depth - 1);
503 }
504
505 static int fio_ioring_init(struct thread_data *td)
506 {
507         struct ioring_data *ld;
508
509         ld = calloc(1, sizeof(*ld));
510
511         /* ring depth must be a power-of-2 */
512         ld->iodepth = td->o.iodepth;
513         td->o.iodepth = roundup_pow2(td->o.iodepth);
514
515         /* io_u index */
516         ld->io_u_index = calloc(td->o.iodepth, sizeof(struct io_u *));
517         ld->iovecs = calloc(td->o.iodepth, sizeof(struct iovec));
518
519         td->io_ops_data = ld;
520         return 0;
521 }
522
523 static int fio_ioring_io_u_init(struct thread_data *td, struct io_u *io_u)
524 {
525         struct ioring_data *ld = td->io_ops_data;
526
527         ld->io_u_index[io_u->index] = io_u;
528         return 0;
529 }
530
531 static struct ioengine_ops ioengine = {
532         .name                   = "io_uring",
533         .version                = FIO_IOOPS_VERSION,
534         .init                   = fio_ioring_init,
535         .post_init              = fio_ioring_post_init,
536         .io_u_init              = fio_ioring_io_u_init,
537         .prep                   = fio_ioring_prep,
538         .queue                  = fio_ioring_queue,
539         .commit                 = fio_ioring_commit,
540         .getevents              = fio_ioring_getevents,
541         .event                  = fio_ioring_event,
542         .cleanup                = fio_ioring_cleanup,
543         .open_file              = generic_open_file,
544         .close_file             = generic_close_file,
545         .get_file_size          = generic_get_file_size,
546         .options                = options,
547         .option_struct_size     = sizeof(struct ioring_options),
548 };
549
550 static void fio_init fio_ioring_register(void)
551 {
552         register_ioengine(&ioengine);
553 }
554
555 static void fio_exit fio_ioring_unregister(void)
556 {
557         unregister_ioengine(&ioengine);
558 }
559 #endif