Rename thread_options' ->io_limit to io_size
[fio.git] / backend.c
CommitLineData
2e1df07d
JA
1/*
2 * fio - the flexible io tester
3 *
4 * Copyright (C) 2005 Jens Axboe <axboe@suse.de>
5 * Copyright (C) 2006-2012 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>
e43606c2 33#include <inttypes.h>
2e1df07d
JA
34#include <sys/stat.h>
35#include <sys/wait.h>
36#include <sys/ipc.h>
2e1df07d 37#include <sys/mman.h>
ff6bb260 38#include <math.h>
2e1df07d
JA
39
40#include "fio.h"
a5e0ee11
O
41#ifndef FIO_NO_HAVE_SHM_H
42#include <sys/shm.h>
43#endif
2e1df07d
JA
44#include "hash.h"
45#include "smalloc.h"
46#include "verify.h"
47#include "trim.h"
48#include "diskutil.h"
49#include "cgroup.h"
50#include "profile.h"
51#include "lib/rand.h"
f7690c4a 52#include "lib/memalign.h"
2e1df07d 53#include "server.h"
44404c5a 54#include "lib/getrusage.h"
f2a2ce0e 55#include "idletime.h"
002fe734 56#include "err.h"
a9da8ab2 57#include "workqueue.h"
e81ecca3 58#include "lib/mountcheck.h"
40511301 59#include "rate-submit.h"
a39fb9ea 60#include "helper_thread.h"
2e1df07d 61
2e1df07d 62static struct fio_mutex *startup_mutex;
2e1df07d
JA
63static struct flist_head *cgroup_list;
64static char *cgroup_mnt;
65static int exit_value;
66static volatile int fio_abort;
3a5f6bde
JA
67static unsigned int nr_process = 0;
68static unsigned int nr_thread = 0;
2e1df07d 69
6eaf09d6 70struct io_log *agg_io_log[DDIR_RWDIR_CNT];
2e1df07d 71
a3efc919
JA
72int groupid = 0;
73unsigned int thread_number = 0;
108fea77 74unsigned int stat_number = 0;
a3efc919
JA
75int shm_id = 0;
76int temp_stall_ts;
77unsigned long done_secs = 0;
a47591e4 78
2e1df07d
JA
79#define JOB_START_TIMEOUT (5 * 1000)
80
81static void sig_int(int sig)
82{
83 if (threads) {
84 if (is_backend)
85 fio_server_got_signal(sig);
86 else {
87 log_info("\nfio: terminating on signal %d\n", sig);
e411c301 88 log_info_flush();
2e1df07d
JA
89 exit_value = 128;
90 }
91
92 fio_terminate_threads(TERMINATE_ALL);
93 }
94}
95
d2235e56 96void sig_show_status(int sig)
4c6d91e8
JA
97{
98 show_running_run_stats();
99}
100
2e1df07d
JA
101static void set_sig_handlers(void)
102{
103 struct sigaction act;
104
105 memset(&act, 0, sizeof(act));
106 act.sa_handler = sig_int;
107 act.sa_flags = SA_RESTART;
108 sigaction(SIGINT, &act, NULL);
109
110 memset(&act, 0, sizeof(act));
111 act.sa_handler = sig_int;
112 act.sa_flags = SA_RESTART;
113 sigaction(SIGTERM, &act, NULL);
114
2f694507
BC
115/* Windows uses SIGBREAK as a quit signal from other applications */
116#ifdef WIN32
117 memset(&act, 0, sizeof(act));
118 act.sa_handler = sig_int;
119 act.sa_flags = SA_RESTART;
120 sigaction(SIGBREAK, &act, NULL);
121#endif
122
4c6d91e8
JA
123 memset(&act, 0, sizeof(act));
124 act.sa_handler = sig_show_status;
125 act.sa_flags = SA_RESTART;
126 sigaction(SIGUSR1, &act, NULL);
127
2e1df07d
JA
128 if (is_backend) {
129 memset(&act, 0, sizeof(act));
130 act.sa_handler = sig_int;
131 act.sa_flags = SA_RESTART;
132 sigaction(SIGPIPE, &act, NULL);
133 }
134}
135
136/*
137 * Check if we are above the minimum rate given.
138 */
e39c0676
JA
139static bool __check_min_rate(struct thread_data *td, struct timeval *now,
140 enum fio_ddir ddir)
2e1df07d
JA
141{
142 unsigned long long bytes = 0;
143 unsigned long iops = 0;
144 unsigned long spent;
145 unsigned long rate;
146 unsigned int ratemin = 0;
147 unsigned int rate_iops = 0;
148 unsigned int rate_iops_min = 0;
149
150 assert(ddir_rw(ddir));
151
152 if (!td->o.ratemin[ddir] && !td->o.rate_iops_min[ddir])
e39c0676 153 return false;
2e1df07d
JA
154
155 /*
156 * allow a 2 second settle period in the beginning
157 */
158 if (mtime_since(&td->start, now) < 2000)
e39c0676 159 return false;
2e1df07d
JA
160
161 iops += td->this_io_blocks[ddir];
162 bytes += td->this_io_bytes[ddir];
163 ratemin += td->o.ratemin[ddir];
164 rate_iops += td->o.rate_iops[ddir];
165 rate_iops_min += td->o.rate_iops_min[ddir];
166
167 /*
168 * if rate blocks is set, sample is running
169 */
170 if (td->rate_bytes[ddir] || td->rate_blocks[ddir]) {
171 spent = mtime_since(&td->lastrate[ddir], now);
172 if (spent < td->o.ratecycle)
e39c0676 173 return false;
2e1df07d 174
20e37e4a 175 if (td->o.rate[ddir] || td->o.ratemin[ddir]) {
2e1df07d
JA
176 /*
177 * check bandwidth specified rate
178 */
179 if (bytes < td->rate_bytes[ddir]) {
420b104a
RE
180 log_err("%s: rate_min=%uB/s not met, only transferred %lluB\n",
181 td->o.name, ratemin, bytes);
e39c0676 182 return true;
2e1df07d 183 } else {
49cba9b3
JA
184 if (spent)
185 rate = ((bytes - td->rate_bytes[ddir]) * 1000) / spent;
186 else
187 rate = 0;
188
2e1df07d
JA
189 if (rate < ratemin ||
190 bytes < td->rate_bytes[ddir]) {
420b104a
RE
191 log_err("%s: rate_min=%uB/s not met, got %luB/s\n",
192 td->o.name, ratemin, rate);
e39c0676 193 return true;
2e1df07d
JA
194 }
195 }
196 } else {
197 /*
198 * checks iops specified rate
199 */
200 if (iops < rate_iops) {
420b104a
RE
201 log_err("%s: rate_iops_min=%u not met, only performed %lu IOs\n",
202 td->o.name, rate_iops, iops);
e39c0676 203 return true;
2e1df07d 204 } else {
4c707a3b
JA
205 if (spent)
206 rate = ((iops - td->rate_blocks[ddir]) * 1000) / spent;
207 else
208 rate = 0;
209
2e1df07d
JA
210 if (rate < rate_iops_min ||
211 iops < td->rate_blocks[ddir]) {
420b104a
RE
212 log_err("%s: rate_iops_min=%u not met, got %lu IOPS\n",
213 td->o.name, rate_iops_min, rate);
e39c0676 214 return true;
2e1df07d
JA
215 }
216 }
217 }
218 }
219
220 td->rate_bytes[ddir] = bytes;
221 td->rate_blocks[ddir] = iops;
222 memcpy(&td->lastrate[ddir], now, sizeof(*now));
e39c0676 223 return false;
2e1df07d
JA
224}
225
e39c0676 226static bool check_min_rate(struct thread_data *td, struct timeval *now)
2e1df07d 227{
e39c0676 228 bool ret = false;
2e1df07d 229
55312f9f 230 if (td->bytes_done[DDIR_READ])
6eaf09d6 231 ret |= __check_min_rate(td, now, DDIR_READ);
55312f9f 232 if (td->bytes_done[DDIR_WRITE])
6eaf09d6 233 ret |= __check_min_rate(td, now, DDIR_WRITE);
55312f9f 234 if (td->bytes_done[DDIR_TRIM])
6eaf09d6 235 ret |= __check_min_rate(td, now, DDIR_TRIM);
2e1df07d
JA
236
237 return ret;
238}
239
240/*
241 * When job exits, we can cancel the in-flight IO if we are using async
242 * io. Attempt to do so.
243 */
244static void cleanup_pending_aio(struct thread_data *td)
245{
2e1df07d
JA
246 int r;
247
248 /*
249 * get immediately available events, if any
250 */
55312f9f 251 r = io_u_queued_complete(td, 0);
2e1df07d
JA
252 if (r < 0)
253 return;
254
255 /*
256 * now cancel remaining active events
257 */
258 if (td->io_ops->cancel) {
2ae0b204
JA
259 struct io_u *io_u;
260 int i;
2e1df07d 261
2ae0b204
JA
262 io_u_qiter(&td->io_u_all, io_u, i) {
263 if (io_u->flags & IO_U_F_FLIGHT) {
2e1df07d
JA
264 r = td->io_ops->cancel(td, io_u);
265 if (!r)
266 put_io_u(td, io_u);
267 }
268 }
269 }
270
271 if (td->cur_depth)
55312f9f 272 r = io_u_queued_complete(td, td->cur_depth);
2e1df07d
JA
273}
274
275/*
276 * Helper to handle the final sync of a file. Works just like the normal
277 * io path, just does everything sync.
278 */
e39c0676 279static bool fio_io_sync(struct thread_data *td, struct fio_file *f)
2e1df07d
JA
280{
281 struct io_u *io_u = __get_io_u(td);
282 int ret;
283
284 if (!io_u)
e39c0676 285 return true;
2e1df07d
JA
286
287 io_u->ddir = DDIR_SYNC;
288 io_u->file = f;
289
290 if (td_io_prep(td, io_u)) {
291 put_io_u(td, io_u);
e39c0676 292 return true;
2e1df07d
JA
293 }
294
295requeue:
296 ret = td_io_queue(td, io_u);
297 if (ret < 0) {
298 td_verror(td, io_u->error, "td_io_queue");
299 put_io_u(td, io_u);
e39c0676 300 return true;
2e1df07d 301 } else if (ret == FIO_Q_QUEUED) {
42ff945c
JA
302 if (td_io_commit(td))
303 return true;
55312f9f 304 if (io_u_queued_complete(td, 1) < 0)
e39c0676 305 return true;
2e1df07d
JA
306 } else if (ret == FIO_Q_COMPLETED) {
307 if (io_u->error) {
308 td_verror(td, io_u->error, "td_io_queue");
e39c0676 309 return true;
2e1df07d
JA
310 }
311
55312f9f 312 if (io_u_sync_complete(td, io_u) < 0)
e39c0676 313 return true;
2e1df07d
JA
314 } else if (ret == FIO_Q_BUSY) {
315 if (td_io_commit(td))
e39c0676 316 return true;
2e1df07d
JA
317 goto requeue;
318 }
319
e39c0676 320 return false;
2e1df07d 321}
a3efc919 322
61ee0f86
JA
323static int fio_file_fsync(struct thread_data *td, struct fio_file *f)
324{
325 int ret;
326
327 if (fio_file_open(f))
328 return fio_io_sync(td, f);
329
330 if (td_io_open_file(td, f))
331 return 1;
332
333 ret = fio_io_sync(td, f);
334 td_io_close_file(td, f);
335 return ret;
336}
337
2e1df07d
JA
338static inline void __update_tv_cache(struct thread_data *td)
339{
340 fio_gettime(&td->tv_cache, NULL);
341}
342
343static inline void update_tv_cache(struct thread_data *td)
344{
345 if ((++td->tv_cache_nr & td->tv_cache_mask) == td->tv_cache_mask)
346 __update_tv_cache(td);
347}
348
e39c0676 349static inline bool runtime_exceeded(struct thread_data *td, struct timeval *t)
2e1df07d
JA
350{
351 if (in_ramp_time(td))
e39c0676 352 return false;
2e1df07d 353 if (!td->o.timeout)
e39c0676 354 return false;
0de5b26f 355 if (utime_since(&td->epoch, t) >= td->o.timeout)
e39c0676 356 return true;
2e1df07d 357
e39c0676 358 return false;
2e1df07d
JA
359}
360
95603b74
BF
361/*
362 * We need to update the runtime consistently in ms, but keep a running
363 * tally of the current elapsed time in microseconds for sub millisecond
364 * updates.
365 */
366static inline void update_runtime(struct thread_data *td,
367 unsigned long long *elapsed_us,
368 const enum fio_ddir ddir)
369{
d5c8ea29
JA
370 if (ddir == DDIR_WRITE && td_write(td) && td->o.verify_only)
371 return;
372
95603b74
BF
373 td->ts.runtime[ddir] -= (elapsed_us[ddir] + 999) / 1000;
374 elapsed_us[ddir] += utime_since_now(&td->start);
375 td->ts.runtime[ddir] += (elapsed_us[ddir] + 999) / 1000;
376}
377
e39c0676
JA
378static bool break_on_this_error(struct thread_data *td, enum fio_ddir ddir,
379 int *retptr)
2e1df07d
JA
380{
381 int ret = *retptr;
382
383 if (ret < 0 || td->error) {
8b28bd41
DM
384 int err = td->error;
385 enum error_type_bit eb;
2e1df07d
JA
386
387 if (ret < 0)
388 err = -ret;
2e1df07d 389
8b28bd41
DM
390 eb = td_error_type(ddir, err);
391 if (!(td->o.continue_on_error & (1 << eb)))
e39c0676 392 return true;
2e1df07d 393
8b28bd41 394 if (td_non_fatal_error(td, eb, err)) {
2e1df07d
JA
395 /*
396 * Continue with the I/Os in case of
397 * a non fatal error.
398 */
399 update_error_count(td, err);
400 td_clear_error(td);
401 *retptr = 0;
e39c0676 402 return false;
2e1df07d
JA
403 } else if (td->o.fill_device && err == ENOSPC) {
404 /*
405 * We expect to hit this error if
406 * fill_device option is set.
407 */
408 td_clear_error(td);
ebea2133 409 fio_mark_td_terminate(td);
e39c0676 410 return true;
2e1df07d
JA
411 } else {
412 /*
413 * Stop the I/O in case of a fatal
414 * error.
415 */
416 update_error_count(td, err);
e39c0676 417 return true;
2e1df07d
JA
418 }
419 }
420
e39c0676 421 return false;
2e1df07d
JA
422}
423
c97f1ad6
JA
424static void check_update_rusage(struct thread_data *td)
425{
426 if (td->update_rusage) {
427 td->update_rusage = 0;
428 update_rusage_stat(td);
429 fio_mutex_up(td->rusage_sem);
430 }
431}
432
55312f9f 433static int wait_for_completions(struct thread_data *td, struct timeval *time)
69fea81e
JA
434{
435 const int full = queue_full(td);
436 int min_evts = 0;
437 int ret;
438
6be06c46
JA
439 if (td->flags & TD_F_REGROW_LOGS)
440 return io_u_quiesce(td);
1fed2080 441
69fea81e
JA
442 /*
443 * if the queue is full, we MUST reap at least 1 event
444 */
82407585 445 min_evts = min(td->o.iodepth_batch_complete_min, td->cur_depth);
4b157ac6 446 if ((full && !min_evts) || !td->o.iodepth_batch_complete_min)
69fea81e
JA
447 min_evts = 1;
448
66b6c5ef 449 if (time && (__should_check_rate(td, DDIR_READ) ||
69fea81e
JA
450 __should_check_rate(td, DDIR_WRITE) ||
451 __should_check_rate(td, DDIR_TRIM)))
452 fio_gettime(time, NULL);
453
454 do {
55312f9f 455 ret = io_u_queued_complete(td, min_evts);
69fea81e
JA
456 if (ret < 0)
457 break;
458 } while (full && (td->cur_depth > td->o.iodepth_low));
459
460 return ret;
461}
462
e9d512d8
JA
463int io_queue_event(struct thread_data *td, struct io_u *io_u, int *ret,
464 enum fio_ddir ddir, uint64_t *bytes_issued, int from_verify,
465 struct timeval *comp_time)
466{
467 int ret2;
468
469 switch (*ret) {
470 case FIO_Q_COMPLETED:
471 if (io_u->error) {
472 *ret = -io_u->error;
473 clear_io_u(td, io_u);
474 } else if (io_u->resid) {
475 int bytes = io_u->xfer_buflen - io_u->resid;
476 struct fio_file *f = io_u->file;
477
478 if (bytes_issued)
479 *bytes_issued += bytes;
480
481 if (!from_verify)
482 trim_io_piece(td, io_u);
483
484 /*
485 * zero read, fail
486 */
487 if (!bytes) {
488 if (!from_verify)
489 unlog_io_piece(td, io_u);
490 td_verror(td, EIO, "full resid");
491 put_io_u(td, io_u);
492 break;
493 }
494
495 io_u->xfer_buflen = io_u->resid;
496 io_u->xfer_buf += bytes;
497 io_u->offset += bytes;
498
499 if (ddir_rw(io_u->ddir))
500 td->ts.short_io_u[io_u->ddir]++;
501
502 f = io_u->file;
503 if (io_u->offset == f->real_file_size)
504 goto sync_done;
505
506 requeue_io_u(td, &io_u);
507 } else {
508sync_done:
509 if (comp_time && (__should_check_rate(td, DDIR_READ) ||
510 __should_check_rate(td, DDIR_WRITE) ||
511 __should_check_rate(td, DDIR_TRIM)))
512 fio_gettime(comp_time, NULL);
513
514 *ret = io_u_sync_complete(td, io_u);
515 if (*ret < 0)
516 break;
517 }
a0e3e5a6 518
cf2c8d52
JA
519 if (td->flags & TD_F_REGROW_LOGS)
520 regrow_logs(td);
521
a0e3e5a6
EW
522 /*
523 * when doing I/O (not when verifying),
524 * check for any errors that are to be ignored
525 */
526 if (!from_verify)
527 break;
528
e9d512d8
JA
529 return 0;
530 case FIO_Q_QUEUED:
531 /*
532 * if the engine doesn't have a commit hook,
533 * the io_u is really queued. if it does have such
534 * a hook, it has to call io_u_queued() itself.
535 */
536 if (td->io_ops->commit == NULL)
537 io_u_queued(td, io_u);
538 if (bytes_issued)
539 *bytes_issued += io_u->xfer_buflen;
540 break;
541 case FIO_Q_BUSY:
542 if (!from_verify)
543 unlog_io_piece(td, io_u);
544 requeue_io_u(td, &io_u);
545 ret2 = td_io_commit(td);
546 if (ret2 < 0)
547 *ret = ret2;
548 break;
549 default:
b0775325 550 assert(*ret < 0);
e9d512d8
JA
551 td_verror(td, -(*ret), "td_io_queue");
552 break;
553 }
554
555 if (break_on_this_error(td, ddir, ret))
556 return 1;
557
558 return 0;
559}
560
e39c0676 561static inline bool io_in_polling(struct thread_data *td)
82407585
RP
562{
563 return !td->o.iodepth_batch_complete_min &&
564 !td->o.iodepth_batch_complete_max;
565}
39c1c323 566/*
567 * Unlinks files from thread data fio_file structure
568 */
569static int unlink_all_files(struct thread_data *td)
570{
571 struct fio_file *f;
572 unsigned int i;
573 int ret = 0;
574
575 for_each_file(td, f, i) {
576 if (f->filetype != FIO_TYPE_FILE)
577 continue;
578 ret = td_io_unlink_file(td, f);
579 if (ret)
580 break;
581 }
582
583 if (ret)
584 td_verror(td, ret, "unlink_all_files");
585
586 return ret;
587}
82407585 588
2e1df07d
JA
589/*
590 * The main verify engine. Runs over the writes we previously submitted,
591 * reads the blocks back in, and checks the crc/md5 of the data.
592 */
100f49f1 593static void do_verify(struct thread_data *td, uint64_t verify_bytes)
2e1df07d
JA
594{
595 struct fio_file *f;
596 struct io_u *io_u;
597 int ret, min_events;
598 unsigned int i;
599
600 dprint(FD_VERIFY, "starting loop\n");
601
602 /*
603 * sync io first and invalidate cache, to make sure we really
604 * read from disk.
605 */
606 for_each_file(td, f, i) {
607 if (!fio_file_open(f))
608 continue;
609 if (fio_io_sync(td, f))
610 break;
611 if (file_invalidate_cache(td, f))
612 break;
613 }
614
c97f1ad6
JA
615 check_update_rusage(td);
616
2e1df07d
JA
617 if (td->error)
618 return;
619
8f380de2
CJ
620 /*
621 * verify_state needs to be reset before verification
622 * proceeds so that expected random seeds match actual
623 * random seeds in headers. The main loop will reset
624 * all random number generators if randrepeat is set.
625 */
626 if (!td->o.rand_repeatable)
627 td_fill_verify_state_seed(td);
628
2e1df07d
JA
629 td_set_runstate(td, TD_VERIFYING);
630
631 io_u = NULL;
632 while (!td->terminate) {
fbccf46c 633 enum fio_ddir ddir;
e9d512d8 634 int full;
2e1df07d
JA
635
636 update_tv_cache(td);
c97f1ad6 637 check_update_rusage(td);
2e1df07d
JA
638
639 if (runtime_exceeded(td, &td->tv_cache)) {
640 __update_tv_cache(td);
641 if (runtime_exceeded(td, &td->tv_cache)) {
ebea2133 642 fio_mark_td_terminate(td);
2e1df07d
JA
643 break;
644 }
645 }
646
9e684a49
DE
647 if (flow_threshold_exceeded(td))
648 continue;
649
44cbc6da
JA
650 if (!td->o.experimental_verify) {
651 io_u = __get_io_u(td);
652 if (!io_u)
653 break;
2e1df07d 654
44cbc6da
JA
655 if (get_next_verify(td, io_u)) {
656 put_io_u(td, io_u);
657 break;
658 }
2e1df07d 659
44cbc6da
JA
660 if (td_io_prep(td, io_u)) {
661 put_io_u(td, io_u);
662 break;
663 }
664 } else {
55312f9f 665 if (ddir_rw_sum(td->bytes_done) + td->o.rw_min_bs > verify_bytes)
100f49f1
JA
666 break;
667
bcd5abfa 668 while ((io_u = get_io_u(td)) != NULL) {
059b61f2 669 if (IS_ERR_OR_NULL(io_u)) {
002fe734
JA
670 io_u = NULL;
671 ret = FIO_Q_BUSY;
672 goto reap;
673 }
674
bcd5abfa
JA
675 /*
676 * We are only interested in the places where
677 * we wrote or trimmed IOs. Turn those into
678 * reads for verification purposes.
679 */
680 if (io_u->ddir == DDIR_READ) {
681 /*
682 * Pretend we issued it for rwmix
683 * accounting
684 */
685 td->io_issues[DDIR_READ]++;
686 put_io_u(td, io_u);
687 continue;
688 } else if (io_u->ddir == DDIR_TRIM) {
689 io_u->ddir = DDIR_READ;
1651e431 690 io_u_set(td, io_u, IO_U_F_TRIMMED);
bcd5abfa
JA
691 break;
692 } else if (io_u->ddir == DDIR_WRITE) {
693 io_u->ddir = DDIR_READ;
694 break;
695 } else {
696 put_io_u(td, io_u);
697 continue;
698 }
699 }
44cbc6da 700
bcd5abfa 701 if (!io_u)
44cbc6da 702 break;
2e1df07d
JA
703 }
704
ca09be4b
JA
705 if (verify_state_should_stop(td, io_u)) {
706 put_io_u(td, io_u);
707 break;
708 }
709
2e1df07d
JA
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
fbccf46c 715 ddir = io_u->ddir;
10adc4a7
GG
716 if (!td->o.disable_slat)
717 fio_gettime(&io_u->start_time, NULL);
fbccf46c 718
2e1df07d 719 ret = td_io_queue(td, io_u);
2e1df07d 720
e9d512d8 721 if (io_queue_event(td, io_u, &ret, ddir, NULL, 1, NULL))
2e1df07d
JA
722 break;
723
724 /*
725 * if we can queue more, do so. but check if there are
726 * completed io_u's first. Note that we can get BUSY even
727 * without IO queued, if the system is resource starved.
728 */
002fe734 729reap:
2e1df07d 730 full = queue_full(td) || (ret == FIO_Q_BUSY && td->cur_depth);
82407585 731 if (full || io_in_polling(td))
55312f9f 732 ret = wait_for_completions(td, NULL);
2e1df07d 733
2e1df07d
JA
734 if (ret < 0)
735 break;
736 }
737
c97f1ad6
JA
738 check_update_rusage(td);
739
2e1df07d
JA
740 if (!td->error) {
741 min_events = td->cur_depth;
742
743 if (min_events)
55312f9f 744 ret = io_u_queued_complete(td, min_events);
2e1df07d
JA
745 } else
746 cleanup_pending_aio(td);
747
748 td_set_runstate(td, TD_RUNNING);
749
750 dprint(FD_VERIFY, "exiting loop\n");
751}
752
e39c0676 753static bool exceeds_number_ios(struct thread_data *td)
3939fe85
JA
754{
755 unsigned long long number_ios;
756
757 if (!td->o.number_ios)
e39c0676 758 return false;
3939fe85 759
cf8a46a7 760 number_ios = ddir_rw_sum(td->io_blocks);
3939fe85
JA
761 number_ios += td->io_u_queued + td->io_u_in_flight;
762
cf8a46a7 763 return number_ios >= (td->o.number_ios * td->loops);
3939fe85
JA
764}
765
926098e0 766static bool io_bytes_exceeded(struct thread_data *td, uint64_t *this_bytes)
f7078f7b 767{
77731b29 768 unsigned long long bytes, limit;
f7078f7b
JA
769
770 if (td_rw(td))
926098e0 771 bytes = this_bytes[DDIR_READ] + this_bytes[DDIR_WRITE];
f7078f7b 772 else if (td_write(td))
926098e0 773 bytes = this_bytes[DDIR_WRITE];
6eaf09d6 774 else if (td_read(td))
926098e0 775 bytes = this_bytes[DDIR_READ];
f7078f7b 776 else
926098e0 777 bytes = this_bytes[DDIR_TRIM];
f7078f7b 778
5be9bf09
TK
779 if (td->o.io_size)
780 limit = td->o.io_size;
77731b29
JA
781 else
782 limit = td->o.size;
783
cf8a46a7 784 limit *= td->loops;
77731b29 785 return bytes >= limit || exceeds_number_ios(td);
f7078f7b
JA
786}
787
926098e0 788static bool io_issue_bytes_exceeded(struct thread_data *td)
e28dd2cf 789{
926098e0
JA
790 return io_bytes_exceeded(td, td->io_issue_bytes);
791}
e28dd2cf 792
926098e0
JA
793static bool io_complete_bytes_exceeded(struct thread_data *td)
794{
795 return io_bytes_exceeded(td, td->this_io_bytes);
e28dd2cf
JE
796}
797
50a8ce86
D
798/*
799 * used to calculate the next io time for rate control
800 *
801 */
802static long long usec_for_io(struct thread_data *td, enum fio_ddir ddir)
803{
ff6bb260 804 uint64_t secs, remainder, bps, bytes, iops;
50a8ce86
D
805
806 assert(!(td->flags & TD_F_CHILD));
807 bytes = td->rate_io_issue_bytes[ddir];
808 bps = td->rate_bps[ddir];
ff6bb260 809
6de65959 810 if (td->o.rate_process == RATE_PROCESS_POISSON) {
a77d139b 811 uint64_t val;
ff6bb260 812 iops = bps / td->o.bs[ddir];
a77d139b
JA
813 val = (int64_t) (1000000 / iops) *
814 -logf(__rand_0_1(&td->poisson_state));
815 if (val) {
816 dprint(FD_RATE, "poisson rate iops=%llu\n",
817 (unsigned long long) 1000000 / val);
818 }
819 td->last_usec += val;
ff6bb260
SL
820 return td->last_usec;
821 } else if (bps) {
50a8ce86
D
822 secs = bytes / bps;
823 remainder = bytes % bps;
824 return remainder * 1000000 / bps + secs * 1000000;
e7b24047
JA
825 }
826
827 return 0;
50a8ce86
D
828}
829
2e1df07d
JA
830/*
831 * Main IO worker function. It retrieves io_u's to process and queues
832 * and reaps them, checking for rate and errors along the way.
100f49f1
JA
833 *
834 * Returns number of bytes written and trimmed.
2e1df07d 835 */
95b03be0 836static void do_io(struct thread_data *td, uint64_t *bytes_done)
2e1df07d
JA
837{
838 unsigned int i;
839 int ret = 0;
c2703bf3 840 uint64_t total_bytes, bytes_issued = 0;
70c68076 841
95b03be0
JA
842 for (i = 0; i < DDIR_RWDIR_CNT; i++)
843 bytes_done[i] = td->bytes_done[i];
2e1df07d
JA
844
845 if (in_ramp_time(td))
846 td_set_runstate(td, TD_RAMP);
847 else
848 td_set_runstate(td, TD_RUNNING);
849
3e260a46
JA
850 lat_target_init(td);
851
1e564979
JE
852 total_bytes = td->o.size;
853 /*
5be9bf09 854 * Allow random overwrite workloads to write up to io_size
1e564979
JE
855 * before starting verification phase as 'size' doesn't apply.
856 */
857 if (td_write(td) && td_random(td) && td->o.norandommap)
5be9bf09 858 total_bytes = max(total_bytes, (uint64_t) td->o.io_size);
78a6469c
JA
859 /*
860 * If verify_backlog is enabled, we'll run the verify in this
861 * handler as well. For that case, we may need up to twice the
862 * amount of bytes.
863 */
78a6469c
JA
864 if (td->o.verify != VERIFY_NONE &&
865 (td_write(td) && td->o.verify_backlog))
c2703bf3
JA
866 total_bytes += td->o.size;
867
82a90686 868 /* In trimwrite mode, each byte is trimmed and then written, so
0e4dd95c 869 * allow total_bytes to be twice as big */
82a90686 870 if (td_trimwrite(td))
0e4dd95c
DE
871 total_bytes += td->total_io_size;
872
f7078f7b 873 while ((td->o.read_iolog_file && !flist_empty(&td->io_log_list)) ||
e28dd2cf 874 (!flist_empty(&td->trim_list)) || !io_issue_bytes_exceeded(td) ||
c04e4661 875 td->o.time_based) {
2e1df07d 876 struct timeval comp_time;
2e1df07d 877 struct io_u *io_u;
e9d512d8 878 int full;
2e1df07d
JA
879 enum fio_ddir ddir;
880
c97f1ad6
JA
881 check_update_rusage(td);
882
7d7803fa 883 if (td->terminate || td->done)
2e1df07d
JA
884 break;
885
886 update_tv_cache(td);
887
888 if (runtime_exceeded(td, &td->tv_cache)) {
889 __update_tv_cache(td);
890 if (runtime_exceeded(td, &td->tv_cache)) {
ebea2133 891 fio_mark_td_terminate(td);
2e1df07d
JA
892 break;
893 }
894 }
895
9e684a49
DE
896 if (flow_threshold_exceeded(td))
897 continue;
898
f1a32461
JA
899 /*
900 * Break if we exceeded the bytes. The exception is time
901 * based runs, but we still need to break out of the loop
902 * for those to run verification, if enabled.
903 */
904 if (bytes_issued >= total_bytes &&
905 (!td->o.time_based ||
906 (td->o.time_based && td->o.verify != VERIFY_NONE)))
20876c53
JC
907 break;
908
2e1df07d 909 io_u = get_io_u(td);
002fe734
JA
910 if (IS_ERR_OR_NULL(io_u)) {
911 int err = PTR_ERR(io_u);
912
913 io_u = NULL;
914 if (err == -EBUSY) {
915 ret = FIO_Q_BUSY;
916 goto reap;
917 }
3e260a46
JA
918 if (td->o.latency_target)
919 goto reap;
2e1df07d 920 break;
3e260a46 921 }
2e1df07d
JA
922
923 ddir = io_u->ddir;
924
925 /*
82af2a7c
JA
926 * Add verification end_io handler if:
927 * - Asked to verify (!td_rw(td))
928 * - Or the io_u is from our verify list (mixed write/ver)
2e1df07d
JA
929 */
930 if (td->o.verify != VERIFY_NONE && io_u->ddir == DDIR_READ &&
82af2a7c 931 ((io_u->flags & IO_U_F_VER_LIST) || !td_rw(td))) {
c4b6117b
PV
932
933 if (!td->o.verify_pattern_bytes) {
d6b72507 934 io_u->rand_seed = __rand(&td->verify_state);
c4b6117b 935 if (sizeof(int) != sizeof(long *))
d6b72507 936 io_u->rand_seed *= __rand(&td->verify_state);
c4b6117b
PV
937 }
938
ca09be4b
JA
939 if (verify_state_should_stop(td, io_u)) {
940 put_io_u(td, io_u);
941 break;
942 }
943
2e1df07d
JA
944 if (td->o.verify_async)
945 io_u->end_io = verify_io_u_async;
946 else
947 io_u->end_io = verify_io_u;
948 td_set_runstate(td, TD_VERIFYING);
949 } else if (in_ramp_time(td))
950 td_set_runstate(td, TD_RAMP);
951 else
952 td_set_runstate(td, TD_RUNNING);
953
9a50c5c5 954 /*
f9401285
JA
955 * Always log IO before it's issued, so we know the specific
956 * order of it. The logged unit will track when the IO has
957 * completed.
9a50c5c5 958 */
c4b6117b
PV
959 if (td_write(td) && io_u->ddir == DDIR_WRITE &&
960 td->o.do_verify &&
961 td->o.verify != VERIFY_NONE &&
f9401285 962 !td->o.experimental_verify)
c4b6117b
PV
963 log_io_piece(td, io_u);
964
a9da8ab2 965 if (td->o.io_submit_mode == IO_MODE_OFFLOAD) {
0c5df5f9
JA
966 const unsigned long blen = io_u->xfer_buflen;
967 const enum fio_ddir ddir = acct_ddir(io_u);
968
a9da8ab2
JA
969 if (td->error)
970 break;
0c5df5f9 971
26de50cf
JA
972 workqueue_enqueue(&td->io_wq, &io_u->work);
973 ret = FIO_Q_QUEUED;
50a8ce86 974
26de50cf 975 if (ddir_rw(ddir)) {
0c5df5f9
JA
976 td->io_issues[ddir]++;
977 td->io_issue_bytes[ddir] += blen;
978 td->rate_io_issue_bytes[ddir] += blen;
979 }
980
50a8ce86
D
981 if (should_check_rate(td))
982 td->rate_next_io_time[ddir] = usec_for_io(td, ddir);
983
a9da8ab2
JA
984 } else {
985 ret = td_io_queue(td, io_u);
2e1df07d 986
50a8ce86
D
987 if (should_check_rate(td))
988 td->rate_next_io_time[ddir] = usec_for_io(td, ddir);
989
fd727d9d 990 if (io_queue_event(td, io_u, &ret, ddir, &bytes_issued, 0, &comp_time))
a9da8ab2 991 break;
2e1df07d 992
a9da8ab2
JA
993 /*
994 * See if we need to complete some commands. Note that
995 * we can get BUSY even without IO queued, if the
996 * system is resource starved.
997 */
3e260a46 998reap:
a9da8ab2
JA
999 full = queue_full(td) ||
1000 (ret == FIO_Q_BUSY && td->cur_depth);
82407585 1001 if (full || io_in_polling(td))
a9da8ab2
JA
1002 ret = wait_for_completions(td, &comp_time);
1003 }
2e1df07d
JA
1004 if (ret < 0)
1005 break;
55312f9f 1006 if (!ddir_rw_sum(td->bytes_done) &&
9b87f09b 1007 !td_ioengine_flagged(td, FIO_NOIO))
2e1df07d
JA
1008 continue;
1009
55312f9f
JA
1010 if (!in_ramp_time(td) && should_check_rate(td)) {
1011 if (check_min_rate(td, &comp_time)) {
f9cafb12 1012 if (exitall_on_terminate || td->o.exitall_error)
2e1df07d
JA
1013 fio_terminate_threads(td->groupid);
1014 td_verror(td, EIO, "check_min_rate");
1015 break;
1016 }
1017 }
3e260a46
JA
1018 if (!in_ramp_time(td) && td->o.latency_target)
1019 lat_target_check(td);
e155cb64 1020
2e1df07d
JA
1021 if (td->o.thinktime) {
1022 unsigned long long b;
1023
342f4be4 1024 b = ddir_rw_sum(td->io_blocks);
2e1df07d
JA
1025 if (!(b % td->o.thinktime_blocks)) {
1026 int left;
1027
002e7183
JA
1028 io_u_quiesce(td);
1029
2e1df07d
JA
1030 if (td->o.thinktime_spin)
1031 usec_spin(td->o.thinktime_spin);
1032
1033 left = td->o.thinktime - td->o.thinktime_spin;
1034 if (left)
1035 usec_sleep(td, left);
1036 }
1037 }
1038 }
1039
c97f1ad6
JA
1040 check_update_rusage(td);
1041
2e1df07d 1042 if (td->trim_entries)
4e0a8fa2 1043 log_err("fio: %lu trim entries leaked?\n", td->trim_entries);
2e1df07d
JA
1044
1045 if (td->o.fill_device && td->error == ENOSPC) {
1046 td->error = 0;
ebea2133 1047 fio_mark_td_terminate(td);
2e1df07d
JA
1048 }
1049 if (!td->error) {
1050 struct fio_file *f;
1051
a9da8ab2
JA
1052 if (td->o.io_submit_mode == IO_MODE_OFFLOAD) {
1053 workqueue_flush(&td->io_wq);
1054 i = 0;
1055 } else
1056 i = td->cur_depth;
1057
2e1df07d 1058 if (i) {
55312f9f 1059 ret = io_u_queued_complete(td, i);
2e1df07d
JA
1060 if (td->o.fill_device && td->error == ENOSPC)
1061 td->error = 0;
1062 }
1063
1064 if (should_fsync(td) && td->o.end_fsync) {
1065 td_set_runstate(td, TD_FSYNCING);
1066
1067 for_each_file(td, f, i) {
61ee0f86 1068 if (!fio_file_fsync(td, f))
2e1df07d 1069 continue;
61ee0f86
JA
1070
1071 log_err("fio: end_fsync failed for file %s\n",
1072 f->file_name);
2e1df07d
JA
1073 }
1074 }
1075 } else
1076 cleanup_pending_aio(td);
1077
1078 /*
1079 * stop job if we failed doing any IO
1080 */
342f4be4 1081 if (!ddir_rw_sum(td->this_io_bytes))
2e1df07d 1082 td->done = 1;
100f49f1 1083
95b03be0
JA
1084 for (i = 0; i < DDIR_RWDIR_CNT; i++)
1085 bytes_done[i] = td->bytes_done[i] - bytes_done[i];
2e1df07d
JA
1086}
1087
94a6e1bb
JA
1088static void free_file_completion_logging(struct thread_data *td)
1089{
1090 struct fio_file *f;
1091 unsigned int i;
1092
1093 for_each_file(td, f, i) {
1094 if (!f->last_write_comp)
1095 break;
1096 sfree(f->last_write_comp);
1097 }
1098}
1099
1100static int init_file_completion_logging(struct thread_data *td,
1101 unsigned int depth)
1102{
1103 struct fio_file *f;
1104 unsigned int i;
1105
1106 if (td->o.verify == VERIFY_NONE || !td->o.verify_state_save)
1107 return 0;
1108
1109 for_each_file(td, f, i) {
1110 f->last_write_comp = scalloc(depth, sizeof(uint64_t));
1111 if (!f->last_write_comp)
1112 goto cleanup;
1113 }
1114
1115 return 0;
1116
1117cleanup:
1118 free_file_completion_logging(td);
1119 log_err("fio: failed to alloc write comp data\n");
1120 return 1;
1121}
1122
2e1df07d
JA
1123static void cleanup_io_u(struct thread_data *td)
1124{
2e1df07d
JA
1125 struct io_u *io_u;
1126
2ae0b204 1127 while ((io_u = io_u_qpop(&td->io_u_freelist)) != NULL) {
c73ed246
JA
1128
1129 if (td->io_ops->io_u_free)
1130 td->io_ops->io_u_free(td, io_u);
1131
2e1df07d
JA
1132 fio_memfree(io_u, sizeof(*io_u));
1133 }
1134
1135 free_io_mem(td);
2ae0b204
JA
1136
1137 io_u_rexit(&td->io_u_requeues);
1138 io_u_qexit(&td->io_u_freelist);
1139 io_u_qexit(&td->io_u_all);
ca09be4b 1140
94a6e1bb 1141 free_file_completion_logging(td);
2e1df07d
JA
1142}
1143
1144static int init_io_u(struct thread_data *td)
1145{
1146 struct io_u *io_u;
9c42684e 1147 unsigned int max_bs, min_write;
2e1df07d 1148 int cl_align, i, max_units;
2ae0b204 1149 int data_xfer = 1, err;
2e1df07d
JA
1150 char *p;
1151
1152 max_units = td->o.iodepth;
74f4b020 1153 max_bs = td_max_bs(td);
9c42684e 1154 min_write = td->o.min_bs[DDIR_WRITE];
2e1df07d
JA
1155 td->orig_buffer_size = (unsigned long long) max_bs
1156 * (unsigned long long) max_units;
1157
9b87f09b 1158 if (td_ioengine_flagged(td, FIO_NOIO) || !(td_read(td) || td_write(td)))
59d8d0f5
JA
1159 data_xfer = 0;
1160
2ae0b204
JA
1161 err = 0;
1162 err += io_u_rinit(&td->io_u_requeues, td->o.iodepth);
1163 err += io_u_qinit(&td->io_u_freelist, td->o.iodepth);
1164 err += io_u_qinit(&td->io_u_all, td->o.iodepth);
1165
1166 if (err) {
1167 log_err("fio: failed setting up IO queues\n");
1168 return 1;
1169 }
1170
fd8a09b8 1171 /*
1172 * if we may later need to do address alignment, then add any
1173 * possible adjustment here so that we don't cause a buffer
1174 * overflow later. this adjustment may be too much if we get
1175 * lucky and the allocator gives us an aligned address.
1176 */
d01612f3 1177 if (td->o.odirect || td->o.mem_align || td->o.oatomic ||
9b87f09b 1178 td_ioengine_flagged(td, FIO_RAWIO))
fd8a09b8 1179 td->orig_buffer_size += page_mask + td->o.mem_align;
1180
2e1df07d
JA
1181 if (td->o.mem_type == MEM_SHMHUGE || td->o.mem_type == MEM_MMAPHUGE) {
1182 unsigned long bs;
1183
1184 bs = td->orig_buffer_size + td->o.hugepage_size - 1;
1185 td->orig_buffer_size = bs & ~(td->o.hugepage_size - 1);
1186 }
1187
1188 if (td->orig_buffer_size != (size_t) td->orig_buffer_size) {
1189 log_err("fio: IO memory too large. Reduce max_bs or iodepth\n");
1190 return 1;
1191 }
1192
59d8d0f5 1193 if (data_xfer && allocate_io_mem(td))
2e1df07d
JA
1194 return 1;
1195
d01612f3 1196 if (td->o.odirect || td->o.mem_align || td->o.oatomic ||
9b87f09b 1197 td_ioengine_flagged(td, FIO_RAWIO))
248c9436 1198 p = PTR_ALIGN(td->orig_buffer, page_mask) + td->o.mem_align;
2e1df07d
JA
1199 else
1200 p = td->orig_buffer;
1201
1202 cl_align = os_cache_line_size();
1203
1204 for (i = 0; i < max_units; i++) {
1205 void *ptr;
1206
1207 if (td->terminate)
1208 return 1;
1209
1210 ptr = fio_memalign(cl_align, sizeof(*io_u));
1211 if (!ptr) {
1212 log_err("fio: unable to allocate aligned memory\n");
1213 break;
1214 }
1215
1216 io_u = ptr;
1217 memset(io_u, 0, sizeof(*io_u));
2ae0b204 1218 INIT_FLIST_HEAD(&io_u->verify_list);
2e1df07d
JA
1219 dprint(FD_MEM, "io_u alloc %p, index %u\n", io_u, i);
1220
59d8d0f5 1221 if (data_xfer) {
2e1df07d
JA
1222 io_u->buf = p;
1223 dprint(FD_MEM, "io_u %p, mem %p\n", io_u, io_u->buf);
1224
1225 if (td_write(td))
9c42684e 1226 io_u_fill_buffer(td, io_u, min_write, max_bs);
2e1df07d
JA
1227 if (td_write(td) && td->o.verify_pattern_bytes) {
1228 /*
1229 * Fill the buffer with the pattern if we are
1230 * going to be doing writes.
1231 */
ce35b1ec 1232 fill_verify_pattern(td, io_u->buf, max_bs, io_u, 0, 0);
2e1df07d
JA
1233 }
1234 }
1235
1236 io_u->index = i;
1237 io_u->flags = IO_U_F_FREE;
2ae0b204
JA
1238 io_u_qpush(&td->io_u_freelist, io_u);
1239
1240 /*
1241 * io_u never leaves this stack, used for iteration of all
1242 * io_u buffers.
1243 */
1244 io_u_qpush(&td->io_u_all, io_u);
c73ed246
JA
1245
1246 if (td->io_ops->io_u_init) {
1247 int ret = td->io_ops->io_u_init(td, io_u);
1248
1249 if (ret) {
1250 log_err("fio: failed to init engine data: %d\n", ret);
1251 return 1;
1252 }
1253 }
1254
2e1df07d
JA
1255 p += max_bs;
1256 }
1257
94a6e1bb
JA
1258 if (init_file_completion_logging(td, max_units))
1259 return 1;
ca09be4b 1260
2e1df07d
JA
1261 return 0;
1262}
1263
bd4d9bdc
TK
1264/*
1265 * This function is Linux specific.
1266 * FIO_HAVE_IOSCHED_SWITCH enabled currently means it's Linux.
1267 */
2e1df07d
JA
1268static int switch_ioscheduler(struct thread_data *td)
1269{
db815fb3 1270#ifdef FIO_HAVE_IOSCHED_SWITCH
2e1df07d
JA
1271 char tmp[256], tmp2[128];
1272 FILE *f;
1273 int ret;
1274
9b87f09b 1275 if (td_ioengine_flagged(td, FIO_DISKLESSIO))
2e1df07d
JA
1276 return 0;
1277
bd4d9bdc
TK
1278 assert(td->files && td->files[0]);
1279 sprintf(tmp, "%s/queue/scheduler", td->files[0]->du->sysfs_root);
2e1df07d
JA
1280
1281 f = fopen(tmp, "r+");
1282 if (!f) {
1283 if (errno == ENOENT) {
1284 log_err("fio: os or kernel doesn't support IO scheduler"
1285 " switching\n");
1286 return 0;
1287 }
1288 td_verror(td, errno, "fopen iosched");
1289 return 1;
1290 }
1291
1292 /*
1293 * Set io scheduler.
1294 */
1295 ret = fwrite(td->o.ioscheduler, strlen(td->o.ioscheduler), 1, f);
1296 if (ferror(f) || ret != 1) {
1297 td_verror(td, errno, "fwrite");
1298 fclose(f);
1299 return 1;
1300 }
1301
1302 rewind(f);
1303
1304 /*
1305 * Read back and check that the selected scheduler is now the default.
1306 */
b44b9e45 1307 memset(tmp, 0, sizeof(tmp));
49c6f33d 1308 ret = fread(tmp, sizeof(tmp), 1, f);
2e1df07d
JA
1309 if (ferror(f) || ret < 0) {
1310 td_verror(td, errno, "fread");
1311 fclose(f);
1312 return 1;
1313 }
b44b9e45
TK
1314 /*
1315 * either a list of io schedulers or "none\n" is expected.
1316 */
1317 tmp[strlen(tmp) - 1] = '\0';
49c6f33d 1318
67f28a8c
TK
1319 /*
1320 * Write to "none" entry doesn't fail, so check the result here.
1321 */
1322 if (!strcmp(tmp, "none")) {
1323 log_err("fio: io scheduler is not tunable\n");
1324 fclose(f);
1325 return 0;
1326 }
2e1df07d
JA
1327
1328 sprintf(tmp2, "[%s]", td->o.ioscheduler);
1329 if (!strstr(tmp, tmp2)) {
1330 log_err("fio: io scheduler %s not found\n", td->o.ioscheduler);
1331 td_verror(td, EINVAL, "iosched_switch");
1332 fclose(f);
1333 return 1;
1334 }
1335
1336 fclose(f);
1337 return 0;
db815fb3
TK
1338#else
1339 return 0;
1340#endif
2e1df07d
JA
1341}
1342
e39c0676 1343static bool keep_running(struct thread_data *td)
2e1df07d 1344{
77731b29
JA
1345 unsigned long long limit;
1346
2e1df07d 1347 if (td->done)
e39c0676 1348 return false;
2e1df07d 1349 if (td->o.time_based)
e39c0676 1350 return true;
2e1df07d
JA
1351 if (td->o.loops) {
1352 td->o.loops--;
e39c0676 1353 return true;
2e1df07d 1354 }
3939fe85 1355 if (exceeds_number_ios(td))
e39c0676 1356 return false;
26251d8d 1357
5be9bf09
TK
1358 if (td->o.io_size)
1359 limit = td->o.io_size;
77731b29
JA
1360 else
1361 limit = td->o.size;
1362
1363 if (limit != -1ULL && ddir_rw_sum(td->io_bytes) < limit) {
5bd5f71a
JA
1364 uint64_t diff;
1365
1366 /*
589d5d4f 1367 * If the difference is less than the maximum IO size, we
5bd5f71a
JA
1368 * are done.
1369 */
77731b29 1370 diff = limit - ddir_rw_sum(td->io_bytes);
74f4b020 1371 if (diff < td_max_bs(td))
e39c0676 1372 return false;
5bd5f71a 1373
5be9bf09 1374 if (fio_files_done(td) && !td->o.io_size)
e39c0676 1375 return false;
002fe734 1376
e39c0676 1377 return true;
5bd5f71a 1378 }
2e1df07d 1379
e39c0676 1380 return false;
2e1df07d
JA
1381}
1382
ce486495 1383static int exec_string(struct thread_options *o, const char *string, const char *mode)
2e1df07d 1384{
6c9ce469 1385 size_t newlen = strlen(string) + strlen(o->name) + strlen(mode) + 9 + 1;
f491a907 1386 int ret;
2e1df07d
JA
1387 char *str;
1388
1389 str = malloc(newlen);
ce486495 1390 sprintf(str, "%s &> %s.%s.txt", string, o->name, mode);
2e1df07d 1391
ce486495 1392 log_info("%s : Saving output of %s in %s.%s.txt\n",o->name, mode, o->name, mode);
2e1df07d
JA
1393 ret = system(str);
1394 if (ret == -1)
1395 log_err("fio: exec of cmd <%s> failed\n", str);
1396
1397 free(str);
1398 return ret;
1399}
1400
62167762
JC
1401/*
1402 * Dry run to compute correct state of numberio for verification.
1403 */
1404static uint64_t do_dry_run(struct thread_data *td)
1405{
62167762
JC
1406 td_set_runstate(td, TD_RUNNING);
1407
1408 while ((td->o.read_iolog_file && !flist_empty(&td->io_log_list)) ||
e28dd2cf 1409 (!flist_empty(&td->trim_list)) || !io_complete_bytes_exceeded(td)) {
62167762
JC
1410 struct io_u *io_u;
1411 int ret;
1412
1413 if (td->terminate || td->done)
1414 break;
1415
1416 io_u = get_io_u(td);
99955d3d 1417 if (IS_ERR_OR_NULL(io_u))
62167762
JC
1418 break;
1419
1651e431 1420 io_u_set(td, io_u, IO_U_F_FLIGHT);
62167762
JC
1421 io_u->error = 0;
1422 io_u->resid = 0;
1423 if (ddir_rw(acct_ddir(io_u)))
1424 td->io_issues[acct_ddir(io_u)]++;
1425 if (ddir_rw(io_u->ddir)) {
1426 io_u_mark_depth(td, 1);
1427 td->ts.total_io_u[io_u->ddir]++;
1428 }
1429
2e63e96b
PV
1430 if (td_write(td) && io_u->ddir == DDIR_WRITE &&
1431 td->o.do_verify &&
1432 td->o.verify != VERIFY_NONE &&
1433 !td->o.experimental_verify)
1434 log_io_piece(td, io_u);
1435
55312f9f 1436 ret = io_u_sync_complete(td, io_u);
62167762
JC
1437 (void) ret;
1438 }
1439
55312f9f 1440 return td->bytes_done[DDIR_WRITE] + td->bytes_done[DDIR_TRIM];
62167762
JC
1441}
1442
24660963
JA
1443struct fork_data {
1444 struct thread_data *td;
1445 struct sk_out *sk_out;
1446};
1447
2e1df07d
JA
1448/*
1449 * Entry point for the thread based jobs. The process based jobs end up
1450 * here as well, after a little setup.
1451 */
1452static void *thread_main(void *data)
1453{
24660963 1454 struct fork_data *fd = data;
95603b74 1455 unsigned long long elapsed_us[DDIR_RWDIR_CNT] = { 0, };
24660963 1456 struct thread_data *td = fd->td;
4896473e 1457 struct thread_options *o = &td->o;
24660963 1458 struct sk_out *sk_out = fd->sk_out;
ca0122d8 1459 int deadlock_loop_cnt;
2e1df07d 1460 int clear_state;
28727df7 1461 int ret;
2e1df07d 1462
24660963
JA
1463 sk_out_assign(sk_out);
1464 free(fd);
1465
4896473e 1466 if (!o->use_thread) {
2e1df07d
JA
1467 setsid();
1468 td->pid = getpid();
1469 } else
1470 td->pid = gettid();
1471
4896473e 1472 fio_local_clock_init(o->use_thread);
5d879392 1473
2e1df07d
JA
1474 dprint(FD_PROCESS, "jobs pid=%d started\n", (int) td->pid);
1475
122c7725
JA
1476 if (is_backend)
1477 fio_server_send_start(td);
1478
2e1df07d
JA
1479 INIT_FLIST_HEAD(&td->io_log_list);
1480 INIT_FLIST_HEAD(&td->io_hist_list);
1481 INIT_FLIST_HEAD(&td->verify_list);
1482 INIT_FLIST_HEAD(&td->trim_list);
1ae83d45 1483 INIT_FLIST_HEAD(&td->next_rand_list);
2e1df07d
JA
1484 td->io_hist_tree = RB_ROOT;
1485
34febb23 1486 ret = mutex_cond_init_pshared(&td->io_u_lock, &td->free_cond);
f9e5b5ee 1487 if (ret) {
34febb23 1488 td_verror(td, ret, "mutex_cond_init_pshared");
f9e5b5ee
JK
1489 goto err;
1490 }
34febb23 1491 ret = cond_init_pshared(&td->verify_cond);
f9e5b5ee 1492 if (ret) {
34febb23 1493 td_verror(td, ret, "mutex_cond_pshared");
f9e5b5ee
JK
1494 goto err;
1495 }
2e1df07d
JA
1496
1497 td_set_runstate(td, TD_INITIALIZED);
1498 dprint(FD_MUTEX, "up startup_mutex\n");
1499 fio_mutex_up(startup_mutex);
1500 dprint(FD_MUTEX, "wait on td->mutex\n");
1501 fio_mutex_down(td->mutex);
1502 dprint(FD_MUTEX, "done waiting on td->mutex\n");
1503
2e1df07d
JA
1504 /*
1505 * A new gid requires privilege, so we need to do this before setting
1506 * the uid.
1507 */
4896473e 1508 if (o->gid != -1U && setgid(o->gid)) {
2e1df07d
JA
1509 td_verror(td, errno, "setgid");
1510 goto err;
1511 }
4896473e 1512 if (o->uid != -1U && setuid(o->uid)) {
2e1df07d
JA
1513 td_verror(td, errno, "setuid");
1514 goto err;
1515 }
1516
2354d810
JA
1517 /*
1518 * Do this early, we don't want the compress threads to be limited
1519 * to the same CPUs as the IO workers. So do this before we set
1520 * any potential CPU affinity
1521 */
1522 if (iolog_compress_init(td, sk_out))
1523 goto err;
1524
2e1df07d
JA
1525 /*
1526 * If we have a gettimeofday() thread, make sure we exclude that
1527 * thread from this job
1528 */
4896473e
JA
1529 if (o->gtod_cpu)
1530 fio_cpu_clear(&o->cpumask, o->gtod_cpu);
2e1df07d
JA
1531
1532 /*
1533 * Set affinity first, in case it has an impact on the memory
1534 * allocations.
1535 */
b2a9e649 1536 if (fio_option_is_set(o, cpumask)) {
c2acfbac 1537 if (o->cpus_allowed_policy == FIO_CPUS_SPLIT) {
30cb4c65 1538 ret = fio_cpus_split(&o->cpumask, td->thread_number - 1);
c2acfbac
JA
1539 if (!ret) {
1540 log_err("fio: no CPUs set\n");
1541 log_err("fio: Try increasing number of available CPUs\n");
1542 td_verror(td, EINVAL, "cpus_split");
1543 goto err;
1544 }
1545 }
28727df7
JA
1546 ret = fio_setaffinity(td->pid, o->cpumask);
1547 if (ret == -1) {
4896473e
JA
1548 td_verror(td, errno, "cpu_set_affinity");
1549 goto err;
1550 }
2e1df07d
JA
1551 }
1552
67bf9823 1553#ifdef CONFIG_LIBNUMA
d0b937ed 1554 /* numa node setup */
b2a9e649
JA
1555 if (fio_option_is_set(o, numa_cpunodes) ||
1556 fio_option_is_set(o, numa_memnodes)) {
43522848 1557 struct bitmask *mask;
d0b937ed
YR
1558
1559 if (numa_available() < 0) {
1560 td_verror(td, errno, "Does not support NUMA API\n");
1561 goto err;
1562 }
1563
b2a9e649 1564 if (fio_option_is_set(o, numa_cpunodes)) {
43522848
DG
1565 mask = numa_parse_nodestring(o->numa_cpunodes);
1566 ret = numa_run_on_node_mask(mask);
1567 numa_free_nodemask(mask);
d0b937ed
YR
1568 if (ret == -1) {
1569 td_verror(td, errno, \
1570 "numa_run_on_node_mask failed\n");
1571 goto err;
1572 }
1573 }
1574
b2a9e649 1575 if (fio_option_is_set(o, numa_memnodes)) {
43522848
DG
1576 mask = NULL;
1577 if (o->numa_memnodes)
1578 mask = numa_parse_nodestring(o->numa_memnodes);
1579
4896473e 1580 switch (o->numa_mem_mode) {
d0b937ed 1581 case MPOL_INTERLEAVE:
43522848 1582 numa_set_interleave_mask(mask);
d0b937ed
YR
1583 break;
1584 case MPOL_BIND:
43522848 1585 numa_set_membind(mask);
d0b937ed
YR
1586 break;
1587 case MPOL_LOCAL:
1588 numa_set_localalloc();
1589 break;
1590 case MPOL_PREFERRED:
4896473e 1591 numa_set_preferred(o->numa_mem_prefer_node);
d0b937ed
YR
1592 break;
1593 case MPOL_DEFAULT:
1594 default:
1595 break;
1596 }
1597
43522848
DG
1598 if (mask)
1599 numa_free_nodemask(mask);
1600
d0b937ed
YR
1601 }
1602 }
1603#endif
1604
9a3f1100
JA
1605 if (fio_pin_memory(td))
1606 goto err;
1607
2e1df07d
JA
1608 /*
1609 * May alter parameters that init_io_u() will use, so we need to
1610 * do this first.
1611 */
1612 if (init_iolog(td))
1613 goto err;
1614
1615 if (init_io_u(td))
1616 goto err;
1617
4896473e 1618 if (o->verify_async && verify_async_init(td))
2e1df07d
JA
1619 goto err;
1620
27d74836
JA
1621 if (fio_option_is_set(o, ioprio) ||
1622 fio_option_is_set(o, ioprio_class)) {
28727df7
JA
1623 ret = ioprio_set(IOPRIO_WHO_PROCESS, 0, o->ioprio_class, o->ioprio);
1624 if (ret == -1) {
2e1df07d
JA
1625 td_verror(td, errno, "ioprio_set");
1626 goto err;
1627 }
1628 }
1629
4896473e 1630 if (o->cgroup && cgroup_setup(td, cgroup_list, &cgroup_mnt))
2e1df07d
JA
1631 goto err;
1632
649c10c5 1633 errno = 0;
4896473e 1634 if (nice(o->nice) == -1 && errno != 0) {
2e1df07d
JA
1635 td_verror(td, errno, "nice");
1636 goto err;
1637 }
1638
4896473e 1639 if (o->ioscheduler && switch_ioscheduler(td))
2e1df07d
JA
1640 goto err;
1641
4896473e 1642 if (!o->create_serialize && setup_files(td))
2e1df07d
JA
1643 goto err;
1644
1645 if (td_io_init(td))
1646 goto err;
1647
1648 if (init_random_map(td))
1649 goto err;
1650
ce486495 1651 if (o->exec_prerun && exec_string(o, o->exec_prerun, (const char *)"prerun"))
4896473e 1652 goto err;
2e1df07d 1653
4896473e 1654 if (o->pre_read) {
2e1df07d
JA
1655 if (pre_read_files(td) < 0)
1656 goto err;
1657 }
1658
dc5bfbb2
JA
1659 fio_verify_init(td);
1660
24660963 1661 if (rate_submit_init(td, sk_out))
a9da8ab2
JA
1662 goto err;
1663
3aea75b1 1664 set_epoch_time(td, o->log_unix_epoch);
44404c5a 1665 fio_getrusage(&td->ru_start);
ac28d905
JA
1666 memcpy(&td->bw_sample_time, &td->epoch, sizeof(td->epoch));
1667 memcpy(&td->iops_sample_time, &td->epoch, sizeof(td->epoch));
16e56d25 1668 memcpy(&td->ss.prev_time, &td->epoch, sizeof(td->epoch));
ac28d905
JA
1669
1670 if (o->ratemin[DDIR_READ] || o->ratemin[DDIR_WRITE] ||
1671 o->ratemin[DDIR_TRIM]) {
1672 memcpy(&td->lastrate[DDIR_READ], &td->bw_sample_time,
1673 sizeof(td->bw_sample_time));
1674 memcpy(&td->lastrate[DDIR_WRITE], &td->bw_sample_time,
1675 sizeof(td->bw_sample_time));
1676 memcpy(&td->lastrate[DDIR_TRIM], &td->bw_sample_time,
1677 sizeof(td->bw_sample_time));
1678 }
1679
2e1df07d
JA
1680 clear_state = 0;
1681 while (keep_running(td)) {
100f49f1
JA
1682 uint64_t verify_bytes;
1683
2e1df07d 1684 fio_gettime(&td->start, NULL);
2e1df07d
JA
1685 memcpy(&td->tv_cache, &td->start, sizeof(td->start));
1686
39c1c323 1687 if (clear_state) {
ac28d905 1688 clear_io_state(td, 0);
2e1df07d 1689
39c1c323 1690 if (o->unlink_each_loop && unlink_all_files(td))
1691 break;
1692 }
1693
2e1df07d
JA
1694 prune_io_piece_log(td);
1695
62167762
JC
1696 if (td->o.verify_only && (td_write(td) || td_rw(td)))
1697 verify_bytes = do_dry_run(td);
095196b1 1698 else {
95b03be0
JA
1699 uint64_t bytes_done[DDIR_RWDIR_CNT];
1700
1701 do_io(td, bytes_done);
1702
1703 if (!ddir_rw_sum(bytes_done)) {
095196b1 1704 fio_mark_td_terminate(td);
95b03be0
JA
1705 verify_bytes = 0;
1706 } else {
1707 verify_bytes = bytes_done[DDIR_WRITE] +
1708 bytes_done[DDIR_TRIM];
1709 }
095196b1 1710 }
2e1df07d 1711
49f0b8e0
JA
1712 /*
1713 * If we took too long to shut down, the main thread could
1714 * already consider us reaped/exited. If that happens, break
1715 * out and clean up.
1716 */
1717 if (td->runstate >= TD_EXITED)
1718 break;
1719
2e1df07d
JA
1720 clear_state = 1;
1721
40c666c8
JA
1722 /*
1723 * Make sure we've successfully updated the rusage stats
1724 * before waiting on the stat mutex. Otherwise we could have
1725 * the stat thread holding stat mutex and waiting for
1726 * the rusage_sem, which would never get upped because
1727 * this thread is waiting for the stat mutex.
1728 */
ca0122d8 1729 deadlock_loop_cnt = 0;
39d13e67
JA
1730 do {
1731 check_update_rusage(td);
1732 if (!fio_mutex_down_trylock(stat_mutex))
1733 break;
1734 usleep(1000);
ca0122d8
TT
1735 if (deadlock_loop_cnt++ > 5000) {
1736 log_err("fio seems to be stuck grabbing stat_mutex, forcibly exiting\n");
43f248c4 1737 td->error = EDEADLK;
ca0122d8
TT
1738 goto err;
1739 }
39d13e67 1740 } while (1);
40c666c8 1741
95603b74
BF
1742 if (td_read(td) && td->io_bytes[DDIR_READ])
1743 update_runtime(td, elapsed_us, DDIR_READ);
1744 if (td_write(td) && td->io_bytes[DDIR_WRITE])
1745 update_runtime(td, elapsed_us, DDIR_WRITE);
1746 if (td_trim(td) && td->io_bytes[DDIR_TRIM])
1747 update_runtime(td, elapsed_us, DDIR_TRIM);
e5437a07
VT
1748 fio_gettime(&td->start, NULL);
1749 fio_mutex_up(stat_mutex);
2e1df07d
JA
1750
1751 if (td->error || td->terminate)
1752 break;
1753
4896473e
JA
1754 if (!o->do_verify ||
1755 o->verify == VERIFY_NONE ||
9b87f09b 1756 td_ioengine_flagged(td, FIO_UNIDIR))
2e1df07d
JA
1757 continue;
1758
ac28d905 1759 clear_io_state(td, 0);
2e1df07d
JA
1760
1761 fio_gettime(&td->start, NULL);
1762
100f49f1 1763 do_verify(td, verify_bytes);
2e1df07d 1764
40c666c8
JA
1765 /*
1766 * See comment further up for why this is done here.
1767 */
1768 check_update_rusage(td);
1769
e5437a07 1770 fio_mutex_down(stat_mutex);
95603b74 1771 update_runtime(td, elapsed_us, DDIR_READ);
e5437a07
VT
1772 fio_gettime(&td->start, NULL);
1773 fio_mutex_up(stat_mutex);
2e1df07d
JA
1774
1775 if (td->error || td->terminate)
1776 break;
1777 }
1778
2cea0b4c
JA
1779 td_set_runstate(td, TD_FINISHING);
1780
2e1df07d 1781 update_rusage_stat(td);
2e1df07d 1782 td->ts.total_run_time = mtime_since_now(&td->epoch);
6eaf09d6
SL
1783 td->ts.io_bytes[DDIR_READ] = td->io_bytes[DDIR_READ];
1784 td->ts.io_bytes[DDIR_WRITE] = td->io_bytes[DDIR_WRITE];
1785 td->ts.io_bytes[DDIR_TRIM] = td->io_bytes[DDIR_TRIM];
2e1df07d 1786
ca09be4b 1787 if (td->o.verify_state_save && !(td->flags & TD_F_VSTATE_SAVED) &&
d264264a
JA
1788 (td->o.verify != VERIFY_NONE && td_write(td)))
1789 verify_save_state(td->thread_number);
ca09be4b 1790
9a3f1100
JA
1791 fio_unpin_memory(td);
1792
a47591e4 1793 td_writeout_logs(td, true);
2e1df07d 1794
155f2f02 1795 iolog_compress_exit(td);
103b174e 1796 rate_submit_exit(td);
aee2ab67 1797
4896473e 1798 if (o->exec_postrun)
ce486495 1799 exec_string(o, o->exec_postrun, (const char *)"postrun");
2e1df07d 1800
f9cafb12 1801 if (exitall_on_terminate || (o->exitall_error && td->error))
2e1df07d
JA
1802 fio_terminate_threads(td->groupid);
1803
1804err:
1805 if (td->error)
1806 log_info("fio: pid=%d, err=%d/%s\n", (int) td->pid, td->error,
1807 td->verror);
1808
4896473e 1809 if (o->verify_async)
2e1df07d
JA
1810 verify_async_exit(td);
1811
1812 close_and_free_files(td);
2e1df07d 1813 cleanup_io_u(td);
32dbca2c 1814 close_ioengine(td);
2e1df07d 1815 cgroup_shutdown(td, &cgroup_mnt);
ca09be4b 1816 verify_free_state(td);
2e1df07d 1817
7c4d0fb7
JA
1818 if (td->zone_state_index) {
1819 int i;
1820
1821 for (i = 0; i < DDIR_RWDIR_CNT; i++)
1822 free(td->zone_state_index[i]);
1823 free(td->zone_state_index);
1824 td->zone_state_index = NULL;
1825 }
1826
b2a9e649 1827 if (fio_option_is_set(o, cpumask)) {
8a1db9a1
JA
1828 ret = fio_cpuset_exit(&o->cpumask);
1829 if (ret)
1830 td_verror(td, ret, "fio_cpuset_exit");
2e1df07d
JA
1831 }
1832
1833 /*
1834 * do this very late, it will log file closing as well
1835 */
4896473e 1836 if (o->write_iolog_file)
2e1df07d
JA
1837 write_iolog_close(td);
1838
ea66e04f
JA
1839 fio_mutex_remove(td->mutex);
1840 td->mutex = NULL;
1841
2e1df07d 1842 td_set_runstate(td, TD_EXITED);
fda2cfac
JA
1843
1844 /*
1845 * Do this last after setting our runstate to exited, so we
1846 * know that the stat thread is signaled.
1847 */
1848 check_update_rusage(td);
1849
24660963 1850 sk_out_drop();
e43606c2 1851 return (void *) (uintptr_t) td->error;
2e1df07d
JA
1852}
1853
cba5460c
JA
1854static void dump_td_info(struct thread_data *td)
1855{
a507505d
JA
1856 log_err("fio: job '%s' (state=%d) hasn't exited in %lu seconds, it "
1857 "appears to be stuck. Doing forceful exit of this job.\n",
1858 td->o.name, td->runstate,
cba5460c
JA
1859 (unsigned long) time_since_now(&td->terminate_time));
1860}
1861
2e1df07d
JA
1862/*
1863 * Run over the job map and reap the threads that have exited, if any.
1864 */
90eff1c9
SW
1865static void reap_threads(unsigned int *nr_running, uint64_t *t_rate,
1866 uint64_t *m_rate)
2e1df07d
JA
1867{
1868 struct thread_data *td;
1869 unsigned int cputhreads, realthreads, pending;
1870 int i, status, ret;
1871
1872 /*
1873 * reap exited threads (TD_EXITED -> TD_REAPED)
1874 */
1875 realthreads = pending = cputhreads = 0;
1876 for_each_td(td, i) {
1877 int flags = 0;
1878
1879 /*
1880 * ->io_ops is NULL for a thread that has closed its
1881 * io engine
1882 */
1883 if (td->io_ops && !strcmp(td->io_ops->name, "cpuio"))
1884 cputhreads++;
1885 else
1886 realthreads++;
1887
1888 if (!td->pid) {
1889 pending++;
1890 continue;
1891 }
1892 if (td->runstate == TD_REAPED)
1893 continue;
1894 if (td->o.use_thread) {
1895 if (td->runstate == TD_EXITED) {
1896 td_set_runstate(td, TD_REAPED);
1897 goto reaped;
1898 }
1899 continue;
1900 }
1901
1902 flags = WNOHANG;
1903 if (td->runstate == TD_EXITED)
1904 flags = 0;
1905
1906 /*
1907 * check if someone quit or got killed in an unusual way
1908 */
1909 ret = waitpid(td->pid, &status, flags);
1910 if (ret < 0) {
1911 if (errno == ECHILD) {
1912 log_err("fio: pid=%d disappeared %d\n",
1913 (int) td->pid, td->runstate);
a5e371a6 1914 td->sig = ECHILD;
2e1df07d
JA
1915 td_set_runstate(td, TD_REAPED);
1916 goto reaped;
1917 }
1918 perror("waitpid");
1919 } else if (ret == td->pid) {
1920 if (WIFSIGNALED(status)) {
1921 int sig = WTERMSIG(status);
1922
36d80bc7 1923 if (sig != SIGTERM && sig != SIGUSR2)
2e1df07d
JA
1924 log_err("fio: pid=%d, got signal=%d\n",
1925 (int) td->pid, sig);
a5e371a6 1926 td->sig = sig;
2e1df07d
JA
1927 td_set_runstate(td, TD_REAPED);
1928 goto reaped;
1929 }
1930 if (WIFEXITED(status)) {
1931 if (WEXITSTATUS(status) && !td->error)
1932 td->error = WEXITSTATUS(status);
1933
1934 td_set_runstate(td, TD_REAPED);
1935 goto reaped;
1936 }
1937 }
1938
cba5460c
JA
1939 /*
1940 * If the job is stuck, do a forceful timeout of it and
1941 * move on.
1942 */
1943 if (td->terminate &&
2c45a4ac 1944 td->runstate < TD_FSYNCING &&
cba5460c
JA
1945 time_since_now(&td->terminate_time) >= FIO_REAP_TIMEOUT) {
1946 dump_td_info(td);
1947 td_set_runstate(td, TD_REAPED);
1948 goto reaped;
1949 }
1950
2e1df07d
JA
1951 /*
1952 * thread is not dead, continue
1953 */
1954 pending++;
1955 continue;
1956reaped:
1957 (*nr_running)--;
342f4be4
JA
1958 (*m_rate) -= ddir_rw_sum(td->o.ratemin);
1959 (*t_rate) -= ddir_rw_sum(td->o.rate);
2e1df07d
JA
1960 if (!td->pid)
1961 pending--;
1962
1963 if (td->error)
1964 exit_value++;
1965
1966 done_secs += mtime_since_now(&td->epoch) / 1000;
4a88752a 1967 profile_td_exit(td);
2e1df07d
JA
1968 }
1969
1970 if (*nr_running == cputhreads && !pending && realthreads)
1971 fio_terminate_threads(TERMINATE_ALL);
1972}
1973
e39c0676 1974static bool __check_trigger_file(void)
ca09be4b
JA
1975{
1976 struct stat sb;
1977
1978 if (!trigger_file)
e39c0676 1979 return false;
ca09be4b
JA
1980
1981 if (stat(trigger_file, &sb))
e39c0676 1982 return false;
ca09be4b
JA
1983
1984 if (unlink(trigger_file) < 0)
1985 log_err("fio: failed to unlink %s: %s\n", trigger_file,
1986 strerror(errno));
1987
e39c0676 1988 return true;
ca09be4b
JA
1989}
1990
e39c0676 1991static bool trigger_timedout(void)
ca09be4b
JA
1992{
1993 if (trigger_timeout)
1994 return time_since_genesis() >= trigger_timeout;
1995
e39c0676 1996 return false;
ca09be4b
JA
1997}
1998
1999void exec_trigger(const char *cmd)
2000{
2001 int ret;
2002
2003 if (!cmd)
2004 return;
2005
2006 ret = system(cmd);
2007 if (ret == -1)
2008 log_err("fio: failed executing %s trigger\n", cmd);
2009}
2010
2011void check_trigger_file(void)
2012{
2013 if (__check_trigger_file() || trigger_timedout()) {
796fb3ce
JA
2014 if (nr_clients)
2015 fio_clients_send_trigger(trigger_remote_cmd);
2016 else {
d264264a 2017 verify_save_state(IO_LIST_ALL);
ca09be4b
JA
2018 fio_terminate_threads(TERMINATE_ALL);
2019 exec_trigger(trigger_cmd);
2020 }
2021 }
2022}
2023
2024static int fio_verify_load_state(struct thread_data *td)
2025{
2026 int ret;
2027
2028 if (!td->o.verify_state)
2029 return 0;
2030
2031 if (is_backend) {
2032 void *data;
2033
2034 ret = fio_server_get_verify_state(td->o.name,
94a6e1bb 2035 td->thread_number - 1, &data);
ca09be4b 2036 if (!ret)
94a6e1bb 2037 verify_assign_state(td, data);
ca09be4b
JA
2038 } else
2039 ret = verify_load_state(td, "local");
2040
2041 return ret;
2042}
2043
06464907
JA
2044static void do_usleep(unsigned int usecs)
2045{
2046 check_for_running_stats();
ca09be4b 2047 check_trigger_file();
06464907
JA
2048 usleep(usecs);
2049}
2050
e39c0676 2051static bool check_mount_writes(struct thread_data *td)
e81ecca3
JA
2052{
2053 struct fio_file *f;
2054 unsigned int i;
2055
2056 if (!td_write(td) || td->o.allow_mounted_write)
e39c0676 2057 return false;
e81ecca3
JA
2058
2059 for_each_file(td, f, i) {
686fbd31 2060 if (f->filetype != FIO_TYPE_BLOCK)
e81ecca3
JA
2061 continue;
2062 if (device_is_mounted(f->file_name))
2063 goto mounted;
2064 }
2065
e39c0676 2066 return false;
e81ecca3 2067mounted:
c5839011 2068 log_err("fio: %s appears mounted, and 'allow_mounted_write' isn't set. Aborting.\n", f->file_name);
e39c0676 2069 return true;
e81ecca3
JA
2070}
2071
9cc8cb91
AK
2072static bool waitee_running(struct thread_data *me)
2073{
2074 const char *waitee = me->o.wait_for;
2075 const char *self = me->o.name;
2076 struct thread_data *td;
2077 int i;
2078
2079 if (!waitee)
2080 return false;
2081
2082 for_each_td(td, i) {
2083 if (!strcmp(td->o.name, self) || strcmp(td->o.name, waitee))
2084 continue;
2085
2086 if (td->runstate < TD_EXITED) {
2087 dprint(FD_PROCESS, "%s fenced by %s(%s)\n",
2088 self, td->o.name,
2089 runstate_to_name(td->runstate));
2090 return true;
2091 }
2092 }
2093
2094 dprint(FD_PROCESS, "%s: %s completed, can run\n", self, waitee);
2095 return false;
2096}
2097
2e1df07d
JA
2098/*
2099 * Main function for kicking off and reaping jobs, as needed.
2100 */
24660963 2101static void run_threads(struct sk_out *sk_out)
2e1df07d
JA
2102{
2103 struct thread_data *td;
90eff1c9
SW
2104 unsigned int i, todo, nr_running, nr_started;
2105 uint64_t m_rate, t_rate;
0de5b26f 2106 uint64_t spent;
2e1df07d 2107
2e1df07d
JA
2108 if (fio_gtod_offload && fio_start_gtod_thread())
2109 return;
334185e9 2110
f2a2ce0e 2111 fio_idle_prof_init();
2e1df07d
JA
2112
2113 set_sig_handlers();
2114
3a5f6bde
JA
2115 nr_thread = nr_process = 0;
2116 for_each_td(td, i) {
e81ecca3
JA
2117 if (check_mount_writes(td))
2118 return;
3a5f6bde
JA
2119 if (td->o.use_thread)
2120 nr_thread++;
2121 else
2122 nr_process++;
2123 }
2124
01cfefcc 2125 if (output_format & FIO_OUTPUT_NORMAL) {
2e1df07d
JA
2126 log_info("Starting ");
2127 if (nr_thread)
2128 log_info("%d thread%s", nr_thread,
2129 nr_thread > 1 ? "s" : "");
2130 if (nr_process) {
2131 if (nr_thread)
2132 log_info(" and ");
2133 log_info("%d process%s", nr_process,
2134 nr_process > 1 ? "es" : "");
2135 }
2136 log_info("\n");
e411c301 2137 log_info_flush();
2e1df07d
JA
2138 }
2139
2140 todo = thread_number;
2141 nr_running = 0;
2142 nr_started = 0;
2143 m_rate = t_rate = 0;
2144
2145 for_each_td(td, i) {
2146 print_status_init(td->thread_number - 1);
2147
2148 if (!td->o.create_serialize)
2149 continue;
2150
ca09be4b
JA
2151 if (fio_verify_load_state(td))
2152 goto reap;
2153
2e1df07d
JA
2154 /*
2155 * do file setup here so it happens sequentially,
2156 * we don't want X number of threads getting their
2157 * client data interspersed on disk
2158 */
2159 if (setup_files(td)) {
ca09be4b 2160reap:
2e1df07d
JA
2161 exit_value++;
2162 if (td->error)
2163 log_err("fio: pid=%d, err=%d/%s\n",
2164 (int) td->pid, td->error, td->verror);
2165 td_set_runstate(td, TD_REAPED);
2166 todo--;
2167 } else {
2168 struct fio_file *f;
2169 unsigned int j;
2170
2171 /*
2172 * for sharing to work, each job must always open
2173 * its own files. so close them, if we opened them
2174 * for creation
2175 */
2176 for_each_file(td, f, j) {
2177 if (fio_file_open(f))
2178 td_io_close_file(td, f);
2179 }
2180 }
2181 }
2182
f2a2ce0e
HL
2183 /* start idle threads before io threads start to run */
2184 fio_idle_prof_start();
2185
2e1df07d
JA
2186 set_genesis_time();
2187
2188 while (todo) {
2189 struct thread_data *map[REAL_MAX_JOBS];
2190 struct timeval this_start;
2191 int this_jobs = 0, left;
ce8b6139 2192 struct fork_data *fd;
2e1df07d
JA
2193
2194 /*
2195 * create threads (TD_NOT_CREATED -> TD_CREATED)
2196 */
2197 for_each_td(td, i) {
2198 if (td->runstate != TD_NOT_CREATED)
2199 continue;
2200
2201 /*
2202 * never got a chance to start, killed by other
2203 * thread for some reason
2204 */
2205 if (td->terminate) {
2206 todo--;
2207 continue;
2208 }
2209
2210 if (td->o.start_delay) {
0de5b26f 2211 spent = utime_since_genesis();
2e1df07d 2212
74454ce4 2213 if (td->o.start_delay > spent)
2e1df07d
JA
2214 continue;
2215 }
2216
2217 if (td->o.stonewall && (nr_started || nr_running)) {
2218 dprint(FD_PROCESS, "%s: stonewall wait\n",
2219 td->o.name);
2220 break;
2221 }
2222
9cc8cb91
AK
2223 if (waitee_running(td)) {
2224 dprint(FD_PROCESS, "%s: waiting for %s\n",
2225 td->o.name, td->o.wait_for);
5c74fc76 2226 continue;
9cc8cb91
AK
2227 }
2228
2e1df07d
JA
2229 init_disk_util(td);
2230
c97f1ad6
JA
2231 td->rusage_sem = fio_mutex_init(FIO_MUTEX_LOCKED);
2232 td->update_rusage = 0;
2233
2e1df07d
JA
2234 /*
2235 * Set state to created. Thread will transition
2236 * to TD_INITIALIZED when it's done setting up.
2237 */
2238 td_set_runstate(td, TD_CREATED);
2239 map[this_jobs++] = td;
2240 nr_started++;
2241
ce8b6139
JA
2242 fd = calloc(1, sizeof(*fd));
2243 fd->td = td;
2244 fd->sk_out = sk_out;
2245
2e1df07d
JA
2246 if (td->o.use_thread) {
2247 int ret;
2248
2249 dprint(FD_PROCESS, "will pthread_create\n");
2250 ret = pthread_create(&td->thread, NULL,
24660963 2251 thread_main, fd);
2e1df07d
JA
2252 if (ret) {
2253 log_err("pthread_create: %s\n",
2254 strerror(ret));
24660963 2255 free(fd);
2e1df07d
JA
2256 nr_started--;
2257 break;
2258 }
2259 ret = pthread_detach(td->thread);
2260 if (ret)
2261 log_err("pthread_detach: %s",
2262 strerror(ret));
2263 } else {
2264 pid_t pid;
2265 dprint(FD_PROCESS, "will fork\n");
2266 pid = fork();
2267 if (!pid) {
ce8b6139 2268 int ret;
2e1df07d 2269
ce8b6139 2270 ret = (int)(uintptr_t)thread_main(fd);
2e1df07d
JA
2271 _exit(ret);
2272 } else if (i == fio_debug_jobno)
2273 *fio_debug_jobp = pid;
2274 }
2275 dprint(FD_MUTEX, "wait on startup_mutex\n");
09400a60 2276 if (fio_mutex_down_timeout(startup_mutex, 10000)) {
2e1df07d
JA
2277 log_err("fio: job startup hung? exiting.\n");
2278 fio_terminate_threads(TERMINATE_ALL);
2279 fio_abort = 1;
2280 nr_started--;
2281 break;
2282 }
2283 dprint(FD_MUTEX, "done waiting on startup_mutex\n");
2284 }
2285
2286 /*
2287 * Wait for the started threads to transition to
2288 * TD_INITIALIZED.
2289 */
2290 fio_gettime(&this_start, NULL);
2291 left = this_jobs;
2292 while (left && !fio_abort) {
2293 if (mtime_since_now(&this_start) > JOB_START_TIMEOUT)
2294 break;
2295
06464907 2296 do_usleep(100000);
2e1df07d
JA
2297
2298 for (i = 0; i < this_jobs; i++) {
2299 td = map[i];
2300 if (!td)
2301 continue;
2302 if (td->runstate == TD_INITIALIZED) {
2303 map[i] = NULL;
2304 left--;
2305 } else if (td->runstate >= TD_EXITED) {
2306 map[i] = NULL;
2307 left--;
2308 todo--;
2309 nr_running++; /* work-around... */
2310 }
2311 }
2312 }
2313
2314 if (left) {
4e87c37a
JA
2315 log_err("fio: %d job%s failed to start\n", left,
2316 left > 1 ? "s" : "");
2e1df07d
JA
2317 for (i = 0; i < this_jobs; i++) {
2318 td = map[i];
2319 if (!td)
2320 continue;
2321 kill(td->pid, SIGTERM);
2322 }
2323 break;
2324 }
2325
2326 /*
2327 * start created threads (TD_INITIALIZED -> TD_RUNNING).
2328 */
2329 for_each_td(td, i) {
2330 if (td->runstate != TD_INITIALIZED)
2331 continue;
2332
2333 if (in_ramp_time(td))
2334 td_set_runstate(td, TD_RAMP);
2335 else
2336 td_set_runstate(td, TD_RUNNING);
2337 nr_running++;
2338 nr_started--;
342f4be4
JA
2339 m_rate += ddir_rw_sum(td->o.ratemin);
2340 t_rate += ddir_rw_sum(td->o.rate);
2e1df07d
JA
2341 todo--;
2342 fio_mutex_up(td->mutex);
2343 }
2344
2345 reap_threads(&nr_running, &t_rate, &m_rate);
2346
122c7725 2347 if (todo)
06464907 2348 do_usleep(100000);
2e1df07d
JA
2349 }
2350
2351 while (nr_running) {
2352 reap_threads(&nr_running, &t_rate, &m_rate);
06464907 2353 do_usleep(10000);
2e1df07d
JA
2354 }
2355
f2a2ce0e
HL
2356 fio_idle_prof_stop();
2357
2e1df07d 2358 update_io_ticks();
2e1df07d
JA
2359}
2360
27357187
JA
2361static void free_disk_util(void)
2362{
27357187 2363 disk_util_prune_entries();
a39fb9ea 2364 helper_thread_destroy();
2e1df07d
JA
2365}
2366
24660963 2367int fio_backend(struct sk_out *sk_out)
2e1df07d
JA
2368{
2369 struct thread_data *td;
2370 int i;
2371
2372 if (exec_profile) {
2373 if (load_profile(exec_profile))
2374 return 1;
2375 free(exec_profile);
2376 exec_profile = NULL;
2377 }
2378 if (!thread_number)
2379 return 0;
2380
2381 if (write_bw_log) {
aee2ab67
JA
2382 struct log_params p = {
2383 .log_type = IO_LOG_TYPE_BW,
2384 };
2385
2386 setup_log(&agg_io_log[DDIR_READ], &p, "agg-read_bw.log");
2387 setup_log(&agg_io_log[DDIR_WRITE], &p, "agg-write_bw.log");
2388 setup_log(&agg_io_log[DDIR_TRIM], &p, "agg-trim_bw.log");
2e1df07d
JA
2389 }
2390
521da527 2391 startup_mutex = fio_mutex_init(FIO_MUTEX_LOCKED);
2e1df07d
JA
2392 if (startup_mutex == NULL)
2393 return 1;
2e1df07d
JA
2394
2395 set_genesis_time();
cef9175e 2396 stat_init();
a39fb9ea 2397 helper_thread_create(startup_mutex, sk_out);
2e1df07d
JA
2398
2399 cgroup_list = smalloc(sizeof(*cgroup_list));
2400 INIT_FLIST_HEAD(cgroup_list);
2401
24660963 2402 run_threads(sk_out);
2e1df07d 2403
a39fb9ea 2404 helper_thread_exit();
8aab824f 2405
2e1df07d 2406 if (!fio_abort) {
83f7b64e 2407 __show_run_stats();
2e1df07d 2408 if (write_bw_log) {
cb7e0ace
JA
2409 for (i = 0; i < DDIR_RWDIR_CNT; i++) {
2410 struct io_log *log = agg_io_log[i];
2411
60a25727 2412 flush_log(log, false);
518dac09 2413 free_log(log);
cb7e0ace 2414 }
2e1df07d
JA
2415 }
2416 }
2417
fda2cfac 2418 for_each_td(td, i) {
ba8fb6f6
VF
2419 if (td->ss.dur) {
2420 if (td->ss.iops_data != NULL) {
2421 free(td->ss.iops_data);
2422 free(td->ss.bw_data);
2423 }
2424 }
2e1df07d 2425 fio_options_free(td);
8049adc1
JA
2426 if (td->rusage_sem) {
2427 fio_mutex_remove(td->rusage_sem);
2428 td->rusage_sem = NULL;
2429 }
fda2cfac 2430 }
2e1df07d 2431
a462baef 2432 free_disk_util();
2e1df07d
JA
2433 cgroup_kill(cgroup_list);
2434 sfree(cgroup_list);
2435 sfree(cgroup_mnt);
2436
2437 fio_mutex_remove(startup_mutex);
cef9175e 2438 stat_exit();
2e1df07d
JA
2439 return exit_value;
2440}