[fio.git] / workqueue.c

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

#include "fio.h"
#include "ioengine.h"
#include "flist.h"
#include "workqueue.h"
#include "lib/getrusage.h"

struct submit_worker {
	pthread_t thread;
	pthread_mutex_t lock;
	pthread_cond_t cond;
	struct flist_head work_list;
	unsigned int flags;
	unsigned int index;
	uint64_t seq;
	struct workqueue *wq;
	struct thread_data td;
};

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 int 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 0;
	}

	return 1;
}

/*
 * 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.
 */
int workqueue_enqueue(struct workqueue *wq, struct io_u *io_u)
{
	struct submit_worker *sw;

	sw = get_submit_worker(wq);
	if (sw) {
		const enum fio_ddir ddir = acct_ddir(io_u);
		struct thread_data *parent = wq->td;

		if (ddir_rw(ddir)) {
			parent->io_issues[ddir]++;
			parent->io_issue_bytes[ddir] += io_u->xfer_buflen;
			parent->rate_io_issue_bytes[ddir] += io_u->xfer_buflen;
		}

		pthread_mutex_lock(&sw->lock);
		flist_add_tail(&io_u->verify_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 FIO_Q_QUEUED;
	}

	return FIO_Q_BUSY;
}

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

	while (!flist_empty(list)) {
		io_u = flist_first_entry(list, struct io_u, verify_list);
		flist_del_init(&io_u->verify_list);
		wq->fn(&sw->td, io_u);
	}
}

static int init_submit_worker(struct submit_worker *sw)
{
	struct thread_data *parent = sw->wq->td;
	struct thread_data *td = &sw->td;
	int fio_unused ret;

	memcpy(&td->o, &parent->o, sizeof(td->o));
	memcpy(&td->ts, &parent->ts, sizeof(td->ts));
	td->o.uid = td->o.gid = -1U;
	dup_files(td, parent);
	fio_options_mem_dupe(td);

	if (ioengine_load(td))
		goto err;

	if (td->o.odirect)
		td->io_ops->flags |= FIO_RAWIO;

	td->pid = gettid();

	INIT_FLIST_HEAD(&td->io_log_list);
	INIT_FLIST_HEAD(&td->io_hist_list);
	INIT_FLIST_HEAD(&td->verify_list);
	INIT_FLIST_HEAD(&td->trim_list);
	INIT_FLIST_HEAD(&td->next_rand_list);
	td->io_hist_tree = RB_ROOT;

	td->o.iodepth = 1;
	if (td_io_init(td))
		goto err_io_init;

	fio_gettime(&td->epoch, NULL);
	fio_getrusage(&td->ru_start);
	clear_io_state(td);

	td_set_runstate(td, TD_RUNNING);
	td->flags |= TD_F_CHILD;
	td->parent = parent;
	return 0;

err_io_init:
	close_ioengine(td);
err:
	return 1;
}

#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 sum_ddir(struct thread_data *dst, struct thread_data *src,
		     enum fio_ddir ddir)
{
#ifndef CONFIG_SFAA
	pthread_mutex_lock(&dst->io_wq.stat_lock);
	pthread_mutex_lock(&src->io_wq.stat_lock);
#endif

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

#ifndef CONFIG_SFAA
	pthread_mutex_unlock(&src->io_wq.stat_lock);
	pthread_mutex_unlock(&dst->io_wq.stat_lock);
#endif
}

static void update_accounting(struct submit_worker *sw)
{
	struct thread_data *src = &sw->td;
	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;
	struct thread_data *td = &sw->td;
	unsigned int eflags = 0, ret;
	FLIST_HEAD(local_list);

	ret = init_submit_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 (td->io_u_queued || td->cur_depth ||
			    td->io_u_in_flight) {
				pthread_mutex_unlock(&sw->lock);
				io_u_quiesce(td);
				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 thread_data *td = &sw->td;

	fio_options_free(td);
	close_and_free_files(td);
	if (td->io_ops)
		close_ioengine(td);
	td_set_runstate(td, TD_EXITED);

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

static void shutdown_worker(struct submit_worker *sw, unsigned int *sum_cnt)
{
	struct thread_data *parent = sw->wq->td;

	pthread_join(sw->thread, NULL);
	(*sum_cnt)++;
	sum_thread_stats(&parent->ts, &sw->td.ts, *sum_cnt);
	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;
			sw->flags |= SW_F_ACCOUNTED;
			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;

	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,
		   workqueue_fn *fn, unsigned max_pending)
{
	unsigned int running;
	int i, error;

	wq->max_workers = max_pending;
	wq->td = td;
	wq->fn = fn;
	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	* Rated submission helpers
	3	*
	4	* Copyright (C) 2015 Jens Axboe <axboe@kernel.dk>
	5	*
	6	*/
	7	#include <unistd.h>
	8
	9	#include "fio.h"
	10	#include "ioengine.h"
	11	#include "flist.h"
	12	#include "workqueue.h"
	13	#include "lib/getrusage.h"
	14
	15	struct submit_worker {
	16	pthread_t thread;
	17	pthread_mutex_t lock;
	18	pthread_cond_t cond;
	19	struct flist_head work_list;
	20	unsigned int flags;
	21	unsigned int index;
	22	uint64_t seq;
	23	struct workqueue *wq;
	24	struct thread_data td;
	25	};
	26
	27	enum {
	28	SW_F_IDLE = 1 << 0,
	29	SW_F_RUNNING = 1 << 1,
	30	SW_F_EXIT = 1 << 2,
	31	SW_F_EXITED = 1 << 3,
	32	SW_F_ACCOUNTED = 1 << 4,
	33	SW_F_ERROR = 1 << 5,
	34	};
	35
	36	static struct submit_worker __get_submit_worker(struct workqueue wq,
	37	unsigned int start,
	38	unsigned int end,
	39	struct submit_worker **best)
	40	{
	41	struct submit_worker *sw = NULL;
	42
	43	while (start <= end) {
	44	sw = &wq->workers[start];
	45	if (sw->flags & SW_F_IDLE)
	46	return sw;
	47	if (!(best) \|\| sw->seq < (best)->seq)
	48	*best = sw;
	49	start++;
	50	}
	51
	52	return NULL;
	53	}
	54
	55	static struct submit_worker get_submit_worker(struct workqueue wq)
	56	{
	57	unsigned int next = wq->next_free_worker;
	58	struct submit_worker sw, best = NULL;
	59
	60	assert(next < wq->max_workers);
	61
	62	sw = __get_submit_worker(wq, next, wq->max_workers - 1, &best);
	63	if (!sw && next)
	64	sw = __get_submit_worker(wq, 0, next - 1, &best);
	65
	66	/*
	67	* No truly idle found, use best match
	68	*/
	69	if (!sw)
	70	sw = best;
	71
	72	if (sw->index == wq->next_free_worker) {
	73	if (sw->index + 1 < wq->max_workers)
	74	wq->next_free_worker = sw->index + 1;
	75	else
	76	wq->next_free_worker = 0;
	77	}
	78
	79	return sw;
	80	}
	81
	82	static int all_sw_idle(struct workqueue *wq)
	83	{
	84	int i;
	85
	86	for (i = 0; i < wq->max_workers; i++) {
	87	struct submit_worker *sw = &wq->workers[i];
	88
	89	if (!(sw->flags & SW_F_IDLE))
	90	return 0;
	91	}
	92
	93	return 1;
	94	}
	95
	96	/*
	97	* Must be serialized wrt workqueue_enqueue() by caller
	98	*/
	99	void workqueue_flush(struct workqueue *wq)
	100	{
	101	wq->wake_idle = 1;
	102
	103	while (!all_sw_idle(wq)) {
	104	pthread_mutex_lock(&wq->flush_lock);
	105	pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
	106	pthread_mutex_unlock(&wq->flush_lock);
	107	}
	108
	109	wq->wake_idle = 0;
	110	}
	111
	112	/*
	113	* Must be serialized by caller.
	114	*/
	115	int workqueue_enqueue(struct workqueue wq, struct io_u io_u)
	116	{
	117	struct submit_worker *sw;
	118
	119	sw = get_submit_worker(wq);
	120	if (sw) {
	121	const enum fio_ddir ddir = acct_ddir(io_u);
	122	struct thread_data *parent = wq->td;
	123
	124	if (ddir_rw(ddir)) {
	125	parent->io_issues[ddir]++;
	126	parent->io_issue_bytes[ddir] += io_u->xfer_buflen;
	127	parent->rate_io_issue_bytes[ddir] += io_u->xfer_buflen;
	128	}
	129
	130	pthread_mutex_lock(&sw->lock);
	131	flist_add_tail(&io_u->verify_list, &sw->work_list);
	132	sw->seq = ++wq->work_seq;
	133	sw->flags &= ~SW_F_IDLE;
	134	pthread_mutex_unlock(&sw->lock);
	135
	136	pthread_cond_signal(&sw->cond);
	137	return FIO_Q_QUEUED;
	138	}
	139
	140	return FIO_Q_BUSY;
	141	}
	142
	143	static void handle_list(struct submit_worker sw, struct flist_head list)
	144	{
	145	struct workqueue *wq = sw->wq;
	146	struct io_u *io_u;
	147
	148	while (!flist_empty(list)) {
	149	io_u = flist_first_entry(list, struct io_u, verify_list);
	150	flist_del_init(&io_u->verify_list);
	151	wq->fn(&sw->td, io_u);
	152	}
	153	}
	154
	155	static int init_submit_worker(struct submit_worker *sw)
	156	{
	157	struct thread_data *parent = sw->wq->td;
	158	struct thread_data *td = &sw->td;
	159	int fio_unused ret;
	160
	161	memcpy(&td->o, &parent->o, sizeof(td->o));
	162	memcpy(&td->ts, &parent->ts, sizeof(td->ts));
	163	td->o.uid = td->o.gid = -1U;
	164	dup_files(td, parent);
	165	fio_options_mem_dupe(td);
	166
	167	if (ioengine_load(td))
	168	goto err;
	169
	170	if (td->o.odirect)
	171	td->io_ops->flags \|= FIO_RAWIO;
	172
	173	td->pid = gettid();
	174
	175	INIT_FLIST_HEAD(&td->io_log_list);
	176	INIT_FLIST_HEAD(&td->io_hist_list);
	177	INIT_FLIST_HEAD(&td->verify_list);
	178	INIT_FLIST_HEAD(&td->trim_list);
	179	INIT_FLIST_HEAD(&td->next_rand_list);
	180	td->io_hist_tree = RB_ROOT;
	181
	182	td->o.iodepth = 1;
	183	if (td_io_init(td))
	184	goto err_io_init;
	185
	186	fio_gettime(&td->epoch, NULL);
	187	fio_getrusage(&td->ru_start);
	188	clear_io_state(td);
	189
	190	td_set_runstate(td, TD_RUNNING);
	191	td->flags \|= TD_F_CHILD;
	192	td->parent = parent;
	193	return 0;
	194
	195	err_io_init:
	196	close_ioengine(td);
	197	err:
	198	return 1;
	199	}
	200
	201	#ifdef CONFIG_SFAA
	202	static void sum_val(uint64_t dst, uint64_t src)
	203	{
	204	if (*src) {
	205	__sync_fetch_and_add(dst, *src);
	206	*src = 0;
	207	}
	208	}
	209	#else
	210	static void sum_val(uint64_t dst, uint64_t src)
	211	{
	212	if (*src) {
	213	dst += src;
	214	*src = 0;
	215	}
	216	}
	217	#endif
	218
	219	static void sum_ddir(struct thread_data dst, struct thread_data src,
	220	enum fio_ddir ddir)
	221	{
	222	#ifndef CONFIG_SFAA
	223	pthread_mutex_lock(&dst->io_wq.stat_lock);
	224	pthread_mutex_lock(&src->io_wq.stat_lock);
	225	#endif
	226
	227	sum_val(&dst->io_bytes[ddir], &src->io_bytes[ddir]);
	228	sum_val(&dst->io_blocks[ddir], &src->io_blocks[ddir]);
	229	sum_val(&dst->this_io_blocks[ddir], &src->this_io_blocks[ddir]);
	230	sum_val(&dst->this_io_bytes[ddir], &src->this_io_bytes[ddir]);
	231	sum_val(&dst->bytes_done[ddir], &src->bytes_done[ddir]);
	232
	233	#ifndef CONFIG_SFAA
	234	pthread_mutex_unlock(&src->io_wq.stat_lock);
	235	pthread_mutex_unlock(&dst->io_wq.stat_lock);
	236	#endif
	237	}
	238
	239	static void update_accounting(struct submit_worker *sw)
	240	{
	241	struct thread_data *src = &sw->td;
	242	struct thread_data *dst = sw->wq->td;
	243
	244	if (td_read(src))
	245	sum_ddir(dst, src, DDIR_READ);
	246	if (td_write(src))
	247	sum_ddir(dst, src, DDIR_WRITE);
	248	if (td_trim(src))
	249	sum_ddir(dst, src, DDIR_TRIM);
	250	}
	251
	252	static void worker_thread(void data)
	253	{
	254	struct submit_worker *sw = data;
	255	struct workqueue *wq = sw->wq;
	256	struct thread_data *td = &sw->td;
	257	unsigned int eflags = 0, ret;
	258	FLIST_HEAD(local_list);
	259
	260	ret = init_submit_worker(sw);
	261	pthread_mutex_lock(&sw->lock);
	262	sw->flags \|= SW_F_RUNNING;
	263	if (ret)
	264	sw->flags \|= SW_F_ERROR;
	265	pthread_mutex_unlock(&sw->lock);
	266
	267	pthread_mutex_lock(&wq->flush_lock);
	268	pthread_cond_signal(&wq->flush_cond);
	269	pthread_mutex_unlock(&wq->flush_lock);
	270
	271	if (sw->flags & SW_F_ERROR)
	272	goto done;
	273
	274	while (1) {
	275	pthread_mutex_lock(&sw->lock);
	276
	277	if (flist_empty(&sw->work_list)) {
	278	if (sw->flags & SW_F_EXIT) {
	279	pthread_mutex_unlock(&sw->lock);
	280	break;
	281	}
	282
	283	if (td->io_u_queued \|\| td->cur_depth \|\|
	284	td->io_u_in_flight) {
	285	pthread_mutex_unlock(&sw->lock);
	286	io_u_quiesce(td);
	287	pthread_mutex_lock(&sw->lock);
	288	}
	289
	290	/*
	291	* We dropped and reaquired the lock, check
	292	* state again.
	293	*/
	294	if (!flist_empty(&sw->work_list))
	295	goto handle_work;
	296
	297	if (sw->flags & SW_F_EXIT) {
	298	pthread_mutex_unlock(&sw->lock);
	299	break;
	300	} else if (!(sw->flags & SW_F_IDLE)) {
	301	sw->flags \|= SW_F_IDLE;
	302	wq->next_free_worker = sw->index;
	303	if (wq->wake_idle)
	304	pthread_cond_signal(&wq->flush_cond);
	305	}
	306	update_accounting(sw);
	307	pthread_cond_wait(&sw->cond, &sw->lock);
	308	} else {
	309	handle_work:
	310	flist_splice_init(&sw->work_list, &local_list);
	311	}
	312	pthread_mutex_unlock(&sw->lock);
	313	handle_list(sw, &local_list);
	314	}
	315
	316	update_accounting(sw);
	317
	318	done:
	319	pthread_mutex_lock(&sw->lock);
	320	sw->flags \|= (SW_F_EXITED \| eflags);
	321	pthread_mutex_unlock(&sw->lock);
	322	return NULL;
	323	}
	324
	325	static void free_worker(struct submit_worker *sw)
	326	{
	327	struct thread_data *td = &sw->td;
	328
	329	fio_options_free(td);
	330	close_and_free_files(td);
	331	if (td->io_ops)
	332	close_ioengine(td);
	333	td_set_runstate(td, TD_EXITED);
	334
	335	pthread_cond_destroy(&sw->cond);
	336	pthread_mutex_destroy(&sw->lock);
	337	}
	338
	339	static void shutdown_worker(struct submit_worker sw, unsigned int sum_cnt)
	340	{
	341	struct thread_data *parent = sw->wq->td;
	342
	343	pthread_join(sw->thread, NULL);
	344	(*sum_cnt)++;
	345	sum_thread_stats(&parent->ts, &sw->td.ts, *sum_cnt);
	346	free_worker(sw);
	347	}
	348
	349	void workqueue_exit(struct workqueue *wq)
	350	{
	351	unsigned int shutdown, sum_cnt = 0;
	352	struct submit_worker *sw;
	353	int i;
	354
	355	for (i = 0; i < wq->max_workers; i++) {
	356	sw = &wq->workers[i];
	357
	358	pthread_mutex_lock(&sw->lock);
	359	sw->flags \|= SW_F_EXIT;
	360	pthread_cond_signal(&sw->cond);
	361	pthread_mutex_unlock(&sw->lock);
	362	}
	363
	364	do {
	365	shutdown = 0;
	366	for (i = 0; i < wq->max_workers; i++) {
	367	sw = &wq->workers[i];
	368	if (sw->flags & SW_F_ACCOUNTED)
	369	continue;
	370	sw->flags \|= SW_F_ACCOUNTED;
	371	shutdown_worker(sw, &sum_cnt);
	372	shutdown++;
	373	}
	374	} while (shutdown && shutdown != wq->max_workers);
	375
	376	free(wq->workers);
	377	pthread_mutex_destroy(&wq->flush_lock);
	378	pthread_cond_destroy(&wq->flush_cond);
	379	pthread_mutex_destroy(&wq->stat_lock);
	380	}
	381
	382	static int start_worker(struct workqueue *wq, unsigned int index)
	383	{
	384	struct submit_worker *sw = &wq->workers[index];
	385	int ret;
	386
	387	INIT_FLIST_HEAD(&sw->work_list);
	388	pthread_cond_init(&sw->cond, NULL);
	389	pthread_mutex_init(&sw->lock, NULL);
	390	sw->wq = wq;
	391	sw->index = index;
	392
	393	ret = pthread_create(&sw->thread, NULL, worker_thread, sw);
	394	if (!ret) {
	395	pthread_mutex_lock(&sw->lock);
	396	sw->flags = SW_F_IDLE;
	397	pthread_mutex_unlock(&sw->lock);
	398	return 0;
	399	}
	400
	401	free_worker(sw);
	402	return 1;
	403	}
	404
	405	int workqueue_init(struct thread_data td, struct workqueue wq,
	406	workqueue_fn *fn, unsigned max_pending)
	407	{
	408	unsigned int running;
	409	int i, error;
	410
	411	wq->max_workers = max_pending;
	412	wq->td = td;
	413	wq->fn = fn;
	414	wq->work_seq = 0;
	415	wq->next_free_worker = 0;
	416	pthread_cond_init(&wq->flush_cond, NULL);
	417	pthread_mutex_init(&wq->flush_lock, NULL);
	418	pthread_mutex_init(&wq->stat_lock, NULL);
	419
	420	wq->workers = calloc(wq->max_workers, sizeof(struct submit_worker));
	421
	422	for (i = 0; i < wq->max_workers; i++)
	423	if (start_worker(wq, i))
	424	break;
	425
	426	wq->max_workers = i;
	427	if (!wq->max_workers)
	428	goto err;
	429
	430	/*
	431	* Wait for them all to be started and initialized
	432	*/
	433	error = 0;
	434	do {
	435	struct submit_worker *sw;
	436
	437	running = 0;
	438	pthread_mutex_lock(&wq->flush_lock);
	439	for (i = 0; i < wq->max_workers; i++) {
	440	sw = &wq->workers[i];
	441	pthread_mutex_lock(&sw->lock);
	442	if (sw->flags & SW_F_RUNNING)
	443	running++;
	444	if (sw->flags & SW_F_ERROR)
	445	error++;
	446	pthread_mutex_unlock(&sw->lock);
	447	}
	448
	449	if (error \|\| running == wq->max_workers) {
	450	pthread_mutex_unlock(&wq->flush_lock);
	451	break;
	452	}
	453
	454	pthread_cond_wait(&wq->flush_cond, &wq->flush_lock);
	455	pthread_mutex_unlock(&wq->flush_lock);
	456	} while (1);
	457
	458	if (!error)
	459	return 0;
	460
	461	err:
	462	log_err("Can't create rate workqueue\n");
	463	td_verror(td, ESRCH, "workqueue_init");
	464	workqueue_exit(wq);
	465	return 1;
	466	}