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