192 	stage_info_t    *sp = wp->stage;  in manager_fn()  local
193 	boolean_t       is_producer = (sp->stagenum == 0);  in manager_fn()
217 	DBG("Starting %s %d, stage: %d\n", sp->name, wp->setnum, sp->stagenum);  in manager_fn()
248 		pthread_mutex_lock(&sp->input->mtx);  in manager_fn()
250 			if (sp->work_todo == 0) {  in manager_fn()
251 				pthread_mutex_unlock(&sp->input->mtx);  in manager_fn()
254 			workp = TAILQ_FIRST(&(sp->input->queue));  in manager_fn()
259 			    sp->name, wp->setnum, sp->stagenum, sp->work_todo);  in manager_fn()
260 			sp->input->waiters++;  in manager_fn()
261 			pthread_cond_wait(&sp->input->cnd, &sp->input->mtx);  in manager_fn()
262 			sp->input->waiters--;  in manager_fn()
264 		TAILQ_REMOVE(&(sp->input->queue), workp, link);  in manager_fn()
265 		iteration = sp->work_todo--;  in manager_fn()
266 		pthread_mutex_unlock(&sp->input->mtx);  in manager_fn()
279 		    sp->name, wp->setnum, sp->stagenum, iteration, workp, workp->data);  in manager_fn()
282 		(void) sp->fn(workp->data, workp->isize);  in manager_fn()
288 		pthread_mutex_lock(&sp->output->mtx);  in manager_fn()
289 		TAILQ_INSERT_TAIL(&(sp->output->queue), workp, link);  in manager_fn()
290 		if (sp->output->waiters) {  in manager_fn()
292 			    sp->name, wp->setnum, sp->stagenum, iteration);  in manager_fn()
293 			pthread_cond_signal(&sp->output->cnd);  in manager_fn()
295 		pthread_mutex_unlock(&sp->output->mtx);  in manager_fn()
299 	pthread_cond_broadcast(&sp->output->cnd);  in manager_fn()
301 	DBG("Ending %s[%d,%d]\n", sp->name, wp->setnum, sp->stagenum);  in manager_fn()
322 	stage_info_t            *sp;  in main()  local
407 		sp = &stage_info[i];  in main()
408 		sp->stagenum = i;  in main()
409 		pthread_mutex_init(&sp->bufq.mtx, NULL);  in main()
410 		pthread_cond_init(&sp->bufq.cnd, NULL);  in main()
411 		TAILQ_INIT(&sp->bufq.queue);  in main()
412 		sp->bufq.waiters = 0;  in main()
414 			sp->fn = producer_fnp;  in main()
415 			sp->name = "producer";  in main()
417 			sp->fn = consumer_fnp;  in main()
418 			sp->name = "consumer";  in main()
420 		sp->input = &sp->bufq;  in main()
421 		sp->output = &stage_info[(i + 1) % stages].bufq;  in main()