2 * Copyright (c) 2004 SuSE, Inc. All Rights Reserved.
4 * This program is free software; you can redistribute it and/or modify it
5 * under the terms of version 2 of the GNU General Public License as
6 * published by the Free Software Foundation.
8 * This program is distributed in the hope that it would be useful, but
9 * WITHOUT ANY WARRANTY; without even the implied warranty of
10 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
12 * Further, this software is distributed without any warranty that it is
13 * free of the rightful claim of any third person regarding infringement
14 * or the like. Any license provided herein, whether implied or
15 * otherwise, applies only to this software file. Patent licenses, if
16 * any, provided herein do not apply to combinations of this program with
17 * other software, or any other product whatsoever.
19 * You should have received a copy of the GNU General Public License along
20 * with this program; if not, write the Free Software Foundation, Inc., 59
21 * Temple Place - Suite 330, Boston MA 02111-1307, USA.
23 * Contact information: Silicon Graphics, Inc., 1600 Amphitheatre Pkwy,
24 * Mountain View, CA 94043, or:
29 * will open or create each file on the command line, and start a series
32 * aio is done in a rotating loop. first file1 gets 8 requests, then
33 * file2, then file3 etc. As each file finishes writing, it is switched
36 * io buffers are aligned in case you want to do raw io
38 * compile with gcc -Wall -laio -lpthread -o aio-stress aio-stress.c
40 * run aio-stress -h to see the options
42 * Please mail Chris Mason (mason@suse.com) with bug reports or patches
44 #define _FILE_OFFSET_BITS 64
45 #define PROG_VERSION "0.18"
53 #include <sys/types.h>
67 #define RUN_FOREVER -1
70 #define O_DIRECT 040000 /* direct disk access hint */
86 * various globals, these are effectively read only by the time the threads
90 unsigned long page_size_mask;
93 int latency_stats = 0;
95 int iterations = RUN_FOREVER;
96 int max_io_submit = 0;
97 long rec_len = 64 * 1024;
100 int num_contexts = 1;
101 off_t context_offset = 2 * 1024 * 1024;
102 int fsync_stages = 1;
105 char *unaligned_buffer = NULL;
106 char *aligned_buffer = NULL;
107 int padded_reclen = 0;
110 char *verify_buf = NULL;
115 /* pthread mutexes and other globals for keeping the threads in sync */
116 pthread_cond_t stage_cond = PTHREAD_COND_INITIALIZER;
117 pthread_mutex_t stage_mutex = PTHREAD_MUTEX_INITIALIZER;
118 int threads_ending = 0;
119 int threads_starting = 0;
120 struct timeval global_stage_start_time;
121 struct thread_info *global_thread_info;
124 * latencies during io_submit are measured, these are the
125 * granularities for deviations
128 int deviations[DEVIATIONS] = { 100, 250, 500, 1000, 5000, 10000 };
134 double deviations[DEVIATIONS];
137 /* container for a series of operations to a file */
139 /* already open file descriptor, valid for whatever operation you want */
142 /* starting byte of the operation */
145 /* ending byte of the operation */
148 /* size of the read/write buffer */
151 /* max number of pending requests before a wait is triggered */
154 /* current number of pending requests */
157 /* last error, zero if there were none */
160 /* total number of errors hit. */
163 /* read,write, random, etc */
166 /* number of ios that will get sent to aio */
169 /* number of ios we've already sent */
172 /* last offset used in an io operation */
175 /* stonewalled = 1 when we got cut off before submitting all our ios */
178 /* list management */
179 struct io_oper *next;
180 struct io_oper *prev;
182 struct timeval start_time;
187 /* a single io, and all the tracking needed for it */
189 /* note, iocb must go first! */
192 /* pointer to parent io operation struct */
193 struct io_oper *io_oper;
198 /* size of the aligned buffer (record size) */
201 /* state of this io unit (free, pending, done) */
204 /* result of last operation */
207 struct io_unit *next;
214 /* allocated array of io_unit structs */
217 /* list of io units available for io */
218 struct io_unit *free_ious;
220 /* number of io units in the ios array */
223 /* number of io units in flight */
224 int num_global_pending;
226 /* preallocated array of iocb pointers, only used in run_active */
229 /* preallocated array of events */
230 struct io_event *events;
232 /* size of the events array */
233 int num_global_events;
235 /* latency stats for io_submit */
236 struct io_latency io_submit_latency;
238 /* list of operations still in progress, and of those finished */
239 struct io_oper *active_opers;
240 struct io_oper *finished_opers;
242 /* number of files this thread is doing io on */
245 /* how much io this thread did in the last stage */
246 double stage_mb_trans;
249 static double time_since(struct timeval *tv) {
253 gettimeofday(&stop, NULL);
254 sec = stop.tv_sec - tv->tv_sec;
255 usec = stop.tv_usec - tv->tv_usec;
256 if (sec > 0 && usec < 0) {
260 ret = sec + usec / (double)1000000;
266 static void calc_latency(struct timeval *tv, struct io_latency *lat)
270 delta = time_since(tv);
271 delta = delta * 1000;
273 if (delta > lat->max)
275 if (!lat->min || delta < lat->min)
278 lat->total_lat += delta;
279 for (i = 0 ; i < DEVIATIONS ; i++) {
280 if (delta < deviations[i]) {
281 lat->deviations[i]++;
287 static void oper_list_add(struct io_oper *oper, struct io_oper **list)
291 oper->prev = oper->next = oper;
294 oper->prev = (*list)->prev;
296 (*list)->prev->next = oper;
297 (*list)->prev = oper;
301 static void oper_list_del(struct io_oper *oper, struct io_oper **list)
303 if ((*list)->next == (*list)->prev && *list == (*list)->next) {
307 oper->prev->next = oper->next;
308 oper->next->prev = oper->prev;
313 /* worker func to check error fields in the io unit */
314 static int check_finished_io(struct io_unit *io) {
316 if (io->res != io->buf_size) {
319 fstat(io->io_oper->fd, &s);
322 * If file size is large enough for the read, then this short
325 if ((io->io_oper->rw == READ || io->io_oper->rw == RREAD) &&
326 s.st_size > (io->iocb.u.c.offset + io->res)) {
328 fprintf(stderr, "io err %lu (%s) op %d, off %Lu size %d\n",
329 io->res, strerror(-io->res), io->iocb.aio_lio_opcode,
330 io->iocb.u.c.offset, io->buf_size);
331 io->io_oper->last_err = io->res;
332 io->io_oper->num_err++;
336 if (verify && io->io_oper->rw == READ) {
337 if (memcmp(io->buf, verify_buf, io->io_oper->reclen)) {
338 fprintf(stderr, "verify error, file %s offset %Lu contents (offset:bad:good):\n",
339 io->io_oper->file_name, io->iocb.u.c.offset);
341 for (i = 0 ; i < io->io_oper->reclen ; i++) {
342 if (io->buf[i] != verify_buf[i]) {
343 fprintf(stderr, "%d:%c:%c ", i, io->buf[i], verify_buf[i]);
346 fprintf(stderr, "\n");
353 /* worker func to check the busy bits and get an io unit ready for use */
354 static int grab_iou(struct io_unit *io, struct io_oper *oper) {
355 if (io->busy == IO_PENDING)
358 io->busy = IO_PENDING;
364 char *stage_name(int rw) {
371 return "random write";
373 return "random read";
378 static inline double oper_mb_trans(struct io_oper *oper) {
379 return ((double)oper->started_ios * (double)oper->reclen) /
380 (double)(1024 * 1024);
383 static void print_time(struct io_oper *oper) {
388 runtime = time_since(&oper->start_time);
389 mb = oper_mb_trans(oper);
391 fprintf(stderr, "%s on %s (%.2f MB/s) %.2f MB in %.2fs\n",
392 stage_name(oper->rw), oper->file_name, tput, mb, runtime);
395 static void print_latency(struct thread_info *t) {
396 struct io_latency *lat = &t->io_submit_latency;
397 double avg = lat->total_lat / lat->total_io;
399 double total_counted = 0;
400 fprintf(stderr, "latency min %.2f avg %.2f max %.2f\n\t",
401 lat->min, avg, lat->max);
403 for (i = 0 ; i < DEVIATIONS ; i++) {
404 fprintf(stderr, " %.0f < %d", lat->deviations[i], deviations[i]);
405 total_counted += lat->deviations[i];
407 if (total_counted && lat->total_io - total_counted)
408 fprintf(stderr, " < %.0f", lat->total_io - total_counted);
409 fprintf(stderr, "\n");
410 memset(&t->io_submit_latency, 0, sizeof(t->io_submit_latency));
414 * updates the fields in the io operation struct that belongs to this
415 * io unit, and make the io unit reusable again
417 void finish_io(struct thread_info *t, struct io_unit *io, long result) {
418 struct io_oper *oper = io->io_oper;
422 io->next = t->free_ious;
425 t->num_global_pending--;
426 check_finished_io(io);
427 if (oper->num_pending == 0 &&
428 (oper->started_ios == oper->total_ios || oper->stonewalled))
434 int read_some_events(struct thread_info *t) {
435 struct io_unit *event_io;
436 struct io_event *event;
439 int min_nr = io_iter;
441 if (t->num_global_pending < io_iter)
442 min_nr = t->num_global_pending;
445 nr = io_getevents(t->io_ctx, min_nr, t->num_global_events, t->events,NULL);
447 nr = io_getevents(t->io_ctx, t->num_global_events, t->events, NULL);
452 for (i = 0 ; i < nr ; i++) {
453 event = t->events + i;
454 event_io = (struct io_unit *)((unsigned long)event->obj);
455 finish_io(t, event_io, event->res);
461 * finds a free io unit, waiting for pending requests if required. returns
462 * null if none could be found
464 static struct io_unit *find_iou(struct thread_info *t, struct io_oper *oper)
466 struct io_unit *event_io;
471 event_io = t->free_ious;
472 t->free_ious = t->free_ious->next;
473 if (grab_iou(event_io, oper)) {
474 fprintf(stderr, "io unit on free list but not free\n");
479 nr = read_some_events(t);
483 fprintf(stderr, "no free ious after read_some_events\n");
488 * wait for all pending requests for this io operation to finish
490 static int io_oper_wait(struct thread_info *t, struct io_oper *oper) {
491 struct io_event event;
492 struct io_unit *event_io;
498 if (oper->num_pending == 0)
501 /* this func is not speed sensitive, no need to go wild reading
502 * more than one event at a time
505 while(io_getevents(t->io_ctx, 1, 1, &event, NULL) > 0) {
507 while(io_getevents(t->io_ctx, 1, &event, NULL) > 0) {
509 event_io = (struct io_unit *)((unsigned long)event.obj);
511 finish_io(t, event_io, event.res);
513 if (oper->num_pending == 0)
518 fprintf(stderr, "%u errors on oper, last %u\n",
519 oper->num_err, oper->last_err);
524 off_t random_byte_offset(struct io_oper *oper) {
526 off_t rand_byte = oper->start;
530 range = (oper->end - oper->start) / (1024 * 1024);
531 if ((page_size_mask+1) > (1024 * 1024))
532 offset = (page_size_mask+1) / (1024 * 1024);
538 /* find a random mb offset */
539 num = 1 + (int)((double)range * rand() / (RAND_MAX + 1.0 ));
540 rand_byte += num * 1024 * 1024;
542 /* find a random byte offset */
543 num = 1 + (int)((double)(1024 * 1024) * rand() / (RAND_MAX + 1.0));
546 num = (num + page_size_mask) & ~page_size_mask;
549 if (rand_byte + oper->reclen > oper->end) {
550 rand_byte -= oper->reclen;
556 * build an aio iocb for an operation, based on oper->rw and the
557 * last offset used. This finds the struct io_unit that will be attached
558 * to the iocb, and things are ready for submission to aio after this
561 * returns null on error
563 static struct io_unit *build_iocb(struct thread_info *t, struct io_oper *oper)
568 io = find_iou(t, oper);
570 fprintf(stderr, "unable to find io unit\n");
576 io_prep_pwrite(&io->iocb,oper->fd, io->buf, oper->reclen,
578 oper->last_offset += oper->reclen;
581 io_prep_pread(&io->iocb,oper->fd, io->buf, oper->reclen,
583 oper->last_offset += oper->reclen;
586 rand_byte = random_byte_offset(oper);
587 oper->last_offset = rand_byte;
588 io_prep_pread(&io->iocb,oper->fd, io->buf, oper->reclen,
592 rand_byte = random_byte_offset(oper);
593 oper->last_offset = rand_byte;
594 io_prep_pwrite(&io->iocb,oper->fd, io->buf, oper->reclen,
604 * wait for any pending requests, and then free all ram associated with
605 * an operation. returns the last error the operation hit (zero means none)
608 finish_oper(struct thread_info *t, struct io_oper *oper)
610 unsigned long last_err;
612 io_oper_wait(t, oper);
613 last_err = oper->last_err;
614 if (oper->num_pending > 0) {
615 fprintf(stderr, "oper num_pending is %d\n", oper->num_pending);
623 * allocates an io operation and fills in all the fields. returns
626 static struct io_oper *
627 create_oper(int fd, int rw, off_t start, off_t end, int reclen, int depth,
628 int iter, char *file_name)
630 struct io_oper *oper;
632 oper = malloc (sizeof(*oper));
634 fprintf(stderr, "unable to allocate io oper\n");
637 memset(oper, 0, sizeof(*oper));
642 oper->last_offset = oper->start;
644 oper->reclen = reclen;
646 oper->total_ios = (oper->end - oper->start) / oper->reclen;
647 oper->file_name = file_name;
653 * does setup on num_ios worth of iocbs, but does not actually
656 int build_oper(struct thread_info *t, struct io_oper *oper, int num_ios,
657 struct iocb **my_iocbs)
662 if (oper->started_ios == 0)
663 gettimeofday(&oper->start_time, NULL);
666 num_ios = oper->total_ios;
668 if ((oper->started_ios + num_ios) > oper->total_ios)
669 num_ios = oper->total_ios - oper->started_ios;
671 for( i = 0 ; i < num_ios ; i++) {
672 io = build_iocb(t, oper);
676 my_iocbs[i] = &io->iocb;
682 * runs through the iocbs in the array provided and updates
683 * counters in the associated oper struct
685 static void update_iou_counters(struct iocb **my_iocbs, int nr)
689 for (i = 0 ; i < nr ; i++) {
690 io = (struct io_unit *)(my_iocbs[i]);
691 io->io_oper->num_pending++;
692 io->io_oper->started_ios++;
696 /* starts some io for a given file, returns zero if all went well */
697 int run_built(struct thread_info *t, int num_ios, struct iocb **my_iocbs)
700 struct timeval start_time;
703 gettimeofday(&start_time, NULL);
704 ret = io_submit(t->io_ctx, num_ios, my_iocbs);
705 calc_latency(&start_time, &t->io_submit_latency);
706 if (ret != num_ios) {
707 /* some ios got through */
709 update_iou_counters(my_iocbs, ret);
711 t->num_global_pending += ret;
715 * we've used all the requests allocated in aio_init, wait and
718 if (ret > 0 || ret == -EAGAIN) {
719 if ((ret = read_some_events(t) > 0)) {
724 fprintf(stderr, "ret %d (%s) on io_submit\n", ret, strerror(-ret));
727 update_iou_counters(my_iocbs, ret);
728 t->num_global_pending += ret;
733 * changes oper->rw to the next in a command sequence, or returns zero
734 * to say this operation is really, completely done for
736 static int restart_oper(struct io_oper *oper) {
741 /* this switch falls through */
744 if (stages & (1 << READ))
747 if (!new_rw && stages & (1 << RWRITE))
750 if (!new_rw && stages & (1 << RREAD))
755 oper->started_ios = 0;
756 oper->last_offset = oper->start;
757 oper->stonewalled = 0;
760 * we're restarting an operation with pending requests, so the
761 * timing info won't be printed by finish_io. Printing it here
763 if (oper->num_pending)
772 static int oper_runnable(struct io_oper *oper) {
776 /* first context is always runnable, if started_ios > 0, no need to
777 * redo the calculations
779 if (oper->started_ios || oper->start == 0)
782 * only the sequential phases force delays in starting */
783 if (oper->rw >= RWRITE)
785 ret = fstat(oper->fd, &buf);
790 if (S_ISREG(buf.st_mode) && buf.st_size < oper->start)
796 * runs through all the io operations on the active list, and starts
797 * a chunk of io on each. If any io operations are completely finished,
798 * it either switches them to the next stage or puts them on the
801 * this function stops after max_io_submit iocbs are sent down the
802 * pipe, even if it has not yet touched all the operations on the
803 * active list. Any operations that have finished are moved onto
804 * the finished_opers list.
806 static int run_active_list(struct thread_info *t,
810 struct io_oper *oper;
811 struct io_oper *built_opers = NULL;
812 struct iocb **my_iocbs = t->iocbs;
816 oper = t->active_opers;
818 if (!oper_runnable(oper)) {
820 if (oper == t->active_opers)
824 ret = build_oper(t, oper, io_iter, my_iocbs);
828 oper_list_del(oper, &t->active_opers);
829 oper_list_add(oper, &built_opers);
830 oper = t->active_opers;
831 if (num_built + io_iter > max_io_submit)
837 ret = run_built(t, num_built, t->iocbs);
839 fprintf(stderr, "error %d on run_built\n", ret);
844 oper_list_del(oper, &built_opers);
845 oper_list_add(oper, &t->active_opers);
846 if (oper->started_ios == oper->total_ios) {
847 oper_list_del(oper, &t->active_opers);
848 oper_list_add(oper, &t->finished_opers);
858 if (use_shm != USE_SHM)
861 ret = shmctl(shm_id, IPC_RMID, &ds);
863 perror("shmctl IPC_RMID");
867 void aio_setup(io_context_t *io_ctx, int n)
869 int res = io_queue_init(n, io_ctx);
871 fprintf(stderr, "io_queue_setup(%d) returned %d (%s)\n",
872 n, res, strerror(-res));
878 * allocate io operation and event arrays for a given thread
880 int setup_ious(struct thread_info *t,
881 int num_files, int depth,
882 int reclen, int max_io_submit) {
884 size_t bytes = num_files * depth * sizeof(*t->ios);
886 t->ios = malloc(bytes);
888 fprintf(stderr, "unable to allocate io units\n");
891 memset(t->ios, 0, bytes);
893 for (i = 0 ; i < depth * num_files; i++) {
894 t->ios[i].buf = aligned_buffer;
895 aligned_buffer += padded_reclen;
896 t->ios[i].buf_size = reclen;
898 memset(t->ios[i].buf, 'b', reclen);
900 memset(t->ios[i].buf, 0, reclen);
901 t->ios[i].next = t->free_ious;
902 t->free_ious = t->ios + i;
905 verify_buf = aligned_buffer;
906 memset(verify_buf, 'b', reclen);
909 t->iocbs = malloc(sizeof(struct iocb *) * max_io_submit);
911 fprintf(stderr, "unable to allocate iocbs\n");
915 memset(t->iocbs, 0, max_io_submit * sizeof(struct iocb *));
917 t->events = malloc(sizeof(struct io_event) * depth * num_files);
919 fprintf(stderr, "unable to allocate ram for events\n");
922 memset(t->events, 0, num_files * sizeof(struct io_event)*depth);
924 t->num_global_ios = num_files * depth;
925 t->num_global_events = t->num_global_ios;
939 * The buffers used for file data are allocated as a single big
940 * malloc, and then each thread and operation takes a piece and uses
941 * that for file data. This lets us do a large shm or bigpages alloc
942 * and without trying to find a special place in each thread to map the
945 int setup_shared_mem(int num_threads, int num_files, int depth,
946 int reclen, int max_io_submit)
951 padded_reclen = (reclen + page_size_mask) / (page_size_mask+1);
952 padded_reclen = padded_reclen * (page_size_mask+1);
953 total_ram = num_files * depth * padded_reclen + num_threads;
955 total_ram += padded_reclen;
957 if (use_shm == USE_MALLOC) {
958 p = malloc(total_ram + page_size_mask);
959 } else if (use_shm == USE_SHM) {
960 shm_id = shmget(IPC_PRIVATE, total_ram, IPC_CREAT | 0700);
966 p = shmat(shm_id, (char *)0x50000000, 0);
971 /* won't really be dropped until we shmdt */
973 } else if (use_shm == USE_SHMFS) {
974 char mmap_name[16]; /* /dev/shm/ + null + XXXXXX */
977 strcpy(mmap_name, "/dev/shm/XXXXXX");
978 fd = mkstemp(mmap_name);
984 ftruncate(fd, total_ram);
986 p = mmap((char *)0x50000000, total_ram,
987 PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
989 if (p == MAP_FAILED) {
995 fprintf(stderr, "unable to allocate buffers\n");
998 unaligned_buffer = p;
999 p = (char*)((intptr_t) (p + page_size_mask) & ~page_size_mask);
1005 if (unaligned_buffer)
1006 free(unaligned_buffer);
1011 * runs through all the thread_info structs and calculates a combined
1014 void global_thread_throughput(struct thread_info *t, char *this_stage) {
1016 double runtime = time_since(&global_stage_start_time);
1017 double total_mb = 0;
1018 double min_trans = 0;
1020 for (i = 0 ; i < num_threads ; i++) {
1021 total_mb += global_thread_info[i].stage_mb_trans;
1022 if (!min_trans || t->stage_mb_trans < min_trans)
1023 min_trans = t->stage_mb_trans;
1026 fprintf(stderr, "%s throughput (%.2f MB/s) ", this_stage,
1027 total_mb / runtime);
1028 fprintf(stderr, "%.2f MB in %.2fs", total_mb, runtime);
1030 fprintf(stderr, " min transfer %.2fMB", min_trans);
1031 fprintf(stderr, "\n");
1036 /* this is the meat of the state machine. There is a list of
1037 * active operations structs, and as each one finishes the required
1038 * io it is moved to a list of finished operations. Once they have
1039 * all finished whatever stage they were in, they are given the chance
1040 * to restart and pick a different stage (read/write/random read etc)
1042 * various timings are printed in between the stages, along with
1043 * thread synchronization if there are more than one threads.
1045 int worker(struct thread_info *t)
1047 struct io_oper *oper;
1048 char *this_stage = NULL;
1049 struct timeval stage_time;
1054 aio_setup(&t->io_ctx, 512);
1057 printf("Starting %s iter:%d \n", __FUNCTION__,iteration);
1058 if (num_threads > 1) {
1059 printf("num_threads %d \n", num_threads);
1060 pthread_mutex_lock(&stage_mutex);
1062 if (threads_starting == num_threads) {
1064 gettimeofday(&global_stage_start_time, NULL);
1065 pthread_cond_broadcast(&stage_cond);
1067 while (threads_starting != num_threads)
1068 pthread_cond_wait(&stage_cond, &stage_mutex);
1069 pthread_mutex_unlock(&stage_mutex);
1071 if (t->active_opers) {
1072 // printf("active_opers %p line:%d\n", t->active_opers, __LINE__);
1073 this_stage = stage_name(t->active_opers->rw);
1074 gettimeofday(&stage_time, NULL);
1075 t->stage_mb_trans = 0;
1078 /* first we send everything through aio */
1079 // printf("cnt:%d max_iterations:%d oper:%p\n",cnt, iterations,oper);
1081 while (t->active_opers && (cnt < iterations || iterations == RUN_FOREVER)) {
1082 // printf("active_opers %p line:%d cnt:%d ", t->active_opers,__LINE__,cnt);
1083 if (stonewall && threads_ending) {
1084 oper = t->active_opers;
1085 oper->stonewalled = 1;
1086 oper_list_del(oper, &t->active_opers);
1087 oper_list_add(oper, &t->finished_opers);
1088 // printf(" if branch\n");
1090 run_active_list(t, io_iter, max_io_submit);
1091 // printf(" else branch\n");
1099 /* then we wait for all the operations to finish */
1100 oper = t->finished_opers;
1101 // printf("line:%d oper:%p\n", __LINE__, oper);
1103 io_oper_wait(t, oper);
1107 } while (oper != t->finished_opers);
1108 // printf("finished_opers %p line:%d\n", t->finished_opers,__LINE__);
1110 /* then we do an fsync to get the timing for any future operations
1111 * right, and check to see if any of these need to get restarted
1113 oper = t->finished_opers;
1114 // printf("oper %p line:%d\n", oper,__LINE__);
1118 t->stage_mb_trans += oper_mb_trans(oper);
1119 if (restart_oper(oper)) {
1120 oper_list_del(oper, &t->finished_opers);
1121 oper_list_add(oper, &t->active_opers);
1122 oper = t->finished_opers;
1126 if (oper == t->finished_opers)
1130 if (t->stage_mb_trans && t->num_files > 0) {
1131 // printf("num_files %d line:%d\n", t->num_files,__LINE__);
1132 double seconds = time_since(&stage_time);
1133 fprintf(stderr, "thread %d %s totals (%.2f MB/s) %.2f MB in %.2fs\n",
1134 t - global_thread_info, this_stage, t->stage_mb_trans/seconds,
1135 t->stage_mb_trans, seconds);
1138 if (num_threads > 1) {
1139 // printf("num_threads %d line:%d\n", num_threads,__LINE__);
1140 pthread_mutex_lock(&stage_mutex);
1142 if (threads_ending == num_threads) {
1143 threads_starting = 0;
1144 pthread_cond_broadcast(&stage_cond);
1145 global_thread_throughput(t, this_stage);
1147 // printf("threads_ending %d line:%d\n", threads_ending,__LINE__);
1148 while (threads_ending != num_threads)
1149 pthread_cond_wait(&stage_cond, &stage_mutex);
1150 pthread_mutex_unlock(&stage_mutex);
1153 /* someone got restarted, go back to the beginning */
1154 if (t->active_opers && (cnt < iterations || iterations == RUN_FOREVER)) {
1159 /* finally, free all the ram */
1160 // printf("finished_opers %p line:%d\n", t->finished_opers,__LINE__);
1161 while (t->finished_opers) {
1162 oper = t->finished_opers;
1163 oper_list_del(oper, &t->finished_opers);
1164 status = finish_oper(t, oper);
1167 if (t->num_global_pending) {
1168 fprintf(stderr, "global num pending is %d\n", t->num_global_pending);
1170 io_queue_release(t->io_ctx);
1175 typedef void * (*start_routine)(void *);
1176 int run_workers(struct thread_info *t, int num_threads)
1182 // printf("%s num_threads %d line:%d\n", __FUNCTION__,num_threads,__LINE__);
1183 for(i = 0 ; i < num_threads ; i++) {
1184 ret = pthread_create(&t[i].tid, NULL, (start_routine)worker, t + i);
1186 perror("pthread_create");
1190 for(i = 0 ; i < num_threads ; i++) {
1191 ret = pthread_join(t[i].tid, (void *)&thread_ret);
1193 perror("pthread_join");
1200 off_t parse_size(char *size_arg, off_t mult) {
1204 c = size_arg[strlen(size_arg) - 1];
1206 size_arg[strlen(size_arg) - 1] = '\0';
1208 num = atoi(size_arg);
1212 mult = 1024 * 1024 * 1024;
1227 void print_usage(void) {
1228 printf("usage: aio-stress [-s size] [-r size] [-a size] [-d num] [-b num]\n");
1229 printf(" [-i num] [-t num] [-c num] [-C size] [-nxhOS ]\n");
1230 printf(" file1 [file2 ...]\n");
1231 printf("\t-a size in KB at which to align buffers\n");
1232 printf("\t-b max number of iocbs to give io_submit at once\n");
1233 printf("\t-c number of io contexts per file\n");
1234 printf("\t-C offset between contexts, default 2MB\n");
1235 printf("\t-s size in MB of the test file(s), default 1024MB\n");
1236 printf("\t-r record size in KB used for each io, default 64KB\n");
1237 printf("\t-d number of pending aio requests for each file, default 64\n");
1238 printf("\t-i number of ios per file sent before switching\n\t to the next file, default 8\n");
1239 printf("\t-I total number of ayncs IOs the program will run, default is run until Cntl-C\n");
1240 printf("\t-O Use O_DIRECT (not available in 2.4 kernels),\n");
1241 printf("\t-S Use O_SYNC for writes\n");
1242 printf("\t-o add an operation to the list: write=0, read=1,\n");
1243 printf("\t random write=2, random read=3.\n");
1244 printf("\t repeat -o to specify multiple ops: -o 0 -o 1 etc.\n");
1245 printf("\t-m shm use ipc shared memory for io buffers instead of malloc\n");
1246 printf("\t-m shmfs mmap a file in /dev/shm for io buffers\n");
1247 printf("\t-n no fsyncs between write stage and read stage\n");
1248 printf("\t-l print io_submit latencies after each stage\n");
1249 printf("\t-t number of threads to run\n");
1250 printf("\t-v verification of bytes written\n");
1251 printf("\t-x turn off thread stonewalling\n");
1252 printf("\t-h this message\n");
1253 printf("\n\t the size options (-a -s and -r) allow modifiers -s 400{k,m,g}\n");
1254 printf("\t translate to 400KB, 400MB and 400GB\n");
1255 printf("version %s\n", PROG_VERSION);
1258 int main(int ac, char **av)
1265 off_t file_size = 1 * 1024 * 1024 * 1024;
1266 int first_stage = WRITE;
1267 struct io_oper *oper;
1271 struct thread_info *t;
1273 page_size_mask = getpagesize() - 1;
1276 c = getopt(ac, av, "a:b:c:C:m:s:r:d:i:I:o:t:lnhOSxv");
1282 page_size_mask = parse_size(optarg, 1024);
1286 num_contexts = atoi(optarg);
1289 context_offset = parse_size(optarg, 1024 * 1024);
1291 max_io_submit = atoi(optarg);
1294 file_size = parse_size(optarg, 1024 * 1024);
1297 depth = atoi(optarg);
1300 rec_len = parse_size(optarg, 1024);
1303 io_iter = atoi(optarg);
1306 iterations = atoi(optarg);
1315 if (!strcmp(optarg, "shm")) {
1316 fprintf(stderr, "using ipc shm\n");
1318 } else if (!strcmp(optarg, "shmfs")) {
1319 fprintf(stderr, "using /dev/shm for buffers\n");
1320 use_shm = USE_SHMFS;
1326 fprintf(stderr, "adding stage %s\n", stage_name(i));
1329 o_direct = O_DIRECT;
1335 num_threads = atoi(optarg);
1351 * make sure we don't try to submit more ios than we have allocated
1354 if (depth < io_iter) {
1356 fprintf(stderr, "dropping io_iter to %d\n", io_iter);
1364 num_files = ac - optind;
1366 if (num_threads > (num_files * num_contexts)) {
1367 num_threads = num_files * num_contexts;
1368 fprintf(stderr, "dropping thread count to the number of contexts %d\n",
1372 t = malloc(num_threads * sizeof(*t));
1377 global_thread_info = t;
1379 /* by default, allow a huge number of iocbs to be sent towards
1383 max_io_submit = num_files * io_iter * num_contexts;
1386 * make sure we don't try to submit more ios than max_io_submit allows
1388 if (max_io_submit < io_iter) {
1389 io_iter = max_io_submit;
1390 fprintf(stderr, "dropping io_iter to %d\n", io_iter);
1394 stages = (1 << WRITE) | (1 << READ) | (1 << RREAD) | (1 << RWRITE);
1396 for (i = 0 ; i < LAST_STAGE; i++) {
1397 if (stages & (1 << i)) {
1399 fprintf(stderr, "starting with %s\n", stage_name(i));
1405 if (file_size < num_contexts * context_offset) {
1406 fprintf(stderr, "file size %Lu too small for %d contexts\n",
1407 file_size, num_contexts);
1411 fprintf(stderr, "file size %LuMB, record size %luKB, depth %d, ios per iteration %d\n", file_size / (1024 * 1024), rec_len / 1024, depth, io_iter);
1412 fprintf(stderr, "max io_submit %d, buffer alignment set to %luKB\n",
1413 max_io_submit, (page_size_mask + 1)/1024);
1414 fprintf(stderr, "threads %d files %d contexts %d context offset %LuMB verification %s\n",
1415 num_threads, num_files, num_contexts,
1416 context_offset / (1024 * 1024), verify ? "on" : "off");
1417 /* open all the files and do any required setup for them */
1418 for (i = optind ; i < ac ; i++) {
1420 for (j = 0 ; j < num_contexts ; j++) {
1421 thread_index = open_fds % num_threads;
1423 // fprintf(stderr, "adding file %s thread %d\n", av[i], thread_index);
1425 rwfd = open(av[i], O_CREAT | O_RDWR | o_direct | o_sync, 0600);
1428 oper = create_oper(rwfd, first_stage, j * context_offset,
1429 file_size - j * context_offset, rec_len,
1430 depth, io_iter, av[i]);
1432 fprintf(stderr, "error in create_oper\n");
1435 oper_list_add(oper, &t[thread_index].active_opers);
1436 t[thread_index].num_files++;
1439 if (setup_shared_mem(num_threads, num_files * num_contexts,
1440 depth, rec_len, max_io_submit))
1444 for (i = 0 ; i < num_threads ; i++) {
1445 if (setup_ious(&t[i], t[i].num_files, depth, rec_len, max_io_submit))
1448 if (num_threads > 1){
1449 printf("Running multi thread version num_threads:%d\n", num_threads);
1450 run_workers(t, num_threads);
1453 printf("Running single thread version \n");
1458 for (i = optind ; i < ac ; i++) {
1459 printf("Cleaning up file %s \n", av[i]);
1464 printf("non zero return %d \n", status);
1467 printf("aio-stress Completed successfully %d \n", status);