8cf338584c9a823ba60ca24012fcf49d980e2442
[fio.git] / t / io_uring.c
1 #include <stdio.h>
2 #include <errno.h>
3 #include <assert.h>
4 #include <stdlib.h>
5 #include <stddef.h>
6 #include <signal.h>
7 #include <inttypes.h>
8 #include <math.h>
9
10 #ifdef CONFIG_LIBAIO
11 #include <libaio.h>
12 #endif
13
14 #include <sys/types.h>
15 #include <sys/stat.h>
16 #include <sys/ioctl.h>
17 #include <sys/syscall.h>
18 #include <sys/resource.h>
19 #include <sys/mman.h>
20 #include <sys/uio.h>
21 #include <linux/fs.h>
22 #include <fcntl.h>
23 #include <unistd.h>
24 #include <string.h>
25 #include <pthread.h>
26 #include <sched.h>
27
28 #include "../arch/arch.h"
29 #include "../lib/types.h"
30 #include "../lib/roundup.h"
31 #include "../lib/rand.h"
32 #include "../minmax.h"
33 #include "../os/linux/io_uring.h"
34
35 struct io_sq_ring {
36         unsigned *head;
37         unsigned *tail;
38         unsigned *ring_mask;
39         unsigned *ring_entries;
40         unsigned *flags;
41         unsigned *array;
42 };
43
44 struct io_cq_ring {
45         unsigned *head;
46         unsigned *tail;
47         unsigned *ring_mask;
48         unsigned *ring_entries;
49         struct io_uring_cqe *cqes;
50 };
51
52 #define DEPTH                   128
53 #define BATCH_SUBMIT            32
54 #define BATCH_COMPLETE          32
55 #define BS                      4096
56
57 #define MAX_FDS                 16
58
59 static unsigned sq_ring_mask, cq_ring_mask;
60
61 struct file {
62         unsigned long max_blocks;
63         unsigned long max_size;
64         unsigned long cur_off;
65         unsigned pending_ios;
66         int real_fd;
67         int fixed_fd;
68         int fileno;
69 };
70
71 #define PLAT_BITS               6
72 #define PLAT_VAL                (1 << PLAT_BITS)
73 #define PLAT_GROUP_NR           29
74 #define PLAT_NR                 (PLAT_GROUP_NR * PLAT_VAL)
75
76 struct submitter {
77         pthread_t thread;
78         int ring_fd;
79         int index;
80         struct io_sq_ring sq_ring;
81         struct io_uring_sqe *sqes;
82         struct io_cq_ring cq_ring;
83         int inflight;
84         int tid;
85         unsigned long reaps;
86         unsigned long done;
87         unsigned long calls;
88         volatile int finish;
89
90         __s32 *fds;
91
92         struct taus258_state rand_state;
93
94         unsigned long *clock_batch;
95         int clock_index;
96         unsigned long *plat;
97
98 #ifdef CONFIG_LIBAIO
99         io_context_t aio_ctx;
100 #endif
101
102         struct file files[MAX_FDS];
103         unsigned nr_files;
104         unsigned cur_file;
105         struct iovec iovecs[];
106 };
107
108 static struct submitter *submitter;
109 static volatile int finish;
110 static int stats_running;
111 static unsigned long max_iops;
112
113 static int depth = DEPTH;
114 static int batch_submit = BATCH_SUBMIT;
115 static int batch_complete = BATCH_COMPLETE;
116 static int bs = BS;
117 static int polled = 1;          /* use IO polling */
118 static int fixedbufs = 1;       /* use fixed user buffers */
119 static int dma_map;             /* pre-map DMA buffers */
120 static int register_files = 1;  /* use fixed files */
121 static int buffered = 0;        /* use buffered IO, not O_DIRECT */
122 static int sq_thread_poll = 0;  /* use kernel submission/poller thread */
123 static int sq_thread_cpu = -1;  /* pin above thread to this CPU */
124 static int do_nop = 0;          /* no-op SQ ring commands */
125 static int nthreads = 1;
126 static int stats = 0;           /* generate IO stats */
127 static int aio = 0;             /* use libaio */
128 static int runtime = 0;         /* runtime */
129 static int random_io = 1;       /* random or sequential IO */
130
131 static unsigned long tsc_rate;
132
133 #define TSC_RATE_FILE   "tsc-rate"
134
135 static int vectored = 1;
136
137 static float plist[] = { 1.0, 5.0, 10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0,
138                         80.0, 90.0, 95.0, 99.0, 99.5, 99.9, 99.95, 99.99 };
139 static int plist_len = 17;
140
141 #ifndef IORING_REGISTER_MAP_BUFFERS
142 #define IORING_REGISTER_MAP_BUFFERS     22
143 struct io_uring_map_buffers {
144         __s32   fd;
145         __u32   buf_start;
146         __u32   buf_end;
147         __u32   flags;
148         __u64   rsvd[2];
149 };
150 #endif
151
152 static unsigned long cycles_to_nsec(unsigned long cycles)
153 {
154         uint64_t val;
155
156         if (!tsc_rate)
157                 return cycles;
158
159         val = cycles * 1000000000ULL;
160         return val / tsc_rate;
161 }
162
163 static unsigned long plat_idx_to_val(unsigned int idx)
164 {
165         unsigned int error_bits;
166         unsigned long k, base;
167
168         assert(idx < PLAT_NR);
169
170         /* MSB <= (PLAT_BITS-1), cannot be rounded off. Use
171          * all bits of the sample as index */
172         if (idx < (PLAT_VAL << 1))
173                 return cycles_to_nsec(idx);
174
175         /* Find the group and compute the minimum value of that group */
176         error_bits = (idx >> PLAT_BITS) - 1;
177         base = ((unsigned long) 1) << (error_bits + PLAT_BITS);
178
179         /* Find its bucket number of the group */
180         k = idx % PLAT_VAL;
181
182         /* Return the mean of the range of the bucket */
183         return cycles_to_nsec(base + ((k + 0.5) * (1 << error_bits)));
184 }
185
186 unsigned int calc_clat_percentiles(unsigned long *io_u_plat, unsigned long nr,
187                                    unsigned long **output,
188                                    unsigned long *maxv, unsigned long *minv)
189 {
190         unsigned long sum = 0;
191         unsigned int len = plist_len, i, j = 0;
192         unsigned long *ovals = NULL;
193         bool is_last;
194
195         *minv = -1UL;
196         *maxv = 0;
197
198         ovals = malloc(len * sizeof(*ovals));
199         if (!ovals)
200                 return 0;
201
202         /*
203          * Calculate bucket values, note down max and min values
204          */
205         is_last = false;
206         for (i = 0; i < PLAT_NR && !is_last; i++) {
207                 sum += io_u_plat[i];
208                 while (sum >= ((long double) plist[j] / 100.0 * nr)) {
209                         assert(plist[j] <= 100.0);
210
211                         ovals[j] = plat_idx_to_val(i);
212                         if (ovals[j] < *minv)
213                                 *minv = ovals[j];
214                         if (ovals[j] > *maxv)
215                                 *maxv = ovals[j];
216
217                         is_last = (j == len - 1) != 0;
218                         if (is_last)
219                                 break;
220
221                         j++;
222                 }
223         }
224
225         if (!is_last)
226                 fprintf(stderr, "error calculating latency percentiles\n");
227
228         *output = ovals;
229         return len;
230 }
231
232 static void show_clat_percentiles(unsigned long *io_u_plat, unsigned long nr,
233                                   unsigned int precision)
234 {
235         unsigned int divisor, len, i, j = 0;
236         unsigned long minv, maxv;
237         unsigned long *ovals;
238         int per_line, scale_down, time_width;
239         bool is_last;
240         char fmt[32];
241
242         len = calc_clat_percentiles(io_u_plat, nr, &ovals, &maxv, &minv);
243         if (!len || !ovals)
244                 goto out;
245
246         if (!tsc_rate) {
247                 scale_down = 0;
248                 divisor = 1;
249                 printf("    percentiles (tsc ticks):\n     |");
250         } else if (minv > 2000 && maxv > 99999) {
251                 scale_down = 1;
252                 divisor = 1000;
253                 printf("    percentiles (usec):\n     |");
254         } else {
255                 scale_down = 0;
256                 divisor = 1;
257                 printf("    percentiles (nsec):\n     |");
258         }
259
260         time_width = max(5, (int) (log10(maxv / divisor) + 1));
261         snprintf(fmt, sizeof(fmt), " %%%u.%ufth=[%%%dllu]%%c", precision + 3,
262                         precision, time_width);
263         /* fmt will be something like " %5.2fth=[%4llu]%c" */
264         per_line = (80 - 7) / (precision + 10 + time_width);
265
266         for (j = 0; j < len; j++) {
267                 /* for formatting */
268                 if (j != 0 && (j % per_line) == 0)
269                         printf("     |");
270
271                 /* end of the list */
272                 is_last = (j == len - 1) != 0;
273
274                 for (i = 0; i < scale_down; i++)
275                         ovals[j] = (ovals[j] + 999) / 1000;
276
277                 printf(fmt, plist[j], ovals[j], is_last ? '\n' : ',');
278
279                 if (is_last)
280                         break;
281
282                 if ((j % per_line) == per_line - 1)     /* for formatting */
283                         printf("\n");
284         }
285
286 out:
287         free(ovals);
288 }
289
290 #ifdef ARCH_HAVE_CPU_CLOCK
291 static unsigned int plat_val_to_idx(unsigned long val)
292 {
293         unsigned int msb, error_bits, base, offset, idx;
294
295         /* Find MSB starting from bit 0 */
296         if (val == 0)
297                 msb = 0;
298         else
299                 msb = (sizeof(val)*8) - __builtin_clzll(val) - 1;
300
301         /*
302          * MSB <= (PLAT_BITS-1), cannot be rounded off. Use
303          * all bits of the sample as index
304          */
305         if (msb <= PLAT_BITS)
306                 return val;
307
308         /* Compute the number of error bits to discard*/
309         error_bits = msb - PLAT_BITS;
310
311         /* Compute the number of buckets before the group */
312         base = (error_bits + 1) << PLAT_BITS;
313
314         /*
315          * Discard the error bits and apply the mask to find the
316          * index for the buckets in the group
317          */
318         offset = (PLAT_VAL - 1) & (val >> error_bits);
319
320         /* Make sure the index does not exceed (array size - 1) */
321         idx = (base + offset) < (PLAT_NR - 1) ?
322                 (base + offset) : (PLAT_NR - 1);
323
324         return idx;
325 }
326 #endif
327
328 static void add_stat(struct submitter *s, int clock_index, int nr)
329 {
330 #ifdef ARCH_HAVE_CPU_CLOCK
331         unsigned long cycles;
332         unsigned int pidx;
333
334         if (!s->finish && clock_index) {
335                 cycles = get_cpu_clock();
336                 cycles -= s->clock_batch[clock_index];
337                 pidx = plat_val_to_idx(cycles);
338                 s->plat[pidx] += nr;
339         }
340 #endif
341 }
342
343 static int io_uring_map_buffers(struct submitter *s)
344 {
345         struct io_uring_map_buffers map = {
346                 .fd             = s->files[0].real_fd,
347                 .buf_end        = depth,
348         };
349
350         if (do_nop)
351                 return 0;
352         if (s->nr_files > 1) {
353                 fprintf(stderr, "Can't map buffers with multiple files\n");
354                 return -1;
355         }
356
357         return syscall(__NR_io_uring_register, s->ring_fd,
358                         IORING_REGISTER_MAP_BUFFERS, &map, 1);
359 }
360
361 static int io_uring_register_buffers(struct submitter *s)
362 {
363         if (do_nop)
364                 return 0;
365
366         return syscall(__NR_io_uring_register, s->ring_fd,
367                         IORING_REGISTER_BUFFERS, s->iovecs, roundup_pow2(depth));
368 }
369
370 static int io_uring_register_files(struct submitter *s)
371 {
372         int i;
373
374         if (do_nop)
375                 return 0;
376
377         s->fds = calloc(s->nr_files, sizeof(__s32));
378         for (i = 0; i < s->nr_files; i++) {
379                 s->fds[i] = s->files[i].real_fd;
380                 s->files[i].fixed_fd = i;
381         }
382
383         return syscall(__NR_io_uring_register, s->ring_fd,
384                         IORING_REGISTER_FILES, s->fds, s->nr_files);
385 }
386
387 static int io_uring_setup(unsigned entries, struct io_uring_params *p)
388 {
389         /*
390          * Clamp CQ ring size at our SQ ring size, we don't need more entries
391          * than that.
392          */
393         p->flags |= IORING_SETUP_CQSIZE;
394         p->cq_entries = entries;
395
396         return syscall(__NR_io_uring_setup, entries, p);
397 }
398
399 static void io_uring_probe(int fd)
400 {
401         struct io_uring_probe *p;
402         int ret;
403
404         p = malloc(sizeof(*p) + 256 * sizeof(struct io_uring_probe_op));
405         if (!p)
406                 return;
407
408         memset(p, 0, sizeof(*p) + 256 * sizeof(struct io_uring_probe_op));
409         ret = syscall(__NR_io_uring_register, fd, IORING_REGISTER_PROBE, p, 256);
410         if (ret < 0)
411                 goto out;
412
413         if (IORING_OP_READ > p->ops_len)
414                 goto out;
415
416         if ((p->ops[IORING_OP_READ].flags & IO_URING_OP_SUPPORTED))
417                 vectored = 0;
418 out:
419         free(p);
420 }
421
422 static int io_uring_enter(struct submitter *s, unsigned int to_submit,
423                           unsigned int min_complete, unsigned int flags)
424 {
425 #ifdef FIO_ARCH_HAS_SYSCALL
426         return __do_syscall6(__NR_io_uring_enter, s->ring_fd, to_submit,
427                                 min_complete, flags, NULL, 0);
428 #else
429         return syscall(__NR_io_uring_enter, s->ring_fd, to_submit, min_complete,
430                         flags, NULL, 0);
431 #endif
432 }
433
434 #ifndef CONFIG_HAVE_GETTID
435 static int gettid(void)
436 {
437         return syscall(__NR_gettid);
438 }
439 #endif
440
441 static unsigned file_depth(struct submitter *s)
442 {
443         return (depth + s->nr_files - 1) / s->nr_files;
444 }
445
446 static void init_io(struct submitter *s, unsigned index)
447 {
448         struct io_uring_sqe *sqe = &s->sqes[index];
449         unsigned long offset;
450         struct file *f;
451         long r;
452
453         if (do_nop) {
454                 sqe->opcode = IORING_OP_NOP;
455                 return;
456         }
457
458         if (s->nr_files == 1) {
459                 f = &s->files[0];
460         } else {
461                 f = &s->files[s->cur_file];
462                 if (f->pending_ios >= file_depth(s)) {
463                         s->cur_file++;
464                         if (s->cur_file == s->nr_files)
465                                 s->cur_file = 0;
466                         f = &s->files[s->cur_file];
467                 }
468         }
469         f->pending_ios++;
470
471         if (random_io) {
472                 r = __rand64(&s->rand_state);
473                 offset = (r % (f->max_blocks - 1)) * bs;
474         } else {
475                 offset = f->cur_off;
476                 f->cur_off += bs;
477                 if (f->cur_off + bs > f->max_size)
478                         f->cur_off = 0;
479         }
480
481         if (register_files) {
482                 sqe->flags = IOSQE_FIXED_FILE;
483                 sqe->fd = f->fixed_fd;
484         } else {
485                 sqe->flags = 0;
486                 sqe->fd = f->real_fd;
487         }
488         if (fixedbufs) {
489                 sqe->opcode = IORING_OP_READ_FIXED;
490                 sqe->addr = (unsigned long) s->iovecs[index].iov_base;
491                 sqe->len = bs;
492                 sqe->buf_index = index;
493         } else if (!vectored) {
494                 sqe->opcode = IORING_OP_READ;
495                 sqe->addr = (unsigned long) s->iovecs[index].iov_base;
496                 sqe->len = bs;
497                 sqe->buf_index = 0;
498         } else {
499                 sqe->opcode = IORING_OP_READV;
500                 sqe->addr = (unsigned long) &s->iovecs[index];
501                 sqe->len = 1;
502                 sqe->buf_index = 0;
503         }
504         sqe->ioprio = 0;
505         sqe->off = offset;
506         sqe->user_data = (unsigned long) f->fileno;
507         if (stats && stats_running)
508                 sqe->user_data |= ((uint64_t)s->clock_index << 32);
509 }
510
511 static int prep_more_ios_uring(struct submitter *s, int max_ios)
512 {
513         struct io_sq_ring *ring = &s->sq_ring;
514         unsigned index, tail, next_tail, prepped = 0;
515
516         next_tail = tail = *ring->tail;
517         do {
518                 next_tail++;
519                 if (next_tail == atomic_load_acquire(ring->head))
520                         break;
521
522                 index = tail & sq_ring_mask;
523                 init_io(s, index);
524                 ring->array[index] = index;
525                 prepped++;
526                 tail = next_tail;
527         } while (prepped < max_ios);
528
529         if (prepped)
530                 atomic_store_release(ring->tail, tail);
531         return prepped;
532 }
533
534 static int get_file_size(struct file *f)
535 {
536         struct stat st;
537
538         if (fstat(f->real_fd, &st) < 0)
539                 return -1;
540         if (S_ISBLK(st.st_mode)) {
541                 unsigned long long bytes;
542
543                 if (ioctl(f->real_fd, BLKGETSIZE64, &bytes) != 0)
544                         return -1;
545
546                 f->max_blocks = bytes / bs;
547                 f->max_size = bytes;
548                 return 0;
549         } else if (S_ISREG(st.st_mode)) {
550                 f->max_blocks = st.st_size / bs;
551                 f->max_size = st.st_size;
552                 return 0;
553         }
554
555         return -1;
556 }
557
558 static int reap_events_uring(struct submitter *s)
559 {
560         struct io_cq_ring *ring = &s->cq_ring;
561         struct io_uring_cqe *cqe;
562         unsigned head, reaped = 0;
563         int last_idx = -1, stat_nr = 0;
564
565         head = *ring->head;
566         do {
567                 struct file *f;
568
569                 read_barrier();
570                 if (head == atomic_load_acquire(ring->tail))
571                         break;
572                 cqe = &ring->cqes[head & cq_ring_mask];
573                 if (!do_nop) {
574                         int fileno = cqe->user_data & 0xffffffff;
575
576                         f = &s->files[fileno];
577                         f->pending_ios--;
578                         if (cqe->res != bs) {
579                                 printf("io: unexpected ret=%d\n", cqe->res);
580                                 if (polled && cqe->res == -EOPNOTSUPP)
581                                         printf("Your filesystem/driver/kernel doesn't support polled IO\n");
582                                 return -1;
583                         }
584                 }
585                 if (stats) {
586                         int clock_index = cqe->user_data >> 32;
587
588                         if (last_idx != clock_index) {
589                                 if (last_idx != -1) {
590                                         add_stat(s, last_idx, stat_nr);
591                                         stat_nr = 0;
592                                 }
593                                 last_idx = clock_index;
594                         }
595                         stat_nr++;
596                 }
597                 reaped++;
598                 head++;
599         } while (1);
600
601         if (stat_nr)
602                 add_stat(s, last_idx, stat_nr);
603
604         if (reaped) {
605                 s->inflight -= reaped;
606                 atomic_store_release(ring->head, head);
607         }
608         return reaped;
609 }
610
611 static int submitter_init(struct submitter *s)
612 {
613         int i, nr_batch;
614
615         s->tid = gettid();
616         printf("submitter=%d, tid=%d\n", s->index, s->tid);
617
618         __init_rand64(&s->rand_state, pthread_self());
619         srand48(pthread_self());
620
621         for (i = 0; i < MAX_FDS; i++)
622                 s->files[i].fileno = i;
623
624         if (stats) {
625                 nr_batch = roundup_pow2(depth / batch_submit);
626                 if (nr_batch < 2)
627                         nr_batch = 2;
628                 s->clock_batch = calloc(nr_batch, sizeof(unsigned long));
629                 s->clock_index = 1;
630
631                 s->plat = calloc(PLAT_NR, sizeof(unsigned long));
632         } else {
633                 s->clock_batch = NULL;
634                 s->plat = NULL;
635                 nr_batch = 0;
636         }
637
638         return nr_batch;
639 }
640
641 #ifdef CONFIG_LIBAIO
642 static int prep_more_ios_aio(struct submitter *s, int max_ios, struct iocb *iocbs)
643 {
644         uint64_t data;
645         long long offset;
646         struct file *f;
647         unsigned index;
648         long r;
649
650         index = 0;
651         while (index < max_ios) {
652                 struct iocb *iocb = &iocbs[index];
653
654                 if (s->nr_files == 1) {
655                         f = &s->files[0];
656                 } else {
657                         f = &s->files[s->cur_file];
658                         if (f->pending_ios >= file_depth(s)) {
659                                 s->cur_file++;
660                                 if (s->cur_file == s->nr_files)
661                                         s->cur_file = 0;
662                                 f = &s->files[s->cur_file];
663                         }
664                 }
665                 f->pending_ios++;
666
667                 r = lrand48();
668                 offset = (r % (f->max_blocks - 1)) * bs;
669                 io_prep_pread(iocb, f->real_fd, s->iovecs[index].iov_base,
670                                 s->iovecs[index].iov_len, offset);
671
672                 data = f->fileno;
673                 if (stats && stats_running)
674                         data |= (((uint64_t) s->clock_index) << 32);
675                 iocb->data = (void *) (uintptr_t) data;
676                 index++;
677         }
678         return index;
679 }
680
681 static int reap_events_aio(struct submitter *s, struct io_event *events, int evs)
682 {
683         int last_idx = -1, stat_nr = 0;
684         int reaped = 0;
685
686         while (evs) {
687                 uint64_t data = (uintptr_t) events[reaped].data;
688                 struct file *f = &s->files[data & 0xffffffff];
689
690                 f->pending_ios--;
691                 if (events[reaped].res != bs) {
692                         printf("io: unexpected ret=%ld\n", events[reaped].res);
693                         return -1;
694                 }
695                 if (stats) {
696                         int clock_index = data >> 32;
697
698                         if (last_idx != clock_index) {
699                                 if (last_idx != -1) {
700                                         add_stat(s, last_idx, stat_nr);
701                                         stat_nr = 0;
702                                 }
703                                 last_idx = clock_index;
704                         }
705                         stat_nr++;
706                 }
707                 reaped++;
708                 evs--;
709         }
710
711         if (stat_nr)
712                 add_stat(s, last_idx, stat_nr);
713
714         s->inflight -= reaped;
715         s->done += reaped;
716         return reaped;
717 }
718
719 static void *submitter_aio_fn(void *data)
720 {
721         struct submitter *s = data;
722         int i, ret, prepped;
723         struct iocb **iocbsptr;
724         struct iocb *iocbs;
725         struct io_event *events;
726 #ifdef ARCH_HAVE_CPU_CLOCK
727         int nr_batch = submitter_init(s);
728 #else
729         submitter_init(s);
730 #endif
731
732         iocbsptr = calloc(depth, sizeof(struct iocb *));
733         iocbs = calloc(depth, sizeof(struct iocb));
734         events = calloc(depth, sizeof(struct io_event));
735
736         for (i = 0; i < depth; i++)
737                 iocbsptr[i] = &iocbs[i];
738
739         prepped = 0;
740         do {
741                 int to_wait, to_submit, to_prep;
742
743                 if (!prepped && s->inflight < depth) {
744                         to_prep = min(depth - s->inflight, batch_submit);
745                         prepped = prep_more_ios_aio(s, to_prep, iocbs);
746 #ifdef ARCH_HAVE_CPU_CLOCK
747                         if (prepped && stats) {
748                                 s->clock_batch[s->clock_index] = get_cpu_clock();
749                                 s->clock_index = (s->clock_index + 1) & (nr_batch - 1);
750                         }
751 #endif
752                 }
753                 s->inflight += prepped;
754                 to_submit = prepped;
755
756                 if (to_submit && (s->inflight + to_submit <= depth))
757                         to_wait = 0;
758                 else
759                         to_wait = min(s->inflight + to_submit, batch_complete);
760
761                 ret = io_submit(s->aio_ctx, to_submit, iocbsptr);
762                 s->calls++;
763                 if (ret < 0) {
764                         perror("io_submit");
765                         break;
766                 } else if (ret != to_submit) {
767                         printf("submitted %d, wanted %d\n", ret, to_submit);
768                         break;
769                 }
770                 prepped = 0;
771
772                 while (to_wait) {
773                         int r;
774
775                         s->calls++;
776                         r = io_getevents(s->aio_ctx, to_wait, to_wait, events, NULL);
777                         if (r < 0) {
778                                 perror("io_getevents");
779                                 break;
780                         } else if (r != to_wait) {
781                                 printf("r=%d, wait=%d\n", r, to_wait);
782                                 break;
783                         }
784                         r = reap_events_aio(s, events, r);
785                         s->reaps += r;
786                         to_wait -= r;
787                 }
788         } while (!s->finish);
789
790         free(iocbsptr);
791         free(iocbs);
792         free(events);
793         finish = 1;
794         return NULL;
795 }
796 #endif
797
798 static void *submitter_uring_fn(void *data)
799 {
800         struct submitter *s = data;
801         struct io_sq_ring *ring = &s->sq_ring;
802         int ret, prepped;
803 #ifdef ARCH_HAVE_CPU_CLOCK
804         int nr_batch = submitter_init(s);
805 #else
806         submitter_init(s);
807 #endif
808
809         prepped = 0;
810         do {
811                 int to_wait, to_submit, this_reap, to_prep;
812                 unsigned ring_flags = 0;
813
814                 if (!prepped && s->inflight < depth) {
815                         to_prep = min(depth - s->inflight, batch_submit);
816                         prepped = prep_more_ios_uring(s, to_prep);
817 #ifdef ARCH_HAVE_CPU_CLOCK
818                         if (prepped && stats) {
819                                 s->clock_batch[s->clock_index] = get_cpu_clock();
820                                 s->clock_index = (s->clock_index + 1) & (nr_batch - 1);
821                         }
822 #endif
823                 }
824                 s->inflight += prepped;
825 submit_more:
826                 to_submit = prepped;
827 submit:
828                 if (to_submit && (s->inflight + to_submit <= depth))
829                         to_wait = 0;
830                 else
831                         to_wait = min(s->inflight + to_submit, batch_complete);
832
833                 /*
834                  * Only need to call io_uring_enter if we're not using SQ thread
835                  * poll, or if IORING_SQ_NEED_WAKEUP is set.
836                  */
837                 if (sq_thread_poll)
838                         ring_flags = atomic_load_acquire(ring->flags);
839                 if (!sq_thread_poll || ring_flags & IORING_SQ_NEED_WAKEUP) {
840                         unsigned flags = 0;
841
842                         if (to_wait)
843                                 flags = IORING_ENTER_GETEVENTS;
844                         if (ring_flags & IORING_SQ_NEED_WAKEUP)
845                                 flags |= IORING_ENTER_SQ_WAKEUP;
846                         ret = io_uring_enter(s, to_submit, to_wait, flags);
847                         s->calls++;
848                 } else {
849                         /* for SQPOLL, we submitted it all effectively */
850                         ret = to_submit;
851                 }
852
853                 /*
854                  * For non SQ thread poll, we already got the events we needed
855                  * through the io_uring_enter() above. For SQ thread poll, we
856                  * need to loop here until we find enough events.
857                  */
858                 this_reap = 0;
859                 do {
860                         int r;
861
862                         r = reap_events_uring(s);
863                         if (r == -1) {
864                                 s->finish = 1;
865                                 break;
866                         } else if (r > 0)
867                                 this_reap += r;
868                 } while (sq_thread_poll && this_reap < to_wait);
869                 s->reaps += this_reap;
870
871                 if (ret >= 0) {
872                         if (!ret) {
873                                 to_submit = 0;
874                                 if (s->inflight)
875                                         goto submit;
876                                 continue;
877                         } else if (ret < to_submit) {
878                                 int diff = to_submit - ret;
879
880                                 s->done += ret;
881                                 prepped -= diff;
882                                 goto submit_more;
883                         }
884                         s->done += ret;
885                         prepped = 0;
886                         continue;
887                 } else if (ret < 0) {
888                         if (errno == EAGAIN) {
889                                 if (s->finish)
890                                         break;
891                                 if (this_reap)
892                                         goto submit;
893                                 to_submit = 0;
894                                 goto submit;
895                         }
896                         printf("io_submit: %s\n", strerror(errno));
897                         break;
898                 }
899         } while (!s->finish);
900
901         finish = 1;
902         return NULL;
903 }
904
905 static struct submitter *get_submitter(int offset)
906 {
907         void *ret;
908
909         ret = submitter;
910         if (offset)
911                 ret += offset * (sizeof(*submitter) + depth * sizeof(struct iovec));
912         return ret;
913 }
914
915 static void do_finish(const char *reason)
916 {
917         int j;
918         printf("Exiting on %s\n", reason);
919         for (j = 0; j < nthreads; j++) {
920                 struct submitter *s = get_submitter(j);
921                 s->finish = 1;
922         }
923         if (max_iops > 100000)
924                 printf("Maximum IOPS=%luK\n", max_iops / 1000);
925         else if (max_iops)
926                 printf("Maximum IOPS=%lu\n", max_iops);
927         finish = 1;
928 }
929
930 static void sig_int(int sig)
931 {
932         do_finish("signal");
933 }
934
935 static void arm_sig_int(void)
936 {
937         struct sigaction act;
938
939         memset(&act, 0, sizeof(act));
940         act.sa_handler = sig_int;
941         act.sa_flags = SA_RESTART;
942         sigaction(SIGINT, &act, NULL);
943
944         /* Windows uses SIGBREAK as a quit signal from other applications */
945 #ifdef WIN32
946         sigaction(SIGBREAK, &act, NULL);
947 #endif
948 }
949
950 static int setup_aio(struct submitter *s)
951 {
952 #ifdef CONFIG_LIBAIO
953         if (polled) {
954                 fprintf(stderr, "aio does not support polled IO\n");
955                 polled = 0;
956         }
957         if (sq_thread_poll) {
958                 fprintf(stderr, "aio does not support SQPOLL IO\n");
959                 sq_thread_poll = 0;
960         }
961         if (do_nop) {
962                 fprintf(stderr, "aio does not support polled IO\n");
963                 do_nop = 0;
964         }
965         if (fixedbufs || register_files) {
966                 fprintf(stderr, "aio does not support registered files or buffers\n");
967                 fixedbufs = register_files = 0;
968         }
969
970         return io_queue_init(roundup_pow2(depth), &s->aio_ctx);
971 #else
972         fprintf(stderr, "Legacy AIO not available on this system/build\n");
973         errno = EINVAL;
974         return -1;
975 #endif
976 }
977
978 static int setup_ring(struct submitter *s)
979 {
980         struct io_sq_ring *sring = &s->sq_ring;
981         struct io_cq_ring *cring = &s->cq_ring;
982         struct io_uring_params p;
983         int ret, fd;
984         void *ptr;
985
986         memset(&p, 0, sizeof(p));
987
988         if (polled && !do_nop)
989                 p.flags |= IORING_SETUP_IOPOLL;
990         if (sq_thread_poll) {
991                 p.flags |= IORING_SETUP_SQPOLL;
992                 if (sq_thread_cpu != -1) {
993                         p.flags |= IORING_SETUP_SQ_AFF;
994                         p.sq_thread_cpu = sq_thread_cpu;
995                 }
996         }
997
998         fd = io_uring_setup(depth, &p);
999         if (fd < 0) {
1000                 perror("io_uring_setup");
1001                 return 1;
1002         }
1003         s->ring_fd = fd;
1004
1005         io_uring_probe(fd);
1006
1007         if (fixedbufs) {
1008                 struct rlimit rlim;
1009
1010                 rlim.rlim_cur = RLIM_INFINITY;
1011                 rlim.rlim_max = RLIM_INFINITY;
1012                 /* ignore potential error, not needed on newer kernels */
1013                 setrlimit(RLIMIT_MEMLOCK, &rlim);
1014
1015                 ret = io_uring_register_buffers(s);
1016                 if (ret < 0) {
1017                         perror("io_uring_register_buffers");
1018                         return 1;
1019                 }
1020
1021                 if (dma_map) {
1022                         ret = io_uring_map_buffers(s);
1023                         if (ret < 0) {
1024                                 perror("io_uring_map_buffers");
1025                                 return 1;
1026                         }
1027                 }
1028         }
1029
1030         if (register_files) {
1031                 ret = io_uring_register_files(s);
1032                 if (ret < 0) {
1033                         perror("io_uring_register_files");
1034                         return 1;
1035                 }
1036         }
1037
1038         ptr = mmap(0, p.sq_off.array + p.sq_entries * sizeof(__u32),
1039                         PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd,
1040                         IORING_OFF_SQ_RING);
1041         sring->head = ptr + p.sq_off.head;
1042         sring->tail = ptr + p.sq_off.tail;
1043         sring->ring_mask = ptr + p.sq_off.ring_mask;
1044         sring->ring_entries = ptr + p.sq_off.ring_entries;
1045         sring->flags = ptr + p.sq_off.flags;
1046         sring->array = ptr + p.sq_off.array;
1047         sq_ring_mask = *sring->ring_mask;
1048
1049         s->sqes = mmap(0, p.sq_entries * sizeof(struct io_uring_sqe),
1050                         PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd,
1051                         IORING_OFF_SQES);
1052
1053         ptr = mmap(0, p.cq_off.cqes + p.cq_entries * sizeof(struct io_uring_cqe),
1054                         PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd,
1055                         IORING_OFF_CQ_RING);
1056         cring->head = ptr + p.cq_off.head;
1057         cring->tail = ptr + p.cq_off.tail;
1058         cring->ring_mask = ptr + p.cq_off.ring_mask;
1059         cring->ring_entries = ptr + p.cq_off.ring_entries;
1060         cring->cqes = ptr + p.cq_off.cqes;
1061         cq_ring_mask = *cring->ring_mask;
1062         return 0;
1063 }
1064
1065 static void file_depths(char *buf)
1066 {
1067         bool prev = false;
1068         char *p;
1069         int i, j;
1070
1071         buf[0] = '\0';
1072         p = buf;
1073         for (j = 0; j < nthreads; j++) {
1074                 struct submitter *s = get_submitter(j);
1075
1076                 for (i = 0; i < s->nr_files; i++) {
1077                         struct file *f = &s->files[i];
1078
1079                         if (prev)
1080                                 p += sprintf(p, " %d", f->pending_ios);
1081                         else
1082                                 p += sprintf(p, "%d", f->pending_ios);
1083                         prev = true;
1084                 }
1085         }
1086 }
1087
1088 static void usage(char *argv, int status)
1089 {
1090         char runtime_str[16];
1091         snprintf(runtime_str, sizeof(runtime_str), "%d", runtime);
1092         printf("%s [options] -- [filenames]\n"
1093                 " -d <int>  : IO Depth, default %d\n"
1094                 " -s <int>  : Batch submit, default %d\n"
1095                 " -c <int>  : Batch complete, default %d\n"
1096                 " -b <int>  : Block size, default %d\n"
1097                 " -p <bool> : Polled IO, default %d\n"
1098                 " -B <bool> : Fixed buffers, default %d\n"
1099                 " -D <bool> : DMA map fixed buffers, default %d\n"
1100                 " -F <bool> : Register files, default %d\n"
1101                 " -n <int>  : Number of threads, default %d\n"
1102                 " -O <bool> : Use O_DIRECT, default %d\n"
1103                 " -N <bool> : Perform just no-op requests, default %d\n"
1104                 " -t <bool> : Track IO latencies, default %d\n"
1105                 " -T <int>  : TSC rate in HZ\n"
1106                 " -r <int>  : Runtime in seconds, default %s\n"
1107                 " -R <bool> : Use random IO, default %d\n"
1108                 " -a <bool> : Use legacy aio, default %d\n",
1109                 argv, DEPTH, BATCH_SUBMIT, BATCH_COMPLETE, BS, polled,
1110                 fixedbufs, dma_map, register_files, nthreads, !buffered, do_nop,
1111                 stats, runtime == 0 ? "unlimited" : runtime_str, random_io, aio);
1112         exit(status);
1113 }
1114
1115 static void read_tsc_rate(void)
1116 {
1117         char buffer[32];
1118         int fd, ret;
1119
1120         if (tsc_rate)
1121                 return;
1122
1123         fd = open(TSC_RATE_FILE, O_RDONLY);
1124         if (fd < 0)
1125                 return;
1126
1127         ret = read(fd, buffer, sizeof(buffer));
1128         if (ret < 0) {
1129                 close(fd);
1130                 return;
1131         }
1132
1133         tsc_rate = strtoul(buffer, NULL, 10);
1134         printf("Using TSC rate %luHz\n", tsc_rate);
1135         close(fd);
1136 }
1137
1138 static void write_tsc_rate(void)
1139 {
1140         char buffer[32];
1141         struct stat sb;
1142         int fd, ret;
1143
1144         if (!stat(TSC_RATE_FILE, &sb))
1145                 return;
1146
1147         fd = open(TSC_RATE_FILE, O_WRONLY | O_CREAT, 0644);
1148         if (fd < 0)
1149                 return;
1150
1151         memset(buffer, 0, sizeof(buffer));
1152         sprintf(buffer, "%lu", tsc_rate);
1153         ret = write(fd, buffer, strlen(buffer));
1154         if (ret < 0)
1155                 perror("write");
1156         close(fd);
1157 }
1158
1159 int main(int argc, char *argv[])
1160 {
1161         struct submitter *s;
1162         unsigned long done, calls, reap;
1163         int err, i, j, flags, fd, opt, threads_per_f, threads_rem = 0, nfiles;
1164         long page_size;
1165         struct file f;
1166         char *fdepths;
1167         void *ret;
1168
1169         if (!do_nop && argc < 2)
1170                 usage(argv[0], 1);
1171
1172         while ((opt = getopt(argc, argv, "d:s:c:b:p:B:F:n:N:O:t:T:a:r:D:R:h?")) != -1) {
1173                 switch (opt) {
1174                 case 'a':
1175                         aio = !!atoi(optarg);
1176                         break;
1177                 case 'd':
1178                         depth = atoi(optarg);
1179                         break;
1180                 case 's':
1181                         batch_submit = atoi(optarg);
1182                         if (!batch_submit)
1183                                 batch_submit = 1;
1184                         break;
1185                 case 'c':
1186                         batch_complete = atoi(optarg);
1187                         if (!batch_complete)
1188                                 batch_complete = 1;
1189                         break;
1190                 case 'b':
1191                         bs = atoi(optarg);
1192                         break;
1193                 case 'p':
1194                         polled = !!atoi(optarg);
1195                         break;
1196                 case 'B':
1197                         fixedbufs = !!atoi(optarg);
1198                         break;
1199                 case 'F':
1200                         register_files = !!atoi(optarg);
1201                         break;
1202                 case 'n':
1203                         nthreads = atoi(optarg);
1204                         if (!nthreads) {
1205                                 printf("Threads must be non-zero\n");
1206                                 usage(argv[0], 1);
1207                         }
1208                         break;
1209                 case 'N':
1210                         do_nop = !!atoi(optarg);
1211                         break;
1212                 case 'O':
1213                         buffered = !atoi(optarg);
1214                         break;
1215                 case 't':
1216 #ifndef ARCH_HAVE_CPU_CLOCK
1217                         fprintf(stderr, "Stats not supported on this CPU\n");
1218                         return 1;
1219 #endif
1220                         stats = !!atoi(optarg);
1221                         break;
1222                 case 'T':
1223 #ifndef ARCH_HAVE_CPU_CLOCK
1224                         fprintf(stderr, "Stats not supported on this CPU\n");
1225                         return 1;
1226 #endif
1227                         tsc_rate = strtoul(optarg, NULL, 10);
1228                         write_tsc_rate();
1229                         break;
1230                 case 'r':
1231                         runtime = atoi(optarg);
1232                         break;
1233                 case 'D':
1234                         dma_map = !!atoi(optarg);
1235                         break;
1236                 case 'R':
1237                         random_io = !!atoi(optarg);
1238                         break;
1239                 case 'h':
1240                 case '?':
1241                 default:
1242                         usage(argv[0], 0);
1243                         break;
1244                 }
1245         }
1246
1247         if (stats)
1248                 read_tsc_rate();
1249
1250         if (batch_complete > depth)
1251                 batch_complete = depth;
1252         if (batch_submit > depth)
1253                 batch_submit = depth;
1254         if (!fixedbufs && dma_map)
1255                 dma_map = 0;
1256
1257         submitter = calloc(nthreads, sizeof(*submitter) +
1258                                 roundup_pow2(depth) * sizeof(struct iovec));
1259         for (j = 0; j < nthreads; j++) {
1260                 s = get_submitter(j);
1261                 s->index = j;
1262                 s->done = s->calls = s->reaps = 0;
1263         }
1264
1265         flags = O_RDONLY | O_NOATIME;
1266         if (!buffered)
1267                 flags |= O_DIRECT;
1268
1269         j = 0;
1270         i = optind;
1271         nfiles = argc - i;
1272         if (!do_nop) {
1273                 if (!nfiles) {
1274                         printf("No files specified\n");
1275                         usage(argv[0], 1);
1276                 }
1277                 threads_per_f = nthreads / nfiles;
1278                 /* make sure each thread gets assigned files */
1279                 if (threads_per_f == 0) {
1280                         threads_per_f = 1;
1281                 } else {
1282                         threads_rem = nthreads - threads_per_f * nfiles;
1283                 }
1284         }
1285         while (!do_nop && i < argc) {
1286                 int k, limit;
1287
1288                 memset(&f, 0, sizeof(f));
1289
1290                 fd = open(argv[i], flags);
1291                 if (fd < 0) {
1292                         perror("open");
1293                         return 1;
1294                 }
1295                 f.real_fd = fd;
1296                 if (get_file_size(&f)) {
1297                         printf("failed getting size of device/file\n");
1298                         return 1;
1299                 }
1300                 if (f.max_blocks <= 1) {
1301                         printf("Zero file/device size?\n");
1302                         return 1;
1303                 }
1304                 f.max_blocks--;
1305
1306                 limit = threads_per_f;
1307                 limit += threads_rem > 0 ? 1 : 0;
1308                 for (k = 0; k < limit; k++) {
1309                         s = get_submitter((j + k) % nthreads);
1310
1311                         if (s->nr_files == MAX_FDS) {
1312                                 printf("Max number of files (%d) reached\n", MAX_FDS);
1313                                 break;
1314                         }
1315
1316                         memcpy(&s->files[s->nr_files], &f, sizeof(f));
1317
1318                         printf("Added file %s (submitter %d)\n", argv[i], s->index);
1319                         s->nr_files++;
1320                 }
1321                 threads_rem--;
1322                 i++;
1323                 j += limit;
1324         }
1325
1326         arm_sig_int();
1327
1328         page_size = sysconf(_SC_PAGESIZE);
1329         if (page_size < 0)
1330                 page_size = 4096;
1331
1332         for (j = 0; j < nthreads; j++) {
1333                 s = get_submitter(j);
1334                 for (i = 0; i < roundup_pow2(depth); i++) {
1335                         void *buf;
1336
1337                         if (posix_memalign(&buf, page_size, bs)) {
1338                                 printf("failed alloc\n");
1339                                 return 1;
1340                         }
1341                         s->iovecs[i].iov_base = buf;
1342                         s->iovecs[i].iov_len = bs;
1343                 }
1344         }
1345
1346         for (j = 0; j < nthreads; j++) {
1347                 s = get_submitter(j);
1348
1349                 if (!aio)
1350                         err = setup_ring(s);
1351                 else
1352                         err = setup_aio(s);
1353                 if (err) {
1354                         printf("ring setup failed: %s, %d\n", strerror(errno), err);
1355                         return 1;
1356                 }
1357         }
1358         s = get_submitter(0);
1359         printf("polled=%d, fixedbufs=%d/%d, register_files=%d, buffered=%d, QD=%d\n", polled, fixedbufs, dma_map, register_files, buffered, depth);
1360         if (!aio)
1361                 printf("Engine=io_uring, sq_ring=%d, cq_ring=%d\n", *s->sq_ring.ring_entries, *s->cq_ring.ring_entries);
1362         else
1363                 printf("Engine=aio\n");
1364
1365         for (j = 0; j < nthreads; j++) {
1366                 s = get_submitter(j);
1367                 if (!aio)
1368                         pthread_create(&s->thread, NULL, submitter_uring_fn, s);
1369 #ifdef CONFIG_LIBAIO
1370                 else
1371                         pthread_create(&s->thread, NULL, submitter_aio_fn, s);
1372 #endif
1373         }
1374
1375         fdepths = malloc(8 * s->nr_files * nthreads);
1376         reap = calls = done = 0;
1377         do {
1378                 unsigned long this_done = 0;
1379                 unsigned long this_reap = 0;
1380                 unsigned long this_call = 0;
1381                 unsigned long rpc = 0, ipc = 0;
1382                 unsigned long iops, bw;
1383
1384                 sleep(1);
1385                 if (runtime && !--runtime)
1386                         do_finish("timeout");
1387
1388                 /* don't print partial run, if interrupted by signal */
1389                 if (finish)
1390                         break;
1391
1392                 /* one second in to the run, enable stats */
1393                 if (stats)
1394                         stats_running = 1;
1395
1396                 for (j = 0; j < nthreads; j++) {
1397                         s = get_submitter(j);
1398                         this_done += s->done;
1399                         this_call += s->calls;
1400                         this_reap += s->reaps;
1401                 }
1402                 if (this_call - calls) {
1403                         rpc = (this_done - done) / (this_call - calls);
1404                         ipc = (this_reap - reap) / (this_call - calls);
1405                 } else
1406                         rpc = ipc = -1;
1407                 file_depths(fdepths);
1408                 iops = this_done - done;
1409                 if (bs > 1048576)
1410                         bw = iops * (bs / 1048576);
1411                 else
1412                         bw = iops / (1048576 / bs);
1413                 if (iops > 100000)
1414                         printf("IOPS=%luK, ", iops / 1000);
1415                 else
1416                         printf("IOPS=%lu, ", iops);
1417                 max_iops = max(max_iops, iops);
1418                 if (!do_nop)
1419                         printf("BW=%luMiB/s, ", bw);
1420                 printf("IOS/call=%ld/%ld, inflight=(%s)\n", rpc, ipc, fdepths);
1421                 done = this_done;
1422                 calls = this_call;
1423                 reap = this_reap;
1424         } while (!finish);
1425
1426         for (j = 0; j < nthreads; j++) {
1427                 s = get_submitter(j);
1428                 pthread_join(s->thread, &ret);
1429                 close(s->ring_fd);
1430
1431                 if (stats) {
1432                         unsigned long nr;
1433
1434                         printf("%d: Latency percentiles:\n", s->tid);
1435                         for (i = 0, nr = 0; i < PLAT_NR; i++)
1436                                 nr += s->plat[i];
1437                         show_clat_percentiles(s->plat, nr, 4);
1438                         free(s->clock_batch);
1439                         free(s->plat);
1440                 }
1441         }
1442
1443         free(fdepths);
1444         free(submitter);
1445         return 0;
1446 }