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