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