From 405113013b076aeb89634a333af96f948044b110 Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Tue, 8 Dec 2015 08:51:43 -0700 Subject: [PATCH] backend: move rated submit code to its own file Signed-off-by: Jens Axboe --- Makefile | 2 +- backend.c | 226 +------------------------------------------------ fio.h | 4 + rate-submit.c | 228 ++++++++++++++++++++++++++++++++++++++++++++++++++ rate-submit.h | 8 ++ 5 files changed, 242 insertions(+), 226 deletions(-) create mode 100644 rate-submit.c create mode 100644 rate-submit.h diff --git a/Makefile b/Makefile index 102cc69c..2ff88a1d 100644 --- a/Makefile +++ b/Makefile @@ -45,7 +45,7 @@ SOURCE := $(patsubst $(SRCDIR)/%,%,$(wildcard $(SRCDIR)/crc/*.c)) \ server.c client.c iolog.c backend.c libfio.c flow.c cconv.c \ gettime-thread.c helpers.c json.c idletime.c td_error.c \ profiles/tiobench.c profiles/act.c io_u_queue.c filelock.c \ - workqueue.c + workqueue.c rate-submit.c ifdef CONFIG_LIBHDFS HDFSFLAGS= -I $(JAVA_HOME)/include -I $(JAVA_HOME)/include/linux -I $(FIO_LIBHDFS_INCLUDE) diff --git a/backend.c b/backend.c index c18cf209..01e0b3ac 100644 --- a/backend.c +++ b/backend.c @@ -57,6 +57,7 @@ #include "lib/tp.h" #include "workqueue.h" #include "lib/mountcheck.h" +#include "rate-submit.h" static pthread_t helper_thread; static pthread_mutex_t helper_lock; @@ -1361,231 +1362,6 @@ static uint64_t do_dry_run(struct thread_data *td) return td->bytes_done[DDIR_WRITE] + td->bytes_done[DDIR_TRIM]; } -static void io_workqueue_fn(struct submit_worker *sw, - struct workqueue_work *work) -{ - struct io_u *io_u = container_of(work, struct io_u, work); - const enum fio_ddir ddir = io_u->ddir; - struct thread_data *td = sw->private; - int ret; - - dprint(FD_RATE, "io_u %p queued by %u\n", io_u, gettid()); - - io_u_set(io_u, IO_U_F_NO_FILE_PUT); - - td->cur_depth++; - - do { - ret = td_io_queue(td, io_u); - if (ret != FIO_Q_BUSY) - break; - ret = io_u_queued_complete(td, 1); - if (ret > 0) - td->cur_depth -= ret; - io_u_clear(io_u, IO_U_F_FLIGHT); - } while (1); - - dprint(FD_RATE, "io_u %p ret %d by %u\n", io_u, ret, gettid()); - - io_queue_event(td, io_u, &ret, ddir, NULL, 0, NULL); - - if (ret == FIO_Q_COMPLETED) - td->cur_depth--; - else if (ret == FIO_Q_QUEUED) { - unsigned int min_evts; - - if (td->o.iodepth == 1) - min_evts = 1; - else - min_evts = 0; - - ret = io_u_queued_complete(td, min_evts); - if (ret > 0) - td->cur_depth -= ret; - } else if (ret == FIO_Q_BUSY) { - ret = io_u_queued_complete(td, td->cur_depth); - if (ret > 0) - td->cur_depth -= ret; - } -} - -static bool io_workqueue_pre_sleep_flush_fn(struct submit_worker *sw) -{ - struct thread_data *td = sw->private; - - if (td->io_u_queued || td->cur_depth || td->io_u_in_flight) - return true; - - return false; -} - -static void io_workqueue_pre_sleep_fn(struct submit_worker *sw) -{ - struct thread_data *td = sw->private; - int ret; - - ret = io_u_quiesce(td); - if (ret > 0) - td->cur_depth -= ret; -} - -static int io_workqueue_alloc_fn(struct submit_worker *sw) -{ - struct thread_data *td; - - td = calloc(1, sizeof(*td)); - sw->private = td; - return 0; -} - -static void io_workqueue_free_fn(struct submit_worker *sw) -{ - free(sw->private); - sw->private = NULL; -} - -static int io_workqueue_init_worker_fn(struct submit_worker *sw) -{ - struct thread_data *parent = sw->wq->td; - struct thread_data *td = sw->private; - int fio_unused ret; - - memcpy(&td->o, &parent->o, sizeof(td->o)); - memcpy(&td->ts, &parent->ts, sizeof(td->ts)); - td->o.uid = td->o.gid = -1U; - dup_files(td, parent); - td->eo = parent->eo; - fio_options_mem_dupe(td); - - if (ioengine_load(td)) - goto err; - - if (td->o.odirect) - td->io_ops->flags |= FIO_RAWIO; - - td->pid = gettid(); - - INIT_FLIST_HEAD(&td->io_log_list); - INIT_FLIST_HEAD(&td->io_hist_list); - INIT_FLIST_HEAD(&td->verify_list); - INIT_FLIST_HEAD(&td->trim_list); - INIT_FLIST_HEAD(&td->next_rand_list); - td->io_hist_tree = RB_ROOT; - - td->o.iodepth = 1; - if (td_io_init(td)) - goto err_io_init; - - fio_gettime(&td->epoch, NULL); - fio_getrusage(&td->ru_start); - clear_io_state(td, 1); - - td_set_runstate(td, TD_RUNNING); - td->flags |= TD_F_CHILD; - td->parent = parent; - return 0; - -err_io_init: - close_ioengine(td); -err: - return 1; - -} - -static void io_workqueue_exit_worker_fn(struct submit_worker *sw, - unsigned int *sum_cnt) -{ - struct thread_data *td = sw->private; - - (*sum_cnt)++; - sum_thread_stats(&sw->wq->td->ts, &td->ts, *sum_cnt == 1); - - fio_options_free(td); - close_and_free_files(td); - if (td->io_ops) - close_ioengine(td); - td_set_runstate(td, TD_EXITED); -} - -#ifdef CONFIG_SFAA -static void sum_val(uint64_t *dst, uint64_t *src) -{ - if (*src) { - __sync_fetch_and_add(dst, *src); - *src = 0; - } -} -#else -static void sum_val(uint64_t *dst, uint64_t *src) -{ - if (*src) { - *dst += *src; - *src = 0; - } -} -#endif - -static void pthread_double_unlock(pthread_mutex_t *lock1, - pthread_mutex_t *lock2) -{ -#ifndef CONFIG_SFAA - pthread_mutex_unlock(lock1); - pthread_mutex_unlock(lock2); -#endif -} - -static void pthread_double_lock(pthread_mutex_t *lock1, pthread_mutex_t *lock2) -{ -#ifndef CONFIG_SFAA - if (lock1 < lock2) { - pthread_mutex_lock(lock1); - pthread_mutex_lock(lock2); - } else { - pthread_mutex_lock(lock2); - pthread_mutex_lock(lock1); - } -#endif -} - -static void sum_ddir(struct thread_data *dst, struct thread_data *src, - enum fio_ddir ddir) -{ - pthread_double_lock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock); - - sum_val(&dst->io_bytes[ddir], &src->io_bytes[ddir]); - sum_val(&dst->io_blocks[ddir], &src->io_blocks[ddir]); - sum_val(&dst->this_io_blocks[ddir], &src->this_io_blocks[ddir]); - sum_val(&dst->this_io_bytes[ddir], &src->this_io_bytes[ddir]); - sum_val(&dst->bytes_done[ddir], &src->bytes_done[ddir]); - - pthread_double_unlock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock); -} - -static void io_workqueue_update_acct_fn(struct submit_worker *sw) -{ - struct thread_data *src = sw->private; - struct thread_data *dst = sw->wq->td; - - if (td_read(src)) - sum_ddir(dst, src, DDIR_READ); - if (td_write(src)) - sum_ddir(dst, src, DDIR_WRITE); - if (td_trim(src)) - sum_ddir(dst, src, DDIR_TRIM); - -} - -struct workqueue_ops rated_wq_ops = { - .fn = io_workqueue_fn, - .pre_sleep_flush_fn = io_workqueue_pre_sleep_flush_fn, - .pre_sleep_fn = io_workqueue_pre_sleep_fn, - .update_acct_fn = io_workqueue_update_acct_fn, - .alloc_worker_fn = io_workqueue_alloc_fn, - .free_worker_fn = io_workqueue_free_fn, - .init_worker_fn = io_workqueue_init_worker_fn, - .exit_worker_fn = io_workqueue_exit_worker_fn, -}; - /* * Entry point for the thread based jobs. The process based jobs end up * here as well, after a little setup. diff --git a/fio.h b/fio.h index 2e13b548..e4f778eb 100644 --- a/fio.h +++ b/fio.h @@ -565,6 +565,10 @@ extern int is_blktrace(const char *, int *); extern int load_blktrace(struct thread_data *, const char *, int); #endif +extern int io_queue_event(struct thread_data *td, struct io_u *io_u, int *ret, + enum fio_ddir ddir, uint64_t *bytes_issued, int from_verify, + struct timeval *comp_time); + /* * Latency target helpers */ diff --git a/rate-submit.c b/rate-submit.c new file mode 100644 index 00000000..a68a3dd3 --- /dev/null +++ b/rate-submit.c @@ -0,0 +1,228 @@ +#include "fio.h" +#include "ioengine.h" +#include "lib/getrusage.h" + +static void io_workqueue_fn(struct submit_worker *sw, + struct workqueue_work *work) +{ + struct io_u *io_u = container_of(work, struct io_u, work); + const enum fio_ddir ddir = io_u->ddir; + struct thread_data *td = sw->private; + int ret; + + dprint(FD_RATE, "io_u %p queued by %u\n", io_u, gettid()); + + io_u_set(io_u, IO_U_F_NO_FILE_PUT); + + td->cur_depth++; + + do { + ret = td_io_queue(td, io_u); + if (ret != FIO_Q_BUSY) + break; + ret = io_u_queued_complete(td, 1); + if (ret > 0) + td->cur_depth -= ret; + io_u_clear(io_u, IO_U_F_FLIGHT); + } while (1); + + dprint(FD_RATE, "io_u %p ret %d by %u\n", io_u, ret, gettid()); + + io_queue_event(td, io_u, &ret, ddir, NULL, 0, NULL); + + if (ret == FIO_Q_COMPLETED) + td->cur_depth--; + else if (ret == FIO_Q_QUEUED) { + unsigned int min_evts; + + if (td->o.iodepth == 1) + min_evts = 1; + else + min_evts = 0; + + ret = io_u_queued_complete(td, min_evts); + if (ret > 0) + td->cur_depth -= ret; + } else if (ret == FIO_Q_BUSY) { + ret = io_u_queued_complete(td, td->cur_depth); + if (ret > 0) + td->cur_depth -= ret; + } +} + +static bool io_workqueue_pre_sleep_flush_fn(struct submit_worker *sw) +{ + struct thread_data *td = sw->private; + + if (td->io_u_queued || td->cur_depth || td->io_u_in_flight) + return true; + + return false; +} + +static void io_workqueue_pre_sleep_fn(struct submit_worker *sw) +{ + struct thread_data *td = sw->private; + int ret; + + ret = io_u_quiesce(td); + if (ret > 0) + td->cur_depth -= ret; +} + +static int io_workqueue_alloc_fn(struct submit_worker *sw) +{ + struct thread_data *td; + + td = calloc(1, sizeof(*td)); + sw->private = td; + return 0; +} + +static void io_workqueue_free_fn(struct submit_worker *sw) +{ + free(sw->private); + sw->private = NULL; +} + +static int io_workqueue_init_worker_fn(struct submit_worker *sw) +{ + struct thread_data *parent = sw->wq->td; + struct thread_data *td = sw->private; + int fio_unused ret; + + memcpy(&td->o, &parent->o, sizeof(td->o)); + memcpy(&td->ts, &parent->ts, sizeof(td->ts)); + td->o.uid = td->o.gid = -1U; + dup_files(td, parent); + td->eo = parent->eo; + fio_options_mem_dupe(td); + + if (ioengine_load(td)) + goto err; + + if (td->o.odirect) + td->io_ops->flags |= FIO_RAWIO; + + td->pid = gettid(); + + INIT_FLIST_HEAD(&td->io_log_list); + INIT_FLIST_HEAD(&td->io_hist_list); + INIT_FLIST_HEAD(&td->verify_list); + INIT_FLIST_HEAD(&td->trim_list); + INIT_FLIST_HEAD(&td->next_rand_list); + td->io_hist_tree = RB_ROOT; + + td->o.iodepth = 1; + if (td_io_init(td)) + goto err_io_init; + + fio_gettime(&td->epoch, NULL); + fio_getrusage(&td->ru_start); + clear_io_state(td, 1); + + td_set_runstate(td, TD_RUNNING); + td->flags |= TD_F_CHILD; + td->parent = parent; + return 0; + +err_io_init: + close_ioengine(td); +err: + return 1; + +} + +static void io_workqueue_exit_worker_fn(struct submit_worker *sw, + unsigned int *sum_cnt) +{ + struct thread_data *td = sw->private; + + (*sum_cnt)++; + sum_thread_stats(&sw->wq->td->ts, &td->ts, *sum_cnt == 1); + + fio_options_free(td); + close_and_free_files(td); + if (td->io_ops) + close_ioengine(td); + td_set_runstate(td, TD_EXITED); +} + +#ifdef CONFIG_SFAA +static void sum_val(uint64_t *dst, uint64_t *src) +{ + if (*src) { + __sync_fetch_and_add(dst, *src); + *src = 0; + } +} +#else +static void sum_val(uint64_t *dst, uint64_t *src) +{ + if (*src) { + *dst += *src; + *src = 0; + } +} +#endif + +static void pthread_double_unlock(pthread_mutex_t *lock1, + pthread_mutex_t *lock2) +{ +#ifndef CONFIG_SFAA + pthread_mutex_unlock(lock1); + pthread_mutex_unlock(lock2); +#endif +} + +static void pthread_double_lock(pthread_mutex_t *lock1, pthread_mutex_t *lock2) +{ +#ifndef CONFIG_SFAA + if (lock1 < lock2) { + pthread_mutex_lock(lock1); + pthread_mutex_lock(lock2); + } else { + pthread_mutex_lock(lock2); + pthread_mutex_lock(lock1); + } +#endif +} + +static void sum_ddir(struct thread_data *dst, struct thread_data *src, + enum fio_ddir ddir) +{ + pthread_double_lock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock); + + sum_val(&dst->io_bytes[ddir], &src->io_bytes[ddir]); + sum_val(&dst->io_blocks[ddir], &src->io_blocks[ddir]); + sum_val(&dst->this_io_blocks[ddir], &src->this_io_blocks[ddir]); + sum_val(&dst->this_io_bytes[ddir], &src->this_io_bytes[ddir]); + sum_val(&dst->bytes_done[ddir], &src->bytes_done[ddir]); + + pthread_double_unlock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock); +} + +static void io_workqueue_update_acct_fn(struct submit_worker *sw) +{ + struct thread_data *src = sw->private; + struct thread_data *dst = sw->wq->td; + + if (td_read(src)) + sum_ddir(dst, src, DDIR_READ); + if (td_write(src)) + sum_ddir(dst, src, DDIR_WRITE); + if (td_trim(src)) + sum_ddir(dst, src, DDIR_TRIM); + +} + +struct workqueue_ops rated_wq_ops = { + .fn = io_workqueue_fn, + .pre_sleep_flush_fn = io_workqueue_pre_sleep_flush_fn, + .pre_sleep_fn = io_workqueue_pre_sleep_fn, + .update_acct_fn = io_workqueue_update_acct_fn, + .alloc_worker_fn = io_workqueue_alloc_fn, + .free_worker_fn = io_workqueue_free_fn, + .init_worker_fn = io_workqueue_init_worker_fn, + .exit_worker_fn = io_workqueue_exit_worker_fn, +}; diff --git a/rate-submit.h b/rate-submit.h new file mode 100644 index 00000000..33bb9d77 --- /dev/null +++ b/rate-submit.h @@ -0,0 +1,8 @@ +#ifndef FIO_RATE_SUBMIT +#define FIO_RATE_SUBMIT + +#include "workqueue.h" + +extern struct workqueue_ops rated_wq_ops; + +#endif -- 2.25.1