t/io_uring.c

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