lustre/tests/aiocp.c

   1 // SPDX-License-Identifier: GPL-2.0+
   2 /*
   3  * Copyright (c) 2004 Daniel McNeil <daniel@osdl.org>
   4  *               2004 Open Source Development Lab
   5  *
   6  * Copy file by using a async I/O state machine.
   7  * 1. Start read request
   8  * 2. When read completes turn it into a write request
   9  * 3. When write completes decrement counter and free resources
  10  *
  11  * Usage: aiocp [-b blksize] -n [num_aio] [-w] [-z] [-s filesize]
  12  *              [-f DIRECT|TRUNC|CREAT|SYNC|LARGEFILE] src dest
  13  *
  14  * Change History:
  15  *
  16  * version of copy command using async I/O
  17  * From:        Stephen Hemminger <shemminger@osdl.org>
  18  * Modified by Daniel McNeil <daniel@osdl.org> for testing aio.
  19  *      - added -a alignment
  20  *      - added -b blksize option
  21  *      - added -s size option
  22  *      - added -f open_flag option
  23  *      - added -w (no write) option (reads from source only)
  24  *      - added -n (num aio) option
  25  *      - added -z (zero dest) opton (writes zeros to dest only)
  26  *      - added -D delay_ms option
  27  *  - 2/2004  Marty Ridgeway (mridge@us.ibm.com) Changes to adapt to LTP
  28  */
  29
  30 /* #define _GNU_SOURCE */
  31 /* #define DEBUG 1 */
  32 #undef DEBUG
  33
  34 #include <unistd.h>
  35 #include <stdio.h>
  36 #include <sys/types.h>
  37 #include <sys/stat.h>
  38 #include <sys/param.h>
  39 #include <fcntl.h>
  40 #include <errno.h>
  41 #include <stdlib.h>
  42 #include <sys/select.h>
  43 #include <lustre/lustreapi.h>
  44 #include <libaio.h>
  45
  46 #define AIO_BLKSIZE     (64*1024)
  47 #define AIO_MAXIO       32
  48
  49 static int aio_blksize = AIO_BLKSIZE;
  50 static int aio_maxio = AIO_MAXIO;
  51
  52 static int busy;                /* # of I/O's in flight */
  53 static int tocopy;              /* # of blocks left to copy */
  54 static int srcfd;               /* source fd */
  55 static int dstfd = -1;          /* destination file descriptor */
  56 static const char *dstname;
  57 static const char *srcname;
  58 static int source_open_flag = O_RDONLY; /* open flags on source file */
  59 static int dest_open_flag = O_WRONLY;   /* open flags on dest file */
  60 static int no_write;                    /* do not write */
  61 static int zero;                        /* write zero's only */
  62
  63 static int debug;
  64 static int count_io_q_waits;    /* how many times io_queue_wait called */
  65
  66 struct iocb **iocb_free;        /* array of pointers to iocb */
  67 int iocb_free_count;            /* current free count */
  68 int alignment = 512;            /* buffer alignment */
  69
  70 struct timeval delay;           /* delay between I/O */
  71
  72 int init_iocb(int n, int iosize)
  73 {
  74         void *buf;
  75         int i;
  76
  77         iocb_free = malloc(n * sizeof(struct iocb *));
  78         if (iocb_free == 0)
  79                 return -1;
  80
  81         for (i = 0; i < n; i++) {
  82                 iocb_free[i] = (struct iocb *) malloc(sizeof(struct iocb));
  83                 if (!iocb_free[i])
  84                         return -1;
  85                 if (posix_memalign(&buf, alignment, iosize))
  86                         return -1;
  87                 if (debug > 1) {
  88                         printf("buf allocated at 0x%p, align:%d\n",
  89                                         buf, alignment);
  90                 }
  91                 if (zero) {
  92                         /*
  93                          * We are writing zero's to dstfd
  94                          */
  95                         memset(buf, 0, iosize);
  96                 }
  97                 io_prep_pread(iocb_free[i], -1, buf, iosize, 0);
  98         }
  99         iocb_free_count = i;
 100         return 0;
 101 }
 102
 103 struct iocb *alloc_iocb()
 104 {
 105         if (!iocb_free_count)
 106                 return 0;
 107         return iocb_free[--iocb_free_count];
 108 }
 109
 110 void free_iocb(struct iocb *io)
 111 {
 112         iocb_free[iocb_free_count++] = io;
 113 }
 114
 115 /*
 116  * io_wait_run() - wait for an io_event and then call the callback.
 117  */
 118 int io_wait_run(io_context_t ctx, struct timespec *to)
 119 {
 120         struct io_event events[aio_maxio];
 121         struct io_event *ep;
 122         int ret, n;
 123
 124         /*
 125          * get up to aio_maxio events at a time.
 126          */
 127         ret = n = io_getevents(ctx, 1, aio_maxio, events, to);
 128
 129         /*
 130          * Call the callback functions for each event.
 131          */
 132         for (ep = events; n-- > 0; ep++) {
 133                 io_callback_t cb = (io_callback_t)ep->data;
 134                 struct iocb *iocb = (struct iocb *)ep->obj;
 135
 136                 cb(ctx, iocb, ep->res, ep->res2);
 137         }
 138         return ret;
 139 }
 140
 141 /* Fatal error handler */
 142 static void io_error(const char *func, int rc)
 143 {
 144         if (rc < 0)
 145                 fprintf(stderr, "%s: %s\n", func, strerror(-rc));
 146         else
 147                 fprintf(stderr, "%s: error %d\n", func, rc);
 148
 149         if (dstfd > 0)
 150                 close(dstfd);
 151         if (dstname && dest_open_flag & O_CREAT)
 152                 unlink(dstname);
 153         exit(1);
 154 }
 155
 156 /*
 157  * Write complete callback.
 158  * Adjust counts and free resources
 159  */
 160 static void wr_done(io_context_t ctx, struct iocb *iocb, long res, long res2)
 161 {
 162         if (res2 != 0)
 163                 io_error("aio write", res2);
 164
 165         if (res != iocb->u.c.nbytes) {
 166                 fprintf(stderr, "write missed bytes expected %lu got %ld\n",
 167                         iocb->u.c.nbytes, res2);
 168                 exit(1);
 169         }
 170         --tocopy;
 171         --busy;
 172         free_iocb(iocb);
 173         if (debug)
 174                 fprintf(stderr, "w");
 175 }
 176
 177 /*
 178  * Read complete callback.
 179  * Change read iocb into a write iocb and start it.
 180  */
 181 static void rd_done(io_context_t ctx, struct iocb *iocb, long res, long res2)
 182 {
 183         /* library needs accessors to look at iocb? */
 184         int iosize = iocb->u.c.nbytes;
 185         char *buf = iocb->u.c.buf;
 186         off_t offset = iocb->u.c.offset;
 187
 188         if (res2 != 0)
 189                 io_error("aio read", res2);
 190         if (res != iosize) {
 191                 fprintf(stderr, "read missed bytes expected %lu got %ld\n",
 192                         iocb->u.c.nbytes, res);
 193                 exit(1);
 194         }
 195
 196
 197         /* turn read into write */
 198         if (no_write) {
 199                 --tocopy;
 200                 --busy;
 201                 free_iocb(iocb);
 202         } else {
 203                 io_prep_pwrite(iocb, dstfd, buf, iosize, offset);
 204                 io_set_callback(iocb, wr_done);
 205                 res = io_submit(ctx, 1, &iocb);
 206                 if (res != 1)
 207                         io_error("io_submit write", res);
 208         }
 209         if (debug)
 210                 fprintf(stderr, "r");
 211         if (debug > 1)
 212                 printf("%d", iosize);
 213 }
 214
 215 void usage(void)
 216 {
 217         fprintf(stderr,
 218                 "Usage: aiocp [-a align] [-s size] [-b blksize] [-n num_io] [-f open_flag] SOURCE DEST\n"
 219                 "This copies from SOURCE to DEST using AIO.\n\n"
 220                 "Usage: aiocp [options] -w SOURCE\n"
 221                 "This does sequential AIO reads (no writes).\n\n"
 222                 "Usage: aiocp [options] -z DEST\n"
 223                 "This does sequential AIO writes of zeros.\n");
 224         exit(1);
 225 }
 226
 227 /*
 228  * Scale value by kilo, mega, or giga.
 229  */
 230 long long scale_by_kmg(long long value, char scale)
 231 {
 232         switch (scale) {
 233         case 'g':
 234         case 'G':
 235                 value *= 1024;
 236                 fallthrough;
 237         case 'm':
 238         case 'M':
 239                 value *= 1024;
 240                 fallthrough;
 241         case 'k':
 242         case 'K':
 243                 value *= 1024;
 244                 break;
 245         case '\0':
 246                 break;
 247         default:
 248                 usage();
 249                 break;
 250         }
 251         return value;
 252 }
 253
 254 int main(int argc, char *const *argv)
 255 {
 256         struct stat st;
 257         off_t length = 0, offset = 0;
 258         io_context_t myctx;
 259         int c;
 260
 261         while ((c = getopt(argc, argv, "a:b:df:n:s:wzD:")) != -1) {
 262                 char *endp;
 263
 264                 switch (c) {
 265                 case 'a':       /* alignment of data buffer */
 266                         alignment = strtol(optarg, &endp, 0);
 267                         alignment = (long)scale_by_kmg((long long)alignment,
 268                                                         *endp);
 269                         break;
 270                 case 'f':       /* use these open flags */
 271                         if (strcmp(optarg, "LARGEFILE") == 0 ||
 272                             strcmp(optarg, "O_LARGEFILE") == 0) {
 273                                 source_open_flag |= O_LARGEFILE;
 274                                 dest_open_flag |= O_LARGEFILE;
 275                         } else if (strcmp(optarg, "TRUNC") == 0 ||
 276                                    strcmp(optarg, "O_TRUNC") == 0) {
 277                                 dest_open_flag |= O_TRUNC;
 278                         } else if (strcmp(optarg, "SYNC") == 0 ||
 279                                    strcmp(optarg, "O_SYNC") == 0) {
 280                                 dest_open_flag |= O_SYNC | O_NONBLOCK;
 281                         } else if (strcmp(optarg, "DIRECT") == 0 ||
 282                                    strcmp(optarg, "O_DIRECT") == 0) {
 283                                 source_open_flag |= O_DIRECT;
 284                                 dest_open_flag |= O_DIRECT;
 285                         } else if (strncmp(optarg, "CREAT", 5) == 0 ||
 286                                    strncmp(optarg, "O_CREAT", 5) == 0) {
 287                                 dest_open_flag |= O_CREAT;
 288                         }
 289                         break;
 290                 case 'd':
 291                         debug++;
 292                         break;
 293                 case 'D':
 294                         delay.tv_usec = atoi(optarg);
 295                         break;
 296                 case 'b':       /* block size */
 297                         aio_blksize = strtol(optarg, &endp, 0);
 298                         aio_blksize = (long)scale_by_kmg(
 299                                         (long long)aio_blksize, *endp);
 300                         break;
 301                 case 'n':       /* num io */
 302                         aio_maxio = strtol(optarg, &endp, 0);
 303                         break;
 304                 case 's':       /* size to transfer */
 305                         length = strtoll(optarg, &endp, 0);
 306                         length = scale_by_kmg(length, *endp);
 307                         break;
 308                 case 'w':       /* no write */
 309                         no_write = 1;
 310                         break;
 311                 case 'z':       /* write zero's */
 312                         zero = 1;
 313                         break;
 314
 315                 default:
 316                         usage();
 317                 }
 318         }
 319
 320         argc -= optind;
 321         argv += optind;
 322
 323 #ifndef DEBUG
 324         if (argc < 1)
 325                 usage();
 326 #else
 327         source_open_flag |= O_DIRECT;
 328         dest_open_flag |= O_DIRECT;
 329         aio_blksize = 1;
 330         aio_maxio = 1;
 331         srcname = "junkdata";
 332         dstname = "ff2";
 333 #endif
 334         if (!zero) {
 335 #ifndef DEBUG
 336                 srcfd = open(srcname = *argv, source_open_flag);
 337                 if (srcfd < 0) {
 338 #else
 339                 srcfd = open(srcname, source_open_flag);
 340                 if (srcfd < 0) {
 341 #endif
 342                         perror(srcname);
 343                         exit(1);
 344                 }
 345                 argv++;
 346                 argc--;
 347                 if (fstat(srcfd, &st) < 0) {
 348                         perror("fstat");
 349                         exit(1);
 350                 }
 351                 if (length == 0)
 352                         length = st.st_size;
 353         }
 354
 355         if (!no_write) {
 356                 /*
 357                  * We are either copying or writing zeros to dstname
 358                  */
 359 #ifndef DEBUG
 360                 if (argc < 1)
 361                         usage();
 362                 dstfd = open(dstname = *argv, dest_open_flag, 0666);
 363                 if (dstfd < 0) {
 364 #else
 365                 dstfd = open(dstname, dest_open_flag, 0666);
 366                 if (dstfd < 0) {
 367 #endif
 368                         perror(dstname);
 369                         exit(1);
 370                 }
 371                 if (zero) {
 372                         /*
 373                          * get size of dest, if we are zeroing it.
 374                          * TODO: handle devices.
 375                          */
 376                         if (fstat(dstfd, &st) < 0) {
 377                                 perror("fstat");
 378                                 exit(1);
 379                         }
 380                         if (length == 0)
 381                                 length = st.st_size;
 382                 }
 383         }
 384
 385         /* initialize state machine */
 386         memset(&myctx, 0, sizeof(myctx));
 387         io_queue_init(aio_maxio, &myctx);
 388         tocopy = howmany(length, aio_blksize);
 389         if (init_iocb(aio_maxio, aio_blksize) < 0) {
 390                 fprintf(stderr, "Error allocating the I/O buffers\n");
 391                 exit(1);
 392         }
 393
 394         while (tocopy > 0) {
 395                 int i, rc;
 396                 /* Submit as many reads as once as possible upto aio_maxio */
 397                 int n = MIN(MIN(aio_maxio - busy, aio_maxio),
 398                                 howmany(length - offset, aio_blksize));
 399                 if (n > 0) {
 400                         struct iocb *ioq[n];
 401
 402                         for (i = 0; i < n; i++) {
 403                                 struct iocb *io = alloc_iocb();
 404                                 int iosize = MIN(length - offset, aio_blksize);
 405
 406                                 if (zero) {
 407                                         /*
 408                                          * We are writing zero's to dstfd
 409                                          */
 410                                         io_prep_pwrite(io, dstfd, io->u.c.buf,
 411                                                         iosize, offset);
 412                                         io_set_callback(io, wr_done);
 413                                 } else {
 414                                         io_prep_pread(io, srcfd, io->u.c.buf,
 415                                                         iosize, offset);
 416                                         io_set_callback(io, rd_done);
 417                                 }
 418                                 ioq[i] = io;
 419                                 offset += iosize;
 420                         }
 421
 422                         rc = io_submit(myctx, n, ioq);
 423                         if (rc < 0)
 424                                 io_error("io_submit", rc);
 425
 426                         busy += n;
 427                         if (debug > 1)
 428                                 printf("io_submit(%d) busy:%d\n", n, busy);
 429                         if (delay.tv_usec) {
 430                                 struct timeval t = delay;
 431                                 (void) select(0, 0, 0, 0, &t);
 432                         }
 433                 }
 434
 435                 /*
 436                  * We have submitted all the I/O requests.
 437                  * Wait for at least one to complete and call the callbacks.
 438                  */
 439                 count_io_q_waits++;
 440                 rc = io_wait_run(myctx, 0);
 441                 if (rc < 0)
 442                         io_error("io_wait_run", rc);
 443
 444                 if (debug > 1) {
 445                         printf("io_wait_run: rc == %d\n", rc);
 446                         printf("busy:%d aio_maxio:%d tocopy:%d\n",
 447                                         busy, aio_maxio, tocopy);
 448                 }
 449         }
 450
 451         if (srcfd != -1)
 452                 close(srcfd);
 453         if (dstfd != -1)
 454                 close(dstfd);
 455         exit(0);
 456 }
 457
 458 /*
 459  * Results look like:
 460  * [alanm@toolbox ~/MOT3]$ ../taio -d kernel-source-2.4.8-0.4g.ppc.rpm abc
 461  * rrrrrrrrrrrrrrrwwwrwrrwwrrwrwwrrwrwrwwrrwrwrrrrwwrwwwrrwrrrwwwwwwwwwwwwwwwww
 462  * rrrrrrrrrrrrrrwwwrrwrwrwrwrrwwwwwwwwwwwwwwrrrrrrrrrrrrrrrrrrwwwwrwrwwrwrwrwr
 463  * wrrrrrrrwwwwwwwwwwwwwrrrwrrrwrrwrwwwwwwwwwwrrrrwwrwrrrrrrrrrrrwwwwwwwwwwwrww
 464  * wwwrrrrrrrrwwrrrwwrwrwrwwwrrrrrrrwwwrrwwwrrwrwwwwwwwwrrrrrrrwwwrrrrrrrwwwwww
 465  * wwwwwwwrwrrrrrrrrwrrwrrwrrwrwrrrwrrrwrrrwrwwwwwwwwwwwwwwwwwwrrrwwwrrrrrrrrrr
 466  * rrwrrrrrrwrrwwwwwwwwwwwwwwwwrwwwrrwrwwrrrrrrrrrrrrrrrrrrrwwwwwwwwwwwwwwwwwww
 467  * rrrrrwrrwrwrwrrwrrrwwwwwwwwrrrrwrrrwrwwrwrrrwrrwrrrrwwwwwwwrwrwwwwrwwrrrwrrr
 468  * rrrwwwwwwwrrrrwwrrrrrrrrrrrrwrwrrrrwwwwwwwwwwwwwwrwrrrrwwwwrwrrrrwrwwwrrrwww
 469  * rwwrrrrrrrwrrrrrrrrrrrrwwwwrrrwwwrwrrwwwwwwwwwwwwwwwwwwwwwrrrrrrrwwwwwwwrw
 470  */