1 // SPDX-License-Identifier: GPL-2.0
3 * Copyright (c) 2004 SuSE, Inc. All Rights Reserved.
9 * will open or create each file on the command line, and start a series
12 * aio is done in a rotating loop. first file1 gets 8 requests, then
13 * file2, then file3 etc. As each file finishes writing, it is switched
16 * io buffers are aligned in case you want to do raw io
18 * compile with gcc -Wall -laio -lpthread -o aio-stress aio-stress.c
20 * run aio-stress -h to see the options
22 * Please mail Chris Mason (mason@suse.com) with bug reports or patches
24 #define _FILE_OFFSET_BITS 64
25 #define PROG_VERSION "0.21"
33 #include <sys/types.h>
47 #define RUN_FOREVER -1
50 #define O_DIRECT 040000 /* direct disk access hint */
66 * various globals, these are effectively read only by the time the threads
70 unsigned long page_size_mask;
73 int latency_stats = 0;
74 int completion_latency_stats = 0;
76 int iterations = RUN_FOREVER;
77 int max_io_submit = 0;
78 long rec_len = 64 * 1024;
82 off_t context_offset = 2 * 1024 * 1024;
86 char *unaligned_buffer = NULL;
87 char *aligned_buffer = NULL;
88 int padded_reclen = 0;
91 char *verify_buf = NULL;
97 /* pthread mutexes and other globals for keeping the threads in sync */
98 pthread_cond_t stage_cond = PTHREAD_COND_INITIALIZER;
99 pthread_mutex_t stage_mutex = PTHREAD_MUTEX_INITIALIZER;
100 int threads_ending = 0;
101 int threads_starting = 0;
102 struct timeval global_stage_start_time;
103 struct thread_info *global_thread_info;
106 * latencies during io_submit are measured, these are the
107 * granularities for deviations
110 int deviations[DEVIATIONS] = { 100, 250, 500, 1000, 5000, 10000 };
116 double deviations[DEVIATIONS];
119 /* container for a series of operations to a file */
121 /* already open file descriptor, valid for whatever operation you want */
124 /* starting byte of the operation */
127 /* ending byte of the operation */
130 /* size of the read/write buffer */
133 /* max number of pending requests before a wait is triggered */
136 /* current number of pending requests */
139 /* last error, zero if there were none */
142 /* total number of errors hit. */
145 /* read,write, random, etc */
148 /* number of ios that will get sent to aio */
151 /* number of ios we've already sent */
154 /* last offset used in an io operation */
157 /* stonewalled = 1 when we got cut off before submitting all our ios */
160 /* list management */
161 struct io_oper *next;
162 struct io_oper *prev;
164 struct timeval start_time;
169 /* a single io, and all the tracking needed for it */
171 /* note, iocb must go first! */
174 /* pointer to parent io operation struct */
175 struct io_oper *io_oper;
180 /* size of the aligned buffer (record size) */
183 /* state of this io unit (free, pending, done) */
186 /* result of last operation */
189 struct io_unit *next;
191 struct timeval io_start_time; /* time of io_submit */
198 /* allocated array of io_unit structs */
201 /* list of io units available for io */
202 struct io_unit *free_ious;
204 /* number of io units in the ios array */
207 /* number of io units in flight */
208 int num_global_pending;
210 /* preallocated array of iocb pointers, only used in run_active */
213 /* preallocated array of events */
214 struct io_event *events;
216 /* size of the events array */
217 int num_global_events;
219 /* latency stats for io_submit */
220 struct io_latency io_submit_latency;
222 /* list of operations still in progress, and of those finished */
223 struct io_oper *active_opers;
224 struct io_oper *finished_opers;
226 /* number of files this thread is doing io on */
229 /* how much io this thread did in the last stage */
230 double stage_mb_trans;
232 /* latency completion stats i/o time from io_submit until io_getevents */
233 struct io_latency io_completion_latency;
237 * return seconds between start_tv and stop_tv in double precision
239 static double time_since(struct timeval *start_tv, struct timeval *stop_tv)
243 sec = stop_tv->tv_sec - start_tv->tv_sec;
244 usec = stop_tv->tv_usec - start_tv->tv_usec;
245 if (sec > 0 && usec < 0) {
249 ret = sec + usec / (double)1000000;
256 * return seconds between start_tv and now in double precision
258 static double time_since_now(struct timeval *start_tv)
260 struct timeval stop_time;
261 gettimeofday(&stop_time, NULL);
262 return time_since(start_tv, &stop_time);
266 * Add latency info to latency struct
268 static void calc_latency(struct timeval *start_tv, struct timeval *stop_tv,
269 struct io_latency *lat)
273 delta = time_since(start_tv, stop_tv);
274 delta = delta * 1000;
276 if (delta > lat->max)
278 if (!lat->min || delta < lat->min)
281 lat->total_lat += delta;
282 for (i = 0 ; i < DEVIATIONS ; i++) {
283 if (delta < deviations[i]) {
284 lat->deviations[i]++;
290 static void oper_list_add(struct io_oper *oper, struct io_oper **list)
294 oper->prev = oper->next = oper;
297 oper->prev = (*list)->prev;
299 (*list)->prev->next = oper;
300 (*list)->prev = oper;
304 static void oper_list_del(struct io_oper *oper, struct io_oper **list)
306 if ((*list)->next == (*list)->prev && *list == (*list)->next) {
310 oper->prev->next = oper->next;
311 oper->next->prev = oper->prev;
316 /* worker func to check error fields in the io unit */
317 static int check_finished_io(struct io_unit *io) {
319 if (io->res != io->buf_size) {
322 fstat(io->io_oper->fd, &s);
325 * If file size is large enough for the read, then this short
328 if ((io->io_oper->rw == READ || io->io_oper->rw == RREAD) &&
329 s.st_size > (io->iocb.u.c.offset + io->res)) {
331 fprintf(stderr, "io err %lu (%s) op %d, off %Lu size %d\n",
332 io->res, strerror(-io->res), io->iocb.aio_lio_opcode,
333 io->iocb.u.c.offset, io->buf_size);
334 io->io_oper->last_err = io->res;
335 io->io_oper->num_err++;
339 if (verify && io->io_oper->rw == READ) {
340 if (memcmp(io->buf, verify_buf, io->io_oper->reclen)) {
341 fprintf(stderr, "verify error, file %s offset %Lu contents (offset:bad:good):\n",
342 io->io_oper->file_name, io->iocb.u.c.offset);
344 for (i = 0 ; i < io->io_oper->reclen ; i++) {
345 if (io->buf[i] != verify_buf[i]) {
346 fprintf(stderr, "%d:%c:%c ", i, io->buf[i], verify_buf[i]);
349 fprintf(stderr, "\n");
356 /* worker func to check the busy bits and get an io unit ready for use */
357 static int grab_iou(struct io_unit *io, struct io_oper *oper) {
358 if (io->busy == IO_PENDING)
361 io->busy = IO_PENDING;
367 char *stage_name(int rw) {
374 return "random write";
376 return "random read";
381 static inline double oper_mb_trans(struct io_oper *oper) {
382 return ((double)oper->started_ios * (double)oper->reclen) /
383 (double)(1024 * 1024);
386 static void print_time(struct io_oper *oper) {
391 runtime = time_since_now(&oper->start_time);
392 mb = oper_mb_trans(oper);
394 fprintf(stderr, "%s on %s (%.2f MB/s) %.2f MB in %.2fs\n",
395 stage_name(oper->rw), oper->file_name, tput, mb, runtime);
398 static void print_lat(char *str, struct io_latency *lat) {
399 double avg = lat->total_lat / lat->total_io;
401 double total_counted = 0;
402 fprintf(stderr, "%s min %.2f avg %.2f max %.2f\n\t",
403 str, lat->min, avg, lat->max);
405 for (i = 0 ; i < DEVIATIONS ; i++) {
406 fprintf(stderr, " %.0f < %d", lat->deviations[i], deviations[i]);
407 total_counted += lat->deviations[i];
409 if (total_counted && lat->total_io - total_counted)
410 fprintf(stderr, " < %.0f", lat->total_io - total_counted);
411 fprintf(stderr, "\n");
412 memset(lat, 0, sizeof(*lat));
415 static void print_latency(struct thread_info *t)
417 struct io_latency *lat = &t->io_submit_latency;
418 print_lat("latency", lat);
421 static void print_completion_latency(struct thread_info *t)
423 struct io_latency *lat = &t->io_completion_latency;
424 print_lat("completion latency", lat);
428 * updates the fields in the io operation struct that belongs to this
429 * io unit, and make the io unit reusable again
431 void finish_io(struct thread_info *t, struct io_unit *io, long result,
432 struct timeval *tv_now) {
433 struct io_oper *oper = io->io_oper;
435 calc_latency(&io->io_start_time, tv_now, &t->io_completion_latency);
438 io->next = t->free_ious;
441 t->num_global_pending--;
442 check_finished_io(io);
443 if (oper->num_pending == 0 &&
444 (oper->started_ios == oper->total_ios || oper->stonewalled))
450 int read_some_events(struct thread_info *t) {
451 struct io_unit *event_io;
452 struct io_event *event;
455 int min_nr = io_iter;
456 struct timeval stop_time;
458 if (t->num_global_pending < io_iter)
459 min_nr = t->num_global_pending;
462 nr = io_getevents(t->io_ctx, min_nr, t->num_global_events, t->events,NULL);
464 nr = io_getevents(t->io_ctx, t->num_global_events, t->events, NULL);
469 gettimeofday(&stop_time, NULL);
470 for (i = 0 ; i < nr ; i++) {
471 event = t->events + i;
472 event_io = (struct io_unit *)((unsigned long)event->obj);
473 finish_io(t, event_io, event->res, &stop_time);
479 * finds a free io unit, waiting for pending requests if required. returns
480 * null if none could be found
482 static struct io_unit *find_iou(struct thread_info *t, struct io_oper *oper)
484 struct io_unit *event_io;
489 event_io = t->free_ious;
490 t->free_ious = t->free_ious->next;
491 if (grab_iou(event_io, oper)) {
492 fprintf(stderr, "io unit on free list but not free\n");
497 nr = read_some_events(t);
501 fprintf(stderr, "no free ious after read_some_events\n");
506 * wait for all pending requests for this io operation to finish
508 static int io_oper_wait(struct thread_info *t, struct io_oper *oper) {
509 struct io_event event;
510 struct io_unit *event_io;
516 if (oper->num_pending == 0)
519 /* this func is not speed sensitive, no need to go wild reading
520 * more than one event at a time
523 while(io_getevents(t->io_ctx, 1, 1, &event, NULL) > 0) {
525 while(io_getevents(t->io_ctx, 1, &event, NULL) > 0) {
527 struct timeval tv_now;
528 event_io = (struct io_unit *)((unsigned long)event.obj);
530 gettimeofday(&tv_now, NULL);
531 finish_io(t, event_io, event.res, &tv_now);
533 if (oper->num_pending == 0)
538 fprintf(stderr, "%u errors on oper, last %u\n",
539 oper->num_err, oper->last_err);
544 off_t random_byte_offset(struct io_oper *oper) {
546 off_t rand_byte = oper->start;
550 range = (oper->end - oper->start) / (1024 * 1024);
551 if ((page_size_mask+1) > (1024 * 1024))
552 offset = (page_size_mask+1) / (1024 * 1024);
558 /* find a random mb offset */
559 num = 1 + (int)((double)range * rand() / (RAND_MAX + 1.0 ));
560 rand_byte += num * 1024 * 1024;
562 /* find a random byte offset */
563 num = 1 + (int)((double)(1024 * 1024) * rand() / (RAND_MAX + 1.0));
566 num = (num + page_size_mask) & ~page_size_mask;
569 if (rand_byte + oper->reclen > oper->end) {
570 rand_byte -= oper->reclen;
576 * build an aio iocb for an operation, based on oper->rw and the
577 * last offset used. This finds the struct io_unit that will be attached
578 * to the iocb, and things are ready for submission to aio after this
581 * returns null on error
583 static struct io_unit *build_iocb(struct thread_info *t, struct io_oper *oper)
588 io = find_iou(t, oper);
590 fprintf(stderr, "unable to find io unit\n");
596 io_prep_pwrite(&io->iocb,oper->fd, io->buf, oper->reclen,
598 oper->last_offset += oper->reclen;
601 io_prep_pread(&io->iocb,oper->fd, io->buf, oper->reclen,
603 oper->last_offset += oper->reclen;
606 rand_byte = random_byte_offset(oper);
607 oper->last_offset = rand_byte;
608 io_prep_pread(&io->iocb,oper->fd, io->buf, oper->reclen,
612 rand_byte = random_byte_offset(oper);
613 oper->last_offset = rand_byte;
614 io_prep_pwrite(&io->iocb,oper->fd, io->buf, oper->reclen,
624 * wait for any pending requests, and then free all ram associated with
625 * an operation. returns the last error the operation hit (zero means none)
628 finish_oper(struct thread_info *t, struct io_oper *oper)
630 unsigned long last_err;
632 io_oper_wait(t, oper);
633 last_err = oper->last_err;
634 if (oper->num_pending > 0) {
635 fprintf(stderr, "oper num_pending is %d\n", oper->num_pending);
643 * allocates an io operation and fills in all the fields. returns
646 static struct io_oper *
647 create_oper(int fd, int rw, off_t start, off_t end, int reclen, int depth,
648 int iter, char *file_name)
650 struct io_oper *oper;
652 oper = malloc (sizeof(*oper));
654 fprintf(stderr, "unable to allocate io oper\n");
657 memset(oper, 0, sizeof(*oper));
662 oper->last_offset = oper->start;
664 oper->reclen = reclen;
666 oper->total_ios = (oper->end - oper->start) / oper->reclen;
667 oper->file_name = file_name;
673 * does setup on num_ios worth of iocbs, but does not actually
676 int build_oper(struct thread_info *t, struct io_oper *oper, int num_ios,
677 struct iocb **my_iocbs)
682 if (oper->started_ios == 0)
683 gettimeofday(&oper->start_time, NULL);
686 num_ios = oper->total_ios;
688 if ((oper->started_ios + num_ios) > oper->total_ios)
689 num_ios = oper->total_ios - oper->started_ios;
691 for( i = 0 ; i < num_ios ; i++) {
692 io = build_iocb(t, oper);
696 my_iocbs[i] = &io->iocb;
702 * runs through the iocbs in the array provided and updates
703 * counters in the associated oper struct
705 static void update_iou_counters(struct iocb **my_iocbs, int nr,
706 struct timeval *tv_now)
710 for (i = 0 ; i < nr ; i++) {
711 io = (struct io_unit *)(my_iocbs[i]);
712 io->io_oper->num_pending++;
713 io->io_oper->started_ios++;
714 io->io_start_time = *tv_now; /* set time of io_submit */
718 /* starts some io for a given file, returns zero if all went well */
719 int run_built(struct thread_info *t, int num_ios, struct iocb **my_iocbs)
722 struct timeval start_time;
723 struct timeval stop_time;
726 gettimeofday(&start_time, NULL);
727 ret = io_submit(t->io_ctx, num_ios, my_iocbs);
728 gettimeofday(&stop_time, NULL);
729 calc_latency(&start_time, &stop_time, &t->io_submit_latency);
731 if (ret != num_ios) {
732 /* some ios got through */
734 update_iou_counters(my_iocbs, ret, &stop_time);
736 t->num_global_pending += ret;
740 * we've used all the requests allocated in aio_init, wait and
743 if (ret > 0 || ret == -EAGAIN) {
745 if ((ret = read_some_events(t) > 0)) {
748 fprintf(stderr, "ret was %d and now is %d\n", ret, old_ret);
753 fprintf(stderr, "ret %d (%s) on io_submit\n", ret, strerror(-ret));
756 update_iou_counters(my_iocbs, ret, &stop_time);
757 t->num_global_pending += ret;
762 * changes oper->rw to the next in a command sequence, or returns zero
763 * to say this operation is really, completely done for
765 static int restart_oper(struct io_oper *oper) {
770 /* this switch falls through */
773 if (stages & (1 << READ))
776 if (!new_rw && stages & (1 << RWRITE))
779 if (!new_rw && stages & (1 << RREAD))
784 oper->started_ios = 0;
785 oper->last_offset = oper->start;
786 oper->stonewalled = 0;
789 * we're restarting an operation with pending requests, so the
790 * timing info won't be printed by finish_io. Printing it here
792 if (oper->num_pending)
801 static int oper_runnable(struct io_oper *oper) {
805 /* first context is always runnable, if started_ios > 0, no need to
806 * redo the calculations
808 if (oper->started_ios || oper->start == 0)
811 * only the sequential phases force delays in starting */
812 if (oper->rw >= RWRITE)
814 ret = fstat(oper->fd, &buf);
819 if (S_ISREG(buf.st_mode) && buf.st_size < oper->start)
825 * runs through all the io operations on the active list, and starts
826 * a chunk of io on each. If any io operations are completely finished,
827 * it either switches them to the next stage or puts them on the
830 * this function stops after max_io_submit iocbs are sent down the
831 * pipe, even if it has not yet touched all the operations on the
832 * active list. Any operations that have finished are moved onto
833 * the finished_opers list.
835 static int run_active_list(struct thread_info *t,
839 struct io_oper *oper;
840 struct io_oper *built_opers = NULL;
841 struct iocb **my_iocbs = t->iocbs;
845 oper = t->active_opers;
847 if (!oper_runnable(oper)) {
849 if (oper == t->active_opers)
853 ret = build_oper(t, oper, io_iter, my_iocbs);
857 oper_list_del(oper, &t->active_opers);
858 oper_list_add(oper, &built_opers);
859 oper = t->active_opers;
860 if (num_built + io_iter > max_io_submit)
866 ret = run_built(t, num_built, t->iocbs);
868 fprintf(stderr, "error %d on run_built\n", ret);
873 oper_list_del(oper, &built_opers);
874 oper_list_add(oper, &t->active_opers);
875 if (oper->started_ios == oper->total_ios) {
876 oper_list_del(oper, &t->active_opers);
877 oper_list_add(oper, &t->finished_opers);
887 if (use_shm != USE_SHM)
890 ret = shmctl(shm_id, IPC_RMID, &ds);
892 perror("shmctl IPC_RMID");
896 void aio_setup(io_context_t *io_ctx, int n)
898 int res = io_queue_init(n, io_ctx);
900 fprintf(stderr, "io_queue_setup(%d) returned %d (%s)\n",
901 n, res, strerror(-res));
907 * allocate io operation and event arrays for a given thread
909 int setup_ious(struct thread_info *t,
910 int num_files, int depth,
911 int reclen, int max_io_submit) {
913 size_t bytes = num_files * depth * sizeof(*t->ios);
915 t->ios = malloc(bytes);
917 fprintf(stderr, "unable to allocate io units\n");
920 memset(t->ios, 0, bytes);
922 for (i = 0 ; i < depth * num_files; i++) {
923 t->ios[i].buf = aligned_buffer;
924 aligned_buffer += padded_reclen;
925 t->ios[i].buf_size = reclen;
927 memset(t->ios[i].buf, 'b', reclen);
929 memset(t->ios[i].buf, 0, reclen);
930 t->ios[i].next = t->free_ious;
931 t->free_ious = t->ios + i;
934 verify_buf = aligned_buffer;
935 memset(verify_buf, 'b', reclen);
938 t->iocbs = malloc(sizeof(struct iocb *) * max_io_submit);
940 fprintf(stderr, "unable to allocate iocbs\n");
944 memset(t->iocbs, 0, max_io_submit * sizeof(struct iocb *));
946 t->events = malloc(sizeof(struct io_event) * depth * num_files);
948 fprintf(stderr, "unable to allocate ram for events\n");
951 memset(t->events, 0, num_files * sizeof(struct io_event)*depth);
953 t->num_global_ios = num_files * depth;
954 t->num_global_events = t->num_global_ios;
968 * The buffers used for file data are allocated as a single big
969 * malloc, and then each thread and operation takes a piece and uses
970 * that for file data. This lets us do a large shm or bigpages alloc
971 * and without trying to find a special place in each thread to map the
974 int setup_shared_mem(int num_threads, int num_files, int depth,
975 int reclen, int max_io_submit)
980 padded_reclen = (reclen + page_size_mask) / (page_size_mask+1);
981 padded_reclen = padded_reclen * (page_size_mask+1);
982 total_ram = num_files * depth * padded_reclen + num_threads;
984 total_ram += padded_reclen;
986 if (use_shm == USE_MALLOC) {
987 p = malloc(total_ram + page_size_mask);
988 } else if (use_shm == USE_SHM) {
989 shm_id = shmget(IPC_PRIVATE, total_ram, IPC_CREAT | 0700);
995 p = shmat(shm_id, (char *)0x50000000, 0);
1000 /* won't really be dropped until we shmdt */
1002 } else if (use_shm == USE_SHMFS) {
1003 char mmap_name[16]; /* /dev/shm/ + null + XXXXXX */
1006 strcpy(mmap_name, "/dev/shm/XXXXXX");
1007 fd = mkstemp(mmap_name);
1013 ftruncate(fd, total_ram);
1015 p = mmap((char *)0x50000000, total_ram,
1016 PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
1018 if (p == MAP_FAILED) {
1024 fprintf(stderr, "unable to allocate buffers\n");
1027 unaligned_buffer = p;
1028 p = (char*)((intptr_t) (p + page_size_mask) & ~page_size_mask);
1034 if (unaligned_buffer)
1035 free(unaligned_buffer);
1040 * runs through all the thread_info structs and calculates a combined
1043 void global_thread_throughput(struct thread_info *t, char *this_stage) {
1045 double runtime = time_since_now(&global_stage_start_time);
1046 double total_mb = 0;
1047 double min_trans = 0;
1049 for (i = 0 ; i < num_threads ; i++) {
1050 total_mb += global_thread_info[i].stage_mb_trans;
1051 if (!min_trans || t->stage_mb_trans < min_trans)
1052 min_trans = t->stage_mb_trans;
1055 fprintf(stderr, "%s throughput (%.2f MB/s) ", this_stage,
1056 total_mb / runtime);
1057 fprintf(stderr, "%.2f MB in %.2fs", total_mb, runtime);
1059 fprintf(stderr, " min transfer %.2fMB", min_trans);
1060 fprintf(stderr, "\n");
1065 /* this is the meat of the state machine. There is a list of
1066 * active operations structs, and as each one finishes the required
1067 * io it is moved to a list of finished operations. Once they have
1068 * all finished whatever stage they were in, they are given the chance
1069 * to restart and pick a different stage (read/write/random read etc)
1071 * various timings are printed in between the stages, along with
1072 * thread synchronization if there are more than one threads.
1074 int worker(struct thread_info *t)
1076 struct io_oper *oper;
1077 char *this_stage = NULL;
1078 struct timeval stage_time;
1083 aio_setup(&t->io_ctx, 512);
1086 if (num_threads > 1) {
1087 pthread_mutex_lock(&stage_mutex);
1089 if (threads_starting == num_threads) {
1091 gettimeofday(&global_stage_start_time, NULL);
1092 pthread_cond_broadcast(&stage_cond);
1094 while (threads_starting != num_threads)
1095 pthread_cond_wait(&stage_cond, &stage_mutex);
1096 pthread_mutex_unlock(&stage_mutex);
1098 if (t->active_opers) {
1099 this_stage = stage_name(t->active_opers->rw);
1100 gettimeofday(&stage_time, NULL);
1101 t->stage_mb_trans = 0;
1105 /* first we send everything through aio */
1106 while(t->active_opers && (cnt < iterations || iterations == RUN_FOREVER)) {
1107 if (stonewall && threads_ending) {
1108 oper = t->active_opers;
1109 oper->stonewalled = 1;
1110 oper_list_del(oper, &t->active_opers);
1111 oper_list_add(oper, &t->finished_opers);
1113 run_active_list(t, io_iter, max_io_submit);
1120 if (completion_latency_stats)
1121 print_completion_latency(t);
1123 /* then we wait for all the operations to finish */
1124 oper = t->finished_opers;
1128 io_oper_wait(t, oper);
1130 } while(oper != t->finished_opers);
1132 /* then we do an fsync to get the timing for any future operations
1133 * right, and check to see if any of these need to get restarted
1135 oper = t->finished_opers;
1139 t->stage_mb_trans += oper_mb_trans(oper);
1140 if (restart_oper(oper)) {
1141 oper_list_del(oper, &t->finished_opers);
1142 oper_list_add(oper, &t->active_opers);
1143 oper = t->finished_opers;
1147 if (oper == t->finished_opers)
1151 if (t->stage_mb_trans && t->num_files > 0) {
1152 double seconds = time_since_now(&stage_time);
1153 fprintf(stderr, "thread %llu %s totals (%.2f MB/s) %.2f MB in %.2fs\n",
1154 (unsigned long long)(t - global_thread_info), this_stage,
1155 t->stage_mb_trans/seconds, t->stage_mb_trans, seconds);
1158 if (num_threads > 1) {
1159 pthread_mutex_lock(&stage_mutex);
1161 if (threads_ending == num_threads) {
1162 threads_starting = 0;
1163 pthread_cond_broadcast(&stage_cond);
1164 global_thread_throughput(t, this_stage);
1166 while(threads_ending != num_threads)
1167 pthread_cond_wait(&stage_cond, &stage_mutex);
1168 pthread_mutex_unlock(&stage_mutex);
1171 /* someone got restarted, go back to the beginning */
1172 if (t->active_opers && (cnt < iterations || iterations == RUN_FOREVER)) {
1177 /* finally, free all the ram */
1178 while(t->finished_opers) {
1179 oper = t->finished_opers;
1180 oper_list_del(oper, &t->finished_opers);
1181 status = finish_oper(t, oper);
1184 if (t->num_global_pending) {
1185 fprintf(stderr, "global num pending is %d\n", t->num_global_pending);
1187 io_queue_release(t->io_ctx);
1192 typedef void * (*start_routine)(void *);
1193 int run_workers(struct thread_info *t, int num_threads)
1198 for(i = 0 ; i < num_threads ; i++) {
1199 ret = pthread_create(&t[i].tid, NULL, (start_routine)worker, t + i);
1201 perror("pthread_create");
1205 for(i = 0 ; i < num_threads ; i++) {
1206 ret = pthread_join(t[i].tid, NULL);
1208 perror("pthread_join");
1215 off_t parse_size(char *size_arg, off_t mult) {
1219 c = size_arg[strlen(size_arg) - 1];
1221 size_arg[strlen(size_arg) - 1] = '\0';
1223 num = atoi(size_arg);
1227 mult = 1024 * 1024 * 1024;
1246 void print_usage(void) {
1247 printf("usage: aio-stress [-s size] [-r size] [-a size] [-d num] [-b num]\n");
1248 printf(" [-i num] [-t num] [-c num] [-C size] [-nxhOS ]\n");
1249 printf(" file1 [file2 ...]\n");
1250 printf("\t-a size in KB at which to align buffers\n");
1251 printf("\t-b max number of iocbs to give io_submit at once\n");
1252 printf("\t-c number of io contexts per file\n");
1253 printf("\t-C offset between contexts, default 2MB\n");
1254 printf("\t-s size in MB of the test file(s), default 1024MB\n");
1255 printf("\t-r record size in KB used for each io, default 64KB\n");
1256 printf("\t-d number of pending aio requests for each file, default 64\n");
1257 printf("\t-i number of ios per file sent before switching\n\t to the next file, default 8\n");
1258 printf("\t-I total number of ayncs IOs the program will run, default is run until Cntl-C\n");
1259 printf("\t-O Use O_DIRECT (not available in 2.4 kernels),\n");
1260 printf("\t-S Use O_SYNC for writes\n");
1261 printf("\t-o add an operation to the list: write=0, read=1,\n");
1262 printf("\t random write=2, random read=3.\n");
1263 printf("\t repeat -o to specify multiple ops: -o 0 -o 1 etc.\n");
1264 printf("\t-m shm use ipc shared memory for io buffers instead of malloc\n");
1265 printf("\t-m shmfs mmap a file in /dev/shm for io buffers\n");
1266 printf("\t-n no fsyncs between write stage and read stage\n");
1267 printf("\t-l print io_submit latencies after each stage\n");
1268 printf("\t-L print io completion latencies after each stage\n");
1269 printf("\t-t number of threads to run\n");
1270 printf("\t-u unlink files after completion\n");
1271 printf("\t-v verification of bytes written\n");
1272 printf("\t-x turn off thread stonewalling\n");
1273 printf("\t-h this message\n");
1274 printf("\n\t the size options (-a -s and -r) allow modifiers -s 400{k,m,g}\n");
1275 printf("\t translate to 400KB, 400MB and 400GB\n");
1276 printf("version %s\n", PROG_VERSION);
1279 int main(int ac, char **av)
1286 off_t file_size = 1 * 1024 * 1024 * 1024;
1287 int first_stage = WRITE;
1288 struct io_oper *oper;
1292 struct thread_info *t;
1294 page_size_mask = getpagesize() - 1;
1297 c = getopt(ac, av, "a:b:c:C:m:s:r:d:i:I:o:t:lLnhOSxvu");
1303 page_size_mask = parse_size(optarg, 1024);
1307 num_contexts = atoi(optarg);
1310 context_offset = parse_size(optarg, 1024 * 1024);
1312 max_io_submit = atoi(optarg);
1315 file_size = parse_size(optarg, 1024 * 1024);
1318 depth = atoi(optarg);
1321 rec_len = parse_size(optarg, 1024);
1324 io_iter = atoi(optarg);
1327 iterations = atoi(optarg);
1336 completion_latency_stats = 1;
1339 if (!strcmp(optarg, "shm")) {
1340 fprintf(stderr, "using ipc shm\n");
1342 } else if (!strcmp(optarg, "shmfs")) {
1343 fprintf(stderr, "using /dev/shm for buffers\n");
1344 use_shm = USE_SHMFS;
1350 fprintf(stderr, "adding stage %s\n", stage_name(i));
1353 o_direct = O_DIRECT;
1359 num_threads = atoi(optarg);
1378 * make sure we don't try to submit more ios than we have allocated
1381 if (depth < io_iter) {
1383 fprintf(stderr, "dropping io_iter to %d\n", io_iter);
1391 num_files = ac - optind;
1393 if (num_threads > (num_files * num_contexts)) {
1394 num_threads = num_files * num_contexts;
1395 fprintf(stderr, "dropping thread count to the number of contexts %d\n",
1399 t = calloc(num_threads, sizeof(*t));
1404 global_thread_info = t;
1406 /* by default, allow a huge number of iocbs to be sent towards
1410 max_io_submit = num_files * io_iter * num_contexts;
1413 * make sure we don't try to submit more ios than max_io_submit allows
1415 if (max_io_submit < io_iter) {
1416 io_iter = max_io_submit;
1417 fprintf(stderr, "dropping io_iter to %d\n", io_iter);
1421 stages = (1 << WRITE) | (1 << READ) | (1 << RREAD) | (1 << RWRITE);
1423 for (i = 0 ; i < LAST_STAGE; i++) {
1424 if (stages & (1 << i)) {
1426 fprintf(stderr, "starting with %s\n", stage_name(i));
1432 if (file_size < num_contexts * context_offset) {
1433 fprintf(stderr, "file size %Lu too small for %d contexts\n",
1434 (unsigned long long)file_size, num_contexts);
1438 fprintf(stderr, "file size %LuMB, record size %luKB, depth %d, ios per iteration %d\n",
1439 (unsigned long long)file_size / (1024 * 1024),
1440 rec_len / 1024, depth, io_iter);
1441 fprintf(stderr, "max io_submit %d, buffer alignment set to %luKB\n",
1442 max_io_submit, (page_size_mask + 1)/1024);
1443 fprintf(stderr, "threads %d files %d contexts %d context offset %LuMB verification %s\n",
1444 num_threads, num_files, num_contexts,
1445 (unsigned long long)context_offset / (1024 * 1024),
1446 verify ? "on" : "off");
1447 /* open all the files and do any required setup for them */
1448 for (i = optind ; i < ac ; i++) {
1450 for (j = 0 ; j < num_contexts ; j++) {
1451 thread_index = open_fds % num_threads;
1454 rwfd = open(av[i], O_CREAT | O_RDWR | o_direct | o_sync, 0600);
1457 oper = create_oper(rwfd, first_stage, j * context_offset,
1458 file_size - j * context_offset, rec_len,
1459 depth, io_iter, av[i]);
1461 fprintf(stderr, "error in create_oper\n");
1464 oper_list_add(oper, &t[thread_index].active_opers);
1465 t[thread_index].num_files++;
1468 if (setup_shared_mem(num_threads, num_files * num_contexts,
1469 depth, rec_len, max_io_submit))
1473 for (i = 0 ; i < num_threads ; i++) {
1474 if (setup_ious(&t[i], t[i].num_files, depth, rec_len, max_io_submit))
1477 if (num_threads > 1){
1478 printf("Running multi thread version num_threads:%d\n", num_threads);
1479 run_workers(t, num_threads);
1481 printf("Running single thread version \n");
1485 for (i = optind ; i < ac ; i++) {
1486 printf("Cleaning up file %s \n", av[i]);