xref: /freebsd/sys/kern/subr_gtaskqueue.c (revision 59d50fe5ef5e670a3b4a7565e2fccde0098da160)
1 /*-
2  * Copyright (c) 2000 Doug Rabson
3  * Copyright (c) 2014 Jeff Roberson
4  * Copyright (c) 2016 Matthew Macy
5  * All rights reserved.
6  *
7  * Redistribution and use in source and binary forms, with or without
8  * modification, are permitted provided that the following conditions
9  * are met:
10  * 1. Redistributions of source code must retain the above copyright
11  *    notice, this list of conditions and the following disclaimer.
12  * 2. Redistributions in binary form must reproduce the above copyright
13  *    notice, this list of conditions and the following disclaimer in the
14  *    documentation and/or other materials provided with the distribution.
15  *
16  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
17  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
18  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
19  * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
20  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
21  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
22  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
23  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
24  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
25  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
26  * SUCH DAMAGE.
27  */
28 
29 #include <sys/cdefs.h>
30 __FBSDID("$FreeBSD$");
31 
32 #include <sys/param.h>
33 #include <sys/systm.h>
34 #include <sys/bus.h>
35 #include <sys/cpuset.h>
36 #include <sys/kernel.h>
37 #include <sys/kthread.h>
38 #include <sys/libkern.h>
39 #include <sys/limits.h>
40 #include <sys/lock.h>
41 #include <sys/malloc.h>
42 #include <sys/mutex.h>
43 #include <sys/proc.h>
44 #include <sys/epoch.h>
45 #include <sys/sched.h>
46 #include <sys/smp.h>
47 #include <sys/gtaskqueue.h>
48 #include <sys/unistd.h>
49 #include <machine/stdarg.h>
50 
51 static MALLOC_DEFINE(M_GTASKQUEUE, "gtaskqueue", "Group Task Queues");
52 static void	gtaskqueue_thread_enqueue(void *);
53 static void	gtaskqueue_thread_loop(void *arg);
54 static int	task_is_running(struct gtaskqueue *queue, struct gtask *gtask);
55 static void	gtaskqueue_drain_locked(struct gtaskqueue *queue, struct gtask *gtask);
56 
57 TASKQGROUP_DEFINE(softirq, mp_ncpus, 1);
58 TASKQGROUP_DEFINE(config, 1, 1);
59 
60 struct gtaskqueue_busy {
61 	struct gtask		*tb_running;
62 	u_int			 tb_seq;
63 	LIST_ENTRY(gtaskqueue_busy) tb_link;
64 };
65 
66 typedef void (*gtaskqueue_enqueue_fn)(void *context);
67 
68 struct gtaskqueue {
69 	STAILQ_HEAD(, gtask)	tq_queue;
70 	LIST_HEAD(, gtaskqueue_busy) tq_active;
71 	u_int			tq_seq;
72 	int			tq_callouts;
73 	struct mtx_padalign	tq_mutex;
74 	gtaskqueue_enqueue_fn	tq_enqueue;
75 	void			*tq_context;
76 	char			*tq_name;
77 	struct thread		**tq_threads;
78 	int			tq_tcount;
79 	int			tq_spin;
80 	int			tq_flags;
81 	taskqueue_callback_fn	tq_callbacks[TASKQUEUE_NUM_CALLBACKS];
82 	void			*tq_cb_contexts[TASKQUEUE_NUM_CALLBACKS];
83 };
84 
85 #define	TQ_FLAGS_ACTIVE		(1 << 0)
86 #define	TQ_FLAGS_BLOCKED	(1 << 1)
87 #define	TQ_FLAGS_UNLOCKED_ENQUEUE	(1 << 2)
88 
89 #define	DT_CALLOUT_ARMED	(1 << 0)
90 
91 #define	TQ_LOCK(tq)							\
92 	do {								\
93 		if ((tq)->tq_spin)					\
94 			mtx_lock_spin(&(tq)->tq_mutex);			\
95 		else							\
96 			mtx_lock(&(tq)->tq_mutex);			\
97 	} while (0)
98 #define	TQ_ASSERT_LOCKED(tq)	mtx_assert(&(tq)->tq_mutex, MA_OWNED)
99 
100 #define	TQ_UNLOCK(tq)							\
101 	do {								\
102 		if ((tq)->tq_spin)					\
103 			mtx_unlock_spin(&(tq)->tq_mutex);		\
104 		else							\
105 			mtx_unlock(&(tq)->tq_mutex);			\
106 	} while (0)
107 #define	TQ_ASSERT_UNLOCKED(tq)	mtx_assert(&(tq)->tq_mutex, MA_NOTOWNED)
108 
109 #ifdef INVARIANTS
110 static void
111 gtask_dump(struct gtask *gtask)
112 {
113 	printf("gtask: %p ta_flags=%x ta_priority=%d ta_func=%p ta_context=%p\n",
114 	       gtask, gtask->ta_flags, gtask->ta_priority, gtask->ta_func, gtask->ta_context);
115 }
116 #endif
117 
118 static __inline int
119 TQ_SLEEP(struct gtaskqueue *tq, void *p, const char *wm)
120 {
121 	if (tq->tq_spin)
122 		return (msleep_spin(p, (struct mtx *)&tq->tq_mutex, wm, 0));
123 	return (msleep(p, &tq->tq_mutex, 0, wm, 0));
124 }
125 
126 static struct gtaskqueue *
127 _gtaskqueue_create(const char *name, int mflags,
128 		 taskqueue_enqueue_fn enqueue, void *context,
129 		 int mtxflags, const char *mtxname __unused)
130 {
131 	struct gtaskqueue *queue;
132 	char *tq_name;
133 
134 	tq_name = malloc(TASKQUEUE_NAMELEN, M_GTASKQUEUE, mflags | M_ZERO);
135 	if (!tq_name)
136 		return (NULL);
137 
138 	snprintf(tq_name, TASKQUEUE_NAMELEN, "%s", (name) ? name : "taskqueue");
139 
140 	queue = malloc(sizeof(struct gtaskqueue), M_GTASKQUEUE, mflags | M_ZERO);
141 	if (!queue) {
142 		free(tq_name, M_GTASKQUEUE);
143 		return (NULL);
144 	}
145 
146 	STAILQ_INIT(&queue->tq_queue);
147 	LIST_INIT(&queue->tq_active);
148 	queue->tq_enqueue = enqueue;
149 	queue->tq_context = context;
150 	queue->tq_name = tq_name;
151 	queue->tq_spin = (mtxflags & MTX_SPIN) != 0;
152 	queue->tq_flags |= TQ_FLAGS_ACTIVE;
153 	if (enqueue == gtaskqueue_thread_enqueue)
154 		queue->tq_flags |= TQ_FLAGS_UNLOCKED_ENQUEUE;
155 	mtx_init(&queue->tq_mutex, tq_name, NULL, mtxflags);
156 
157 	return (queue);
158 }
159 
160 /*
161  * Signal a taskqueue thread to terminate.
162  */
163 static void
164 gtaskqueue_terminate(struct thread **pp, struct gtaskqueue *tq)
165 {
166 
167 	while (tq->tq_tcount > 0 || tq->tq_callouts > 0) {
168 		wakeup(tq);
169 		TQ_SLEEP(tq, pp, "gtq_destroy");
170 	}
171 }
172 
173 static void __unused
174 gtaskqueue_free(struct gtaskqueue *queue)
175 {
176 
177 	TQ_LOCK(queue);
178 	queue->tq_flags &= ~TQ_FLAGS_ACTIVE;
179 	gtaskqueue_terminate(queue->tq_threads, queue);
180 	KASSERT(LIST_EMPTY(&queue->tq_active), ("Tasks still running?"));
181 	KASSERT(queue->tq_callouts == 0, ("Armed timeout tasks"));
182 	mtx_destroy(&queue->tq_mutex);
183 	free(queue->tq_threads, M_GTASKQUEUE);
184 	free(queue->tq_name, M_GTASKQUEUE);
185 	free(queue, M_GTASKQUEUE);
186 }
187 
188 /*
189  * Wait for all to complete, then prevent it from being enqueued
190  */
191 void
192 grouptask_block(struct grouptask *grouptask)
193 {
194 	struct gtaskqueue *queue = grouptask->gt_taskqueue;
195 	struct gtask *gtask = &grouptask->gt_task;
196 
197 #ifdef INVARIANTS
198 	if (queue == NULL) {
199 		gtask_dump(gtask);
200 		panic("queue == NULL");
201 	}
202 #endif
203 	TQ_LOCK(queue);
204 	gtask->ta_flags |= TASK_NOENQUEUE;
205   	gtaskqueue_drain_locked(queue, gtask);
206 	TQ_UNLOCK(queue);
207 }
208 
209 void
210 grouptask_unblock(struct grouptask *grouptask)
211 {
212 	struct gtaskqueue *queue = grouptask->gt_taskqueue;
213 	struct gtask *gtask = &grouptask->gt_task;
214 
215 #ifdef INVARIANTS
216 	if (queue == NULL) {
217 		gtask_dump(gtask);
218 		panic("queue == NULL");
219 	}
220 #endif
221 	TQ_LOCK(queue);
222 	gtask->ta_flags &= ~TASK_NOENQUEUE;
223 	TQ_UNLOCK(queue);
224 }
225 
226 int
227 grouptaskqueue_enqueue(struct gtaskqueue *queue, struct gtask *gtask)
228 {
229 #ifdef INVARIANTS
230 	if (queue == NULL) {
231 		gtask_dump(gtask);
232 		panic("queue == NULL");
233 	}
234 #endif
235 	TQ_LOCK(queue);
236 	if (gtask->ta_flags & TASK_ENQUEUED) {
237 		TQ_UNLOCK(queue);
238 		return (0);
239 	}
240 	if (gtask->ta_flags & TASK_NOENQUEUE) {
241 		TQ_UNLOCK(queue);
242 		return (EAGAIN);
243 	}
244 	STAILQ_INSERT_TAIL(&queue->tq_queue, gtask, ta_link);
245 	gtask->ta_flags |= TASK_ENQUEUED;
246 	TQ_UNLOCK(queue);
247 	if ((queue->tq_flags & TQ_FLAGS_BLOCKED) == 0)
248 		queue->tq_enqueue(queue->tq_context);
249 	return (0);
250 }
251 
252 static void
253 gtaskqueue_task_nop_fn(void *context)
254 {
255 }
256 
257 /*
258  * Block until all currently queued tasks in this taskqueue
259  * have begun execution.  Tasks queued during execution of
260  * this function are ignored.
261  */
262 static void
263 gtaskqueue_drain_tq_queue(struct gtaskqueue *queue)
264 {
265 	struct gtask t_barrier;
266 
267 	if (STAILQ_EMPTY(&queue->tq_queue))
268 		return;
269 
270 	/*
271 	 * Enqueue our barrier after all current tasks, but with
272 	 * the highest priority so that newly queued tasks cannot
273 	 * pass it.  Because of the high priority, we can not use
274 	 * taskqueue_enqueue_locked directly (which drops the lock
275 	 * anyway) so just insert it at tail while we have the
276 	 * queue lock.
277 	 */
278 	GTASK_INIT(&t_barrier, 0, USHRT_MAX, gtaskqueue_task_nop_fn, &t_barrier);
279 	STAILQ_INSERT_TAIL(&queue->tq_queue, &t_barrier, ta_link);
280 	t_barrier.ta_flags |= TASK_ENQUEUED;
281 
282 	/*
283 	 * Once the barrier has executed, all previously queued tasks
284 	 * have completed or are currently executing.
285 	 */
286 	while (t_barrier.ta_flags & TASK_ENQUEUED)
287 		TQ_SLEEP(queue, &t_barrier, "gtq_qdrain");
288 }
289 
290 /*
291  * Block until all currently executing tasks for this taskqueue
292  * complete.  Tasks that begin execution during the execution
293  * of this function are ignored.
294  */
295 static void
296 gtaskqueue_drain_tq_active(struct gtaskqueue *queue)
297 {
298 	struct gtaskqueue_busy *tb;
299 	u_int seq;
300 
301 	if (LIST_EMPTY(&queue->tq_active))
302 		return;
303 
304 	/* Block taskq_terminate().*/
305 	queue->tq_callouts++;
306 
307 	/* Wait for any active task with sequence from the past. */
308 	seq = queue->tq_seq;
309 restart:
310 	LIST_FOREACH(tb, &queue->tq_active, tb_link) {
311 		if ((int)(tb->tb_seq - seq) <= 0) {
312 			TQ_SLEEP(queue, tb->tb_running, "gtq_adrain");
313 			goto restart;
314 		}
315 	}
316 
317 	/* Release taskqueue_terminate(). */
318 	queue->tq_callouts--;
319 	if ((queue->tq_flags & TQ_FLAGS_ACTIVE) == 0)
320 		wakeup_one(queue->tq_threads);
321 }
322 
323 void
324 gtaskqueue_block(struct gtaskqueue *queue)
325 {
326 
327 	TQ_LOCK(queue);
328 	queue->tq_flags |= TQ_FLAGS_BLOCKED;
329 	TQ_UNLOCK(queue);
330 }
331 
332 void
333 gtaskqueue_unblock(struct gtaskqueue *queue)
334 {
335 
336 	TQ_LOCK(queue);
337 	queue->tq_flags &= ~TQ_FLAGS_BLOCKED;
338 	if (!STAILQ_EMPTY(&queue->tq_queue))
339 		queue->tq_enqueue(queue->tq_context);
340 	TQ_UNLOCK(queue);
341 }
342 
343 static void
344 gtaskqueue_run_locked(struct gtaskqueue *queue)
345 {
346 	struct epoch_tracker et;
347 	struct gtaskqueue_busy tb;
348 	struct gtask *gtask;
349 	bool in_net_epoch;
350 
351 	KASSERT(queue != NULL, ("tq is NULL"));
352 	TQ_ASSERT_LOCKED(queue);
353 	tb.tb_running = NULL;
354 	LIST_INSERT_HEAD(&queue->tq_active, &tb, tb_link);
355 	in_net_epoch = false;
356 
357 	while ((gtask = STAILQ_FIRST(&queue->tq_queue)) != NULL) {
358 		STAILQ_REMOVE_HEAD(&queue->tq_queue, ta_link);
359 		gtask->ta_flags &= ~TASK_ENQUEUED;
360 		tb.tb_running = gtask;
361 		tb.tb_seq = ++queue->tq_seq;
362 		TQ_UNLOCK(queue);
363 
364 		KASSERT(gtask->ta_func != NULL, ("task->ta_func is NULL"));
365 		if (!in_net_epoch && TASK_IS_NET(gtask)) {
366 			in_net_epoch = true;
367 			NET_EPOCH_ENTER(et);
368 		} else if (in_net_epoch && !TASK_IS_NET(gtask)) {
369 			NET_EPOCH_EXIT(et);
370 			in_net_epoch = false;
371 		}
372 		gtask->ta_func(gtask->ta_context);
373 
374 		TQ_LOCK(queue);
375 		wakeup(gtask);
376 	}
377 	if (in_net_epoch)
378 		NET_EPOCH_EXIT(et);
379 	LIST_REMOVE(&tb, tb_link);
380 }
381 
382 static int
383 task_is_running(struct gtaskqueue *queue, struct gtask *gtask)
384 {
385 	struct gtaskqueue_busy *tb;
386 
387 	TQ_ASSERT_LOCKED(queue);
388 	LIST_FOREACH(tb, &queue->tq_active, tb_link) {
389 		if (tb->tb_running == gtask)
390 			return (1);
391 	}
392 	return (0);
393 }
394 
395 static int
396 gtaskqueue_cancel_locked(struct gtaskqueue *queue, struct gtask *gtask)
397 {
398 
399 	if (gtask->ta_flags & TASK_ENQUEUED)
400 		STAILQ_REMOVE(&queue->tq_queue, gtask, gtask, ta_link);
401 	gtask->ta_flags &= ~TASK_ENQUEUED;
402 	return (task_is_running(queue, gtask) ? EBUSY : 0);
403 }
404 
405 int
406 gtaskqueue_cancel(struct gtaskqueue *queue, struct gtask *gtask)
407 {
408 	int error;
409 
410 	TQ_LOCK(queue);
411 	error = gtaskqueue_cancel_locked(queue, gtask);
412 	TQ_UNLOCK(queue);
413 
414 	return (error);
415 }
416 
417 static void
418 gtaskqueue_drain_locked(struct gtaskqueue *queue, struct gtask *gtask)
419 {
420 	while ((gtask->ta_flags & TASK_ENQUEUED) || task_is_running(queue, gtask))
421 		TQ_SLEEP(queue, gtask, "gtq_drain");
422 }
423 
424 void
425 gtaskqueue_drain(struct gtaskqueue *queue, struct gtask *gtask)
426 {
427 
428 	if (!queue->tq_spin)
429 		WITNESS_WARN(WARN_GIANTOK | WARN_SLEEPOK, NULL, __func__);
430 
431 	TQ_LOCK(queue);
432 	gtaskqueue_drain_locked(queue, gtask);
433 	TQ_UNLOCK(queue);
434 }
435 
436 void
437 gtaskqueue_drain_all(struct gtaskqueue *queue)
438 {
439 
440 	if (!queue->tq_spin)
441 		WITNESS_WARN(WARN_GIANTOK | WARN_SLEEPOK, NULL, __func__);
442 
443 	TQ_LOCK(queue);
444 	gtaskqueue_drain_tq_queue(queue);
445 	gtaskqueue_drain_tq_active(queue);
446 	TQ_UNLOCK(queue);
447 }
448 
449 static int
450 _gtaskqueue_start_threads(struct gtaskqueue **tqp, int count, int pri,
451     cpuset_t *mask, const char *name, va_list ap)
452 {
453 	char ktname[MAXCOMLEN + 1];
454 	struct thread *td;
455 	struct gtaskqueue *tq;
456 	int i, error;
457 
458 	if (count <= 0)
459 		return (EINVAL);
460 
461 	vsnprintf(ktname, sizeof(ktname), name, ap);
462 	tq = *tqp;
463 
464 	tq->tq_threads = malloc(sizeof(struct thread *) * count, M_GTASKQUEUE,
465 	    M_NOWAIT | M_ZERO);
466 	if (tq->tq_threads == NULL) {
467 		printf("%s: no memory for %s threads\n", __func__, ktname);
468 		return (ENOMEM);
469 	}
470 
471 	for (i = 0; i < count; i++) {
472 		if (count == 1)
473 			error = kthread_add(gtaskqueue_thread_loop, tqp, NULL,
474 			    &tq->tq_threads[i], RFSTOPPED, 0, "%s", ktname);
475 		else
476 			error = kthread_add(gtaskqueue_thread_loop, tqp, NULL,
477 			    &tq->tq_threads[i], RFSTOPPED, 0,
478 			    "%s_%d", ktname, i);
479 		if (error) {
480 			/* should be ok to continue, taskqueue_free will dtrt */
481 			printf("%s: kthread_add(%s): error %d", __func__,
482 			    ktname, error);
483 			tq->tq_threads[i] = NULL;		/* paranoid */
484 		} else
485 			tq->tq_tcount++;
486 	}
487 	for (i = 0; i < count; i++) {
488 		if (tq->tq_threads[i] == NULL)
489 			continue;
490 		td = tq->tq_threads[i];
491 		if (mask) {
492 			error = cpuset_setthread(td->td_tid, mask);
493 			/*
494 			 * Failing to pin is rarely an actual fatal error;
495 			 * it'll just affect performance.
496 			 */
497 			if (error)
498 				printf("%s: curthread=%llu: can't pin; "
499 				    "error=%d\n",
500 				    __func__,
501 				    (unsigned long long) td->td_tid,
502 				    error);
503 		}
504 		thread_lock(td);
505 		sched_prio(td, pri);
506 		sched_add(td, SRQ_BORING);
507 	}
508 
509 	return (0);
510 }
511 
512 static int
513 gtaskqueue_start_threads(struct gtaskqueue **tqp, int count, int pri,
514     const char *name, ...)
515 {
516 	va_list ap;
517 	int error;
518 
519 	va_start(ap, name);
520 	error = _gtaskqueue_start_threads(tqp, count, pri, NULL, name, ap);
521 	va_end(ap);
522 	return (error);
523 }
524 
525 static inline void
526 gtaskqueue_run_callback(struct gtaskqueue *tq,
527     enum taskqueue_callback_type cb_type)
528 {
529 	taskqueue_callback_fn tq_callback;
530 
531 	TQ_ASSERT_UNLOCKED(tq);
532 	tq_callback = tq->tq_callbacks[cb_type];
533 	if (tq_callback != NULL)
534 		tq_callback(tq->tq_cb_contexts[cb_type]);
535 }
536 
537 static void
538 gtaskqueue_thread_loop(void *arg)
539 {
540 	struct gtaskqueue **tqp, *tq;
541 
542 	tqp = arg;
543 	tq = *tqp;
544 	gtaskqueue_run_callback(tq, TASKQUEUE_CALLBACK_TYPE_INIT);
545 	TQ_LOCK(tq);
546 	while ((tq->tq_flags & TQ_FLAGS_ACTIVE) != 0) {
547 		/* XXX ? */
548 		gtaskqueue_run_locked(tq);
549 		/*
550 		 * Because taskqueue_run() can drop tq_mutex, we need to
551 		 * check if the TQ_FLAGS_ACTIVE flag wasn't removed in the
552 		 * meantime, which means we missed a wakeup.
553 		 */
554 		if ((tq->tq_flags & TQ_FLAGS_ACTIVE) == 0)
555 			break;
556 		TQ_SLEEP(tq, tq, "-");
557 	}
558 	gtaskqueue_run_locked(tq);
559 	/*
560 	 * This thread is on its way out, so just drop the lock temporarily
561 	 * in order to call the shutdown callback.  This allows the callback
562 	 * to look at the taskqueue, even just before it dies.
563 	 */
564 	TQ_UNLOCK(tq);
565 	gtaskqueue_run_callback(tq, TASKQUEUE_CALLBACK_TYPE_SHUTDOWN);
566 	TQ_LOCK(tq);
567 
568 	/* rendezvous with thread that asked us to terminate */
569 	tq->tq_tcount--;
570 	wakeup_one(tq->tq_threads);
571 	TQ_UNLOCK(tq);
572 	kthread_exit();
573 }
574 
575 static void
576 gtaskqueue_thread_enqueue(void *context)
577 {
578 	struct gtaskqueue **tqp, *tq;
579 
580 	tqp = context;
581 	tq = *tqp;
582 	wakeup_any(tq);
583 }
584 
585 static struct gtaskqueue *
586 gtaskqueue_create_fast(const char *name, int mflags,
587 		 taskqueue_enqueue_fn enqueue, void *context)
588 {
589 	return _gtaskqueue_create(name, mflags, enqueue, context,
590 			MTX_SPIN, "fast_taskqueue");
591 }
592 
593 struct taskqgroup_cpu {
594 	LIST_HEAD(, grouptask) tgc_tasks;
595 	struct gtaskqueue *tgc_taskq;
596 	int		tgc_cnt;
597 	int		tgc_cpu;
598 };
599 
600 struct taskqgroup {
601 	struct taskqgroup_cpu tqg_queue[MAXCPU];
602 	struct mtx	tqg_lock;
603 	const char *	tqg_name;
604 	int		tqg_cnt;
605 };
606 
607 struct taskq_bind_task {
608 	struct gtask bt_task;
609 	int	bt_cpuid;
610 };
611 
612 static void
613 taskqgroup_cpu_create(struct taskqgroup *qgroup, int idx, int cpu)
614 {
615 	struct taskqgroup_cpu *qcpu;
616 
617 	qcpu = &qgroup->tqg_queue[idx];
618 	LIST_INIT(&qcpu->tgc_tasks);
619 	qcpu->tgc_taskq = gtaskqueue_create_fast(NULL, M_WAITOK,
620 	    taskqueue_thread_enqueue, &qcpu->tgc_taskq);
621 	gtaskqueue_start_threads(&qcpu->tgc_taskq, 1, PI_SOFT,
622 	    "%s_%d", qgroup->tqg_name, idx);
623 	qcpu->tgc_cpu = cpu;
624 }
625 
626 /*
627  * Find the taskq with least # of tasks that doesn't currently have any
628  * other queues from the uniq identifier.
629  */
630 static int
631 taskqgroup_find(struct taskqgroup *qgroup, void *uniq)
632 {
633 	struct grouptask *n;
634 	int i, idx, mincnt;
635 	int strict;
636 
637 	mtx_assert(&qgroup->tqg_lock, MA_OWNED);
638 	KASSERT(qgroup->tqg_cnt != 0,
639 	    ("qgroup %s has no queues", qgroup->tqg_name));
640 
641 	/*
642 	 * Two passes: first scan for a queue with the least tasks that
643 	 * does not already service this uniq id.  If that fails simply find
644 	 * the queue with the least total tasks.
645 	 */
646 	for (idx = -1, mincnt = INT_MAX, strict = 1; mincnt == INT_MAX;
647 	    strict = 0) {
648 		for (i = 0; i < qgroup->tqg_cnt; i++) {
649 			if (qgroup->tqg_queue[i].tgc_cnt > mincnt)
650 				continue;
651 			if (strict) {
652 				LIST_FOREACH(n, &qgroup->tqg_queue[i].tgc_tasks,
653 				    gt_list)
654 					if (n->gt_uniq == uniq)
655 						break;
656 				if (n != NULL)
657 					continue;
658 			}
659 			mincnt = qgroup->tqg_queue[i].tgc_cnt;
660 			idx = i;
661 		}
662 	}
663 	if (idx == -1)
664 		panic("%s: failed to pick a qid.", __func__);
665 
666 	return (idx);
667 }
668 
669 void
670 taskqgroup_attach(struct taskqgroup *qgroup, struct grouptask *gtask,
671     void *uniq, device_t dev, struct resource *irq, const char *name)
672 {
673 	int cpu, qid, error;
674 
675 	KASSERT(qgroup->tqg_cnt > 0,
676 	    ("qgroup %s has no queues", qgroup->tqg_name));
677 
678 	gtask->gt_uniq = uniq;
679 	snprintf(gtask->gt_name, GROUPTASK_NAMELEN, "%s", name ? name : "grouptask");
680 	gtask->gt_dev = dev;
681 	gtask->gt_irq = irq;
682 	gtask->gt_cpu = -1;
683 	mtx_lock(&qgroup->tqg_lock);
684 	qid = taskqgroup_find(qgroup, uniq);
685 	qgroup->tqg_queue[qid].tgc_cnt++;
686 	LIST_INSERT_HEAD(&qgroup->tqg_queue[qid].tgc_tasks, gtask, gt_list);
687 	gtask->gt_taskqueue = qgroup->tqg_queue[qid].tgc_taskq;
688 	if (dev != NULL && irq != NULL) {
689 		cpu = qgroup->tqg_queue[qid].tgc_cpu;
690 		gtask->gt_cpu = cpu;
691 		mtx_unlock(&qgroup->tqg_lock);
692 		error = bus_bind_intr(dev, irq, cpu);
693 		if (error)
694 			printf("%s: binding interrupt failed for %s: %d\n",
695 			    __func__, gtask->gt_name, error);
696 	} else
697 		mtx_unlock(&qgroup->tqg_lock);
698 }
699 
700 int
701 taskqgroup_attach_cpu(struct taskqgroup *qgroup, struct grouptask *gtask,
702     void *uniq, int cpu, device_t dev, struct resource *irq, const char *name)
703 {
704 	int i, qid, error;
705 
706 	gtask->gt_uniq = uniq;
707 	snprintf(gtask->gt_name, GROUPTASK_NAMELEN, "%s", name ? name : "grouptask");
708 	gtask->gt_dev = dev;
709 	gtask->gt_irq = irq;
710 	gtask->gt_cpu = cpu;
711 	mtx_lock(&qgroup->tqg_lock);
712 	for (i = 0, qid = -1; i < qgroup->tqg_cnt; i++)
713 		if (qgroup->tqg_queue[i].tgc_cpu == cpu) {
714 			qid = i;
715 			break;
716 		}
717 	if (qid == -1) {
718 		mtx_unlock(&qgroup->tqg_lock);
719 		printf("%s: qid not found for %s cpu=%d\n", __func__, gtask->gt_name, cpu);
720 		return (EINVAL);
721 	}
722 	qgroup->tqg_queue[qid].tgc_cnt++;
723 	LIST_INSERT_HEAD(&qgroup->tqg_queue[qid].tgc_tasks, gtask, gt_list);
724 	gtask->gt_taskqueue = qgroup->tqg_queue[qid].tgc_taskq;
725 	cpu = qgroup->tqg_queue[qid].tgc_cpu;
726 	mtx_unlock(&qgroup->tqg_lock);
727 
728 	if (dev != NULL && irq != NULL) {
729 		error = bus_bind_intr(dev, irq, cpu);
730 		if (error)
731 			printf("%s: binding interrupt failed for %s: %d\n",
732 			    __func__, gtask->gt_name, error);
733 	}
734 	return (0);
735 }
736 
737 void
738 taskqgroup_detach(struct taskqgroup *qgroup, struct grouptask *gtask)
739 {
740 	int i;
741 
742 	grouptask_block(gtask);
743 	mtx_lock(&qgroup->tqg_lock);
744 	for (i = 0; i < qgroup->tqg_cnt; i++)
745 		if (qgroup->tqg_queue[i].tgc_taskq == gtask->gt_taskqueue)
746 			break;
747 	if (i == qgroup->tqg_cnt)
748 		panic("%s: task %s not in group", __func__, gtask->gt_name);
749 	qgroup->tqg_queue[i].tgc_cnt--;
750 	LIST_REMOVE(gtask, gt_list);
751 	mtx_unlock(&qgroup->tqg_lock);
752 	gtask->gt_taskqueue = NULL;
753 	gtask->gt_task.ta_flags &= ~TASK_NOENQUEUE;
754 }
755 
756 static void
757 taskqgroup_binder(void *ctx)
758 {
759 	struct taskq_bind_task *gtask;
760 	cpuset_t mask;
761 	int error;
762 
763 	gtask = ctx;
764 	CPU_ZERO(&mask);
765 	CPU_SET(gtask->bt_cpuid, &mask);
766 	error = cpuset_setthread(curthread->td_tid, &mask);
767 	thread_lock(curthread);
768 	sched_bind(curthread, gtask->bt_cpuid);
769 	thread_unlock(curthread);
770 
771 	if (error)
772 		printf("%s: binding curthread failed: %d\n", __func__, error);
773 	free(gtask, M_DEVBUF);
774 }
775 
776 void
777 taskqgroup_bind(struct taskqgroup *qgroup)
778 {
779 	struct taskq_bind_task *gtask;
780 	int i;
781 
782 	/*
783 	 * Bind taskqueue threads to specific CPUs, if they have been assigned
784 	 * one.
785 	 */
786 	if (qgroup->tqg_cnt == 1)
787 		return;
788 
789 	for (i = 0; i < qgroup->tqg_cnt; i++) {
790 		gtask = malloc(sizeof(*gtask), M_DEVBUF, M_WAITOK);
791 		GTASK_INIT(&gtask->bt_task, 0, 0, taskqgroup_binder, gtask);
792 		gtask->bt_cpuid = qgroup->tqg_queue[i].tgc_cpu;
793 		grouptaskqueue_enqueue(qgroup->tqg_queue[i].tgc_taskq,
794 		    &gtask->bt_task);
795 	}
796 }
797 
798 struct taskqgroup *
799 taskqgroup_create(const char *name, int cnt, int stride)
800 {
801 	struct taskqgroup *qgroup;
802 	int cpu, i, j;
803 
804 	qgroup = malloc(sizeof(*qgroup), M_GTASKQUEUE, M_WAITOK | M_ZERO);
805 	mtx_init(&qgroup->tqg_lock, "taskqgroup", NULL, MTX_DEF);
806 	qgroup->tqg_name = name;
807 	qgroup->tqg_cnt = cnt;
808 
809 	for (cpu = i = 0; i < cnt; i++) {
810 		taskqgroup_cpu_create(qgroup, i, cpu);
811 		for (j = 0; j < stride; j++)
812 			cpu = CPU_NEXT(cpu);
813 	}
814 	return (qgroup);
815 }
816 
817 void
818 taskqgroup_destroy(struct taskqgroup *qgroup)
819 {
820 
821 }
822 
823 void
824 taskqgroup_config_gtask_init(void *ctx, struct grouptask *gtask, gtask_fn_t *fn,
825     const char *name)
826 {
827 
828 	GROUPTASK_INIT(gtask, 0, fn, ctx);
829 	taskqgroup_attach(qgroup_config, gtask, gtask, NULL, NULL, name);
830 }
831 
832 void
833 taskqgroup_config_gtask_deinit(struct grouptask *gtask)
834 {
835 
836 	taskqgroup_detach(qgroup_config, gtask);
837 }
838