178 	stage_info_t                    *sp = (stage_info_t *) arg;  in manager_fn()  local
179 	line_info_t                     *lp = sp->set;  in manager_fn()
210 	DBG("Starting %s set: %d stage: %d\n", sp->name, lp->setnum, sp->stagenum);  in manager_fn()
241 		pthread_mutex_lock(&sp->input->mtx);  in manager_fn()
243 			workp = TAILQ_FIRST(&(sp->input->queue));  in manager_fn()
248 			    sp->name, lp->setnum, sp->stagenum, iteration);  in manager_fn()
249 			sp->input->waiters = TRUE;  in manager_fn()
250 			pthread_cond_wait(&sp->input->cnd, &sp->input->mtx);  in manager_fn()
251 			sp->input->waiters = FALSE;  in manager_fn()
253 		TAILQ_REMOVE(&(sp->input->queue), workp, link);  in manager_fn()
254 		pthread_mutex_unlock(&sp->input->mtx);  in manager_fn()
257 		    sp->name, lp->setnum, sp->stagenum, iteration, workp, workp->data);  in manager_fn()
260 		(void) sp->fn(workp->data, lp->isize);  in manager_fn()
266 		pthread_mutex_lock(&sp->output->mtx);  in manager_fn()
267 		TAILQ_INSERT_TAIL(&(sp->output->queue), workp, link);  in manager_fn()
268 		if (sp->output->waiters) {  in manager_fn()
270 			    sp->name, lp->setnum, sp->stagenum, iteration);  in manager_fn()
271 			pthread_cond_signal(&sp->output->cnd);  in manager_fn()
273 		pthread_mutex_unlock(&sp->output->mtx);  in manager_fn()
276 	DBG("Ending %s[%d,%d]\n", sp->name, lp->setnum, sp->stagenum);  in manager_fn()
348 	stage_info_t            *sp;  in main()  local
438 			sp = &stage_info[(i * stages) + j];  in main()
439 			sp->stagenum = j;  in main()
440 			sp->set = lp;  in main()
441 			lp->stage[j] = sp;  in main()
442 			pthread_mutex_init(&sp->bufq.mtx, NULL);  in main()
443 			pthread_cond_init(&sp->bufq.cnd, NULL);  in main()
444 			TAILQ_INIT(&sp->bufq.queue);  in main()
445 			sp->bufq.waiters = FALSE;  in main()
453 			sp = lp->stage[j];  in main()
455 				sp->fn = producer_fnp;  in main()
456 				sp->name = "producer";  in main()
458 				sp->fn = consumer_fnp;  in main()
459 				sp->name = "consumer";  in main()
461 			sp->input = &lp->stage[j]->bufq;  in main()
462 			sp->output = &lp->stage[(j + 1) % stages]->bufq;  in main()