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