1 /*
2 * CDDL HEADER START
3 *
4 * The contents of this file are subject to the terms of the
5 * Common Development and Distribution License (the "License").
6 * You may not use this file except in compliance with the License.
7 *
8 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
9 * or http://www.opensolaris.org/os/licensing.
10 * See the License for the specific language governing permissions
11 * and limitations under the License.
12 *
13 * When distributing Covered Code, include this CDDL HEADER in each
14 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
15 * If applicable, add the following below this CDDL HEADER, with the
16 * fields enclosed by brackets "[]" replaced with your own identifying
17 * information: Portions Copyright [yyyy] [name of copyright owner]
18 *
19 * CDDL HEADER END
20 */
21 /*
22 * Copyright 2010 Sun Microsystems, Inc. All rights reserved.
23 * Use is subject to license terms.
24 */
25
26 /*
27 * Copyright 2015 Nexenta Systems, Inc. All rights reserved.
28 * Copyright (c) 2017 by Delphix. All rights reserved.
29 * Copyright (c) 2017, Joyent, Inc.
30 */
31
32 /*
33 * Kernel task queues: general-purpose asynchronous task scheduling.
34 *
35 * A common problem in kernel programming is the need to schedule tasks
36 * to be performed later, by another thread. There are several reasons
37 * you may want or need to do this:
38 *
39 * (1) The task isn't time-critical, but your current code path is.
40 *
41 * (2) The task may require grabbing locks that you already hold.
42 *
43 * (3) The task may need to block (e.g. to wait for memory), but you
44 * cannot block in your current context.
45 *
46 * (4) Your code path can't complete because of some condition, but you can't
47 * sleep or fail, so you queue the task for later execution when condition
48 * disappears.
49 *
50 * (5) You just want a simple way to launch multiple tasks in parallel.
51 *
52 * Task queues provide such a facility. In its simplest form (used when
53 * performance is not a critical consideration) a task queue consists of a
54 * single list of tasks, together with one or more threads to service the
55 * list. There are some cases when this simple queue is not sufficient:
56 *
57 * (1) The task queues are very hot and there is a need to avoid data and lock
58 * contention over global resources.
59 *
60 * (2) Some tasks may depend on other tasks to complete, so they can't be put in
61 * the same list managed by the same thread.
62 *
63 * (3) Some tasks may block for a long time, and this should not block other
64 * tasks in the queue.
65 *
66 * To provide useful service in such cases we define a "dynamic task queue"
67 * which has an individual thread for each of the tasks. These threads are
68 * dynamically created as they are needed and destroyed when they are not in
69 * use. The API for managing task pools is the same as for managing task queues
70 * with the exception of a taskq creation flag TASKQ_DYNAMIC which tells that
71 * dynamic task pool behavior is desired.
72 *
73 * Dynamic task queues may also place tasks in the normal queue (called "backing
74 * queue") when task pool runs out of resources. Users of task queues may
75 * disallow such queued scheduling by specifying TQ_NOQUEUE in the dispatch
76 * flags.
77 *
78 * The backing task queue is also used for scheduling internal tasks needed for
79 * dynamic task queue maintenance.
80 *
81 * INTERFACES ==================================================================
82 *
83 * taskq_t *taskq_create(name, nthreads, pri, minalloc, maxalloc, flags);
84 *
85 * Create a taskq with specified properties.
86 * Possible 'flags':
87 *
88 * TASKQ_DYNAMIC: Create task pool for task management. If this flag is
89 * specified, 'nthreads' specifies the maximum number of threads in
90 * the task queue. Task execution order for dynamic task queues is
91 * not predictable.
92 *
93 * If this flag is not specified (default case) a
94 * single-list task queue is created with 'nthreads' threads
95 * servicing it. Entries in this queue are managed by
96 * taskq_ent_alloc() and taskq_ent_free() which try to keep the
97 * task population between 'minalloc' and 'maxalloc', but the
98 * latter limit is only advisory for TQ_SLEEP dispatches and the
99 * former limit is only advisory for TQ_NOALLOC dispatches. If
100 * TASKQ_PREPOPULATE is set in 'flags', the taskq will be
101 * prepopulated with 'minalloc' task structures.
102 *
103 * Since non-DYNAMIC taskqs are queues, tasks are guaranteed to be
104 * executed in the order they are scheduled if nthreads == 1.
105 * If nthreads > 1, task execution order is not predictable.
106 *
107 * TASKQ_PREPOPULATE: Prepopulate task queue with threads.
108 * Also prepopulate the task queue with 'minalloc' task structures.
109 *
110 * TASKQ_THREADS_CPU_PCT: This flag specifies that 'nthreads' should be
111 * interpreted as a percentage of the # of online CPUs on the
112 * system. The taskq subsystem will automatically adjust the
113 * number of threads in the taskq in response to CPU online
114 * and offline events, to keep the ratio. nthreads must be in
115 * the range [0,100].
116 *
117 * The calculation used is:
118 *
119 * MAX((ncpus_online * percentage)/100, 1)
120 *
121 * This flag is not supported for DYNAMIC task queues.
122 * This flag is not compatible with TASKQ_CPR_SAFE.
123 *
124 * TASKQ_CPR_SAFE: This flag specifies that users of the task queue will
125 * use their own protocol for handling CPR issues. This flag is not
126 * supported for DYNAMIC task queues. This flag is not compatible
127 * with TASKQ_THREADS_CPU_PCT.
128 *
129 * The 'pri' field specifies the default priority for the threads that
130 * service all scheduled tasks.
131 *
132 * taskq_t *taskq_create_instance(name, instance, nthreads, pri, minalloc,
133 * maxalloc, flags);
134 *
135 * Like taskq_create(), but takes an instance number (or -1 to indicate
136 * no instance).
137 *
138 * taskq_t *taskq_create_proc(name, nthreads, pri, minalloc, maxalloc, proc,
139 * flags);
140 *
141 * Like taskq_create(), but creates the taskq threads in the specified
142 * system process. If proc != &p0, this must be called from a thread
143 * in that process.
144 *
145 * taskq_t *taskq_create_sysdc(name, nthreads, minalloc, maxalloc, proc,
146 * dc, flags);
147 *
148 * Like taskq_create_proc(), but the taskq threads will use the
149 * System Duty Cycle (SDC) scheduling class with a duty cycle of dc.
150 *
151 * void taskq_destroy(tap):
152 *
153 * Waits for any scheduled tasks to complete, then destroys the taskq.
154 * Caller should guarantee that no new tasks are scheduled in the closing
155 * taskq.
156 *
157 * taskqid_t taskq_dispatch(tq, func, arg, flags):
158 *
159 * Dispatches the task "func(arg)" to taskq. The 'flags' indicates whether
160 * the caller is willing to block for memory. The function returns an
161 * opaque value which is zero iff dispatch fails. If flags is TQ_NOSLEEP
162 * or TQ_NOALLOC and the task can't be dispatched, taskq_dispatch() fails
163 * and returns (taskqid_t)0.
164 *
165 * ASSUMES: func != NULL.
166 *
167 * Possible flags:
168 * TQ_NOSLEEP: Do not wait for resources; may fail.
169 *
170 * TQ_NOALLOC: Do not allocate memory; may fail. May only be used with
171 * non-dynamic task queues.
172 *
173 * TQ_NOQUEUE: Do not enqueue a task if it can't dispatch it due to
174 * lack of available resources and fail. If this flag is not
175 * set, and the task pool is exhausted, the task may be scheduled
176 * in the backing queue. This flag may ONLY be used with dynamic
177 * task queues.
178 *
179 * NOTE: This flag should always be used when a task queue is used
180 * for tasks that may depend on each other for completion.
181 * Enqueueing dependent tasks may create deadlocks.
182 *
183 * TQ_SLEEP: May block waiting for resources. May still fail for
184 * dynamic task queues if TQ_NOQUEUE is also specified, otherwise
185 * always succeed.
186 *
187 * TQ_FRONT: Puts the new task at the front of the queue. Be careful.
188 *
189 * NOTE: Dynamic task queues are much more likely to fail in
190 * taskq_dispatch() (especially if TQ_NOQUEUE was specified), so it
191 * is important to have backup strategies handling such failures.
192 *
193 * void taskq_dispatch_ent(tq, func, arg, flags, tqent)
194 *
195 * This is a light-weight form of taskq_dispatch(), that uses a
196 * preallocated taskq_ent_t structure for scheduling. As a
197 * result, it does not perform allocations and cannot ever fail.
198 * Note especially that it cannot be used with TASKQ_DYNAMIC
199 * taskqs. The memory for the tqent must not be modified or used
200 * until the function (func) is called. (However, func itself
201 * may safely modify or free this memory, once it is called.)
202 * Note that the taskq framework will NOT free this memory.
203 *
204 * void taskq_wait(tq):
205 *
206 * Waits for all previously scheduled tasks to complete.
207 *
208 * NOTE: It does not stop any new task dispatches.
209 * Do NOT call taskq_wait() from a task: it will cause deadlock.
210 *
211 * void taskq_suspend(tq)
212 *
213 * Suspend all task execution. Tasks already scheduled for a dynamic task
214 * queue will still be executed, but all new scheduled tasks will be
215 * suspended until taskq_resume() is called.
216 *
217 * int taskq_suspended(tq)
218 *
219 * Returns 1 if taskq is suspended and 0 otherwise. It is intended to
220 * ASSERT that the task queue is suspended.
221 *
222 * void taskq_resume(tq)
223 *
224 * Resume task queue execution.
225 *
226 * int taskq_member(tq, thread)
227 *
228 * Returns 1 if 'thread' belongs to taskq 'tq' and 0 otherwise. The
229 * intended use is to ASSERT that a given function is called in taskq
230 * context only.
231 *
232 * system_taskq
233 *
234 * Global system-wide dynamic task queue for common uses. It may be used by
235 * any subsystem that needs to schedule tasks and does not need to manage
236 * its own task queues. It is initialized quite early during system boot.
237 *
238 * IMPLEMENTATION ==============================================================
239 *
240 * This is schematic representation of the task queue structures.
241 *
242 * taskq:
243 * +-------------+
244 * | tq_lock | +---< taskq_ent_free()
245 * +-------------+ |
246 * |... | | tqent: tqent:
247 * +-------------+ | +------------+ +------------+
248 * | tq_freelist |-->| tqent_next |--> ... ->| tqent_next |
249 * +-------------+ +------------+ +------------+
250 * |... | | ... | | ... |
251 * +-------------+ +------------+ +------------+
252 * | tq_task | |
253 * | | +-------------->taskq_ent_alloc()
254 * +--------------------------------------------------------------------------+
255 * | | | tqent tqent |
256 * | +---------------------+ +--> +------------+ +--> +------------+ |
257 * | | ... | | | func, arg | | | func, arg | |
258 * +>+---------------------+ <---|-+ +------------+ <---|-+ +------------+ |
259 * | tq_taskq.tqent_next | ----+ | | tqent_next | --->+ | | tqent_next |--+
260 * +---------------------+ | +------------+ ^ | +------------+
261 * +-| tq_task.tqent_prev | +--| tqent_prev | | +--| tqent_prev | ^
262 * | +---------------------+ +------------+ | +------------+ |
263 * | |... | | ... | | | ... | |
264 * | +---------------------+ +------------+ | +------------+ |
265 * | ^ | |
266 * | | | |
267 * +--------------------------------------+--------------+ TQ_APPEND() -+
268 * | | |
269 * |... | taskq_thread()-----+
270 * +-------------+
271 * | tq_buckets |--+-------> [ NULL ] (for regular task queues)
272 * +-------------+ |
273 * | DYNAMIC TASK QUEUES:
274 * |
275 * +-> taskq_bucket[nCPU] taskq_bucket_dispatch()
276 * +-------------------+ ^
277 * +--->| tqbucket_lock | |
278 * | +-------------------+ +--------+ +--------+
279 * | | tqbucket_freelist |-->| tqent |-->...| tqent | ^
280 * | +-------------------+<--+--------+<--...+--------+ |
281 * | | ... | | thread | | thread | |
282 * | +-------------------+ +--------+ +--------+ |
283 * | +-------------------+ |
284 * taskq_dispatch()--+--->| tqbucket_lock | TQ_APPEND()------+
285 * TQ_HASH() | +-------------------+ +--------+ +--------+
286 * | | tqbucket_freelist |-->| tqent |-->...| tqent |
287 * | +-------------------+<--+--------+<--...+--------+
288 * | | ... | | thread | | thread |
289 * | +-------------------+ +--------+ +--------+
290 * +---> ...
291 *
292 *
293 * Task queues use tq_task field to link new entry in the queue. The queue is a
294 * circular doubly-linked list. Entries are put in the end of the list with
295 * TQ_APPEND() and processed from the front of the list by taskq_thread() in
296 * FIFO order. Task queue entries are cached in the free list managed by
297 * taskq_ent_alloc() and taskq_ent_free() functions.
298 *
299 * All threads used by task queues mark t_taskq field of the thread to
300 * point to the task queue.
301 *
302 * Taskq Thread Management -----------------------------------------------------
303 *
304 * Taskq's non-dynamic threads are managed with several variables and flags:
305 *
306 * * tq_nthreads - The number of threads in taskq_thread() for the
307 * taskq.
308 *
309 * * tq_active - The number of threads not waiting on a CV in
310 * taskq_thread(); includes newly created threads
311 * not yet counted in tq_nthreads.
312 *
313 * * tq_nthreads_target
314 * - The number of threads desired for the taskq.
315 *
316 * * tq_flags & TASKQ_CHANGING
317 * - Indicates that tq_nthreads != tq_nthreads_target.
318 *
319 * * tq_flags & TASKQ_THREAD_CREATED
320 * - Indicates that a thread is being created in the taskq.
321 *
322 * During creation, tq_nthreads and tq_active are set to 0, and
323 * tq_nthreads_target is set to the number of threads desired. The
324 * TASKQ_CHANGING flag is set, and taskq_thread_create() is called to
325 * create the first thread. taskq_thread_create() increments tq_active,
326 * sets TASKQ_THREAD_CREATED, and creates the new thread.
327 *
328 * Each thread starts in taskq_thread(), clears the TASKQ_THREAD_CREATED
329 * flag, and increments tq_nthreads. It stores the new value of
330 * tq_nthreads as its "thread_id", and stores its thread pointer in the
331 * tq_threadlist at the (thread_id - 1). We keep the thread_id space
332 * densely packed by requiring that only the largest thread_id can exit during
333 * normal adjustment. The exception is during the destruction of the
334 * taskq; once tq_nthreads_target is set to zero, no new threads will be created
335 * for the taskq queue, so every thread can exit without any ordering being
336 * necessary.
337 *
338 * Threads will only process work if their thread id is <= tq_nthreads_target.
339 *
340 * When TASKQ_CHANGING is set, threads will check the current thread target
341 * whenever they wake up, and do whatever they can to apply its effects.
342 *
343 * TASKQ_THREAD_CPU_PCT --------------------------------------------------------
344 *
345 * When a taskq is created with TASKQ_THREAD_CPU_PCT, we store their requested
346 * percentage in tq_threads_ncpus_pct, start them off with the correct thread
347 * target, and add them to the taskq_cpupct_list for later adjustment.
348 *
349 * We register taskq_cpu_setup() to be called whenever a CPU changes state. It
350 * walks the list of TASKQ_THREAD_CPU_PCT taskqs, adjusts their nthread_target
351 * if need be, and wakes up all of the threads to process the change.
352 *
353 * Dynamic Task Queues Implementation ------------------------------------------
354 *
355 * For a dynamic task queues there is a 1-to-1 mapping between a thread and
356 * taskq_ent_structure. Each entry is serviced by its own thread and each thread
357 * is controlled by a single entry.
358 *
359 * Entries are distributed over a set of buckets. To avoid using modulo
360 * arithmetics the number of buckets is 2^n and is determined as the nearest
361 * power of two roundown of the number of CPUs in the system. Tunable
362 * variable 'taskq_maxbuckets' limits the maximum number of buckets. Each entry
363 * is attached to a bucket for its lifetime and can't migrate to other buckets.
364 *
365 * Entries that have scheduled tasks are not placed in any list. The dispatch
366 * function sets their "func" and "arg" fields and signals the corresponding
367 * thread to execute the task. Once the thread executes the task it clears the
368 * "func" field and places an entry on the bucket cache of free entries pointed
369 * by "tqbucket_freelist" field. ALL entries on the free list should have "func"
370 * field equal to NULL. The free list is a circular doubly-linked list identical
371 * in structure to the tq_task list above, but entries are taken from it in LIFO
372 * order - the last freed entry is the first to be allocated. The
373 * taskq_bucket_dispatch() function gets the most recently used entry from the
374 * free list, sets its "func" and "arg" fields and signals a worker thread.
375 *
376 * After executing each task a per-entry thread taskq_d_thread() places its
377 * entry on the bucket free list and goes to a timed sleep. If it wakes up
378 * without getting new task it removes the entry from the free list and destroys
379 * itself. The thread sleep time is controlled by a tunable variable
380 * `taskq_thread_timeout'.
381 *
382 * There are various statistics kept in the bucket which allows for later
383 * analysis of taskq usage patterns. Also, a global copy of taskq creation and
384 * death statistics is kept in the global taskq data structure. Since thread
385 * creation and death happen rarely, updating such global data does not present
386 * a performance problem.
387 *
388 * NOTE: Threads are not bound to any CPU and there is absolutely no association
389 * between the bucket and actual thread CPU, so buckets are used only to
390 * split resources and reduce resource contention. Having threads attached
391 * to the CPU denoted by a bucket may reduce number of times the job
392 * switches between CPUs.
393 *
394 * Current algorithm creates a thread whenever a bucket has no free
395 * entries. It would be nice to know how many threads are in the running
396 * state and don't create threads if all CPUs are busy with existing
397 * tasks, but it is unclear how such strategy can be implemented.
398 *
399 * Currently buckets are created statically as an array attached to task
400 * queue. On some system with nCPUs < max_ncpus it may waste system
401 * memory. One solution may be allocation of buckets when they are first
402 * touched, but it is not clear how useful it is.
403 *
404 * SUSPEND/RESUME implementation -----------------------------------------------
405 *
406 * Before executing a task taskq_thread() (executing non-dynamic task
407 * queues) obtains taskq's thread lock as a reader. The taskq_suspend()
408 * function gets the same lock as a writer blocking all non-dynamic task
409 * execution. The taskq_resume() function releases the lock allowing
410 * taskq_thread to continue execution.
411 *
412 * For dynamic task queues, each bucket is marked as TQBUCKET_SUSPEND by
413 * taskq_suspend() function. After that taskq_bucket_dispatch() always
414 * fails, so that taskq_dispatch() will either enqueue tasks for a
415 * suspended backing queue or fail if TQ_NOQUEUE is specified in dispatch
416 * flags.
417 *
418 * NOTE: taskq_suspend() does not immediately block any tasks already
419 * scheduled for dynamic task queues. It only suspends new tasks
420 * scheduled after taskq_suspend() was called.
421 *
422 * taskq_member() function works by comparing a thread t_taskq pointer with
423 * the passed thread pointer.
424 *
425 * LOCKS and LOCK Hierarchy ----------------------------------------------------
426 *
427 * There are three locks used in task queues:
428 *
429 * 1) The taskq_t's tq_lock, protecting global task queue state.
430 *
431 * 2) Each per-CPU bucket has a lock for bucket management.
432 *
433 * 3) The global taskq_cpupct_lock, which protects the list of
434 * TASKQ_THREADS_CPU_PCT taskqs.
435 *
436 * If both (1) and (2) are needed, tq_lock should be taken *after* the bucket
437 * lock.
438 *
439 * If both (1) and (3) are needed, tq_lock should be taken *after*
440 * taskq_cpupct_lock.
441 *
442 * DEBUG FACILITIES ------------------------------------------------------------
443 *
444 * For DEBUG kernels it is possible to induce random failures to
445 * taskq_dispatch() function when it is given TQ_NOSLEEP argument. The value of
446 * taskq_dmtbf and taskq_smtbf tunables control the mean time between induced
447 * failures for dynamic and static task queues respectively.
448 *
449 * Setting TASKQ_STATISTIC to 0 will disable per-bucket statistics.
450 *
451 * TUNABLES --------------------------------------------------------------------
452 *
453 * system_taskq_size - Size of the global system_taskq.
454 * This value is multiplied by nCPUs to determine
455 * actual size.
456 * Default value: 64
457 *
458 * taskq_minimum_nthreads_max
459 * - Minimum size of the thread list for a taskq.
460 * Useful for testing different thread pool
461 * sizes by overwriting tq_nthreads_target.
462 *
463 * taskq_thread_timeout - Maximum idle time for taskq_d_thread()
464 * Default value: 5 minutes
465 *
466 * taskq_maxbuckets - Maximum number of buckets in any task queue
467 * Default value: 128
468 *
469 * taskq_search_depth - Maximum # of buckets searched for a free entry
470 * Default value: 4
471 *
472 * taskq_dmtbf - Mean time between induced dispatch failures
473 * for dynamic task queues.
474 * Default value: UINT_MAX (no induced failures)
475 *
476 * taskq_smtbf - Mean time between induced dispatch failures
477 * for static task queues.
478 * Default value: UINT_MAX (no induced failures)
479 *
480 * CONDITIONAL compilation -----------------------------------------------------
481 *
482 * TASKQ_STATISTIC - If set will enable bucket statistic (default).
483 *
484 */
485
486 #include <sys/taskq_impl.h>
487 #include <sys/thread.h>
488 #include <sys/proc.h>
489 #include <sys/kmem.h>
490 #include <sys/vmem.h>
491 #include <sys/callb.h>
492 #include <sys/class.h>
493 #include <sys/systm.h>
494 #include <sys/cmn_err.h>
495 #include <sys/debug.h>
496 #include <sys/vmsystm.h> /* For throttlefree */
497 #include <sys/sysmacros.h>
498 #include <sys/cpuvar.h>
499 #include <sys/cpupart.h>
500 #include <sys/sdt.h>
501 #include <sys/sysdc.h>
502 #include <sys/note.h>
503
504 static kmem_cache_t *taskq_ent_cache, *taskq_cache;
505
506 /*
507 * Pseudo instance numbers for taskqs without explicitly provided instance.
508 */
509 static vmem_t *taskq_id_arena;
510
511 /* Global system task queue for common use */
512 taskq_t *system_taskq;
513
514 /*
515 * Maximum number of entries in global system taskq is
516 * system_taskq_size * max_ncpus
517 */
518 #define SYSTEM_TASKQ_SIZE 64
519 int system_taskq_size = SYSTEM_TASKQ_SIZE;
520
521 /*
522 * Minimum size for tq_nthreads_max; useful for those who want to play around
523 * with increasing a taskq's tq_nthreads_target.
524 */
525 int taskq_minimum_nthreads_max = 1;
526
527 /*
528 * We want to ensure that when taskq_create() returns, there is at least
529 * one thread ready to handle requests. To guarantee this, we have to wait
530 * for the second thread, since the first one cannot process requests until
531 * the second thread has been created.
532 */
533 #define TASKQ_CREATE_ACTIVE_THREADS 2
534
535 /* Maximum percentage allowed for TASKQ_THREADS_CPU_PCT */
536 #define TASKQ_CPUPCT_MAX_PERCENT 1000
537 int taskq_cpupct_max_percent = TASKQ_CPUPCT_MAX_PERCENT;
538
539 /*
540 * Dynamic task queue threads that don't get any work within
541 * taskq_thread_timeout destroy themselves
542 */
543 #define TASKQ_THREAD_TIMEOUT (60 * 5)
544 int taskq_thread_timeout = TASKQ_THREAD_TIMEOUT;
545
546 #define TASKQ_MAXBUCKETS 128
547 int taskq_maxbuckets = TASKQ_MAXBUCKETS;
548
549 /*
550 * When a bucket has no available entries another buckets are tried.
551 * taskq_search_depth parameter limits the amount of buckets that we search
552 * before failing. This is mostly useful in systems with many CPUs where we may
553 * spend too much time scanning busy buckets.
554 */
555 #define TASKQ_SEARCH_DEPTH 4
556 int taskq_search_depth = TASKQ_SEARCH_DEPTH;
557
558 /*
559 * Hashing function: mix various bits of x. May be pretty much anything.
560 */
561 #define TQ_HASH(x) ((x) ^ ((x) >> 11) ^ ((x) >> 17) ^ ((x) ^ 27))
562
563 /*
564 * We do not create any new threads when the system is low on memory and start
565 * throttling memory allocations. The following macro tries to estimate such
566 * condition.
567 */
568 #define ENOUGH_MEMORY() (freemem > throttlefree)
569
570 /*
571 * Static functions.
572 */
573 static taskq_t *taskq_create_common(const char *, int, int, pri_t, int,
574 int, proc_t *, uint_t, uint_t);
575 static void taskq_thread(void *);
576 static void taskq_d_thread(taskq_ent_t *);
577 static void taskq_bucket_extend(void *);
578 static int taskq_constructor(void *, void *, int);
579 static void taskq_destructor(void *, void *);
580 static int taskq_ent_constructor(void *, void *, int);
581 static void taskq_ent_destructor(void *, void *);
582 static taskq_ent_t *taskq_ent_alloc(taskq_t *, int);
583 static void taskq_ent_free(taskq_t *, taskq_ent_t *);
584 static int taskq_ent_exists(taskq_t *, task_func_t, void *);
585 static taskq_ent_t *taskq_bucket_dispatch(taskq_bucket_t *, task_func_t,
586 void *);
587
588 /*
589 * Task queues kstats.
590 */
591 struct taskq_kstat {
592 kstat_named_t tq_pid;
593 kstat_named_t tq_tasks;
594 kstat_named_t tq_executed;
595 kstat_named_t tq_maxtasks;
596 kstat_named_t tq_totaltime;
597 kstat_named_t tq_nalloc;
598 kstat_named_t tq_nactive;
599 kstat_named_t tq_pri;
600 kstat_named_t tq_nthreads;
601 kstat_named_t tq_nomem;
602 } taskq_kstat = {
603 { "pid", KSTAT_DATA_UINT64 },
604 { "tasks", KSTAT_DATA_UINT64 },
605 { "executed", KSTAT_DATA_UINT64 },
606 { "maxtasks", KSTAT_DATA_UINT64 },
607 { "totaltime", KSTAT_DATA_UINT64 },
608 { "nalloc", KSTAT_DATA_UINT64 },
609 { "nactive", KSTAT_DATA_UINT64 },
610 { "priority", KSTAT_DATA_UINT64 },
611 { "threads", KSTAT_DATA_UINT64 },
612 { "nomem", KSTAT_DATA_UINT64 },
613 };
614
615 struct taskq_d_kstat {
616 kstat_named_t tqd_pri;
617 kstat_named_t tqd_btasks;
618 kstat_named_t tqd_bexecuted;
619 kstat_named_t tqd_bmaxtasks;
620 kstat_named_t tqd_bnalloc;
621 kstat_named_t tqd_bnactive;
622 kstat_named_t tqd_btotaltime;
623 kstat_named_t tqd_hits;
624 kstat_named_t tqd_misses;
625 kstat_named_t tqd_overflows;
626 kstat_named_t tqd_tcreates;
627 kstat_named_t tqd_tdeaths;
628 kstat_named_t tqd_maxthreads;
629 kstat_named_t tqd_nomem;
630 kstat_named_t tqd_disptcreates;
631 kstat_named_t tqd_totaltime;
632 kstat_named_t tqd_nalloc;
633 kstat_named_t tqd_nfree;
634 } taskq_d_kstat = {
635 { "priority", KSTAT_DATA_UINT64 },
636 { "btasks", KSTAT_DATA_UINT64 },
637 { "bexecuted", KSTAT_DATA_UINT64 },
638 { "bmaxtasks", KSTAT_DATA_UINT64 },
639 { "bnalloc", KSTAT_DATA_UINT64 },
640 { "bnactive", KSTAT_DATA_UINT64 },
641 { "btotaltime", KSTAT_DATA_UINT64 },
642 { "hits", KSTAT_DATA_UINT64 },
643 { "misses", KSTAT_DATA_UINT64 },
644 { "overflows", KSTAT_DATA_UINT64 },
645 { "tcreates", KSTAT_DATA_UINT64 },
646 { "tdeaths", KSTAT_DATA_UINT64 },
647 { "maxthreads", KSTAT_DATA_UINT64 },
648 { "nomem", KSTAT_DATA_UINT64 },
649 { "disptcreates", KSTAT_DATA_UINT64 },
650 { "totaltime", KSTAT_DATA_UINT64 },
651 { "nalloc", KSTAT_DATA_UINT64 },
652 { "nfree", KSTAT_DATA_UINT64 },
653 };
654
655 static kmutex_t taskq_kstat_lock;
656 static kmutex_t taskq_d_kstat_lock;
657 static int taskq_kstat_update(kstat_t *, int);
658 static int taskq_d_kstat_update(kstat_t *, int);
659
660 /*
661 * List of all TASKQ_THREADS_CPU_PCT taskqs.
662 */
663 static list_t taskq_cpupct_list; /* protected by cpu_lock */
664
665 /*
666 * Collect per-bucket statistic when TASKQ_STATISTIC is defined.
667 */
668 #define TASKQ_STATISTIC 1
669
670 #if TASKQ_STATISTIC
671 #define TQ_STAT(b, x) b->tqbucket_stat.x++
672 #else
673 #define TQ_STAT(b, x)
674 #endif
675
676 /*
677 * Random fault injection.
678 */
679 uint_t taskq_random;
680 uint_t taskq_dmtbf = UINT_MAX; /* mean time between injected failures */
681 uint_t taskq_smtbf = UINT_MAX; /* mean time between injected failures */
682
683 /*
684 * TQ_NOSLEEP dispatches on dynamic task queues are always allowed to fail.
685 *
686 * TQ_NOSLEEP dispatches on static task queues can't arbitrarily fail because
687 * they could prepopulate the cache and make sure that they do not use more
688 * then minalloc entries. So, fault injection in this case insures that
689 * either TASKQ_PREPOPULATE is not set or there are more entries allocated
690 * than is specified by minalloc. TQ_NOALLOC dispatches are always allowed
691 * to fail, but for simplicity we treat them identically to TQ_NOSLEEP
692 * dispatches.
693 */
694 #ifdef DEBUG
695 #define TASKQ_D_RANDOM_DISPATCH_FAILURE(tq, flag) \
696 taskq_random = (taskq_random * 2416 + 374441) % 1771875;\
697 if ((flag & TQ_NOSLEEP) && \
698 taskq_random < 1771875 / taskq_dmtbf) { \
699 return (NULL); \
700 }
701
702 #define TASKQ_S_RANDOM_DISPATCH_FAILURE(tq, flag) \
703 taskq_random = (taskq_random * 2416 + 374441) % 1771875;\
704 if ((flag & (TQ_NOSLEEP | TQ_NOALLOC)) && \
705 (!(tq->tq_flags & TASKQ_PREPOPULATE) || \
706 (tq->tq_nalloc > tq->tq_minalloc)) && \
707 (taskq_random < (1771875 / taskq_smtbf))) { \
708 mutex_exit(&tq->tq_lock); \
709 return (NULL); \
710 }
711 #else
712 #define TASKQ_S_RANDOM_DISPATCH_FAILURE(tq, flag)
713 #define TASKQ_D_RANDOM_DISPATCH_FAILURE(tq, flag)
714 #endif
715
716 #define IS_EMPTY(l) (((l).tqent_prev == (l).tqent_next) && \
717 ((l).tqent_prev == &(l)))
718
719 /*
720 * Append `tqe' in the end of the doubly-linked list denoted by l.
721 */
722 #define TQ_APPEND(l, tqe) { \
723 tqe->tqent_next = &l; \
724 tqe->tqent_prev = l.tqent_prev; \
725 tqe->tqent_next->tqent_prev = tqe; \
726 tqe->tqent_prev->tqent_next = tqe; \
727 }
728 /*
729 * Prepend 'tqe' to the beginning of l
730 */
731 #define TQ_PREPEND(l, tqe) { \
732 tqe->tqent_next = l.tqent_next; \
733 tqe->tqent_prev = &l; \
734 tqe->tqent_next->tqent_prev = tqe; \
735 tqe->tqent_prev->tqent_next = tqe; \
736 }
737
738 /*
739 * Schedule a task specified by func and arg into the task queue entry tqe.
740 */
741 #define TQ_DO_ENQUEUE(tq, tqe, func, arg, front) { \
742 ASSERT(MUTEX_HELD(&tq->tq_lock)); \
743 _NOTE(CONSTCOND) \
744 if (front) { \
745 TQ_PREPEND(tq->tq_task, tqe); \
746 } else { \
747 TQ_APPEND(tq->tq_task, tqe); \
748 } \
749 tqe->tqent_func = (func); \
750 tqe->tqent_arg = (arg); \
751 tq->tq_tasks++; \
752 if (tq->tq_tasks - tq->tq_executed > tq->tq_maxtasks) \
753 tq->tq_maxtasks = tq->tq_tasks - tq->tq_executed; \
754 cv_signal(&tq->tq_dispatch_cv); \
755 DTRACE_PROBE2(taskq__enqueue, taskq_t *, tq, taskq_ent_t *, tqe); \
756 }
757
758 #define TQ_ENQUEUE(tq, tqe, func, arg) \
759 TQ_DO_ENQUEUE(tq, tqe, func, arg, 0)
760
761 #define TQ_ENQUEUE_FRONT(tq, tqe, func, arg) \
762 TQ_DO_ENQUEUE(tq, tqe, func, arg, 1)
763
764 /*
765 * Do-nothing task which may be used to prepopulate thread caches.
766 */
767 /*ARGSUSED*/
768 void
769 nulltask(void *unused)
770 {
771 }
772
773 /*ARGSUSED*/
774 static int
775 taskq_constructor(void *buf, void *cdrarg, int kmflags)
776 {
777 taskq_t *tq = buf;
778
779 bzero(tq, sizeof (taskq_t));
780
781 mutex_init(&tq->tq_lock, NULL, MUTEX_DEFAULT, NULL);
782 rw_init(&tq->tq_threadlock, NULL, RW_DEFAULT, NULL);
783 cv_init(&tq->tq_dispatch_cv, NULL, CV_DEFAULT, NULL);
784 cv_init(&tq->tq_exit_cv, NULL, CV_DEFAULT, NULL);
785 cv_init(&tq->tq_wait_cv, NULL, CV_DEFAULT, NULL);
786 cv_init(&tq->tq_maxalloc_cv, NULL, CV_DEFAULT, NULL);
787
788 tq->tq_task.tqent_next = &tq->tq_task;
789 tq->tq_task.tqent_prev = &tq->tq_task;
790
791 return (0);
792 }
793
794 /*ARGSUSED*/
795 static void
796 taskq_destructor(void *buf, void *cdrarg)
797 {
798 taskq_t *tq = buf;
799
800 ASSERT(tq->tq_nthreads == 0);
801 ASSERT(tq->tq_buckets == NULL);
802 ASSERT(tq->tq_tcreates == 0);
803 ASSERT(tq->tq_tdeaths == 0);
804
805 mutex_destroy(&tq->tq_lock);
806 rw_destroy(&tq->tq_threadlock);
807 cv_destroy(&tq->tq_dispatch_cv);
808 cv_destroy(&tq->tq_exit_cv);
809 cv_destroy(&tq->tq_wait_cv);
810 cv_destroy(&tq->tq_maxalloc_cv);
811 }
812
813 /*ARGSUSED*/
814 static int
815 taskq_ent_constructor(void *buf, void *cdrarg, int kmflags)
816 {
817 taskq_ent_t *tqe = buf;
818
819 tqe->tqent_thread = NULL;
820 cv_init(&tqe->tqent_cv, NULL, CV_DEFAULT, NULL);
821
822 return (0);
823 }
824
825 /*ARGSUSED*/
826 static void
827 taskq_ent_destructor(void *buf, void *cdrarg)
828 {
829 taskq_ent_t *tqe = buf;
830
831 ASSERT(tqe->tqent_thread == NULL);
832 cv_destroy(&tqe->tqent_cv);
833 }
834
835 void
836 taskq_init(void)
837 {
838 taskq_ent_cache = kmem_cache_create("taskq_ent_cache",
839 sizeof (taskq_ent_t), 0, taskq_ent_constructor,
840 taskq_ent_destructor, NULL, NULL, NULL, 0);
841 taskq_cache = kmem_cache_create("taskq_cache", sizeof (taskq_t),
842 0, taskq_constructor, taskq_destructor, NULL, NULL, NULL, 0);
843 taskq_id_arena = vmem_create("taskq_id_arena",
844 (void *)1, INT32_MAX, 1, NULL, NULL, NULL, 0,
845 VM_SLEEP | VMC_IDENTIFIER);
846
847 list_create(&taskq_cpupct_list, sizeof (taskq_t),
848 offsetof(taskq_t, tq_cpupct_link));
849 }
850
851 static void
852 taskq_update_nthreads(taskq_t *tq, uint_t ncpus)
853 {
854 uint_t newtarget = TASKQ_THREADS_PCT(ncpus, tq->tq_threads_ncpus_pct);
855
856 ASSERT(MUTEX_HELD(&cpu_lock));
857 ASSERT(MUTEX_HELD(&tq->tq_lock));
858
859 /* We must be going from non-zero to non-zero; no exiting. */
860 ASSERT3U(tq->tq_nthreads_target, !=, 0);
861 ASSERT3U(newtarget, !=, 0);
862
863 ASSERT3U(newtarget, <=, tq->tq_nthreads_max);
864 if (newtarget != tq->tq_nthreads_target) {
865 tq->tq_flags |= TASKQ_CHANGING;
866 tq->tq_nthreads_target = newtarget;
867 cv_broadcast(&tq->tq_dispatch_cv);
868 cv_broadcast(&tq->tq_exit_cv);
869 }
870 }
871
872 /* called during task queue creation */
873 static void
874 taskq_cpupct_install(taskq_t *tq, cpupart_t *cpup)
875 {
876 ASSERT(tq->tq_flags & TASKQ_THREADS_CPU_PCT);
877
878 mutex_enter(&cpu_lock);
879 mutex_enter(&tq->tq_lock);
880 tq->tq_cpupart = cpup->cp_id;
881 taskq_update_nthreads(tq, cpup->cp_ncpus);
882 mutex_exit(&tq->tq_lock);
883
884 list_insert_tail(&taskq_cpupct_list, tq);
885 mutex_exit(&cpu_lock);
886 }
887
888 static void
889 taskq_cpupct_remove(taskq_t *tq)
890 {
891 ASSERT(tq->tq_flags & TASKQ_THREADS_CPU_PCT);
892
893 mutex_enter(&cpu_lock);
894 list_remove(&taskq_cpupct_list, tq);
895 mutex_exit(&cpu_lock);
896 }
897
898 /*ARGSUSED*/
899 static int
900 taskq_cpu_setup(cpu_setup_t what, int id, void *arg)
901 {
902 taskq_t *tq;
903 cpupart_t *cp = cpu[id]->cpu_part;
904 uint_t ncpus = cp->cp_ncpus;
905
906 ASSERT(MUTEX_HELD(&cpu_lock));
907 ASSERT(ncpus > 0);
908
909 switch (what) {
910 case CPU_OFF:
911 case CPU_CPUPART_OUT:
912 /* offlines are called *before* the cpu is offlined. */
913 if (ncpus > 1)
914 ncpus--;
915 break;
916
917 case CPU_ON:
918 case CPU_CPUPART_IN:
919 break;
920
921 default:
922 return (0); /* doesn't affect cpu count */
923 }
924
925 for (tq = list_head(&taskq_cpupct_list); tq != NULL;
926 tq = list_next(&taskq_cpupct_list, tq)) {
927
928 mutex_enter(&tq->tq_lock);
929 /*
930 * If the taskq is part of the cpuset which is changing,
931 * update its nthreads_target.
932 */
933 if (tq->tq_cpupart == cp->cp_id) {
934 taskq_update_nthreads(tq, ncpus);
935 }
936 mutex_exit(&tq->tq_lock);
937 }
938 return (0);
939 }
940
941 void
942 taskq_mp_init(void)
943 {
944 mutex_enter(&cpu_lock);
945 register_cpu_setup_func(taskq_cpu_setup, NULL);
946 /*
947 * Make sure we're up to date. At this point in boot, there is only
948 * one processor set, so we only have to update the current CPU.
949 */
950 (void) taskq_cpu_setup(CPU_ON, CPU->cpu_id, NULL);
951 mutex_exit(&cpu_lock);
952 }
953
954 /*
955 * Create global system dynamic task queue.
956 */
957 void
958 system_taskq_init(void)
959 {
960 system_taskq = taskq_create_common("system_taskq", 0,
961 system_taskq_size * max_ncpus, minclsyspri, 4, 512, &p0, 0,
962 TASKQ_DYNAMIC | TASKQ_PREPOPULATE);
963 }
964
965 /*
966 * taskq_ent_alloc()
967 *
968 * Allocates a new taskq_ent_t structure either from the free list or from the
969 * cache. Returns NULL if it can't be allocated.
970 *
971 * Assumes: tq->tq_lock is held.
972 */
973 static taskq_ent_t *
974 taskq_ent_alloc(taskq_t *tq, int flags)
975 {
976 int kmflags = (flags & TQ_NOSLEEP) ? KM_NOSLEEP : KM_SLEEP;
977 taskq_ent_t *tqe;
978 clock_t wait_time;
979 clock_t wait_rv;
980
981 ASSERT(MUTEX_HELD(&tq->tq_lock));
982
983 /*
984 * TQ_NOALLOC allocations are allowed to use the freelist, even if
985 * we are below tq_minalloc.
986 */
987 again: if ((tqe = tq->tq_freelist) != NULL &&
988 ((flags & TQ_NOALLOC) || tq->tq_nalloc >= tq->tq_minalloc)) {
989 tq->tq_freelist = tqe->tqent_next;
990 } else {
991 if (flags & TQ_NOALLOC)
992 return (NULL);
993
994 if (tq->tq_nalloc >= tq->tq_maxalloc) {
995 if (kmflags & KM_NOSLEEP)
996 return (NULL);
997
998 /*
999 * We don't want to exceed tq_maxalloc, but we can't
1000 * wait for other tasks to complete (and thus free up
1001 * task structures) without risking deadlock with
1002 * the caller. So, we just delay for one second
1003 * to throttle the allocation rate. If we have tasks
1004 * complete before one second timeout expires then
1005 * taskq_ent_free will signal us and we will
1006 * immediately retry the allocation (reap free).
1007 */
1008 wait_time = ddi_get_lbolt() + hz;
1009 while (tq->tq_freelist == NULL) {
1010 tq->tq_maxalloc_wait++;
1011 wait_rv = cv_timedwait(&tq->tq_maxalloc_cv,
1012 &tq->tq_lock, wait_time);
1013 tq->tq_maxalloc_wait--;
1014 if (wait_rv == -1)
1015 break;
1016 }
1017 if (tq->tq_freelist)
1018 goto again; /* reap freelist */
1019
1020 }
1021 mutex_exit(&tq->tq_lock);
1022
1023 tqe = kmem_cache_alloc(taskq_ent_cache, kmflags);
1024
1025 mutex_enter(&tq->tq_lock);
1026 if (tqe != NULL)
1027 tq->tq_nalloc++;
1028 }
1029 return (tqe);
1030 }
1031
1032 /*
1033 * taskq_ent_free()
1034 *
1035 * Free taskq_ent_t structure by either putting it on the free list or freeing
1036 * it to the cache.
1037 *
1038 * Assumes: tq->tq_lock is held.
1039 */
1040 static void
1041 taskq_ent_free(taskq_t *tq, taskq_ent_t *tqe)
1042 {
1043 ASSERT(MUTEX_HELD(&tq->tq_lock));
1044
1045 if (tq->tq_nalloc <= tq->tq_minalloc) {
1046 tqe->tqent_next = tq->tq_freelist;
1047 tq->tq_freelist = tqe;
1048 } else {
1049 tq->tq_nalloc--;
1050 mutex_exit(&tq->tq_lock);
1051 kmem_cache_free(taskq_ent_cache, tqe);
1052 mutex_enter(&tq->tq_lock);
1053 }
1054
1055 if (tq->tq_maxalloc_wait)
1056 cv_signal(&tq->tq_maxalloc_cv);
1057 }
1058
1059 /*
1060 * taskq_ent_exists()
1061 *
1062 * Return 1 if taskq already has entry for calling 'func(arg)'.
1063 *
1064 * Assumes: tq->tq_lock is held.
1065 */
1066 static int
1067 taskq_ent_exists(taskq_t *tq, task_func_t func, void *arg)
1068 {
1069 taskq_ent_t *tqe;
1070
1071 ASSERT(MUTEX_HELD(&tq->tq_lock));
1072
1073 for (tqe = tq->tq_task.tqent_next; tqe != &tq->tq_task;
1074 tqe = tqe->tqent_next)
1075 if ((tqe->tqent_func == func) && (tqe->tqent_arg == arg))
1076 return (1);
1077 return (0);
1078 }
1079
1080 /*
1081 * Dispatch a task "func(arg)" to a free entry of bucket b.
1082 *
1083 * Assumes: no bucket locks is held.
1084 *
1085 * Returns: a pointer to an entry if dispatch was successful.
1086 * NULL if there are no free entries or if the bucket is suspended.
1087 */
1088 static taskq_ent_t *
1089 taskq_bucket_dispatch(taskq_bucket_t *b, task_func_t func, void *arg)
1090 {
1091 taskq_ent_t *tqe;
1092
1093 ASSERT(MUTEX_NOT_HELD(&b->tqbucket_lock));
1094 ASSERT(func != NULL);
1095
1096 mutex_enter(&b->tqbucket_lock);
1097
1098 ASSERT(b->tqbucket_nfree != 0 || IS_EMPTY(b->tqbucket_freelist));
1099 ASSERT(b->tqbucket_nfree == 0 || !IS_EMPTY(b->tqbucket_freelist));
1100
1101 /*
1102 * Get en entry from the freelist if there is one.
1103 * Schedule task into the entry.
1104 */
1105 if ((b->tqbucket_nfree != 0) &&
1106 !(b->tqbucket_flags & TQBUCKET_SUSPEND)) {
1107 tqe = b->tqbucket_freelist.tqent_prev;
1108
1109 ASSERT(tqe != &b->tqbucket_freelist);
1110 ASSERT(tqe->tqent_thread != NULL);
1111
1112 tqe->tqent_prev->tqent_next = tqe->tqent_next;
1113 tqe->tqent_next->tqent_prev = tqe->tqent_prev;
1114 b->tqbucket_nalloc++;
1115 b->tqbucket_nfree--;
1116 tqe->tqent_func = func;
1117 tqe->tqent_arg = arg;
1118 TQ_STAT(b, tqs_hits);
1119 cv_signal(&tqe->tqent_cv);
1120 DTRACE_PROBE2(taskq__d__enqueue, taskq_bucket_t *, b,
1121 taskq_ent_t *, tqe);
1122 } else {
1123 tqe = NULL;
1124 TQ_STAT(b, tqs_misses);
1125 }
1126 mutex_exit(&b->tqbucket_lock);
1127 return (tqe);
1128 }
1129
1130 /*
1131 * Dispatch a task.
1132 *
1133 * Assumes: func != NULL
1134 *
1135 * Returns: NULL if dispatch failed.
1136 * non-NULL if task dispatched successfully.
1137 * Actual return value is the pointer to taskq entry that was used to
1138 * dispatch a task. This is useful for debugging.
1139 */
1140 taskqid_t
1141 taskq_dispatch(taskq_t *tq, task_func_t func, void *arg, uint_t flags)
1142 {
1143 taskq_bucket_t *bucket = NULL; /* Which bucket needs extension */
1144 taskq_ent_t *tqe = NULL;
1145 taskq_ent_t *tqe1;
1146 uint_t bsize;
1147
1148 ASSERT(tq != NULL);
1149 ASSERT(func != NULL);
1150
1151 if (!(tq->tq_flags & TASKQ_DYNAMIC)) {
1152 /*
1153 * TQ_NOQUEUE flag can't be used with non-dynamic task queues.
1154 */
1155 ASSERT(!(flags & TQ_NOQUEUE));
1156 /*
1157 * Enqueue the task to the underlying queue.
1158 */
1159 mutex_enter(&tq->tq_lock);
1160
1161 TASKQ_S_RANDOM_DISPATCH_FAILURE(tq, flags);
1162
1163 if ((tqe = taskq_ent_alloc(tq, flags)) == NULL) {
1164 tq->tq_nomem++;
1165 mutex_exit(&tq->tq_lock);
1166 return (NULL);
1167 }
1168 /* Make sure we start without any flags */
1169 tqe->tqent_un.tqent_flags = 0;
1170
1171 if (flags & TQ_FRONT) {
1172 TQ_ENQUEUE_FRONT(tq, tqe, func, arg);
1173 } else {
1174 TQ_ENQUEUE(tq, tqe, func, arg);
1175 }
1176 mutex_exit(&tq->tq_lock);
1177 return ((taskqid_t)tqe);
1178 }
1179
1180 /*
1181 * Dynamic taskq dispatching.
1182 */
1183 ASSERT(!(flags & (TQ_NOALLOC | TQ_FRONT)));
1184 TASKQ_D_RANDOM_DISPATCH_FAILURE(tq, flags);
1185
1186 bsize = tq->tq_nbuckets;
1187
1188 if (bsize == 1) {
1189 /*
1190 * In a single-CPU case there is only one bucket, so get
1191 * entry directly from there.
1192 */
1193 if ((tqe = taskq_bucket_dispatch(tq->tq_buckets, func, arg))
1194 != NULL)
1195 return ((taskqid_t)tqe); /* Fastpath */
1196 bucket = tq->tq_buckets;
1197 } else {
1198 int loopcount;
1199 taskq_bucket_t *b;
1200 uintptr_t h = ((uintptr_t)CPU + (uintptr_t)arg) >> 3;
1201
1202 h = TQ_HASH(h);
1203
1204 /*
1205 * The 'bucket' points to the original bucket that we hit. If we
1206 * can't allocate from it, we search other buckets, but only
1207 * extend this one.
1208 */
1209 b = &tq->tq_buckets[h & (bsize - 1)];
1210 ASSERT(b->tqbucket_taskq == tq); /* Sanity check */
1211
1212 /*
1213 * Do a quick check before grabbing the lock. If the bucket does
1214 * not have free entries now, chances are very small that it
1215 * will after we take the lock, so we just skip it.
1216 */
1217 if (b->tqbucket_nfree != 0) {
1218 if ((tqe = taskq_bucket_dispatch(b, func, arg)) != NULL)
1219 return ((taskqid_t)tqe); /* Fastpath */
1220 } else {
1221 TQ_STAT(b, tqs_misses);
1222 }
1223
1224 bucket = b;
1225 loopcount = MIN(taskq_search_depth, bsize);
1226 /*
1227 * If bucket dispatch failed, search loopcount number of buckets
1228 * before we give up and fail.
1229 */
1230 do {
1231 b = &tq->tq_buckets[++h & (bsize - 1)];
1232 ASSERT(b->tqbucket_taskq == tq); /* Sanity check */
1233 loopcount--;
1234
1235 if (b->tqbucket_nfree != 0) {
1236 tqe = taskq_bucket_dispatch(b, func, arg);
1237 } else {
1238 TQ_STAT(b, tqs_misses);
1239 }
1240 } while ((tqe == NULL) && (loopcount > 0));
1241 }
1242
1243 /*
1244 * At this point we either scheduled a task and (tqe != NULL) or failed
1245 * (tqe == NULL). Try to recover from fails.
1246 */
1247
1248 /*
1249 * For KM_SLEEP dispatches, try to extend the bucket and retry dispatch.
1250 */
1251 if ((tqe == NULL) && !(flags & TQ_NOSLEEP)) {
1252 /*
1253 * taskq_bucket_extend() may fail to do anything, but this is
1254 * fine - we deal with it later. If the bucket was successfully
1255 * extended, there is a good chance that taskq_bucket_dispatch()
1256 * will get this new entry, unless someone is racing with us and
1257 * stealing the new entry from under our nose.
1258 * taskq_bucket_extend() may sleep.
1259 */
1260 taskq_bucket_extend(bucket);
1261 TQ_STAT(bucket, tqs_disptcreates);
1262 if ((tqe = taskq_bucket_dispatch(bucket, func, arg)) != NULL)
1263 return ((taskqid_t)tqe);
1264 }
1265
1266 ASSERT(bucket != NULL);
1267
1268 /*
1269 * Since there are not enough free entries in the bucket, add a
1270 * taskq entry to extend it in the background using backing queue
1271 * (unless we already have a taskq entry to perform that extension).
1272 */
1273 mutex_enter(&tq->tq_lock);
1274 if (!taskq_ent_exists(tq, taskq_bucket_extend, bucket)) {
1275 if ((tqe1 = taskq_ent_alloc(tq, TQ_NOSLEEP)) != NULL) {
1276 TQ_ENQUEUE_FRONT(tq, tqe1, taskq_bucket_extend, bucket);
1277 } else {
1278 tq->tq_nomem++;
1279 }
1280 }
1281
1282 /*
1283 * Dispatch failed and we can't find an entry to schedule a task.
1284 * Revert to the backing queue unless TQ_NOQUEUE was asked.
1285 */
1286 if ((tqe == NULL) && !(flags & TQ_NOQUEUE)) {
1287 if ((tqe = taskq_ent_alloc(tq, flags)) != NULL) {
1288 TQ_ENQUEUE(tq, tqe, func, arg);
1289 } else {
1290 tq->tq_nomem++;
1291 }
1292 }
1293 mutex_exit(&tq->tq_lock);
1294
1295 return ((taskqid_t)tqe);
1296 }
1297
1298 void
1299 taskq_dispatch_ent(taskq_t *tq, task_func_t func, void *arg, uint_t flags,
1300 taskq_ent_t *tqe)
1301 {
1302 ASSERT(func != NULL);
1303 ASSERT(!(tq->tq_flags & TASKQ_DYNAMIC));
1304
1305 /*
1306 * Mark it as a prealloc'd task. This is important
1307 * to ensure that we don't free it later.
1308 */
1309 tqe->tqent_un.tqent_flags |= TQENT_FLAG_PREALLOC;
1310 /*
1311 * Enqueue the task to the underlying queue.
1312 */
1313 mutex_enter(&tq->tq_lock);
1314
1315 if (flags & TQ_FRONT) {
1316 TQ_ENQUEUE_FRONT(tq, tqe, func, arg);
1317 } else {
1318 TQ_ENQUEUE(tq, tqe, func, arg);
1319 }
1320 mutex_exit(&tq->tq_lock);
1321 }
1322
1323 /*
1324 * Allow our caller to ask if there are tasks pending on the queue.
1325 */
1326 boolean_t
1327 taskq_empty(taskq_t *tq)
1328 {
1329 boolean_t rv;
1330
1331 ASSERT3P(tq, !=, curthread->t_taskq);
1332 mutex_enter(&tq->tq_lock);
1333 rv = (tq->tq_task.tqent_next == &tq->tq_task) && (tq->tq_active == 0);
1334 mutex_exit(&tq->tq_lock);
1335
1336 return (rv);
1337 }
1338
1339 /*
1340 * Wait for all pending tasks to complete.
1341 * Calling taskq_wait from a task will cause deadlock.
1342 */
1343 void
1344 taskq_wait(taskq_t *tq)
1345 {
1346 ASSERT(tq != curthread->t_taskq);
1347
1348 mutex_enter(&tq->tq_lock);
1349 while (tq->tq_task.tqent_next != &tq->tq_task || tq->tq_active != 0)
1350 cv_wait(&tq->tq_wait_cv, &tq->tq_lock);
1351 mutex_exit(&tq->tq_lock);
1352
1353 if (tq->tq_flags & TASKQ_DYNAMIC) {
1354 taskq_bucket_t *b = tq->tq_buckets;
1355 int bid = 0;
1356 for (; (b != NULL) && (bid < tq->tq_nbuckets); b++, bid++) {
1357 mutex_enter(&b->tqbucket_lock);
1358 while (b->tqbucket_nalloc > 0)
1359 cv_wait(&b->tqbucket_cv, &b->tqbucket_lock);
1360 mutex_exit(&b->tqbucket_lock);
1361 }
1362 }
1363 }
1364
1365 /*
1366 * Suspend execution of tasks.
1367 *
1368 * Tasks in the queue part will be suspended immediately upon return from this
1369 * function. Pending tasks in the dynamic part will continue to execute, but all
1370 * new tasks will be suspended.
1371 */
1372 void
1373 taskq_suspend(taskq_t *tq)
1374 {
1375 rw_enter(&tq->tq_threadlock, RW_WRITER);
1376
1377 if (tq->tq_flags & TASKQ_DYNAMIC) {
1378 taskq_bucket_t *b = tq->tq_buckets;
1379 int bid = 0;
1380 for (; (b != NULL) && (bid < tq->tq_nbuckets); b++, bid++) {
1381 mutex_enter(&b->tqbucket_lock);
1382 b->tqbucket_flags |= TQBUCKET_SUSPEND;
1383 mutex_exit(&b->tqbucket_lock);
1384 }
1385 }
1386 /*
1387 * Mark task queue as being suspended. Needed for taskq_suspended().
1388 */
1389 mutex_enter(&tq->tq_lock);
1390 ASSERT(!(tq->tq_flags & TASKQ_SUSPENDED));
1391 tq->tq_flags |= TASKQ_SUSPENDED;
1392 mutex_exit(&tq->tq_lock);
1393 }
1394
1395 /*
1396 * returns: 1 if tq is suspended, 0 otherwise.
1397 */
1398 int
1399 taskq_suspended(taskq_t *tq)
1400 {
1401 return ((tq->tq_flags & TASKQ_SUSPENDED) != 0);
1402 }
1403
1404 /*
1405 * Resume taskq execution.
1406 */
1407 void
1408 taskq_resume(taskq_t *tq)
1409 {
1410 ASSERT(RW_WRITE_HELD(&tq->tq_threadlock));
1411
1412 if (tq->tq_flags & TASKQ_DYNAMIC) {
1413 taskq_bucket_t *b = tq->tq_buckets;
1414 int bid = 0;
1415 for (; (b != NULL) && (bid < tq->tq_nbuckets); b++, bid++) {
1416 mutex_enter(&b->tqbucket_lock);
1417 b->tqbucket_flags &= ~TQBUCKET_SUSPEND;
1418 mutex_exit(&b->tqbucket_lock);
1419 }
1420 }
1421 mutex_enter(&tq->tq_lock);
1422 ASSERT(tq->tq_flags & TASKQ_SUSPENDED);
1423 tq->tq_flags &= ~TASKQ_SUSPENDED;
1424 mutex_exit(&tq->tq_lock);
1425
1426 rw_exit(&tq->tq_threadlock);
1427 }
1428
1429 int
1430 taskq_member(taskq_t *tq, kthread_t *thread)
1431 {
1432 return (thread->t_taskq == tq);
1433 }
1434
1435 /*
1436 * Creates a thread in the taskq. We only allow one outstanding create at
1437 * a time. We drop and reacquire the tq_lock in order to avoid blocking other
1438 * taskq activity while thread_create() or lwp_kernel_create() run.
1439 *
1440 * The first time we're called, we do some additional setup, and do not
1441 * return until there are enough threads to start servicing requests.
1442 */
1443 static void
1444 taskq_thread_create(taskq_t *tq)
1445 {
1446 kthread_t *t;
1447 const boolean_t first = (tq->tq_nthreads == 0);
1448
1449 ASSERT(MUTEX_HELD(&tq->tq_lock));
1450 ASSERT(tq->tq_flags & TASKQ_CHANGING);
1451 ASSERT(tq->tq_nthreads < tq->tq_nthreads_target);
1452 ASSERT(!(tq->tq_flags & TASKQ_THREAD_CREATED));
1453
1454
1455 tq->tq_flags |= TASKQ_THREAD_CREATED;
1456 tq->tq_active++;
1457 mutex_exit(&tq->tq_lock);
1458
1459 /*
1460 * With TASKQ_DUTY_CYCLE the new thread must have an LWP
1461 * as explained in ../disp/sysdc.c (for the msacct data).
1462 * Otherwise simple kthreads are preferred.
1463 */
1464 if ((tq->tq_flags & TASKQ_DUTY_CYCLE) != 0) {
1465 /* Enforced in taskq_create_common */
1466 ASSERT3P(tq->tq_proc, !=, &p0);
1467 t = lwp_kernel_create(tq->tq_proc, taskq_thread, tq, TS_RUN,
1468 tq->tq_pri);
1469 } else {
1470 t = thread_create(NULL, 0, taskq_thread, tq, 0, tq->tq_proc,
1471 TS_RUN, tq->tq_pri);
1472 }
1473
1474 if (!first) {
1475 mutex_enter(&tq->tq_lock);
1476 return;
1477 }
1478
1479 /*
1480 * We know the thread cannot go away, since tq cannot be
1481 * destroyed until creation has completed. We can therefore
1482 * safely dereference t.
1483 */
1484 if (tq->tq_flags & TASKQ_THREADS_CPU_PCT) {
1485 taskq_cpupct_install(tq, t->t_cpupart);
1486 }
1487 mutex_enter(&tq->tq_lock);
1488
1489 /* Wait until we can service requests. */
1490 while (tq->tq_nthreads != tq->tq_nthreads_target &&
1491 tq->tq_nthreads < TASKQ_CREATE_ACTIVE_THREADS) {
1492 cv_wait(&tq->tq_wait_cv, &tq->tq_lock);
1493 }
1494 }
1495
1496 /*
1497 * Common "sleep taskq thread" function, which handles CPR stuff, as well
1498 * as giving a nice common point for debuggers to find inactive threads.
1499 */
1500 static clock_t
1501 taskq_thread_wait(taskq_t *tq, kmutex_t *mx, kcondvar_t *cv,
1502 callb_cpr_t *cprinfo, clock_t timeout)
1503 {
1504 clock_t ret = 0;
1505
1506 if (!(tq->tq_flags & TASKQ_CPR_SAFE)) {
1507 CALLB_CPR_SAFE_BEGIN(cprinfo);
1508 }
1509 if (timeout < 0)
1510 cv_wait(cv, mx);
1511 else
1512 ret = cv_reltimedwait(cv, mx, timeout, TR_CLOCK_TICK);
1513
1514 if (!(tq->tq_flags & TASKQ_CPR_SAFE)) {
1515 CALLB_CPR_SAFE_END(cprinfo, mx);
1516 }
1517
1518 return (ret);
1519 }
1520
1521 /*
1522 * Worker thread for processing task queue.
1523 */
1524 static void
1525 taskq_thread(void *arg)
1526 {
1527 int thread_id;
1528
1529 taskq_t *tq = arg;
1530 taskq_ent_t *tqe;
1531 callb_cpr_t cprinfo;
1532 hrtime_t start, end;
1533 boolean_t freeit;
1534
1535 curthread->t_taskq = tq; /* mark ourselves for taskq_member() */
1536
1537 if (curproc != &p0 && (tq->tq_flags & TASKQ_DUTY_CYCLE)) {
1538 sysdc_thread_enter(curthread, tq->tq_DC,
1539 (tq->tq_flags & TASKQ_DC_BATCH) ? SYSDC_THREAD_BATCH : 0);
1540 }
1541
1542 if (tq->tq_flags & TASKQ_CPR_SAFE) {
1543 CALLB_CPR_INIT_SAFE(curthread, tq->tq_name);
1544 } else {
1545 CALLB_CPR_INIT(&cprinfo, &tq->tq_lock, callb_generic_cpr,
1546 tq->tq_name);
1547 }
1548 mutex_enter(&tq->tq_lock);
1549 thread_id = ++tq->tq_nthreads;
1550 ASSERT(tq->tq_flags & TASKQ_THREAD_CREATED);
1551 ASSERT(tq->tq_flags & TASKQ_CHANGING);
1552 tq->tq_flags &= ~TASKQ_THREAD_CREATED;
1553
1554 VERIFY3S(thread_id, <=, tq->tq_nthreads_max);
1555
1556 if (tq->tq_nthreads_max == 1)
1557 tq->tq_thread = curthread;
1558 else
1559 tq->tq_threadlist[thread_id - 1] = curthread;
1560
1561 /* Allow taskq_create_common()'s taskq_thread_create() to return. */
1562 if (tq->tq_nthreads == TASKQ_CREATE_ACTIVE_THREADS)
1563 cv_broadcast(&tq->tq_wait_cv);
1564
1565 for (;;) {
1566 if (tq->tq_flags & TASKQ_CHANGING) {
1567 /* See if we're no longer needed */
1568 if (thread_id > tq->tq_nthreads_target) {
1569 /*
1570 * To preserve the one-to-one mapping between
1571 * thread_id and thread, we must exit from
1572 * highest thread ID to least.
1573 *
1574 * However, if everyone is exiting, the order
1575 * doesn't matter, so just exit immediately.
1576 * (this is safe, since you must wait for
1577 * nthreads to reach 0 after setting
1578 * tq_nthreads_target to 0)
1579 */
1580 if (thread_id == tq->tq_nthreads ||
1581 tq->tq_nthreads_target == 0)
1582 break;
1583
1584 /* Wait for higher thread_ids to exit */
1585 (void) taskq_thread_wait(tq, &tq->tq_lock,
1586 &tq->tq_exit_cv, &cprinfo, -1);
1587 continue;
1588 }
1589
1590 /*
1591 * If no thread is starting taskq_thread(), we can
1592 * do some bookkeeping.
1593 */
1594 if (!(tq->tq_flags & TASKQ_THREAD_CREATED)) {
1595 /* Check if we've reached our target */
1596 if (tq->tq_nthreads == tq->tq_nthreads_target) {
1597 tq->tq_flags &= ~TASKQ_CHANGING;
1598 cv_broadcast(&tq->tq_wait_cv);
1599 }
1600 /* Check if we need to create a thread */
1601 if (tq->tq_nthreads < tq->tq_nthreads_target) {
1602 taskq_thread_create(tq);
1603 continue; /* tq_lock was dropped */
1604 }
1605 }
1606 }
1607 if ((tqe = tq->tq_task.tqent_next) == &tq->tq_task) {
1608 if (--tq->tq_active == 0)
1609 cv_broadcast(&tq->tq_wait_cv);
1610 (void) taskq_thread_wait(tq, &tq->tq_lock,
1611 &tq->tq_dispatch_cv, &cprinfo, -1);
1612 tq->tq_active++;
1613 continue;
1614 }
1615
1616 tqe->tqent_prev->tqent_next = tqe->tqent_next;
1617 tqe->tqent_next->tqent_prev = tqe->tqent_prev;
1618 mutex_exit(&tq->tq_lock);
1619
1620 /*
1621 * For prealloc'd tasks, we don't free anything. We
1622 * have to check this now, because once we call the
1623 * function for a prealloc'd taskq, we can't touch the
1624 * tqent any longer (calling the function returns the
1625 * ownershp of the tqent back to caller of
1626 * taskq_dispatch.)
1627 */
1628 if ((!(tq->tq_flags & TASKQ_DYNAMIC)) &&
1629 (tqe->tqent_un.tqent_flags & TQENT_FLAG_PREALLOC)) {
1630 /* clear pointers to assist assertion checks */
1631 tqe->tqent_next = tqe->tqent_prev = NULL;
1632 freeit = B_FALSE;
1633 } else {
1634 freeit = B_TRUE;
1635 }
1636
1637 rw_enter(&tq->tq_threadlock, RW_READER);
1638 start = gethrtime();
1639 DTRACE_PROBE2(taskq__exec__start, taskq_t *, tq,
1640 taskq_ent_t *, tqe);
1641 tqe->tqent_func(tqe->tqent_arg);
1642 DTRACE_PROBE2(taskq__exec__end, taskq_t *, tq,
1643 taskq_ent_t *, tqe);
1644 end = gethrtime();
1645 rw_exit(&tq->tq_threadlock);
1646
1647 mutex_enter(&tq->tq_lock);
1648 tq->tq_totaltime += end - start;
1649 tq->tq_executed++;
1650
1651 if (freeit)
1652 taskq_ent_free(tq, tqe);
1653 }
1654
1655 if (tq->tq_nthreads_max == 1)
1656 tq->tq_thread = NULL;
1657 else
1658 tq->tq_threadlist[thread_id - 1] = NULL;
1659
1660 /* We're exiting, and therefore no longer active */
1661 ASSERT(tq->tq_active > 0);
1662 tq->tq_active--;
1663
1664 ASSERT(tq->tq_nthreads > 0);
1665 tq->tq_nthreads--;
1666
1667 /* Wake up anyone waiting for us to exit */
1668 cv_broadcast(&tq->tq_exit_cv);
1669 if (tq->tq_nthreads == tq->tq_nthreads_target) {
1670 if (!(tq->tq_flags & TASKQ_THREAD_CREATED))
1671 tq->tq_flags &= ~TASKQ_CHANGING;
1672
1673 cv_broadcast(&tq->tq_wait_cv);
1674 }
1675
1676 ASSERT(!(tq->tq_flags & TASKQ_CPR_SAFE));
1677 CALLB_CPR_EXIT(&cprinfo); /* drops tq->tq_lock */
1678 if (curthread->t_lwp != NULL) {
1679 mutex_enter(&curproc->p_lock);
1680 lwp_exit();
1681 } else {
1682 thread_exit();
1683 }
1684 }
1685
1686 /*
1687 * Worker per-entry thread for dynamic dispatches.
1688 */
1689 static void
1690 taskq_d_thread(taskq_ent_t *tqe)
1691 {
1692 taskq_bucket_t *bucket = tqe->tqent_un.tqent_bucket;
1693 taskq_t *tq = bucket->tqbucket_taskq;
1694 kmutex_t *lock = &bucket->tqbucket_lock;
1695 kcondvar_t *cv = &tqe->tqent_cv;
1696 callb_cpr_t cprinfo;
1697 clock_t w;
1698
1699 CALLB_CPR_INIT(&cprinfo, lock, callb_generic_cpr, tq->tq_name);
1700
1701 mutex_enter(lock);
1702
1703 for (;;) {
1704 /*
1705 * If a task is scheduled (func != NULL), execute it, otherwise
1706 * sleep, waiting for a job.
1707 */
1708 if (tqe->tqent_func != NULL) {
1709 hrtime_t start;
1710 hrtime_t end;
1711
1712 ASSERT(bucket->tqbucket_nalloc > 0);
1713
1714 /*
1715 * It is possible to free the entry right away before
1716 * actually executing the task so that subsequent
1717 * dispatches may immediately reuse it. But this,
1718 * effectively, creates a two-length queue in the entry
1719 * and may lead to a deadlock if the execution of the
1720 * current task depends on the execution of the next
1721 * scheduled task. So, we keep the entry busy until the
1722 * task is processed.
1723 */
1724
1725 mutex_exit(lock);
1726 start = gethrtime();
1727 DTRACE_PROBE3(taskq__d__exec__start, taskq_t *, tq,
1728 taskq_bucket_t *, bucket, taskq_ent_t *, tqe);
1729 tqe->tqent_func(tqe->tqent_arg);
1730 DTRACE_PROBE3(taskq__d__exec__end, taskq_t *, tq,
1731 taskq_bucket_t *, bucket, taskq_ent_t *, tqe);
1732 end = gethrtime();
1733 mutex_enter(lock);
1734 bucket->tqbucket_totaltime += end - start;
1735
1736 /*
1737 * Return the entry to the bucket free list.
1738 */
1739 tqe->tqent_func = NULL;
1740 TQ_APPEND(bucket->tqbucket_freelist, tqe);
1741 bucket->tqbucket_nalloc--;
1742 bucket->tqbucket_nfree++;
1743 ASSERT(!IS_EMPTY(bucket->tqbucket_freelist));
1744 /*
1745 * taskq_wait() waits for nalloc to drop to zero on
1746 * tqbucket_cv.
1747 */
1748 cv_signal(&bucket->tqbucket_cv);
1749 }
1750
1751 /*
1752 * At this point the entry must be in the bucket free list -
1753 * either because it was there initially or because it just
1754 * finished executing a task and put itself on the free list.
1755 */
1756 ASSERT(bucket->tqbucket_nfree > 0);
1757 /*
1758 * Go to sleep unless we are closing.
1759 * If a thread is sleeping too long, it dies.
1760 */
1761 if (! (bucket->tqbucket_flags & TQBUCKET_CLOSE)) {
1762 w = taskq_thread_wait(tq, lock, cv,
1763 &cprinfo, taskq_thread_timeout * hz);
1764 }
1765
1766 /*
1767 * At this point we may be in two different states:
1768 *
1769 * (1) tqent_func is set which means that a new task is
1770 * dispatched and we need to execute it.
1771 *
1772 * (2) Thread is sleeping for too long or we are closing. In
1773 * both cases destroy the thread and the entry.
1774 */
1775
1776 /* If func is NULL we should be on the freelist. */
1777 ASSERT((tqe->tqent_func != NULL) ||
1778 (bucket->tqbucket_nfree > 0));
1779 /* If func is non-NULL we should be allocated */
1780 ASSERT((tqe->tqent_func == NULL) ||
1781 (bucket->tqbucket_nalloc > 0));
1782
1783 /* Check freelist consistency */
1784 ASSERT((bucket->tqbucket_nfree > 0) ||
1785 IS_EMPTY(bucket->tqbucket_freelist));
1786 ASSERT((bucket->tqbucket_nfree == 0) ||
1787 !IS_EMPTY(bucket->tqbucket_freelist));
1788
1789 if ((tqe->tqent_func == NULL) &&
1790 ((w == -1) || (bucket->tqbucket_flags & TQBUCKET_CLOSE))) {
1791 /*
1792 * This thread is sleeping for too long or we are
1793 * closing - time to die.
1794 * Thread creation/destruction happens rarely,
1795 * so grabbing the lock is not a big performance issue.
1796 * The bucket lock is dropped by CALLB_CPR_EXIT().
1797 */
1798
1799 /* Remove the entry from the free list. */
1800 tqe->tqent_prev->tqent_next = tqe->tqent_next;
1801 tqe->tqent_next->tqent_prev = tqe->tqent_prev;
1802 ASSERT(bucket->tqbucket_nfree > 0);
1803 bucket->tqbucket_nfree--;
1804
1805 TQ_STAT(bucket, tqs_tdeaths);
1806 cv_signal(&bucket->tqbucket_cv);
1807 tqe->tqent_thread = NULL;
1808 mutex_enter(&tq->tq_lock);
1809 tq->tq_tdeaths++;
1810 mutex_exit(&tq->tq_lock);
1811 CALLB_CPR_EXIT(&cprinfo);
1812 kmem_cache_free(taskq_ent_cache, tqe);
1813 thread_exit();
1814 }
1815 }
1816 }
1817
1818
1819 /*
1820 * Taskq creation. May sleep for memory.
1821 * Always use automatically generated instances to avoid kstat name space
1822 * collisions.
1823 */
1824
1825 taskq_t *
1826 taskq_create(const char *name, int nthreads, pri_t pri, int minalloc,
1827 int maxalloc, uint_t flags)
1828 {
1829 ASSERT((flags & ~TASKQ_INTERFACE_FLAGS) == 0);
1830
1831 return (taskq_create_common(name, 0, nthreads, pri, minalloc,
1832 maxalloc, &p0, 0, flags | TASKQ_NOINSTANCE));
1833 }
1834
1835 /*
1836 * Create an instance of task queue. It is legal to create task queues with the
1837 * same name and different instances.
1838 *
1839 * taskq_create_instance is used by ddi_taskq_create() where it gets the
1840 * instance from ddi_get_instance(). In some cases the instance is not
1841 * initialized and is set to -1. This case is handled as if no instance was
1842 * passed at all.
1843 */
1844 taskq_t *
1845 taskq_create_instance(const char *name, int instance, int nthreads, pri_t pri,
1846 int minalloc, int maxalloc, uint_t flags)
1847 {
1848 ASSERT((flags & ~TASKQ_INTERFACE_FLAGS) == 0);
1849 ASSERT((instance >= 0) || (instance == -1));
1850
1851 if (instance < 0) {
1852 flags |= TASKQ_NOINSTANCE;
1853 }
1854
1855 return (taskq_create_common(name, instance, nthreads,
1856 pri, minalloc, maxalloc, &p0, 0, flags));
1857 }
1858
1859 taskq_t *
1860 taskq_create_proc(const char *name, int nthreads, pri_t pri, int minalloc,
1861 int maxalloc, proc_t *proc, uint_t flags)
1862 {
1863 ASSERT((flags & ~TASKQ_INTERFACE_FLAGS) == 0);
1864 ASSERT(proc->p_flag & SSYS);
1865
1866 return (taskq_create_common(name, 0, nthreads, pri, minalloc,
1867 maxalloc, proc, 0, flags | TASKQ_NOINSTANCE));
1868 }
1869
1870 taskq_t *
1871 taskq_create_sysdc(const char *name, int nthreads, int minalloc,
1872 int maxalloc, proc_t *proc, uint_t dc, uint_t flags)
1873 {
1874 ASSERT((flags & ~TASKQ_INTERFACE_FLAGS) == 0);
1875 ASSERT(proc->p_flag & SSYS);
1876
1877 return (taskq_create_common(name, 0, nthreads, minclsyspri, minalloc,
1878 maxalloc, proc, dc, flags | TASKQ_NOINSTANCE | TASKQ_DUTY_CYCLE));
1879 }
1880
1881 static taskq_t *
1882 taskq_create_common(const char *name, int instance, int nthreads, pri_t pri,
1883 int minalloc, int maxalloc, proc_t *proc, uint_t dc, uint_t flags)
1884 {
1885 taskq_t *tq = kmem_cache_alloc(taskq_cache, KM_SLEEP);
1886 uint_t ncpus = ((boot_max_ncpus == -1) ? max_ncpus : boot_max_ncpus);
1887 uint_t bsize; /* # of buckets - always power of 2 */
1888 int max_nthreads;
1889
1890 /*
1891 * TASKQ_DYNAMIC, TASKQ_CPR_SAFE and TASKQ_THREADS_CPU_PCT are all
1892 * mutually incompatible.
1893 */
1894 IMPLY((flags & TASKQ_DYNAMIC), !(flags & TASKQ_CPR_SAFE));
1895 IMPLY((flags & TASKQ_DYNAMIC), !(flags & TASKQ_THREADS_CPU_PCT));
1896 IMPLY((flags & TASKQ_CPR_SAFE), !(flags & TASKQ_THREADS_CPU_PCT));
1897
1898 /* Cannot have DYNAMIC with DUTY_CYCLE */
1899 IMPLY((flags & TASKQ_DYNAMIC), !(flags & TASKQ_DUTY_CYCLE));
1900
1901 /* Cannot have DUTY_CYCLE with a p0 kernel process */
1902 IMPLY((flags & TASKQ_DUTY_CYCLE), proc != &p0);
1903
1904 /* Cannot have DC_BATCH without DUTY_CYCLE */
1905 ASSERT((flags & (TASKQ_DUTY_CYCLE|TASKQ_DC_BATCH)) != TASKQ_DC_BATCH);
1906
1907 ASSERT(proc != NULL);
1908
1909 bsize = 1 << (highbit(ncpus) - 1);
1910 ASSERT(bsize >= 1);
1911 bsize = MIN(bsize, taskq_maxbuckets);
1912
1913 if (flags & TASKQ_DYNAMIC) {
1914 ASSERT3S(nthreads, >=, 1);
1915 tq->tq_maxsize = nthreads;
1916
1917 /* For dynamic task queues use just one backup thread */
1918 nthreads = max_nthreads = 1;
1919
1920 } else if (flags & TASKQ_THREADS_CPU_PCT) {
1921 uint_t pct;
1922 ASSERT3S(nthreads, >=, 0);
1923 pct = nthreads;
1924
1925 if (pct > taskq_cpupct_max_percent)
1926 pct = taskq_cpupct_max_percent;
1927
1928 /*
1929 * If you're using THREADS_CPU_PCT, the process for the
1930 * taskq threads must be curproc. This allows any pset
1931 * binding to be inherited correctly. If proc is &p0,
1932 * we won't be creating LWPs, so new threads will be assigned
1933 * to the default processor set.
1934 */
1935 ASSERT(curproc == proc || proc == &p0);
1936 tq->tq_threads_ncpus_pct = pct;
1937 nthreads = 1; /* corrected in taskq_thread_create() */
1938 max_nthreads = TASKQ_THREADS_PCT(max_ncpus, pct);
1939
1940 } else {
1941 ASSERT3S(nthreads, >=, 1);
1942 max_nthreads = nthreads;
1943 }
1944
1945 if (max_nthreads < taskq_minimum_nthreads_max)
1946 max_nthreads = taskq_minimum_nthreads_max;
1947
1948 /*
1949 * Make sure the name is 0-terminated, and conforms to the rules for
1950 * C indentifiers
1951 */
1952 (void) strncpy(tq->tq_name, name, TASKQ_NAMELEN + 1);
1953 strident_canon(tq->tq_name, TASKQ_NAMELEN + 1);
1954
1955 tq->tq_flags = flags | TASKQ_CHANGING;
1956 tq->tq_active = 0;
1957 tq->tq_instance = instance;
1958 tq->tq_nthreads_target = nthreads;
1959 tq->tq_nthreads_max = max_nthreads;
1960 tq->tq_minalloc = minalloc;
1961 tq->tq_maxalloc = maxalloc;
1962 tq->tq_nbuckets = bsize;
1963 tq->tq_proc = proc;
1964 tq->tq_pri = pri;
1965 tq->tq_DC = dc;
1966 list_link_init(&tq->tq_cpupct_link);
1967
1968 if (max_nthreads > 1)
1969 tq->tq_threadlist = kmem_alloc(
1970 sizeof (kthread_t *) * max_nthreads, KM_SLEEP);
1971
1972 mutex_enter(&tq->tq_lock);
1973 if (flags & TASKQ_PREPOPULATE) {
1974 while (minalloc-- > 0)
1975 taskq_ent_free(tq, taskq_ent_alloc(tq, TQ_SLEEP));
1976 }
1977
1978 /*
1979 * Before we start creating threads for this taskq, take a
1980 * zone hold so the zone can't go away before taskq_destroy
1981 * makes sure all the taskq threads are gone. This hold is
1982 * similar in purpose to those taken by zthread_create().
1983 */
1984 zone_hold(tq->tq_proc->p_zone);
1985
1986 /*
1987 * Create the first thread, which will create any other threads
1988 * necessary. taskq_thread_create will not return until we have
1989 * enough threads to be able to process requests.
1990 */
1991 taskq_thread_create(tq);
1992 mutex_exit(&tq->tq_lock);
1993
1994 if (flags & TASKQ_DYNAMIC) {
1995 taskq_bucket_t *bucket = kmem_zalloc(sizeof (taskq_bucket_t) *
1996 bsize, KM_SLEEP);
1997 int b_id;
1998
1999 tq->tq_buckets = bucket;
2000
2001 /* Initialize each bucket */
2002 for (b_id = 0; b_id < bsize; b_id++, bucket++) {
2003 mutex_init(&bucket->tqbucket_lock, NULL, MUTEX_DEFAULT,
2004 NULL);
2005 cv_init(&bucket->tqbucket_cv, NULL, CV_DEFAULT, NULL);
2006 bucket->tqbucket_taskq = tq;
2007 bucket->tqbucket_freelist.tqent_next =
2008 bucket->tqbucket_freelist.tqent_prev =
2009 &bucket->tqbucket_freelist;
2010 if (flags & TASKQ_PREPOPULATE)
2011 taskq_bucket_extend(bucket);
2012 }
2013 }
2014
2015 /*
2016 * Install kstats.
2017 * We have two cases:
2018 * 1) Instance is provided to taskq_create_instance(). In this case it
2019 * should be >= 0 and we use it.
2020 *
2021 * 2) Instance is not provided and is automatically generated
2022 */
2023 if (flags & TASKQ_NOINSTANCE) {
2024 instance = tq->tq_instance =
2025 (int)(uintptr_t)vmem_alloc(taskq_id_arena, 1, VM_SLEEP);
2026 }
2027
2028 if (flags & TASKQ_DYNAMIC) {
2029 if ((tq->tq_kstat = kstat_create("unix", instance,
2030 tq->tq_name, "taskq_d", KSTAT_TYPE_NAMED,
2031 sizeof (taskq_d_kstat) / sizeof (kstat_named_t),
2032 KSTAT_FLAG_VIRTUAL)) != NULL) {
2033 tq->tq_kstat->ks_lock = &taskq_d_kstat_lock;
2034 tq->tq_kstat->ks_data = &taskq_d_kstat;
2035 tq->tq_kstat->ks_update = taskq_d_kstat_update;
2036 tq->tq_kstat->ks_private = tq;
2037 kstat_install(tq->tq_kstat);
2038 }
2039 } else {
2040 if ((tq->tq_kstat = kstat_create("unix", instance, tq->tq_name,
2041 "taskq", KSTAT_TYPE_NAMED,
2042 sizeof (taskq_kstat) / sizeof (kstat_named_t),
2043 KSTAT_FLAG_VIRTUAL)) != NULL) {
2044 tq->tq_kstat->ks_lock = &taskq_kstat_lock;
2045 tq->tq_kstat->ks_data = &taskq_kstat;
2046 tq->tq_kstat->ks_update = taskq_kstat_update;
2047 tq->tq_kstat->ks_private = tq;
2048 kstat_install(tq->tq_kstat);
2049 }
2050 }
2051
2052 return (tq);
2053 }
2054
2055 /*
2056 * taskq_destroy().
2057 *
2058 * Assumes: by the time taskq_destroy is called no one will use this task queue
2059 * in any way and no one will try to dispatch entries in it.
2060 */
2061 void
2062 taskq_destroy(taskq_t *tq)
2063 {
2064 taskq_bucket_t *b = tq->tq_buckets;
2065 int bid = 0;
2066
2067 ASSERT(! (tq->tq_flags & TASKQ_CPR_SAFE));
2068
2069 /*
2070 * Destroy kstats.
2071 */
2072 if (tq->tq_kstat != NULL) {
2073 kstat_delete(tq->tq_kstat);
2074 tq->tq_kstat = NULL;
2075 }
2076
2077 /*
2078 * Destroy instance if needed.
2079 */
2080 if (tq->tq_flags & TASKQ_NOINSTANCE) {
2081 vmem_free(taskq_id_arena, (void *)(uintptr_t)(tq->tq_instance),
2082 1);
2083 tq->tq_instance = 0;
2084 }
2085
2086 /*
2087 * Unregister from the cpupct list.
2088 */
2089 if (tq->tq_flags & TASKQ_THREADS_CPU_PCT) {
2090 taskq_cpupct_remove(tq);
2091 }
2092
2093 /*
2094 * Wait for any pending entries to complete.
2095 */
2096 taskq_wait(tq);
2097
2098 mutex_enter(&tq->tq_lock);
2099 ASSERT((tq->tq_task.tqent_next == &tq->tq_task) &&
2100 (tq->tq_active == 0));
2101
2102 /* notify all the threads that they need to exit */
2103 tq->tq_nthreads_target = 0;
2104
2105 tq->tq_flags |= TASKQ_CHANGING;
2106 cv_broadcast(&tq->tq_dispatch_cv);
2107 cv_broadcast(&tq->tq_exit_cv);
2108
2109 while (tq->tq_nthreads != 0)
2110 cv_wait(&tq->tq_wait_cv, &tq->tq_lock);
2111
2112 if (tq->tq_nthreads_max != 1)
2113 kmem_free(tq->tq_threadlist, sizeof (kthread_t *) *
2114 tq->tq_nthreads_max);
2115
2116 tq->tq_minalloc = 0;
2117 while (tq->tq_nalloc != 0)
2118 taskq_ent_free(tq, taskq_ent_alloc(tq, TQ_SLEEP));
2119
2120 mutex_exit(&tq->tq_lock);
2121
2122 /*
2123 * Mark each bucket as closing and wakeup all sleeping threads.
2124 */
2125 for (; (b != NULL) && (bid < tq->tq_nbuckets); b++, bid++) {
2126 taskq_ent_t *tqe;
2127
2128 mutex_enter(&b->tqbucket_lock);
2129
2130 b->tqbucket_flags |= TQBUCKET_CLOSE;
2131 /* Wakeup all sleeping threads */
2132
2133 for (tqe = b->tqbucket_freelist.tqent_next;
2134 tqe != &b->tqbucket_freelist; tqe = tqe->tqent_next)
2135 cv_signal(&tqe->tqent_cv);
2136
2137 ASSERT(b->tqbucket_nalloc == 0);
2138
2139 /*
2140 * At this point we waited for all pending jobs to complete (in
2141 * both the task queue and the bucket and no new jobs should
2142 * arrive. Wait for all threads to die.
2143 */
2144 while (b->tqbucket_nfree > 0)
2145 cv_wait(&b->tqbucket_cv, &b->tqbucket_lock);
2146 mutex_exit(&b->tqbucket_lock);
2147 mutex_destroy(&b->tqbucket_lock);
2148 cv_destroy(&b->tqbucket_cv);
2149 }
2150
2151 if (tq->tq_buckets != NULL) {
2152 ASSERT(tq->tq_flags & TASKQ_DYNAMIC);
2153 kmem_free(tq->tq_buckets,
2154 sizeof (taskq_bucket_t) * tq->tq_nbuckets);
2155
2156 /* Cleanup fields before returning tq to the cache */
2157 tq->tq_buckets = NULL;
2158 tq->tq_tcreates = 0;
2159 tq->tq_tdeaths = 0;
2160 } else {
2161 ASSERT(!(tq->tq_flags & TASKQ_DYNAMIC));
2162 }
2163
2164 /*
2165 * Now that all the taskq threads are gone, we can
2166 * drop the zone hold taken in taskq_create_common
2167 */
2168 zone_rele(tq->tq_proc->p_zone);
2169
2170 tq->tq_threads_ncpus_pct = 0;
2171 tq->tq_totaltime = 0;
2172 tq->tq_tasks = 0;
2173 tq->tq_maxtasks = 0;
2174 tq->tq_executed = 0;
2175 kmem_cache_free(taskq_cache, tq);
2176 }
2177
2178 /*
2179 * Extend a bucket with a new entry on the free list and attach a worker thread
2180 * to it.
2181 *
2182 * Argument: pointer to the bucket.
2183 *
2184 * This function may quietly fail. It is only used by taskq_dispatch() which
2185 * handles such failures properly.
2186 */
2187 static void
2188 taskq_bucket_extend(void *arg)
2189 {
2190 taskq_ent_t *tqe;
2191 taskq_bucket_t *b = (taskq_bucket_t *)arg;
2192 taskq_t *tq = b->tqbucket_taskq;
2193 int nthreads;
2194
2195 mutex_enter(&tq->tq_lock);
2196
2197 if (! ENOUGH_MEMORY()) {
2198 tq->tq_nomem++;
2199 mutex_exit(&tq->tq_lock);
2200 return;
2201 }
2202
2203 /*
2204 * Observe global taskq limits on the number of threads.
2205 */
2206 if (tq->tq_tcreates++ - tq->tq_tdeaths > tq->tq_maxsize) {
2207 tq->tq_tcreates--;
2208 mutex_exit(&tq->tq_lock);
2209 return;
2210 }
2211 mutex_exit(&tq->tq_lock);
2212
2213 tqe = kmem_cache_alloc(taskq_ent_cache, KM_NOSLEEP);
2214
2215 if (tqe == NULL) {
2216 mutex_enter(&tq->tq_lock);
2217 tq->tq_nomem++;
2218 tq->tq_tcreates--;
2219 mutex_exit(&tq->tq_lock);
2220 return;
2221 }
2222
2223 ASSERT(tqe->tqent_thread == NULL);
2224
2225 tqe->tqent_un.tqent_bucket = b;
2226
2227 /*
2228 * Create a thread in a TS_STOPPED state first. If it is successfully
2229 * created, place the entry on the free list and start the thread.
2230 */
2231 tqe->tqent_thread = thread_create(NULL, 0, taskq_d_thread, tqe,
2232 0, tq->tq_proc, TS_STOPPED, tq->tq_pri);
2233
2234 /*
2235 * Once the entry is ready, link it to the the bucket free list.
2236 */
2237 mutex_enter(&b->tqbucket_lock);
2238 tqe->tqent_func = NULL;
2239 TQ_APPEND(b->tqbucket_freelist, tqe);
2240 b->tqbucket_nfree++;
2241 TQ_STAT(b, tqs_tcreates);
2242
2243 #if TASKQ_STATISTIC
2244 nthreads = b->tqbucket_stat.tqs_tcreates -
2245 b->tqbucket_stat.tqs_tdeaths;
2246 b->tqbucket_stat.tqs_maxthreads = MAX(nthreads,
2247 b->tqbucket_stat.tqs_maxthreads);
2248 #endif
2249
2250 mutex_exit(&b->tqbucket_lock);
2251 /*
2252 * Start the stopped thread.
2253 */
2254 thread_lock(tqe->tqent_thread);
2255 tqe->tqent_thread->t_taskq = tq;
2256 tqe->tqent_thread->t_schedflag |= TS_ALLSTART;
2257 setrun_locked(tqe->tqent_thread);
2258 thread_unlock(tqe->tqent_thread);
2259 }
2260
2261 static int
2262 taskq_kstat_update(kstat_t *ksp, int rw)
2263 {
2264 struct taskq_kstat *tqsp = &taskq_kstat;
2265 taskq_t *tq = ksp->ks_private;
2266
2267 if (rw == KSTAT_WRITE)
2268 return (EACCES);
2269
2270 tqsp->tq_pid.value.ui64 = tq->tq_proc->p_pid;
2271 tqsp->tq_tasks.value.ui64 = tq->tq_tasks;
2272 tqsp->tq_executed.value.ui64 = tq->tq_executed;
2273 tqsp->tq_maxtasks.value.ui64 = tq->tq_maxtasks;
2274 tqsp->tq_totaltime.value.ui64 = tq->tq_totaltime;
2275 tqsp->tq_nactive.value.ui64 = tq->tq_active;
2276 tqsp->tq_nalloc.value.ui64 = tq->tq_nalloc;
2277 tqsp->tq_pri.value.ui64 = tq->tq_pri;
2278 tqsp->tq_nthreads.value.ui64 = tq->tq_nthreads;
2279 tqsp->tq_nomem.value.ui64 = tq->tq_nomem;
2280 return (0);
2281 }
2282
2283 static int
2284 taskq_d_kstat_update(kstat_t *ksp, int rw)
2285 {
2286 struct taskq_d_kstat *tqsp = &taskq_d_kstat;
2287 taskq_t *tq = ksp->ks_private;
2288 taskq_bucket_t *b = tq->tq_buckets;
2289 int bid = 0;
2290
2291 if (rw == KSTAT_WRITE)
2292 return (EACCES);
2293
2294 ASSERT(tq->tq_flags & TASKQ_DYNAMIC);
2295
2296 tqsp->tqd_btasks.value.ui64 = tq->tq_tasks;
2297 tqsp->tqd_bexecuted.value.ui64 = tq->tq_executed;
2298 tqsp->tqd_bmaxtasks.value.ui64 = tq->tq_maxtasks;
2299 tqsp->tqd_bnalloc.value.ui64 = tq->tq_nalloc;
2300 tqsp->tqd_bnactive.value.ui64 = tq->tq_active;
2301 tqsp->tqd_btotaltime.value.ui64 = tq->tq_totaltime;
2302 tqsp->tqd_pri.value.ui64 = tq->tq_pri;
2303 tqsp->tqd_nomem.value.ui64 = tq->tq_nomem;
2304
2305 tqsp->tqd_hits.value.ui64 = 0;
2306 tqsp->tqd_misses.value.ui64 = 0;
2307 tqsp->tqd_overflows.value.ui64 = 0;
2308 tqsp->tqd_tcreates.value.ui64 = 0;
2309 tqsp->tqd_tdeaths.value.ui64 = 0;
2310 tqsp->tqd_maxthreads.value.ui64 = 0;
2311 tqsp->tqd_nomem.value.ui64 = 0;
2312 tqsp->tqd_disptcreates.value.ui64 = 0;
2313 tqsp->tqd_totaltime.value.ui64 = 0;
2314 tqsp->tqd_nalloc.value.ui64 = 0;
2315 tqsp->tqd_nfree.value.ui64 = 0;
2316
2317 for (; (b != NULL) && (bid < tq->tq_nbuckets); b++, bid++) {
2318 tqsp->tqd_hits.value.ui64 += b->tqbucket_stat.tqs_hits;
2319 tqsp->tqd_misses.value.ui64 += b->tqbucket_stat.tqs_misses;
2320 tqsp->tqd_overflows.value.ui64 += b->tqbucket_stat.tqs_overflow;
2321 tqsp->tqd_tcreates.value.ui64 += b->tqbucket_stat.tqs_tcreates;
2322 tqsp->tqd_tdeaths.value.ui64 += b->tqbucket_stat.tqs_tdeaths;
2323 tqsp->tqd_maxthreads.value.ui64 +=
2324 b->tqbucket_stat.tqs_maxthreads;
2325 tqsp->tqd_disptcreates.value.ui64 +=
2326 b->tqbucket_stat.tqs_disptcreates;
2327 tqsp->tqd_totaltime.value.ui64 += b->tqbucket_totaltime;
2328 tqsp->tqd_nalloc.value.ui64 += b->tqbucket_nalloc;
2329 tqsp->tqd_nfree.value.ui64 += b->tqbucket_nfree;
2330 }
2331 return (0);
2332 }