[fio.git] / workqueue.c

/*
 * Generic workqueue offload mechanism
 *
 * Copyright (C) 2015 Jens Axboe <axboe@kernel.dk>
 *
 */
#include <unistd.h>

#include "fio.h"
#include "flist.h"
#include "workqueue.h"

enum {
	SW_F_IDLE	= 1 << 0,
	SW_F_RUNNING	= 1 << 1,
	SW_F_EXIT	= 1 << 2,
	SW_F_EXITED	= 1 << 3,
	SW_F_ACCOUNTED	= 1 << 4,
	SW_F_ERROR	= 1 << 5,
};

static struct submit_worker *__get_submit_worker(struct workqueue *wq,
						 unsigned int start,
						 unsigned int end,
						 struct submit_worker **best)
{
	struct submit_worker *sw = NULL;

	while (start <= end) {
		sw = &wq->workers[start];
		if (sw->flags & SW_F_IDLE)
			return sw;
		if (!(*best) || sw->seq < (*best)->seq)
			*best = sw;
		start++;
	}

	return NULL;
}

static struct submit_worker *get_submit_worker(struct workqueue *wq)
{
	unsigned int next = wq->next_free_worker;
	struct submit_worker *sw, *best = NULL;

	assert(next < wq->max_workers);

	sw = __get_submit_worker(wq, next, wq->max_workers - 1, &best);
	if (!sw && next)
		sw = __get_submit_worker(wq, 0, next - 1, &best);

	/*
	 * No truly idle found, use best match
	 */
	if (!sw)
		sw = best;

	if (sw->index == wq->next_free_worker) {
		if (sw->index + 1 < wq->max_workers)
			wq->next_free_worker = sw->index + 1;
		else
			wq->next_free_worker = 0;
	}

	return sw;
}

static bool all_sw_idle(struct workqueue *wq)
{
	int i;

	for (i = 0; i < wq->max_workers; i++) {
		struct submit_worker *sw = &wq->workers[i];

		if (!(sw->flags & SW_F_IDLE))
			return false;
	}

	return true;
}

/*
 * Must be serialized wrt workqueue_enqueue() by caller
 */
void workqueue_flush(struct workqueue *wq)
{
	wq->wake_idle = 1;

	while (!all_sw_idle(wq)) {
		pthread_mutex_lock(&wq->flush_lock);
		pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
		pthread_mutex_unlock(&wq->flush_lock);
	}

	wq->wake_idle = 0;
}

/*
 * Must be serialized by caller. Returns true for queued, false for busy.
 */
void workqueue_enqueue(struct workqueue *wq, struct workqueue_work *work)
{
	struct submit_worker *sw;

	sw = get_submit_worker(wq);
	assert(sw);

	pthread_mutex_lock(&sw->lock);
	flist_add_tail(&work->list, &sw->work_list);
	sw->seq = ++wq->work_seq;
	sw->flags &= ~SW_F_IDLE;
	pthread_mutex_unlock(&sw->lock);

	pthread_cond_signal(&sw->cond);
}

static void handle_list(struct submit_worker *sw, struct flist_head *list)
{
	struct workqueue *wq = sw->wq;
	struct workqueue_work *work;

	while (!flist_empty(list)) {
		work = flist_first_entry(list, struct workqueue_work, list);
		flist_del_init(&work->list);
		wq->ops.fn(sw, work);
	}
}

static void *worker_thread(void *data)
{
	struct submit_worker *sw = data;
	struct workqueue *wq = sw->wq;
	unsigned int eflags = 0, ret = 0;
	FLIST_HEAD(local_list);

	if (wq->ops.nice) {
		if (nice(wq->ops.nice) < 0) {
			log_err("workqueue: nice %s\n", strerror(errno));
			ret = 1;
		}
	}

	if (!ret)
		ret = workqueue_init_worker(sw);

	pthread_mutex_lock(&sw->lock);
	sw->flags |= SW_F_RUNNING;
	if (ret)
		sw->flags |= SW_F_ERROR;
	pthread_mutex_unlock(&sw->lock);

	pthread_mutex_lock(&wq->flush_lock);
	pthread_cond_signal(&wq->flush_cond);
	pthread_mutex_unlock(&wq->flush_lock);

	if (sw->flags & SW_F_ERROR)
		goto done;

	while (1) {
		pthread_mutex_lock(&sw->lock);

		if (flist_empty(&sw->work_list)) {
			if (sw->flags & SW_F_EXIT) {
				pthread_mutex_unlock(&sw->lock);
				break;
			}

			if (workqueue_pre_sleep_check(sw)) {
				pthread_mutex_unlock(&sw->lock);
				workqueue_pre_sleep(sw);
				pthread_mutex_lock(&sw->lock);
			}

			/*
			 * We dropped and reaquired the lock, check
			 * state again.
			 */
			if (!flist_empty(&sw->work_list))
				goto handle_work;

			if (sw->flags & SW_F_EXIT) {
				pthread_mutex_unlock(&sw->lock);
				break;
			} else if (!(sw->flags & SW_F_IDLE)) {
				sw->flags |= SW_F_IDLE;
				wq->next_free_worker = sw->index;
				if (wq->wake_idle)
					pthread_cond_signal(&wq->flush_cond);
			}
			if (wq->ops.update_acct_fn)
				wq->ops.update_acct_fn(sw);

			pthread_cond_wait(&sw->cond, &sw->lock);
		} else {
handle_work:
			flist_splice_init(&sw->work_list, &local_list);
		}
		pthread_mutex_unlock(&sw->lock);
		handle_list(sw, &local_list);
	}

	if (wq->ops.update_acct_fn)
		wq->ops.update_acct_fn(sw);

done:
	pthread_mutex_lock(&sw->lock);
	sw->flags |= (SW_F_EXITED | eflags);
	pthread_mutex_unlock(&sw->lock);
	return NULL;
}

static void free_worker(struct submit_worker *sw, unsigned int *sum_cnt)
{
	struct workqueue *wq = sw->wq;

	workqueue_exit_worker(sw, sum_cnt);

	pthread_cond_destroy(&sw->cond);
	pthread_mutex_destroy(&sw->lock);

	if (wq->ops.free_worker_fn)
		wq->ops.free_worker_fn(sw);
}

static void shutdown_worker(struct submit_worker *sw, unsigned int *sum_cnt)
{
	pthread_join(sw->thread, NULL);
	free_worker(sw, sum_cnt);
}

void workqueue_exit(struct workqueue *wq)
{
	unsigned int shutdown, sum_cnt = 0;
	struct submit_worker *sw;
	int i;

	for (i = 0; i < wq->max_workers; i++) {
		sw = &wq->workers[i];

		pthread_mutex_lock(&sw->lock);
		sw->flags |= SW_F_EXIT;
		pthread_cond_signal(&sw->cond);
		pthread_mutex_unlock(&sw->lock);
	}

	do {
		shutdown = 0;
		for (i = 0; i < wq->max_workers; i++) {
			sw = &wq->workers[i];
			if (sw->flags & SW_F_ACCOUNTED)
				continue;
			pthread_mutex_lock(&sw->lock);
			sw->flags |= SW_F_ACCOUNTED;
			pthread_mutex_unlock(&sw->lock);
			shutdown_worker(sw, &sum_cnt);
			shutdown++;
		}
	} while (shutdown && shutdown != wq->max_workers);

	free(wq->workers);
	pthread_mutex_destroy(&wq->flush_lock);
	pthread_cond_destroy(&wq->flush_cond);
	pthread_mutex_destroy(&wq->stat_lock);
}

static int start_worker(struct workqueue *wq, unsigned int index)
{
	struct submit_worker *sw = &wq->workers[index];
	int ret;

	INIT_FLIST_HEAD(&sw->work_list);
	pthread_cond_init(&sw->cond, NULL);
	pthread_mutex_init(&sw->lock, NULL);
	sw->wq = wq;
	sw->index = index;

	if (wq->ops.alloc_worker_fn) {
		ret = wq->ops.alloc_worker_fn(sw);
		if (ret)
			return ret;
	}

	ret = pthread_create(&sw->thread, NULL, worker_thread, sw);
	if (!ret) {
		pthread_mutex_lock(&sw->lock);
		sw->flags = SW_F_IDLE;
		pthread_mutex_unlock(&sw->lock);
		return 0;
	}

	free_worker(sw, NULL);
	return 1;
}

int workqueue_init(struct thread_data *td, struct workqueue *wq,
		   struct workqueue_ops *ops, unsigned max_pending)
{
	unsigned int running;
	int i, error;

	wq->max_workers = max_pending;
	wq->td = td;
	wq->ops = *ops;
	wq->work_seq = 0;
	wq->next_free_worker = 0;
	pthread_cond_init(&wq->flush_cond, NULL);
	pthread_mutex_init(&wq->flush_lock, NULL);
	pthread_mutex_init(&wq->stat_lock, NULL);

	wq->workers = calloc(wq->max_workers, sizeof(struct submit_worker));

	for (i = 0; i < wq->max_workers; i++)
		if (start_worker(wq, i))
			break;

	wq->max_workers = i;
	if (!wq->max_workers)
		goto err;

	/*
	 * Wait for them all to be started and initialized
	 */
	error = 0;
	do {
		struct submit_worker *sw;

		running = 0;
		pthread_mutex_lock(&wq->flush_lock);
		for (i = 0; i < wq->max_workers; i++) {
			sw = &wq->workers[i];
			pthread_mutex_lock(&sw->lock);
			if (sw->flags & SW_F_RUNNING)
				running++;
			if (sw->flags & SW_F_ERROR)
				error++;
			pthread_mutex_unlock(&sw->lock);
		}

		if (error || running == wq->max_workers) {
			pthread_mutex_unlock(&wq->flush_lock);
			break;
		}

		pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
		pthread_mutex_unlock(&wq->flush_lock);
	} while (1);

	if (!error)
		return 0;

err:
	log_err("Can't create rate workqueue\n");
	td_verror(td, ESRCH, "workqueue_init");
	workqueue_exit(wq);
	return 1;
}
Commit	Line	Data
	1	/*
	2	* Generic workqueue offload mechanism
	3	*
	4	* Copyright (C) 2015 Jens Axboe <axboe@kernel.dk>
	5	*
	6	*/
	7	#include <unistd.h>
	8
	9	#include "fio.h"
	10	#include "flist.h"
	11	#include "workqueue.h"
	12
	13	enum {
	14	SW_F_IDLE = 1 << 0,
	15	SW_F_RUNNING = 1 << 1,
	16	SW_F_EXIT = 1 << 2,
	17	SW_F_EXITED = 1 << 3,
	18	SW_F_ACCOUNTED = 1 << 4,
	19	SW_F_ERROR = 1 << 5,
	20	};
	21
	22	static struct submit_worker __get_submit_worker(struct workqueue wq,
	23	unsigned int start,
	24	unsigned int end,
	25	struct submit_worker **best)
	26	{
	27	struct submit_worker *sw = NULL;
	28
	29	while (start <= end) {
	30	sw = &wq->workers[start];
	31	if (sw->flags & SW_F_IDLE)
	32	return sw;
	33	if (!(best) \|\| sw->seq < (best)->seq)
	34	*best = sw;
	35	start++;
	36	}
	37
	38	return NULL;
	39	}
	40
	41	static struct submit_worker get_submit_worker(struct workqueue wq)
	42	{
	43	unsigned int next = wq->next_free_worker;
	44	struct submit_worker sw, best = NULL;
	45
	46	assert(next < wq->max_workers);
	47
	48	sw = __get_submit_worker(wq, next, wq->max_workers - 1, &best);
	49	if (!sw && next)
	50	sw = __get_submit_worker(wq, 0, next - 1, &best);
	51
	52	/*
	53	* No truly idle found, use best match
	54	*/
	55	if (!sw)
	56	sw = best;
	57
	58	if (sw->index == wq->next_free_worker) {
	59	if (sw->index + 1 < wq->max_workers)
	60	wq->next_free_worker = sw->index + 1;
	61	else
	62	wq->next_free_worker = 0;
	63	}
	64
	65	return sw;
	66	}
	67
	68	static bool all_sw_idle(struct workqueue *wq)
	69	{
	70	int i;
	71
	72	for (i = 0; i < wq->max_workers; i++) {
	73	struct submit_worker *sw = &wq->workers[i];
	74
	75	if (!(sw->flags & SW_F_IDLE))
	76	return false;
	77	}
	78
	79	return true;
	80	}
	81
	82	/*
	83	* Must be serialized wrt workqueue_enqueue() by caller
	84	*/
	85	void workqueue_flush(struct workqueue *wq)
	86	{
	87	wq->wake_idle = 1;
	88
	89	while (!all_sw_idle(wq)) {
	90	pthread_mutex_lock(&wq->flush_lock);
	91	pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
	92	pthread_mutex_unlock(&wq->flush_lock);
	93	}
	94
	95	wq->wake_idle = 0;
	96	}
	97
	98	/*
	99	* Must be serialized by caller. Returns true for queued, false for busy.
	100	*/
	101	void workqueue_enqueue(struct workqueue wq, struct workqueue_work work)
	102	{
	103	struct submit_worker *sw;
	104
	105	sw = get_submit_worker(wq);
	106	assert(sw);
	107
	108	pthread_mutex_lock(&sw->lock);
	109	flist_add_tail(&work->list, &sw->work_list);
	110	sw->seq = ++wq->work_seq;
	111	sw->flags &= ~SW_F_IDLE;
	112	pthread_mutex_unlock(&sw->lock);
	113
	114	pthread_cond_signal(&sw->cond);
	115	}
	116
	117	static void handle_list(struct submit_worker sw, struct flist_head list)
	118	{
	119	struct workqueue *wq = sw->wq;
	120	struct workqueue_work *work;
	121
	122	while (!flist_empty(list)) {
	123	work = flist_first_entry(list, struct workqueue_work, list);
	124	flist_del_init(&work->list);
	125	wq->ops.fn(sw, work);
	126	}
	127	}
	128
	129	static void worker_thread(void data)
	130	{
	131	struct submit_worker *sw = data;
	132	struct workqueue *wq = sw->wq;
	133	unsigned int eflags = 0, ret = 0;
	134	FLIST_HEAD(local_list);
	135
	136	if (wq->ops.nice) {
	137	if (nice(wq->ops.nice) < 0) {
	138	log_err("workqueue: nice %s\n", strerror(errno));
	139	ret = 1;
	140	}
	141	}
	142
	143	if (!ret)
	144	ret = workqueue_init_worker(sw);
	145
	146	pthread_mutex_lock(&sw->lock);
	147	sw->flags \|= SW_F_RUNNING;
	148	if (ret)
	149	sw->flags \|= SW_F_ERROR;
	150	pthread_mutex_unlock(&sw->lock);
	151
	152	pthread_mutex_lock(&wq->flush_lock);
	153	pthread_cond_signal(&wq->flush_cond);
	154	pthread_mutex_unlock(&wq->flush_lock);
	155
	156	if (sw->flags & SW_F_ERROR)
	157	goto done;
	158
	159	while (1) {
	160	pthread_mutex_lock(&sw->lock);
	161
	162	if (flist_empty(&sw->work_list)) {
	163	if (sw->flags & SW_F_EXIT) {
	164	pthread_mutex_unlock(&sw->lock);
	165	break;
	166	}
	167
	168	if (workqueue_pre_sleep_check(sw)) {
	169	pthread_mutex_unlock(&sw->lock);
	170	workqueue_pre_sleep(sw);
	171	pthread_mutex_lock(&sw->lock);
	172	}
	173
	174	/*
	175	* We dropped and reaquired the lock, check
	176	* state again.
	177	*/
	178	if (!flist_empty(&sw->work_list))
	179	goto handle_work;
	180
	181	if (sw->flags & SW_F_EXIT) {
	182	pthread_mutex_unlock(&sw->lock);
	183	break;
	184	} else if (!(sw->flags & SW_F_IDLE)) {
	185	sw->flags \|= SW_F_IDLE;
	186	wq->next_free_worker = sw->index;
	187	if (wq->wake_idle)
	188	pthread_cond_signal(&wq->flush_cond);
	189	}
	190	if (wq->ops.update_acct_fn)
	191	wq->ops.update_acct_fn(sw);
	192
	193	pthread_cond_wait(&sw->cond, &sw->lock);
	194	} else {
	195	handle_work:
	196	flist_splice_init(&sw->work_list, &local_list);
	197	}
	198	pthread_mutex_unlock(&sw->lock);
	199	handle_list(sw, &local_list);
	200	}
	201
	202	if (wq->ops.update_acct_fn)
	203	wq->ops.update_acct_fn(sw);
	204
	205	done:
	206	pthread_mutex_lock(&sw->lock);
	207	sw->flags \|= (SW_F_EXITED \| eflags);
	208	pthread_mutex_unlock(&sw->lock);
	209	return NULL;
	210	}
	211
	212	static void free_worker(struct submit_worker sw, unsigned int sum_cnt)
	213	{
	214	struct workqueue *wq = sw->wq;
	215
	216	workqueue_exit_worker(sw, sum_cnt);
	217
	218	pthread_cond_destroy(&sw->cond);
	219	pthread_mutex_destroy(&sw->lock);
	220
	221	if (wq->ops.free_worker_fn)
	222	wq->ops.free_worker_fn(sw);
	223	}
	224
	225	static void shutdown_worker(struct submit_worker sw, unsigned int sum_cnt)
	226	{
	227	pthread_join(sw->thread, NULL);
	228	free_worker(sw, sum_cnt);
	229	}
	230
	231	void workqueue_exit(struct workqueue *wq)
	232	{
	233	unsigned int shutdown, sum_cnt = 0;
	234	struct submit_worker *sw;
	235	int i;
	236
	237	for (i = 0; i < wq->max_workers; i++) {
	238	sw = &wq->workers[i];
	239
	240	pthread_mutex_lock(&sw->lock);
	241	sw->flags \|= SW_F_EXIT;
	242	pthread_cond_signal(&sw->cond);
	243	pthread_mutex_unlock(&sw->lock);
	244	}
	245
	246	do {
	247	shutdown = 0;
	248	for (i = 0; i < wq->max_workers; i++) {
	249	sw = &wq->workers[i];
	250	if (sw->flags & SW_F_ACCOUNTED)
	251	continue;
	252	pthread_mutex_lock(&sw->lock);
	253	sw->flags \|= SW_F_ACCOUNTED;
	254	pthread_mutex_unlock(&sw->lock);
	255	shutdown_worker(sw, &sum_cnt);
	256	shutdown++;
	257	}
	258	} while (shutdown && shutdown != wq->max_workers);
	259
	260	free(wq->workers);
	261	pthread_mutex_destroy(&wq->flush_lock);
	262	pthread_cond_destroy(&wq->flush_cond);
	263	pthread_mutex_destroy(&wq->stat_lock);
	264	}
	265
	266	static int start_worker(struct workqueue *wq, unsigned int index)
	267	{
	268	struct submit_worker *sw = &wq->workers[index];
	269	int ret;
	270
	271	INIT_FLIST_HEAD(&sw->work_list);
	272	pthread_cond_init(&sw->cond, NULL);
	273	pthread_mutex_init(&sw->lock, NULL);
	274	sw->wq = wq;
	275	sw->index = index;
	276
	277	if (wq->ops.alloc_worker_fn) {
	278	ret = wq->ops.alloc_worker_fn(sw);
	279	if (ret)
	280	return ret;
	281	}
	282
	283	ret = pthread_create(&sw->thread, NULL, worker_thread, sw);
	284	if (!ret) {
	285	pthread_mutex_lock(&sw->lock);
	286	sw->flags = SW_F_IDLE;
	287	pthread_mutex_unlock(&sw->lock);
	288	return 0;
	289	}
	290
	291	free_worker(sw, NULL);
	292	return 1;
	293	}
	294
	295	int workqueue_init(struct thread_data td, struct workqueue wq,
	296	struct workqueue_ops *ops, unsigned max_pending)
	297	{
	298	unsigned int running;
	299	int i, error;
	300
	301	wq->max_workers = max_pending;
	302	wq->td = td;
	303	wq->ops = *ops;
	304	wq->work_seq = 0;
	305	wq->next_free_worker = 0;
	306	pthread_cond_init(&wq->flush_cond, NULL);
	307	pthread_mutex_init(&wq->flush_lock, NULL);
	308	pthread_mutex_init(&wq->stat_lock, NULL);
	309
	310	wq->workers = calloc(wq->max_workers, sizeof(struct submit_worker));
	311
	312	for (i = 0; i < wq->max_workers; i++)
	313	if (start_worker(wq, i))
	314	break;
	315
	316	wq->max_workers = i;
	317	if (!wq->max_workers)
	318	goto err;
	319
	320	/*
	321	* Wait for them all to be started and initialized
	322	*/
	323	error = 0;
	324	do {
	325	struct submit_worker *sw;
	326
	327	running = 0;
	328	pthread_mutex_lock(&wq->flush_lock);
	329	for (i = 0; i < wq->max_workers; i++) {
	330	sw = &wq->workers[i];
	331	pthread_mutex_lock(&sw->lock);
	332	if (sw->flags & SW_F_RUNNING)
	333	running++;
	334	if (sw->flags & SW_F_ERROR)
	335	error++;
	336	pthread_mutex_unlock(&sw->lock);
	337	}
	338
	339	if (error \|\| running == wq->max_workers) {
	340	pthread_mutex_unlock(&wq->flush_lock);
	341	break;
	342	}
	343
	344	pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
	345	pthread_mutex_unlock(&wq->flush_lock);
	346	} while (1);
	347
	348	if (!error)
	349	return 0;
	350
	351	err:
	352	log_err("Can't create rate workqueue\n");
	353	td_verror(td, ESRCH, "workqueue_init");
	354	workqueue_exit(wq);
	355	return 1;
	356	}