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