ucw/sorter/debug/radix-asio-test.c

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