Merge branch 'master' of https://github.com/celestinechen/fio
[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
9 #include <sys/types.h>
10 #include <sys/stat.h>
11 #include <sys/ioctl.h>
12 #include <sys/syscall.h>
13 #include <sys/resource.h>
14 #include <sys/mman.h>
15 #include <sys/uio.h>
16 #include <linux/fs.h>
17 #include <fcntl.h>
18 #include <unistd.h>
19 #include <string.h>
20 #include <pthread.h>
21 #include <sched.h>
22
23 #include "../arch/arch.h"
24 #include "../lib/types.h"
25 #include "../os/linux/io_uring.h"
26
27 #define min(a, b)               ((a < b) ? (a) : (b))
28
29 struct io_sq_ring {
30         unsigned *head;
31         unsigned *tail;
32         unsigned *ring_mask;
33         unsigned *ring_entries;
34         unsigned *flags;
35         unsigned *array;
36 };
37
38 struct io_cq_ring {
39         unsigned *head;
40         unsigned *tail;
41         unsigned *ring_mask;
42         unsigned *ring_entries;
43         struct io_uring_cqe *cqes;
44 };
45
46 #define DEPTH                   128
47 #define BATCH_SUBMIT            32
48 #define BATCH_COMPLETE          32
49
50 #define BS                      4096
51
52 #define MAX_FDS                 16
53
54 static unsigned sq_ring_mask, cq_ring_mask;
55
56 struct file {
57         unsigned long max_blocks;
58         unsigned pending_ios;
59         int real_fd;
60         int fixed_fd;
61 };
62
63 struct submitter {
64         pthread_t thread;
65         int ring_fd;
66         struct io_sq_ring sq_ring;
67         struct io_uring_sqe *sqes;
68         struct io_cq_ring cq_ring;
69         int inflight;
70         unsigned long reaps;
71         unsigned long done;
72         unsigned long calls;
73         volatile int finish;
74
75         __s32 *fds;
76
77         struct file files[MAX_FDS];
78         unsigned nr_files;
79         unsigned cur_file;
80         struct iovec iovecs[];
81 };
82
83 static struct submitter *submitter;
84 static volatile int finish;
85
86 static int depth = DEPTH;
87 static int batch_submit = BATCH_SUBMIT;
88 static int batch_complete = BATCH_COMPLETE;
89 static int polled = 1;          /* use IO polling */
90 static int fixedbufs = 1;       /* use fixed user buffers */
91 static int register_files = 1;  /* use fixed files */
92 static int buffered = 0;        /* use buffered IO, not O_DIRECT */
93 static int sq_thread_poll = 0;  /* use kernel submission/poller thread */
94 static int sq_thread_cpu = -1;  /* pin above thread to this CPU */
95 static int do_nop = 0;          /* no-op SQ ring commands */
96
97 static int io_uring_register_buffers(struct submitter *s)
98 {
99         if (do_nop)
100                 return 0;
101
102         return syscall(__NR_io_uring_register, s->ring_fd,
103                         IORING_REGISTER_BUFFERS, s->iovecs, depth);
104 }
105
106 static int io_uring_register_files(struct submitter *s)
107 {
108         int i;
109
110         if (do_nop)
111                 return 0;
112
113         s->fds = calloc(s->nr_files, sizeof(__s32));
114         for (i = 0; i < s->nr_files; i++) {
115                 s->fds[i] = s->files[i].real_fd;
116                 s->files[i].fixed_fd = i;
117         }
118
119         return syscall(__NR_io_uring_register, s->ring_fd,
120                         IORING_REGISTER_FILES, s->fds, s->nr_files);
121 }
122
123 static int io_uring_setup(unsigned entries, struct io_uring_params *p)
124 {
125         return syscall(__NR_io_uring_setup, entries, p);
126 }
127
128 static int io_uring_enter(struct submitter *s, unsigned int to_submit,
129                           unsigned int min_complete, unsigned int flags)
130 {
131         return syscall(__NR_io_uring_enter, s->ring_fd, to_submit, min_complete,
132                         flags, NULL, 0);
133 }
134
135 #ifndef CONFIG_HAVE_GETTID
136 static int gettid(void)
137 {
138         return syscall(__NR_gettid);
139 }
140 #endif
141
142 static unsigned file_depth(struct submitter *s)
143 {
144         return (depth + s->nr_files - 1) / s->nr_files;
145 }
146
147 static void init_io(struct submitter *s, unsigned index)
148 {
149         struct io_uring_sqe *sqe = &s->sqes[index];
150         unsigned long offset;
151         struct file *f;
152         long r;
153
154         if (do_nop) {
155                 sqe->opcode = IORING_OP_NOP;
156                 return;
157         }
158
159         if (s->nr_files == 1) {
160                 f = &s->files[0];
161         } else {
162                 f = &s->files[s->cur_file];
163                 if (f->pending_ios >= file_depth(s)) {
164                         s->cur_file++;
165                         if (s->cur_file == s->nr_files)
166                                 s->cur_file = 0;
167                         f = &s->files[s->cur_file];
168                 }
169         }
170         f->pending_ios++;
171
172         r = lrand48();
173         offset = (r % (f->max_blocks - 1)) * BS;
174
175         if (register_files) {
176                 sqe->flags = IOSQE_FIXED_FILE;
177                 sqe->fd = f->fixed_fd;
178         } else {
179                 sqe->flags = 0;
180                 sqe->fd = f->real_fd;
181         }
182         if (fixedbufs) {
183                 sqe->opcode = IORING_OP_READ_FIXED;
184                 sqe->addr = (unsigned long) s->iovecs[index].iov_base;
185                 sqe->len = BS;
186                 sqe->buf_index = index;
187         } else {
188                 sqe->opcode = IORING_OP_READV;
189                 sqe->addr = (unsigned long) &s->iovecs[index];
190                 sqe->len = 1;
191                 sqe->buf_index = 0;
192         }
193         sqe->ioprio = 0;
194         sqe->off = offset;
195         sqe->user_data = (unsigned long) f;
196 }
197
198 static int prep_more_ios(struct submitter *s, int max_ios)
199 {
200         struct io_sq_ring *ring = &s->sq_ring;
201         unsigned index, tail, next_tail, prepped = 0;
202
203         next_tail = tail = *ring->tail;
204         do {
205                 next_tail++;
206                 read_barrier();
207                 if (next_tail == *ring->head)
208                         break;
209
210                 index = tail & sq_ring_mask;
211                 init_io(s, index);
212                 ring->array[index] = index;
213                 prepped++;
214                 tail = next_tail;
215         } while (prepped < max_ios);
216
217         if (*ring->tail != tail) {
218                 *ring->tail = tail;
219                 write_barrier();
220         }
221         return prepped;
222 }
223
224 static int get_file_size(struct file *f)
225 {
226         struct stat st;
227
228         if (fstat(f->real_fd, &st) < 0)
229                 return -1;
230         if (S_ISBLK(st.st_mode)) {
231                 unsigned long long bytes;
232
233                 if (ioctl(f->real_fd, BLKGETSIZE64, &bytes) != 0)
234                         return -1;
235
236                 f->max_blocks = bytes / BS;
237                 return 0;
238         } else if (S_ISREG(st.st_mode)) {
239                 f->max_blocks = st.st_size / BS;
240                 return 0;
241         }
242
243         return -1;
244 }
245
246 static int reap_events(struct submitter *s)
247 {
248         struct io_cq_ring *ring = &s->cq_ring;
249         struct io_uring_cqe *cqe;
250         unsigned head, reaped = 0;
251
252         head = *ring->head;
253         do {
254                 struct file *f;
255
256                 read_barrier();
257                 if (head == *ring->tail)
258                         break;
259                 cqe = &ring->cqes[head & cq_ring_mask];
260                 if (!do_nop) {
261                         f = (struct file *) (uintptr_t) cqe->user_data;
262                         f->pending_ios--;
263                         if (cqe->res != BS) {
264                                 printf("io: unexpected ret=%d\n", cqe->res);
265                                 if (polled && cqe->res == -EOPNOTSUPP)
266                                         printf("Your filesystem/driver/kernel doesn't support polled IO\n");
267                                 return -1;
268                         }
269                 }
270                 reaped++;
271                 head++;
272         } while (1);
273
274         s->inflight -= reaped;
275         *ring->head = head;
276         write_barrier();
277         return reaped;
278 }
279
280 static void *submitter_fn(void *data)
281 {
282         struct submitter *s = data;
283         struct io_sq_ring *ring = &s->sq_ring;
284         int ret, prepped;
285
286         printf("submitter=%d\n", gettid());
287
288         srand48(pthread_self());
289
290         prepped = 0;
291         do {
292                 int to_wait, to_submit, this_reap, to_prep;
293
294                 if (!prepped && s->inflight < depth) {
295                         to_prep = min(depth - s->inflight, batch_submit);
296                         prepped = prep_more_ios(s, to_prep);
297                 }
298                 s->inflight += prepped;
299 submit_more:
300                 to_submit = prepped;
301 submit:
302                 if (to_submit && (s->inflight + to_submit <= depth))
303                         to_wait = 0;
304                 else
305                         to_wait = min(s->inflight + to_submit, batch_complete);
306
307                 /*
308                  * Only need to call io_uring_enter if we're not using SQ thread
309                  * poll, or if IORING_SQ_NEED_WAKEUP is set.
310                  */
311                 if (!sq_thread_poll || (*ring->flags & IORING_SQ_NEED_WAKEUP)) {
312                         unsigned flags = 0;
313
314                         if (to_wait)
315                                 flags = IORING_ENTER_GETEVENTS;
316                         if ((*ring->flags & IORING_SQ_NEED_WAKEUP))
317                                 flags |= IORING_ENTER_SQ_WAKEUP;
318                         ret = io_uring_enter(s, to_submit, to_wait, flags);
319                         s->calls++;
320                 }
321
322                 /*
323                  * For non SQ thread poll, we already got the events we needed
324                  * through the io_uring_enter() above. For SQ thread poll, we
325                  * need to loop here until we find enough events.
326                  */
327                 this_reap = 0;
328                 do {
329                         int r;
330                         r = reap_events(s);
331                         if (r == -1) {
332                                 s->finish = 1;
333                                 break;
334                         } else if (r > 0)
335                                 this_reap += r;
336                 } while (sq_thread_poll && this_reap < to_wait);
337                 s->reaps += this_reap;
338
339                 if (ret >= 0) {
340                         if (!ret) {
341                                 to_submit = 0;
342                                 if (s->inflight)
343                                         goto submit;
344                                 continue;
345                         } else if (ret < to_submit) {
346                                 int diff = to_submit - ret;
347
348                                 s->done += ret;
349                                 prepped -= diff;
350                                 goto submit_more;
351                         }
352                         s->done += ret;
353                         prepped = 0;
354                         continue;
355                 } else if (ret < 0) {
356                         if (errno == EAGAIN) {
357                                 if (s->finish)
358                                         break;
359                                 if (this_reap)
360                                         goto submit;
361                                 to_submit = 0;
362                                 goto submit;
363                         }
364                         printf("io_submit: %s\n", strerror(errno));
365                         break;
366                 }
367         } while (!s->finish);
368
369         finish = 1;
370         return NULL;
371 }
372
373 static void sig_int(int sig)
374 {
375         printf("Exiting on signal %d\n", sig);
376         submitter->finish = 1;
377         finish = 1;
378 }
379
380 static void arm_sig_int(void)
381 {
382         struct sigaction act;
383
384         memset(&act, 0, sizeof(act));
385         act.sa_handler = sig_int;
386         act.sa_flags = SA_RESTART;
387         sigaction(SIGINT, &act, NULL);
388 }
389
390 static int setup_ring(struct submitter *s)
391 {
392         struct io_sq_ring *sring = &s->sq_ring;
393         struct io_cq_ring *cring = &s->cq_ring;
394         struct io_uring_params p;
395         int ret, fd;
396         void *ptr;
397
398         memset(&p, 0, sizeof(p));
399
400         if (polled && !do_nop)
401                 p.flags |= IORING_SETUP_IOPOLL;
402         if (sq_thread_poll) {
403                 p.flags |= IORING_SETUP_SQPOLL;
404                 if (sq_thread_cpu != -1) {
405                         p.flags |= IORING_SETUP_SQ_AFF;
406                         p.sq_thread_cpu = sq_thread_cpu;
407                 }
408         }
409
410         fd = io_uring_setup(depth, &p);
411         if (fd < 0) {
412                 perror("io_uring_setup");
413                 return 1;
414         }
415         s->ring_fd = fd;
416
417         if (fixedbufs) {
418                 ret = io_uring_register_buffers(s);
419                 if (ret < 0) {
420                         perror("io_uring_register_buffers");
421                         return 1;
422                 }
423         }
424
425         if (register_files) {
426                 ret = io_uring_register_files(s);
427                 if (ret < 0) {
428                         perror("io_uring_register_files");
429                         return 1;
430                 }
431         }
432
433         ptr = mmap(0, p.sq_off.array + p.sq_entries * sizeof(__u32),
434                         PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd,
435                         IORING_OFF_SQ_RING);
436         printf("sq_ring ptr = 0x%p\n", ptr);
437         sring->head = ptr + p.sq_off.head;
438         sring->tail = ptr + p.sq_off.tail;
439         sring->ring_mask = ptr + p.sq_off.ring_mask;
440         sring->ring_entries = ptr + p.sq_off.ring_entries;
441         sring->flags = ptr + p.sq_off.flags;
442         sring->array = ptr + p.sq_off.array;
443         sq_ring_mask = *sring->ring_mask;
444
445         s->sqes = mmap(0, p.sq_entries * sizeof(struct io_uring_sqe),
446                         PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd,
447                         IORING_OFF_SQES);
448         printf("sqes ptr    = 0x%p\n", s->sqes);
449
450         ptr = mmap(0, p.cq_off.cqes + p.cq_entries * sizeof(struct io_uring_cqe),
451                         PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd,
452                         IORING_OFF_CQ_RING);
453         printf("cq_ring ptr = 0x%p\n", ptr);
454         cring->head = ptr + p.cq_off.head;
455         cring->tail = ptr + p.cq_off.tail;
456         cring->ring_mask = ptr + p.cq_off.ring_mask;
457         cring->ring_entries = ptr + p.cq_off.ring_entries;
458         cring->cqes = ptr + p.cq_off.cqes;
459         cq_ring_mask = *cring->ring_mask;
460         return 0;
461 }
462
463 static void file_depths(char *buf)
464 {
465         struct submitter *s = submitter;
466         char *p;
467         int i;
468
469         buf[0] = '\0';
470         p = buf;
471         for (i = 0; i < s->nr_files; i++) {
472                 struct file *f = &s->files[i];
473
474                 if (i + 1 == s->nr_files)
475                         p += sprintf(p, "%d", f->pending_ios);
476                 else
477                         p += sprintf(p, "%d, ", f->pending_ios);
478         }
479 }
480
481 static void usage(char *argv)
482 {
483         printf("%s [options] -- [filenames]\n"
484                 " -d <int> : IO Depth, default %d\n"
485                 " -s <int> : Batch submit, default %d\n"
486                 " -c <int> : Batch complete, default %d\n",
487                 argv, DEPTH, BATCH_SUBMIT, BATCH_COMPLETE);
488         exit(0);
489 }
490
491 int main(int argc, char *argv[])
492 {
493         struct submitter *s;
494         unsigned long done, calls, reap;
495         int err, i, flags, fd, opt;
496         char *fdepths;
497         void *ret;
498
499         if (!do_nop && argc < 2) {
500                 printf("%s: filename [options]\n", argv[0]);
501                 return 1;
502         }
503
504         while ((opt = getopt(argc, argv, "d:s:c:h?")) != -1) {
505                 switch (opt) {
506                 case 'd':
507                         depth = atoi(optarg);
508                         break;
509                 case 's':
510                         batch_submit = atoi(optarg);
511                         break;
512                 case 'c':
513                         batch_complete = atoi(optarg);
514                         break;
515                 case 'h':
516                 case '?':
517                 default:
518                         usage(argv[0]);
519                         break;
520                 }
521         }
522
523         submitter = malloc(sizeof(*submitter) + depth * sizeof(struct iovec));
524         memset(submitter, 0, sizeof(*submitter) + depth * sizeof(struct iovec));
525         s = submitter;
526
527         flags = O_RDONLY | O_NOATIME;
528         if (!buffered)
529                 flags |= O_DIRECT;
530
531         i = optind;
532         while (!do_nop && i < argc) {
533                 struct file *f;
534
535                 if (s->nr_files == MAX_FDS) {
536                         printf("Max number of files (%d) reached\n", MAX_FDS);
537                         break;
538                 }
539                 fd = open(argv[i], flags);
540                 if (fd < 0) {
541                         perror("open");
542                         return 1;
543                 }
544
545                 f = &s->files[s->nr_files];
546                 f->real_fd = fd;
547                 if (get_file_size(f)) {
548                         printf("failed getting size of device/file\n");
549                         return 1;
550                 }
551                 if (f->max_blocks <= 1) {
552                         printf("Zero file/device size?\n");
553                         return 1;
554                 }
555                 f->max_blocks--;
556
557                 printf("Added file %s\n", argv[i]);
558                 s->nr_files++;
559                 i++;
560         }
561
562         if (fixedbufs) {
563                 struct rlimit rlim;
564
565                 rlim.rlim_cur = RLIM_INFINITY;
566                 rlim.rlim_max = RLIM_INFINITY;
567                 if (setrlimit(RLIMIT_MEMLOCK, &rlim) < 0) {
568                         perror("setrlimit");
569                         return 1;
570                 }
571         }
572
573         arm_sig_int();
574
575         for (i = 0; i < depth; i++) {
576                 void *buf;
577
578                 if (posix_memalign(&buf, BS, BS)) {
579                         printf("failed alloc\n");
580                         return 1;
581                 }
582                 s->iovecs[i].iov_base = buf;
583                 s->iovecs[i].iov_len = BS;
584         }
585
586         err = setup_ring(s);
587         if (err) {
588                 printf("ring setup failed: %s, %d\n", strerror(errno), err);
589                 return 1;
590         }
591         printf("polled=%d, fixedbufs=%d, buffered=%d", polled, fixedbufs, buffered);
592         printf(" QD=%d, sq_ring=%d, cq_ring=%d\n", depth, *s->sq_ring.ring_entries, *s->cq_ring.ring_entries);
593
594         pthread_create(&s->thread, NULL, submitter_fn, s);
595
596         fdepths = malloc(8 * s->nr_files);
597         reap = calls = done = 0;
598         do {
599                 unsigned long this_done = 0;
600                 unsigned long this_reap = 0;
601                 unsigned long this_call = 0;
602                 unsigned long rpc = 0, ipc = 0;
603
604                 sleep(1);
605                 this_done += s->done;
606                 this_call += s->calls;
607                 this_reap += s->reaps;
608                 if (this_call - calls) {
609                         rpc = (this_done - done) / (this_call - calls);
610                         ipc = (this_reap - reap) / (this_call - calls);
611                 } else
612                         rpc = ipc = -1;
613                 file_depths(fdepths);
614                 printf("IOPS=%lu, IOS/call=%ld/%ld, inflight=%u (%s)\n",
615                                 this_done - done, rpc, ipc, s->inflight,
616                                 fdepths);
617                 done = this_done;
618                 calls = this_call;
619                 reap = this_reap;
620         } while (!finish);
621
622         pthread_join(s->thread, &ret);
623         close(s->ring_fd);
624         free(fdepths);
625         return 0;
626 }