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