#include "lib/mountcheck.h"
#include "rate-submit.h"
-static pthread_t helper_thread;
-static pthread_mutex_t helper_lock;
-pthread_cond_t helper_cond;
-int helper_do_stat = 0;
-
static struct fio_mutex *startup_mutex;
static struct flist_head *cgroup_list;
static char *cgroup_mnt;
int shm_id = 0;
int temp_stall_ts;
unsigned long done_secs = 0;
-volatile int helper_exit = 0;
+
+static struct helper_data {
+ volatile int exit;
+ volatile int reset;
+ volatile int do_stat;
+ struct sk_out *sk_out;
+ pthread_t thread;
+ pthread_mutex_t lock;
+ pthread_cond_t cond;
+} *helper_data;
#define PAGE_ALIGN(buf) \
(char *) (((uintptr_t) (buf) + page_mask) & ~page_mask)
put_io_u(td, io_u);
return true;
} else if (ret == FIO_Q_QUEUED) {
+ if (td_io_commit(td))
+ return true;
if (io_u_queued_complete(td, 1) < 0)
return true;
} else if (ret == FIO_Q_COMPLETED) {
if (*ret < 0)
break;
}
+
+ /*
+ * when doing I/O (not when verifying),
+ * check for any errors that are to be ignored
+ */
+ if (!from_verify)
+ break;
+
return 0;
case FIO_Q_QUEUED:
/*
if (flow_threshold_exceeded(td))
continue;
- if (!td->o.time_based && bytes_issued >= total_bytes)
+ /*
+ * Break if we exceeded the bytes. The exception is time
+ * based runs, but we still need to break out of the loop
+ * for those to run verification, if enabled.
+ */
+ if (bytes_issued >= total_bytes &&
+ (!td->o.time_based ||
+ (td->o.time_based && td->o.verify != VERIFY_NONE)))
break;
io_u = get_io_u(td);
bytes_done[i] = td->bytes_done[i] - bytes_done[i];
}
+static void free_file_completion_logging(struct thread_data *td)
+{
+ struct fio_file *f;
+ unsigned int i;
+
+ for_each_file(td, f, i) {
+ if (!f->last_write_comp)
+ break;
+ sfree(f->last_write_comp);
+ }
+}
+
+static int init_file_completion_logging(struct thread_data *td,
+ unsigned int depth)
+{
+ struct fio_file *f;
+ unsigned int i;
+
+ if (td->o.verify == VERIFY_NONE || !td->o.verify_state_save)
+ return 0;
+
+ for_each_file(td, f, i) {
+ f->last_write_comp = scalloc(depth, sizeof(uint64_t));
+ if (!f->last_write_comp)
+ goto cleanup;
+ }
+
+ return 0;
+
+cleanup:
+ free_file_completion_logging(td);
+ log_err("fio: failed to alloc write comp data\n");
+ return 1;
+}
+
static void cleanup_io_u(struct thread_data *td)
{
struct io_u *io_u;
io_u_qexit(&td->io_u_freelist);
io_u_qexit(&td->io_u_all);
- if (td->last_write_comp)
- sfree(td->last_write_comp);
+ free_file_completion_logging(td);
}
static int init_io_u(struct thread_data *td)
p += max_bs;
}
- if (td->o.verify != VERIFY_NONE) {
- td->last_write_comp = scalloc(max_units, sizeof(uint64_t));
- if (!td->last_write_comp) {
- log_err("fio: failed to alloc write comp data\n");
- return 1;
- }
- }
+ if (init_file_completion_logging(td, max_units))
+ return 1;
return 0;
}
if (diff < td_max_bs(td))
return false;
- if (fio_files_done(td))
+ if (fio_files_done(td) && !td->o.io_limit)
return false;
return true;
fio_unpin_memory(td);
- fio_writeout_logs(td);
+ td_writeout_logs(td, true);
iolog_compress_exit(td);
rate_submit_exit(td);
cgroup_shutdown(td, &cgroup_mnt);
verify_free_state(td);
+ if (td->zone_state_index) {
+ int i;
+
+ for (i = 0; i < DDIR_RWDIR_CNT; i++)
+ free(td->zone_state_index[i]);
+ free(td->zone_state_index);
+ td->zone_state_index = NULL;
+ }
+
if (fio_option_is_set(o, cpumask)) {
ret = fio_cpuset_exit(&o->cpumask);
if (ret)
if (is_backend) {
void *data;
- int ver;
ret = fio_server_get_verify_state(td->o.name,
- td->thread_number - 1, &data, &ver);
+ td->thread_number - 1, &data);
if (!ret)
- verify_convert_assign_state(td, data, ver);
+ verify_assign_state(td, data);
} else
ret = verify_load_state(td, "local");
return true;
}
+static bool waitee_running(struct thread_data *me)
+{
+ const char *waitee = me->o.wait_for;
+ const char *self = me->o.name;
+ struct thread_data *td;
+ int i;
+
+ if (!waitee)
+ return false;
+
+ for_each_td(td, i) {
+ if (!strcmp(td->o.name, self) || strcmp(td->o.name, waitee))
+ continue;
+
+ if (td->runstate < TD_EXITED) {
+ dprint(FD_PROCESS, "%s fenced by %s(%s)\n",
+ self, td->o.name,
+ runstate_to_name(td->runstate));
+ return true;
+ }
+ }
+
+ dprint(FD_PROCESS, "%s: %s completed, can run\n", self, waitee);
+ return false;
+}
+
/*
* Main function for kicking off and reaping jobs, as needed.
*/
break;
}
+ if (waitee_running(td)) {
+ dprint(FD_PROCESS, "%s: waiting for %s\n",
+ td->o.name, td->o.wait_for);
+ continue;
+ }
+
init_disk_util(td);
td->rusage_sem = fio_mutex_init(FIO_MUTEX_LOCKED);
update_io_ticks();
}
+void helper_reset(void)
+{
+ if (!helper_data)
+ return;
+
+ pthread_mutex_lock(&helper_data->lock);
+
+ if (!helper_data->reset) {
+ helper_data->reset = 1;
+ pthread_cond_signal(&helper_data->cond);
+ }
+
+ pthread_mutex_unlock(&helper_data->lock);
+}
+
+void helper_do_stat(void)
+{
+ if (!helper_data)
+ return;
+
+ pthread_mutex_lock(&helper_data->lock);
+ helper_data->do_stat = 1;
+ pthread_cond_signal(&helper_data->cond);
+ pthread_mutex_unlock(&helper_data->lock);
+}
+
+bool helper_should_exit(void)
+{
+ if (!helper_data)
+ return true;
+
+ return helper_data->exit;
+}
+
static void wait_for_helper_thread_exit(void)
{
void *ret;
- helper_exit = 1;
- pthread_cond_signal(&helper_cond);
- pthread_join(helper_thread, &ret);
+ pthread_mutex_lock(&helper_data->lock);
+ helper_data->exit = 1;
+ pthread_cond_signal(&helper_data->cond);
+ pthread_mutex_unlock(&helper_data->lock);
+
+ pthread_join(helper_data->thread, &ret);
}
static void free_disk_util(void)
{
disk_util_prune_entries();
- pthread_cond_destroy(&helper_cond);
+ pthread_cond_destroy(&helper_data->cond);
+ pthread_mutex_destroy(&helper_data->lock);
+ sfree(helper_data);
}
static void *helper_thread_main(void *data)
{
- struct sk_out *sk_out = data;
+ struct helper_data *hd = data;
+ unsigned int msec_to_next_event, next_log;
+ struct timeval tv, last_du;
int ret = 0;
- sk_out_assign(sk_out);
+ sk_out_assign(hd->sk_out);
+
+ gettimeofday(&tv, NULL);
+ memcpy(&last_du, &tv, sizeof(tv));
fio_mutex_up(startup_mutex);
- while (!ret) {
- uint64_t sec = DISK_UTIL_MSEC / 1000;
- uint64_t nsec = (DISK_UTIL_MSEC % 1000) * 1000000;
+ msec_to_next_event = DISK_UTIL_MSEC;
+ while (!ret && !hd->exit) {
struct timespec ts;
- struct timeval tv;
+ struct timeval now;
+ uint64_t since_du;
+
+ timeval_add_msec(&tv, msec_to_next_event);
+ ts.tv_sec = tv.tv_sec;
+ ts.tv_nsec = tv.tv_usec * 1000;
+
+ pthread_mutex_lock(&hd->lock);
+ pthread_cond_timedwait(&hd->cond, &hd->lock, &ts);
- gettimeofday(&tv, NULL);
- ts.tv_sec = tv.tv_sec + sec;
- ts.tv_nsec = (tv.tv_usec * 1000) + nsec;
+ gettimeofday(&now, NULL);
- if (ts.tv_nsec >= 1000000000ULL) {
- ts.tv_nsec -= 1000000000ULL;
- ts.tv_sec++;
+ if (hd->reset) {
+ memcpy(&tv, &now, sizeof(tv));
+ memcpy(&last_du, &now, sizeof(last_du));
+ hd->reset = 0;
}
- pthread_cond_timedwait(&helper_cond, &helper_lock, &ts);
+ pthread_mutex_unlock(&hd->lock);
- ret = update_io_ticks();
+ since_du = mtime_since(&last_du, &now);
+ if (since_du >= DISK_UTIL_MSEC || DISK_UTIL_MSEC - since_du < 10) {
+ ret = update_io_ticks();
+ timeval_add_msec(&last_du, DISK_UTIL_MSEC);
+ msec_to_next_event = DISK_UTIL_MSEC;
+ if (since_du >= DISK_UTIL_MSEC)
+ msec_to_next_event -= (since_du - DISK_UTIL_MSEC);
+ } else {
+ if (since_du >= DISK_UTIL_MSEC)
+ msec_to_next_event = DISK_UTIL_MSEC - (DISK_UTIL_MSEC - since_du);
+ else
+ msec_to_next_event = DISK_UTIL_MSEC;
+ }
- if (helper_do_stat) {
- helper_do_stat = 0;
+ if (hd->do_stat) {
+ hd->do_stat = 0;
__show_running_run_stats();
}
+ next_log = calc_log_samples();
+ if (!next_log)
+ next_log = DISK_UTIL_MSEC;
+
+ msec_to_next_event = min(next_log, msec_to_next_event);
+
if (!is_backend)
print_thread_status();
}
+ fio_writeout_logs(false);
+
sk_out_drop();
return NULL;
}
static int create_helper_thread(struct sk_out *sk_out)
{
+ struct helper_data *hd;
int ret;
+ hd = smalloc(sizeof(*hd));
+
setup_disk_util();
- pthread_cond_init(&helper_cond, NULL);
- pthread_mutex_init(&helper_lock, NULL);
+ hd->sk_out = sk_out;
+ pthread_cond_init(&hd->cond, NULL);
+ pthread_mutex_init(&hd->lock, NULL);
- ret = pthread_create(&helper_thread, NULL, helper_thread_main, sk_out);
+ ret = pthread_create(&hd->thread, NULL, helper_thread_main, hd);
if (ret) {
log_err("Can't create helper thread: %s\n", strerror(ret));
return 1;
}
+ helper_data = hd;
+
dprint(FD_MUTEX, "wait on startup_mutex\n");
fio_mutex_down(startup_mutex);
dprint(FD_MUTEX, "done waiting on startup_mutex\n");
projects / fio.git / blobdiff