[fio.git] / rate-submit.c

/*
 * Rated submission helpers
 *
 * Copyright (C) 2015 Jens Axboe <axboe@kernel.dk>
 *
 */
#include "fio.h"
#include "ioengine.h"
#include "lib/getrusage.h"
#include "rate-submit.h"

static int 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->priv;
	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;
	}

	return 0;
}

static bool io_workqueue_pre_sleep_flush_fn(struct submit_worker *sw)
{
	struct thread_data *td = sw->priv;

	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->priv;
	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->priv = td;
	return 0;
}

static void io_workqueue_free_fn(struct submit_worker *sw)
{
	free(sw->priv);
	sw->priv = NULL;
}

static int io_workqueue_init_worker_fn(struct submit_worker *sw)
{
	struct thread_data *parent = sw->wq->td;
	struct thread_data *td = sw->priv;
	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->priv;

	(*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->priv;
	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);

}

static 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,
};

int rate_submit_init(struct thread_data *td, struct sk_out *sk_out)
{
	if (td->o.io_submit_mode != IO_MODE_OFFLOAD)
		return 0;

	return workqueue_init(td, &td->io_wq, &rated_wq_ops, td->o.iodepth, sk_out);
}

void rate_submit_exit(struct thread_data *td)
{
	if (td->o.io_submit_mode != IO_MODE_OFFLOAD)
		return;

	workqueue_exit(&td->io_wq);
}
Commit	Line	Data
51575029 JA	1	/*
	2	* Rated submission helpers
	3	*
	4	* Copyright (C) 2015 Jens Axboe <axboe@kernel.dk>
	5	*
	6	*/
40511301 JA	7	#include "fio.h"
	8	#include "ioengine.h"
	9	#include "lib/getrusage.h"
24660963	10	#include "rate-submit.h"
40511301	11
155f2f02 JA	12	static int io_workqueue_fn(struct submit_worker *sw,
155f2f02 JA	13	struct workqueue_work *work)
40511301 JA	14	{
	15	struct io_u *io_u = container_of(work, struct io_u, work);
	16	const enum fio_ddir ddir = io_u->ddir;
b86ad8f1	17	struct thread_data *td = sw->priv;
40511301 JA	18	int ret;
	19
	20	dprint(FD_RATE, "io_u %p queued by %u\n", io_u, gettid());
	21
	22	io_u_set(io_u, IO_U_F_NO_FILE_PUT);
	23
	24	td->cur_depth++;
	25
	26	do {
	27	ret = td_io_queue(td, io_u);
	28	if (ret != FIO_Q_BUSY)
	29	break;
	30	ret = io_u_queued_complete(td, 1);
	31	if (ret > 0)
	32	td->cur_depth -= ret;
	33	io_u_clear(io_u, IO_U_F_FLIGHT);
	34	} while (1);
	35
	36	dprint(FD_RATE, "io_u %p ret %d by %u\n", io_u, ret, gettid());
	37
	38	io_queue_event(td, io_u, &ret, ddir, NULL, 0, NULL);
	39
	40	if (ret == FIO_Q_COMPLETED)
	41	td->cur_depth--;
	42	else if (ret == FIO_Q_QUEUED) {
	43	unsigned int min_evts;
	44
	45	if (td->o.iodepth == 1)
	46	min_evts = 1;
	47	else
	48	min_evts = 0;
	49
	50	ret = io_u_queued_complete(td, min_evts);
	51	if (ret > 0)
	52	td->cur_depth -= ret;
	53	} else if (ret == FIO_Q_BUSY) {
	54	ret = io_u_queued_complete(td, td->cur_depth);
	55	if (ret > 0)
	56	td->cur_depth -= ret;
	57	}
155f2f02 JA	58
155f2f02 JA	59	return 0;
40511301 JA	60	}
	61
	62	static bool io_workqueue_pre_sleep_flush_fn(struct submit_worker *sw)
	63	{
b86ad8f1	64	struct thread_data *td = sw->priv;
40511301 JA	65
	66	if (td->io_u_queued \|\| td->cur_depth \|\| td->io_u_in_flight)
	67	return true;
	68
	69	return false;
	70	}
	71
	72	static void io_workqueue_pre_sleep_fn(struct submit_worker *sw)
	73	{
b86ad8f1	74	struct thread_data *td = sw->priv;
40511301 JA	75	int ret;
	76
	77	ret = io_u_quiesce(td);
	78	if (ret > 0)
	79	td->cur_depth -= ret;
	80	}
	81
	82	static int io_workqueue_alloc_fn(struct submit_worker *sw)
	83	{
	84	struct thread_data *td;
	85
	86	td = calloc(1, sizeof(*td));
b86ad8f1	87	sw->priv = td;
40511301 JA	88	return 0;
	89	}
	90
	91	static void io_workqueue_free_fn(struct submit_worker *sw)
	92	{
b86ad8f1 CB	93	free(sw->priv);
b86ad8f1 CB	94	sw->priv = NULL;
40511301 JA	95	}
	96
	97	static int io_workqueue_init_worker_fn(struct submit_worker *sw)
	98	{
	99	struct thread_data *parent = sw->wq->td;
b86ad8f1	100	struct thread_data *td = sw->priv;
40511301 JA	101	int fio_unused ret;
	102
	103	memcpy(&td->o, &parent->o, sizeof(td->o));
	104	memcpy(&td->ts, &parent->ts, sizeof(td->ts));
	105	td->o.uid = td->o.gid = -1U;
	106	dup_files(td, parent);
	107	td->eo = parent->eo;
	108	fio_options_mem_dupe(td);
	109
	110	if (ioengine_load(td))
	111	goto err;
	112
	113	if (td->o.odirect)
	114	td->io_ops->flags \|= FIO_RAWIO;
	115
	116	td->pid = gettid();
	117
	118	INIT_FLIST_HEAD(&td->io_log_list);
	119	INIT_FLIST_HEAD(&td->io_hist_list);
	120	INIT_FLIST_HEAD(&td->verify_list);
	121	INIT_FLIST_HEAD(&td->trim_list);
	122	INIT_FLIST_HEAD(&td->next_rand_list);
	123	td->io_hist_tree = RB_ROOT;
	124
	125	td->o.iodepth = 1;
	126	if (td_io_init(td))
	127	goto err_io_init;
	128
	129	fio_gettime(&td->epoch, NULL);
	130	fio_getrusage(&td->ru_start);
	131	clear_io_state(td, 1);
	132
	133	td_set_runstate(td, TD_RUNNING);
	134	td->flags \|= TD_F_CHILD;
	135	td->parent = parent;
	136	return 0;
	137
	138	err_io_init:
	139	close_ioengine(td);
	140	err:
	141	return 1;
	142
	143	}
	144
	145	static void io_workqueue_exit_worker_fn(struct submit_worker *sw,
	146	unsigned int *sum_cnt)
	147	{
b86ad8f1	148	struct thread_data *td = sw->priv;
40511301 JA	149
	150	(*sum_cnt)++;
	151	sum_thread_stats(&sw->wq->td->ts, &td->ts, *sum_cnt == 1);
	152
	153	fio_options_free(td);
	154	close_and_free_files(td);
	155	if (td->io_ops)
	156	close_ioengine(td);
	157	td_set_runstate(td, TD_EXITED);
	158	}
	159
	160	#ifdef CONFIG_SFAA
	161	static void sum_val(uint64_t dst, uint64_t src)
	162	{
	163	if (*src) {
	164	__sync_fetch_and_add(dst, *src);
	165	*src = 0;
	166	}
	167	}
	168	#else
	169	static void sum_val(uint64_t dst, uint64_t src)
	170	{
	171	if (*src) {
	172	dst += src;
	173	*src = 0;
	174	}
	175	}
	176	#endif
	177
	178	static void pthread_double_unlock(pthread_mutex_t *lock1,
	179	pthread_mutex_t *lock2)
	180	{
	181	#ifndef CONFIG_SFAA
	182	pthread_mutex_unlock(lock1);
	183	pthread_mutex_unlock(lock2);
	184	#endif
	185	}
	186
	187	static void pthread_double_lock(pthread_mutex_t lock1, pthread_mutex_t lock2)
	188	{
	189	#ifndef CONFIG_SFAA
	190	if (lock1 < lock2) {
	191	pthread_mutex_lock(lock1);
	192	pthread_mutex_lock(lock2);
	193	} else {
	194	pthread_mutex_lock(lock2);
	195	pthread_mutex_lock(lock1);
	196	}
	197	#endif
	198	}
	199
	200	static void sum_ddir(struct thread_data dst, struct thread_data src,
	201	enum fio_ddir ddir)
	202	{
	203	pthread_double_lock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
	204
	205	sum_val(&dst->io_bytes[ddir], &src->io_bytes[ddir]);
	206	sum_val(&dst->io_blocks[ddir], &src->io_blocks[ddir]);
	207	sum_val(&dst->this_io_blocks[ddir], &src->this_io_blocks[ddir]);
	208	sum_val(&dst->this_io_bytes[ddir], &src->this_io_bytes[ddir]);
	209	sum_val(&dst->bytes_done[ddir], &src->bytes_done[ddir]);
	210
	211	pthread_double_unlock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
	212	}
213
214	static void io_workqueue_update_acct_fn(struct submit_worker *sw)
215	{
b86ad8f1	216	struct thread_data *src = sw->priv;
40511301 JA	217	struct thread_data *dst = sw->wq->td;
	218
	219	if (td_read(src))
	220	sum_ddir(dst, src, DDIR_READ);
	221	if (td_write(src))
	222	sum_ddir(dst, src, DDIR_WRITE);
	223	if (td_trim(src))
	224	sum_ddir(dst, src, DDIR_TRIM);
	225
	226	}
	227
103b174e	228	static struct workqueue_ops rated_wq_ops = {
40511301 JA	229	.fn = io_workqueue_fn,
	230	.pre_sleep_flush_fn = io_workqueue_pre_sleep_flush_fn,
	231	.pre_sleep_fn = io_workqueue_pre_sleep_fn,
	232	.update_acct_fn = io_workqueue_update_acct_fn,
	233	.alloc_worker_fn = io_workqueue_alloc_fn,
	234	.free_worker_fn = io_workqueue_free_fn,
	235	.init_worker_fn = io_workqueue_init_worker_fn,
	236	.exit_worker_fn = io_workqueue_exit_worker_fn,
	237	};
103b174e	238
24660963	239	int rate_submit_init(struct thread_data td, struct sk_out sk_out)
103b174e JA	240	{
	241	if (td->o.io_submit_mode != IO_MODE_OFFLOAD)
	242	return 0;
	243
24660963	244	return workqueue_init(td, &td->io_wq, &rated_wq_ops, td->o.iodepth, sk_out);
103b174e JA	245	}
	246
	247	void rate_submit_exit(struct thread_data *td)
	248	{
	249	if (td->o.io_submit_mode != IO_MODE_OFFLOAD)
	250	return;
	251
	252	workqueue_exit(&td->io_wq);
	253	}