debug/sorter/asio-test.c

   1 /*
   2  *  An experiment with parallel reading and writing of files using ASIO.
   3  */
   4
   5 #include "lib/lib.h"
   6 #include "lib/lfs.h"
   7 #include "lib/asio.h"
   8
   9 #include <stdio.h>
  10 #include <stdlib.h>
  11 #include <fcntl.h>
  12 #include <unistd.h>
  13
  14 #define COPY
  15 #define DIRECT O_DIRECT
  16
  17 #define P_INIT do { cnt = 0; cnt_rep = 0; cnt_ms = 1; } while(0)
  18 #define P_UPDATE(cc) do { \
  19   cnt += cc; \
  20   if (cnt >= cnt_rep) { cnt_ms += get_timer(); \
  21     printf("%d of %d MB (%.2f MB/sec)\r", (int)(cnt >> 20), (int)(total_size >> 20), (double)cnt / 1048576 * 1000 / cnt_ms); \
  22     fflush(stdout); cnt_rep += 1<<26; } } while(0)
  23 #define P_FINAL do { \
  24   cnt_ms += get_timer(); \
  25   log(L_INFO, "Spent %.3f sec (%.2f MB/sec)", (double)cnt_ms/1000, (double)cnt / 1048576 * 1000 / cnt_ms); \
  26 } while(0)
  27
  28 static struct asio_queue io_queue;
  29
  30 int main(int argc, char **argv)
  31 {
  32   ASSERT(argc == 4);
  33   uns files = atol(argv[1]);
  34   uns bufsize = atol(argv[2]) * 1024;                           // Kbytes
  35   u64 total_size = (u64)atol(argv[3]) * 1024*1024*1024;         // Gbytes
  36   u64 cnt, cnt_rep;
  37   uns cnt_ms;
  38   int fd[files];
  39   byte name[files][16];
  40
  41   init_timer();
  42
  43   io_queue.buffer_size = bufsize;
  44   io_queue.max_writebacks = 2;
  45   asio_init_queue(&io_queue);
  46
  47 #ifdef COPY
  48   log(L_INFO, "Creating input file");
  49   int in_fd = sh_open("tmp/ft-in", O_RDWR | O_CREAT | O_TRUNC | DIRECT, 0666);
  50   ASSERT(in_fd >= 0);
  51   ASSERT(!(total_size % bufsize));
  52   P_INIT;
  53   for (uns i=0; i<total_size/bufsize; i++)
  54     {
  55       struct asio_request *r = asio_get(&io_queue);
  56       r->op = ASIO_WRITE_BACK;
  57       r->fd = in_fd;
  58       r->len = bufsize;
  59       byte *xbuf = r->buffer;
  60       for (uns j=0; j<bufsize; j++)
  61         xbuf[j] = i+j;
  62       asio_submit(r);
  63       P_UPDATE(bufsize);
  64     }
  65   asio_sync(&io_queue);
  66   lseek(in_fd, 0, SEEK_SET);
  67   sync();
  68   P_FINAL;
  69 #endif
  70
  71   log(L_INFO, "Initializing output files");
  72   for (uns i=0; i<files; i++)
  73     {
  74       sprintf(name[i], "tmp/ft-%d", i);
  75       fd[i] = sh_open(name[i], O_RDWR | O_CREAT | O_TRUNC | DIRECT, 0666);
  76       if (fd[i] < 0)
  77         die("Cannot create %s: %m", name[i]);
  78     }
  79   sync();
  80   get_timer();
  81
  82   log(L_INFO, "Writing %d MB to %d files in parallel with %d byte buffers", (int)(total_size >> 20), files, bufsize);
  83   P_INIT;
  84   for (uns round=0; round<total_size/bufsize/files; round++)
  85     {
  86       for (uns i=0; i<files; i++)
  87         {
  88           struct asio_request *r = asio_get(&io_queue);
  89 #ifdef COPY
  90           r->op = ASIO_READ;
  91           r->fd = in_fd;
  92           r->len = bufsize;
  93           asio_submit(r);
  94           struct asio_request *rr = asio_wait(&io_queue);
  95           ASSERT(rr == r && r->status == (int)r->len);
  96 #else
  97           for (uns j=0; j<bufsize; j++)
  98             r->buffer[j] = round+i+j;
  99 #endif
 100           r->op = ASIO_WRITE_BACK;
 101           r->fd = fd[i];
 102           r->len = bufsize;
 103           asio_submit(r);
 104           P_UPDATE(bufsize);
 105         }
 106     }
 107   asio_sync(&io_queue);
 108 #ifdef COPY
 109   close(in_fd);
 110 #endif
 111   log(L_INFO, "Syncing");
 112   sync();
 113   P_FINAL;
 114
 115   log(L_INFO, "Reading the files sequentially");
 116   P_INIT;
 117   for (uns i=0; i<files; i++)
 118     {
 119       lseek(fd[i], 0, SEEK_SET);
 120       for (uns round=0; round<total_size/bufsize/files; round++)
 121         {
 122           struct asio_request *rr, *r = asio_get(&io_queue);
 123           r->op = ASIO_READ;
 124           r->fd = fd[i];
 125           r->len = bufsize;
 126           asio_submit(r);
 127           rr = asio_wait(&io_queue);
 128           ASSERT(rr == r && r->status == (int)bufsize);
 129           asio_put(r);
 130           P_UPDATE(bufsize);
 131         }
 132       close(fd[i]);
 133     }
 134   P_FINAL;
 135
 136   for (uns i=0; i<files; i++)
 137     unlink(name[i]);
 138 #ifdef COPY
 139   unlink("tmp/ft-in");
 140 #endif
 141
 142   asio_cleanup_queue(&io_queue);
 143   log(L_INFO, "Done");
 144   return 0;
 145 }