[fio.git] / workqueue.c

/*
 * Rated submission helpers
 *
 * Copyright (C) 2015 Jens Axboe <axboe@kernel.dk>
 *
 */
#include <unistd.h>

#include "fio.h"
#include "flist.h"
#include "workqueue.h"
#include "lib/getrusage.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.
 */
bool workqueue_enqueue(struct workqueue *wq, struct workqueue_work *work)
{
	struct submit_worker *sw;

	sw = get_submit_worker(wq);
	if (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);
		return true;
	}

	return false;
}

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

#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 update_accounting(struct submit_worker *sw)
{
	struct thread_data *src = sw->private;
	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 void *worker_thread(void *data)
{
	struct submit_worker *sw = data;
	struct workqueue *wq = sw->wq;
	unsigned int eflags = 0, ret;
	FLIST_HEAD(local_list);

	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);
			}
			update_accounting(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);
	}

	update_accounting(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)
{
	struct workqueue *wq = sw->wq;

	workqueue_exit_worker(sw);

	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)
{
	struct thread_data *parent = sw->wq->td;
	struct thread_data *td = sw->private;

	pthread_join(sw->thread, NULL);
	(*sum_cnt)++;
	sum_thread_stats(&parent->ts, &td->ts, *sum_cnt == 1);
	free_worker(sw);
}

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);
	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
a9da8ab2 JA	1	/*
	2	* Rated submission helpers
	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"
	11	#include "workqueue.h"
	12	#include "lib/getrusage.h"
	13
a9da8ab2 JA	14	enum {
	15	SW_F_IDLE = 1 << 0,
	16	SW_F_RUNNING = 1 << 1,
	17	SW_F_EXIT = 1 << 2,
	18	SW_F_EXITED = 1 << 3,
	19	SW_F_ACCOUNTED = 1 << 4,
	20	SW_F_ERROR = 1 << 5,
	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	{
	88	wq->wake_idle = 1;
	89
	90	while (!all_sw_idle(wq)) {
	91	pthread_mutex_lock(&wq->flush_lock);
	92	pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
	93	pthread_mutex_unlock(&wq->flush_lock);
	94	}
	95
	96	wq->wake_idle = 0;
	97	}
	98
	99	/*
b07f6ad1	100	* Must be serialized by caller. Returns true for queued, false for busy.
a9da8ab2	101	*/
88271841	102	bool workqueue_enqueue(struct workqueue wq, struct workqueue_work work)
a9da8ab2 JA	103	{
	104	struct submit_worker *sw;
	105
	106	sw = get_submit_worker(wq);
	107	if (sw) {
a9da8ab2	108	pthread_mutex_lock(&sw->lock);
88271841	109	flist_add_tail(&work->list, &sw->work_list);
a9da8ab2 JA	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);
b07f6ad1	115	return true;
a9da8ab2 JA	116	}
a9da8ab2 JA	117
b07f6ad1	118	return false;
a9da8ab2 JA	119	}
	120
	121	static void handle_list(struct submit_worker sw, struct flist_head list)
	122	{
	123	struct workqueue *wq = sw->wq;
88271841	124	struct workqueue_work *work;
a9da8ab2 JA	125
a9da8ab2 JA	126	while (!flist_empty(list)) {
88271841 JA	127	work = flist_first_entry(list, struct workqueue_work, list);
88271841 JA	128	flist_del_init(&work->list);
ee2b6d6e	129	wq->ops.fn(sw, work);
a9da8ab2 JA	130	}
	131	}
	132
f5cd2907	133	#ifdef CONFIG_SFAA
a9da8ab2 JA	134	static void sum_val(uint64_t dst, uint64_t src)
	135	{
	136	if (*src) {
	137	__sync_fetch_and_add(dst, *src);
	138	*src = 0;
	139	}
	140	}
2a274336 JA	141	#else
	142	static void sum_val(uint64_t dst, uint64_t src)
	143	{
	144	if (*src) {
	145	dst += src;
	146	*src = 0;
	147	}
	148	}
	149	#endif
a9da8ab2	150
a6a3469e JA	151	static void pthread_double_unlock(pthread_mutex_t *lock1,
a6a3469e JA	152	pthread_mutex_t *lock2)
a9da8ab2	153	{
f5cd2907	154	#ifndef CONFIG_SFAA
a6a3469e JA	155	pthread_mutex_unlock(lock1);
a6a3469e JA	156	pthread_mutex_unlock(lock2);
2a274336	157	#endif
a6a3469e JA	158	}
	159
	160	static void pthread_double_lock(pthread_mutex_t lock1, pthread_mutex_t lock2)
	161	{
	162	#ifndef CONFIG_SFAA
	163	if (lock1 < lock2) {
	164	pthread_mutex_lock(lock1);
	165	pthread_mutex_lock(lock2);
	166	} else {
	167	pthread_mutex_lock(lock2);
	168	pthread_mutex_lock(lock1);
	169	}
	170	#endif
	171	}
	172
	173	static void sum_ddir(struct thread_data dst, struct thread_data src,
	174	enum fio_ddir ddir)
	175	{
	176	pthread_double_lock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
2a274336	177
a9da8ab2 JA	178	sum_val(&dst->io_bytes[ddir], &src->io_bytes[ddir]);
	179	sum_val(&dst->io_blocks[ddir], &src->io_blocks[ddir]);
	180	sum_val(&dst->this_io_blocks[ddir], &src->this_io_blocks[ddir]);
	181	sum_val(&dst->this_io_bytes[ddir], &src->this_io_bytes[ddir]);
	182	sum_val(&dst->bytes_done[ddir], &src->bytes_done[ddir]);
2a274336	183
a6a3469e	184	pthread_double_unlock(&dst->io_wq.stat_lock, &src->io_wq.stat_lock);
a9da8ab2 JA	185	}
	186
	187	static void update_accounting(struct submit_worker *sw)
	188	{
ee2b6d6e	189	struct thread_data *src = sw->private;
a9da8ab2 JA	190	struct thread_data *dst = sw->wq->td;
	191
	192	if (td_read(src))
	193	sum_ddir(dst, src, DDIR_READ);
	194	if (td_write(src))
	195	sum_ddir(dst, src, DDIR_WRITE);
	196	if (td_trim(src))
	197	sum_ddir(dst, src, DDIR_TRIM);
	198	}
	199
	200	static void worker_thread(void data)
	201	{
	202	struct submit_worker *sw = data;
	203	struct workqueue *wq = sw->wq;
a9da8ab2 JA	204	unsigned int eflags = 0, ret;
	205	FLIST_HEAD(local_list);
	206
80ea26a8	207	ret = workqueue_init_worker(sw);
a9da8ab2 JA	208	pthread_mutex_lock(&sw->lock);
	209	sw->flags \|= SW_F_RUNNING;
	210	if (ret)
	211	sw->flags \|= SW_F_ERROR;
	212	pthread_mutex_unlock(&sw->lock);
	213
	214	pthread_mutex_lock(&wq->flush_lock);
	215	pthread_cond_signal(&wq->flush_cond);
	216	pthread_mutex_unlock(&wq->flush_lock);
	217
	218	if (sw->flags & SW_F_ERROR)
	219	goto done;
	220
	221	while (1) {
	222	pthread_mutex_lock(&sw->lock);
	223
	224	if (flist_empty(&sw->work_list)) {
	225	if (sw->flags & SW_F_EXIT) {
	226	pthread_mutex_unlock(&sw->lock);
	227	break;
	228	}
	229
ee2b6d6e	230	if (workqueue_pre_sleep_check(sw)) {
a9da8ab2	231	pthread_mutex_unlock(&sw->lock);
ee2b6d6e	232	workqueue_pre_sleep(sw);
a9da8ab2 JA	233	pthread_mutex_lock(&sw->lock);
	234	}
	235
	236	/*
	237	* We dropped and reaquired the lock, check
	238	* state again.
	239	*/
	240	if (!flist_empty(&sw->work_list))
	241	goto handle_work;
	242
	243	if (sw->flags & SW_F_EXIT) {
	244	pthread_mutex_unlock(&sw->lock);
	245	break;
	246	} else if (!(sw->flags & SW_F_IDLE)) {
	247	sw->flags \|= SW_F_IDLE;
	248	wq->next_free_worker = sw->index;
	249	if (wq->wake_idle)
	250	pthread_cond_signal(&wq->flush_cond);
	251	}
	252	update_accounting(sw);
	253	pthread_cond_wait(&sw->cond, &sw->lock);
	254	} else {
	255	handle_work:
	256	flist_splice_init(&sw->work_list, &local_list);
	257	}
	258	pthread_mutex_unlock(&sw->lock);
	259	handle_list(sw, &local_list);
	260	}
	261
	262	update_accounting(sw);
	263
	264	done:
	265	pthread_mutex_lock(&sw->lock);
	266	sw->flags \|= (SW_F_EXITED \| eflags);
	267	pthread_mutex_unlock(&sw->lock);
	268	return NULL;
	269	}
	270
	271	static void free_worker(struct submit_worker *sw)
	272	{
ee2b6d6e	273	struct workqueue *wq = sw->wq;
a9da8ab2	274
df8472e1	275	workqueue_exit_worker(sw);
a9da8ab2 JA	276
	277	pthread_cond_destroy(&sw->cond);
	278	pthread_mutex_destroy(&sw->lock);
ee2b6d6e JA	279
	280	if (wq->ops.free_worker_fn)
	281	wq->ops.free_worker_fn(sw);
a9da8ab2 JA	282	}
	283
	284	static void shutdown_worker(struct submit_worker sw, unsigned int sum_cnt)
	285	{
	286	struct thread_data *parent = sw->wq->td;
ee2b6d6e	287	struct thread_data *td = sw->private;
a9da8ab2 JA	288
	289	pthread_join(sw->thread, NULL);
	290	(*sum_cnt)++;
ee2b6d6e	291	sum_thread_stats(&parent->ts, &td->ts, *sum_cnt == 1);
a9da8ab2 JA	292	free_worker(sw);
	293	}
	294
	295	void workqueue_exit(struct workqueue *wq)
	296	{
	297	unsigned int shutdown, sum_cnt = 0;
	298	struct submit_worker *sw;
	299	int i;
	300
	301	for (i = 0; i < wq->max_workers; i++) {
	302	sw = &wq->workers[i];
	303
	304	pthread_mutex_lock(&sw->lock);
	305	sw->flags \|= SW_F_EXIT;
	306	pthread_cond_signal(&sw->cond);
	307	pthread_mutex_unlock(&sw->lock);
	308	}
	309
	310	do {
	311	shutdown = 0;
	312	for (i = 0; i < wq->max_workers; i++) {
	313	sw = &wq->workers[i];
	314	if (sw->flags & SW_F_ACCOUNTED)
	315	continue;
b7d0bbf1	316	pthread_mutex_lock(&sw->lock);
a9da8ab2	317	sw->flags \|= SW_F_ACCOUNTED;
b7d0bbf1	318	pthread_mutex_unlock(&sw->lock);
a9da8ab2 JA	319	shutdown_worker(sw, &sum_cnt);
	320	shutdown++;
	321	}
	322	} while (shutdown && shutdown != wq->max_workers);
	323
	324	free(wq->workers);
	325	pthread_mutex_destroy(&wq->flush_lock);
	326	pthread_cond_destroy(&wq->flush_cond);
2a274336	327	pthread_mutex_destroy(&wq->stat_lock);
a9da8ab2 JA	328	}
	329
	330	static int start_worker(struct workqueue *wq, unsigned int index)
	331	{
	332	struct submit_worker *sw = &wq->workers[index];
	333	int ret;
	334
	335	INIT_FLIST_HEAD(&sw->work_list);
	336	pthread_cond_init(&sw->cond, NULL);
	337	pthread_mutex_init(&sw->lock, NULL);
	338	sw->wq = wq;
	339	sw->index = index;
	340
ee2b6d6e JA	341	if (wq->ops.alloc_worker_fn) {
	342	ret = wq->ops.alloc_worker_fn(sw);
	343	if (ret)
	344	return ret;
	345	}
	346
a9da8ab2 JA	347	ret = pthread_create(&sw->thread, NULL, worker_thread, sw);
	348	if (!ret) {
	349	pthread_mutex_lock(&sw->lock);
	350	sw->flags = SW_F_IDLE;
	351	pthread_mutex_unlock(&sw->lock);
	352	return 0;
	353	}
	354
	355	free_worker(sw);
	356	return 1;
	357	}
	358
	359	int workqueue_init(struct thread_data td, struct workqueue wq,
5bb79f69	360	struct workqueue_ops *ops, unsigned max_pending)
a9da8ab2 JA	361	{
	362	unsigned int running;
	363	int i, error;
	364
	365	wq->max_workers = max_pending;
	366	wq->td = td;
5bb79f69	367	wq->ops = *ops;
a9da8ab2 JA	368	wq->work_seq = 0;
	369	wq->next_free_worker = 0;
	370	pthread_cond_init(&wq->flush_cond, NULL);
	371	pthread_mutex_init(&wq->flush_lock, NULL);
2a274336	372	pthread_mutex_init(&wq->stat_lock, NULL);
a9da8ab2 JA	373
	374	wq->workers = calloc(wq->max_workers, sizeof(struct submit_worker));
	375
	376	for (i = 0; i < wq->max_workers; i++)
	377	if (start_worker(wq, i))
	378	break;
	379
	380	wq->max_workers = i;
bddc8d16 JA	381	if (!wq->max_workers)
bddc8d16 JA	382	goto err;
a9da8ab2 JA	383
	384	/*
	385	* Wait for them all to be started and initialized
	386	*/
	387	error = 0;
	388	do {
	389	struct submit_worker *sw;
	390
	391	running = 0;
	392	pthread_mutex_lock(&wq->flush_lock);
	393	for (i = 0; i < wq->max_workers; i++) {
	394	sw = &wq->workers[i];
	395	pthread_mutex_lock(&sw->lock);
	396	if (sw->flags & SW_F_RUNNING)
	397	running++;
	398	if (sw->flags & SW_F_ERROR)
	399	error++;
	400	pthread_mutex_unlock(&sw->lock);
	401	}
	402
	403	if (error \|\| running == wq->max_workers) {
	404	pthread_mutex_unlock(&wq->flush_lock);
	405	break;
	406	}
	407
	408	pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
	409	pthread_mutex_unlock(&wq->flush_lock);
	410	} while (1);
	411
bddc8d16 JA	412	if (!error)
bddc8d16 JA	413	return 0;
a9da8ab2	414
bddc8d16 JA	415	err:
	416	log_err("Can't create rate workqueue\n");
	417	td_verror(td, ESRCH, "workqueue_init");
	418	workqueue_exit(wq);
	419	return 1;
a9da8ab2	420	}