xref: /freebsd/sys/kern/subr_gtaskqueue.c (revision 61a74c5ccd65d1a00a96779f16eda8c41ff3a426)
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/sched.h>
45 #include <sys/smp.h>
46 #include <sys/gtaskqueue.h>
47 #include <sys/unistd.h>
48 #include <machine/stdarg.h>
49 
50 static MALLOC_DEFINE(M_GTASKQUEUE, "gtaskqueue", "Group Task Queues");
51 static void	gtaskqueue_thread_enqueue(void *);
52 static void	gtaskqueue_thread_loop(void *arg);
53 static int	task_is_running(struct gtaskqueue *queue, struct gtask *gtask);
54 static void	gtaskqueue_drain_locked(struct gtaskqueue *queue, struct gtask *gtask);
55 
56 TASKQGROUP_DEFINE(softirq, mp_ncpus, 1);
57 TASKQGROUP_DEFINE(config, 1, 1);
58 
59 struct gtaskqueue_busy {
60 	struct gtask		*tb_running;
61 	u_int			 tb_seq;
62 	LIST_ENTRY(gtaskqueue_busy) tb_link;
63 };
64 
65 typedef void (*gtaskqueue_enqueue_fn)(void *context);
66 
67 struct gtaskqueue {
68 	STAILQ_HEAD(, gtask)	tq_queue;
69 	LIST_HEAD(, gtaskqueue_busy) tq_active;
70 	u_int			tq_seq;
71 	int			tq_callouts;
72 	struct mtx_padalign	tq_mutex;
73 	gtaskqueue_enqueue_fn	tq_enqueue;
74 	void			*tq_context;
75 	char			*tq_name;
76 	struct thread		**tq_threads;
77 	int			tq_tcount;
78 	int			tq_spin;
79 	int			tq_flags;
80 	taskqueue_callback_fn	tq_callbacks[TASKQUEUE_NUM_CALLBACKS];
81 	void			*tq_cb_contexts[TASKQUEUE_NUM_CALLBACKS];
82 };
83 
84 #define	TQ_FLAGS_ACTIVE		(1 << 0)
85 #define	TQ_FLAGS_BLOCKED	(1 << 1)
86 #define	TQ_FLAGS_UNLOCKED_ENQUEUE	(1 << 2)
87 
88 #define	DT_CALLOUT_ARMED	(1 << 0)
89 
90 #define	TQ_LOCK(tq)							\
91 	do {								\
92 		if ((tq)->tq_spin)					\
93 			mtx_lock_spin(&(tq)->tq_mutex);			\
94 		else							\
95 			mtx_lock(&(tq)->tq_mutex);			\
96 	} while (0)
97 #define	TQ_ASSERT_LOCKED(tq)	mtx_assert(&(tq)->tq_mutex, MA_OWNED)
98 
99 #define	TQ_UNLOCK(tq)							\
100 	do {								\
101 		if ((tq)->tq_spin)					\
102 			mtx_unlock_spin(&(tq)->tq_mutex);		\
103 		else							\
104 			mtx_unlock(&(tq)->tq_mutex);			\
105 	} while (0)
106 #define	TQ_ASSERT_UNLOCKED(tq)	mtx_assert(&(tq)->tq_mutex, MA_NOTOWNED)
107 
108 #ifdef INVARIANTS
109 static void
110 gtask_dump(struct gtask *gtask)
111 {
112 	printf("gtask: %p ta_flags=%x ta_priority=%d ta_func=%p ta_context=%p\n",
113 	       gtask, gtask->ta_flags, gtask->ta_priority, gtask->ta_func, gtask->ta_context);
114 }
115 #endif
116 
117 static __inline int
118 TQ_SLEEP(struct gtaskqueue *tq, void *p, const char *wm)
119 {
120 	if (tq->tq_spin)
121 		return (msleep_spin(p, (struct mtx *)&tq->tq_mutex, wm, 0));
122 	return (msleep(p, &tq->tq_mutex, 0, wm, 0));
123 }
124 
125 static struct gtaskqueue *
126 _gtaskqueue_create(const char *name, int mflags,
127 		 taskqueue_enqueue_fn enqueue, void *context,
128 		 int mtxflags, const char *mtxname __unused)
129 {
130 	struct gtaskqueue *queue;
131 	char *tq_name;
132 
133 	tq_name = malloc(TASKQUEUE_NAMELEN, M_GTASKQUEUE, mflags | M_ZERO);
134 	if (!tq_name)
135 		return (NULL);
136 
137 	snprintf(tq_name, TASKQUEUE_NAMELEN, "%s", (name) ? name : "taskqueue");
138 
139 	queue = malloc(sizeof(struct gtaskqueue), M_GTASKQUEUE, mflags | M_ZERO);
140 	if (!queue) {
141 		free(tq_name, M_GTASKQUEUE);
142 		return (NULL);
143 	}
144 
145 	STAILQ_INIT(&queue->tq_queue);
146 	LIST_INIT(&queue->tq_active);
147 	queue->tq_enqueue = enqueue;
148 	queue->tq_context = context;
149 	queue->tq_name = tq_name;
150 	queue->tq_spin = (mtxflags & MTX_SPIN) != 0;
151 	queue->tq_flags |= TQ_FLAGS_ACTIVE;
152 	if (enqueue == gtaskqueue_thread_enqueue)
153 		queue->tq_flags |= TQ_FLAGS_UNLOCKED_ENQUEUE;
154 	mtx_init(&queue->tq_mutex, tq_name, NULL, mtxflags);
155 
156 	return (queue);
157 }
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
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 gtaskqueue_busy tb;
347 	struct gtask *gtask;
348 
349 	KASSERT(queue != NULL, ("tq is NULL"));
350 	TQ_ASSERT_LOCKED(queue);
351 	tb.tb_running = NULL;
352 	LIST_INSERT_HEAD(&queue->tq_active, &tb, tb_link);
353 
354 	while ((gtask = STAILQ_FIRST(&queue->tq_queue)) != NULL) {
355 		STAILQ_REMOVE_HEAD(&queue->tq_queue, ta_link);
356 		gtask->ta_flags &= ~TASK_ENQUEUED;
357 		tb.tb_running = gtask;
358 		tb.tb_seq = ++queue->tq_seq;
359 		TQ_UNLOCK(queue);
360 
361 		KASSERT(gtask->ta_func != NULL, ("task->ta_func is NULL"));
362 		gtask->ta_func(gtask->ta_context);
363 
364 		TQ_LOCK(queue);
365 		wakeup(gtask);
366 	}
367 	LIST_REMOVE(&tb, tb_link);
368 }
369 
370 static int
371 task_is_running(struct gtaskqueue *queue, struct gtask *gtask)
372 {
373 	struct gtaskqueue_busy *tb;
374 
375 	TQ_ASSERT_LOCKED(queue);
376 	LIST_FOREACH(tb, &queue->tq_active, tb_link) {
377 		if (tb->tb_running == gtask)
378 			return (1);
379 	}
380 	return (0);
381 }
382 
383 static int
384 gtaskqueue_cancel_locked(struct gtaskqueue *queue, struct gtask *gtask)
385 {
386 
387 	if (gtask->ta_flags & TASK_ENQUEUED)
388 		STAILQ_REMOVE(&queue->tq_queue, gtask, gtask, ta_link);
389 	gtask->ta_flags &= ~TASK_ENQUEUED;
390 	return (task_is_running(queue, gtask) ? EBUSY : 0);
391 }
392 
393 int
394 gtaskqueue_cancel(struct gtaskqueue *queue, struct gtask *gtask)
395 {
396 	int error;
397 
398 	TQ_LOCK(queue);
399 	error = gtaskqueue_cancel_locked(queue, gtask);
400 	TQ_UNLOCK(queue);
401 
402 	return (error);
403 }
404 
405 static void
406 gtaskqueue_drain_locked(struct gtaskqueue *queue, struct gtask *gtask)
407 {
408 	while ((gtask->ta_flags & TASK_ENQUEUED) || task_is_running(queue, gtask))
409 		TQ_SLEEP(queue, gtask, "gtq_drain");
410 }
411 
412 void
413 gtaskqueue_drain(struct gtaskqueue *queue, struct gtask *gtask)
414 {
415 
416 	if (!queue->tq_spin)
417 		WITNESS_WARN(WARN_GIANTOK | WARN_SLEEPOK, NULL, __func__);
418 
419 	TQ_LOCK(queue);
420 	gtaskqueue_drain_locked(queue, gtask);
421 	TQ_UNLOCK(queue);
422 }
423 
424 void
425 gtaskqueue_drain_all(struct gtaskqueue *queue)
426 {
427 
428 	if (!queue->tq_spin)
429 		WITNESS_WARN(WARN_GIANTOK | WARN_SLEEPOK, NULL, __func__);
430 
431 	TQ_LOCK(queue);
432 	gtaskqueue_drain_tq_queue(queue);
433 	gtaskqueue_drain_tq_active(queue);
434 	TQ_UNLOCK(queue);
435 }
436 
437 static int
438 _gtaskqueue_start_threads(struct gtaskqueue **tqp, int count, int pri,
439     cpuset_t *mask, const char *name, va_list ap)
440 {
441 	char ktname[MAXCOMLEN + 1];
442 	struct thread *td;
443 	struct gtaskqueue *tq;
444 	int i, error;
445 
446 	if (count <= 0)
447 		return (EINVAL);
448 
449 	vsnprintf(ktname, sizeof(ktname), name, ap);
450 	tq = *tqp;
451 
452 	tq->tq_threads = malloc(sizeof(struct thread *) * count, M_GTASKQUEUE,
453 	    M_NOWAIT | M_ZERO);
454 	if (tq->tq_threads == NULL) {
455 		printf("%s: no memory for %s threads\n", __func__, ktname);
456 		return (ENOMEM);
457 	}
458 
459 	for (i = 0; i < count; i++) {
460 		if (count == 1)
461 			error = kthread_add(gtaskqueue_thread_loop, tqp, NULL,
462 			    &tq->tq_threads[i], RFSTOPPED, 0, "%s", ktname);
463 		else
464 			error = kthread_add(gtaskqueue_thread_loop, tqp, NULL,
465 			    &tq->tq_threads[i], RFSTOPPED, 0,
466 			    "%s_%d", ktname, i);
467 		if (error) {
468 			/* should be ok to continue, taskqueue_free will dtrt */
469 			printf("%s: kthread_add(%s): error %d", __func__,
470 			    ktname, error);
471 			tq->tq_threads[i] = NULL;		/* paranoid */
472 		} else
473 			tq->tq_tcount++;
474 	}
475 	for (i = 0; i < count; i++) {
476 		if (tq->tq_threads[i] == NULL)
477 			continue;
478 		td = tq->tq_threads[i];
479 		if (mask) {
480 			error = cpuset_setthread(td->td_tid, mask);
481 			/*
482 			 * Failing to pin is rarely an actual fatal error;
483 			 * it'll just affect performance.
484 			 */
485 			if (error)
486 				printf("%s: curthread=%llu: can't pin; "
487 				    "error=%d\n",
488 				    __func__,
489 				    (unsigned long long) td->td_tid,
490 				    error);
491 		}
492 		thread_lock(td);
493 		sched_prio(td, pri);
494 		sched_add(td, SRQ_BORING);
495 	}
496 
497 	return (0);
498 }
499 
500 static int
501 gtaskqueue_start_threads(struct gtaskqueue **tqp, int count, int pri,
502     const char *name, ...)
503 {
504 	va_list ap;
505 	int error;
506 
507 	va_start(ap, name);
508 	error = _gtaskqueue_start_threads(tqp, count, pri, NULL, name, ap);
509 	va_end(ap);
510 	return (error);
511 }
512 
513 static inline void
514 gtaskqueue_run_callback(struct gtaskqueue *tq,
515     enum taskqueue_callback_type cb_type)
516 {
517 	taskqueue_callback_fn tq_callback;
518 
519 	TQ_ASSERT_UNLOCKED(tq);
520 	tq_callback = tq->tq_callbacks[cb_type];
521 	if (tq_callback != NULL)
522 		tq_callback(tq->tq_cb_contexts[cb_type]);
523 }
524 
525 static void
526 gtaskqueue_thread_loop(void *arg)
527 {
528 	struct gtaskqueue **tqp, *tq;
529 
530 	tqp = arg;
531 	tq = *tqp;
532 	gtaskqueue_run_callback(tq, TASKQUEUE_CALLBACK_TYPE_INIT);
533 	TQ_LOCK(tq);
534 	while ((tq->tq_flags & TQ_FLAGS_ACTIVE) != 0) {
535 		/* XXX ? */
536 		gtaskqueue_run_locked(tq);
537 		/*
538 		 * Because taskqueue_run() can drop tq_mutex, we need to
539 		 * check if the TQ_FLAGS_ACTIVE flag wasn't removed in the
540 		 * meantime, which means we missed a wakeup.
541 		 */
542 		if ((tq->tq_flags & TQ_FLAGS_ACTIVE) == 0)
543 			break;
544 		TQ_SLEEP(tq, tq, "-");
545 	}
546 	gtaskqueue_run_locked(tq);
547 	/*
548 	 * This thread is on its way out, so just drop the lock temporarily
549 	 * in order to call the shutdown callback.  This allows the callback
550 	 * to look at the taskqueue, even just before it dies.
551 	 */
552 	TQ_UNLOCK(tq);
553 	gtaskqueue_run_callback(tq, TASKQUEUE_CALLBACK_TYPE_SHUTDOWN);
554 	TQ_LOCK(tq);
555 
556 	/* rendezvous with thread that asked us to terminate */
557 	tq->tq_tcount--;
558 	wakeup_one(tq->tq_threads);
559 	TQ_UNLOCK(tq);
560 	kthread_exit();
561 }
562 
563 static void
564 gtaskqueue_thread_enqueue(void *context)
565 {
566 	struct gtaskqueue **tqp, *tq;
567 
568 	tqp = context;
569 	tq = *tqp;
570 	wakeup_any(tq);
571 }
572 
573 
574 static struct gtaskqueue *
575 gtaskqueue_create_fast(const char *name, int mflags,
576 		 taskqueue_enqueue_fn enqueue, void *context)
577 {
578 	return _gtaskqueue_create(name, mflags, enqueue, context,
579 			MTX_SPIN, "fast_taskqueue");
580 }
581 
582 
583 struct taskqgroup_cpu {
584 	LIST_HEAD(, grouptask)	tgc_tasks;
585 	struct gtaskqueue	*tgc_taskq;
586 	int	tgc_cnt;
587 	int	tgc_cpu;
588 };
589 
590 struct taskqgroup {
591 	struct taskqgroup_cpu tqg_queue[MAXCPU];
592 	struct mtx	tqg_lock;
593 	const char *	tqg_name;
594 	int		tqg_adjusting;
595 	int		tqg_stride;
596 	int		tqg_cnt;
597 };
598 
599 struct taskq_bind_task {
600 	struct gtask bt_task;
601 	int	bt_cpuid;
602 };
603 
604 static void
605 taskqgroup_cpu_create(struct taskqgroup *qgroup, int idx, int cpu)
606 {
607 	struct taskqgroup_cpu *qcpu;
608 
609 	qcpu = &qgroup->tqg_queue[idx];
610 	LIST_INIT(&qcpu->tgc_tasks);
611 	qcpu->tgc_taskq = gtaskqueue_create_fast(NULL, M_WAITOK,
612 	    taskqueue_thread_enqueue, &qcpu->tgc_taskq);
613 	gtaskqueue_start_threads(&qcpu->tgc_taskq, 1, PI_SOFT,
614 	    "%s_%d", qgroup->tqg_name, idx);
615 	qcpu->tgc_cpu = cpu;
616 }
617 
618 static void
619 taskqgroup_cpu_remove(struct taskqgroup *qgroup, int idx)
620 {
621 
622 	gtaskqueue_free(qgroup->tqg_queue[idx].tgc_taskq);
623 }
624 
625 /*
626  * Find the taskq with least # of tasks that doesn't currently have any
627  * other queues from the uniq identifier.
628  */
629 static int
630 taskqgroup_find(struct taskqgroup *qgroup, void *uniq)
631 {
632 	struct grouptask *n;
633 	int i, idx, mincnt;
634 	int strict;
635 
636 	mtx_assert(&qgroup->tqg_lock, MA_OWNED);
637 	if (qgroup->tqg_cnt == 0)
638 		return (0);
639 	idx = -1;
640 	mincnt = INT_MAX;
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 (strict = 1; mincnt == INT_MAX; strict = 0) {
647 		for (i = 0; i < qgroup->tqg_cnt; i++) {
648 			if (qgroup->tqg_queue[i].tgc_cnt > mincnt)
649 				continue;
650 			if (strict) {
651 				LIST_FOREACH(n,
652 				    &qgroup->tqg_queue[i].tgc_tasks, gt_list)
653 					if (n->gt_uniq == uniq)
654 						break;
655 				if (n != NULL)
656 					continue;
657 			}
658 			mincnt = qgroup->tqg_queue[i].tgc_cnt;
659 			idx = i;
660 		}
661 	}
662 	if (idx == -1)
663 		panic("%s: failed to pick a qid.", __func__);
664 
665 	return (idx);
666 }
667 
668 /*
669  * smp_started is unusable since it is not set for UP kernels or even for
670  * SMP kernels when there is 1 CPU.  This is usually handled by adding a
671  * (mp_ncpus == 1) test, but that would be broken here since we need to
672  * to synchronize with the SI_SUB_SMP ordering.  Even in the pure SMP case
673  * smp_started only gives a fuzzy ordering relative to SI_SUB_SMP.
674  *
675  * So maintain our own flag.  It must be set after all CPUs are started
676  * and before SI_SUB_SMP:SI_ORDER_ANY so that the SYSINIT for delayed
677  * adjustment is properly delayed.  SI_ORDER_FOURTH is clearly before
678  * SI_ORDER_ANY and unclearly after the CPUs are started.  It would be
679  * simpler for adjustment to pass a flag indicating if it is delayed.
680  */
681 
682 static int tqg_smp_started;
683 
684 static void
685 tqg_record_smp_started(void *arg)
686 {
687 	tqg_smp_started = 1;
688 }
689 
690 SYSINIT(tqg_record_smp_started, SI_SUB_SMP, SI_ORDER_FOURTH,
691 	tqg_record_smp_started, NULL);
692 
693 void
694 taskqgroup_attach(struct taskqgroup *qgroup, struct grouptask *gtask,
695     void *uniq, device_t dev, struct resource *irq, const char *name)
696 {
697 	int cpu, qid, error;
698 
699 	gtask->gt_uniq = uniq;
700 	snprintf(gtask->gt_name, GROUPTASK_NAMELEN, "%s", name ? name : "grouptask");
701 	gtask->gt_dev = dev;
702 	gtask->gt_irq = irq;
703 	gtask->gt_cpu = -1;
704 	mtx_lock(&qgroup->tqg_lock);
705 	qid = taskqgroup_find(qgroup, uniq);
706 	qgroup->tqg_queue[qid].tgc_cnt++;
707 	LIST_INSERT_HEAD(&qgroup->tqg_queue[qid].tgc_tasks, gtask, gt_list);
708 	gtask->gt_taskqueue = qgroup->tqg_queue[qid].tgc_taskq;
709 	if (dev != NULL && irq != NULL && tqg_smp_started) {
710 		cpu = qgroup->tqg_queue[qid].tgc_cpu;
711 		gtask->gt_cpu = cpu;
712 		mtx_unlock(&qgroup->tqg_lock);
713 		error = bus_bind_intr(dev, irq, cpu);
714 		if (error)
715 			printf("%s: binding interrupt failed for %s: %d\n",
716 			    __func__, gtask->gt_name, error);
717 	} else
718 		mtx_unlock(&qgroup->tqg_lock);
719 }
720 
721 static void
722 taskqgroup_attach_deferred(struct taskqgroup *qgroup, struct grouptask *gtask)
723 {
724 	int qid, cpu, error;
725 
726 	mtx_lock(&qgroup->tqg_lock);
727 	qid = taskqgroup_find(qgroup, gtask->gt_uniq);
728 	cpu = qgroup->tqg_queue[qid].tgc_cpu;
729 	if (gtask->gt_dev != NULL && gtask->gt_irq != NULL) {
730 		mtx_unlock(&qgroup->tqg_lock);
731 		error = bus_bind_intr(gtask->gt_dev, gtask->gt_irq, cpu);
732 		mtx_lock(&qgroup->tqg_lock);
733 		if (error)
734 			printf("%s: binding interrupt failed for %s: %d\n",
735 			    __func__, gtask->gt_name, error);
736 
737 	}
738 	qgroup->tqg_queue[qid].tgc_cnt++;
739 	LIST_INSERT_HEAD(&qgroup->tqg_queue[qid].tgc_tasks, gtask, gt_list);
740 	MPASS(qgroup->tqg_queue[qid].tgc_taskq != NULL);
741 	gtask->gt_taskqueue = qgroup->tqg_queue[qid].tgc_taskq;
742 	mtx_unlock(&qgroup->tqg_lock);
743 }
744 
745 int
746 taskqgroup_attach_cpu(struct taskqgroup *qgroup, struct grouptask *gtask,
747     void *uniq, int cpu, device_t dev, struct resource *irq, const char *name)
748 {
749 	int i, qid, error;
750 
751 	qid = -1;
752 	gtask->gt_uniq = uniq;
753 	snprintf(gtask->gt_name, GROUPTASK_NAMELEN, "%s", name ? name : "grouptask");
754 	gtask->gt_dev = dev;
755 	gtask->gt_irq = irq;
756 	gtask->gt_cpu = cpu;
757 	mtx_lock(&qgroup->tqg_lock);
758 	if (tqg_smp_started) {
759 		for (i = 0; i < qgroup->tqg_cnt; i++)
760 			if (qgroup->tqg_queue[i].tgc_cpu == cpu) {
761 				qid = i;
762 				break;
763 			}
764 		if (qid == -1) {
765 			mtx_unlock(&qgroup->tqg_lock);
766 			printf("%s: qid not found for %s cpu=%d\n", __func__, gtask->gt_name, cpu);
767 			return (EINVAL);
768 		}
769 	} else
770 		qid = 0;
771 	qgroup->tqg_queue[qid].tgc_cnt++;
772 	LIST_INSERT_HEAD(&qgroup->tqg_queue[qid].tgc_tasks, gtask, gt_list);
773 	gtask->gt_taskqueue = qgroup->tqg_queue[qid].tgc_taskq;
774 	cpu = qgroup->tqg_queue[qid].tgc_cpu;
775 	mtx_unlock(&qgroup->tqg_lock);
776 
777 	if (dev != NULL && irq != NULL && tqg_smp_started) {
778 		error = bus_bind_intr(dev, irq, cpu);
779 		if (error)
780 			printf("%s: binding interrupt failed for %s: %d\n",
781 			    __func__, gtask->gt_name, error);
782 	}
783 	return (0);
784 }
785 
786 static int
787 taskqgroup_attach_cpu_deferred(struct taskqgroup *qgroup, struct grouptask *gtask)
788 {
789 	device_t dev;
790 	struct resource *irq;
791 	int cpu, error, i, qid;
792 
793 	qid = -1;
794 	dev = gtask->gt_dev;
795 	irq = gtask->gt_irq;
796 	cpu = gtask->gt_cpu;
797 	MPASS(tqg_smp_started);
798 	mtx_lock(&qgroup->tqg_lock);
799 	for (i = 0; i < qgroup->tqg_cnt; i++)
800 		if (qgroup->tqg_queue[i].tgc_cpu == cpu) {
801 			qid = i;
802 			break;
803 		}
804 	if (qid == -1) {
805 		mtx_unlock(&qgroup->tqg_lock);
806 		printf("%s: qid not found for %s cpu=%d\n", __func__, gtask->gt_name, cpu);
807 		return (EINVAL);
808 	}
809 	qgroup->tqg_queue[qid].tgc_cnt++;
810 	LIST_INSERT_HEAD(&qgroup->tqg_queue[qid].tgc_tasks, gtask, gt_list);
811 	MPASS(qgroup->tqg_queue[qid].tgc_taskq != NULL);
812 	gtask->gt_taskqueue = qgroup->tqg_queue[qid].tgc_taskq;
813 	mtx_unlock(&qgroup->tqg_lock);
814 
815 	if (dev != NULL && irq != NULL) {
816 		error = bus_bind_intr(dev, irq, cpu);
817 		if (error)
818 			printf("%s: binding interrupt failed for %s: %d\n",
819 			    __func__, gtask->gt_name, error);
820 	}
821 	return (0);
822 }
823 
824 void
825 taskqgroup_detach(struct taskqgroup *qgroup, struct grouptask *gtask)
826 {
827 	int i;
828 
829 	grouptask_block(gtask);
830 	mtx_lock(&qgroup->tqg_lock);
831 	for (i = 0; i < qgroup->tqg_cnt; i++)
832 		if (qgroup->tqg_queue[i].tgc_taskq == gtask->gt_taskqueue)
833 			break;
834 	if (i == qgroup->tqg_cnt)
835 		panic("%s: task %s not in group", __func__, gtask->gt_name);
836 	qgroup->tqg_queue[i].tgc_cnt--;
837 	LIST_REMOVE(gtask, gt_list);
838 	mtx_unlock(&qgroup->tqg_lock);
839 	gtask->gt_taskqueue = NULL;
840 	gtask->gt_task.ta_flags &= ~TASK_NOENQUEUE;
841 }
842 
843 static void
844 taskqgroup_binder(void *ctx)
845 {
846 	struct taskq_bind_task *gtask = (struct taskq_bind_task *)ctx;
847 	cpuset_t mask;
848 	int error;
849 
850 	CPU_ZERO(&mask);
851 	CPU_SET(gtask->bt_cpuid, &mask);
852 	error = cpuset_setthread(curthread->td_tid, &mask);
853 	thread_lock(curthread);
854 	sched_bind(curthread, gtask->bt_cpuid);
855 	thread_unlock(curthread);
856 
857 	if (error)
858 		printf("%s: binding curthread failed: %d\n", __func__, error);
859 	free(gtask, M_DEVBUF);
860 }
861 
862 static void
863 taskqgroup_bind(struct taskqgroup *qgroup)
864 {
865 	struct taskq_bind_task *gtask;
866 	int i;
867 
868 	/*
869 	 * Bind taskqueue threads to specific CPUs, if they have been assigned
870 	 * one.
871 	 */
872 	if (qgroup->tqg_cnt == 1)
873 		return;
874 
875 	for (i = 0; i < qgroup->tqg_cnt; i++) {
876 		gtask = malloc(sizeof (*gtask), M_DEVBUF, M_WAITOK);
877 		GTASK_INIT(&gtask->bt_task, 0, 0, taskqgroup_binder, gtask);
878 		gtask->bt_cpuid = qgroup->tqg_queue[i].tgc_cpu;
879 		grouptaskqueue_enqueue(qgroup->tqg_queue[i].tgc_taskq,
880 		    &gtask->bt_task);
881 	}
882 }
883 
884 static void
885 taskqgroup_config_init(void *arg)
886 {
887 	struct taskqgroup *qgroup = qgroup_config;
888 	LIST_HEAD(, grouptask) gtask_head = LIST_HEAD_INITIALIZER(NULL);
889 
890 	LIST_SWAP(&gtask_head, &qgroup->tqg_queue[0].tgc_tasks,
891 	    grouptask, gt_list);
892 	qgroup->tqg_queue[0].tgc_cnt = 0;
893 	taskqgroup_cpu_create(qgroup, 0, 0);
894 
895 	qgroup->tqg_cnt = 1;
896 	qgroup->tqg_stride = 1;
897 }
898 
899 SYSINIT(taskqgroup_config_init, SI_SUB_TASKQ, SI_ORDER_SECOND,
900 	taskqgroup_config_init, NULL);
901 
902 static int
903 _taskqgroup_adjust(struct taskqgroup *qgroup, int cnt, int stride)
904 {
905 	LIST_HEAD(, grouptask) gtask_head = LIST_HEAD_INITIALIZER(NULL);
906 	struct grouptask *gtask;
907 	int i, k, old_cnt, old_cpu, cpu;
908 
909 	mtx_assert(&qgroup->tqg_lock, MA_OWNED);
910 
911 	if (cnt < 1 || cnt * stride > mp_ncpus || !tqg_smp_started) {
912 		printf("%s: failed cnt: %d stride: %d "
913 		    "mp_ncpus: %d tqg_smp_started: %d\n",
914 		    __func__, cnt, stride, mp_ncpus, tqg_smp_started);
915 		return (EINVAL);
916 	}
917 	if (qgroup->tqg_adjusting) {
918 		printf("%s failed: adjusting\n", __func__);
919 		return (EBUSY);
920 	}
921 	qgroup->tqg_adjusting = 1;
922 	old_cnt = qgroup->tqg_cnt;
923 	old_cpu = 0;
924 	if (old_cnt < cnt)
925 		old_cpu = qgroup->tqg_queue[old_cnt].tgc_cpu;
926 	mtx_unlock(&qgroup->tqg_lock);
927 	/*
928 	 * Set up queue for tasks added before boot.
929 	 */
930 	if (old_cnt == 0) {
931 		LIST_SWAP(&gtask_head, &qgroup->tqg_queue[0].tgc_tasks,
932 		    grouptask, gt_list);
933 		qgroup->tqg_queue[0].tgc_cnt = 0;
934 	}
935 
936 	/*
937 	 * If new taskq threads have been added.
938 	 */
939 	cpu = old_cpu;
940 	for (i = old_cnt; i < cnt; i++) {
941 		taskqgroup_cpu_create(qgroup, i, cpu);
942 
943 		for (k = 0; k < stride; k++)
944 			cpu = CPU_NEXT(cpu);
945 	}
946 	mtx_lock(&qgroup->tqg_lock);
947 	qgroup->tqg_cnt = cnt;
948 	qgroup->tqg_stride = stride;
949 
950 	/*
951 	 * Adjust drivers to use new taskqs.
952 	 */
953 	for (i = 0; i < old_cnt; i++) {
954 		while ((gtask = LIST_FIRST(&qgroup->tqg_queue[i].tgc_tasks))) {
955 			LIST_REMOVE(gtask, gt_list);
956 			qgroup->tqg_queue[i].tgc_cnt--;
957 			LIST_INSERT_HEAD(&gtask_head, gtask, gt_list);
958 		}
959 	}
960 	mtx_unlock(&qgroup->tqg_lock);
961 
962 	while ((gtask = LIST_FIRST(&gtask_head))) {
963 		LIST_REMOVE(gtask, gt_list);
964 		if (gtask->gt_cpu == -1)
965 			taskqgroup_attach_deferred(qgroup, gtask);
966 		else if (taskqgroup_attach_cpu_deferred(qgroup, gtask))
967 			taskqgroup_attach_deferred(qgroup, gtask);
968 	}
969 
970 #ifdef INVARIANTS
971 	mtx_lock(&qgroup->tqg_lock);
972 	for (i = 0; i < qgroup->tqg_cnt; i++) {
973 		MPASS(qgroup->tqg_queue[i].tgc_taskq != NULL);
974 		LIST_FOREACH(gtask, &qgroup->tqg_queue[i].tgc_tasks, gt_list)
975 			MPASS(gtask->gt_taskqueue != NULL);
976 	}
977 	mtx_unlock(&qgroup->tqg_lock);
978 #endif
979 	/*
980 	 * If taskq thread count has been reduced.
981 	 */
982 	for (i = cnt; i < old_cnt; i++)
983 		taskqgroup_cpu_remove(qgroup, i);
984 
985 	taskqgroup_bind(qgroup);
986 
987 	mtx_lock(&qgroup->tqg_lock);
988 	qgroup->tqg_adjusting = 0;
989 
990 	return (0);
991 }
992 
993 int
994 taskqgroup_adjust(struct taskqgroup *qgroup, int cnt, int stride)
995 {
996 	int error;
997 
998 	mtx_lock(&qgroup->tqg_lock);
999 	error = _taskqgroup_adjust(qgroup, cnt, stride);
1000 	mtx_unlock(&qgroup->tqg_lock);
1001 
1002 	return (error);
1003 }
1004 
1005 struct taskqgroup *
1006 taskqgroup_create(const char *name)
1007 {
1008 	struct taskqgroup *qgroup;
1009 
1010 	qgroup = malloc(sizeof(*qgroup), M_GTASKQUEUE, M_WAITOK | M_ZERO);
1011 	mtx_init(&qgroup->tqg_lock, "taskqgroup", NULL, MTX_DEF);
1012 	qgroup->tqg_name = name;
1013 	LIST_INIT(&qgroup->tqg_queue[0].tgc_tasks);
1014 
1015 	return (qgroup);
1016 }
1017 
1018 void
1019 taskqgroup_destroy(struct taskqgroup *qgroup)
1020 {
1021 
1022 }
1023 
1024 void
1025 taskqgroup_config_gtask_init(void *ctx, struct grouptask *gtask, gtask_fn_t *fn,
1026     const char *name)
1027 {
1028 
1029 	GROUPTASK_INIT(gtask, 0, fn, ctx);
1030 	taskqgroup_attach(qgroup_config, gtask, gtask, NULL, NULL, name);
1031 }
1032 
1033 void
1034 taskqgroup_config_gtask_deinit(struct grouptask *gtask)
1035 {
1036 
1037 	taskqgroup_detach(qgroup_config, gtask);
1038 }
1039