blkiotest.c

/*
 * Copyright (c) 2016--2021  Wu, Xingbo <wuxb45@gmail.com>
 *
 * All rights reserved. No warranty, explicit or implicit, provided.
 */
#define _GNU_SOURCE

#include "lib.h"
#include "blkio.h"
#include "ctypes.h"

#include <fcntl.h>
#include <sys/ioctl.h>
#if defined(__linux__)
#include <linux/fs.h>
#endif

#if defined(SPDK)
#include <spdk/stdinc.h>
#include <spdk/nvme.h>
#include <spdk/env.h>
#include <spdk/log.h>
#include <rte_log.h>
#endif // SPDK

#ifndef O_DIRECT
#define O_DIRECT 0
#endif

// rcache {{{
#define RC_BATCH ((1000lu))

static au64 rcache_nr_op = 0;
  static void *
rcache_worker(void * ptr)
{
  srandom_u64(time_nsec());
  u64 nr_op = 0;
  struct rcache * const c = (typeof(c))ptr;
  u64 * ptrs[16] = {};
  char fn[64];
  for (u64 e = 0; e < RC_BATCH; e++) {
    const u64 fid = random_u64() & 0x1fflu;
    sprintf(fn, "/tmp/rc-%lu.txt", fid);
    int fd = open(fn, O_RDONLY);
    if (fd < 0)
      debug_die();
    for (u64 i = 0; i < RC_BATCH; i++) {
      const u64 x = random_u64() & 0xflu;
      if (ptrs[x]) {
        //usleep(1);
        if ((i & 0xflu) == 0) {
          rcache_retain(c, (const u8 *)(ptrs[x]));
          rcache_release(c, (const u8 *)(ptrs[x]));
        }
        rcache_release(c, (const u8 *)(ptrs[x]));
        ptrs[x] = NULL;
      } else {
        const u32 r = (u32)(random_u64() & 0x1fflu);
        ptrs[x] = rcache_acquire(c, fd, r);
        if (ptrs[x][0] != fid || ptrs[x][1] != r) {
          debug_die();
        }
        nr_op++;
      }
    }
    for (u64 i = 0; i < 16; i++) {
      if (ptrs[i]) {
        rcache_release(c, (const u8 *)(ptrs[i]));
        ptrs[i] = NULL;
      }
    }
    rcache_close(c, fd);
  }
  rcache_nr_op += nr_op;
  return NULL;
}

  static void
test_rcache(void)
{
  char * const mem = calloc(1, PGSZ);
  // 4k x 512 x 512 (1GB total)
  for (u64 i = 0; i < 512; i++) {
    sprintf(mem, "/tmp/rc-%lu.txt", i);
    int fd = open(mem, O_WRONLY|O_CREAT, 00644);
    for (u64 j = 0; j < 512; j++) {
      *(u64*)mem = i;
      *(u64*)(mem+8) = j;
      write(fd, mem, PGSZ);
    }
    close(fd);
  }
  struct rcache * const c = rcache_create(4, 4);
  debug_perf_switch();
  thread_fork_join(4, rcache_worker, false, c);
  for (u32 t = 1; t < 8; t++) {
    rcache_nr_op = 0;
    const u64 dt = thread_fork_join(t, rcache_worker, false, c);
    const double ops = (double)(rcache_nr_op) / ((double)dt) * 1e9;
    printf("nr_th %u dt-ms %lu; %.3lf op/s\n", t, dt / (u64)1e6, ops);
    debug_perf_switch();
  }
  rcache_destroy(c);

  for (u64 i = 0; i < 512; i++) { // clean up
    sprintf(mem, "/tmp/rc-%lu.txt", i);
    unlink(mem);
  }
  free(mem);
}
// }}} rcache

// rwcache {{{
#define RC2_NF ((512))
#define RC2_NP ((512))
#define RC2_NWORKER ((4))
  static void *
rwcache_worker(void * ptr)
{
  srandom_u64(time_nsec());
  struct rwcache * const c = (typeof(c))ptr;
  u64 * ptrs[16] = {};
  char fn[64];
  u64 nr_op = 0;
  for (u64 e = 0; e < RC_BATCH; e++) {
    const u64 fid = random_u64() % RC2_NF;
    sprintf(fn, "/tmp/rc-%lu.txt", fid);
    int fd = open(fn, O_RDONLY);
    if (fd < 0)
      debug_die();
    for (u64 i = 0; i < RC_BATCH; i++) {
      const u64 x = random_u64() % 16;
      if (ptrs[x]) {
        //usleep(1);
        rwcache_release(c, ptrs[x]);
        ptrs[x] = NULL;
      } else {
        const u32 r = (u32)(random_u64() % RC2_NP);
        ptrs[x] = rwcache_acquire(c, fd, r);
        if (ptrs[x][0] != fid || ptrs[x][1] != r) {
          debug_die();
        }
        nr_op++;
      }
    }
    for (u64 i = 0; i < 16; i++) {
      if (ptrs[i]) {
        rwcache_release(c, ptrs[i]);
        ptrs[i] = NULL;
      }
    }
    rwcache_close(c, fd);
  }
  rcache_nr_op += nr_op;
  return NULL;
}

static au64 rwcache_seq = 0;
  static void *
rwcache_worker2(void * ptr)
{
  struct rwcache * const c = (typeof(c))ptr;
  char fn[64];
  const u64 seq = atomic_fetch_add(&rwcache_seq, 1);
  for (u64 e = 0; e < 1000; e++) {
    if (!seq) {
      fprintf(stderr, "%lu\n", e);
    }
    for (u64 i = seq; i < RC2_NF; i += RC2_NWORKER) {
      sprintf(fn, "/tmp/rc-%lu.txt", i);
      int fd = open(fn, O_RDWR); // RW
      if (fd < 0)
        debug_die();
      for (u32 k = 0; k < RC2_NP; k++) {
        u64 * const mem = rwcache_acquire(c, fd, k);
        if (mem[0] != i || mem[1] != k || mem[2] != e) {
          printf("e %lu i %lu k %u -- in file %lu %lu %lu\n", e, i, k, mem[2], mem[0], mem[1]);
          debug_die();
        }
        mem[2] = e + 1;
        rwcache_dirty(c, mem);
        if (random_u64() & 0x800)
          rwcache_sync(c, mem);
        rwcache_release(c, mem);
      }
      rwcache_close(c, fd);
    }
  }
  return NULL;
}

  static void
test_rwcache(void)
{
  char * const mem = calloc(1, PGSZ);
  // 4k x 512 x 512 (1GB total)
  for (u64 i = 0; i < RC2_NF; i++) {
    sprintf(mem, "/tmp/rc-%lu.txt", i);
    int fd = open(mem, O_WRONLY|O_CREAT, 00644);
    for (u64 j = 0; j < RC2_NP; j++) {
      ((u64*)mem)[0] = i;
      ((u64*)mem)[1] = j;
      write(fd, mem, PGSZ);
    }
    close(fd);
  }
  free(mem);
  struct rwcache * const c = rwcache_create(1, 12);
  thread_fork_join(4, rwcache_worker, false, c);

  for (u32 t = 1; t < 8; t++) {
    rcache_nr_op = 0;
    const u64 dt = thread_fork_join(t, rwcache_worker, false, c);
    const double ops = (double)(rcache_nr_op) / ((double)dt) * 1000000000lu;
    printf("nr_th %u dt %lu; %.3lf op/s\n", t, dt / (u64)1e6, ops);
  }

  // update
  const u64 dtw = thread_fork_join(RC2_NWORKER, rwcache_worker2, false, c);
  printf("update %lu\n", dtw);
  rwcache_destroy(c);
}
// }}} rwcache

#if defined(LIBURING)

// coqrw {{{
struct coqrw_worker_info {
  u64 bufsize;
  u64 rand_mask;
  u64 end_time;
  u64 cpt;

  int fd;
  au64 seq; // reader if atomic_fetch_add(1) < nr_r
  u64 nr_r;
  au64 io_r; // stat, output
  au64 io_w; // stat, output
};

struct coqrw_co_info {
  struct coqrw_worker_info * wi;
  struct coq * coq;
  struct io_uring * ring;
};

  static void
coqrw_co_worker(void)
{
  struct coqrw_co_info * const ci = (typeof(ci))co_priv();
  struct coqrw_worker_info * const wi = ci->wi;
  printf("%s %p\n", __func__, ci);
  const bool is_reader = atomic_fetch_add(&(wi->seq), 1) < wi->nr_r;
  const int fd = wi->fd;
  u8 * const buf = xalloc(512, wi->bufsize);
  u64 succ = 0;
  u64 err = 0;
  do {
    for (u64 i = 0; i < 100; i++) {
      const u64 blknum = random_u64() & wi->rand_mask;
      const u64 offset = blknum * wi->bufsize;
      const ssize_t ret = is_reader ?
        coq_pread_uring(ci->coq, ci->ring, fd, buf, wi->bufsize, (off_t)offset):
        coq_pwrite_uring(ci->coq, ci->ring, fd, buf, wi->bufsize, (off_t)offset);

      if (ret < 0)
        err++;
      else
        succ++;
    }
  } while (time_nsec() < wi->end_time);
  printf("%s succ %lu err %lu\n", __func__, succ, err);
  atomic_fetch_add(is_reader ? (&(wi->io_r)) : (&(wi->io_w)), succ);
}

  static void *
coqrw_th_worker(void * const ptr)
{
  struct coqrw_worker_info * const wi = (typeof(wi))ptr;
  srandom_u64(time_nsec());
  struct coqrw_co_info ci;
  ci.coq = coq_create();
  ci.ring = coq_uring_create(64);
  ci.wi = wi;
  u64 hostrsp = 0;

  for (u64 i = 0; i < wi->cpt; i++) {
    struct co * const co = co_create(8192*8, coqrw_co_worker, &ci, &hostrsp);
    corq_enqueue(ci.coq, co);
  }
  coq_run(ci.coq);
  coq_uring_destroy(ci.ring);
  coq_destroy(ci.coq);
  return NULL;
}

  static int
test_coqrw(int argc, char ** argv)
{
  if (argc < 8) {
    fprintf(stderr, "usage: %s <dev> <nr_r> <nr_w> <time> <buflen_power> <range_power> <co-per-th>\n", argv[0]);
    fprintf(stderr, "example: %s /dev/sdb 4 0 10 12 5 10 4\n", argv[0]);
    exit(1);
  }

  //const int fd = open(argv[1], O_RDWR | O_DIRECT);
  const int fd = open(argv[1], O_RDWR);
  if (fd < 0) {
    fprintf(stderr, "open %s failed\n", argv[1]);
    exit(1);
  }

  const u32 nr_r = a2u32(argv[2]);
  const u32 nr_w = a2u32(argv[3]);
  const u64 sec = a2u64(argv[4]);
  const u64 p0 = a2u64(argv[5]);
  const u64 p1 = (p0 < 12) ? 12 : p0;
  const u64 p2 = (p1 > 25) ? 25 : p1;
  const u64 bufsize = 1lu << p2;
  const u64 r0 = a2u64(argv[6]);
  const u64 r1 = (r0 < p2) ? p2 : r0;
  const u64 mask = (1lu << (r1 - p2)) - 1;
  const u64 cpt = a2u64(argv[7]);
  printf("buflen power %lu range power %lu cpt %lu\n", p2, r1, cpt);

  // warning: buf is currently shared between all readers/writers
  struct coqrw_worker_info wi = {
    .bufsize = bufsize,
    .rand_mask = mask,
    .end_time = time_nsec() + (sec * 1000000000lu),
    .cpt = cpt,
    .fd = fd,
    .seq = 0,
    .nr_r = nr_r * cpt,
    .io_r = 0,
    .io_w = 0
  };

  const u64 dt = thread_fork_join(nr_r+nr_w, coqrw_th_worker, false, &wi);
  const double miops = (double)(wi.io_r + wi.io_w) * 1e3 / ((double)dt);
  printf("million IOPS %.3lf\n", miops);
  return 0;
}
// }}} coqrw

// rawcoring {{{
// use once to open a file for testing
  static int
prep_big_file(const char * const filename, const bool dio, const u64 size)
{
  printf("Preparing %s\n", filename);
  struct stat st;
  int ret = stat(filename, &st);
  bool is_blk = (ret == 0) && S_ISBLK(st.st_mode);
  if (is_blk) {
    // blk device
    int fd = open(filename, O_RDWR|(dio?O_DIRECT:0));
    if (fd < 0) {
      printf("open blk file failed\n");
      return -1;
    }
    u64 devsize = 0;
    ioctl(fd, BLKGETSIZE64, &devsize);
    if (devsize < size) {
      printf("device size < access range\n");
      close(fd);
      return -1;
    }
    return fd;
  } else {
    // file
    int fd = open(filename, O_CREAT|O_RDWR, 00666);
    if (fd < 0) {
      printf("open regular file failed\n");
      return -1;
    }
    if ((u64)st.st_size >= size) {
      return fd;
    } else {
      printf("fallocate %ld -> %lu\n", st.st_size, size);
      ret = fallocate(fd, 0, st.st_size, (off_t)size-st.st_size);
      if (ret != 0) {
        printf("fallocate err: %d %s\n", ret, strerror(-errno));
        close(fd);
        return -1;
      }
    }
    if (dio) { // reopen for Direct IO
      close(fd);
      fd = open(filename, O_RDWR|O_DIRECT);
    }
    return fd;
  }
}

// private metadata for a worker coroutine
struct rawcoring_worker_info {
  struct io_uring * ring;
  struct co * co;
  struct iovec vec; // registered iovec
  u32 iosize;
  int vecidx; // index of the buffer
  u64 nrblk;
  u64 depth;
  u64 * pseq;
  u64 nr_err; // out
  u64 nr_op; // out
};

// the worker thread performs one query task and return
// a query task need a few I/O operations
  static void
rawcoring_worker(void)
{
  struct rawcoring_worker_info * const wi = co_priv();

  for (u64 i = 0; i < 64; i++) {
    struct io_uring_sqe * const sqe = io_uring_get_sqe(wi->ring);
    debug_assert(sqe); // may fail if queue size was initialized to be very small
    const u64 r = random_u64() % wi->nrblk;

    // fixed requires the vec index
    io_uring_prep_read_fixed(sqe, 0, wi->vec.iov_base, wi->iosize, (off_t)(r*wi->iosize), wi->vecidx);
    io_uring_sqe_set_flags(sqe, IOSQE_FIXED_FILE); // the 0 above means the index
    io_uring_sqe_set_data(sqe, wi);

    // submit the request to uring
    (*wi->pseq)++;
    if ((*wi->pseq) == wi->depth) {
      const int nsub = io_uring_submit(wi->ring); // it enters kernel on demand
      // the test programs in liburing usually don't care...
      if ((u64)nsub != wi->depth)
        debug_die();
      *wi->pseq = 0;
    }
    // wait for completion--returns to the host routine now
    // when the host routine received a cqe for the request, this coroutine will resume
    // the return value from co_back is the cqe to be consumed
    struct io_uring_cqe * const cqe = (typeof(cqe))co_back(0);

    if (cqe->res <= 0) {
      printf("cqe %p data %p res %d flags %u\n", cqe, (void *)cqe->user_data, cqe->res, cqe->flags);
      exit(-1);
    }
    // consume the data in iovec ...
    // do nothing
    io_uring_cqe_seen(wi->ring, cqe);
    wi->nr_op++;
  }
}

  static int
test_rawcoring(int argc, char ** argv)
{
  if (argc < 8) {
    printf("Usage: <file> <iosize (bytes)> <nrblk> <depth> <runtime> <sqpoll (0/1)> <dio (0/1)>\n");
    printf("<file> will be sized to <iosize> x <nrblk>\n");
    printf("kcpu will be used only in sqpoll mode; need SYSADMIN cap to run\n");
    exit(0);
  }
  const char * const filename = argv[1];
  const u64 iosize = a2u64(argv[2]);
  const u64 nrblk = a2u64(argv[3]);
  const u32 depth = a2u32(argv[4]);
  const u64 runtime = a2u64(argv[5]);
  const bool sqpoll = a2u64(argv[6]) ? true : false;
  const bool dio = a2u64(argv[7]) ? true : false;

  const int fdrw = prep_big_file(filename, dio, iosize * nrblk);
  if (fdrw < 0) {
    printf("prep-file failed\n");
    exit(-1);
  }

  struct io_uring_params p = {};
  if (sqpoll) {
    p.flags |= IORING_SETUP_SQPOLL | IORING_SETUP_SQ_AFF;
    p.sq_thread_cpu = 1; // hard-coded
  }

  struct io_uring ring = {};
  // at least doubled sqes
  const u32 qdepth = (u32)bits_p2_up_u64((depth < 4 ? 4 : depth) * 2);
  int ret = io_uring_queue_init_params(qdepth, &ring, &p);
  if (ret) {
    printf("mmap err %d\n", ret);
    exit(-1);
  }

  // register the file
  ret = io_uring_register_files(&ring, &fdrw, 1);
  if (ret) {
    printf("reg file err %d\n", ret);
    exit(-1);
  }

  // iovecs & registering
  struct iovec * vecs = malloc(sizeof(*vecs) * depth);
  for (u64 i = 0; i < depth; i++) {
    vecs[i].iov_base = xalloc(512, iosize);
    vecs[i].iov_len = iosize;
  }
  ret = io_uring_register_buffers(&ring, vecs, depth);
  if (ret) {
    printf("reg vecs err %d\n", ret);
    exit(-1);
  }

  // parameters are ready
  // start the first batch of requests
  struct rawcoring_worker_info * const wis = calloc(depth, sizeof(wis[0]));
  u64 seq = 0;
  u64 hostrsp = 0;
  const u64 t0 = time_nsec();
  const u64 endtime = t0 + (runtime * 1000000000lu);
  // init wi and fill the queue
  for (u32 i = 0; i < depth; i++) {
    struct rawcoring_worker_info * const wi = &(wis[i]);
    wi->ring = &ring;
    wi->vec = vecs[i];
    wi->vecidx = (int)i;
    wi->iosize = (u32)iosize;
    wi->nrblk = nrblk;
    wi->depth = depth;
    wi->pseq = &seq;
    wi->co = co_create(8192, rawcoring_worker, wi, &hostrsp);
    (void)co_enter(wi->co, 0); // submit job and return
  }

  // keep the queue full until time out
  do {
    for (u64 i = 0; i < 100; i++) {
      struct io_uring_cqe * cqe = NULL;
      ret = io_uring_wait_cqe(&ring, &cqe);

      // wait error
      if (ret) {
        printf("wait_cqe err %d\n", ret);
        exit(-1);
      }

      // obtain the worker
      struct rawcoring_worker_info * const wi = (typeof(wi))io_uring_cqe_get_data(cqe);
      co_enter(wi->co, (u64)cqe);
      while (!co_valid(wi->co)) {
        // submit a new task
        co_reuse(wi->co, rawcoring_worker, wi, &hostrsp);
        co_enter(wi->co, 0);
      }
    }
  } while (time_nsec() < endtime);

  // about to finish. wait for completion of all flying queries
  u64 todo = depth;
  while (todo) {
    struct io_uring_cqe * cqe = NULL;
    ret = io_uring_wait_cqe(&ring, &cqe);

    // wait error
    if (ret) {
      printf("wait_cqe err %d\n", ret);
      exit(-1);
    }

    // obtain the worker
    struct rawcoring_worker_info * const wi = (typeof(wi))io_uring_cqe_get_data(cqe);
    co_enter(wi->co, (u64)cqe);
    if (!co_valid(wi->co))
      todo--;
    // else: the query is still running
  }

  const u64 dt = time_diff_nsec(t0);

  u64 nr_op = 0;
  u64 nr_err = 0;
  for (u64 i = 0; i < depth; i++) {
    printf("%lu\n", wis[i].nr_op);
    nr_op += wis[i].nr_op;
    nr_err += wis[i].nr_err;
  }
  const double iops = ((double)nr_op) * 1000000000lu / ((double)dt);
  printf("nops %lu iops %.0lf tpt %.3lf MB/s err %lu\n", nr_op, iops, (iops * (double)iosize) / 1048576, nr_err);

  // TODO: clean up
  exit(0);
}
// }}} rawcoring

// ring {{{
struct ring_worker_info {
  int fd;
  bool iopoll;
  bool sqpoll;

  u64 nrblk;
  u32 iosize;
  u32 depth;

  u32 ucpu;
  u32 kcpu;

  u64 endtime;
  u64 nr_op;
};

  static u64
ring_wait_one(struct io_uring * const ring)
{
  struct io_uring_cqe *cqe = NULL;
  int ret = io_uring_wait_cqe(ring, &cqe);

  if (ret) {
    printf("wait_cqe err %d\n", ret);
    exit(-1);
  }

  debug_assert(cqe); // high user address
  if (cqe->res <= 0) {
    printf("cqe %p data %lu res %d flags %u\n", cqe, (u64)cqe->user_data, cqe->res, cqe->flags);
    exit(-1);
  }
  const u64 id = (u64) io_uring_cqe_get_data(cqe);
  io_uring_cqe_seen(ring, cqe);
  return id;
}

  static void *
pread_worker(void * const ptr)
{
  srandom_u64(time_nsec());
  struct ring_worker_info * const wi = (typeof(wi))ptr;
  thread_pin(wi->ucpu);

  void * const buf = xalloc(512, wi->iosize);

  u64 complete = 0;
  u64 nerr = 0;
  // keep the queue full
  do {
    for (u64 i = 0; i < 1000; i++) {
      const u64 r = random_u64() % wi->nrblk;
      if (wi->iosize != (u32)pread(wi->fd, buf, wi->iosize, (off_t)(r*wi->iosize)))
        nerr++;
      complete++;
    }
  } while (time_nsec() < wi->endtime);
  if (nerr)
    printf("nsub #err %lu\n", nerr);

  wi->nr_op = complete;
  return NULL;
}

  static void *
ring_worker(void * const ptr)
{
  srandom_u64(time_nsec());
  struct ring_worker_info * const wi = (typeof(wi))ptr;
  thread_pin(wi->ucpu);

  struct iovec * vecs = malloc(sizeof(*vecs) * wi->depth);
  for (u32 i = 0; i < wi->depth; i++) {
    vecs[i].iov_base = xalloc(512, wi->iosize);
    vecs[i].iov_len = wi->iosize;
  }

  struct io_uring_params p = {};
  // sqpoll needs CAP_SYS_ADMIN or root user
  if (wi->sqpoll) {
    p.flags |= IORING_SETUP_SQPOLL | IORING_SETUP_SQ_AFF;
    p.sq_thread_cpu = wi->kcpu;
  }
  //p.sq_thread_idle = 0; // kernel defaults to 1s
  struct io_uring ring = {};
  int ret = io_uring_queue_init_params(wi->depth*2, &ring, &p);
  if (ret) {
    printf("mmap err %d\n", ret);
    exit(-1);
  }
  ret = io_uring_register_files(&ring, &wi->fd, 1);
  if (ret) {
    printf("reg fd err %d\n", ret);
    exit(-1);
  }
  ret = io_uring_register_buffers(&ring, vecs, wi->depth);
  if (ret) {
    printf("reg vecs err %d\n", ret);
    exit(-1);
  }

  //printf("sqes %p cqes %p\n", ring.sq.sqes, ring.cq.cqes);

  // fill the queue
  for (u32 i = 0; i < wi->depth; i++) {
    struct io_uring_sqe * sqe = io_uring_get_sqe(&ring);
    debug_assert(sqe);
    const u64 r = random_u64() % wi->nrblk;
    io_uring_prep_read_fixed(sqe, 0, vecs[i].iov_base, wi->iosize, (off_t)(r*wi->iosize), (int)i);
    io_uring_sqe_set_flags(sqe, IOSQE_FIXED_FILE); // fd above replaced with the index
    io_uring_sqe_set_data(sqe, (void *)(u64)i);
  }
  const u32 nsub = (u32)io_uring_submit(&ring); // it enters kernel on demand
  if (nsub != wi->depth) {
    printf("first submit %u/%u\n", nsub, wi->depth);
    exit(-1);
  }

  u64 complete = 0;
  u64 nerr = 0;
  // keep the queue full
  do {
    for (u64 i = 0; i < 1000; i++) {
      const u64 id = ring_wait_one(&ring);
      complete++;

      struct io_uring_sqe * const sqe = io_uring_get_sqe(&ring);
      debug_assert(sqe);
      const u64 r = random_u64() % wi->nrblk;
      io_uring_prep_read_fixed(sqe, 0, vecs[id].iov_base, wi->iosize, (off_t)(r*wi->iosize), (int)id);
      io_uring_sqe_set_flags(sqe, IOSQE_FIXED_FILE);
      io_uring_sqe_set_data(sqe, (void *)id);
      const int nsub1 = io_uring_submit(&ring);
      (void)nsub1;
      // SQPOLL: occasionally has nsub1 > 1, usually many 2s and a few 3s
      if (nsub1 != 1)
        nerr++;
    }
  } while (time_nsec() < wi->endtime);

  // empty the queue
  for (u32 i = 0; i < wi->depth; i++) {
    const u64 id = ring_wait_one(&ring);
    (void)id;
    complete++;
  }
  if (nerr)
    printf("nsub #err %lu\n", nerr);

  ret = io_uring_unregister_files(&ring);
  if (ret) {
    printf("unreg fd err %d\n", ret);
    exit(-1);
  }

  ret = io_uring_unregister_buffers(&ring);
  if (ret) {
    printf("unreg vecs err %d\n", ret);
    exit(-1);
  }
  io_uring_queue_exit(&ring);
  wi->nr_op = complete;
  return NULL;
}

  static int
test_ring(int argc, char ** argv)
{
  if (argc < 9) {
    printf("Read-only I/O tests with io_uring. Can use a dev or a file path\n");
    printf("Usage: <path> <iosize (bytes)> <nrblk> <depth> <runtime> <sqpoll (0/1)> <dio (0/1)> {<ucpu> <kcpu>} ...\n");
    printf("<file> will be sized to <iosize> x <nrblk>\n");
    printf("kcpu will be used only in sqpoll mode; sqpoll needs root or CAP_SYS_ADMIN\n");
    exit(0);
  }
  const char * const filename = argv[1];
  const u32 iosize = a2u32(argv[2]);
  const u64 nrblk = a2u64(argv[3]);
  const u32 depth = a2u32(argv[4]);
  const u64 runtime = a2u64(argv[5]);
  const bool sqpoll = a2u64(argv[6]) ? true : false;
  const bool dio = a2u64(argv[7]) ? true : false;
  const u32 nrth = (u32)((argc - 8)/2);

  if (nrth == 0 || nrth >= 64) {
    printf("need 1 to 64 <ucpu> <kcpu> pairs\n");
    exit(-1);
  }

  const int fdrw = prep_big_file(filename, dio, iosize * nrblk);
  if (fdrw < 0) {
    printf("prep-file failed\n");
    exit(-1);
  }
  struct ring_worker_info wis[64];
  struct ring_worker_info * pwis[64];
  const u64 endtime = runtime * 1000000000lu + time_nsec();
  for (u32 i = 0; i < nrth; i++) {
    wis[i].fd = dup(fdrw);
    wis[i].sqpoll = sqpoll;
    wis[i].iosize = iosize;
    wis[i].nrblk = nrblk;
    wis[i].depth = depth;
    wis[i].ucpu = a2u32(argv[8+i*2]);
    wis[i].kcpu = a2u32(argv[9+i*2]);
    wis[i].endtime = endtime;
    wis[i].nr_op = 0;
    pwis[i] = &wis[i];
  }
  const u64 dt = thread_fork_join(nrth, depth ? ring_worker : pread_worker, true, pwis);
  u64 ops = 0;
  for (u32 i = 0; i < nrth; i++) {
    ops += wis[i].nr_op;
    close(wis[i].fd);
  }

  const double iops = ((double)ops) * 1000000000lu / ((double)dt);
  printf("iops %.0lf tpt %.3lf MB/s\n", iops, (iops * (double)iosize) / 1048576);
  exit(0);
}
// }}} ring

// coqread {{{
struct coqread_worker_info {
  struct coq * coq;
  struct io_uring * ring;
  int fd;
  u64 mask;
};

  static void
coq_worker(void)
{
  struct coqread_worker_info * const wi = (typeof(wi))co_priv();
  const u64 x = 100000 + (random_u64() & 0xff);
  const double t0 = time_sec();
  u8 * const buf = xalloc(PGSZ, PGSZ);
  for (u64 i = 0; i < x; i++) {
    const u64 r = random_u64() & wi->mask;
    const ssize_t n = coq_pread_uring(wi->coq, wi->ring, wi->fd, buf, PGSZ, (off_t)(r * PGSZ));
    if (n != PGSZ) {
      printf("%ld != PGSZ\n", (s64)n);
      exit(0);
    }
  }
  const double dt = time_diff_sec(t0);
  printf("done %p nop %lu iops %.3lf\n", co_self(), x, (double)x / dt);
}

  static void
test_coqread(int argc, char ** argv)
{
  srandom_u64(time_nsec());
  if (argc < 5) {
    printf("Usage: <existig-file/dev> <dio(0|1)> <nr-coroutines> <power>\n");
    exit(0);
  }

  char * const path = argv[1];
  const bool dio = argv[2][0] == '1';
  const u64 nt = a2u64(argv[3]);
  const u64 power = a2u64(argv[4]);

  const int fd = open(path, O_RDONLY | (dio ? O_DIRECT : 0), 00666);
  if (fd < 0) {
    printf("open failed\n");
    exit(0);
  }

  // create separate coq and ring
  //struct coq * const coq = coq_create();
  //struct io_uring * const ring = coq_uring_create(64);
  //struct coqread_worker_info wi = {.coq = coq, .ring = ring, .fd = fd, .mask = (1lu << power) - 1, };

  // create bundled coq+uring
  struct coq * const coq = coq_uring_create_pair(64);
  struct coqread_worker_info wi = {.coq = coq, .ring = NULL, .fd = fd, .mask = (1lu << power) - 1, };
  u64 hostrsp = 0;
  for (u64 i = 0; i < nt; i++) {
    struct co * const co = co_create(PGSZ * 7, coq_worker, &wi, &hostrsp);
    corq_enqueue(coq, co);
  }

  coq_run(coq);
  close(fd);
}
// }}} coqread

#endif // LIBURING

// wring {{{
  static int
test_wring(int argc, char ** argv)
{
  if (argc < 3) {
    fprintf(stderr, "usage: <path> <direct?>\n");
    return 0;
  }
  int flags = O_CREAT| O_RDWR;
  if (argv[2][0] == '1')
    flags |= O_DIRECT;
  int fd = open(argv[1], flags, 00644);
  debug_assert(fd >= 0);

#define WRING_DP ((64))
  char zeroes[PGSZ] = {};
  for (u32 i = 0; i < WRING_DP; i++)
    write(fd, zeroes, PGSZ);
  struct wring * const wring = wring_create(fd, PGSZ, WRING_DP);
  wring_fsync(wring);
  for (u64 i = 0; i < WRING_DP; i++) {
    u8 * const buf = wring_acquire(wring);
    buf[0] = (u8)i;
    wring_write(wring, (off_t)(i*PGSZ), buf);
  }
  wring_fsync(wring);
  wring_flush(wring);
  u8 * buffer = xalloc(PGSZ, PGSZ);
  for (u64 i = 0; i < WRING_DP; i++) {
    pread(fd, buffer, 1, (off_t)(i*PGSZ));
    if (buffer[0] != i)
      printf("err!\n");
  }
#define WRING_NR ((8192))
  for (u32 i = 0; i < WRING_DP; i++)
    write(fd, zeroes, PGSZ);
  wring_fsync(wring);
  for (u64 e = 0; e < 3; e++) {
    u64 t0, dt;
    t0 = time_nsec();
    for (u64 i = 0; i < WRING_NR; i++) {
      u8 * const buf = wring_acquire(wring);
      buf[0]=(u8)i;
      wring_write(wring, (off_t)(i*PGSZ), buf);
      //usleep(1);
    }
    dt = time_diff_nsec(t0);
    wring_fsync(wring);
    wring_flush(wring);
    printf("ns/op %lu\n", dt/WRING_NR);

    t0 = time_nsec();
    for (u64 i = 0; i < WRING_NR; i++) {
      buffer[0]=(u8)i;
      pwrite(fd, buffer, PGSZ, (off_t)(i*PGSZ));
      //usleep(1);
    }
    dt = time_diff_nsec(t0);
    wring_fsync(wring);
    printf("ns/op %lu\n", dt/WRING_NR);
  }

  free(buffer);
  wring_flush(wring);
  close(fd);
  wring_destroy(wring);
  return 0;
}
// }}} wring

#if defined(SPDK)

// nvme {{{
  static void
coq_process_nvme_cb(void * const priv, const struct spdk_nvme_cpl * const cpl)
{
  struct co * const co = (typeof(co))priv;
  debug_assert(co);
  const u64 res = spdk_nvme_cpl_is_error(cpl) ? UINT64_MAX : 0;
  co_enter(co, res); // 0 means ok
  if (!co_valid(co))
    co_destroy(co);
}

// may process mutliple completions
  static bool
cowq_process_nvme(void * const priv)
{
  struct spdk_nvme_qpair * const qp = (typeof(qp))priv;
  do {
    const int r = spdk_nvme_qpair_process_completions(qp, 0); // 0 or 1 cpl
    if (r > 0)
      return true;
    else if (r < 0)
      return false;

    // got nothing; retry
    cpu_pause();
  } while (true);
}

  bool
coq_pread_nvme(struct coq * const q, struct spdk_nvme_qpair * const qp,
    struct spdk_nvme_ns * const ns, void * const buf, const u64 lba, const u32 nr)
{
  const int rc = spdk_nvme_ns_cmd_read(ns, qp, buf, lba, nr, coq_process_nvme_cb, co_self(), 0);
  if (rc)
    return false;

  // callback
  const u32 i = cowq_enqueue(q, cowq_process_nvme, (void *)qp);

  const u64 res = co_back(0);
  cowq_remove(q, i);
  return res == 0;
}

  bool
coq_pwrite_nvme(struct coq * const q, struct spdk_nvme_qpair * const qp,
    struct spdk_nvme_ns * const ns, void * const buf, const u64 lba, const u32 nr)
{
  const int rc = spdk_nvme_ns_cmd_write(ns, qp, buf, lba, nr, coq_process_nvme_cb, co_self(), 0);
  if (rc)
    return false;

  // callback
  const u32 i = cowq_enqueue(q, cowq_process_nvme, (void *)qp);

  const u64 res = co_back(0);
  cowq_remove(q, i);
  return res == 0;
}
// }}} nvme

// spdk-helpers {{{
struct ctrlr_entry {
  struct spdk_nvme_ctrlr  *ctrlr;
  struct ctrlr_entry  *next; // chain
  char      name[1024];
};

// a ring of name spaces
struct ns_entry {
  struct spdk_nvme_ctrlr  *ctrlr;
  struct spdk_nvme_ns  *ns;
  struct ns_entry    *next; // chain
  struct spdk_nvme_qpair  *qpair;
};

struct nvmeset {
  bool seen_ctrlr;
  struct ctrlr_entry *cts;
  struct ns_entry *nss;
};

  static bool
probe_cb(void *cb_ctx, const struct spdk_nvme_transport_id *trid,
    struct spdk_nvme_ctrlr_opts *opts)
{
  printf("%s %s\n", __func__, trid->traddr);
  (void)opts;
  struct nvmeset * const pset = (typeof(pset))cb_ctx;

  // only use one ctrlr
  if (pset->seen_ctrlr == false) {
    pset->seen_ctrlr = true;
    return true;
  } else {
    return false;
  }
}

  static void
register_ns(struct nvmeset * const pset, struct spdk_nvme_ctrlr *ctrlr, struct spdk_nvme_ns *ns)
{
  if (!spdk_nvme_ns_is_active(ns))
    return;

  struct ns_entry * const entry = malloc(sizeof(*entry));
  debug_assert(entry);

  entry->ctrlr = ctrlr;
  entry->ns = ns;
  entry->next = pset->nss;
  pset->nss = entry;

  printf("Namespace ID: %u size: %luGB #sectors %lu sector size %u\n",
      spdk_nvme_ns_get_id(ns), spdk_nvme_ns_get_size(ns) >> 30,
      spdk_nvme_ns_get_num_sectors(ns), spdk_nvme_ns_get_sector_size(ns));
}

  static void
attach_cb(void *cb_ctx, const struct spdk_nvme_transport_id *trid,
    struct spdk_nvme_ctrlr *ctrlr, const struct spdk_nvme_ctrlr_opts *opts)
{
  (void)opts;
  struct ctrlr_entry *entry = malloc(sizeof(struct ctrlr_entry));
  debug_assert(entry);

  const struct spdk_nvme_ctrlr_data * const cdata = spdk_nvme_ctrlr_get_data(ctrlr);
  snprintf(entry->name, sizeof(entry->name), "%-20.20s (%-20.20s)", cdata->mn, cdata->sn);

  entry->ctrlr = ctrlr;
  // link to nvmeset
  struct nvmeset * const pset = (typeof(pset))cb_ctx;
  entry->next = pset->cts;
  pset->cts = entry;

  const u32 num_ns = spdk_nvme_ctrlr_get_num_ns(ctrlr);
  printf("%s %s with %u namespaces.\n", trid->traddr, entry->name, num_ns);
  // only use the first ns
  struct spdk_nvme_ns * const ns = spdk_nvme_ctrlr_get_ns(ctrlr, 1); // nsid == 1
  if (ns)
    register_ns(pset, ctrlr, ns);
}

  static void
cleanup(struct nvmeset * const pset)
{
  // free all ns_entry
  struct ns_entry * ns = pset->nss;
  while (ns) {
    struct ns_entry *next = ns->next;
    free(ns);
    ns = next;
  }

  // detach and free all ctrlr_entry
  struct ctrlr_entry * ct = pset->cts;
  while (ct) {
    struct ctrlr_entry *next = ct->next;
    spdk_nvme_detach(ct->ctrlr);
    free(ct);
    ct = next;
  }
}
// }}} spdk-helpers

// rawspdk {{{
struct rawspdk_worker_data {
  struct ns_entry  *ns;
  char *buf;
  u64 nsectors;
  u64 * count;
  u64 maxcount;
  volatile u64 * palive;
  u64 batch;
};

  static void
rawspdk_read_complete(void *arg, const struct spdk_nvme_cpl *completion)
{
  struct rawspdk_worker_data * const wd = arg;
  struct ns_entry * const ns = wd->ns;

  if (spdk_nvme_cpl_is_error(completion)) {
    spdk_nvme_qpair_print_completion(wd->ns->qpair, (struct spdk_nvme_cpl *)completion);
    fprintf(stderr, "I/O error status: %s\n", spdk_nvme_cpl_get_status_string(&completion->status));
    fprintf(stderr, "I/O failed, aborting run\n");
    exit(1);
  }
  if ((wd->count[0]++) >= wd->maxcount) {
    spdk_free(wd->buf);
    (*wd->palive)--;
    return;
  }
  (void)(wd->buf[0]);

  int rc = spdk_nvme_ns_cmd_read(ns->ns, ns->qpair, wd->buf,
      (random_u64() % wd->nsectors) * wd->batch, wd->batch, rawspdk_read_complete, arg, 0);
  if (rc != 0) {
    fprintf(stderr, "starting read I/O failed\n");
    exit(1);
  }
}

  static double
rawspdk_main(struct ns_entry * const ns, const u64 qlen, const u64 maxcount, const u64 batch)
{
  ns->qpair = spdk_nvme_ctrlr_alloc_io_qpair(ns->ctrlr, NULL, 0);
  if (ns->qpair == NULL) {
    printf("ERROR: spdk_nvme_ctrlr_alloc_io_qpair() failed\n");
    return 1.0;
  }

  const double t0 = time_sec();
  struct rawspdk_worker_data  data[qlen];
  u64 count = 0;
  volatile u64 alive = qlen;

  for (u64 i = 0; i < qlen; i++) {
    data[i].ns = ns;
    data[i].buf = spdk_zmalloc(spdk_nvme_ns_get_sector_size(ns->ns) * batch, 0x1000, NULL, SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA);
    data[i].nsectors = spdk_nvme_ns_get_num_sectors(ns->ns) / batch;
    data[i].count = &count;
    data[i].maxcount = maxcount;
    data[i].palive = &alive;
    data[i].batch = batch;
    int rc = spdk_nvme_ns_cmd_read(ns->ns, ns->qpair, data[i].buf,
        (random_u64() % data[i].nsectors) * batch, batch, rawspdk_read_complete, &data[i], 0);
    if (rc != 0) {
      fprintf(stderr, "starting write I/O failed\n");
      exit(1);
    }
  }

  while (alive != 0) {
    spdk_nvme_qpair_process_completions(ns->qpair, 0);
  }

  spdk_nvme_ctrlr_free_io_qpair(ns->qpair);
  const double dt = time_diff_sec(t0);
  printf("iops %.3lf\n", (double)maxcount /dt);
  return dt;
}

  static int
test_rawspdk(int argc, char **argv)
{
  if (argc < 4) {
    printf("usage <queue-size (6 is best for Intel 905p)> <total reqs> <# of 512B sectors per op>\n");
    exit(0);
  }
  const u64 qlen = a2u64(argv[1]);
  const u64 nr = a2u64(argv[2]);
  const u64 batch = a2u64(argv[3]);
  int rc;
  struct spdk_env_opts opts;

  rte_log_set_global_level(1);
  spdk_log_set_print_level(SPDK_LOG_DISABLED);
  spdk_env_opts_init(&opts);
  if (spdk_env_init(&opts) < 0) {
    fprintf(stderr, "Unable to initialize SPDK env\n");
    return 1;
  }

  struct nvmeset set = {};
  rc = spdk_nvme_probe(NULL, &set, probe_cb, attach_cb, NULL);
  if (rc != 0 || set.cts == NULL || set.nss == NULL) {
    fprintf(stderr, "spdk_nvme_probe() failed or no device\n");
    cleanup(&set);
    return 1;
  }

  rawspdk_main(set.nss, qlen, nr, batch);
  cleanup(&set);
  return 0;
}
// }}} rawspdk

// rawcospdk {{{
struct rawcospdk_worker_data {
  struct ns_entry  *ns;
  u64 maxsecs;
  u64 nsec;
  au64 count;
  u64 end_time;
  au64 nw;
};

  static void
rawcospdk_io_complete(void *arg, const struct spdk_nvme_cpl *completion)
{
  if (spdk_nvme_cpl_is_error(completion)) {
    fprintf(stderr, "I/O failed, aborting run\n");
    exit(1);
  }

  co_enter((struct co *)arg, 0);
}

  static void
rawcospdk_worker(void)
{
  struct rawcospdk_worker_data * const wd = co_priv();
  const u64 mod = wd->maxsecs / wd->nsec;
  struct ns_entry * const e = wd->ns;
  const size_t bufsize = spdk_nvme_ns_get_sector_size(e->ns) * wd->nsec;
  u8 * buf = spdk_zmalloc(bufsize, 0x1000, NULL, SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA);
  if (!buf)
    co_exit(0);
  do {
    for (u64 i = 0; i < 100; i++) {
      u64 lba = (random_u64() % mod) * wd->nsec;
      // submit I/O
      int rc = spdk_nvme_ns_cmd_read(e->ns, e->qpair, buf, lba,
          wd->nsec, rawcospdk_io_complete, co_self(), 0);
      if (rc != 0) {
        fprintf(stderr, "starting write I/O failed\n");
        exit(1);
      }
      // yield
      co_back(0);
      wd->count++;
    }
  } while (time_nsec() < wd->end_time);
  wd->nw--;
}

  static void
rawcospdk_main(struct ns_entry * const ns, const u64 qlen, const u64 runtime, const u64 batch)
{
  ns->qpair = spdk_nvme_ctrlr_alloc_io_qpair(ns->ctrlr, NULL, 0);
  if (ns->qpair == NULL) {
    printf("ERROR: spdk_nvme_ctrlr_alloc_io_qpair() failed\n");
    return;
  }

  struct rawcospdk_worker_data wd;
  wd.ns = ns;
  wd.maxsecs = spdk_nvme_ns_get_num_sectors(ns->ns);
  wd.nsec = batch;
  wd.count = 0;
  wd.end_time = time_nsec() + (1000000000lu * runtime);
  wd.nw = 0;
  u64 hostrsp = 0;

  const double t0 = time_sec();
  for (u64 i = 0; i < qlen; i++) {
    struct co * const co = co_create(8192, rawcospdk_worker, &wd, &hostrsp);
    co_enter(co, 0);
    if (!co_valid(co))
      co_destroy(co);
    else
      wd.nw++;
  }

  while (wd.nw)
    spdk_nvme_qpair_process_completions(ns->qpair, 0);

  const double dt = time_diff_sec(t0);
  printf("iops %.3lf\n", (double)wd.count /dt);
  spdk_nvme_ctrlr_free_io_qpair(ns->qpair);
}

  static int
test_rawcospdk(int argc, char **argv)
{
  if (argc < 4) {
    printf("usage <queue-size (6 is best for Intel 905p)> <time (sec)> <# of 512B sectors per op>\n");
    exit(0);
  }
  const u64 qlen = a2u64(argv[1]);
  const u64 runtime = a2u64(argv[2]);
  const u64 batch = a2u64(argv[3]);
  int rc;
  struct spdk_env_opts opts;

  rte_log_set_global_level(1);
  spdk_log_set_print_level(SPDK_LOG_DISABLED);
  spdk_env_opts_init(&opts);
  if (spdk_env_init(&opts) < 0) {
    fprintf(stderr, "Unable to initialize SPDK env\n");
    return 1;
  }

  struct nvmeset set = {};
  rc = spdk_nvme_probe(NULL, &set, probe_cb, attach_cb, NULL);
  if (rc != 0 || set.cts == NULL || set.nss == NULL) {
    fprintf(stderr, "spdk_nvme_probe() failed or no device\n");
    cleanup(&set);
    return 1;
  }

  rawcospdk_main(set.nss, qlen, runtime, batch);
  cleanup(&set);
  return 0;
}
// }}} rawcospdk

// cospdk {{{
struct cospdk_worker_info {
  struct coq * coq;
  struct spdk_nvme_ctrlr * ct;
  struct spdk_nvme_ns * ns;
  struct spdk_nvme_qpair  *qp;
};

struct cospdk_probe {
  bool dev_seen;
  // one controller
  struct spdk_nvme_ctrlr * ct;
  struct spdk_nvme_ns * ns;
};

  static bool
nvme_probe_cb(struct cospdk_probe * const p, const struct spdk_nvme_transport_id *trid,
    struct spdk_nvme_ctrlr_opts *opts)
{
  (void)trid;
  (void)opts;
  if (p->dev_seen == false) {
    p->dev_seen = true;
    return true;
  } else {
    return false;
  }
}

  static void
nvme_attach_cb(struct cospdk_probe * const p, const struct spdk_nvme_transport_id *trid,
    struct spdk_nvme_ctrlr *ct, const struct spdk_nvme_ctrlr_opts *opts)
{
  (void)trid;
  (void)opts;
  struct spdk_nvme_ns * const ns = spdk_nvme_ctrlr_get_ns(ct, 1); // nsid == 1
  if (ns == NULL) {
    printf("ns not found!\n");
  } else {
    printf("ns at %p\n", ns);
  }
  p->ct = ct;
  p->ns = ns;
}

  static void
cospdk_worker(void)
{
  struct cospdk_worker_info * const wi = (typeof(wi))co_priv();
  const u64 nsecs = spdk_nvme_ns_get_num_sectors(wi->ns);
  const u64 secsz = spdk_nvme_ns_get_sector_size(wi->ns);
  u8 * buf = spdk_zmalloc(secsz, secsz, NULL, SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA);
  for (u64 i = 0; i < 3; i++) {
    const bool r = coq_pread_nvme(wi->coq, wi->qp, wi->ns, buf, random_u64() % nsecs, 1);
    if (!r) {
      printf("read error %p\n", co_self());
    } else {
      printf("read done %p\n", co_self());
    }
  }
  printf("all done %p\n", co_self());
}

  static void
test_cospdk(void)
{
  struct spdk_env_opts opts;
  spdk_env_opts_init(&opts);
  if (spdk_env_init(&opts) < 0) {
    fprintf(stderr, "Unable to initialize SPDK env\n");
    exit(0);
  }

  struct cospdk_probe p = {};
  const int rc = spdk_nvme_probe(NULL, &p, (void *)nvme_probe_cb, (void *)nvme_attach_cb, NULL);
  if (rc) {
    printf("probe error\n");
    exit(0);
  }

  struct cospdk_worker_info wi = {};
  wi.coq = coq_create();
  wi.ct = p.ct;
  wi.ns = p.ns;
  wi.qp = spdk_nvme_ctrlr_alloc_io_qpair(wi.ct, NULL, 0);

  // enter loop
  u64 hostrsp = 0;
  for (u64 i = 0; i < 2; i++) {
    struct co * const co = co_create(8192, cospdk_worker, &wi, &hostrsp);
    corq_enqueue(wi.coq, co);
  }

  coq_run(wi.coq);

  // clean up
  spdk_nvme_detach(p.ct);
}
// }}} cospdk

// cospdkperf {{{
struct cospdkperf_th_info {
  bool dev_seen;
  // one controller
  struct spdk_nvme_ctrlr * ct;
  struct spdk_nvme_ns * ns;
  u64 sectsz;
  u64 nsects;

  u64 endtime;
  u64 bufsect;
  u64 rangesect;
  u64 nr_r;
  u64 cpt;

  au64 seq;
  au64 io_r;
  au64 io_w;
};

struct cospdkperf_co_info {
  struct coq * coq;
  struct spdk_nvme_qpair  *qp;
  struct cospdkperf_th_info * ti;
};

// coroutine
  static void
cospdk_co_worker(void)
{
  printf("co entering\n");
  struct cospdkperf_co_info * const wi = (typeof(wi))co_priv();
  struct cospdkperf_th_info * const ti = wi->ti;
  const u64 seq = atomic_fetch_add(&(ti->seq), 1);
  const bool is_reader = seq < ti->nr_r;
  u8 * buf = spdk_zmalloc(ti->sectsz*ti->bufsect, 0x1000, NULL, SPDK_ENV_SOCKET_ID_ANY, SPDK_MALLOC_DMA);
  const u64 mask = bits_p2_up_u64(ti->rangesect) - 8; // 4096 aligned
  typeof(coq_pread_nvme) * iofunc = is_reader ? coq_pread_nvme : coq_pwrite_nvme;
  u64 nrop = 0;

  do {
#define BATCHSIZE ((1000))
    for (u64 i = 0; i < BATCHSIZE; i++) {
      const bool r = iofunc(wi->coq, wi->qp, ti->ns, buf, random_u64() & mask, ti->bufsect);
      if (!r)
        printf("read error %p\n", co_self());
    }
    nrop += BATCHSIZE;
  } while (time_nsec() < ti->endtime);

  if (is_reader) {
    atomic_fetch_add(&(ti->io_r), nrop);
  } else {
    atomic_fetch_add(&(ti->io_w), nrop);
  }
  printf("worker done %p %c %lu\n", co_self(), is_reader ? 'r':'w', nrop);
}

  static void *
cospdk_th_worker(void * const ptr)
{
  printf("thread entering\n");
  struct cospdkperf_th_info * ti = (typeof(ti))ptr;
  struct cospdkperf_co_info wi = {};
  wi.coq = coq_create();
  // is this thread-safe?
  wi.qp = spdk_nvme_ctrlr_alloc_io_qpair(ti->ct, NULL, 0);
  wi.ti = ti;

  // enter loop
  u64 hostrsp = 0;
  for (u64 i = 0; i < ti->cpt; i++) {
    struct co * const co = co_create(8192, cospdk_co_worker, &wi, &hostrsp);
    corq_enqueue(wi.coq, co);
  }

  coq_run(wi.coq);
  spdk_nvme_ctrlr_free_io_qpair(wi.qp);
  return NULL;
}

  static bool
cospdk_nvme_probe_cb(struct cospdkperf_th_info * const p, const struct spdk_nvme_transport_id *trid,
    struct spdk_nvme_ctrlr_opts *opts)
{
  (void)trid;
  (void)opts;
  if (p->dev_seen == false) {
    p->dev_seen = true;
    return true;
  } else {
    return false;
  }
}

  static void
cospdk_nvme_attach_cb(struct cospdkperf_th_info * const p, const struct spdk_nvme_transport_id *trid,
    struct spdk_nvme_ctrlr *ct, const struct spdk_nvme_ctrlr_opts *opts)
{
  (void)trid;
  (void)opts;
  struct spdk_nvme_ns * const ns = spdk_nvme_ctrlr_get_ns(ct, 1); // nsid == 1
  if (ns == NULL) {
    printf("ns not found!\n");
  } else {
    printf("ns at %p\n", ns);
  }
  p->ct = ct;
  p->ns = ns;
  p->sectsz = spdk_nvme_ns_get_sector_size(ns);
  p->nsects = spdk_nvme_ns_get_num_sectors(ns);
}

  static int
test_cospdkperf(int argc, char ** argv)
{
  if (argc < 7) {
    fprintf(stderr, "usage: %s <nr_r> <nr_w> <time> <bufsect> <rangesect> <co-per-th>\n", argv[0]);
    fprintf(stderr, "example: %s 4 0 5 1 1024 4\n", argv[0]);
    exit(1);
  }

  struct spdk_env_opts opts;
  spdk_env_opts_init(&opts);
  if (spdk_env_init(&opts) < 0) {
    fprintf(stderr, "Unable to initialize SPDK env\n");
    exit(0);
  }

  struct cospdkperf_th_info ti = {};
  const int rc = spdk_nvme_probe(NULL, &ti, (void *)cospdk_nvme_probe_cb, (void *)cospdk_nvme_attach_cb, NULL);
  if (rc) {
    printf("probe error\n");
    exit(0);
  }

  const u32 nr_r = a2u32(argv[1]);
  const u32 nr_w = a2u32(argv[2]);
  const u64 runtime = a2u64(argv[3]);
  const u64 bufsect = a2u64(argv[4]);
  const u64 rangesect = a2u64(argv[5]);
  const u64 cpt = a2u64(argv[6]);

  printf("sectsz %lu nsects %lu bufsect %lu rangesect %lu\n", ti.sectsz, ti.nsects, bufsect, rangesect);
  ti.endtime = time_nsec() + (1000000000lu * runtime);
  ti.bufsect = bufsect;
  ti.rangesect = rangesect;
  ti.nr_r = nr_r * cpt;
  ti.cpt = cpt;

  const u64 dt = thread_fork_join(nr_r+nr_w, cospdk_th_worker, false, &ti);
  const double miops = (double)(ti.io_r + ti.io_w) * 1e3 / ((double)dt);
  printf("million IOPS %.3lf\n", miops);

  spdk_nvme_detach(ti.ct);
  return 0;
}
// }}} cospdkperf

#endif // SPDK

// mmap {{{
#define TESTNR ((1 << 18))
#define TESTMASK (((TESTNR - 1) << 12))
  static void *
test_mmap_worker_seq(void * ptr)
{
  u8 * const mem = ptr;
  u8 x = 0;
  for (u64 i = 0; i < TESTNR; i++) {
    x ^= mem[i*PGSZ];
  }
  printf("[%hhu]", x); // make sure memory is really accessed
  return NULL;
}

  static void *
test_mmap_worker_rnd(void * ptr)
{
  u8 * const mem = ptr;
  srandom_u64(time_nsec());
  u8 x = 0;
  for (u64 i = 0; i < TESTNR; i++) {
    x ^= mem[random_u64() & TESTMASK];
  }
  printf("[%hhu]", x); // make sure memory is really accessed
  return NULL;
}

  static int
test_mmap(int argc, char ** argv)
{
  if (argc < 2) {
    printf("usage: <devname>\n");
    return 0;
  }

  int fd = open(argv[1], O_RDONLY);
  if (fd < 0) {
    printf("open failed\n");
    return 0;
  }

  const size_t fsize = fdsize(fd);
  if (fsize < (PGSZ * TESTNR)) {
    printf("file is too small\n");
    exit(0);
  }

  u8 * const mem = mmap(NULL, fsize, PROT_READ, MAP_PRIVATE, fd, 0);
  close(fd);
  if (mem == MAP_FAILED) {
    printf("mmap failed\n");
    return 0;
  }

  for (u32 t = 1; t <= 8; t *= 2) {
    for (u64 s = 0; s < 2; s++) {
      const u64 dt = thread_fork_join(t, test_mmap_worker_seq, false, mem);
      printf("seq read th %u dt %lu ns/fault %lu ns/fault/th %lu\n", t, dt, dt / TESTNR / t, dt / TESTNR);
    }
    for (u64 s = 0; s < 2; s++) {
      const u64 dt = thread_fork_join(t, test_mmap_worker_rnd, false, mem);
      printf("rnd read th %u dt %lu ns/fault %lu ns/fault/th %lu\n", t, dt, dt / TESTNR / t, dt / TESTNR);
    }
  }

  munmap(mem, fsize);
  return 0;
}
// }}} mmap

// pread {{{
  static void *
test_pread_worker_seq(void * ptr)
{
  int fd = *(int *)ptr;
  u8 * pg = xalloc(PGSZ, PGSZ);
  for (u64 i = 0; i < TESTNR; i++) {
    pread(fd, pg, PGSZ, (off_t)(i * PGSZ));
  }
  free(pg);
  return NULL;
}

  static void *
test_pread_worker_rnd(void * ptr)
{
  int fd = *(int *)ptr;
  u8 * pg = xalloc(PGSZ, PGSZ);
  srandom_u64(time_nsec());
  for (u64 i = 0; i < TESTNR; i++) {
    pread(fd, pg, PGSZ, (off_t)(random_u64() & TESTMASK));
  }
  free(pg);
  return NULL;
}

  static int
test_pread(int argc, char ** argv)
{
  if (argc < 3) {
    printf("usage: <devname> <direct?[01]>\n");
    return 0;
  }

  const int dio = argv[2][0] == '1' ? O_DIRECT : 0;
  int fd = open(argv[1], O_RDONLY|dio);
  if (fd < 0) {
    printf("open failed\n");
    return 0;
  }

  const size_t fsize = fdsize(fd);
  if (fsize < (PGSZ * TESTNR)) {
    printf("file is too small\n");
    exit(0);
  }
  for (u32 t = 1; t <= 8; t *= 2) {
    for (u64 s = 0; s < 2; s++) {
      const u64 dt = thread_fork_join(t, test_pread_worker_seq, false, &fd);
      printf("seq read th %u dt %lu throughput-MB/s %lu latency-ns %lu\n", t, dt, (t * TESTNR * PGSZ)/(dt/1000), dt/TESTNR);
    }
    for (u64 s = 0; s < 2; s++) {
      const u64 dt = thread_fork_join(t, test_pread_worker_rnd, false, &fd);
      printf("rnd read th %u dt %lu throughput-MB/s %lu latency-ns %lu\n", t, dt, (t * TESTNR * PGSZ)/(dt/1000), dt/TESTNR);
    }
  }
  close(fd);
  return 0;
}
// }}} pread

// main {{{
  int
main(int argc, char ** argv)
{
  struct tests {
    char * name;
    typeof(main) * func;
  };
#define TC(name) { #name, (void *) test_ ## name, }
  static const struct tests tests[] = {
    TC(rcache), TC(rwcache),
#if defined(__linux__)
    TC(coqrw), TC(rawcoring), TC(ring), TC(coqread),
#endif // __linux__
    TC(wring),
#if defined(SPDK)
    TC(rawspdk), TC(rawcospdk), TC(cospdk), TC(cospdkperf),
#endif // SPDK
    TC(mmap), TC(pread),
    {NULL, NULL},
  };
#undef TC

  if (argc == 1) {
    printf("tests: ");
    for (int i = 0; tests[i].name; i++)
      printf(" %s", tests[i].name);
    printf("\n");
    return 0;
  }
  for (int j = 0; tests[j].name; j++)
    if (0 == strcmp(tests[j].name, argv[1]))
      return tests[j].func(argc-1, argv+1);

  printf("%s command not recognized\n", argv[1]);
  return 0;
}
// }}} main

// vim:fdm=marker