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

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

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;
	double flow_counter_ratio, flow_weight_ratio;

	if (!flow)
		return 0;

	flow_counter_ratio = (double)td->flow_counter /
		atomic_load_relaxed(&flow->flow_counter);
	flow_weight_ratio = (double)td->o.flow /
		atomic_load_relaxed(&flow->total_weight);

	/*
	 * each thread/process executing a fio job will stall based on the
	 * expected  user ratio for a given flow_id group. the idea is to keep
	 * 2 counters, flow and job-specific counter to test if the
	 * ratio between them is proportional to other jobs in the same flow_id
	 */
	if (flow_counter_ratio > flow_weight_ratio) {
		if (td->o.flow_sleep) {
			io_u_quiesce(td);
			usleep(td->o.flow_sleep);
		} else if (td->o.zone_mode == ZONE_MODE_ZBD) {
			io_u_quiesce(td);
		}

		return 1;
	}

	/*
	 * increment flow(shared counter, therefore atomically)
	 * and job-specific counter
	 */
	atomic_add(&flow->flow_counter, 1);
	++td->flow_counter;

	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 = 1;
		flow->total_weight = 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, unsigned long flow_counter,
				        unsigned int weight)
{
	if (!flow_lock)
		return;

	fio_sem_down(flow_lock);

	atomic_sub(&flow->flow_counter, flow_counter);
	atomic_sub(&flow->total_weight, weight);

	if (!--flow->refs) {
		assert(flow->flow_counter == 1);
		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);
		td->flow_counter = 0;
		atomic_add(&td->flow->total_weight, td->o.flow);
	}
}

void flow_exit_job(struct thread_data *td)
{
	if (td->flow) {
		flow_put(td->flow, td->flow_counter, td->o.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);
}