opensolaris/misc/thread_pool.c

40 delete_pool(tpool_t *tpool)
47 	for (job = tpool->tp_head; job != NULL; job = tpool->tp_head) {
48 		tpool->tp_head = job->tpj_next;
51 	(void) pthread_attr_destroy(&tpool->tp_attr);
52 	free(tpool);
61 	tpool_t *tpool = arg;
63 	if (--tpool->tp_current == 0 &&
64 	    (tpool->tp_flags & (TP_DESTROY | TP_ABANDON))) {
65 		if (tpool->tp_flags & TP_ABANDON) {
66 			pthread_mutex_unlock(&tpool->tp_mutex);
67 			delete_pool(tpool);
70 		if (tpool->tp_flags & TP_DESTROY)
71 			(void) pthread_cond_broadcast(&tpool->tp_busycv);
73 	pthread_mutex_unlock(&tpool->tp_mutex);
77 notify_waiters(tpool_t *tpool)
79 	if (tpool->tp_head == NULL && tpool->tp_active == NULL) {
80 		tpool->tp_flags &= ~TP_WAIT;
81 		(void) pthread_cond_broadcast(&tpool->tp_waitcv);
91 	tpool_t *tpool = arg;
96 	pthread_mutex_lock(&tpool->tp_mutex);
98 	for (activepp = &tpool->tp_active;; activepp = &activep->tpa_next) {
105 	if (tpool->tp_flags & TP_WAIT)
106 		notify_waiters(tpool);
112 	tpool_t *tpool = (tpool_t *)arg;
119 	pthread_mutex_lock(&tpool->tp_mutex);
120 	pthread_cleanup_push(worker_cleanup, tpool);
129 		tpool->tp_idle++;
130 		if (tpool->tp_flags & TP_WAIT)
131 			notify_waiters(tpool);
132 		while ((tpool->tp_head == NULL ||
133 		    (tpool->tp_flags & TP_SUSPEND)) &&
134 		    !(tpool->tp_flags & (TP_DESTROY | TP_ABANDON))) {
135 			if (tpool->tp_current <= tpool->tp_minimum ||
136 			    tpool->tp_linger == 0) {
137 				(void) pthread_cond_wait(&tpool->tp_workcv,
138 				    &tpool->tp_mutex);
143 				timeout.tv_sec += tpool->tp_linger;
144 				if (pthread_cond_timedwait(&tpool->tp_workcv,
145 				    &tpool->tp_mutex, &timeout) != 0) {
151 		tpool->tp_idle--;
152 		if (tpool->tp_flags & TP_DESTROY)
154 		if (tpool->tp_flags & TP_ABANDON) {
156 			if (tpool->tp_flags & TP_SUSPEND) {
157 				tpool->tp_flags &= ~TP_SUSPEND;
158 				(void) pthread_cond_broadcast(&tpool->tp_workcv);
160 			if (tpool->tp_head == NULL)
163 		if ((job = tpool->tp_head) != NULL &&
164 		    !(tpool->tp_flags & TP_SUSPEND)) {
168 			tpool->tp_head = job->tpj_next;
169 			if (job == tpool->tp_tail)
170 				tpool->tp_tail = NULL;
171 			tpool->tp_njobs--;
172 			active.tpa_next = tpool->tp_active;
173 			tpool->tp_active = &active;
174 			pthread_mutex_unlock(&tpool->tp_mutex);
175 			pthread_cleanup_push(job_cleanup, tpool);
194 		if (elapsed && tpool->tp_current > tpool->tp_minimum) {
211 create_worker(tpool_t *tpool)
219 	error = pthread_create(&thread, &tpool->tp_attr, tpool_worker, tpool);
228 	tpool_t	*tpool;
236 	tpool = calloc(1, sizeof (*tpool));
237 	if (tpool == NULL) {
241 	(void) pthread_mutex_init(&tpool->tp_mutex, NULL);
242 	(void) pthread_cond_init(&tpool->tp_busycv, NULL);
243 	(void) pthread_cond_init(&tpool->tp_workcv, NULL);
244 	(void) pthread_cond_init(&tpool->tp_waitcv, NULL);
245 	tpool->tp_minimum = min_threads;
246 	tpool->tp_maximum = max_threads;
247 	tpool->tp_linger = linger;
250 	(void) pthread_attr_init(&tpool->tp_attr);
251 	(void) pthread_attr_setdetachstate(&tpool->tp_attr,
254 	return (tpool);
265 tpool_dispatch(tpool_t *tpool, void (*func)(void *), void *arg)
275 	pthread_mutex_lock(&tpool->tp_mutex);
277 	if (tpool->tp_head == NULL)
278 		tpool->tp_head = job;
280 		tpool->tp_tail->tpj_next = job;
281 	tpool->tp_tail = job;
282 	tpool->tp_njobs++;
284 	if (!(tpool->tp_flags & TP_SUSPEND)) {
285 		if (tpool->tp_idle > 0)
286 			(void) pthread_cond_signal(&tpool->tp_workcv);
287 		else if (tpool->tp_current < tpool->tp_maximum &&
288 		    create_worker(tpool) == 0)
289 			tpool->tp_current++;
292 	pthread_mutex_unlock(&tpool->tp_mutex);
302 tpool_destroy(tpool_t *tpool)
306 	pthread_mutex_lock(&tpool->tp_mutex);
307 	pthread_cleanup_push((_Voidfp)pthread_mutex_unlock, &tpool->tp_mutex);
310 	tpool->tp_flags |= TP_DESTROY;
311 	tpool->tp_flags &= ~TP_SUSPEND;
312 	(void) pthread_cond_broadcast(&tpool->tp_workcv);
315 	for (activep = tpool->tp_active; activep; activep = activep->tpa_next)
319 	while (tpool->tp_active != NULL) {
320 		tpool->tp_flags |= TP_WAIT;
321 		(void) pthread_cond_wait(&tpool->tp_waitcv, &tpool->tp_mutex);
325 	while (tpool->tp_current != 0)
326 		(void) pthread_cond_wait(&tpool->tp_busycv, &tpool->tp_mutex);
328 	pthread_cleanup_pop(1);	/* pthread_mutex_unlock(&tpool->tp_mutex); */
329 	delete_pool(tpool);
337 tpool_abandon(tpool_t *tpool)
340 	pthread_mutex_lock(&tpool->tp_mutex);
341 	if (tpool->tp_current == 0) {
343 		pthread_mutex_unlock(&tpool->tp_mutex);
344 		delete_pool(tpool);
347 		tpool->tp_flags |= TP_ABANDON;
348 		tpool->tp_flags &= ~TP_SUSPEND;
349 		(void) pthread_cond_broadcast(&tpool->tp_workcv);
350 		pthread_mutex_unlock(&tpool->tp_mutex);
359 tpool_wait(tpool_t *tpool)
362 	pthread_mutex_lock(&tpool->tp_mutex);
363 	pthread_cleanup_push((_Voidfp)pthread_mutex_unlock, &tpool->tp_mutex);
364 	while (tpool->tp_head != NULL || tpool->tp_active != NULL) {
365 		tpool->tp_flags |= TP_WAIT;
366 		(void) pthread_cond_wait(&tpool->tp_waitcv, &tpool->tp_mutex);
368 	pthread_cleanup_pop(1);	/* pthread_mutex_unlock(&tpool->tp_mutex); */
372 tpool_suspend(tpool_t *tpool)
375 	pthread_mutex_lock(&tpool->tp_mutex);
376 	tpool->tp_flags |= TP_SUSPEND;
377 	pthread_mutex_unlock(&tpool->tp_mutex);
381 tpool_suspended(tpool_t *tpool)
385 	pthread_mutex_lock(&tpool->tp_mutex);
386 	suspended = (tpool->tp_flags & TP_SUSPEND) != 0;
387 	pthread_mutex_unlock(&tpool->tp_mutex);
393 tpool_resume(tpool_t *tpool)
397 	pthread_mutex_lock(&tpool->tp_mutex);
398 	if (!(tpool->tp_flags & TP_SUSPEND)) {
399 		pthread_mutex_unlock(&tpool->tp_mutex);
402 	tpool->tp_flags &= ~TP_SUSPEND;
403 	(void) pthread_cond_broadcast(&tpool->tp_workcv);
404 	excess = tpool->tp_njobs - tpool->tp_idle;
405 	while (excess-- > 0 && tpool->tp_current < tpool->tp_maximum) {
406 		if (create_worker(tpool) != 0)
408 		tpool->tp_current++;
410 	pthread_mutex_unlock(&tpool->tp_mutex);
414 tpool_member(tpool_t *tpool)
419 	pthread_mutex_lock(&tpool->tp_mutex);
420 	for (activep = tpool->tp_active; activep; activep = activep->tpa_next) {
422 			pthread_mutex_unlock(&tpool->tp_mutex);
426 	pthread_mutex_unlock(&tpool->tp_mutex);