fio.c

   1 /*
   2  * fio - the flexible io tester
   3  *
   4  * Copyright (C) 2005 Jens Axboe <axboe@suse.de>
   5  * Copyright (C) 2006 Jens Axboe <axboe@kernel.dk>
   6  *
   7  * The license below covers all files distributed with fio unless otherwise
   8  * noted in the file itself.
   9  *
  10  *  This program is free software; you can redistribute it and/or modify
  11  *  it under the terms of the GNU General Public License version 2 as
  12  *  published by the Free Software Foundation.
  13  *
  14  *  This program is distributed in the hope that it will be useful,
  15  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  *  GNU General Public License for more details.
  18  *
  19  *  You should have received a copy of the GNU General Public License
  20  *  along with this program; if not, write to the Free Software
  21  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  22  *
  23  */
  24 #include <unistd.h>
  25 #include <fcntl.h>
  26 #include <string.h>
  27 #include <limits.h>
  28 #include <signal.h>
  29 #include <time.h>
  30 #include <locale.h>
  31 #include <assert.h>
  32 #include <time.h>
  33 #include <sys/stat.h>
  34 #include <sys/wait.h>
  35 #include <sys/ipc.h>
  36 #include <sys/shm.h>
  37 #include <sys/mman.h>
  38
  39 #include "fio.h"
  40 #include "hash.h"
  41 #include "smalloc.h"
  42 #include "verify.h"
  43 #include "trim.h"
  44 #include "diskutil.h"
  45 #include "cgroup.h"
  46 #include "profile.h"
  47 #include "lib/rand.h"
  48 #include "memalign.h"
  49 #include "server.h"
  50
  51 unsigned long page_mask;
  52 unsigned long page_size;
  53
  54 #define PAGE_ALIGN(buf) \
  55         (char *) (((unsigned long) (buf) + page_mask) & ~page_mask)
  56
  57 int groupid = 0;
  58 unsigned int thread_number = 0;
  59 unsigned int nr_process = 0;
  60 unsigned int nr_thread = 0;
  61 int shm_id = 0;
  62 int temp_stall_ts;
  63 unsigned long done_secs = 0;
  64
  65 /*
  66  * Just expose an empty list, if the OS does not support disk util stats
  67  */
  68 #ifndef FIO_HAVE_DISK_UTIL
  69 FLIST_HEAD(disk_list);
  70 #endif
  71
  72 static struct fio_mutex *startup_mutex;
  73 static struct fio_mutex *writeout_mutex;
  74 static volatile int fio_abort;
  75 static int exit_value;
  76 static pthread_t disk_util_thread;
  77 static struct flist_head *cgroup_list;
  78 static char *cgroup_mnt;
  79
  80 unsigned long arch_flags = 0;
  81
  82 struct io_log *agg_io_log[2];
  83
  84 #define JOB_START_TIMEOUT       (5 * 1000)
  85
  86 static const char *fio_os_strings[os_nr] = {
  87         "Invalid",
  88         "Linux",
  89         "AIX",
  90         "FreeBSD",
  91         "HP-UX",
  92         "OSX",
  93         "NetBSD",
  94         "Solaris",
  95         "Windows"
  96 };
  97
  98 static const char *fio_arch_strings[arch_nr] = {
  99         "Invalid",
 100         "x86-64",
 101         "x86",
 102         "ppc",
 103         "ia64",
 104         "s390",
 105         "alpha",
 106         "sparc",
 107         "sparc64",
 108         "arm",
 109         "sh",
 110         "hppa",
 111         "generic"
 112 };
 113
 114 const char *fio_get_os_string(int nr)
 115 {
 116         if (nr < os_nr)
 117                 return fio_os_strings[nr];
 118
 119         return NULL;
 120 }
 121
 122 const char *fio_get_arch_string(int nr)
 123 {
 124         if (nr < arch_nr)
 125                 return fio_arch_strings[nr];
 126
 127         return NULL;
 128 }
 129
 130 void td_set_runstate(struct thread_data *td, int runstate)
 131 {
 132         if (td->runstate == runstate)
 133                 return;
 134
 135         dprint(FD_PROCESS, "pid=%d: runstate %d -> %d\n", (int) td->pid,
 136                                                 td->runstate, runstate);
 137         td->runstate = runstate;
 138 }
 139
 140 void fio_terminate_threads(int group_id)
 141 {
 142         struct thread_data *td;
 143         int i;
 144
 145         dprint(FD_PROCESS, "terminate group_id=%d\n", group_id);
 146
 147         for_each_td(td, i) {
 148                 if (group_id == TERMINATE_ALL || groupid == td->groupid) {
 149                         dprint(FD_PROCESS, "setting terminate on %s/%d\n",
 150                                                 td->o.name, (int) td->pid);
 151                         td->terminate = 1;
 152                         td->o.start_delay = 0;
 153
 154                         /*
 155                          * if the thread is running, just let it exit
 156                          */
 157                         if (!td->pid)
 158                                 continue;
 159                         else if (td->runstate < TD_RAMP)
 160                                 kill(td->pid, SIGTERM);
 161                         else {
 162                                 struct ioengine_ops *ops = td->io_ops;
 163
 164                                 if (ops && (ops->flags & FIO_SIGTERM))
 165                                         kill(td->pid, SIGTERM);
 166                         }
 167                 }
 168         }
 169 }
 170
 171 static void sig_int(int sig)
 172 {
 173         if (threads) {
 174                 if (is_backend)
 175                         fio_server_got_signal(sig);
 176                 else {
 177                         log_info("\nfio: terminating on signal %d\n", sig);
 178                         fflush(stdout);
 179                         exit_value = 128;
 180                 }
 181
 182                 fio_terminate_threads(TERMINATE_ALL);
 183         }
 184 }
 185
 186 static void *disk_thread_main(void *data)
 187 {
 188         fio_mutex_up(startup_mutex);
 189
 190         while (threads) {
 191                 usleep(DISK_UTIL_MSEC * 1000);
 192                 if (!threads)
 193                         break;
 194                 update_io_ticks();
 195
 196                 if (!is_backend)
 197                         print_thread_status();
 198         }
 199
 200         return NULL;
 201 }
 202
 203 static int create_disk_util_thread(void)
 204 {
 205         int ret;
 206
 207         ret = pthread_create(&disk_util_thread, NULL, disk_thread_main, NULL);
 208         if (ret) {
 209                 log_err("Can't create disk util thread: %s\n", strerror(ret));
 210                 return 1;
 211         }
 212
 213         ret = pthread_detach(disk_util_thread);
 214         if (ret) {
 215                 log_err("Can't detatch disk util thread: %s\n", strerror(ret));
 216                 return 1;
 217         }
 218
 219         dprint(FD_MUTEX, "wait on startup_mutex\n");
 220         fio_mutex_down(startup_mutex);
 221         dprint(FD_MUTEX, "done waiting on startup_mutex\n");
 222         return 0;
 223 }
 224
 225 static void set_sig_handlers(void)
 226 {
 227         struct sigaction act;
 228
 229         memset(&act, 0, sizeof(act));
 230         act.sa_handler = sig_int;
 231         act.sa_flags = SA_RESTART;
 232         sigaction(SIGINT, &act, NULL);
 233
 234         memset(&act, 0, sizeof(act));
 235         act.sa_handler = sig_int;
 236         act.sa_flags = SA_RESTART;
 237         sigaction(SIGTERM, &act, NULL);
 238
 239         if (is_backend) {
 240                 memset(&act, 0, sizeof(act));
 241                 act.sa_handler = sig_int;
 242                 act.sa_flags = SA_RESTART;
 243                 sigaction(SIGPIPE, &act, NULL);
 244         }
 245 }
 246
 247 /*
 248  * Check if we are above the minimum rate given.
 249  */
 250 static int __check_min_rate(struct thread_data *td, struct timeval *now,
 251                             enum fio_ddir ddir)
 252 {
 253         unsigned long long bytes = 0;
 254         unsigned long iops = 0;
 255         unsigned long spent;
 256         unsigned long rate;
 257         unsigned int ratemin = 0;
 258         unsigned int rate_iops = 0;
 259         unsigned int rate_iops_min = 0;
 260
 261         assert(ddir_rw(ddir));
 262
 263         if (!td->o.ratemin[ddir] && !td->o.rate_iops_min[ddir])
 264                 return 0;
 265
 266         /*
 267          * allow a 2 second settle period in the beginning
 268          */
 269         if (mtime_since(&td->start, now) < 2000)
 270                 return 0;
 271
 272         iops += td->this_io_blocks[ddir];
 273         bytes += td->this_io_bytes[ddir];
 274         ratemin += td->o.ratemin[ddir];
 275         rate_iops += td->o.rate_iops[ddir];
 276         rate_iops_min += td->o.rate_iops_min[ddir];
 277
 278         /*
 279          * if rate blocks is set, sample is running
 280          */
 281         if (td->rate_bytes[ddir] || td->rate_blocks[ddir]) {
 282                 spent = mtime_since(&td->lastrate[ddir], now);
 283                 if (spent < td->o.ratecycle)
 284                         return 0;
 285
 286                 if (td->o.rate[ddir]) {
 287                         /*
 288                          * check bandwidth specified rate
 289                          */
 290                         if (bytes < td->rate_bytes[ddir]) {
 291                                 log_err("%s: min rate %u not met\n", td->o.name,
 292                                                                 ratemin);
 293                                 return 1;
 294                         } else {
 295                                 rate = ((bytes - td->rate_bytes[ddir]) * 1000) / spent;
 296                                 if (rate < ratemin ||
 297                                     bytes < td->rate_bytes[ddir]) {
 298                                         log_err("%s: min rate %u not met, got"
 299                                                 " %luKB/sec\n", td->o.name,
 300                                                         ratemin, rate);
 301                                         return 1;
 302                                 }
 303                         }
 304                 } else {
 305                         /*
 306                          * checks iops specified rate
 307                          */
 308                         if (iops < rate_iops) {
 309                                 log_err("%s: min iops rate %u not met\n",
 310                                                 td->o.name, rate_iops);
 311                                 return 1;
 312                         } else {
 313                                 rate = ((iops - td->rate_blocks[ddir]) * 1000) / spent;
 314                                 if (rate < rate_iops_min ||
 315                                     iops < td->rate_blocks[ddir]) {
 316                                         log_err("%s: min iops rate %u not met,"
 317                                                 " got %lu\n", td->o.name,
 318                                                         rate_iops_min, rate);
 319                                 }
 320                         }
 321                 }
 322         }
 323
 324         td->rate_bytes[ddir] = bytes;
 325         td->rate_blocks[ddir] = iops;
 326         memcpy(&td->lastrate[ddir], now, sizeof(*now));
 327         return 0;
 328 }
 329
 330 static int check_min_rate(struct thread_data *td, struct timeval *now,
 331                           unsigned long *bytes_done)
 332 {
 333         int ret = 0;
 334
 335         if (bytes_done[0])
 336                 ret |= __check_min_rate(td, now, 0);
 337         if (bytes_done[1])
 338                 ret |= __check_min_rate(td, now, 1);
 339
 340         return ret;
 341 }
 342
 343 static inline int runtime_exceeded(struct thread_data *td, struct timeval *t)
 344 {
 345         if (in_ramp_time(td))
 346                 return 0;
 347         if (!td->o.timeout)
 348                 return 0;
 349         if (mtime_since(&td->epoch, t) >= td->o.timeout * 1000)
 350                 return 1;
 351
 352         return 0;
 353 }
 354
 355 /*
 356  * When job exits, we can cancel the in-flight IO if we are using async
 357  * io. Attempt to do so.
 358  */
 359 static void cleanup_pending_aio(struct thread_data *td)
 360 {
 361         struct flist_head *entry, *n;
 362         struct io_u *io_u;
 363         int r;
 364
 365         /*
 366          * get immediately available events, if any
 367          */
 368         r = io_u_queued_complete(td, 0, NULL);
 369         if (r < 0)
 370                 return;
 371
 372         /*
 373          * now cancel remaining active events
 374          */
 375         if (td->io_ops->cancel) {
 376                 flist_for_each_safe(entry, n, &td->io_u_busylist) {
 377                         io_u = flist_entry(entry, struct io_u, list);
 378
 379                         /*
 380                          * if the io_u isn't in flight, then that generally
 381                          * means someone leaked an io_u. complain but fix
 382                          * it up, so we don't stall here.
 383                          */
 384                         if ((io_u->flags & IO_U_F_FLIGHT) == 0) {
 385                                 log_err("fio: non-busy IO on busy list\n");
 386                                 put_io_u(td, io_u);
 387                         } else {
 388                                 r = td->io_ops->cancel(td, io_u);
 389                                 if (!r)
 390                                         put_io_u(td, io_u);
 391                         }
 392                 }
 393         }
 394
 395         if (td->cur_depth)
 396                 r = io_u_queued_complete(td, td->cur_depth, NULL);
 397 }
 398
 399 /*
 400  * Helper to handle the final sync of a file. Works just like the normal
 401  * io path, just does everything sync.
 402  */
 403 static int fio_io_sync(struct thread_data *td, struct fio_file *f)
 404 {
 405         struct io_u *io_u = __get_io_u(td);
 406         int ret;
 407
 408         if (!io_u)
 409                 return 1;
 410
 411         io_u->ddir = DDIR_SYNC;
 412         io_u->file = f;
 413
 414         if (td_io_prep(td, io_u)) {
 415                 put_io_u(td, io_u);
 416                 return 1;
 417         }
 418
 419 requeue:
 420         ret = td_io_queue(td, io_u);
 421         if (ret < 0) {
 422                 td_verror(td, io_u->error, "td_io_queue");
 423                 put_io_u(td, io_u);
 424                 return 1;
 425         } else if (ret == FIO_Q_QUEUED) {
 426                 if (io_u_queued_complete(td, 1, NULL) < 0)
 427                         return 1;
 428         } else if (ret == FIO_Q_COMPLETED) {
 429                 if (io_u->error) {
 430                         td_verror(td, io_u->error, "td_io_queue");
 431                         return 1;
 432                 }
 433
 434                 if (io_u_sync_complete(td, io_u, NULL) < 0)
 435                         return 1;
 436         } else if (ret == FIO_Q_BUSY) {
 437                 if (td_io_commit(td))
 438                         return 1;
 439                 goto requeue;
 440         }
 441
 442         return 0;
 443 }
 444
 445 static inline void __update_tv_cache(struct thread_data *td)
 446 {
 447         fio_gettime(&td->tv_cache, NULL);
 448 }
 449
 450 static inline void update_tv_cache(struct thread_data *td)
 451 {
 452         if ((++td->tv_cache_nr & td->tv_cache_mask) == td->tv_cache_mask)
 453                 __update_tv_cache(td);
 454 }
 455
 456 static int break_on_this_error(struct thread_data *td, enum fio_ddir ddir,
 457                                int *retptr)
 458 {
 459         int ret = *retptr;
 460
 461         if (ret < 0 || td->error) {
 462                 int err;
 463
 464                 if (ret < 0)
 465                         err = -ret;
 466                 else
 467                         err = td->error;
 468
 469                 if (!(td->o.continue_on_error & td_error_type(ddir, err)))
 470                         return 1;
 471
 472                 if (td_non_fatal_error(err)) {
 473                         /*
 474                          * Continue with the I/Os in case of
 475                          * a non fatal error.
 476                          */
 477                         update_error_count(td, err);
 478                         td_clear_error(td);
 479                         *retptr = 0;
 480                         return 0;
 481                 } else if (td->o.fill_device && err == ENOSPC) {
 482                         /*
 483                          * We expect to hit this error if
 484                          * fill_device option is set.
 485                          */
 486                         td_clear_error(td);
 487                         td->terminate = 1;
 488                         return 1;
 489                 } else {
 490                         /*
 491                          * Stop the I/O in case of a fatal
 492                          * error.
 493                          */
 494                         update_error_count(td, err);
 495                         return 1;
 496                 }
 497         }
 498
 499         return 0;
 500 }
 501
 502 /*
 503  * The main verify engine. Runs over the writes we previously submitted,
 504  * reads the blocks back in, and checks the crc/md5 of the data.
 505  */
 506 static void do_verify(struct thread_data *td)
 507 {
 508         struct fio_file *f;
 509         struct io_u *io_u;
 510         int ret, min_events;
 511         unsigned int i;
 512
 513         dprint(FD_VERIFY, "starting loop\n");
 514
 515         /*
 516          * sync io first and invalidate cache, to make sure we really
 517          * read from disk.
 518          */
 519         for_each_file(td, f, i) {
 520                 if (!fio_file_open(f))
 521                         continue;
 522                 if (fio_io_sync(td, f))
 523                         break;
 524                 if (file_invalidate_cache(td, f))
 525                         break;
 526         }
 527
 528         if (td->error)
 529                 return;
 530
 531         td_set_runstate(td, TD_VERIFYING);
 532
 533         io_u = NULL;
 534         while (!td->terminate) {
 535                 int ret2, full;
 536
 537                 update_tv_cache(td);
 538
 539                 if (runtime_exceeded(td, &td->tv_cache)) {
 540                         __update_tv_cache(td);
 541                         if (runtime_exceeded(td, &td->tv_cache)) {
 542                                 td->terminate = 1;
 543                                 break;
 544                         }
 545                 }
 546
 547                 io_u = __get_io_u(td);
 548                 if (!io_u)
 549                         break;
 550
 551                 if (get_next_verify(td, io_u)) {
 552                         put_io_u(td, io_u);
 553                         break;
 554                 }
 555
 556                 if (td_io_prep(td, io_u)) {
 557                         put_io_u(td, io_u);
 558                         break;
 559                 }
 560
 561                 if (td->o.verify_async)
 562                         io_u->end_io = verify_io_u_async;
 563                 else
 564                         io_u->end_io = verify_io_u;
 565
 566                 ret = td_io_queue(td, io_u);
 567                 switch (ret) {
 568                 case FIO_Q_COMPLETED:
 569                         if (io_u->error) {
 570                                 ret = -io_u->error;
 571                                 clear_io_u(td, io_u);
 572                         } else if (io_u->resid) {
 573                                 int bytes = io_u->xfer_buflen - io_u->resid;
 574
 575                                 /*
 576                                  * zero read, fail
 577                                  */
 578                                 if (!bytes) {
 579                                         td_verror(td, EIO, "full resid");
 580                                         put_io_u(td, io_u);
 581                                         break;
 582                                 }
 583
 584                                 io_u->xfer_buflen = io_u->resid;
 585                                 io_u->xfer_buf += bytes;
 586                                 io_u->offset += bytes;
 587
 588                                 if (ddir_rw(io_u->ddir))
 589                                         td->ts.short_io_u[io_u->ddir]++;
 590
 591                                 f = io_u->file;
 592                                 if (io_u->offset == f->real_file_size)
 593                                         goto sync_done;
 594
 595                                 requeue_io_u(td, &io_u);
 596                         } else {
 597 sync_done:
 598                                 ret = io_u_sync_complete(td, io_u, NULL);
 599                                 if (ret < 0)
 600                                         break;
 601                         }
 602                         continue;
 603                 case FIO_Q_QUEUED:
 604                         break;
 605                 case FIO_Q_BUSY:
 606                         requeue_io_u(td, &io_u);
 607                         ret2 = td_io_commit(td);
 608                         if (ret2 < 0)
 609                                 ret = ret2;
 610                         break;
 611                 default:
 612                         assert(ret < 0);
 613                         td_verror(td, -ret, "td_io_queue");
 614                         break;
 615                 }
 616
 617                 if (break_on_this_error(td, io_u->ddir, &ret))
 618                         break;
 619
 620                 /*
 621                  * if we can queue more, do so. but check if there are
 622                  * completed io_u's first. Note that we can get BUSY even
 623                  * without IO queued, if the system is resource starved.
 624                  */
 625                 full = queue_full(td) || (ret == FIO_Q_BUSY && td->cur_depth);
 626                 if (full || !td->o.iodepth_batch_complete) {
 627                         min_events = min(td->o.iodepth_batch_complete,
 628                                          td->cur_depth);
 629                         if (full && !min_events && td->o.iodepth_batch_complete != 0)
 630                                 min_events = 1;
 631
 632                         do {
 633                                 /*
 634                                  * Reap required number of io units, if any,
 635                                  * and do the verification on them through
 636                                  * the callback handler
 637                                  */
 638                                 if (io_u_queued_complete(td, min_events, NULL) < 0) {
 639                                         ret = -1;
 640                                         break;
 641                                 }
 642                         } while (full && (td->cur_depth > td->o.iodepth_low));
 643                 }
 644                 if (ret < 0)
 645                         break;
 646         }
 647
 648         if (!td->error) {
 649                 min_events = td->cur_depth;
 650
 651                 if (min_events)
 652                         ret = io_u_queued_complete(td, min_events, NULL);
 653         } else
 654                 cleanup_pending_aio(td);
 655
 656         td_set_runstate(td, TD_RUNNING);
 657
 658         dprint(FD_VERIFY, "exiting loop\n");
 659 }
 660
 661 /*
 662  * Main IO worker function. It retrieves io_u's to process and queues
 663  * and reaps them, checking for rate and errors along the way.
 664  */
 665 static void do_io(struct thread_data *td)
 666 {
 667         unsigned int i;
 668         int ret = 0;
 669
 670         if (in_ramp_time(td))
 671                 td_set_runstate(td, TD_RAMP);
 672         else
 673                 td_set_runstate(td, TD_RUNNING);
 674
 675         while ( (td->o.read_iolog_file && !flist_empty(&td->io_log_list)) ||
 676                 (!flist_empty(&td->trim_list)) ||
 677                 ((td->this_io_bytes[0] + td->this_io_bytes[1]) < td->o.size) ) {
 678                 struct timeval comp_time;
 679                 unsigned long bytes_done[2] = { 0, 0 };
 680                 int min_evts = 0;
 681                 struct io_u *io_u;
 682                 int ret2, full;
 683                 enum fio_ddir ddir;
 684
 685                 if (td->terminate)
 686                         break;
 687
 688                 update_tv_cache(td);
 689
 690                 if (runtime_exceeded(td, &td->tv_cache)) {
 691                         __update_tv_cache(td);
 692                         if (runtime_exceeded(td, &td->tv_cache)) {
 693                                 td->terminate = 1;
 694                                 break;
 695                         }
 696                 }
 697
 698                 io_u = get_io_u(td);
 699                 if (!io_u)
 700                         break;
 701
 702                 ddir = io_u->ddir;
 703
 704                 /*
 705                  * Add verification end_io handler, if asked to verify
 706                  * a previously written file.
 707                  */
 708                 if (td->o.verify != VERIFY_NONE && io_u->ddir == DDIR_READ &&
 709                     !td_rw(td)) {
 710                         if (td->o.verify_async)
 711                                 io_u->end_io = verify_io_u_async;
 712                         else
 713                                 io_u->end_io = verify_io_u;
 714                         td_set_runstate(td, TD_VERIFYING);
 715                 } else if (in_ramp_time(td))
 716                         td_set_runstate(td, TD_RAMP);
 717                 else
 718                         td_set_runstate(td, TD_RUNNING);
 719
 720                 ret = td_io_queue(td, io_u);
 721                 switch (ret) {
 722                 case FIO_Q_COMPLETED:
 723                         if (io_u->error) {
 724                                 ret = -io_u->error;
 725                                 clear_io_u(td, io_u);
 726                         } else if (io_u->resid) {
 727                                 int bytes = io_u->xfer_buflen - io_u->resid;
 728                                 struct fio_file *f = io_u->file;
 729
 730                                 /*
 731                                  * zero read, fail
 732                                  */
 733                                 if (!bytes) {
 734                                         td_verror(td, EIO, "full resid");
 735                                         put_io_u(td, io_u);
 736                                         break;
 737                                 }
 738
 739                                 io_u->xfer_buflen = io_u->resid;
 740                                 io_u->xfer_buf += bytes;
 741                                 io_u->offset += bytes;
 742
 743                                 if (ddir_rw(io_u->ddir))
 744                                         td->ts.short_io_u[io_u->ddir]++;
 745
 746                                 if (io_u->offset == f->real_file_size)
 747                                         goto sync_done;
 748
 749                                 requeue_io_u(td, &io_u);
 750                         } else {
 751 sync_done:
 752                                 if (__should_check_rate(td, 0) ||
 753                                     __should_check_rate(td, 1))
 754                                         fio_gettime(&comp_time, NULL);
 755
 756                                 ret = io_u_sync_complete(td, io_u, bytes_done);
 757                                 if (ret < 0)
 758                                         break;
 759                         }
 760                         break;
 761                 case FIO_Q_QUEUED:
 762                         /*
 763                          * if the engine doesn't have a commit hook,
 764                          * the io_u is really queued. if it does have such
 765                          * a hook, it has to call io_u_queued() itself.
 766                          */
 767                         if (td->io_ops->commit == NULL)
 768                                 io_u_queued(td, io_u);
 769                         break;
 770                 case FIO_Q_BUSY:
 771                         requeue_io_u(td, &io_u);
 772                         ret2 = td_io_commit(td);
 773                         if (ret2 < 0)
 774                                 ret = ret2;
 775                         break;
 776                 default:
 777                         assert(ret < 0);
 778                         put_io_u(td, io_u);
 779                         break;
 780                 }
 781
 782                 if (break_on_this_error(td, ddir, &ret))
 783                         break;
 784
 785                 /*
 786                  * See if we need to complete some commands. Note that we
 787                  * can get BUSY even without IO queued, if the system is
 788                  * resource starved.
 789                  */
 790                 full = queue_full(td) || (ret == FIO_Q_BUSY && td->cur_depth);
 791                 if (full || !td->o.iodepth_batch_complete) {
 792                         min_evts = min(td->o.iodepth_batch_complete,
 793                                         td->cur_depth);
 794                         if (full && !min_evts && td->o.iodepth_batch_complete != 0)
 795                                 min_evts = 1;
 796
 797                         if (__should_check_rate(td, 0) ||
 798                             __should_check_rate(td, 1))
 799                                 fio_gettime(&comp_time, NULL);
 800
 801                         do {
 802                                 ret = io_u_queued_complete(td, min_evts, bytes_done);
 803                                 if (ret < 0)
 804                                         break;
 805
 806                         } while (full && (td->cur_depth > td->o.iodepth_low));
 807                 }
 808
 809                 if (ret < 0)
 810                         break;
 811                 if (!(bytes_done[0] + bytes_done[1]))
 812                         continue;
 813
 814                 if (!in_ramp_time(td) && should_check_rate(td, bytes_done)) {
 815                         if (check_min_rate(td, &comp_time, bytes_done)) {
 816                                 if (exitall_on_terminate)
 817                                         fio_terminate_threads(td->groupid);
 818                                 td_verror(td, EIO, "check_min_rate");
 819                                 break;
 820                         }
 821                 }
 822
 823                 if (td->o.thinktime) {
 824                         unsigned long long b;
 825
 826                         b = td->io_blocks[0] + td->io_blocks[1];
 827                         if (!(b % td->o.thinktime_blocks)) {
 828                                 int left;
 829
 830                                 if (td->o.thinktime_spin)
 831                                         usec_spin(td->o.thinktime_spin);
 832
 833                                 left = td->o.thinktime - td->o.thinktime_spin;
 834                                 if (left)
 835                                         usec_sleep(td, left);
 836                         }
 837                 }
 838         }
 839
 840         if (td->trim_entries)
 841                 log_err("fio: %d trim entries leaked?\n", td->trim_entries);
 842
 843         if (td->o.fill_device && td->error == ENOSPC) {
 844                 td->error = 0;
 845                 td->terminate = 1;
 846         }
 847         if (!td->error) {
 848                 struct fio_file *f;
 849
 850                 i = td->cur_depth;
 851                 if (i) {
 852                         ret = io_u_queued_complete(td, i, NULL);
 853                         if (td->o.fill_device && td->error == ENOSPC)
 854                                 td->error = 0;
 855                 }
 856
 857                 if (should_fsync(td) && td->o.end_fsync) {
 858                         td_set_runstate(td, TD_FSYNCING);
 859
 860                         for_each_file(td, f, i) {
 861                                 if (!fio_file_open(f))
 862                                         continue;
 863                                 fio_io_sync(td, f);
 864                         }
 865                 }
 866         } else
 867                 cleanup_pending_aio(td);
 868
 869         /*
 870          * stop job if we failed doing any IO
 871          */
 872         if ((td->this_io_bytes[0] + td->this_io_bytes[1]) == 0)
 873                 td->done = 1;
 874 }
 875
 876 static void cleanup_io_u(struct thread_data *td)
 877 {
 878         struct flist_head *entry, *n;
 879         struct io_u *io_u;
 880
 881         flist_for_each_safe(entry, n, &td->io_u_freelist) {
 882                 io_u = flist_entry(entry, struct io_u, list);
 883
 884                 flist_del(&io_u->list);
 885                 fio_memfree(io_u, sizeof(*io_u));
 886         }
 887
 888         free_io_mem(td);
 889 }
 890
 891 static int init_io_u(struct thread_data *td)
 892 {
 893         struct io_u *io_u;
 894         unsigned int max_bs;
 895         int cl_align, i, max_units;
 896         char *p;
 897
 898         max_units = td->o.iodepth;
 899         max_bs = max(td->o.max_bs[DDIR_READ], td->o.max_bs[DDIR_WRITE]);
 900         td->orig_buffer_size = (unsigned long long) max_bs
 901                                         * (unsigned long long) max_units;
 902
 903         if (td->o.mem_type == MEM_SHMHUGE || td->o.mem_type == MEM_MMAPHUGE) {
 904                 unsigned long bs;
 905
 906                 bs = td->orig_buffer_size + td->o.hugepage_size - 1;
 907                 td->orig_buffer_size = bs & ~(td->o.hugepage_size - 1);
 908         }
 909
 910         if (td->orig_buffer_size != (size_t) td->orig_buffer_size) {
 911                 log_err("fio: IO memory too large. Reduce max_bs or iodepth\n");
 912                 return 1;
 913         }
 914
 915         if (allocate_io_mem(td))
 916                 return 1;
 917
 918         if (td->o.odirect || td->o.mem_align ||
 919             (td->io_ops->flags & FIO_RAWIO))
 920                 p = PAGE_ALIGN(td->orig_buffer) + td->o.mem_align;
 921         else
 922                 p = td->orig_buffer;
 923
 924         cl_align = os_cache_line_size();
 925
 926         for (i = 0; i < max_units; i++) {
 927                 void *ptr;
 928
 929                 if (td->terminate)
 930                         return 1;
 931
 932                 ptr = fio_memalign(cl_align, sizeof(*io_u));
 933                 if (!ptr) {
 934                         log_err("fio: unable to allocate aligned memory\n");
 935                         break;
 936                 }
 937
 938                 io_u = ptr;
 939                 memset(io_u, 0, sizeof(*io_u));
 940                 INIT_FLIST_HEAD(&io_u->list);
 941                 dprint(FD_MEM, "io_u alloc %p, index %u\n", io_u, i);
 942
 943                 if (!(td->io_ops->flags & FIO_NOIO)) {
 944                         io_u->buf = p;
 945                         dprint(FD_MEM, "io_u %p, mem %p\n", io_u, io_u->buf);
 946
 947                         if (td_write(td))
 948                                 io_u_fill_buffer(td, io_u, max_bs);
 949                         if (td_write(td) && td->o.verify_pattern_bytes) {
 950                                 /*
 951                                  * Fill the buffer with the pattern if we are
 952                                  * going to be doing writes.
 953                                  */
 954                                 fill_pattern(td, io_u->buf, max_bs, io_u, 0, 0);
 955                         }
 956                 }
 957
 958                 io_u->index = i;
 959                 io_u->flags = IO_U_F_FREE;
 960                 flist_add(&io_u->list, &td->io_u_freelist);
 961                 p += max_bs;
 962         }
 963
 964         return 0;
 965 }
 966
 967 static int switch_ioscheduler(struct thread_data *td)
 968 {
 969         char tmp[256], tmp2[128];
 970         FILE *f;
 971         int ret;
 972
 973         if (td->io_ops->flags & FIO_DISKLESSIO)
 974                 return 0;
 975
 976         sprintf(tmp, "%s/queue/scheduler", td->sysfs_root);
 977
 978         f = fopen(tmp, "r+");
 979         if (!f) {
 980                 if (errno == ENOENT) {
 981                         log_err("fio: os or kernel doesn't support IO scheduler"
 982                                 " switching\n");
 983                         return 0;
 984                 }
 985                 td_verror(td, errno, "fopen iosched");
 986                 return 1;
 987         }
 988
 989         /*
 990          * Set io scheduler.
 991          */
 992         ret = fwrite(td->o.ioscheduler, strlen(td->o.ioscheduler), 1, f);
 993         if (ferror(f) || ret != 1) {
 994                 td_verror(td, errno, "fwrite");
 995                 fclose(f);
 996                 return 1;
 997         }
 998
 999         rewind(f);
1000
1001         /*
1002          * Read back and check that the selected scheduler is now the default.
1003          */
1004         ret = fread(tmp, 1, sizeof(tmp), f);
1005         if (ferror(f) || ret < 0) {
1006                 td_verror(td, errno, "fread");
1007                 fclose(f);
1008                 return 1;
1009         }
1010
1011         sprintf(tmp2, "[%s]", td->o.ioscheduler);
1012         if (!strstr(tmp, tmp2)) {
1013                 log_err("fio: io scheduler %s not found\n", td->o.ioscheduler);
1014                 td_verror(td, EINVAL, "iosched_switch");
1015                 fclose(f);
1016                 return 1;
1017         }
1018
1019         fclose(f);
1020         return 0;
1021 }
1022
1023 static int keep_running(struct thread_data *td)
1024 {
1025         unsigned long long io_done;
1026
1027         if (td->done)
1028                 return 0;
1029         if (td->o.time_based)
1030                 return 1;
1031         if (td->o.loops) {
1032                 td->o.loops--;
1033                 return 1;
1034         }
1035
1036         io_done = td->io_bytes[DDIR_READ] + td->io_bytes[DDIR_WRITE]
1037                         + td->io_skip_bytes;
1038         if (io_done < td->o.size)
1039                 return 1;
1040
1041         return 0;
1042 }
1043
1044 static void reset_io_counters(struct thread_data *td)
1045 {
1046         td->stat_io_bytes[0] = td->stat_io_bytes[1] = 0;
1047         td->this_io_bytes[0] = td->this_io_bytes[1] = 0;
1048         td->stat_io_blocks[0] = td->stat_io_blocks[1] = 0;
1049         td->this_io_blocks[0] = td->this_io_blocks[1] = 0;
1050         td->zone_bytes = 0;
1051         td->rate_bytes[0] = td->rate_bytes[1] = 0;
1052         td->rate_blocks[0] = td->rate_blocks[1] = 0;
1053
1054         td->last_was_sync = 0;
1055
1056         /*
1057          * reset file done count if we are to start over
1058          */
1059         if (td->o.time_based || td->o.loops)
1060                 td->nr_done_files = 0;
1061 }
1062
1063 void reset_all_stats(struct thread_data *td)
1064 {
1065         struct timeval tv;
1066         int i;
1067
1068         reset_io_counters(td);
1069
1070         for (i = 0; i < 2; i++) {
1071                 td->io_bytes[i] = 0;
1072                 td->io_blocks[i] = 0;
1073                 td->io_issues[i] = 0;
1074                 td->ts.total_io_u[i] = 0;
1075         }
1076
1077         fio_gettime(&tv, NULL);
1078         td->ts.runtime[0] = 0;
1079         td->ts.runtime[1] = 0;
1080         memcpy(&td->epoch, &tv, sizeof(tv));
1081         memcpy(&td->start, &tv, sizeof(tv));
1082 }
1083
1084 static void clear_io_state(struct thread_data *td)
1085 {
1086         struct fio_file *f;
1087         unsigned int i;
1088
1089         reset_io_counters(td);
1090
1091         close_files(td);
1092         for_each_file(td, f, i)
1093                 fio_file_clear_done(f);
1094
1095         /*
1096          * Set the same seed to get repeatable runs
1097          */
1098         td_fill_rand_seeds(td);
1099 }
1100
1101 static int exec_string(const char *string)
1102 {
1103         int ret, newlen = strlen(string) + 1 + 8;
1104         char *str;
1105
1106         str = malloc(newlen);
1107         sprintf(str, "sh -c %s", string);
1108
1109         ret = system(str);
1110         if (ret == -1)
1111                 log_err("fio: exec of cmd <%s> failed\n", str);
1112
1113         free(str);
1114         return ret;
1115 }
1116
1117 /*
1118  * Entry point for the thread based jobs. The process based jobs end up
1119  * here as well, after a little setup.
1120  */
1121 static void *thread_main(void *data)
1122 {
1123         unsigned long long elapsed;
1124         struct thread_data *td = data;
1125         pthread_condattr_t attr;
1126         int clear_state;
1127
1128         if (!td->o.use_thread) {
1129                 setsid();
1130                 td->pid = getpid();
1131         } else
1132                 td->pid = gettid();
1133
1134         dprint(FD_PROCESS, "jobs pid=%d started\n", (int) td->pid);
1135
1136         INIT_FLIST_HEAD(&td->io_u_freelist);
1137         INIT_FLIST_HEAD(&td->io_u_busylist);
1138         INIT_FLIST_HEAD(&td->io_u_requeues);
1139         INIT_FLIST_HEAD(&td->io_log_list);
1140         INIT_FLIST_HEAD(&td->io_hist_list);
1141         INIT_FLIST_HEAD(&td->verify_list);
1142         INIT_FLIST_HEAD(&td->trim_list);
1143         pthread_mutex_init(&td->io_u_lock, NULL);
1144         td->io_hist_tree = RB_ROOT;
1145
1146         pthread_condattr_init(&attr);
1147         pthread_cond_init(&td->verify_cond, &attr);
1148         pthread_cond_init(&td->free_cond, &attr);
1149
1150         td_set_runstate(td, TD_INITIALIZED);
1151         dprint(FD_MUTEX, "up startup_mutex\n");
1152         fio_mutex_up(startup_mutex);
1153         dprint(FD_MUTEX, "wait on td->mutex\n");
1154         fio_mutex_down(td->mutex);
1155         dprint(FD_MUTEX, "done waiting on td->mutex\n");
1156
1157         /*
1158          * the ->mutex mutex is now no longer used, close it to avoid
1159          * eating a file descriptor
1160          */
1161         fio_mutex_remove(td->mutex);
1162
1163         /*
1164          * A new gid requires privilege, so we need to do this before setting
1165          * the uid.
1166          */
1167         if (td->o.gid != -1U && setgid(td->o.gid)) {
1168                 td_verror(td, errno, "setgid");
1169                 goto err;
1170         }
1171         if (td->o.uid != -1U && setuid(td->o.uid)) {
1172                 td_verror(td, errno, "setuid");
1173                 goto err;
1174         }
1175
1176         /*
1177          * If we have a gettimeofday() thread, make sure we exclude that
1178          * thread from this job
1179          */
1180         if (td->o.gtod_cpu)
1181                 fio_cpu_clear(&td->o.cpumask, td->o.gtod_cpu);
1182
1183         /*
1184          * Set affinity first, in case it has an impact on the memory
1185          * allocations.
1186          */
1187         if (td->o.cpumask_set && fio_setaffinity(td->pid, td->o.cpumask) == -1) {
1188                 td_verror(td, errno, "cpu_set_affinity");
1189                 goto err;
1190         }
1191
1192         /*
1193          * May alter parameters that init_io_u() will use, so we need to
1194          * do this first.
1195          */
1196         if (init_iolog(td))
1197                 goto err;
1198
1199         if (init_io_u(td))
1200                 goto err;
1201
1202         if (td->o.verify_async && verify_async_init(td))
1203                 goto err;
1204
1205         if (td->ioprio_set) {
1206                 if (ioprio_set(IOPRIO_WHO_PROCESS, 0, td->ioprio) == -1) {
1207                         td_verror(td, errno, "ioprio_set");
1208                         goto err;
1209                 }
1210         }
1211
1212         if (td->o.cgroup_weight && cgroup_setup(td, cgroup_list, &cgroup_mnt))
1213                 goto err;
1214
1215         if (nice(td->o.nice) == -1) {
1216                 td_verror(td, errno, "nice");
1217                 goto err;
1218         }
1219
1220         if (td->o.ioscheduler && switch_ioscheduler(td))
1221                 goto err;
1222
1223         if (!td->o.create_serialize && setup_files(td))
1224                 goto err;
1225
1226         if (td_io_init(td))
1227                 goto err;
1228
1229         if (init_random_map(td))
1230                 goto err;
1231
1232         if (td->o.exec_prerun) {
1233                 if (exec_string(td->o.exec_prerun))
1234                         goto err;
1235         }
1236
1237         if (td->o.pre_read) {
1238                 if (pre_read_files(td) < 0)
1239                         goto err;
1240         }
1241
1242         fio_gettime(&td->epoch, NULL);
1243         getrusage(RUSAGE_SELF, &td->ru_start);
1244
1245         clear_state = 0;
1246         while (keep_running(td)) {
1247                 fio_gettime(&td->start, NULL);
1248                 memcpy(&td->bw_sample_time, &td->start, sizeof(td->start));
1249                 memcpy(&td->iops_sample_time, &td->start, sizeof(td->start));
1250                 memcpy(&td->tv_cache, &td->start, sizeof(td->start));
1251
1252                 if (td->o.ratemin[0] || td->o.ratemin[1]) {
1253                         memcpy(&td->lastrate[0], &td->bw_sample_time,
1254                                                 sizeof(td->bw_sample_time));
1255                         memcpy(&td->lastrate[1], &td->bw_sample_time,
1256                                                 sizeof(td->bw_sample_time));
1257                 }
1258
1259                 if (clear_state)
1260                         clear_io_state(td);
1261
1262                 prune_io_piece_log(td);
1263
1264                 do_io(td);
1265
1266                 clear_state = 1;
1267
1268                 if (td_read(td) && td->io_bytes[DDIR_READ]) {
1269                         elapsed = utime_since_now(&td->start);
1270                         td->ts.runtime[DDIR_READ] += elapsed;
1271                 }
1272                 if (td_write(td) && td->io_bytes[DDIR_WRITE]) {
1273                         elapsed = utime_since_now(&td->start);
1274                         td->ts.runtime[DDIR_WRITE] += elapsed;
1275                 }
1276
1277                 if (td->error || td->terminate)
1278                         break;
1279
1280                 if (!td->o.do_verify ||
1281                     td->o.verify == VERIFY_NONE ||
1282                     (td->io_ops->flags & FIO_UNIDIR))
1283                         continue;
1284
1285                 clear_io_state(td);
1286
1287                 fio_gettime(&td->start, NULL);
1288
1289                 do_verify(td);
1290
1291                 td->ts.runtime[DDIR_READ] += utime_since_now(&td->start);
1292
1293                 if (td->error || td->terminate)
1294                         break;
1295         }
1296
1297         update_rusage_stat(td);
1298         td->ts.runtime[0] = (td->ts.runtime[0] + 999) / 1000;
1299         td->ts.runtime[1] = (td->ts.runtime[1] + 999) / 1000;
1300         td->ts.total_run_time = mtime_since_now(&td->epoch);
1301         td->ts.io_bytes[0] = td->io_bytes[0];
1302         td->ts.io_bytes[1] = td->io_bytes[1];
1303
1304         fio_mutex_down(writeout_mutex);
1305         if (td->bw_log) {
1306                 if (td->o.bw_log_file) {
1307                         finish_log_named(td, td->bw_log,
1308                                                 td->o.bw_log_file, "bw");
1309                 } else
1310                         finish_log(td, td->bw_log, "bw");
1311         }
1312         if (td->lat_log) {
1313                 if (td->o.lat_log_file) {
1314                         finish_log_named(td, td->lat_log,
1315                                                 td->o.lat_log_file, "lat");
1316                 } else
1317                         finish_log(td, td->lat_log, "lat");
1318         }
1319         if (td->slat_log) {
1320                 if (td->o.lat_log_file) {
1321                         finish_log_named(td, td->slat_log,
1322                                                 td->o.lat_log_file, "slat");
1323                 } else
1324                         finish_log(td, td->slat_log, "slat");
1325         }
1326         if (td->clat_log) {
1327                 if (td->o.lat_log_file) {
1328                         finish_log_named(td, td->clat_log,
1329                                                 td->o.lat_log_file, "clat");
1330                 } else
1331                         finish_log(td, td->clat_log, "clat");
1332         }
1333         if (td->iops_log) {
1334                 if (td->o.iops_log_file) {
1335                         finish_log_named(td, td->iops_log,
1336                                                 td->o.iops_log_file, "iops");
1337                 } else
1338                         finish_log(td, td->iops_log, "iops");
1339         }
1340
1341         fio_mutex_up(writeout_mutex);
1342         if (td->o.exec_postrun)
1343                 exec_string(td->o.exec_postrun);
1344
1345         if (exitall_on_terminate)
1346                 fio_terminate_threads(td->groupid);
1347
1348 err:
1349         if (td->error)
1350                 log_info("fio: pid=%d, err=%d/%s\n", (int) td->pid, td->error,
1351                                                         td->verror);
1352
1353         if (td->o.verify_async)
1354                 verify_async_exit(td);
1355
1356         close_and_free_files(td);
1357         close_ioengine(td);
1358         cleanup_io_u(td);
1359         cgroup_shutdown(td, &cgroup_mnt);
1360
1361         if (td->o.cpumask_set) {
1362                 int ret = fio_cpuset_exit(&td->o.cpumask);
1363
1364                 td_verror(td, ret, "fio_cpuset_exit");
1365         }
1366
1367         /*
1368          * do this very late, it will log file closing as well
1369          */
1370         if (td->o.write_iolog_file)
1371                 write_iolog_close(td);
1372
1373         td_set_runstate(td, TD_EXITED);
1374         return (void *) (unsigned long) td->error;
1375 }
1376
1377 /*
1378  * We cannot pass the td data into a forked process, so attach the td and
1379  * pass it to the thread worker.
1380  */
1381 static int fork_main(int shmid, int offset)
1382 {
1383         struct thread_data *td;
1384         void *data, *ret;
1385
1386 #ifndef __hpux
1387         data = shmat(shmid, NULL, 0);
1388         if (data == (void *) -1) {
1389                 int __err = errno;
1390
1391                 perror("shmat");
1392                 return __err;
1393         }
1394 #else
1395         /*
1396          * HP-UX inherits shm mappings?
1397          */
1398         data = threads;
1399 #endif
1400
1401         td = data + offset * sizeof(struct thread_data);
1402         ret = thread_main(td);
1403         shmdt(data);
1404         return (int) (unsigned long) ret;
1405 }
1406
1407 /*
1408  * Run over the job map and reap the threads that have exited, if any.
1409  */
1410 static void reap_threads(unsigned int *nr_running, unsigned int *t_rate,
1411                          unsigned int *m_rate)
1412 {
1413         struct thread_data *td;
1414         unsigned int cputhreads, realthreads, pending;
1415         int i, status, ret;
1416
1417         /*
1418          * reap exited threads (TD_EXITED -> TD_REAPED)
1419          */
1420         realthreads = pending = cputhreads = 0;
1421         for_each_td(td, i) {
1422                 int flags = 0;
1423
1424                 /*
1425                  * ->io_ops is NULL for a thread that has closed its
1426                  * io engine
1427                  */
1428                 if (td->io_ops && !strcmp(td->io_ops->name, "cpuio"))
1429                         cputhreads++;
1430                 else
1431                         realthreads++;
1432
1433                 if (!td->pid) {
1434                         pending++;
1435                         continue;
1436                 }
1437                 if (td->runstate == TD_REAPED)
1438                         continue;
1439                 if (td->o.use_thread) {
1440                         if (td->runstate == TD_EXITED) {
1441                                 td_set_runstate(td, TD_REAPED);
1442                                 goto reaped;
1443                         }
1444                         continue;
1445                 }
1446
1447                 flags = WNOHANG;
1448                 if (td->runstate == TD_EXITED)
1449                         flags = 0;
1450
1451                 /*
1452                  * check if someone quit or got killed in an unusual way
1453                  */
1454                 ret = waitpid(td->pid, &status, flags);
1455                 if (ret < 0) {
1456                         if (errno == ECHILD) {
1457                                 log_err("fio: pid=%d disappeared %d\n",
1458                                                 (int) td->pid, td->runstate);
1459                                 td_set_runstate(td, TD_REAPED);
1460                                 goto reaped;
1461                         }
1462                         perror("waitpid");
1463                 } else if (ret == td->pid) {
1464                         if (WIFSIGNALED(status)) {
1465                                 int sig = WTERMSIG(status);
1466
1467                                 if (sig != SIGTERM)
1468                                         log_err("fio: pid=%d, got signal=%d\n",
1469                                                         (int) td->pid, sig);
1470                                 td_set_runstate(td, TD_REAPED);
1471                                 goto reaped;
1472                         }
1473                         if (WIFEXITED(status)) {
1474                                 if (WEXITSTATUS(status) && !td->error)
1475                                         td->error = WEXITSTATUS(status);
1476
1477                                 td_set_runstate(td, TD_REAPED);
1478                                 goto reaped;
1479                         }
1480                 }
1481
1482                 /*
1483                  * thread is not dead, continue
1484                  */
1485                 pending++;
1486                 continue;
1487 reaped:
1488                 (*nr_running)--;
1489                 (*m_rate) -= (td->o.ratemin[0] + td->o.ratemin[1]);
1490                 (*t_rate) -= (td->o.rate[0] + td->o.rate[1]);
1491                 if (!td->pid)
1492                         pending--;
1493
1494                 if (td->error)
1495                         exit_value++;
1496
1497                 done_secs += mtime_since_now(&td->epoch) / 1000;
1498         }
1499
1500         if (*nr_running == cputhreads && !pending && realthreads)
1501                 fio_terminate_threads(TERMINATE_ALL);
1502 }
1503
1504 /*
1505  * Main function for kicking off and reaping jobs, as needed.
1506  */
1507 static void run_threads(void)
1508 {
1509         struct thread_data *td;
1510         unsigned long spent;
1511         unsigned int i, todo, nr_running, m_rate, t_rate, nr_started;
1512
1513         if (fio_pin_memory())
1514                 return;
1515
1516         if (fio_gtod_offload && fio_start_gtod_thread())
1517                 return;
1518
1519         set_sig_handlers();
1520
1521         if (!terse_output) {
1522                 log_info("Starting ");
1523                 if (nr_thread)
1524                         log_info("%d thread%s", nr_thread,
1525                                                 nr_thread > 1 ? "s" : "");
1526                 if (nr_process) {
1527                         if (nr_thread)
1528                                 log_info(" and ");
1529                         log_info("%d process%s", nr_process,
1530                                                 nr_process > 1 ? "es" : "");
1531                 }
1532                 log_info("\n");
1533                 fflush(stdout);
1534         }
1535
1536         todo = thread_number;
1537         nr_running = 0;
1538         nr_started = 0;
1539         m_rate = t_rate = 0;
1540
1541         for_each_td(td, i) {
1542                 print_status_init(td->thread_number - 1);
1543
1544                 if (!td->o.create_serialize)
1545                         continue;
1546
1547                 /*
1548                  * do file setup here so it happens sequentially,
1549                  * we don't want X number of threads getting their
1550                  * client data interspersed on disk
1551                  */
1552                 if (setup_files(td)) {
1553                         exit_value++;
1554                         if (td->error)
1555                                 log_err("fio: pid=%d, err=%d/%s\n",
1556                                         (int) td->pid, td->error, td->verror);
1557                         td_set_runstate(td, TD_REAPED);
1558                         todo--;
1559                 } else {
1560                         struct fio_file *f;
1561                         unsigned int j;
1562
1563                         /*
1564                          * for sharing to work, each job must always open
1565                          * its own files. so close them, if we opened them
1566                          * for creation
1567                          */
1568                         for_each_file(td, f, j) {
1569                                 if (fio_file_open(f))
1570                                         td_io_close_file(td, f);
1571                         }
1572                 }
1573         }
1574
1575         set_genesis_time();
1576
1577         while (todo) {
1578                 struct thread_data *map[REAL_MAX_JOBS];
1579                 struct timeval this_start;
1580                 int this_jobs = 0, left;
1581
1582                 /*
1583                  * create threads (TD_NOT_CREATED -> TD_CREATED)
1584                  */
1585                 for_each_td(td, i) {
1586                         if (td->runstate != TD_NOT_CREATED)
1587                                 continue;
1588
1589                         /*
1590                          * never got a chance to start, killed by other
1591                          * thread for some reason
1592                          */
1593                         if (td->terminate) {
1594                                 todo--;
1595                                 continue;
1596                         }
1597
1598                         if (td->o.start_delay) {
1599                                 spent = mtime_since_genesis();
1600
1601                                 if (td->o.start_delay * 1000 > spent)
1602                                         continue;
1603                         }
1604
1605                         if (td->o.stonewall && (nr_started || nr_running)) {
1606                                 dprint(FD_PROCESS, "%s: stonewall wait\n",
1607                                                         td->o.name);
1608                                 break;
1609                         }
1610
1611                         init_disk_util(td);
1612
1613                         /*
1614                          * Set state to created. Thread will transition
1615                          * to TD_INITIALIZED when it's done setting up.
1616                          */
1617                         td_set_runstate(td, TD_CREATED);
1618                         map[this_jobs++] = td;
1619                         nr_started++;
1620
1621                         if (td->o.use_thread) {
1622                                 int ret;
1623
1624                                 dprint(FD_PROCESS, "will pthread_create\n");
1625                                 ret = pthread_create(&td->thread, NULL,
1626                                                         thread_main, td);
1627                                 if (ret) {
1628                                         log_err("pthread_create: %s\n",
1629                                                         strerror(ret));
1630                                         nr_started--;
1631                                         break;
1632                                 }
1633                                 ret = pthread_detach(td->thread);
1634                                 if (ret)
1635                                         log_err("pthread_detach: %s",
1636                                                         strerror(ret));
1637                         } else {
1638                                 pid_t pid;
1639                                 dprint(FD_PROCESS, "will fork\n");
1640                                 pid = fork();
1641                                 if (!pid) {
1642                                         int ret = fork_main(shm_id, i);
1643
1644                                         _exit(ret);
1645                                 } else if (i == fio_debug_jobno)
1646                                         *fio_debug_jobp = pid;
1647                         }
1648                         dprint(FD_MUTEX, "wait on startup_mutex\n");
1649                         if (fio_mutex_down_timeout(startup_mutex, 10)) {
1650                                 log_err("fio: job startup hung? exiting.\n");
1651                                 fio_terminate_threads(TERMINATE_ALL);
1652                                 fio_abort = 1;
1653                                 nr_started--;
1654                                 break;
1655                         }
1656                         dprint(FD_MUTEX, "done waiting on startup_mutex\n");
1657                 }
1658
1659                 /*
1660                  * Wait for the started threads to transition to
1661                  * TD_INITIALIZED.
1662                  */
1663                 fio_gettime(&this_start, NULL);
1664                 left = this_jobs;
1665                 while (left && !fio_abort) {
1666                         if (mtime_since_now(&this_start) > JOB_START_TIMEOUT)
1667                                 break;
1668
1669                         usleep(100000);
1670
1671                         for (i = 0; i < this_jobs; i++) {
1672                                 td = map[i];
1673                                 if (!td)
1674                                         continue;
1675                                 if (td->runstate == TD_INITIALIZED) {
1676                                         map[i] = NULL;
1677                                         left--;
1678                                 } else if (td->runstate >= TD_EXITED) {
1679                                         map[i] = NULL;
1680                                         left--;
1681                                         todo--;
1682                                         nr_running++; /* work-around... */
1683                                 }
1684                         }
1685                 }
1686
1687                 if (left) {
1688                         log_err("fio: %d jobs failed to start\n", left);
1689                         for (i = 0; i < this_jobs; i++) {
1690                                 td = map[i];
1691                                 if (!td)
1692                                         continue;
1693                                 kill(td->pid, SIGTERM);
1694                         }
1695                         break;
1696                 }
1697
1698                 /*
1699                  * start created threads (TD_INITIALIZED -> TD_RUNNING).
1700                  */
1701                 for_each_td(td, i) {
1702                         if (td->runstate != TD_INITIALIZED)
1703                                 continue;
1704
1705                         if (in_ramp_time(td))
1706                                 td_set_runstate(td, TD_RAMP);
1707                         else
1708                                 td_set_runstate(td, TD_RUNNING);
1709                         nr_running++;
1710                         nr_started--;
1711                         m_rate += td->o.ratemin[0] + td->o.ratemin[1];
1712                         t_rate += td->o.rate[0] + td->o.rate[1];
1713                         todo--;
1714                         fio_mutex_up(td->mutex);
1715                 }
1716
1717                 reap_threads(&nr_running, &t_rate, &m_rate);
1718
1719                 if (todo) {
1720                         if (is_backend)
1721                                 fio_server_idle_loop();
1722                         else
1723                                 usleep(100000);
1724                 }
1725         }
1726
1727         while (nr_running) {
1728                 reap_threads(&nr_running, &t_rate, &m_rate);
1729
1730                 if (is_backend)
1731                         fio_server_idle_loop();
1732                 else
1733                         usleep(10000);
1734         }
1735
1736         update_io_ticks();
1737         fio_unpin_memory();
1738 }
1739
1740 int exec_run(void)
1741 {
1742         struct thread_data *td;
1743         int i;
1744
1745         if (nr_clients)
1746                 return fio_handle_clients();
1747         if (exec_profile) {
1748                 if (load_profile(exec_profile))
1749                         return 1;
1750                 free(exec_profile);
1751                 exec_profile = NULL;
1752         }
1753         if (!thread_number)
1754                 return 0;
1755
1756         if (write_bw_log) {
1757                 setup_log(&agg_io_log[DDIR_READ], 0);
1758                 setup_log(&agg_io_log[DDIR_WRITE], 0);
1759         }
1760
1761         startup_mutex = fio_mutex_init(0);
1762         if (startup_mutex == NULL)
1763                 return 1;
1764         writeout_mutex = fio_mutex_init(1);
1765         if (writeout_mutex == NULL)
1766                 return 1;
1767
1768         set_genesis_time();
1769         create_disk_util_thread();
1770
1771         cgroup_list = smalloc(sizeof(*cgroup_list));
1772         INIT_FLIST_HEAD(cgroup_list);
1773
1774         run_threads();
1775
1776         if (!fio_abort) {
1777                 show_run_stats();
1778                 if (write_bw_log) {
1779                         __finish_log(agg_io_log[DDIR_READ], "agg-read_bw.log");
1780                         __finish_log(agg_io_log[DDIR_WRITE],
1781                                         "agg-write_bw.log");
1782                 }
1783         }
1784
1785         for_each_td(td, i)
1786                 fio_options_free(td);
1787
1788         cgroup_kill(cgroup_list);
1789         sfree(cgroup_list);
1790         sfree(cgroup_mnt);
1791
1792         fio_mutex_remove(startup_mutex);
1793         fio_mutex_remove(writeout_mutex);
1794         return exit_value;
1795 }
1796
1797 void reset_fio_state(void)
1798 {
1799         groupid = 0;
1800         thread_number = 0;
1801         nr_process = 0;
1802         nr_thread = 0;
1803         done_secs = 0;
1804 }
1805
1806 static int endian_check(void)
1807 {
1808         union {
1809                 uint8_t c[8];
1810                 uint64_t v;
1811         } u;
1812         int le = 0, be = 0;
1813
1814         u.v = 0x12;
1815         if (u.c[7] == 0x12)
1816                 be = 1;
1817         else if (u.c[0] == 0x12)
1818                 le = 1;
1819
1820 #if defined(FIO_LITTLE_ENDIAN)
1821         if (be)
1822                 return 1;
1823 #elif defined(FIO_BIG_ENDIAN)
1824         if (le)
1825                 return 1;
1826 #else
1827         return 1;
1828 #endif
1829
1830         if (!le && !be)
1831                 return 1;
1832
1833         return 0;
1834 }
1835
1836 int main(int argc, char *argv[], char *envp[])
1837 {
1838         long ps;
1839
1840         if (endian_check()) {
1841                 log_err("fio: endianness settings appear wrong.\n");
1842                 log_err("fio: please report this to fio@vger.kernel.org\n");
1843                 return 1;
1844         }
1845
1846         arch_init(envp);
1847
1848         sinit();
1849
1850         /*
1851          * We need locale for number printing, if it isn't set then just
1852          * go with the US format.
1853          */
1854         if (!getenv("LC_NUMERIC"))
1855                 setlocale(LC_NUMERIC, "en_US");
1856
1857         ps = sysconf(_SC_PAGESIZE);
1858         if (ps < 0) {
1859                 log_err("Failed to get page size\n");
1860                 return 1;
1861         }
1862
1863         page_size = ps;
1864         page_mask = ps - 1;
1865
1866         fio_keywords_init();
1867
1868         if (parse_options(argc, argv))
1869                 return 1;
1870
1871         return exec_run();
1872 }