[fio.git] / flow.c

#include "fio.h"
#include "fio_sem.h"
#include "smalloc.h"
#include "flist.h"

struct fio_flow {
	unsigned int refs;
	struct flist_head list;
	unsigned int id;
	long long int flow_counter;
};

static struct flist_head *flow_list;
static struct fio_sem *flow_lock;

int flow_threshold_exceeded(struct thread_data *td)
{
	struct fio_flow *flow = td->flow;
	long long flow_counter;

	if (!flow)
		return 0;

	if (td->o.flow > 0)
		flow_counter = flow->flow_counter;
	else
		flow_counter = -flow->flow_counter;

	if (flow_counter > td->o.flow_watermark) {
		if (td->o.flow_sleep) {
			io_u_quiesce(td);
			usleep(td->o.flow_sleep);
		}

		return 1;
	}

	/* No synchronization needed because it doesn't
	 * matter if the flow count is slightly inaccurate */
	flow->flow_counter += td->o.flow;
	return 0;
}

static struct fio_flow *flow_get(unsigned int id)
{
	struct fio_flow *flow = NULL;
	struct flist_head *n;

	if (!flow_lock)
		return NULL;

	fio_sem_down(flow_lock);

	flist_for_each(n, flow_list) {
		flow = flist_entry(n, struct fio_flow, list);
		if (flow->id == id)
			break;

		flow = NULL;
	}

	if (!flow) {
		flow = smalloc(sizeof(*flow));
		if (!flow) {
			fio_sem_up(flow_lock);
			return NULL;
		}
		flow->refs = 0;
		INIT_FLIST_HEAD(&flow->list);
		flow->id = id;
		flow->flow_counter = 0;

		flist_add_tail(&flow->list, flow_list);
	}

	flow->refs++;
	fio_sem_up(flow_lock);
	return flow;
}

static void flow_put(struct fio_flow *flow)
{
	if (!flow_lock)
		return;

	fio_sem_down(flow_lock);

	if (!--flow->refs) {
		flist_del(&flow->list);
		sfree(flow);
	}

	fio_sem_up(flow_lock);
}

void flow_init_job(struct thread_data *td)
{
	if (td->o.flow)
		td->flow = flow_get(td->o.flow_id);
}

void flow_exit_job(struct thread_data *td)
{
	if (td->flow) {
		flow_put(td->flow);
		td->flow = NULL;
	}
}

void flow_init(void)
{
	flow_list = smalloc(sizeof(*flow_list));
	if (!flow_list) {
		log_err("fio: smalloc pool exhausted\n");
		return;
	}

	flow_lock = fio_sem_init(FIO_SEM_UNLOCKED);
	if (!flow_lock) {
		log_err("fio: failed to allocate flow lock\n");
		sfree(flow_list);
		return;
	}

	INIT_FLIST_HEAD(flow_list);
}

void flow_exit(void)
{
	if (flow_lock)
		fio_sem_remove(flow_lock);
	if (flow_list)
		sfree(flow_list);
}
Commit	Line	Data
9e684a49	1	#include "fio.h"
971caeb1	2	#include "fio_sem.h"
9e684a49 DE	3	#include "smalloc.h"
	4	#include "flist.h"
	5
	6	struct fio_flow {
	7	unsigned int refs;
	8	struct flist_head list;
	9	unsigned int id;
	10	long long int flow_counter;
	11	};
	12
	13	static struct flist_head *flow_list;
971caeb1	14	static struct fio_sem *flow_lock;
9e684a49 DE	15
	16	int flow_threshold_exceeded(struct thread_data *td)
	17	{
	18	struct fio_flow *flow = td->flow;
c13a60ce	19	long long flow_counter;
9e684a49 DE	20
	21	if (!flow)
	22	return 0;
	23
c13a60ce JA	24	if (td->o.flow > 0)
	25	flow_counter = flow->flow_counter;
	26	else
	27	flow_counter = -flow->flow_counter;
	28
	29	if (flow_counter > td->o.flow_watermark) {
cb44aa1f JA	30	if (td->o.flow_sleep) {
cb44aa1f JA	31	io_u_quiesce(td);
9e684a49	32	usleep(td->o.flow_sleep);
cb44aa1f JA	33	}
cb44aa1f JA	34
9e684a49 DE	35	return 1;
	36	}
	37
	38	/* No synchronization needed because it doesn't
	39	* matter if the flow count is slightly inaccurate */
	40	flow->flow_counter += td->o.flow;
	41	return 0;
	42	}
	43
	44	static struct fio_flow *flow_get(unsigned int id)
	45	{
8e8b225d	46	struct fio_flow *flow = NULL;
9e684a49 DE	47	struct flist_head *n;
9e684a49 DE	48
fba5c5ff JA	49	if (!flow_lock)
	50	return NULL;
	51
971caeb1	52	fio_sem_down(flow_lock);
9e684a49 DE	53
	54	flist_for_each(n, flow_list) {
	55	flow = flist_entry(n, struct fio_flow, list);
	56	if (flow->id == id)
	57	break;
	58
	59	flow = NULL;
	60	}
	61
	62	if (!flow) {
	63	flow = smalloc(sizeof(*flow));
fba5c5ff	64	if (!flow) {
971caeb1	65	fio_sem_up(flow_lock);
fba5c5ff JA	66	return NULL;
fba5c5ff JA	67	}
9e684a49 DE	68	flow->refs = 0;
	69	INIT_FLIST_HEAD(&flow->list);
	70	flow->id = id;
	71	flow->flow_counter = 0;
	72
	73	flist_add_tail(&flow->list, flow_list);
	74	}
	75
	76	flow->refs++;
971caeb1	77	fio_sem_up(flow_lock);
9e684a49 DE	78	return flow;
	79	}
	80
	81	static void flow_put(struct fio_flow *flow)
	82	{
fba5c5ff JA	83	if (!flow_lock)
	84	return;
	85
971caeb1	86	fio_sem_down(flow_lock);
9e684a49 DE	87
	88	if (!--flow->refs) {
	89	flist_del(&flow->list);
	90	sfree(flow);
	91	}
	92
971caeb1	93	fio_sem_up(flow_lock);
9e684a49 DE	94	}
	95
	96	void flow_init_job(struct thread_data *td)
	97	{
	98	if (td->o.flow)
	99	td->flow = flow_get(td->o.flow_id);
	100	}
	101
	102	void flow_exit_job(struct thread_data *td)
	103	{
	104	if (td->flow) {
	105	flow_put(td->flow);
	106	td->flow = NULL;
	107	}
	108	}
	109
	110	void flow_init(void)
	111	{
9e684a49	112	flow_list = smalloc(sizeof(*flow_list));
fba5c5ff JA	113	if (!flow_list) {
	114	log_err("fio: smalloc pool exhausted\n");
	115	return;
	116	}
	117
971caeb1	118	flow_lock = fio_sem_init(FIO_SEM_UNLOCKED);
fba5c5ff JA	119	if (!flow_lock) {
	120	log_err("fio: failed to allocate flow lock\n");
	121	sfree(flow_list);
	122	return;
	123	}
	124
9e684a49 DE	125	INIT_FLIST_HEAD(flow_list);
	126	}
	127
	128	void flow_exit(void)
	129	{
fba5c5ff	130	if (flow_lock)
971caeb1	131	fio_sem_remove(flow_lock);
fba5c5ff JA	132	if (flow_list)
fba5c5ff JA	133	sfree(flow_list);
9e684a49	134	}