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, Version 1.0 only 6 * (the "License"). You may not use this file except in compliance 7 * with the License. 8 * 9 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE 10 * or http://www.opensolaris.org/os/licensing. 11 * See the License for the specific language governing permissions 12 * and limitations under the License. 13 * 14 * When distributing Covered Code, include this CDDL HEADER in each 15 * file and include the License file at usr/src/OPENSOLARIS.LICENSE. 16 * If applicable, add the following below this CDDL HEADER, with the 17 * fields enclosed by brackets "[]" replaced with your own identifying 18 * information: Portions Copyright [yyyy] [name of copyright owner] 19 * 20 * CDDL HEADER END 21 */ 22 /* 23 * Copyright 2005 Sun Microsystems, Inc. All rights reserved. 24 * Use is subject to license terms. 25 */ 26 27 #pragma ident "%Z%%M% %I% %E% SMI" 28 29 /* 30 * Kernel task queues: general-purpose asynchronous task scheduling. 31 * 32 * A common problem in kernel programming is the need to schedule tasks 33 * to be performed later, by another thread. There are several reasons 34 * you may want or need to do this: 35 * 36 * (1) The task isn't time-critical, but your current code path is. 37 * 38 * (2) The task may require grabbing locks that you already hold. 39 * 40 * (3) The task may need to block (e.g. to wait for memory), but you 41 * cannot block in your current context. 42 * 43 * (4) Your code path can't complete because of some condition, but you can't 44 * sleep or fail, so you queue the task for later execution when condition 45 * disappears. 46 * 47 * (5) You just want a simple way to launch multiple tasks in parallel. 48 * 49 * Task queues provide such a facility. In its simplest form (used when 50 * performance is not a critical consideration) a task queue consists of a 51 * single list of tasks, together with one or more threads to service the 52 * list. There are some cases when this simple queue is not sufficient: 53 * 54 * (1) The task queues are very hot and there is a need to avoid data and lock 55 * contention over global resources. 56 * 57 * (2) Some tasks may depend on other tasks to complete, so they can't be put in 58 * the same list managed by the same thread. 59 * 60 * (3) Some tasks may block for a long time, and this should not block other 61 * tasks in the queue. 62 * 63 * To provide useful service in such cases we define a "dynamic task queue" 64 * which has an individual thread for each of the tasks. These threads are 65 * dynamically created as they are needed and destroyed when they are not in 66 * use. The API for managing task pools is the same as for managing task queues 67 * with the exception of a taskq creation flag TASKQ_DYNAMIC which tells that 68 * dynamic task pool behavior is desired. 69 * 70 * Dynamic task queues may also place tasks in the normal queue (called "backing 71 * queue") when task pool runs out of resources. Users of task queues may 72 * disallow such queued scheduling by specifying TQ_NOQUEUE in the dispatch 73 * flags. 74 * 75 * The backing task queue is also used for scheduling internal tasks needed for 76 * dynamic task queue maintenance. 77 * 78 * INTERFACES: 79 * 80 * taskq_t *taskq_create(name, nthreads, pri_t pri, minalloc, maxall, flags); 81 * 82 * Create a taskq with specified properties. 83 * Possible 'flags': 84 * 85 * TASKQ_DYNAMIC: Create task pool for task management. If this flag is 86 * specified, 'nthreads' specifies the maximum number of threads in 87 * the task queue. Task execution order for dynamic task queues is 88 * not predictable. 89 * 90 * If this flag is not specified (default case) a 91 * single-list task queue is created with 'nthreads' threads 92 * servicing it. Entries in this queue are managed by 93 * taskq_ent_alloc() and taskq_ent_free() which try to keep the 94 * task population between 'minalloc' and 'maxalloc', but the 95 * latter limit is only advisory for TQ_SLEEP dispatches and the 96 * former limit is only advisory for TQ_NOALLOC dispatches. If 97 * TASKQ_PREPOPULATE is set in 'flags', the taskq will be 98 * prepopulated with 'minalloc' task structures. 99 * 100 * Since non-DYNAMIC taskqs are queues, tasks are guaranteed to be 101 * executed in the order they are scheduled if nthreads == 1. 102 * If nthreads > 1, task execution order is not predictable. 103 * 104 * TASKQ_PREPOPULATE: Prepopulate task queue with threads. 105 * Also prepopulate the task queue with 'minalloc' task structures. 106 * 107 * TASKQ_CPR_SAFE: This flag specifies that users of the task queue will 108 * use their own protocol for handling CPR issues. This flag is not 109 * supported for DYNAMIC task queues. 110 * 111 * The 'pri' field specifies the default priority for the threads that 112 * service all scheduled tasks. 113 * 114 * void taskq_destroy(tap): 115 * 116 * Waits for any scheduled tasks to complete, then destroys the taskq. 117 * Caller should guarantee that no new tasks are scheduled in the closing 118 * taskq. 119 * 120 * taskqid_t taskq_dispatch(tq, func, arg, flags): 121 * 122 * Dispatches the task "func(arg)" to taskq. The 'flags' indicates whether 123 * the caller is willing to block for memory. The function returns an 124 * opaque value which is zero iff dispatch fails. If flags is TQ_NOSLEEP 125 * or TQ_NOALLOC and the task can't be dispatched, taskq_dispatch() fails 126 * and returns (taskqid_t)0. 127 * 128 * ASSUMES: func != NULL. 129 * 130 * Possible flags: 131 * TQ_NOSLEEP: Do not wait for resources; may fail. 132 * 133 * TQ_NOALLOC: Do not allocate memory; may fail. May only be used with 134 * non-dynamic task queues. 135 * 136 * TQ_NOQUEUE: Do not enqueue a task if it can't dispatch it due to 137 * lack of available resources and fail. If this flag is not 138 * set, and the task pool is exhausted, the task may be scheduled 139 * in the backing queue. This flag may ONLY be used with dynamic 140 * task queues. 141 * 142 * NOTE: This flag should always be used when a task queue is used 143 * for tasks that may depend on each other for completion. 144 * Enqueueing dependent tasks may create deadlocks. 145 * 146 * TQ_SLEEP: May block waiting for resources. May still fail for 147 * dynamic task queues if TQ_NOQUEUE is also specified, otherwise 148 * always succeed. 149 * 150 * NOTE: Dynamic task queues are much more likely to fail in 151 * taskq_dispatch() (especially if TQ_NOQUEUE was specified), so it 152 * is important to have backup strategies handling such failures. 153 * 154 * void taskq_wait(tq): 155 * 156 * Waits for all previously scheduled tasks to complete. 157 * 158 * NOTE: It does not stop any new task dispatches. 159 * Do NOT call taskq_wait() from a task: it will cause deadlock. 160 * 161 * void taskq_suspend(tq) 162 * 163 * Suspend all task execution. Tasks already scheduled for a dynamic task 164 * queue will still be executed, but all new scheduled tasks will be 165 * suspended until taskq_resume() is called. 166 * 167 * int taskq_suspended(tq) 168 * 169 * Returns 1 if taskq is suspended and 0 otherwise. It is intended to 170 * ASSERT that the task queue is suspended. 171 * 172 * void taskq_resume(tq) 173 * 174 * Resume task queue execution. 175 * 176 * int taskq_member(tq, thread) 177 * 178 * Returns 1 if 'thread' belongs to taskq 'tq' and 0 otherwise. The 179 * intended use is to ASSERT that a given function is called in taskq 180 * context only. 181 * 182 * system_taskq 183 * 184 * Global system-wide dynamic task queue for common uses. It may be used by 185 * any subsystem that needs to schedule tasks and does not need to manage 186 * its own task queues. It is initialized quite early during system boot. 187 * 188 * IMPLEMENTATION. 189 * 190 * This is schematic representation of the task queue structures. 191 * 192 * taskq: 193 * +-------------+ 194 * |tq_lock | +---< taskq_ent_free() 195 * +-------------+ | 196 * |... | | tqent: tqent: 197 * +-------------+ | +------------+ +------------+ 198 * | tq_freelist |-->| tqent_next |--> ... ->| tqent_next | 199 * +-------------+ +------------+ +------------+ 200 * |... | | ... | | ... | 201 * +-------------+ +------------+ +------------+ 202 * | tq_task | | 203 * | | +-------------->taskq_ent_alloc() 204 * +--------------------------------------------------------------------------+ 205 * | | | tqent tqent | 206 * | +---------------------+ +--> +------------+ +--> +------------+ | 207 * | | ... | | | func, arg | | | func, arg | | 208 * +>+---------------------+ <---|-+ +------------+ <---|-+ +------------+ | 209 * | tq_taskq.tqent_next | ----+ | | tqent_next | --->+ | | tqent_next |--+ 210 * +---------------------+ | +------------+ ^ | +------------+ 211 * +-| tq_task.tqent_prev | +--| tqent_prev | | +--| tqent_prev | ^ 212 * | +---------------------+ +------------+ | +------------+ | 213 * | |... | | ... | | | ... | | 214 * | +---------------------+ +------------+ | +------------+ | 215 * | ^ | | 216 * | | | | 217 * +--------------------------------------+--------------+ TQ_APPEND() -+ 218 * | | | 219 * |... | taskq_thread()-----+ 220 * +-------------+ 221 * | tq_buckets |--+-------> [ NULL ] (for regular task queues) 222 * +-------------+ | 223 * | DYNAMIC TASK QUEUES: 224 * | 225 * +-> taskq_bucket[nCPU] taskq_bucket_dispatch() 226 * +-------------------+ ^ 227 * +--->| tqbucket_lock | | 228 * | +-------------------+ +--------+ +--------+ 229 * | | tqbucket_freelist |-->| tqent |-->...| tqent | ^ 230 * | +-------------------+<--+--------+<--...+--------+ | 231 * | | ... | | thread | | thread | | 232 * | +-------------------+ +--------+ +--------+ | 233 * | +-------------------+ | 234 * taskq_dispatch()--+--->| tqbucket_lock | TQ_APPEND()------+ 235 * TQ_HASH() | +-------------------+ +--------+ +--------+ 236 * | | tqbucket_freelist |-->| tqent |-->...| tqent | 237 * | +-------------------+<--+--------+<--...+--------+ 238 * | | ... | | thread | | thread | 239 * | +-------------------+ +--------+ +--------+ 240 * +---> ... 241 * 242 * 243 * Task queues use tq_task field to link new entry in the queue. The queue is a 244 * circular doubly-linked list. Entries are put in the end of the list with 245 * TQ_APPEND() and processed from the front of the list by taskq_thread() in 246 * FIFO order. Task queue entries are cached in the free list managed by 247 * taskq_ent_alloc() and taskq_ent_free() functions. 248 * 249 * All threads used by task queues mark t_taskq field of the thread to 250 * point to the task queue. 251 * 252 * Dynamic Task Queues Implementation. 253 * 254 * For a dynamic task queues there is a 1-to-1 mapping between a thread and 255 * taskq_ent_structure. Each entry is serviced by its own thread and each thread 256 * is controlled by a single entry. 257 * 258 * Entries are distributed over a set of buckets. To avoid using modulo 259 * arithmetics the number of buckets is 2^n and is determined as the nearest 260 * power of two roundown of the number of CPUs in the system. Tunable 261 * variable 'taskq_maxbuckets' limits the maximum number of buckets. Each entry 262 * is attached to a bucket for its lifetime and can't migrate to other buckets. 263 * 264 * Entries that have scheduled tasks are not placed in any list. The dispatch 265 * function sets their "func" and "arg" fields and signals the corresponding 266 * thread to execute the task. Once the thread executes the task it clears the 267 * "func" field and places an entry on the bucket cache of free entries pointed 268 * by "tqbucket_freelist" field. ALL entries on the free list should have "func" 269 * field equal to NULL. The free list is a circular doubly-linked list identical 270 * in structure to the tq_task list above, but entries are taken from it in LIFO 271 * order - the last freed entry is the first to be allocated. The 272 * taskq_bucket_dispatch() function gets the most recently used entry from the 273 * free list, sets its "func" and "arg" fields and signals a worker thread. 274 * 275 * After executing each task a per-entry thread taskq_d_thread() places its 276 * entry on the bucket free list and goes to a timed sleep. If it wakes up 277 * without getting new task it removes the entry from the free list and destroys 278 * itself. The thread sleep time is controlled by a tunable variable 279 * `taskq_thread_timeout'. 280 * 281 * There is various statistics kept in the bucket which allows for later 282 * analysis of taskq usage patterns. Also, a global copy of taskq creation and 283 * death statistics is kept in the global taskq data structure. Since thread 284 * creation and death happen rarely, updating such global data does not present 285 * a performance problem. 286 * 287 * NOTE: Threads are not bound to any CPU and there is absolutely no association 288 * between the bucket and actual thread CPU, so buckets are used only to 289 * split resources and reduce resource contention. Having threads attached 290 * to the CPU denoted by a bucket may reduce number of times the job 291 * switches between CPUs. 292 * 293 * Current algorithm creates a thread whenever a bucket has no free 294 * entries. It would be nice to know how many threads are in the running 295 * state and don't create threads if all CPUs are busy with existing 296 * tasks, but it is unclear how such strategy can be implemented. 297 * 298 * Currently buckets are created statically as an array attached to task 299 * queue. On some system with nCPUs < max_ncpus it may waste system 300 * memory. One solution may be allocation of buckets when they are first 301 * touched, but it is not clear how useful it is. 302 * 303 * SUSPEND/RESUME implementation. 304 * 305 * Before executing a task taskq_thread() (executing non-dynamic task 306 * queues) obtains taskq's thread lock as a reader. The taskq_suspend() 307 * function gets the same lock as a writer blocking all non-dynamic task 308 * execution. The taskq_resume() function releases the lock allowing 309 * taskq_thread to continue execution. 310 * 311 * For dynamic task queues, each bucket is marked as TQBUCKET_SUSPEND by 312 * taskq_suspend() function. After that taskq_bucket_dispatch() always 313 * fails, so that taskq_dispatch() will either enqueue tasks for a 314 * suspended backing queue or fail if TQ_NOQUEUE is specified in dispatch 315 * flags. 316 * 317 * NOTE: taskq_suspend() does not immediately block any tasks already 318 * scheduled for dynamic task queues. It only suspends new tasks 319 * scheduled after taskq_suspend() was called. 320 * 321 * taskq_member() function works by comparing a thread t_taskq pointer with 322 * the passed thread pointer. 323 * 324 * LOCKS and LOCK Hierarchy: 325 * 326 * There are two locks used in task queues. 327 * 328 * 1) Task queue structure has a lock, protecting global task queue state. 329 * 330 * 2) Each per-CPU bucket has a lock for bucket management. 331 * 332 * If both locks are needed, task queue lock should be taken only after bucket 333 * lock. 334 * 335 * DEBUG FACILITIES. 336 * 337 * For DEBUG kernels it is possible to induce random failures to 338 * taskq_dispatch() function when it is given TQ_NOSLEEP argument. The value of 339 * taskq_dmtbf and taskq_smtbf tunables control the mean time between induced 340 * failures for dynamic and static task queues respectively. 341 * 342 * Setting TASKQ_STATISTIC to 0 will disable per-bucket statistics. 343 * 344 * TUNABLES 345 * 346 * system_taskq_size - Size of the global system_taskq. 347 * This value is multiplied by nCPUs to determine 348 * actual size. 349 * Default value: 64 350 * 351 * taskq_thread_timeout - Maximum idle time for taskq_d_thread() 352 * Default value: 5 minutes 353 * 354 * taskq_maxbuckets - Maximum number of buckets in any task queue 355 * Default value: 128 356 * 357 * taskq_search_depth - Maximum # of buckets searched for a free entry 358 * Default value: 4 359 * 360 * taskq_dmtbf - Mean time between induced dispatch failures 361 * for dynamic task queues. 362 * Default value: UINT_MAX (no induced failures) 363 * 364 * taskq_smtbf - Mean time between induced dispatch failures 365 * for static task queues. 366 * Default value: UINT_MAX (no induced failures) 367 * 368 * CONDITIONAL compilation. 369 * 370 * TASKQ_STATISTIC - If set will enable bucket statistic (default). 371 * 372 */ 373 374 #include <sys/taskq_impl.h> 375 #include <sys/thread.h> 376 #include <sys/proc.h> 377 #include <sys/kmem.h> 378 #include <sys/vmem.h> 379 #include <sys/callb.h> 380 #include <sys/systm.h> 381 #include <sys/cmn_err.h> 382 #include <sys/debug.h> 383 #include <sys/vmsystm.h> /* For throttlefree */ 384 #include <sys/sysmacros.h> 385 #include <sys/cpuvar.h> 386 #include <sys/sdt.h> 387 388 static kmem_cache_t *taskq_ent_cache, *taskq_cache; 389 390 /* 391 * Pseudo instance numbers for taskqs without explicitely provided instance. 392 */ 393 static vmem_t *taskq_id_arena; 394 395 /* Global system task queue for common use */ 396 taskq_t *system_taskq; 397 398 /* 399 * Maxmimum number of entries in global system taskq is 400 * system_taskq_size * max_ncpus 401 */ 402 #define SYSTEM_TASKQ_SIZE 64 403 int system_taskq_size = SYSTEM_TASKQ_SIZE; 404 405 /* 406 * Dynamic task queue threads that don't get any work within 407 * taskq_thread_timeout destroy themselves 408 */ 409 #define TASKQ_THREAD_TIMEOUT (60 * 5) 410 int taskq_thread_timeout = TASKQ_THREAD_TIMEOUT; 411 412 #define TASKQ_MAXBUCKETS 128 413 int taskq_maxbuckets = TASKQ_MAXBUCKETS; 414 415 /* 416 * When a bucket has no available entries another buckets are tried. 417 * taskq_search_depth parameter limits the amount of buckets that we search 418 * before failing. This is mostly useful in systems with many CPUs where we may 419 * spend too much time scanning busy buckets. 420 */ 421 #define TASKQ_SEARCH_DEPTH 4 422 int taskq_search_depth = TASKQ_SEARCH_DEPTH; 423 424 /* 425 * Hashing function: mix various bits of x. May be pretty much anything. 426 */ 427 #define TQ_HASH(x) ((x) ^ ((x) >> 11) ^ ((x) >> 17) ^ ((x) ^ 27)) 428 429 /* 430 * We do not create any new threads when the system is low on memory and start 431 * throttling memory allocations. The following macro tries to estimate such 432 * condition. 433 */ 434 #define ENOUGH_MEMORY() (freemem > throttlefree) 435 436 /* 437 * Static functions. 438 */ 439 static taskq_t *taskq_create_common(const char *, int, int, pri_t, int, 440 int, uint_t); 441 static void taskq_thread(void *); 442 static void taskq_d_thread(taskq_ent_t *); 443 static void taskq_bucket_extend(void *); 444 static int taskq_constructor(void *, void *, int); 445 static void taskq_destructor(void *, void *); 446 static int taskq_ent_constructor(void *, void *, int); 447 static void taskq_ent_destructor(void *, void *); 448 static taskq_ent_t *taskq_ent_alloc(taskq_t *, int); 449 static void taskq_ent_free(taskq_t *, taskq_ent_t *); 450 static taskq_ent_t *taskq_bucket_dispatch(taskq_bucket_t *, task_func_t, 451 void *); 452 453 /* 454 * Task queues kstats. 455 */ 456 struct taskq_kstat { 457 kstat_named_t tq_tasks; 458 kstat_named_t tq_executed; 459 kstat_named_t tq_maxtasks; 460 kstat_named_t tq_totaltime; 461 kstat_named_t tq_nalloc; 462 kstat_named_t tq_nactive; 463 kstat_named_t tq_pri; 464 kstat_named_t tq_nthreads; 465 } taskq_kstat = { 466 { "tasks", KSTAT_DATA_UINT64 }, 467 { "executed", KSTAT_DATA_UINT64 }, 468 { "maxtasks", KSTAT_DATA_UINT64 }, 469 { "totaltime", KSTAT_DATA_UINT64 }, 470 { "nactive", KSTAT_DATA_UINT64 }, 471 { "nalloc", KSTAT_DATA_UINT64 }, 472 { "priority", KSTAT_DATA_UINT64 }, 473 { "threads", KSTAT_DATA_UINT64 }, 474 }; 475 476 struct taskq_d_kstat { 477 kstat_named_t tqd_pri; 478 kstat_named_t tqd_btasks; 479 kstat_named_t tqd_bexecuted; 480 kstat_named_t tqd_bmaxtasks; 481 kstat_named_t tqd_bnalloc; 482 kstat_named_t tqd_bnactive; 483 kstat_named_t tqd_btotaltime; 484 kstat_named_t tqd_hits; 485 kstat_named_t tqd_misses; 486 kstat_named_t tqd_overflows; 487 kstat_named_t tqd_tcreates; 488 kstat_named_t tqd_tdeaths; 489 kstat_named_t tqd_maxthreads; 490 kstat_named_t tqd_nomem; 491 kstat_named_t tqd_disptcreates; 492 kstat_named_t tqd_totaltime; 493 kstat_named_t tqd_nalloc; 494 kstat_named_t tqd_nfree; 495 } taskq_d_kstat = { 496 { "priority", KSTAT_DATA_UINT64 }, 497 { "btasks", KSTAT_DATA_UINT64 }, 498 { "bexecuted", KSTAT_DATA_UINT64 }, 499 { "bmaxtasks", KSTAT_DATA_UINT64 }, 500 { "bnalloc", KSTAT_DATA_UINT64 }, 501 { "bnactive", KSTAT_DATA_UINT64 }, 502 { "btotaltime", KSTAT_DATA_UINT64 }, 503 { "hits", KSTAT_DATA_UINT64 }, 504 { "misses", KSTAT_DATA_UINT64 }, 505 { "overflows", KSTAT_DATA_UINT64 }, 506 { "tcreates", KSTAT_DATA_UINT64 }, 507 { "tdeaths", KSTAT_DATA_UINT64 }, 508 { "maxthreads", KSTAT_DATA_UINT64 }, 509 { "nomem", KSTAT_DATA_UINT64 }, 510 { "disptcreates", KSTAT_DATA_UINT64 }, 511 { "totaltime", KSTAT_DATA_UINT64 }, 512 { "nalloc", KSTAT_DATA_UINT64 }, 513 { "nfree", KSTAT_DATA_UINT64 }, 514 }; 515 516 static kmutex_t taskq_kstat_lock; 517 static kmutex_t taskq_d_kstat_lock; 518 static int taskq_kstat_update(kstat_t *, int); 519 static int taskq_d_kstat_update(kstat_t *, int); 520 521 522 /* 523 * Collect per-bucket statistic when TASKQ_STATISTIC is defined. 524 */ 525 #define TASKQ_STATISTIC 1 526 527 #if TASKQ_STATISTIC 528 #define TQ_STAT(b, x) b->tqbucket_stat.x++ 529 #else 530 #define TQ_STAT(b, x) 531 #endif 532 533 /* 534 * Random fault injection. 535 */ 536 uint_t taskq_random; 537 uint_t taskq_dmtbf = UINT_MAX; /* mean time between injected failures */ 538 uint_t taskq_smtbf = UINT_MAX; /* mean time between injected failures */ 539 540 /* 541 * TQ_NOSLEEP dispatches on dynamic task queues are always allowed to fail. 542 * 543 * TQ_NOSLEEP dispatches on static task queues can't arbitrarily fail because 544 * they could prepopulate the cache and make sure that they do not use more 545 * then minalloc entries. So, fault injection in this case insures that 546 * either TASKQ_PREPOPULATE is not set or there are more entries allocated 547 * than is specified by minalloc. TQ_NOALLOC dispatches are always allowed 548 * to fail, but for simplicity we treat them identically to TQ_NOSLEEP 549 * dispatches. 550 */ 551 #ifdef DEBUG 552 #define TASKQ_D_RANDOM_DISPATCH_FAILURE(tq, flag) \ 553 taskq_random = (taskq_random * 2416 + 374441) % 1771875;\ 554 if ((flag & TQ_NOSLEEP) && \ 555 taskq_random < 1771875 / taskq_dmtbf) { \ 556 return (NULL); \ 557 } 558 559 #define TASKQ_S_RANDOM_DISPATCH_FAILURE(tq, flag) \ 560 taskq_random = (taskq_random * 2416 + 374441) % 1771875;\ 561 if ((flag & (TQ_NOSLEEP | TQ_NOALLOC)) && \ 562 (!(tq->tq_flags & TASKQ_PREPOPULATE) || \ 563 (tq->tq_nalloc > tq->tq_minalloc)) && \ 564 (taskq_random < (1771875 / taskq_smtbf))) { \ 565 mutex_exit(&tq->tq_lock); \ 566 return (NULL); \ 567 } 568 #else 569 #define TASKQ_S_RANDOM_DISPATCH_FAILURE(tq, flag) 570 #define TASKQ_D_RANDOM_DISPATCH_FAILURE(tq, flag) 571 #endif 572 573 #define IS_EMPTY(l) (((l).tqent_prev == (l).tqent_next) && \ 574 ((l).tqent_prev == &(l))) 575 576 /* 577 * Append `tqe' in the end of the doubly-linked list denoted by l. 578 */ 579 #define TQ_APPEND(l, tqe) { \ 580 tqe->tqent_next = &l; \ 581 tqe->tqent_prev = l.tqent_prev; \ 582 tqe->tqent_next->tqent_prev = tqe; \ 583 tqe->tqent_prev->tqent_next = tqe; \ 584 } 585 586 /* 587 * Schedule a task specified by func and arg into the task queue entry tqe. 588 */ 589 #define TQ_ENQUEUE(tq, tqe, func, arg) { \ 590 ASSERT(MUTEX_HELD(&tq->tq_lock)); \ 591 TQ_APPEND(tq->tq_task, tqe); \ 592 tqe->tqent_func = (func); \ 593 tqe->tqent_arg = (arg); \ 594 tq->tq_tasks++; \ 595 if (tq->tq_tasks - tq->tq_executed > tq->tq_maxtasks) \ 596 tq->tq_maxtasks = tq->tq_tasks - tq->tq_executed; \ 597 cv_signal(&tq->tq_dispatch_cv); \ 598 DTRACE_PROBE2(taskq__enqueue, taskq_t *, tq, taskq_ent_t *, tqe); \ 599 } 600 601 /* 602 * Do-nothing task which may be used to prepopulate thread caches. 603 */ 604 /*ARGSUSED*/ 605 void 606 nulltask(void *unused) 607 { 608 } 609 610 611 /*ARGSUSED*/ 612 static int 613 taskq_constructor(void *buf, void *cdrarg, int kmflags) 614 { 615 taskq_t *tq = buf; 616 617 bzero(tq, sizeof (taskq_t)); 618 619 mutex_init(&tq->tq_lock, NULL, MUTEX_DEFAULT, NULL); 620 rw_init(&tq->tq_threadlock, NULL, RW_DEFAULT, NULL); 621 cv_init(&tq->tq_dispatch_cv, NULL, CV_DEFAULT, NULL); 622 cv_init(&tq->tq_wait_cv, NULL, CV_DEFAULT, NULL); 623 624 tq->tq_task.tqent_next = &tq->tq_task; 625 tq->tq_task.tqent_prev = &tq->tq_task; 626 627 return (0); 628 } 629 630 /*ARGSUSED*/ 631 static void 632 taskq_destructor(void *buf, void *cdrarg) 633 { 634 taskq_t *tq = buf; 635 636 mutex_destroy(&tq->tq_lock); 637 rw_destroy(&tq->tq_threadlock); 638 cv_destroy(&tq->tq_dispatch_cv); 639 cv_destroy(&tq->tq_wait_cv); 640 } 641 642 /*ARGSUSED*/ 643 static int 644 taskq_ent_constructor(void *buf, void *cdrarg, int kmflags) 645 { 646 taskq_ent_t *tqe = buf; 647 648 tqe->tqent_thread = NULL; 649 cv_init(&tqe->tqent_cv, NULL, CV_DEFAULT, NULL); 650 651 return (0); 652 } 653 654 /*ARGSUSED*/ 655 static void 656 taskq_ent_destructor(void *buf, void *cdrarg) 657 { 658 taskq_ent_t *tqe = buf; 659 660 ASSERT(tqe->tqent_thread == NULL); 661 cv_destroy(&tqe->tqent_cv); 662 } 663 664 void 665 taskq_init(void) 666 { 667 taskq_ent_cache = kmem_cache_create("taskq_ent_cache", 668 sizeof (taskq_ent_t), 0, taskq_ent_constructor, 669 taskq_ent_destructor, NULL, NULL, NULL, 0); 670 taskq_cache = kmem_cache_create("taskq_cache", sizeof (taskq_t), 671 0, taskq_constructor, taskq_destructor, NULL, NULL, NULL, 0); 672 taskq_id_arena = vmem_create("taskq_id_arena", 673 (void *)1, INT32_MAX, 1, NULL, NULL, NULL, 0, 674 VM_SLEEP | VMC_IDENTIFIER); 675 } 676 677 /* 678 * Create global system dynamic task queue. 679 */ 680 void 681 system_taskq_init(void) 682 { 683 system_taskq = taskq_create_common("system_taskq", 0, 684 system_taskq_size * max_ncpus, minclsyspri, 4, 512, 685 TASKQ_DYNAMIC | TASKQ_PREPOPULATE); 686 } 687 688 /* 689 * taskq_ent_alloc() 690 * 691 * Allocates a new taskq_ent_t structure either from the free list or from the 692 * cache. Returns NULL if it can't be allocated. 693 * 694 * Assumes: tq->tq_lock is held. 695 */ 696 static taskq_ent_t * 697 taskq_ent_alloc(taskq_t *tq, int flags) 698 { 699 int kmflags = (flags & TQ_NOSLEEP) ? KM_NOSLEEP : KM_SLEEP; 700 701 taskq_ent_t *tqe; 702 703 ASSERT(MUTEX_HELD(&tq->tq_lock)); 704 705 /* 706 * TQ_NOALLOC allocations are allowed to use the freelist, even if 707 * we are below tq_minalloc. 708 */ 709 if ((tqe = tq->tq_freelist) != NULL && 710 ((flags & TQ_NOALLOC) || tq->tq_nalloc >= tq->tq_minalloc)) { 711 tq->tq_freelist = tqe->tqent_next; 712 } else { 713 if (flags & TQ_NOALLOC) 714 return (NULL); 715 716 mutex_exit(&tq->tq_lock); 717 if (tq->tq_nalloc >= tq->tq_maxalloc) { 718 if (kmflags & KM_NOSLEEP) { 719 mutex_enter(&tq->tq_lock); 720 return (NULL); 721 } 722 /* 723 * We don't want to exceed tq_maxalloc, but we can't 724 * wait for other tasks to complete (and thus free up 725 * task structures) without risking deadlock with 726 * the caller. So, we just delay for one second 727 * to throttle the allocation rate. 728 */ 729 delay(hz); 730 } 731 tqe = kmem_cache_alloc(taskq_ent_cache, kmflags); 732 mutex_enter(&tq->tq_lock); 733 if (tqe != NULL) 734 tq->tq_nalloc++; 735 } 736 return (tqe); 737 } 738 739 /* 740 * taskq_ent_free() 741 * 742 * Free taskq_ent_t structure by either putting it on the free list or freeing 743 * it to the cache. 744 * 745 * Assumes: tq->tq_lock is held. 746 */ 747 static void 748 taskq_ent_free(taskq_t *tq, taskq_ent_t *tqe) 749 { 750 ASSERT(MUTEX_HELD(&tq->tq_lock)); 751 752 if (tq->tq_nalloc <= tq->tq_minalloc) { 753 tqe->tqent_next = tq->tq_freelist; 754 tq->tq_freelist = tqe; 755 } else { 756 tq->tq_nalloc--; 757 mutex_exit(&tq->tq_lock); 758 kmem_cache_free(taskq_ent_cache, tqe); 759 mutex_enter(&tq->tq_lock); 760 } 761 } 762 763 /* 764 * Dispatch a task "func(arg)" to a free entry of bucket b. 765 * 766 * Assumes: no bucket locks is held. 767 * 768 * Returns: a pointer to an entry if dispatch was successful. 769 * NULL if there are no free entries or if the bucket is suspended. 770 */ 771 static taskq_ent_t * 772 taskq_bucket_dispatch(taskq_bucket_t *b, task_func_t func, void *arg) 773 { 774 taskq_ent_t *tqe; 775 776 ASSERT(MUTEX_NOT_HELD(&b->tqbucket_lock)); 777 ASSERT(func != NULL); 778 779 mutex_enter(&b->tqbucket_lock); 780 781 ASSERT(b->tqbucket_nfree != 0 || IS_EMPTY(b->tqbucket_freelist)); 782 ASSERT(b->tqbucket_nfree == 0 || !IS_EMPTY(b->tqbucket_freelist)); 783 784 /* 785 * Get en entry from the freelist if there is one. 786 * Schedule task into the entry. 787 */ 788 if ((b->tqbucket_nfree != 0) && 789 !(b->tqbucket_flags & TQBUCKET_SUSPEND)) { 790 tqe = b->tqbucket_freelist.tqent_prev; 791 792 ASSERT(tqe != &b->tqbucket_freelist); 793 ASSERT(tqe->tqent_thread != NULL); 794 795 tqe->tqent_prev->tqent_next = tqe->tqent_next; 796 tqe->tqent_next->tqent_prev = tqe->tqent_prev; 797 b->tqbucket_nalloc++; 798 b->tqbucket_nfree--; 799 tqe->tqent_func = func; 800 tqe->tqent_arg = arg; 801 TQ_STAT(b, tqs_hits); 802 cv_signal(&tqe->tqent_cv); 803 DTRACE_PROBE2(taskq__d__enqueue, taskq_bucket_t *, b, 804 taskq_ent_t *, tqe); 805 } else { 806 tqe = NULL; 807 TQ_STAT(b, tqs_misses); 808 } 809 mutex_exit(&b->tqbucket_lock); 810 return (tqe); 811 } 812 813 /* 814 * Dispatch a task. 815 * 816 * Assumes: func != NULL 817 * 818 * Returns: NULL if dispatch failed. 819 * non-NULL if task dispatched successfully. 820 * Actual return value is the pointer to taskq entry that was used to 821 * dispatch a task. This is useful for debugging. 822 */ 823 /* ARGSUSED */ 824 taskqid_t 825 taskq_dispatch(taskq_t *tq, task_func_t func, void *arg, uint_t flags) 826 { 827 taskq_bucket_t *bucket = NULL; /* Which bucket needs extension */ 828 taskq_ent_t *tqe = NULL; 829 taskq_ent_t *tqe1; 830 uint_t bsize; 831 832 ASSERT(tq != NULL); 833 ASSERT(func != NULL); 834 835 if (!(tq->tq_flags & TASKQ_DYNAMIC)) { 836 /* 837 * TQ_NOQUEUE flag can't be used with non-dynamic task queues. 838 */ 839 ASSERT(! (flags & TQ_NOQUEUE)); 840 /* 841 * Enqueue the task to the underlying queue. 842 */ 843 mutex_enter(&tq->tq_lock); 844 845 TASKQ_S_RANDOM_DISPATCH_FAILURE(tq, flags); 846 847 if ((tqe = taskq_ent_alloc(tq, flags)) == NULL) { 848 mutex_exit(&tq->tq_lock); 849 return (NULL); 850 } 851 TQ_ENQUEUE(tq, tqe, func, arg); 852 mutex_exit(&tq->tq_lock); 853 return ((taskqid_t)tqe); 854 } 855 856 /* 857 * Dynamic taskq dispatching. 858 */ 859 ASSERT(!(flags & TQ_NOALLOC)); 860 TASKQ_D_RANDOM_DISPATCH_FAILURE(tq, flags); 861 862 bsize = tq->tq_nbuckets; 863 864 if (bsize == 1) { 865 /* 866 * In a single-CPU case there is only one bucket, so get 867 * entry directly from there. 868 */ 869 if ((tqe = taskq_bucket_dispatch(tq->tq_buckets, func, arg)) 870 != NULL) 871 return ((taskqid_t)tqe); /* Fastpath */ 872 bucket = tq->tq_buckets; 873 } else { 874 int loopcount; 875 taskq_bucket_t *b; 876 uintptr_t h = ((uintptr_t)CPU + (uintptr_t)arg) >> 3; 877 878 h = TQ_HASH(h); 879 880 /* 881 * The 'bucket' points to the original bucket that we hit. If we 882 * can't allocate from it, we search other buckets, but only 883 * extend this one. 884 */ 885 b = &tq->tq_buckets[h & (bsize - 1)]; 886 ASSERT(b->tqbucket_taskq == tq); /* Sanity check */ 887 888 /* 889 * Do a quick check before grabbing the lock. If the bucket does 890 * not have free entries now, chances are very small that it 891 * will after we take the lock, so we just skip it. 892 */ 893 if (b->tqbucket_nfree != 0) { 894 if ((tqe = taskq_bucket_dispatch(b, func, arg)) != NULL) 895 return ((taskqid_t)tqe); /* Fastpath */ 896 } else { 897 TQ_STAT(b, tqs_misses); 898 } 899 900 bucket = b; 901 loopcount = MIN(taskq_search_depth, bsize); 902 /* 903 * If bucket dispatch failed, search loopcount number of buckets 904 * before we give up and fail. 905 */ 906 do { 907 b = &tq->tq_buckets[++h & (bsize - 1)]; 908 ASSERT(b->tqbucket_taskq == tq); /* Sanity check */ 909 loopcount--; 910 911 if (b->tqbucket_nfree != 0) { 912 tqe = taskq_bucket_dispatch(b, func, arg); 913 } else { 914 TQ_STAT(b, tqs_misses); 915 } 916 } while ((tqe == NULL) && (loopcount > 0)); 917 } 918 919 /* 920 * At this point we either scheduled a task and (tqe != NULL) or failed 921 * (tqe == NULL). Try to recover from fails. 922 */ 923 924 /* 925 * For KM_SLEEP dispatches, try to extend the bucket and retry dispatch. 926 */ 927 if ((tqe == NULL) && !(flags & TQ_NOSLEEP)) { 928 /* 929 * taskq_bucket_extend() may fail to do anything, but this is 930 * fine - we deal with it later. If the bucket was successfully 931 * extended, there is a good chance that taskq_bucket_dispatch() 932 * will get this new entry, unless someone is racing with us and 933 * stealing the new entry from under our nose. 934 * taskq_bucket_extend() may sleep. 935 */ 936 taskq_bucket_extend(bucket); 937 TQ_STAT(bucket, tqs_disptcreates); 938 if ((tqe = taskq_bucket_dispatch(bucket, func, arg)) != NULL) 939 return ((taskqid_t)tqe); 940 } 941 942 ASSERT(bucket != NULL); 943 /* 944 * Since there are not enough free entries in the bucket, extend it 945 * in the background using backing queue. 946 */ 947 mutex_enter(&tq->tq_lock); 948 if ((tqe1 = taskq_ent_alloc(tq, TQ_NOSLEEP)) != NULL) { 949 TQ_ENQUEUE(tq, tqe1, taskq_bucket_extend, 950 bucket); 951 } else { 952 TQ_STAT(bucket, tqs_nomem); 953 } 954 955 /* 956 * Dispatch failed and we can't find an entry to schedule a task. 957 * Revert to the backing queue unless TQ_NOQUEUE was asked. 958 */ 959 if ((tqe == NULL) && !(flags & TQ_NOQUEUE)) { 960 if ((tqe = taskq_ent_alloc(tq, flags)) != NULL) { 961 TQ_ENQUEUE(tq, tqe, func, arg); 962 } else { 963 TQ_STAT(bucket, tqs_nomem); 964 } 965 } 966 mutex_exit(&tq->tq_lock); 967 968 return ((taskqid_t)tqe); 969 } 970 971 /* 972 * Wait for all pending tasks to complete. 973 * Calling taskq_wait from a task will cause deadlock. 974 */ 975 void 976 taskq_wait(taskq_t *tq) 977 { 978 ASSERT(tq != curthread->t_taskq); 979 980 mutex_enter(&tq->tq_lock); 981 while (tq->tq_task.tqent_next != &tq->tq_task || tq->tq_active != 0) 982 cv_wait(&tq->tq_wait_cv, &tq->tq_lock); 983 mutex_exit(&tq->tq_lock); 984 985 if (tq->tq_flags & TASKQ_DYNAMIC) { 986 taskq_bucket_t *b = tq->tq_buckets; 987 int bid = 0; 988 for (; (b != NULL) && (bid < tq->tq_nbuckets); b++, bid++) { 989 mutex_enter(&b->tqbucket_lock); 990 while (b->tqbucket_nalloc > 0) 991 cv_wait(&b->tqbucket_cv, &b->tqbucket_lock); 992 mutex_exit(&b->tqbucket_lock); 993 } 994 } 995 } 996 997 /* 998 * Suspend execution of tasks. 999 * 1000 * Tasks in the queue part will be suspended immediately upon return from this 1001 * function. Pending tasks in the dynamic part will continue to execute, but all 1002 * new tasks will be suspended. 1003 */ 1004 void 1005 taskq_suspend(taskq_t *tq) 1006 { 1007 rw_enter(&tq->tq_threadlock, RW_WRITER); 1008 1009 if (tq->tq_flags & TASKQ_DYNAMIC) { 1010 taskq_bucket_t *b = tq->tq_buckets; 1011 int bid = 0; 1012 for (; (b != NULL) && (bid < tq->tq_nbuckets); b++, bid++) { 1013 mutex_enter(&b->tqbucket_lock); 1014 b->tqbucket_flags |= TQBUCKET_SUSPEND; 1015 mutex_exit(&b->tqbucket_lock); 1016 } 1017 } 1018 /* 1019 * Mark task queue as being suspended. Needed for taskq_suspended(). 1020 */ 1021 mutex_enter(&tq->tq_lock); 1022 ASSERT(!(tq->tq_flags & TASKQ_SUSPENDED)); 1023 tq->tq_flags |= TASKQ_SUSPENDED; 1024 mutex_exit(&tq->tq_lock); 1025 } 1026 1027 /* 1028 * returns: 1 if tq is suspended, 0 otherwise. 1029 */ 1030 int 1031 taskq_suspended(taskq_t *tq) 1032 { 1033 return ((tq->tq_flags & TASKQ_SUSPENDED) != 0); 1034 } 1035 1036 /* 1037 * Resume taskq execution. 1038 */ 1039 void 1040 taskq_resume(taskq_t *tq) 1041 { 1042 ASSERT(RW_WRITE_HELD(&tq->tq_threadlock)); 1043 1044 if (tq->tq_flags & TASKQ_DYNAMIC) { 1045 taskq_bucket_t *b = tq->tq_buckets; 1046 int bid = 0; 1047 for (; (b != NULL) && (bid < tq->tq_nbuckets); b++, bid++) { 1048 mutex_enter(&b->tqbucket_lock); 1049 b->tqbucket_flags &= ~TQBUCKET_SUSPEND; 1050 mutex_exit(&b->tqbucket_lock); 1051 } 1052 } 1053 mutex_enter(&tq->tq_lock); 1054 ASSERT(tq->tq_flags & TASKQ_SUSPENDED); 1055 tq->tq_flags &= ~TASKQ_SUSPENDED; 1056 mutex_exit(&tq->tq_lock); 1057 1058 rw_exit(&tq->tq_threadlock); 1059 } 1060 1061 int 1062 taskq_member(taskq_t *tq, kthread_t *thread) 1063 { 1064 return (thread->t_taskq == tq); 1065 } 1066 1067 /* 1068 * Worker thread for processing task queue. 1069 */ 1070 static void 1071 taskq_thread(void *arg) 1072 { 1073 taskq_t *tq = arg; 1074 taskq_ent_t *tqe; 1075 callb_cpr_t cprinfo; 1076 hrtime_t start, end; 1077 1078 if (tq->tq_flags & TASKQ_CPR_SAFE) { 1079 CALLB_CPR_INIT_SAFE(curthread, tq->tq_name); 1080 } else { 1081 CALLB_CPR_INIT(&cprinfo, &tq->tq_lock, callb_generic_cpr, 1082 tq->tq_name); 1083 } 1084 mutex_enter(&tq->tq_lock); 1085 while (tq->tq_flags & TASKQ_ACTIVE) { 1086 if ((tqe = tq->tq_task.tqent_next) == &tq->tq_task) { 1087 if (--tq->tq_active == 0) 1088 cv_broadcast(&tq->tq_wait_cv); 1089 if (tq->tq_flags & TASKQ_CPR_SAFE) { 1090 cv_wait(&tq->tq_dispatch_cv, &tq->tq_lock); 1091 } else { 1092 CALLB_CPR_SAFE_BEGIN(&cprinfo); 1093 cv_wait(&tq->tq_dispatch_cv, &tq->tq_lock); 1094 CALLB_CPR_SAFE_END(&cprinfo, &tq->tq_lock); 1095 } 1096 tq->tq_active++; 1097 continue; 1098 } 1099 tqe->tqent_prev->tqent_next = tqe->tqent_next; 1100 tqe->tqent_next->tqent_prev = tqe->tqent_prev; 1101 mutex_exit(&tq->tq_lock); 1102 1103 rw_enter(&tq->tq_threadlock, RW_READER); 1104 start = gethrtime(); 1105 DTRACE_PROBE2(taskq__exec__start, taskq_t *, tq, 1106 taskq_ent_t *, tqe); 1107 tqe->tqent_func(tqe->tqent_arg); 1108 DTRACE_PROBE2(taskq__exec__end, taskq_t *, tq, 1109 taskq_ent_t *, tqe); 1110 end = gethrtime(); 1111 rw_exit(&tq->tq_threadlock); 1112 1113 mutex_enter(&tq->tq_lock); 1114 tq->tq_totaltime += end - start; 1115 tq->tq_executed++; 1116 1117 taskq_ent_free(tq, tqe); 1118 } 1119 tq->tq_nthreads--; 1120 cv_broadcast(&tq->tq_wait_cv); 1121 ASSERT(!(tq->tq_flags & TASKQ_CPR_SAFE)); 1122 CALLB_CPR_EXIT(&cprinfo); 1123 thread_exit(); 1124 } 1125 1126 /* 1127 * Worker per-entry thread for dynamic dispatches. 1128 */ 1129 static void 1130 taskq_d_thread(taskq_ent_t *tqe) 1131 { 1132 taskq_bucket_t *bucket = tqe->tqent_bucket; 1133 taskq_t *tq = bucket->tqbucket_taskq; 1134 kmutex_t *lock = &bucket->tqbucket_lock; 1135 kcondvar_t *cv = &tqe->tqent_cv; 1136 callb_cpr_t cprinfo; 1137 clock_t w; 1138 1139 CALLB_CPR_INIT(&cprinfo, lock, callb_generic_cpr, tq->tq_name); 1140 1141 mutex_enter(lock); 1142 1143 for (;;) { 1144 /* 1145 * If a task is scheduled (func != NULL), execute it, otherwise 1146 * sleep, waiting for a job. 1147 */ 1148 if (tqe->tqent_func != NULL) { 1149 hrtime_t start; 1150 hrtime_t end; 1151 1152 ASSERT(bucket->tqbucket_nalloc > 0); 1153 1154 /* 1155 * It is possible to free the entry right away before 1156 * actually executing the task so that subsequent 1157 * dispatches may immediately reuse it. But this, 1158 * effectively, creates a two-length queue in the entry 1159 * and may lead to a deadlock if the execution of the 1160 * current task depends on the execution of the next 1161 * scheduled task. So, we keep the entry busy until the 1162 * task is processed. 1163 */ 1164 1165 mutex_exit(lock); 1166 start = gethrtime(); 1167 DTRACE_PROBE3(taskq__d__exec__start, taskq_t *, tq, 1168 taskq_bucket_t *, bucket, taskq_ent_t *, tqe); 1169 tqe->tqent_func(tqe->tqent_arg); 1170 DTRACE_PROBE3(taskq__d__exec__end, taskq_t *, tq, 1171 taskq_bucket_t *, bucket, taskq_ent_t *, tqe); 1172 end = gethrtime(); 1173 mutex_enter(lock); 1174 bucket->tqbucket_totaltime += end - start; 1175 1176 /* 1177 * Return the entry to the bucket free list. 1178 */ 1179 tqe->tqent_func = NULL; 1180 TQ_APPEND(bucket->tqbucket_freelist, tqe); 1181 bucket->tqbucket_nalloc--; 1182 bucket->tqbucket_nfree++; 1183 ASSERT(!IS_EMPTY(bucket->tqbucket_freelist)); 1184 /* 1185 * taskq_wait() waits for nalloc to drop to zero on 1186 * tqbucket_cv. 1187 */ 1188 cv_signal(&bucket->tqbucket_cv); 1189 } 1190 1191 /* 1192 * At this point the entry must be in the bucket free list - 1193 * either because it was there initially or because it just 1194 * finished executing a task and put itself on the free list. 1195 */ 1196 ASSERT(bucket->tqbucket_nfree > 0); 1197 /* 1198 * Go to sleep unless we are closing. 1199 * If a thread is sleeping too long, it dies. 1200 */ 1201 if (! (bucket->tqbucket_flags & TQBUCKET_CLOSE)) { 1202 CALLB_CPR_SAFE_BEGIN(&cprinfo); 1203 w = cv_timedwait(cv, lock, lbolt + 1204 taskq_thread_timeout * hz); 1205 CALLB_CPR_SAFE_END(&cprinfo, lock); 1206 } 1207 1208 /* 1209 * At this point we may be in two different states: 1210 * 1211 * (1) tqent_func is set which means that a new task is 1212 * dispatched and we need to execute it. 1213 * 1214 * (2) Thread is sleeping for too long or we are closing. In 1215 * both cases destroy the thread and the entry. 1216 */ 1217 1218 /* If func is NULL we should be on the freelist. */ 1219 ASSERT((tqe->tqent_func != NULL) || 1220 (bucket->tqbucket_nfree > 0)); 1221 /* If func is non-NULL we should be allocated */ 1222 ASSERT((tqe->tqent_func == NULL) || 1223 (bucket->tqbucket_nalloc > 0)); 1224 1225 /* Check freelist consistency */ 1226 ASSERT((bucket->tqbucket_nfree > 0) || 1227 IS_EMPTY(bucket->tqbucket_freelist)); 1228 ASSERT((bucket->tqbucket_nfree == 0) || 1229 !IS_EMPTY(bucket->tqbucket_freelist)); 1230 1231 if ((tqe->tqent_func == NULL) && 1232 ((w == -1) || (bucket->tqbucket_flags & TQBUCKET_CLOSE))) { 1233 /* 1234 * This thread is sleeping for too long or we are 1235 * closing - time to die. 1236 * Thread creation/destruction happens rarely, 1237 * so grabbing the lock is not a big performance issue. 1238 * The bucket lock is dropped by CALLB_CPR_EXIT(). 1239 */ 1240 1241 /* Remove the entry from the free list. */ 1242 tqe->tqent_prev->tqent_next = tqe->tqent_next; 1243 tqe->tqent_next->tqent_prev = tqe->tqent_prev; 1244 ASSERT(bucket->tqbucket_nfree > 0); 1245 bucket->tqbucket_nfree--; 1246 1247 TQ_STAT(bucket, tqs_tdeaths); 1248 cv_signal(&bucket->tqbucket_cv); 1249 tqe->tqent_thread = NULL; 1250 mutex_enter(&tq->tq_lock); 1251 tq->tq_tdeaths++; 1252 mutex_exit(&tq->tq_lock); 1253 CALLB_CPR_EXIT(&cprinfo); 1254 kmem_cache_free(taskq_ent_cache, tqe); 1255 thread_exit(); 1256 } 1257 } 1258 } 1259 1260 1261 /* 1262 * Taskq creation. May sleep for memory. 1263 * Always use automatically generated instances to avoid kstat name space 1264 * collisions. 1265 */ 1266 1267 taskq_t * 1268 taskq_create(const char *name, int nthreads, pri_t pri, int minalloc, 1269 int maxalloc, uint_t flags) 1270 { 1271 return taskq_create_common(name, 0, nthreads, pri, minalloc, 1272 maxalloc, flags | TASKQ_NOINSTANCE); 1273 } 1274 1275 /* 1276 * Create an instance of task queue. It is legal to create task queues with the 1277 * same name and different instances. 1278 * 1279 * taskq_create_instance is used by ddi_taskq_create() where it gets the 1280 * instance from ddi_get_instance(). In some cases the instance is not 1281 * initialized and is set to -1. This case is handled as if no instance was 1282 * passed at all. 1283 */ 1284 taskq_t * 1285 taskq_create_instance(const char *name, int instance, int nthreads, pri_t pri, 1286 int minalloc, int maxalloc, uint_t flags) 1287 { 1288 ASSERT((instance >= 0) || (instance == -1)); 1289 1290 if (instance < 0) { 1291 flags |= TASKQ_NOINSTANCE; 1292 } 1293 1294 return (taskq_create_common(name, instance, nthreads, 1295 pri, minalloc, maxalloc, flags)); 1296 } 1297 1298 static taskq_t * 1299 taskq_create_common(const char *name, int instance, int nthreads, pri_t pri, 1300 int minalloc, int maxalloc, uint_t flags) 1301 { 1302 taskq_t *tq = kmem_cache_alloc(taskq_cache, KM_SLEEP); 1303 uint_t ncpus = ((boot_max_ncpus == -1) ? max_ncpus : boot_max_ncpus); 1304 uint_t bsize; /* # of buckets - always power of 2 */ 1305 1306 /* 1307 * TASKQ_CPR_SAFE and TASKQ_DYNAMIC flags are mutually exclusive. 1308 */ 1309 ASSERT((flags & (TASKQ_DYNAMIC | TASKQ_CPR_SAFE)) != 1310 ((TASKQ_DYNAMIC | TASKQ_CPR_SAFE))); 1311 1312 ASSERT(tq->tq_buckets == NULL); 1313 1314 bsize = 1 << (highbit(ncpus) - 1); 1315 ASSERT(bsize >= 1); 1316 bsize = MIN(bsize, taskq_maxbuckets); 1317 1318 tq->tq_maxsize = nthreads; 1319 1320 /* For non-dynamic task queues use just one backup thread */ 1321 if (flags & TASKQ_DYNAMIC) 1322 nthreads = 1; 1323 1324 (void) strncpy(tq->tq_name, name, TASKQ_NAMELEN + 1); 1325 tq->tq_name[TASKQ_NAMELEN] = '\0'; 1326 /* Make sure the name conforms to the rules for C indentifiers */ 1327 strident_canon(tq->tq_name, TASKQ_NAMELEN); 1328 1329 tq->tq_flags = flags | TASKQ_ACTIVE; 1330 tq->tq_active = nthreads; 1331 tq->tq_instance = instance; 1332 tq->tq_nthreads = nthreads; 1333 tq->tq_minalloc = minalloc; 1334 tq->tq_maxalloc = maxalloc; 1335 tq->tq_nbuckets = bsize; 1336 tq->tq_pri = pri; 1337 1338 if (flags & TASKQ_PREPOPULATE) { 1339 mutex_enter(&tq->tq_lock); 1340 while (minalloc-- > 0) 1341 taskq_ent_free(tq, taskq_ent_alloc(tq, TQ_SLEEP)); 1342 mutex_exit(&tq->tq_lock); 1343 } 1344 1345 if (nthreads == 1) { 1346 tq->tq_thread = thread_create(NULL, 0, taskq_thread, tq, 1347 0, &p0, TS_RUN, pri); 1348 /* 1349 * No need to take thread_lock to change the field: no one can 1350 * reference it at this point. 1351 */ 1352 tq->tq_thread->t_taskq = tq; 1353 } else { 1354 kthread_t **tpp = kmem_alloc(sizeof (kthread_t *) * nthreads, 1355 KM_SLEEP); 1356 1357 tq->tq_threadlist = tpp; 1358 1359 mutex_enter(&tq->tq_lock); 1360 while (nthreads-- > 0) { 1361 *tpp = thread_create(NULL, 0, taskq_thread, tq, 1362 0, &p0, TS_RUN, pri); 1363 (*tpp)->t_taskq = tq; 1364 tpp++; 1365 } 1366 mutex_exit(&tq->tq_lock); 1367 } 1368 1369 if (flags & TASKQ_DYNAMIC) { 1370 taskq_bucket_t *bucket = kmem_zalloc(sizeof (taskq_bucket_t) * 1371 bsize, KM_SLEEP); 1372 int b_id; 1373 1374 tq->tq_buckets = bucket; 1375 1376 /* Initialize each bucket */ 1377 for (b_id = 0; b_id < bsize; b_id++, bucket++) { 1378 mutex_init(&bucket->tqbucket_lock, NULL, MUTEX_DEFAULT, 1379 NULL); 1380 cv_init(&bucket->tqbucket_cv, NULL, CV_DEFAULT, NULL); 1381 bucket->tqbucket_taskq = tq; 1382 bucket->tqbucket_freelist.tqent_next = 1383 bucket->tqbucket_freelist.tqent_prev = 1384 &bucket->tqbucket_freelist; 1385 if (flags & TASKQ_PREPOPULATE) 1386 taskq_bucket_extend(bucket); 1387 } 1388 } 1389 1390 /* 1391 * Install kstats. 1392 * We have two cases: 1393 * 1) Instance is provided to taskq_create_instance(). In this case it 1394 * should be >= 0 and we use it. 1395 * 1396 * 2) Instance is not provided and is automatically generated 1397 */ 1398 if (flags & TASKQ_NOINSTANCE) { 1399 instance = tq->tq_instance = 1400 (int)(uintptr_t)vmem_alloc(taskq_id_arena, 1, VM_SLEEP); 1401 } 1402 1403 if (flags & TASKQ_DYNAMIC) { 1404 if ((tq->tq_kstat = kstat_create("unix", instance, 1405 tq->tq_name, "taskq_d", KSTAT_TYPE_NAMED, 1406 sizeof (taskq_d_kstat) / sizeof (kstat_named_t), 1407 KSTAT_FLAG_VIRTUAL)) != NULL) { 1408 tq->tq_kstat->ks_lock = &taskq_d_kstat_lock; 1409 tq->tq_kstat->ks_data = &taskq_d_kstat; 1410 tq->tq_kstat->ks_update = taskq_d_kstat_update; 1411 tq->tq_kstat->ks_private = tq; 1412 kstat_install(tq->tq_kstat); 1413 } 1414 } else { 1415 if ((tq->tq_kstat = kstat_create("unix", instance, tq->tq_name, 1416 "taskq", KSTAT_TYPE_NAMED, 1417 sizeof (taskq_kstat) / sizeof (kstat_named_t), 1418 KSTAT_FLAG_VIRTUAL)) != NULL) { 1419 tq->tq_kstat->ks_lock = &taskq_kstat_lock; 1420 tq->tq_kstat->ks_data = &taskq_kstat; 1421 tq->tq_kstat->ks_update = taskq_kstat_update; 1422 tq->tq_kstat->ks_private = tq; 1423 kstat_install(tq->tq_kstat); 1424 } 1425 } 1426 1427 return (tq); 1428 } 1429 1430 /* 1431 * taskq_destroy(). 1432 * 1433 * Assumes: by the time taskq_destroy is called no one will use this task queue 1434 * in any way and no one will try to dispatch entries in it. 1435 */ 1436 void 1437 taskq_destroy(taskq_t *tq) 1438 { 1439 taskq_bucket_t *b = tq->tq_buckets; 1440 int bid = 0; 1441 1442 ASSERT(! (tq->tq_flags & TASKQ_CPR_SAFE)); 1443 1444 /* 1445 * Destroy kstats. 1446 */ 1447 if (tq->tq_kstat != NULL) { 1448 kstat_delete(tq->tq_kstat); 1449 tq->tq_kstat = NULL; 1450 } 1451 1452 /* 1453 * Destroy instance if needed. 1454 */ 1455 if (tq->tq_flags & TASKQ_NOINSTANCE) { 1456 vmem_free(taskq_id_arena, (void *)(uintptr_t)(tq->tq_instance), 1457 1); 1458 tq->tq_instance = 0; 1459 } 1460 1461 /* 1462 * Wait for any pending entries to complete. 1463 */ 1464 taskq_wait(tq); 1465 1466 mutex_enter(&tq->tq_lock); 1467 ASSERT((tq->tq_task.tqent_next == &tq->tq_task) && 1468 (tq->tq_active == 0)); 1469 1470 if ((tq->tq_nthreads > 1) && (tq->tq_threadlist != NULL)) 1471 kmem_free(tq->tq_threadlist, sizeof (kthread_t *) * 1472 tq->tq_nthreads); 1473 1474 tq->tq_flags &= ~TASKQ_ACTIVE; 1475 cv_broadcast(&tq->tq_dispatch_cv); 1476 while (tq->tq_nthreads != 0) 1477 cv_wait(&tq->tq_wait_cv, &tq->tq_lock); 1478 1479 tq->tq_minalloc = 0; 1480 while (tq->tq_nalloc != 0) 1481 taskq_ent_free(tq, taskq_ent_alloc(tq, TQ_SLEEP)); 1482 1483 mutex_exit(&tq->tq_lock); 1484 1485 /* 1486 * Mark each bucket as closing and wakeup all sleeping threads. 1487 */ 1488 for (; (b != NULL) && (bid < tq->tq_nbuckets); b++, bid++) { 1489 taskq_ent_t *tqe; 1490 1491 mutex_enter(&b->tqbucket_lock); 1492 1493 b->tqbucket_flags |= TQBUCKET_CLOSE; 1494 /* Wakeup all sleeping threads */ 1495 1496 for (tqe = b->tqbucket_freelist.tqent_next; 1497 tqe != &b->tqbucket_freelist; tqe = tqe->tqent_next) 1498 cv_signal(&tqe->tqent_cv); 1499 1500 ASSERT(b->tqbucket_nalloc == 0); 1501 1502 /* 1503 * At this point we waited for all pending jobs to complete (in 1504 * both the task queue and the bucket and no new jobs should 1505 * arrive. Wait for all threads to die. 1506 */ 1507 while (b->tqbucket_nfree > 0) 1508 cv_wait(&b->tqbucket_cv, &b->tqbucket_lock); 1509 mutex_exit(&b->tqbucket_lock); 1510 mutex_destroy(&b->tqbucket_lock); 1511 cv_destroy(&b->tqbucket_cv); 1512 } 1513 1514 if (tq->tq_buckets != NULL) { 1515 ASSERT(tq->tq_flags & TASKQ_DYNAMIC); 1516 kmem_free(tq->tq_buckets, 1517 sizeof (taskq_bucket_t) * tq->tq_nbuckets); 1518 1519 /* Cleanup fields before returning tq to the cache */ 1520 tq->tq_buckets = NULL; 1521 tq->tq_tcreates = 0; 1522 tq->tq_tdeaths = 0; 1523 } else { 1524 ASSERT(!(tq->tq_flags & TASKQ_DYNAMIC)); 1525 } 1526 1527 tq->tq_totaltime = 0; 1528 tq->tq_tasks = 0; 1529 tq->tq_maxtasks = 0; 1530 tq->tq_executed = 0; 1531 kmem_cache_free(taskq_cache, tq); 1532 } 1533 1534 /* 1535 * Extend a bucket with a new entry on the free list and attach a worker thread 1536 * to it. 1537 * 1538 * Argument: pointer to the bucket. 1539 * 1540 * This function may quietly fail. It is only used by taskq_dispatch() which 1541 * handles such failures properly. 1542 */ 1543 static void 1544 taskq_bucket_extend(void *arg) 1545 { 1546 taskq_ent_t *tqe; 1547 taskq_bucket_t *b = (taskq_bucket_t *)arg; 1548 taskq_t *tq = b->tqbucket_taskq; 1549 int nthreads; 1550 1551 if (! ENOUGH_MEMORY()) { 1552 TQ_STAT(b, tqs_nomem); 1553 return; 1554 } 1555 1556 mutex_enter(&tq->tq_lock); 1557 1558 /* 1559 * Observe global taskq limits on the number of threads. 1560 */ 1561 if (tq->tq_tcreates++ - tq->tq_tdeaths > tq->tq_maxsize) { 1562 tq->tq_tcreates--; 1563 mutex_exit(&tq->tq_lock); 1564 return; 1565 } 1566 mutex_exit(&tq->tq_lock); 1567 1568 tqe = kmem_cache_alloc(taskq_ent_cache, KM_NOSLEEP); 1569 1570 if (tqe == NULL) { 1571 mutex_enter(&tq->tq_lock); 1572 TQ_STAT(b, tqs_nomem); 1573 tq->tq_tcreates--; 1574 mutex_exit(&tq->tq_lock); 1575 return; 1576 } 1577 1578 ASSERT(tqe->tqent_thread == NULL); 1579 1580 tqe->tqent_bucket = b; 1581 1582 /* 1583 * Create a thread in a TS_STOPPED state first. If it is successfully 1584 * created, place the entry on the free list and start the thread. 1585 */ 1586 tqe->tqent_thread = thread_create(NULL, 0, taskq_d_thread, tqe, 1587 0, &p0, TS_STOPPED, tq->tq_pri); 1588 1589 /* 1590 * Once the entry is ready, link it to the the bucket free list. 1591 */ 1592 mutex_enter(&b->tqbucket_lock); 1593 tqe->tqent_func = NULL; 1594 TQ_APPEND(b->tqbucket_freelist, tqe); 1595 b->tqbucket_nfree++; 1596 TQ_STAT(b, tqs_tcreates); 1597 1598 #if TASKQ_STATISTIC 1599 nthreads = b->tqbucket_stat.tqs_tcreates - 1600 b->tqbucket_stat.tqs_tdeaths; 1601 b->tqbucket_stat.tqs_maxthreads = MAX(nthreads, 1602 b->tqbucket_stat.tqs_maxthreads); 1603 #endif 1604 1605 mutex_exit(&b->tqbucket_lock); 1606 /* 1607 * Start the stopped thread. 1608 */ 1609 thread_lock(tqe->tqent_thread); 1610 tqe->tqent_thread->t_taskq = tq; 1611 tqe->tqent_thread->t_schedflag |= TS_ALLSTART; 1612 setrun_locked(tqe->tqent_thread); 1613 thread_unlock(tqe->tqent_thread); 1614 } 1615 1616 static int 1617 taskq_kstat_update(kstat_t *ksp, int rw) 1618 { 1619 struct taskq_kstat *tqsp = &taskq_kstat; 1620 taskq_t *tq = ksp->ks_private; 1621 1622 if (rw == KSTAT_WRITE) 1623 return (EACCES); 1624 1625 tqsp->tq_tasks.value.ui64 = tq->tq_tasks; 1626 tqsp->tq_executed.value.ui64 = tq->tq_executed; 1627 tqsp->tq_maxtasks.value.ui64 = tq->tq_maxtasks; 1628 tqsp->tq_totaltime.value.ui64 = tq->tq_totaltime; 1629 tqsp->tq_nactive.value.ui64 = tq->tq_active; 1630 tqsp->tq_nalloc.value.ui64 = tq->tq_nalloc; 1631 tqsp->tq_pri.value.ui64 = tq->tq_pri; 1632 tqsp->tq_nthreads.value.ui64 = tq->tq_nthreads; 1633 return (0); 1634 } 1635 1636 static int 1637 taskq_d_kstat_update(kstat_t *ksp, int rw) 1638 { 1639 struct taskq_d_kstat *tqsp = &taskq_d_kstat; 1640 taskq_t *tq = ksp->ks_private; 1641 taskq_bucket_t *b = tq->tq_buckets; 1642 int bid = 0; 1643 1644 if (rw == KSTAT_WRITE) 1645 return (EACCES); 1646 1647 ASSERT(tq->tq_flags & TASKQ_DYNAMIC); 1648 1649 tqsp->tqd_btasks.value.ui64 = tq->tq_tasks; 1650 tqsp->tqd_bexecuted.value.ui64 = tq->tq_executed; 1651 tqsp->tqd_bmaxtasks.value.ui64 = tq->tq_maxtasks; 1652 tqsp->tqd_bnalloc.value.ui64 = tq->tq_nalloc; 1653 tqsp->tqd_bnactive.value.ui64 = tq->tq_active; 1654 tqsp->tqd_btotaltime.value.ui64 = tq->tq_totaltime; 1655 tqsp->tqd_pri.value.ui64 = tq->tq_pri; 1656 1657 tqsp->tqd_hits.value.ui64 = 0; 1658 tqsp->tqd_misses.value.ui64 = 0; 1659 tqsp->tqd_overflows.value.ui64 = 0; 1660 tqsp->tqd_tcreates.value.ui64 = 0; 1661 tqsp->tqd_tdeaths.value.ui64 = 0; 1662 tqsp->tqd_maxthreads.value.ui64 = 0; 1663 tqsp->tqd_nomem.value.ui64 = 0; 1664 tqsp->tqd_disptcreates.value.ui64 = 0; 1665 tqsp->tqd_totaltime.value.ui64 = 0; 1666 tqsp->tqd_nalloc.value.ui64 = 0; 1667 tqsp->tqd_nfree.value.ui64 = 0; 1668 1669 for (; (b != NULL) && (bid < tq->tq_nbuckets); b++, bid++) { 1670 tqsp->tqd_hits.value.ui64 += b->tqbucket_stat.tqs_hits; 1671 tqsp->tqd_misses.value.ui64 += b->tqbucket_stat.tqs_misses; 1672 tqsp->tqd_overflows.value.ui64 += b->tqbucket_stat.tqs_overflow; 1673 tqsp->tqd_tcreates.value.ui64 += b->tqbucket_stat.tqs_tcreates; 1674 tqsp->tqd_tdeaths.value.ui64 += b->tqbucket_stat.tqs_tdeaths; 1675 tqsp->tqd_maxthreads.value.ui64 += 1676 b->tqbucket_stat.tqs_maxthreads; 1677 tqsp->tqd_nomem.value.ui64 += b->tqbucket_stat.tqs_nomem; 1678 tqsp->tqd_disptcreates.value.ui64 += 1679 b->tqbucket_stat.tqs_disptcreates; 1680 tqsp->tqd_totaltime.value.ui64 += b->tqbucket_totaltime; 1681 tqsp->tqd_nalloc.value.ui64 += b->tqbucket_nalloc; 1682 tqsp->tqd_nfree.value.ui64 += b->tqbucket_nfree; 1683 } 1684 return (0); 1685 } 1686