libzpool/common/taskq.c

59 task_alloc(taskq_t *tq, int tqflags)
64 again:	if ((t = tq->tq_freelist) != NULL && tq->tq_nalloc >= tq->tq_minalloc) {
65 		tq->tq_freelist = t->tqent_next;
67 		if (tq->tq_nalloc >= tq->tq_maxalloc) {
81 			tq->tq_maxalloc_wait++;
82 			rv = cv_timedwait(&tq->tq_maxalloc_cv,
83 			    &tq->tq_lock, ddi_get_lbolt() + hz);
84 			tq->tq_maxalloc_wait--;
88 		mutex_exit(&tq->tq_lock);
92 		mutex_enter(&tq->tq_lock);
94 			tq->tq_nalloc++;
100 task_free(taskq_t *tq, taskq_ent_t *t)
102 	if (tq->tq_nalloc <= tq->tq_minalloc) {
103 		t->tqent_next = tq->tq_freelist;
104 		tq->tq_freelist = t;
106 		tq->tq_nalloc--;
107 		mutex_exit(&tq->tq_lock);
109 		mutex_enter(&tq->tq_lock);
112 	if (tq->tq_maxalloc_wait)
113 		cv_signal(&tq->tq_maxalloc_cv);
117 taskq_dispatch(taskq_t *tq, task_func_t func, void *arg, uint_t tqflags)
126 	mutex_enter(&tq->tq_lock);
127 	ASSERT(tq->tq_flags & TASKQ_ACTIVE);
128 	if ((t = task_alloc(tq, tqflags)) == NULL) {
129 		mutex_exit(&tq->tq_lock);
133 		t->tqent_next = tq->tq_task.tqent_next;
134 		t->tqent_prev = &tq->tq_task;
136 		t->tqent_next = &tq->tq_task;
137 		t->tqent_prev = tq->tq_task.tqent_prev;
144 	cv_signal(&tq->tq_dispatch_cv);
145 	mutex_exit(&tq->tq_lock);
150 taskq_dispatch_ent(taskq_t *tq, task_func_t func, void *arg, uint_t flags,
154 	ASSERT(!(tq->tq_flags & TASKQ_DYNAMIC));
164 	mutex_enter(&tq->tq_lock);
167 		t->tqent_next = tq->tq_task.tqent_next;
168 		t->tqent_prev = &tq->tq_task;
170 		t->tqent_next = &tq->tq_task;
171 		t->tqent_prev = tq->tq_task.tqent_prev;
177 	cv_signal(&tq->tq_dispatch_cv);
178 	mutex_exit(&tq->tq_lock);
182 taskq_wait(taskq_t *tq)
184 	mutex_enter(&tq->tq_lock);
185 	while (tq->tq_task.tqent_next != &tq->tq_task || tq->tq_active != 0)
186 		cv_wait(&tq->tq_wait_cv, &tq->tq_lock);
187 	mutex_exit(&tq->tq_lock);
193 	taskq_t *tq = arg;
197 	mutex_enter(&tq->tq_lock);
198 	while (tq->tq_flags & TASKQ_ACTIVE) {
199 		if ((t = tq->tq_task.tqent_next) == &tq->tq_task) {
200 			if (--tq->tq_active == 0)
201 				cv_broadcast(&tq->tq_wait_cv);
202 			cv_wait(&tq->tq_dispatch_cv, &tq->tq_lock);
203 			tq->tq_active++;
211 		mutex_exit(&tq->tq_lock);
213 		rw_enter(&tq->tq_threadlock, RW_READER);
215 		rw_exit(&tq->tq_threadlock);
217 		mutex_enter(&tq->tq_lock);
219 			task_free(tq, t);
221 	tq->tq_nthreads--;
222 	cv_broadcast(&tq->tq_wait_cv);
223 	mutex_exit(&tq->tq_lock);
232 	taskq_t *tq = kmem_zalloc(sizeof (taskq_t), KM_SLEEP);
248 	rw_init(&tq->tq_threadlock, NULL, RW_DEFAULT, NULL);
249 	mutex_init(&tq->tq_lock, NULL, MUTEX_DEFAULT, NULL);
250 	cv_init(&tq->tq_dispatch_cv, NULL, CV_DEFAULT, NULL);
251 	cv_init(&tq->tq_wait_cv, NULL, CV_DEFAULT, NULL);
252 	cv_init(&tq->tq_maxalloc_cv, NULL, CV_DEFAULT, NULL);
253 	(void) strncpy(tq->tq_name, name, TASKQ_NAMELEN + 1);
254 	tq->tq_flags = flags | TASKQ_ACTIVE;
255 	tq->tq_active = nthreads;
256 	tq->tq_nthreads = nthreads;
257 	tq->tq_minalloc = minalloc;
258 	tq->tq_maxalloc = maxalloc;
259 	tq->tq_task.tqent_next = &tq->tq_task;
260 	tq->tq_task.tqent_prev = &tq->tq_task;
261 	tq->tq_threadlist = kmem_alloc(nthreads * sizeof (thread_t), KM_SLEEP);
264 		mutex_enter(&tq->tq_lock);
266 			task_free(tq, task_alloc(tq, KM_SLEEP));
267 		mutex_exit(&tq->tq_lock);
272 		    tq, THR_BOUND, &tq->tq_threadlist[t]);
274 	return (tq);
278 taskq_destroy(taskq_t *tq)
281 	int nthreads = tq->tq_nthreads;
283 	taskq_wait(tq);
285 	mutex_enter(&tq->tq_lock);
287 	tq->tq_flags &= ~TASKQ_ACTIVE;
288 	cv_broadcast(&tq->tq_dispatch_cv);
290 	while (tq->tq_nthreads != 0)
291 		cv_wait(&tq->tq_wait_cv, &tq->tq_lock);
293 	tq->tq_minalloc = 0;
294 	while (tq->tq_nalloc != 0) {
295 		ASSERT(tq->tq_freelist != NULL);
296 		task_free(tq, task_alloc(tq, KM_SLEEP));
299 	mutex_exit(&tq->tq_lock);
302 		(void) thr_join(tq->tq_threadlist[t], NULL, NULL);
304 	kmem_free(tq->tq_threadlist, nthreads * sizeof (thread_t));
306 	rw_destroy(&tq->tq_threadlock);
307 	mutex_destroy(&tq->tq_lock);
308 	cv_destroy(&tq->tq_dispatch_cv);
309 	cv_destroy(&tq->tq_wait_cv);
310 	cv_destroy(&tq->tq_maxalloc_cv);
312 	kmem_free(tq, sizeof (taskq_t));
316 taskq_member(taskq_t *tq, void *t)
323 	for (i = 0; i < tq->tq_nthreads; i++)
324 		if (tq->tq_threadlist[i] == (thread_t)(uintptr_t)t)