text: /*
   * Copyright 2012-2014 Google Inc. All rights reserved.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  /*
   * A program to test/validate realtime disk performance under various
   * conditions.
   */
  #include <assert.h>
  #include <dirent.h>
  #include <errno.h>
  #include <fcntl.h>
  #include <memory.h>
  #include <pthread.h>
  #include <sched.h>
  #include <stdint.h>
  #include <stdio.h>
  #include <stdlib.h>
  #include <sys/sendfile.h>
  #include <sys/mman.h>
  #include <sys/stat.h>
  #include <sys/syscall.h>
  #include <sys/types.h>
  #include <time.h>
  #include <unistd.h>
  #include <netinet/tcp.h>
  #include <arpa/inet.h>
  #include "ioprio.h"
  
  
  #ifndef SCHED_IDLE
  // not defined in glibc nor uclibc for some reason, but in Linux since 2.6.23
  #define SCHED_IDLE  5
  #endif
  
  #define PCT_MIN_INIT 9999 // impossible percentage
  
  static int _posix_fallocate(int fd, __off64_t offset, __off64_t len) {
  #ifdef __UCLIBC__
    return syscall(
        SYS_fallocate, fd, 0,
        __LONG_LONG_PAIR((uint32_t)(offset >> 32), (uint32_t)offset),
        __LONG_LONG_PAIR((uint32_t)(len >> 32), (uint32_t)len));
  #else
    return posix_fallocate(fd, offset, len);
  #endif
  }
  
  
  struct TaskStatus {
    int tasknum;
    volatile long long counter;
    volatile long long total_spare_pct;
    volatile long long spare_pct_cnt;
    volatile long long spare_pct_min;
    int sock_fd; // used by reader/receiver for sendfile option
  };
  
  #define MAX_TASKS 128
  static struct TaskStatus *spinners[MAX_TASKS];
  static struct TaskStatus writers[MAX_TASKS];
  static struct TaskStatus readers[MAX_TASKS];
  static struct TaskStatus receivers[MAX_TASKS];
  
  #define MAX_FILE_SIZE (2*1000*1000*1000)
  #define MAX_BUF (128*1024*1024)
  static char *buf;
  
  // command-line arguments
  static int timeout = -1;
  static int nspins = 0;
  static int nwriters = 0;
  static int nreaders = 0;
  static int blocksize_write = 128*1024;
  static int blocksize_read = 0;
  static int bytes_per_sec = 2*1024*1024;
  static int so_rcvbuf = 0;
  static int so_sndbuf = 0;
  static int keep_old_files = 0;
  static int use_stagger = 0;
  static int use_o_direct_write = 0;
  static int use_o_direct_read = 0;
  static int use_sendfile = 0;
  static int use_mmap = 0;
  static int use_fallocate = 0;
  static int use_fsync = 0;
  static int use_realtime_prio = 0;
  static int use_ionice = 0;
  static int be_verbose = 0;
  static int print_extra_stats = 0;
  
  #define CHECK(x) _check(#x, x)
  
  static void _check(const char *str, int result) {
    if (!result) {
      perror(str);
      assert(result);
    }
  }
  
  
  // Returns the kernel monotonic timestamp in microseconds.
  static long long ustime(void) {
    struct timespec ts;
    if (clock_gettime(CLOCK_MONOTONIC, &ts) < 0) {
      perror("clock_gettime");
      exit(7); // really should never happen, so don't try to recover
    }
    return ts.tv_sec * 1000000LL + ts.tv_nsec / 1000;
  }
  
  
  static void _set_priority(int policy, int prio) {
    struct sched_param sp;
    memset(&sp, 0, sizeof(sp));
    sp.sched_priority = prio;
    CHECK(sched_setscheduler(0, policy, &sp) == 0);
  }
  
  
  static long _pagesize(void) {
    static long pagesize;
    if (!pagesize) {
      pagesize = sysconf(_SC_PAGESIZE);
      fprintf(stderr, "pagesize=%ld\n", pagesize);
    }
    return pagesize;
  }
  
  
  // write one byte every PAGESIZE bytes inside the buffer, thus forcing the
  // kernel to actually page all the accessed pages out to disk (eventually).
  static void _page_out(char *buf, size_t count) {
    static long seed;
    if (!seed) {
      seed = random();
    }
    for (size_t i = 0; i < count; i += _pagesize()) {
      buf[i] = ++seed;
    }
  }
  
  
  // read one byte every PAGESIZE bytes inside the buffer, thus forcing the
  // kernel to actually page all the accessed pages in from disk.
  static volatile char page_tempbyte;
  static void _page_in(char *buf, size_t count) {
    for (size_t i = 0; i < count; i += _pagesize()) {
      page_tempbyte = buf[i];
    }
  }
  
  
  static ssize_t _do_write(int fd, char *buf, size_t count) {
    assert(buf);
    if (use_mmap) {
      off_t oldpos = lseek(fd, 0, SEEK_CUR);
      struct stat st;
      CHECK(fstat(fd, &st) >= 0);
      if (st.st_size < oldpos + (ssize_t)count) {
        if (ftruncate(fd, oldpos + count) < 0) {
          // probably disk full
          return -1;
        }
      }
      char *wbuf;
      CHECK((wbuf = mmap(NULL, count, PROT_WRITE, MAP_SHARED, fd, oldpos))
            != MAP_FAILED);
      off_t newpos = lseek(fd, count, SEEK_CUR);
      count = newpos - oldpos;
      _page_out(wbuf, count);
      CHECK(munmap(wbuf, count) >= 0);
      return count;
    } else {
      // non-mmap version
      return write(fd, buf, count);
    }
  }
  
  
  static ssize_t _do_read(int fd, char **buf, size_t count, int socket_fd) {
    assert(buf);
    if (use_mmap) {
      off_t oldpos = lseek(fd, 0, SEEK_CUR);
      if (*buf) {
        CHECK(munmap(*buf, count) >= 0);
        *buf = NULL;
      }
      CHECK((*buf = mmap(NULL, count, PROT_READ, MAP_SHARED, fd, oldpos))
            != MAP_FAILED);
      off_t newpos = lseek(fd, count, SEEK_CUR);
      count = newpos - oldpos;
      _page_in(*buf, count);
      return count;
    } else if (use_sendfile && socket_fd >= 0) {
      // send the length as 32-bit number, followed by the data block
      uint32_t blocksz = count;
      CHECK(send(socket_fd, &blocksz, sizeof(blocksz), 0) == sizeof(blocksz));
      ssize_t ret = sendfile64(socket_fd, fd, 0, count);
      if (be_verbose) {
        fprintf(stderr, "sendfile sent %ld/%ld bytes to socket %d\n",
                (long)ret, (long)count, socket_fd);
      }
      return ret;
    } else {
      // non-mmap version
      if (!*buf) {
        CHECK(posix_memalign((void **)buf, _pagesize(), blocksize_read) == 0);
      }
      return read(fd, *buf, count);
    }
  }
  
  
  static void *spinner(void *_status) {
    struct TaskStatus *status = _status;
    fprintf(stderr, "s#%d ", status->tasknum);
  
    // use IDLE priority so that this task *never* runs unless nobody else
    // is interested.  Thus the spinners should only count upward if there's
    // an actual available idle CPU core to run the task.
    _set_priority(SCHED_IDLE, 0);
  
    volatile long long *counter = &status->counter;
    while (1) {
      // Note: it is not necessarily safe to increment a counter here without
      // a lock (since it's read from other threads). However, introducing
      // locking or atomic operations would slow down the counter and possibly
      // introduce CPU barriers or cache line flushes, which defeats the main
      // purpose of this counter: counting raw, uninterrupted CPU cycles.
      // Also this number isn't critical to the operation of the program, so
      // occasional mis-reads of the counter should not be harmful and should
      // be pretty obvious (an occasional, wildly wrong reading).
      // Locking would be much more critical if we incremented a given counter
      // from more than one thread, but we never do that.
      (*counter)++;
    }
    return NULL;
  }
  
  
  static void _create_socketpair(int *snd_fd, int *rcv_fd) {
    // create server socket
    int server_fd;
    CHECK((server_fd = socket(AF_INET, SOCK_STREAM, 0)) >= 0);
    int flags = 1;
    CHECK(setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &flags,
                     sizeof(flags)) == 0);
    struct sockaddr_in serveraddr;
    serveraddr.sin_family = AF_INET;
    serveraddr.sin_addr.s_addr = inet_addr("127.0.0.1");
    serveraddr.sin_port = htons(0);
  
    CHECK(bind(server_fd, (struct sockaddr*)&serveraddr,
               sizeof(serveraddr)) == 0);
    socklen_t len = sizeof(struct sockaddr);
    CHECK(getsockname(server_fd, (struct sockaddr *)&serveraddr, &len) == 0);
    int port = ntohs(serveraddr.sin_port);
  
    CHECK(listen(server_fd, 1) == 0);
  
    // create sender socket
    int sender_fd;
    CHECK((sender_fd = socket(AF_INET, SOCK_STREAM, 0)) >= 0);
    flags = 1;
    CHECK(setsockopt(sender_fd, IPPROTO_TCP, TCP_NODELAY, &flags,
                     sizeof(int)) == 0);
    flags = 4;
    CHECK(setsockopt(sender_fd, SOL_SOCKET, SO_PRIORITY, &flags,
                     sizeof(flags)) == 0);
    len = sizeof(int);
    int snd_size, old_snd_size = -1;
    if (so_sndbuf) {
      CHECK(getsockopt(sender_fd, SOL_SOCKET, SO_SNDBUF, &old_snd_size,
                       &len) == 0);
      CHECK(setsockopt(sender_fd, SOL_SOCKET, SO_SNDBUF, &so_sndbuf,
                       sizeof(int)) == 0);
    }
    len = sizeof(int);
    CHECK(getsockopt(sender_fd, SOL_SOCKET, SO_SNDBUF, &snd_size, &len) == 0);
  
    // connect sender to server
    memset(&serveraddr, 0, sizeof(serveraddr));
    serveraddr.sin_family = AF_INET;
    serveraddr.sin_addr.s_addr = inet_addr("127.0.0.1");
    serveraddr.sin_port = htons(port);
    CHECK(connect(sender_fd, (struct sockaddr *)&serveraddr,
                  sizeof(struct sockaddr)) == 0);
  
    // reader accepts connection request from sender
    struct sockaddr clientaddr;
    len = sizeof(struct sockaddr_in);
    int receiver_fd;
    CHECK((receiver_fd = accept(server_fd, &clientaddr, &len)) >= 0);
    close(server_fd);
    flags = 1;
    CHECK(setsockopt(receiver_fd, IPPROTO_TCP, TCP_NODELAY, &flags,
                     sizeof(flags)) == 0);
    flags = 4;
    CHECK(setsockopt(receiver_fd, SOL_SOCKET, SO_PRIORITY, &flags,
                     sizeof(flags)) == 0);
    len = sizeof(int);
    int rcv_size, old_rcv_size = -1;
    if (so_rcvbuf) {
      CHECK(getsockopt(receiver_fd, SOL_SOCKET, SO_RCVBUF, &old_rcv_size,
                       &len) == 0);
      CHECK(setsockopt(receiver_fd, SOL_SOCKET, SO_RCVBUF, &so_rcvbuf,
                       sizeof(int)) == 0);
    }
    len = sizeof(int);
    CHECK(getsockopt(receiver_fd, SOL_SOCKET, SO_RCVBUF, &rcv_size, &len) == 0);
  
    fprintf(stderr, "created socket pair, sender(%d) with so_snd_size:%d "
            "(was %d), receiver(%d) with so_rcv_size:%d (was %d)\n", sender_fd,
            snd_size / 2, old_snd_size / 2, receiver_fd, rcv_size / 2,
            old_rcv_size / 2);
    *snd_fd = sender_fd;
    *rcv_fd = receiver_fd;
  }
  
  
  static void *receiver(void *_status) {
    struct TaskStatus *status = _status;
  
    // use priority higher than the spinner (IDLE) but lower than the writers
    // and readers (FIFO, prio=10)
    if (use_realtime_prio) _set_priority(SCHED_FIFO, 1);
  
    fprintf(stderr, "n#%d ", status->tasknum);
  
    // dummy buffer that we receive all data into
    void *blackhole;
    CHECK((blackhole = malloc(2 * blocksize_read)) != NULL);
  
    while (1) {
      ssize_t bytes = recv(status->sock_fd, blackhole, 2 * blocksize_read, 0);
      CHECK(bytes >= 0);
      if (bytes == 0) {
        // socket was closed
        fprintf(stderr, "receiver socket %d closed\n", status->sock_fd);
        break;
      }
    }
    fprintf(stderr, "receiver thread exiting!\n");
    return NULL;
  }
  
  
  static void *writer(void *_status) {
    struct TaskStatus *status = _status;
    fprintf(stderr, "w#%d ", status->tasknum);
  
    int nblocks = MAX_FILE_SIZE / blocksize_write;
    long long blockdelay = blocksize_write * 1000000LL / bytes_per_sec;
  
    if (use_realtime_prio) _set_priority(SCHED_FIFO, 10);
    if (use_stagger) {
      // the 0.5 is to stagger the writers in between the staggered readers,
      // in case nreaders==nwriters.
      usleep(blockdelay * (0.5 + status->tasknum) / nwriters);
    }
  
    long long starttime = ustime();
    for (int fileno = 0; fileno < 1000000; fileno++) {
      char filename[1024];
      sprintf(filename, "db.%d.%d.tmp", status->tasknum, fileno);
      int fd;
      mode_t mode = O_RDWR|O_CREAT;
      if (use_o_direct_write) mode |= O_DIRECT;
      CHECK((fd = open(filename, mode, 0666)) >= 0);
      for (int blocknum = 0; blocknum < nblocks; blocknum++) {
        if (use_fallocate) {
          struct stat st;
          CHECK(fstat(fd, &st) == 0);
          if (st.st_size <= blocknum * blocksize_write) {
            CHECK(_posix_fallocate(fd, 0, blocknum * blocksize_write +
                                   100*1024*1024)  == 0);
          }
        }
        CHECK(_do_write(fd, buf + blocknum * 4096, blocksize_write) > 0);
        if (use_fsync) fdatasync(fd);
        long long now = ustime();
        starttime += blockdelay;
        long long spare_time = starttime - now;
        long long spare_pct = 100 * spare_time / blockdelay;
        status->total_spare_pct += spare_pct;
        if (spare_pct < status->spare_pct_min) status->spare_pct_min = spare_pct;
        status->spare_pct_cnt++;
        if (spare_time < 0) {
          // disk fell behind
          while (now > starttime) {
            status->counter++;
            starttime += blockdelay;
          }
        } else {
          // ahead of schedule, wait until next timeslot
          usleep(spare_time);
        }
      }
      close(fd);
    }
    assert(!"created an impossible number of files");
    return NULL;
  }
  
  
  static int open_random_file(int mode) {
    DIR *dir;
    struct dirent dentbuf, *dent = NULL;
  
    while (1) {
      CHECK((dir = opendir(".")) != NULL);
  
      int count = 0;
      while (readdir_r(dir, &dentbuf, &dent) == 0 && dent != NULL) {
        struct stat st;
        if (stat(dent->d_name, &st) < 0) continue;
        if (st.st_size > blocksize_read) {
          count++;
        }
      }
      if (!count) {
        fprintf(stderr, "reader: no big files to read yet.\n");
        closedir(dir);
        sleep(1);
        continue;
      }
  
      int want = random() % count, cur = 0;
      rewinddir(dir);
      while (readdir_r(dir, &dentbuf, &dent) == 0 && dent != NULL) {
        struct stat st;
        if (stat(dent->d_name, &st) < 0) continue;
        if (st.st_size > blocksize_read) {
          if (cur == want) {
            closedir(dir);
            return open(dent->d_name, mode);
          }
          cur++;
        }
      }
  
      // if we get here, one of the expected files has disappeared; try again.
      closedir(dir);
    }
    // not reached
    assert(!"should never get here");
  }
  
  
  static void *reader(void *_status) {
    struct TaskStatus *status = _status;
    fprintf(stderr, "r#%d ", status->tasknum);
  
    long long blockdelay = blocksize_read * 1000000LL / bytes_per_sec;
    char *rbuf = NULL;
  
    if (use_realtime_prio) _set_priority(SCHED_FIFO, 10);
    if (use_stagger) usleep(blockdelay * status->tasknum / nreaders);
  
    while (1) {
      int fd;
      mode_t mode = O_RDONLY;
      if (use_o_direct_read) mode |= O_DIRECT;
      CHECK((fd = open_random_file(mode)) >= 0);
      struct stat st;
      CHECK(fstat(fd, &st) == 0);
  
      // start reading at a random offset into the file.  If there aren't too
      // many files and we have lots of parallel readers, we might otherwise
      // get two readers reading the same blocks from the same file at the
      // same time, and if the disk cache kicks in, that reduces disk load
      // unnecessarily.
      off_t start_offset = (random() % (st.st_size / 65536 + 1)) * 65536;
      lseek(fd, start_offset, SEEK_SET);
  
      long long starttime = ustime(), got, totalbytes = start_offset;
      // we intentionally stop reading after we reach the *original* size of
      // the file, even if the file has grown since then.  Continuing to read
      // a growing file (presumably being written by a separate writer thread)
      // seems like a good test, because that's how the system works in real
      // life.  But it turns out to be so beneficial (when not using O_DIRECT,
      // the kernel can avoid doing disk reads) that it gets in the way of our
      // benchmark.  We need to check worst-case performance (reading old files
      // while new ones are being written) not average case.
      while (totalbytes + blocksize_read < st.st_size &&
             (got = _do_read(fd, &rbuf, blocksize_read, status->sock_fd)) > 0) {
        long long now = ustime();
        totalbytes += got;
        starttime += blockdelay;
        long long spare_time = starttime - now;
        long long spare_pct = 100 * spare_time / blockdelay;
        status->total_spare_pct += spare_pct;
        status->spare_pct_cnt++;
        if (spare_pct < status->spare_pct_min) status->spare_pct_min = spare_pct;
        if (spare_time < 0) {
          // disk fell behind
          while (now > starttime) {
            status->counter++;
            starttime += blockdelay;
          }
        } else {
          // ahead of schedule, wait until next timeslot
          usleep(spare_time);
        }
      }
      close(fd);
    }
    return NULL;
  }
  
  
  static long long count_spins(void) {
    static long long last_end, last_total;
    long long total = 0;
    for (int i = 0; i < nspins; i++) {
      total += spinners[i]->counter;
    }
    long long this_end = ustime(), this_spin;
    if (last_end) {
      this_spin = (total - last_total) / (this_end - last_end);
    } else {
      this_spin = 0;
    }
    last_end = this_end;
    last_total = total;
    return this_spin;
  }
  
  
  static long long sum_tasks(struct TaskStatus *array, int nelems) {
    long long total = 0;
    for (int i = 0; i < nelems; i++) {
      total += array[i].counter;
    }
    return total;
  }
  
  
  static long long avg_spare_time(struct TaskStatus *array, int nelems) {
    long long total = 0;
    for (int i = 0; i < nelems; i++) {
      if (array[i].spare_pct_cnt) {
        total += array[i].total_spare_pct / array[i].spare_pct_cnt;
        array[i].total_spare_pct = array[i].spare_pct_cnt = 0;
      }
    }
    return total / nelems;
  }
  
  
  static long long min_spare_time(struct TaskStatus *array, int nelems) {
    long long min_pct = array[0].spare_pct_min;
    array[0].spare_pct_min = PCT_MIN_INIT;
    for (int i = 1; i < nelems; i++) {
      if (array[i].spare_pct_min < min_pct)
        min_pct = array[i].spare_pct_min;
      array[i].spare_pct_min = PCT_MIN_INIT;
    }
    return min_pct;
  }
  
  
  static void usage(void) {
    fprintf(stderr,
            "\n"
            "Usage: diskbench [options]\n"
            "    -h, -?  This help message\n"
            "    -t ...  Timeout (number of seconds to run test)\n"
            "    -i ...  Number of idle spinners (to occupy CPU threads)\n"
            "    -w ...  Number of parallel writers (creating files)\n"
            "    -r ...  Number of parallel readers (reading files)\n"
            "    -b ...  Block size (kbyte size of a single read/write)\n"
            "    -c ...  Alternative block size for reading (kbyte)\n"
            "    -s ...  Speed (kbytes read/written per sec, per stream)\n"
            "    -m ...  Socket receive buffer size in KB (for sendfile)\n"
            "    -z ...  Socket send buffer size in KB (for sendfile)\n"
            "    -K      Keep old temp output files from previous run\n"
            "    -S      Stagger reads and writes evenly (default: clump them)\n"
            "    -D      Use O_DIRECT for writing\n"
            "    -O      Use O_DIRECT for reading\n"
            "    -N      Use sendfile to send read data through a socket\n"
            "            to a local client\n"
            "    -M      Use mmap()\n"
            "    -F      Use fallocate()\n"
            "    -Y      Use fdatasync() after writing\n"
            "    -R      Use CPU real-time priority\n"
            "    -I      Use ionice real-time disk priority\n"
            "    -E      Print extra stats\n"
            "    -v      Verbose output\n");
    exit(99);
  }
  
  
  int main(int argc, char **argv) {
    srandom(time(NULL));
  
    int opt;
    while ((opt = getopt(argc, argv, "?ht:i:w:r:b:c:s:m:z:KSDONMFYRIEv")) != -1) {
      switch (opt) {
      case '?':
      case 'h':
        usage();
        break;
      case 't':
        timeout = atoi(optarg);
        break;
      case 'i':
        nspins = atoi(optarg);
        break;
      case 'w':
        nwriters = atoi(optarg);
        break;
      case 'r':
        nreaders = atoi(optarg);
        break;
      case 'b':
        blocksize_write = atoi(optarg) * 1024;
        break;
      case 'c':
        blocksize_read = atoi(optarg) * 1024;
        break;
      case 's':
        bytes_per_sec = atoi(optarg) * 1024;
        break;
      case 'm':
        so_rcvbuf = atoi(optarg) * 1024;
        break;
      case 'z':
        so_sndbuf = atoi(optarg) * 1024;
        break;
      case 'K':
        keep_old_files = 1;
        break;
      case 'S':
        use_stagger = 1;
        break;
      case 'D':
        use_o_direct_write = 1;
        break;
      case 'O':
        use_o_direct_read = 1;
        break;
      case 'N':
        use_sendfile = 1;
        break;
      case 'M':
        use_mmap = 1;
        break;
      case 'F':
        use_fallocate = 1;
        break;
      case 'Y':
        use_fsync = 1;
        break;
      case 'R':
        use_realtime_prio = 1;
        break;
      case 'I':
        use_ionice = 1;
        break;
      case 'E':
        print_extra_stats = 1;
        break;
      case 'v':
        be_verbose = 1;
        break;
      }
    }
  
    if (nspins > MAX_TASKS || nreaders > MAX_TASKS || nwriters > MAX_TASKS) {
      fprintf(stderr, "\nfatal: idlers, readers, and writers must all be <= %d\n",
              MAX_TASKS);
      return 8;
    }
  
    if (!nspins && !nreaders && !nwriters) {
      fprintf(stderr, "\nfatal: must specify at least one of -i, -r, -w\n");
      return 9;
    }
  
    if (!blocksize_read) blocksize_read = blocksize_write;
  
    CHECK(posix_memalign((void **)&buf, _pagesize(), MAX_BUF) == 0);
    for (int i = 0; i < MAX_BUF; i++) {
      buf[i] = i % 257;
    }
  
    if (nwriters == 0) {
      fprintf(stderr, "not clearing old temp files (-w 0)\n");
    } else if (keep_old_files) {
      fprintf(stderr, "not clearing old temp files (-K)\n");
    } else {
      fprintf(stderr, "clearing old temp files.\n");
      CHECK(system("rm -f db.*.*.tmp") == 0);
    }
  
    fprintf(stderr, "syncing disks.\n");
    sync();
  
    fprintf(stderr, "starting: %d idlers, %d readers, %d writers\n",
            nspins, nreaders, nwriters);
  
    for (int i = 0; i < nspins; i++) {
      // spinners[] could just be an array of structs, instead of pointers to
      // structs, but we want to make sure the counters don't all share the
      // same cache line.  Prevening this sharing more than doubles the
      // counting rate on my x86_64 (Xeon X5650) machine, although it seems
      // to make no difference on BCM7425. I don't want an endless stream of
      // conflicting cache line flushes to artificially inflate CPU usage.
      spinners[i] = calloc(1, sizeof(struct TaskStatus));
      spinners[i]->tasknum = i + 1;
      pthread_t thread;
      CHECK(pthread_create(&thread, NULL, spinner, spinners[i]) == 0);
    }
  
    for (int i = 0; i < nspins; i++) {
      spinners[i]->counter = 0;
    }
    count_spins(); // initialize timings
    sleep(1);  // run for one cycle without any non-spinner activity
    long long best_spin = count_spins();
    if (!best_spin) best_spin = 1;
    fprintf(stderr, "\nidle spins:%lld\n", best_spin);
  
    if (use_ionice) {
      int realtime = IOPRIO_PRIO_VALUE(IOPRIO_CLASS_RT, 0);
      CHECK(ioprio_set(IOPRIO_WHO_PROCESS, getpid(), realtime) != -1);
    }
  
    for (int i = 0; i < nwriters; i++) {
      memset(&writers[i], 0, sizeof(writers[i]));
      writers[i].tasknum = i;
      writers[i].spare_pct_min = PCT_MIN_INIT;
      pthread_t thread;
      CHECK(pthread_create(&thread, NULL, writer, &writers[i]) == 0);
    }
  
    for (int i = 0; i < nreaders; i++) {
      memset(&readers[i], 0, sizeof(readers[i]));
      if (use_sendfile) {
        memset(&receivers[i], 0, sizeof(receivers[i]));
        _create_socketpair(&readers[i].sock_fd, &receivers[i].sock_fd);
        receivers[i].tasknum = i;
        pthread_t thread;
        CHECK(pthread_create(&thread, NULL, receiver, &receivers[i]) == 0);
      } else {
        readers[i].sock_fd = -1; // disable
      }
      readers[i].tasknum = i;
      readers[i].spare_pct_min = PCT_MIN_INIT;
      pthread_t thread;
      CHECK(pthread_create(&thread, NULL, reader, &readers[i]) == 0);
    }
  
    // that cycle was filled with startup traffic, just ignore it
    usleep(100*1000);
    count_spins();
    fprintf(stderr, "\n");
  
    long long count = 0;
    while (timeout == -1 || count < timeout) {
      sleep(1);
      long long this_spin = count_spins();
      if (this_spin > best_spin) best_spin = this_spin;
      if (print_extra_stats) {
        printf("%5lld  spins:%lld/%lld  cpu:%.2f%%  overruns: w=%lld r=%lld "
               "avg/min spare_time: w=%lld/%lld%% r=%lld/%lld%%\n",
               ++count,
               this_spin, best_spin,
               100 * (1-(this_spin*1.0/best_spin)),
               sum_tasks(writers, nwriters),
               sum_tasks(readers, nreaders),
               avg_spare_time(writers, nwriters),
               min_spare_time(writers, nwriters),
               avg_spare_time(readers, nreaders),
               min_spare_time(readers, nreaders));
      } else {
        printf("%5lld  spins:%lld/%lld  cpu:%.2f%%  overruns: w=%lld r=%lld\n",
               ++count,
               this_spin, best_spin,
               100 * (1-(this_spin*1.0/best_spin)),
               sum_tasks(writers, nwriters),
               sum_tasks(readers, nreaders));
      }
      fflush(stdout);
    }
    return 0;
  }
  