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