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 static 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 static struct iocb *alloc_iocb()
 104 {
 105         if (!iocb_free_count)
 106                 return 0;
 107         return iocb_free[--iocb_free_count];
 108 }
 109
 110 static 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 static 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,
 167                         "write missed bytes at %llu expected %lu got %ld\n",
 168                         iocb->u.c.offset, iocb->u.c.nbytes, res);
 169                 exit(1);
 170         }
 171         --tocopy;
 172         --busy;
 173         free_iocb(iocb);
 174         if (debug)
 175                 fprintf(stderr, "w");
 176 }
 177
 178 /*
 179  * Read complete callback.
 180  * Change read iocb into a write iocb and start it.
 181  */
 182 static void rd_done(io_context_t ctx, struct iocb *iocb, long res, long res2)
 183 {
 184         /* library needs accessors to look at iocb? */
 185         int iosize = iocb->u.c.nbytes;
 186         char *buf = iocb->u.c.buf;
 187         off_t offset = iocb->u.c.offset;
 188
 189         if (res2 != 0)
 190                 io_error("aio read", res2);
 191         if (res != iosize) {
 192                 fprintf(stderr,
 193                         "read missed bytes at %llu expected %lu got %ld\n",
 194                         iocb->u.c.offset, iocb->u.c.nbytes, res);
 195                 exit(1);
 196         }
 197
 198         /* turn read into write */
 199         if (no_write) {
 200                 --tocopy;
 201                 --busy;
 202                 free_iocb(iocb);
 203         } else {
 204                 io_prep_pwrite(iocb, dstfd, buf, iosize, offset);
 205                 io_set_callback(iocb, wr_done);
 206                 res = io_submit(ctx, 1, &iocb);
 207                 if (res != 1)
 208                         io_error("io_submit write", res);
 209         }
 210         if (debug)
 211                 fprintf(stderr, "r");
 212         if (debug > 1)
 213                 printf("%d", iosize);
 214 }
 215
 216 static void usage(void)
 217 {
 218         fprintf(stderr,
 219                 "Usage: aiocp [-a align] [-s size] [-b blksize] [-n num_io] [-f open_flag] SOURCE DEST\n"
 220                 "This copies from SOURCE to DEST using AIO.\n\n"
 221                 "Usage: aiocp [options] -w SOURCE\n"
 222                 "This does sequential AIO reads (no writes).\n\n"
 223                 "Usage: aiocp [options] -z DEST\n"
 224                 "This does sequential AIO writes of zeros.\n");
 225         exit(1);
 226 }
 227
 228 /*
 229  * Scale value by kilo, mega, or giga.
 230  */
 231 static long long scale_by_kmg(long long value, char scale)
 232 {
 233         switch (scale) {
 234         case 'g':
 235         case 'G':
 236                 value *= 1024;
 237                 fallthrough;
 238         case 'm':
 239         case 'M':
 240                 value *= 1024;
 241                 fallthrough;
 242         case 'k':
 243         case 'K':
 244                 value *= 1024;
 245                 break;
 246         case '\0':
 247                 break;
 248         default:
 249                 usage();
 250                 break;
 251         }
 252         return value;
 253 }
 254
 255 int main(int argc, char *const *argv)
 256 {
 257         struct stat st;
 258         off_t length = 0, offset = 0;
 259         io_context_t myctx;
 260         int c;
 261
 262         while ((c = getopt(argc, argv, "a:b:df:n:s:wzD:")) != -1) {
 263                 char *endp;
 264
 265                 switch (c) {
 266                 case 'a':       /* alignment of data buffer */
 267                         alignment = strtol(optarg, &endp, 0);
 268                         alignment = (long)scale_by_kmg((long long)alignment,
 269                                                         *endp);
 270                         break;
 271                 case 'f':       /* use these open flags */
 272                         if (strcmp(optarg, "LARGEFILE") == 0 ||
 273                             strcmp(optarg, "O_LARGEFILE") == 0) {
 274                                 source_open_flag |= O_LARGEFILE;
 275                                 dest_open_flag |= O_LARGEFILE;
 276                         } else if (strcmp(optarg, "TRUNC") == 0 ||
 277                                    strcmp(optarg, "O_TRUNC") == 0) {
 278                                 dest_open_flag |= O_TRUNC;
 279                         } else if (strcmp(optarg, "SYNC") == 0 ||
 280                                    strcmp(optarg, "O_SYNC") == 0) {
 281                                 dest_open_flag |= O_SYNC | O_NONBLOCK;
 282                         } else if (strcmp(optarg, "DIRECT") == 0 ||
 283                                    strcmp(optarg, "O_DIRECT") == 0) {
 284                                 source_open_flag |= O_DIRECT;
 285                                 dest_open_flag |= O_DIRECT;
 286                         } else if (strncmp(optarg, "CREAT", 5) == 0 ||
 287                                    strncmp(optarg, "O_CREAT", 5) == 0) {
 288                                 dest_open_flag |= O_CREAT;
 289                         }
 290                         break;
 291                 case 'd':
 292                         debug++;
 293                         break;
 294                 case 'D':
 295                         delay.tv_usec = atoi(optarg);
 296                         break;
 297                 case 'b':       /* block size */
 298                         aio_blksize = strtol(optarg, &endp, 0);
 299                         aio_blksize = (long)scale_by_kmg(
 300                                         (long long)aio_blksize, *endp);
 301                         break;
 302                 case 'n':       /* num io */
 303                         aio_maxio = strtol(optarg, &endp, 0);
 304                         break;
 305                 case 's':       /* size to transfer */
 306                         length = strtoll(optarg, &endp, 0);
 307                         length = scale_by_kmg(length, *endp);
 308                         break;
 309                 case 'w':       /* no write */
 310                         no_write = 1;
 311                         break;
 312                 case 'z':       /* write zero's */
 313                         zero = 1;
 314                         break;
 315
 316                 default:
 317                         usage();
 318                 }
 319         }
 320
 321         argc -= optind;
 322         argv += optind;
 323
 324 #ifndef DEBUG
 325         if (argc < 1)
 326                 usage();
 327 #else
 328         source_open_flag |= O_DIRECT;
 329         dest_open_flag |= O_DIRECT;
 330         aio_blksize = 1;
 331         aio_maxio = 1;
 332         srcname = "junkdata";
 333         dstname = "ff2";
 334 #endif
 335         if (!zero) {
 336 #ifndef DEBUG
 337                 srcfd = open(srcname = *argv, source_open_flag);
 338                 if (srcfd < 0) {
 339 #else
 340                 srcfd = open(srcname, source_open_flag);
 341                 if (srcfd < 0) {
 342 #endif
 343                         perror(srcname);
 344                         exit(1);
 345                 }
 346                 argv++;
 347                 argc--;
 348                 if (fstat(srcfd, &st) < 0) {
 349                         perror("fstat");
 350                         exit(1);
 351                 }
 352                 if (length == 0)
 353                         length = st.st_size;
 354         }
 355
 356         if (!no_write) {
 357                 /*
 358                  * We are either copying or writing zeros to dstname
 359                  */
 360 #ifndef DEBUG
 361                 if (argc < 1)
 362                         usage();
 363                 dstfd = open(dstname = *argv, dest_open_flag, 0666);
 364                 if (dstfd < 0) {
 365 #else
 366                 dstfd = open(dstname, dest_open_flag, 0666);
 367                 if (dstfd < 0) {
 368 #endif
 369                         perror(dstname);
 370                         exit(1);
 371                 }
 372                 if (zero) {
 373                         /*
 374                          * get size of dest, if we are zeroing it.
 375                          * TODO: handle devices.
 376                          */
 377                         if (fstat(dstfd, &st) < 0) {
 378                                 perror("fstat");
 379                                 exit(1);
 380                         }
 381                         if (length == 0)
 382                                 length = st.st_size;
 383                 }
 384         }
 385
 386         /* initialize state machine */
 387         memset(&myctx, 0, sizeof(myctx));
 388         io_queue_init(aio_maxio, &myctx);
 389         tocopy = howmany(length, aio_blksize);
 390         if (init_iocb(aio_maxio, aio_blksize) < 0) {
 391                 fprintf(stderr, "Error allocating the I/O buffers\n");
 392                 exit(1);
 393         }
 394
 395         while (tocopy > 0) {
 396                 int i, rc;
 397                 /* Submit as many reads as once as possible upto aio_maxio */
 398                 int n = MIN(MIN(aio_maxio - busy, aio_maxio),
 399                                 howmany(length - offset, aio_blksize));
 400                 if (n > 0) {
 401                         struct iocb *ioq[n];
 402
 403                         for (i = 0; i < n; i++) {
 404                                 struct iocb *io = alloc_iocb();
 405                                 int iosize = MIN(length - offset, aio_blksize);
 406
 407                                 if (zero) {
 408                                         /*
 409                                          * We are writing zero's to dstfd
 410                                          */
 411                                         io_prep_pwrite(io, dstfd, io->u.c.buf,
 412                                                         iosize, offset);
 413                                         io_set_callback(io, wr_done);
 414                                 } else {
 415                                         io_prep_pread(io, srcfd, io->u.c.buf,
 416                                                         iosize, offset);
 417                                         io_set_callback(io, rd_done);
 418                                 }
 419                                 ioq[i] = io;
 420                                 offset += iosize;
 421                         }
 422
 423                         rc = io_submit(myctx, n, ioq);
 424                         if (rc < 0)
 425                                 io_error("io_submit", rc);
 426
 427                         busy += n;
 428                         if (debug > 1)
 429                                 printf("io_submit(%d) busy:%d\n", n, busy);
 430                         if (delay.tv_usec) {
 431                                 struct timeval t = delay;
 432                                 (void) select(0, 0, 0, 0, &t);
 433                         }
 434                 }
 435
 436                 /*
 437                  * We have submitted all the I/O requests.
 438                  * Wait for at least one to complete and call the callbacks.
 439                  */
 440                 count_io_q_waits++;
 441                 rc = io_wait_run(myctx, 0);
 442                 if (rc < 0)
 443                         io_error("io_wait_run", rc);
 444
 445                 if (debug > 1) {
 446                         printf("io_wait_run: rc == %d\n", rc);
 447                         printf("busy:%d aio_maxio:%d tocopy:%d\n",
 448                                         busy, aio_maxio, tocopy);
 449                 }
 450         }
 451
 452         if (srcfd != -1)
 453                 close(srcfd);
 454         if (dstfd != -1)
 455                 close(dstfd);
 456         exit(0);
 457 }
 458
 459 /*
 460  * Results look like:
 461  * [alanm@toolbox ~/MOT3]$ ../taio -d kernel-source-2.4.8-0.4g.ppc.rpm abc
 462  * rrrrrrrrrrrrrrrwwwrwrrwwrrwrwwrrwrwrwwrrwrwrrrrwwrwwwrrwrrrwwwwwwwwwwwwwwwww
 463  * rrrrrrrrrrrrrrwwwrrwrwrwrwrrwwwwwwwwwwwwwwrrrrrrrrrrrrrrrrrrwwwwrwrwwrwrwrwr
 464  * wrrrrrrrwwwwwwwwwwwwwrrrwrrrwrrwrwwwwwwwwwwrrrrwwrwrrrrrrrrrrrwwwwwwwwwwwrww
 465  * wwwrrrrrrrrwwrrrwwrwrwrwwwrrrrrrrwwwrrwwwrrwrwwwwwwwwrrrrrrrwwwrrrrrrrwwwwww
 466  * wwwwwwwrwrrrrrrrrwrrwrrwrrwrwrrrwrrrwrrrwrwwwwwwwwwwwwwwwwwwrrrwwwrrrrrrrrrr
 467  * rrwrrrrrrwrrwwwwwwwwwwwwwwwwrwwwrrwrwwrrrrrrrrrrrrrrrrrrrwwwwwwwwwwwwwwwwwww
 468  * rrrrrwrrwrwrwrrwrrrwwwwwwwwrrrrwrrrwrwwrwrrrwrrwrrrrwwwwwwwrwrwwwwrwwrrrwrrr
 469  * rrrwwwwwwwrrrrwwrrrrrrrrrrrrwrwrrrrwwwwwwwwwwwwwwrwrrrrwwwwrwrrrrwrwwwrrrwww
 470  * rwwrrrrrrrwrrrrrrrrrrrrwwwwrrrwwwrwrrwwwwwwwwwwwwwwwwwwwwwrrrrrrrwwwwwwwrw
 471  */