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