[fio.git] / rate-submit.c

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

static void check_overlap(struct io_u *io_u)
{
	int i;
	struct thread_data *td;
	bool overlap = false;

	do {
		/*
		 * Allow only one thread to check for overlap at a
		 * time to prevent two threads from thinking the coast
		 * is clear and then submitting IOs that overlap with
		 * each other
		 *
		 * If an overlap is found, release the lock and
		 * re-acquire it before checking again to give other
		 * threads a chance to make progress
		 *
		 * If an overlap is not found, release the lock when the
		 * io_u's IO_U_F_FLIGHT flag is set so that this io_u
		 * can be checked by other threads as they assess overlap
		 */
		pthread_mutex_lock(&overlap_check);
		for_each_td(td, i) {
			if (td->runstate <= TD_SETTING_UP ||
				td->runstate >= TD_FINISHING ||
				!td->o.serialize_overlap ||
				td->o.io_submit_mode != IO_MODE_OFFLOAD)
				continue;

			overlap = in_flight_overlap(&td->io_u_all, io_u);
			if (overlap) {
				pthread_mutex_unlock(&overlap_check);
				break;
			}
		}
	} while (overlap);
}

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;

	if (td->o.serialize_overlap)
		check_overlap(io_u);

	dprint(FD_RATE, "io_u %p queued by %u\n", io_u, gettid());

	io_u_set(td, 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(td, 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;
	}

	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;

	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;

	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);
	td->io_hist_tree = RB_ROOT;

	td->o.iodepth = 1;
	if (td_io_init(td))
		goto err_io_init;

	set_epoch_time(td, td->o.log_unix_epoch);
	fio_getrusage(&td->ru_start);
	clear_io_state(td, 1);

	td_set_runstate(td, TD_RUNNING);
	td->flags |= TD_F_CHILD | TD_F_NEED_LOCK;
	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	7	#include "fio.h"
618ee94c	8	#include "ioengines.h"
40511301	9	#include "lib/getrusage.h"
24660963	10	#include "rate-submit.h"
40511301	11
c06379a6 VF	12	static void check_overlap(struct io_u *io_u)
	13	{
	14	int i;
	15	struct thread_data *td;
	16	bool overlap = false;
	17
	18	do {
	19	/*
	20	* Allow only one thread to check for overlap at a
	21	* time to prevent two threads from thinking the coast
	22	* is clear and then submitting IOs that overlap with
	23	* each other
817ae977 VF	24	*
	25	* If an overlap is found, release the lock and
	26	* re-acquire it before checking again to give other
	27	* threads a chance to make progress
	28	*
	29	* If an overlap is not found, release the lock when the
	30	* io_u's IO_U_F_FLIGHT flag is set so that this io_u
	31	* can be checked by other threads as they assess overlap
c06379a6 VF	32	*/
	33	pthread_mutex_lock(&overlap_check);
	34	for_each_td(td, i) {
	35	if (td->runstate <= TD_SETTING_UP \|\|
	36	td->runstate >= TD_FINISHING \|\|
	37	!td->o.serialize_overlap \|\|
	38	td->o.io_submit_mode != IO_MODE_OFFLOAD)
	39	continue;
	40
	41	overlap = in_flight_overlap(&td->io_u_all, io_u);
	42	if (overlap) {
	43	pthread_mutex_unlock(&overlap_check);
	44	break;
	45	}
	46	}
	47	} while (overlap);
	48	}
	49
155f2f02 JA	50	static int io_workqueue_fn(struct submit_worker *sw,
155f2f02 JA	51	struct workqueue_work *work)
40511301 JA	52	{
	53	struct io_u *io_u = container_of(work, struct io_u, work);
	54	const enum fio_ddir ddir = io_u->ddir;
b86ad8f1	55	struct thread_data *td = sw->priv;
40511301 JA	56	int ret;
40511301 JA	57
c06379a6 VF	58	if (td->o.serialize_overlap)
	59	check_overlap(io_u);
	60
40511301 JA	61	dprint(FD_RATE, "io_u %p queued by %u\n", io_u, gettid());
40511301 JA	62
1651e431	63	io_u_set(td, io_u, IO_U_F_NO_FILE_PUT);
40511301 JA	64
	65	td->cur_depth++;
	66
	67	do {
	68	ret = td_io_queue(td, io_u);
	69	if (ret != FIO_Q_BUSY)
	70	break;
	71	ret = io_u_queued_complete(td, 1);
	72	if (ret > 0)
	73	td->cur_depth -= ret;
1651e431	74	io_u_clear(td, io_u, IO_U_F_FLIGHT);
40511301 JA	75	} while (1);
	76
	77	dprint(FD_RATE, "io_u %p ret %d by %u\n", io_u, ret, gettid());
	78
	79	io_queue_event(td, io_u, &ret, ddir, NULL, 0, NULL);
	80
	81	if (ret == FIO_Q_COMPLETED)
	82	td->cur_depth--;
	83	else if (ret == FIO_Q_QUEUED) {
	84	unsigned int min_evts;
	85
	86	if (td->o.iodepth == 1)
	87	min_evts = 1;
	88	else
	89	min_evts = 0;
	90
	91	ret = io_u_queued_complete(td, min_evts);
	92	if (ret > 0)
	93	td->cur_depth -= ret;
40511301	94	}
155f2f02 JA	95
155f2f02 JA	96	return 0;
40511301 JA	97	}
	98
	99	static bool io_workqueue_pre_sleep_flush_fn(struct submit_worker *sw)
	100	{
b86ad8f1	101	struct thread_data *td = sw->priv;
40511301 JA	102
	103	if (td->io_u_queued \|\| td->cur_depth \|\| td->io_u_in_flight)
	104	return true;
	105
	106	return false;
	107	}
	108
	109	static void io_workqueue_pre_sleep_fn(struct submit_worker *sw)
	110	{
b86ad8f1	111	struct thread_data *td = sw->priv;
40511301 JA	112	int ret;
	113
	114	ret = io_u_quiesce(td);
	115	if (ret > 0)
	116	td->cur_depth -= ret;
	117	}
	118
	119	static int io_workqueue_alloc_fn(struct submit_worker *sw)
	120	{
	121	struct thread_data *td;
	122
	123	td = calloc(1, sizeof(*td));
b86ad8f1	124	sw->priv = td;
40511301 JA	125	return 0;
	126	}
	127
	128	static void io_workqueue_free_fn(struct submit_worker *sw)
	129	{
b86ad8f1 CB	130	free(sw->priv);
b86ad8f1 CB	131	sw->priv = NULL;
40511301 JA	132	}
	133
	134	static int io_workqueue_init_worker_fn(struct submit_worker *sw)
	135	{
	136	struct thread_data *parent = sw->wq->td;
b86ad8f1	137	struct thread_data *td = sw->priv;
40511301 JA	138
	139	memcpy(&td->o, &parent->o, sizeof(td->o));
	140	memcpy(&td->ts, &parent->ts, sizeof(td->ts));
	141	td->o.uid = td->o.gid = -1U;
	142	dup_files(td, parent);
	143	td->eo = parent->eo;
	144	fio_options_mem_dupe(td);
	145
	146	if (ioengine_load(td))
	147	goto err;
	148
40511301 JA	149	td->pid = gettid();
	150
	151	INIT_FLIST_HEAD(&td->io_log_list);
	152	INIT_FLIST_HEAD(&td->io_hist_list);
	153	INIT_FLIST_HEAD(&td->verify_list);
	154	INIT_FLIST_HEAD(&td->trim_list);
40511301 JA	155	td->io_hist_tree = RB_ROOT;
	156
	157	td->o.iodepth = 1;
	158	if (td_io_init(td))
	159	goto err_io_init;
	160
3aea75b1	161	set_epoch_time(td, td->o.log_unix_epoch);
40511301 JA	162	fio_getrusage(&td->ru_start);
	163	clear_io_state(td, 1);
	164
	165	td_set_runstate(td, TD_RUNNING);
b7aae4ba	166	td->flags \|= TD_F_CHILD \| TD_F_NEED_LOCK;
40511301 JA	167	td->parent = parent;
	168	return 0;
	169
	170	err_io_init:
	171	close_ioengine(td);
	172	err:
	173	return 1;
	174
	175	}
	176
	177	static void io_workqueue_exit_worker_fn(struct submit_worker *sw,
	178	unsigned int *sum_cnt)
	179	{
b86ad8f1	180	struct thread_data *td = sw->priv;
40511301 JA	181
	182	(*sum_cnt)++;
	183	sum_thread_stats(&sw->wq->td->ts, &td->ts, *sum_cnt == 1);
	184
	185	fio_options_free(td);
	186	close_and_free_files(td);
	187	if (td->io_ops)
	188	close_ioengine(td);
	189	td_set_runstate(td, TD_EXITED);
	190	}
	191
	192	#ifdef CONFIG_SFAA
	193	static void sum_val(uint64_t dst, uint64_t src)
	194	{
	195	if (*src) {
	196	__sync_fetch_and_add(dst, *src);
	197	*src = 0;
	198	}
	199	}
	200	#else
	201	static void sum_val(uint64_t dst, uint64_t src)
	202	{
	203	if (*src) {
	204	dst += src;
	205	*src = 0;
	206	}
	207	}
	208	#endif
	209
	210	static void pthread_double_unlock(pthread_mutex_t *lock1,
	211	pthread_mutex_t *lock2)
	212	{
	213	#ifndef CONFIG_SFAA
	214	pthread_mutex_unlock(lock1);
	215	pthread_mutex_unlock(lock2);
	216	#endif
	217	}
	218
	219	static void pthread_double_lock(pthread_mutex_t lock1, pthread_mutex_t lock2)
	220	{
	221	#ifndef CONFIG_SFAA
	222	if (lock1 < lock2) {
	223	pthread_mutex_lock(lock1);
	224	pthread_mutex_lock(lock2);
	225	} else {
	226	pthread_mutex_lock(lock2);
	227	pthread_mutex_lock(lock1);
	228	}
	229	#endif
	230	}
	231
	232	static void sum_ddir(struct thread_data dst, struct thread_data src,
	233	enum fio_ddir ddir)
	234	{
	235	pthread_double_lock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
	236
	237	sum_val(&dst->io_bytes[ddir], &src->io_bytes[ddir]);
	238	sum_val(&dst->io_blocks[ddir], &src->io_blocks[ddir]);
	239	sum_val(&dst->this_io_blocks[ddir], &src->this_io_blocks[ddir]);
	240	sum_val(&dst->this_io_bytes[ddir], &src->this_io_bytes[ddir]);
	241	sum_val(&dst->bytes_done[ddir], &src->bytes_done[ddir]);
	242
	243	pthread_double_unlock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
	244	}
245
246	static void io_workqueue_update_acct_fn(struct submit_worker *sw)
247	{
b86ad8f1	248	struct thread_data *src = sw->priv;
40511301 JA	249	struct thread_data *dst = sw->wq->td;
	250
	251	if (td_read(src))
	252	sum_ddir(dst, src, DDIR_READ);
	253	if (td_write(src))
	254	sum_ddir(dst, src, DDIR_WRITE);
	255	if (td_trim(src))
	256	sum_ddir(dst, src, DDIR_TRIM);
	257
	258	}
	259
103b174e	260	static struct workqueue_ops rated_wq_ops = {
40511301 JA	261	.fn = io_workqueue_fn,
	262	.pre_sleep_flush_fn = io_workqueue_pre_sleep_flush_fn,
	263	.pre_sleep_fn = io_workqueue_pre_sleep_fn,
	264	.update_acct_fn = io_workqueue_update_acct_fn,
	265	.alloc_worker_fn = io_workqueue_alloc_fn,
	266	.free_worker_fn = io_workqueue_free_fn,
	267	.init_worker_fn = io_workqueue_init_worker_fn,
	268	.exit_worker_fn = io_workqueue_exit_worker_fn,
	269	};
103b174e	270
24660963	271	int rate_submit_init(struct thread_data td, struct sk_out sk_out)
103b174e JA	272	{
	273	if (td->o.io_submit_mode != IO_MODE_OFFLOAD)
	274	return 0;
	275
24660963	276	return workqueue_init(td, &td->io_wq, &rated_wq_ops, td->o.iodepth, sk_out);
103b174e JA	277	}
	278
	279	void rate_submit_exit(struct thread_data *td)
	280	{
	281	if (td->o.io_submit_mode != IO_MODE_OFFLOAD)
	282	return;
	283
	284	workqueue_exit(&td->io_wq);
	285	}