[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"
#include "smalloc.h"
#include "pshared.h"

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

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)
{
	pthread_mutex_lock(&wq->flush_lock);
	wq->wake_idle = 1;

	while (!all_sw_idle(wq))
		pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);

	wq->wake_idle = 0;
	pthread_mutex_unlock(&wq->flush_lock);
}

/*
 * Must be serialized by caller.
 */
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_cond_signal(&sw->cond);
	pthread_mutex_unlock(&sw->lock);
}

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 ret = 0;
	FLIST_HEAD(local_list);

	sk_out_assign(sw->sk_out);

	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;

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

			if (workqueue_pre_sleep_check(sw)) {
				pthread_mutex_unlock(&sw->lock);
				workqueue_pre_sleep(sw);
				pthread_mutex_lock(&sw->lock);
			}
		}
		/*
		 * We may have dropped and reaquired the lock, check state
		 * again.
		 */
		if (flist_empty(&sw->work_list)) {
			if (sw->flags & SW_F_EXIT) {
				break;
			}
			if (!(sw->flags & SW_F_IDLE)) {
				sw->flags |= SW_F_IDLE;
				wq->next_free_worker = sw->index;
				pthread_mutex_unlock(&sw->lock);
				pthread_mutex_lock(&wq->flush_lock);
				if (wq->wake_idle)
					pthread_cond_signal(&wq->flush_cond);
				pthread_mutex_unlock(&wq->flush_lock);
				pthread_mutex_lock(&sw->lock);
			}
		}
		if (flist_empty(&sw->work_list)) {
			if (sw->flags & SW_F_EXIT) {
				break;
			}
			pthread_cond_wait(&sw->cond, &sw->lock);
		} else {
			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);
		pthread_mutex_lock(&sw->lock);
	}
	pthread_mutex_unlock(&sw->lock);

done:
	sk_out_drop();
	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;

	if (!wq->workers)
		return;

	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);

	sfree(wq->workers);
	wq->workers = NULL;
	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 sk_out *sk_out)
{
	struct submit_worker *sw = &wq->workers[index];
	int ret;

	INIT_FLIST_HEAD(&sw->work_list);

	ret = mutex_cond_init_pshared(&sw->lock, &sw->cond);
	if (ret)
		return ret;

	sw->wq = wq;
	sw->index = index;
	sw->sk_out = sk_out;

	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 int max_workers,
		   struct sk_out *sk_out)
{
	unsigned int running;
	int i, error;
	int ret;

	wq->max_workers = max_workers;
	wq->td = td;
	wq->ops = *ops;
	wq->work_seq = 0;
	wq->next_free_worker = 0;

	ret = mutex_cond_init_pshared(&wq->flush_lock, &wq->flush_cond);
	if (ret)
		goto err;
	ret = mutex_init_pshared(&wq->stat_lock);
	if (ret)
		goto err;

	wq->workers = smalloc(wq->max_workers * sizeof(struct submit_worker));
	if (!wq->workers)
		goto err;

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

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

	/*
	 * Wait for them all to be started and initialized
	 */
	error = 0;
	pthread_mutex_lock(&wq->flush_lock);
	do {
		struct submit_worker *sw;

		running = 0;
		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)
			break;

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

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