2 * Copyright (C) 2007-2010 Lawrence Livermore National Security, LLC.
3 * Copyright (C) 2007 The Regents of the University of California.
4 * Produced at Lawrence Livermore National Laboratory (cf, DISCLAIMER).
5 * Written by Brian Behlendorf <behlendorf1@llnl.gov>.
8 * This file is part of the SPL, Solaris Porting Layer.
10 * The SPL is free software; you can redistribute it and/or modify it
11 * under the terms of the GNU General Public License as published by the
12 * Free Software Foundation; either version 2 of the License, or (at your
13 * option) any later version.
15 * The SPL is distributed in the hope that it will be useful, but WITHOUT
16 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
17 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
20 * You should have received a copy of the GNU General Public License along
21 * with the SPL. If not, see <http://www.gnu.org/licenses/>.
23 * Solaris Porting Layer (SPL) Task Queue Implementation.
26 #include <sys/timer.h>
27 #include <sys/taskq.h>
30 #include <sys/trace_spl.h>
31 #ifdef HAVE_CPU_HOTPLUG
32 #include <linux/cpuhotplug.h>
35 static int spl_taskq_thread_bind
= 0;
36 module_param(spl_taskq_thread_bind
, int, 0644);
37 MODULE_PARM_DESC(spl_taskq_thread_bind
, "Bind taskq thread to CPU by default");
39 static uint_t spl_taskq_thread_timeout_ms
= 10000;
41 module_param(spl_taskq_thread_timeout_ms
, uint
, 0644);
43 MODULE_PARM_DESC(spl_taskq_thread_timeout_ms
,
44 "Time to require a dynamic thread be idle before it gets cleaned up");
46 static int spl_taskq_thread_dynamic
= 1;
47 module_param(spl_taskq_thread_dynamic
, int, 0444);
48 MODULE_PARM_DESC(spl_taskq_thread_dynamic
, "Allow dynamic taskq threads");
50 static int spl_taskq_thread_priority
= 1;
51 module_param(spl_taskq_thread_priority
, int, 0644);
52 MODULE_PARM_DESC(spl_taskq_thread_priority
,
53 "Allow non-default priority for taskq threads");
55 static uint_t spl_taskq_thread_sequential
= 4;
57 module_param(spl_taskq_thread_sequential
, uint
, 0644);
59 MODULE_PARM_DESC(spl_taskq_thread_sequential
,
60 "Create new taskq threads after N sequential tasks");
63 * Global system-wide dynamic task queue available for all consumers. This
64 * taskq is not intended for long-running tasks; instead, a dedicated taskq
67 taskq_t
*system_taskq
;
68 EXPORT_SYMBOL(system_taskq
);
69 /* Global dynamic task queue for long delay */
70 taskq_t
*system_delay_taskq
;
71 EXPORT_SYMBOL(system_delay_taskq
);
73 /* Private dedicated taskq for creating new taskq threads on demand. */
74 static taskq_t
*dynamic_taskq
;
75 static taskq_thread_t
*taskq_thread_create(taskq_t
*);
77 #ifdef HAVE_CPU_HOTPLUG
78 /* Multi-callback id for cpu hotplugging. */
79 static int spl_taskq_cpuhp_state
;
82 /* List of all taskqs */
84 struct rw_semaphore tq_list_sem
;
85 static uint_t taskq_tsd
;
88 task_km_flags(uint_t flags
)
90 if (flags
& TQ_NOSLEEP
)
93 if (flags
& TQ_PUSHPAGE
)
100 * taskq_find_by_name - Find the largest instance number of a named taskq.
103 taskq_find_by_name(const char *name
)
105 struct list_head
*tql
= NULL
;
108 list_for_each_prev(tql
, &tq_list
) {
109 tq
= list_entry(tql
, taskq_t
, tq_taskqs
);
110 if (strcmp(name
, tq
->tq_name
) == 0)
111 return (tq
->tq_instance
);
117 * NOTE: Must be called with tq->tq_lock held, returns a list_t which
118 * is not attached to the free, work, or pending taskq lists.
121 task_alloc(taskq_t
*tq
, uint_t flags
, unsigned long *irqflags
)
128 /* Acquire taskq_ent_t's from free list if available */
129 if (!list_empty(&tq
->tq_free_list
) && !(flags
& TQ_NEW
)) {
130 t
= list_entry(tq
->tq_free_list
.next
, taskq_ent_t
, tqent_list
);
132 ASSERT(!(t
->tqent_flags
& TQENT_FLAG_PREALLOC
));
133 ASSERT(!(t
->tqent_flags
& TQENT_FLAG_CANCEL
));
134 ASSERT(!timer_pending(&t
->tqent_timer
));
136 list_del_init(&t
->tqent_list
);
140 /* Free list is empty and memory allocations are prohibited */
141 if (flags
& TQ_NOALLOC
)
144 /* Hit maximum taskq_ent_t pool size */
145 if (tq
->tq_nalloc
>= tq
->tq_maxalloc
) {
146 if (flags
& TQ_NOSLEEP
)
150 * Sleep periodically polling the free list for an available
151 * taskq_ent_t. Dispatching with TQ_SLEEP should always succeed
152 * but we cannot block forever waiting for an taskq_ent_t to
153 * show up in the free list, otherwise a deadlock can happen.
155 * Therefore, we need to allocate a new task even if the number
156 * of allocated tasks is above tq->tq_maxalloc, but we still
157 * end up delaying the task allocation by one second, thereby
158 * throttling the task dispatch rate.
160 spin_unlock_irqrestore(&tq
->tq_lock
, *irqflags
);
161 schedule_timeout(HZ
/ 100);
162 spin_lock_irqsave_nested(&tq
->tq_lock
, *irqflags
,
170 spin_unlock_irqrestore(&tq
->tq_lock
, *irqflags
);
171 t
= kmem_alloc(sizeof (taskq_ent_t
), task_km_flags(flags
));
172 spin_lock_irqsave_nested(&tq
->tq_lock
, *irqflags
, tq
->tq_lock_class
);
183 * NOTE: Must be called with tq->tq_lock held, expects the taskq_ent_t
184 * to already be removed from the free, work, or pending taskq lists.
187 task_free(taskq_t
*tq
, taskq_ent_t
*t
)
191 ASSERT(list_empty(&t
->tqent_list
));
192 ASSERT(!timer_pending(&t
->tqent_timer
));
194 kmem_free(t
, sizeof (taskq_ent_t
));
199 * NOTE: Must be called with tq->tq_lock held, either destroys the
200 * taskq_ent_t if too many exist or moves it to the free list for later use.
203 task_done(taskq_t
*tq
, taskq_ent_t
*t
)
208 /* Wake tasks blocked in taskq_wait_id() */
209 wake_up_all(&t
->tqent_waitq
);
211 list_del_init(&t
->tqent_list
);
213 if (tq
->tq_nalloc
<= tq
->tq_minalloc
) {
214 t
->tqent_id
= TASKQID_INVALID
;
215 t
->tqent_func
= NULL
;
219 list_add_tail(&t
->tqent_list
, &tq
->tq_free_list
);
226 * When a delayed task timer expires remove it from the delay list and
227 * add it to the priority list in order for immediate processing.
230 task_expire_impl(taskq_ent_t
*t
)
233 taskq_t
*tq
= t
->tqent_taskq
;
234 struct list_head
*l
= NULL
;
237 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
, tq
->tq_lock_class
);
239 if (t
->tqent_flags
& TQENT_FLAG_CANCEL
) {
240 ASSERT(list_empty(&t
->tqent_list
));
241 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
245 t
->tqent_birth
= jiffies
;
246 DTRACE_PROBE1(taskq_ent__birth
, taskq_ent_t
*, t
);
249 * The priority list must be maintained in strict task id order
250 * from lowest to highest for lowest_id to be easily calculable.
252 list_del(&t
->tqent_list
);
253 list_for_each_prev(l
, &tq
->tq_prio_list
) {
254 w
= list_entry(l
, taskq_ent_t
, tqent_list
);
255 if (w
->tqent_id
< t
->tqent_id
) {
256 list_add(&t
->tqent_list
, l
);
260 if (l
== &tq
->tq_prio_list
)
261 list_add(&t
->tqent_list
, &tq
->tq_prio_list
);
263 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
265 wake_up(&tq
->tq_work_waitq
);
269 task_expire(spl_timer_list_t tl
)
271 struct timer_list
*tmr
= (struct timer_list
*)tl
;
272 taskq_ent_t
*t
= from_timer(t
, tmr
, tqent_timer
);
277 * Returns the lowest incomplete taskqid_t. The taskqid_t may
278 * be queued on the pending list, on the priority list, on the
279 * delay list, or on the work list currently being handled, but
280 * it is not 100% complete yet.
283 taskq_lowest_id(taskq_t
*tq
)
285 taskqid_t lowest_id
= tq
->tq_next_id
;
289 if (!list_empty(&tq
->tq_pend_list
)) {
290 t
= list_entry(tq
->tq_pend_list
.next
, taskq_ent_t
, tqent_list
);
291 lowest_id
= MIN(lowest_id
, t
->tqent_id
);
294 if (!list_empty(&tq
->tq_prio_list
)) {
295 t
= list_entry(tq
->tq_prio_list
.next
, taskq_ent_t
, tqent_list
);
296 lowest_id
= MIN(lowest_id
, t
->tqent_id
);
299 if (!list_empty(&tq
->tq_delay_list
)) {
300 t
= list_entry(tq
->tq_delay_list
.next
, taskq_ent_t
, tqent_list
);
301 lowest_id
= MIN(lowest_id
, t
->tqent_id
);
304 if (!list_empty(&tq
->tq_active_list
)) {
305 tqt
= list_entry(tq
->tq_active_list
.next
, taskq_thread_t
,
307 ASSERT(tqt
->tqt_id
!= TASKQID_INVALID
);
308 lowest_id
= MIN(lowest_id
, tqt
->tqt_id
);
315 * Insert a task into a list keeping the list sorted by increasing taskqid.
318 taskq_insert_in_order(taskq_t
*tq
, taskq_thread_t
*tqt
)
321 struct list_head
*l
= NULL
;
326 list_for_each_prev(l
, &tq
->tq_active_list
) {
327 w
= list_entry(l
, taskq_thread_t
, tqt_active_list
);
328 if (w
->tqt_id
< tqt
->tqt_id
) {
329 list_add(&tqt
->tqt_active_list
, l
);
333 if (l
== &tq
->tq_active_list
)
334 list_add(&tqt
->tqt_active_list
, &tq
->tq_active_list
);
338 * Find and return a task from the given list if it exists. The list
339 * must be in lowest to highest task id order.
342 taskq_find_list(taskq_t
*tq
, struct list_head
*lh
, taskqid_t id
)
344 struct list_head
*l
= NULL
;
347 list_for_each(l
, lh
) {
348 t
= list_entry(l
, taskq_ent_t
, tqent_list
);
350 if (t
->tqent_id
== id
)
353 if (t
->tqent_id
> id
)
361 * Find an already dispatched task given the task id regardless of what
362 * state it is in. If a task is still pending it will be returned.
363 * If a task is executing, then -EBUSY will be returned instead.
364 * If the task has already been run then NULL is returned.
367 taskq_find(taskq_t
*tq
, taskqid_t id
)
370 struct list_head
*l
= NULL
;
373 t
= taskq_find_list(tq
, &tq
->tq_delay_list
, id
);
377 t
= taskq_find_list(tq
, &tq
->tq_prio_list
, id
);
381 t
= taskq_find_list(tq
, &tq
->tq_pend_list
, id
);
385 list_for_each(l
, &tq
->tq_active_list
) {
386 tqt
= list_entry(l
, taskq_thread_t
, tqt_active_list
);
387 if (tqt
->tqt_id
== id
) {
389 * Instead of returning tqt_task, we just return a non
390 * NULL value to prevent misuse, since tqt_task only
391 * has two valid fields.
393 return (ERR_PTR(-EBUSY
));
401 * Theory for the taskq_wait_id(), taskq_wait_outstanding(), and
402 * taskq_wait() functions below.
404 * Taskq waiting is accomplished by tracking the lowest outstanding task
405 * id and the next available task id. As tasks are dispatched they are
406 * added to the tail of the pending, priority, or delay lists. As worker
407 * threads become available the tasks are removed from the heads of these
408 * lists and linked to the worker threads. This ensures the lists are
409 * kept sorted by lowest to highest task id.
411 * Therefore the lowest outstanding task id can be quickly determined by
412 * checking the head item from all of these lists. This value is stored
413 * with the taskq as the lowest id. It only needs to be recalculated when
414 * either the task with the current lowest id completes or is canceled.
416 * By blocking until the lowest task id exceeds the passed task id the
417 * taskq_wait_outstanding() function can be easily implemented. Similarly,
418 * by blocking until the lowest task id matches the next task id taskq_wait()
419 * can be implemented.
421 * Callers should be aware that when there are multiple worked threads it
422 * is possible for larger task ids to complete before smaller ones. Also
423 * when the taskq contains delay tasks with small task ids callers may
424 * block for a considerable length of time waiting for them to expire and
428 taskq_wait_id_check(taskq_t
*tq
, taskqid_t id
)
433 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
, tq
->tq_lock_class
);
434 rc
= (taskq_find(tq
, id
) == NULL
);
435 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
441 * The taskq_wait_id() function blocks until the passed task id completes.
442 * This does not guarantee that all lower task ids have completed.
445 taskq_wait_id(taskq_t
*tq
, taskqid_t id
)
447 wait_event(tq
->tq_wait_waitq
, taskq_wait_id_check(tq
, id
));
449 EXPORT_SYMBOL(taskq_wait_id
);
452 taskq_wait_outstanding_check(taskq_t
*tq
, taskqid_t id
)
457 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
, tq
->tq_lock_class
);
458 rc
= (id
< tq
->tq_lowest_id
);
459 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
465 * The taskq_wait_outstanding() function will block until all tasks with a
466 * lower taskqid than the passed 'id' have been completed. Note that all
467 * task id's are assigned monotonically at dispatch time. Zero may be
468 * passed for the id to indicate all tasks dispatch up to this point,
469 * but not after, should be waited for.
472 taskq_wait_outstanding(taskq_t
*tq
, taskqid_t id
)
474 id
= id
? id
: tq
->tq_next_id
- 1;
475 wait_event(tq
->tq_wait_waitq
, taskq_wait_outstanding_check(tq
, id
));
477 EXPORT_SYMBOL(taskq_wait_outstanding
);
480 taskq_wait_check(taskq_t
*tq
)
485 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
, tq
->tq_lock_class
);
486 rc
= (tq
->tq_lowest_id
== tq
->tq_next_id
);
487 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
493 * The taskq_wait() function will block until the taskq is empty.
494 * This means that if a taskq re-dispatches work to itself taskq_wait()
495 * callers will block indefinitely.
498 taskq_wait(taskq_t
*tq
)
500 wait_event(tq
->tq_wait_waitq
, taskq_wait_check(tq
));
502 EXPORT_SYMBOL(taskq_wait
);
505 taskq_member(taskq_t
*tq
, kthread_t
*t
)
507 return (tq
== (taskq_t
*)tsd_get_by_thread(taskq_tsd
, t
));
509 EXPORT_SYMBOL(taskq_member
);
512 taskq_of_curthread(void)
514 return (tsd_get(taskq_tsd
));
516 EXPORT_SYMBOL(taskq_of_curthread
);
519 * Cancel an already dispatched task given the task id. Still pending tasks
520 * will be immediately canceled, and if the task is active the function will
521 * block until it completes. Preallocated tasks which are canceled must be
522 * freed by the caller.
525 taskq_cancel_id(taskq_t
*tq
, taskqid_t id
)
533 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
, tq
->tq_lock_class
);
534 t
= taskq_find(tq
, id
);
535 if (t
&& t
!= ERR_PTR(-EBUSY
)) {
536 list_del_init(&t
->tqent_list
);
537 t
->tqent_flags
|= TQENT_FLAG_CANCEL
;
540 * When canceling the lowest outstanding task id we
541 * must recalculate the new lowest outstanding id.
543 if (tq
->tq_lowest_id
== t
->tqent_id
) {
544 tq
->tq_lowest_id
= taskq_lowest_id(tq
);
545 ASSERT3S(tq
->tq_lowest_id
, >, t
->tqent_id
);
549 * The task_expire() function takes the tq->tq_lock so drop
550 * drop the lock before synchronously cancelling the timer.
552 if (timer_pending(&t
->tqent_timer
)) {
553 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
554 del_timer_sync(&t
->tqent_timer
);
555 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
,
559 if (!(t
->tqent_flags
& TQENT_FLAG_PREALLOC
))
564 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
566 if (t
== ERR_PTR(-EBUSY
)) {
567 taskq_wait_id(tq
, id
);
573 EXPORT_SYMBOL(taskq_cancel_id
);
575 static int taskq_thread_spawn(taskq_t
*tq
);
578 taskq_dispatch(taskq_t
*tq
, task_func_t func
, void *arg
, uint_t flags
)
581 taskqid_t rc
= TASKQID_INVALID
;
582 unsigned long irqflags
;
587 spin_lock_irqsave_nested(&tq
->tq_lock
, irqflags
, tq
->tq_lock_class
);
589 /* Taskq being destroyed and all tasks drained */
590 if (!(tq
->tq_flags
& TASKQ_ACTIVE
))
593 /* Do not queue the task unless there is idle thread for it */
594 ASSERT(tq
->tq_nactive
<= tq
->tq_nthreads
);
595 if ((flags
& TQ_NOQUEUE
) && (tq
->tq_nactive
== tq
->tq_nthreads
)) {
596 /* Dynamic taskq may be able to spawn another thread */
597 if (!(tq
->tq_flags
& TASKQ_DYNAMIC
) ||
598 taskq_thread_spawn(tq
) == 0)
602 if ((t
= task_alloc(tq
, flags
, &irqflags
)) == NULL
)
605 spin_lock(&t
->tqent_lock
);
607 /* Queue to the front of the list to enforce TQ_NOQUEUE semantics */
608 if (flags
& TQ_NOQUEUE
)
609 list_add(&t
->tqent_list
, &tq
->tq_prio_list
);
610 /* Queue to the priority list instead of the pending list */
611 else if (flags
& TQ_FRONT
)
612 list_add_tail(&t
->tqent_list
, &tq
->tq_prio_list
);
614 list_add_tail(&t
->tqent_list
, &tq
->tq_pend_list
);
616 t
->tqent_id
= rc
= tq
->tq_next_id
;
618 t
->tqent_func
= func
;
621 t
->tqent_timer
.function
= NULL
;
622 t
->tqent_timer
.expires
= 0;
624 t
->tqent_birth
= jiffies
;
625 DTRACE_PROBE1(taskq_ent__birth
, taskq_ent_t
*, t
);
627 ASSERT(!(t
->tqent_flags
& TQENT_FLAG_PREALLOC
));
629 spin_unlock(&t
->tqent_lock
);
631 wake_up(&tq
->tq_work_waitq
);
633 /* Spawn additional taskq threads if required. */
634 if (!(flags
& TQ_NOQUEUE
) && tq
->tq_nactive
== tq
->tq_nthreads
)
635 (void) taskq_thread_spawn(tq
);
637 spin_unlock_irqrestore(&tq
->tq_lock
, irqflags
);
640 EXPORT_SYMBOL(taskq_dispatch
);
643 taskq_dispatch_delay(taskq_t
*tq
, task_func_t func
, void *arg
,
644 uint_t flags
, clock_t expire_time
)
646 taskqid_t rc
= TASKQID_INVALID
;
648 unsigned long irqflags
;
653 spin_lock_irqsave_nested(&tq
->tq_lock
, irqflags
, tq
->tq_lock_class
);
655 /* Taskq being destroyed and all tasks drained */
656 if (!(tq
->tq_flags
& TASKQ_ACTIVE
))
659 if ((t
= task_alloc(tq
, flags
, &irqflags
)) == NULL
)
662 spin_lock(&t
->tqent_lock
);
664 /* Queue to the delay list for subsequent execution */
665 list_add_tail(&t
->tqent_list
, &tq
->tq_delay_list
);
667 t
->tqent_id
= rc
= tq
->tq_next_id
;
669 t
->tqent_func
= func
;
672 t
->tqent_timer
.function
= task_expire
;
673 t
->tqent_timer
.expires
= (unsigned long)expire_time
;
674 add_timer(&t
->tqent_timer
);
676 ASSERT(!(t
->tqent_flags
& TQENT_FLAG_PREALLOC
));
678 spin_unlock(&t
->tqent_lock
);
680 /* Spawn additional taskq threads if required. */
681 if (tq
->tq_nactive
== tq
->tq_nthreads
)
682 (void) taskq_thread_spawn(tq
);
683 spin_unlock_irqrestore(&tq
->tq_lock
, irqflags
);
686 EXPORT_SYMBOL(taskq_dispatch_delay
);
689 taskq_dispatch_ent(taskq_t
*tq
, task_func_t func
, void *arg
, uint_t flags
,
692 unsigned long irqflags
;
696 spin_lock_irqsave_nested(&tq
->tq_lock
, irqflags
,
699 /* Taskq being destroyed and all tasks drained */
700 if (!(tq
->tq_flags
& TASKQ_ACTIVE
)) {
701 t
->tqent_id
= TASKQID_INVALID
;
705 if ((flags
& TQ_NOQUEUE
) && (tq
->tq_nactive
== tq
->tq_nthreads
)) {
706 /* Dynamic taskq may be able to spawn another thread */
707 if (!(tq
->tq_flags
& TASKQ_DYNAMIC
) ||
708 taskq_thread_spawn(tq
) == 0)
713 spin_lock(&t
->tqent_lock
);
716 * Make sure the entry is not on some other taskq; it is important to
717 * ASSERT() under lock
719 ASSERT(taskq_empty_ent(t
));
722 * Mark it as a prealloc'd task. This is important
723 * to ensure that we don't free it later.
725 t
->tqent_flags
|= TQENT_FLAG_PREALLOC
;
727 /* Queue to the priority list instead of the pending list */
728 if (flags
& TQ_FRONT
)
729 list_add_tail(&t
->tqent_list
, &tq
->tq_prio_list
);
731 list_add_tail(&t
->tqent_list
, &tq
->tq_pend_list
);
733 t
->tqent_id
= tq
->tq_next_id
;
735 t
->tqent_func
= func
;
739 t
->tqent_birth
= jiffies
;
740 DTRACE_PROBE1(taskq_ent__birth
, taskq_ent_t
*, t
);
742 spin_unlock(&t
->tqent_lock
);
744 wake_up(&tq
->tq_work_waitq
);
746 /* Spawn additional taskq threads if required. */
747 if (tq
->tq_nactive
== tq
->tq_nthreads
)
748 (void) taskq_thread_spawn(tq
);
750 spin_unlock_irqrestore(&tq
->tq_lock
, irqflags
);
752 EXPORT_SYMBOL(taskq_dispatch_ent
);
755 taskq_empty_ent(taskq_ent_t
*t
)
757 return (list_empty(&t
->tqent_list
));
759 EXPORT_SYMBOL(taskq_empty_ent
);
762 taskq_init_ent(taskq_ent_t
*t
)
764 spin_lock_init(&t
->tqent_lock
);
765 init_waitqueue_head(&t
->tqent_waitq
);
766 timer_setup(&t
->tqent_timer
, NULL
, 0);
767 INIT_LIST_HEAD(&t
->tqent_list
);
769 t
->tqent_func
= NULL
;
772 t
->tqent_taskq
= NULL
;
774 EXPORT_SYMBOL(taskq_init_ent
);
777 * Return the next pending task, preference is given to tasks on the
778 * priority list which were dispatched with TQ_FRONT.
781 taskq_next_ent(taskq_t
*tq
)
783 struct list_head
*list
;
785 if (!list_empty(&tq
->tq_prio_list
))
786 list
= &tq
->tq_prio_list
;
787 else if (!list_empty(&tq
->tq_pend_list
))
788 list
= &tq
->tq_pend_list
;
792 return (list_entry(list
->next
, taskq_ent_t
, tqent_list
));
796 * Spawns a new thread for the specified taskq.
799 taskq_thread_spawn_task(void *arg
)
801 taskq_t
*tq
= (taskq_t
*)arg
;
804 if (taskq_thread_create(tq
) == NULL
) {
805 /* restore spawning count if failed */
806 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
,
809 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
814 * Spawn addition threads for dynamic taskqs (TASKQ_DYNAMIC) the current
815 * number of threads is insufficient to handle the pending tasks. These
816 * new threads must be created by the dedicated dynamic_taskq to avoid
817 * deadlocks between thread creation and memory reclaim. The system_taskq
818 * which is also a dynamic taskq cannot be safely used for this.
821 taskq_thread_spawn(taskq_t
*tq
)
825 if (!(tq
->tq_flags
& TASKQ_DYNAMIC
))
828 if ((tq
->tq_nthreads
+ tq
->tq_nspawn
< tq
->tq_maxthreads
) &&
829 (tq
->tq_flags
& TASKQ_ACTIVE
)) {
830 spawning
= (++tq
->tq_nspawn
);
831 taskq_dispatch(dynamic_taskq
, taskq_thread_spawn_task
,
839 * Threads in a dynamic taskq should only exit once it has been completely
840 * drained and no other threads are actively servicing tasks. This prevents
841 * threads from being created and destroyed more than is required.
843 * The first thread is the thread list is treated as the primary thread.
844 * There is nothing special about the primary thread but in order to avoid
845 * all the taskq pids from changing we opt to make it long running.
848 taskq_thread_should_stop(taskq_t
*tq
, taskq_thread_t
*tqt
)
850 if (!(tq
->tq_flags
& TASKQ_DYNAMIC
))
853 if (list_first_entry(&(tq
->tq_thread_list
), taskq_thread_t
,
854 tqt_thread_list
) == tqt
)
858 ((tq
->tq_nspawn
== 0) && /* No threads are being spawned */
859 (tq
->tq_nactive
== 0) && /* No threads are handling tasks */
860 (tq
->tq_nthreads
> 1) && /* More than 1 thread is running */
861 (!taskq_next_ent(tq
)) && /* There are no pending tasks */
862 (spl_taskq_thread_dynamic
)); /* Dynamic taskqs are allowed */
865 * If we would have said stop before, let's instead wait a bit, maybe
866 * we'll see more work come our way soon...
869 /* if it's 0, we want the old behavior. */
870 /* if the taskq is being torn down, we also want to go away. */
871 if (spl_taskq_thread_timeout_ms
== 0 ||
872 !(tq
->tq_flags
& TASKQ_ACTIVE
))
874 unsigned long lasttime
= tq
->lastshouldstop
;
876 if (time_after(jiffies
, lasttime
+
877 msecs_to_jiffies(spl_taskq_thread_timeout_ms
)))
882 tq
->lastshouldstop
= jiffies
;
885 tq
->lastshouldstop
= 0;
891 taskq_thread(void *args
)
893 DECLARE_WAITQUEUE(wait
, current
);
895 taskq_thread_t
*tqt
= args
;
900 taskq_ent_t dup_task
= {};
905 current
->flags
|= PF_NOFREEZE
;
907 (void) spl_fstrans_mark();
909 sigfillset(&blocked
);
910 sigprocmask(SIG_BLOCK
, &blocked
, NULL
);
911 flush_signals(current
);
913 tsd_set(taskq_tsd
, tq
);
914 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
, tq
->tq_lock_class
);
916 * If we are dynamically spawned, decrease spawning count. Note that
917 * we could be created during taskq_create, in which case we shouldn't
918 * do the decrement. But it's fine because taskq_create will reset
921 if (tq
->tq_flags
& TASKQ_DYNAMIC
)
924 /* Immediately exit if more threads than allowed were created. */
925 if (tq
->tq_nthreads
>= tq
->tq_maxthreads
)
929 list_add_tail(&tqt
->tqt_thread_list
, &tq
->tq_thread_list
);
930 wake_up(&tq
->tq_wait_waitq
);
931 set_current_state(TASK_INTERRUPTIBLE
);
933 while (!kthread_should_stop()) {
935 if (list_empty(&tq
->tq_pend_list
) &&
936 list_empty(&tq
->tq_prio_list
)) {
938 if (taskq_thread_should_stop(tq
, tqt
)) {
939 wake_up_all(&tq
->tq_wait_waitq
);
943 add_wait_queue_exclusive(&tq
->tq_work_waitq
, &wait
);
944 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
949 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
,
951 remove_wait_queue(&tq
->tq_work_waitq
, &wait
);
953 __set_current_state(TASK_RUNNING
);
956 if ((t
= taskq_next_ent(tq
)) != NULL
) {
957 list_del_init(&t
->tqent_list
);
960 * A TQENT_FLAG_PREALLOC task may be reused or freed
961 * during the task function call. Store tqent_id and
964 * Also use an on stack taskq_ent_t for tqt_task
965 * assignment in this case; we want to make sure
966 * to duplicate all fields, so the values are
967 * correct when it's accessed via DTRACE_PROBE*.
969 tqt
->tqt_id
= t
->tqent_id
;
970 tqt
->tqt_flags
= t
->tqent_flags
;
972 if (t
->tqent_flags
& TQENT_FLAG_PREALLOC
) {
978 taskq_insert_in_order(tq
, tqt
);
980 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
982 DTRACE_PROBE1(taskq_ent__start
, taskq_ent_t
*, t
);
984 /* Perform the requested task */
985 t
->tqent_func(t
->tqent_arg
);
987 DTRACE_PROBE1(taskq_ent__finish
, taskq_ent_t
*, t
);
989 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
,
992 list_del_init(&tqt
->tqt_active_list
);
993 tqt
->tqt_task
= NULL
;
995 /* For prealloc'd tasks, we don't free anything. */
996 if (!(tqt
->tqt_flags
& TQENT_FLAG_PREALLOC
))
1000 * When the current lowest outstanding taskqid is
1001 * done calculate the new lowest outstanding id
1003 if (tq
->tq_lowest_id
== tqt
->tqt_id
) {
1004 tq
->tq_lowest_id
= taskq_lowest_id(tq
);
1005 ASSERT3S(tq
->tq_lowest_id
, >, tqt
->tqt_id
);
1008 /* Spawn additional taskq threads if required. */
1009 if ((++seq_tasks
) > spl_taskq_thread_sequential
&&
1010 taskq_thread_spawn(tq
))
1013 tqt
->tqt_id
= TASKQID_INVALID
;
1015 wake_up_all(&tq
->tq_wait_waitq
);
1017 if (taskq_thread_should_stop(tq
, tqt
))
1021 set_current_state(TASK_INTERRUPTIBLE
);
1025 __set_current_state(TASK_RUNNING
);
1027 list_del_init(&tqt
->tqt_thread_list
);
1029 kmem_free(tqt
, sizeof (taskq_thread_t
));
1030 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
1032 tsd_set(taskq_tsd
, NULL
);
1038 static taskq_thread_t
*
1039 taskq_thread_create(taskq_t
*tq
)
1041 static int last_used_cpu
= 0;
1042 taskq_thread_t
*tqt
;
1044 tqt
= kmem_alloc(sizeof (*tqt
), KM_PUSHPAGE
);
1045 INIT_LIST_HEAD(&tqt
->tqt_thread_list
);
1046 INIT_LIST_HEAD(&tqt
->tqt_active_list
);
1048 tqt
->tqt_id
= TASKQID_INVALID
;
1050 tqt
->tqt_thread
= spl_kthread_create(taskq_thread
, tqt
,
1052 if (tqt
->tqt_thread
== NULL
) {
1053 kmem_free(tqt
, sizeof (taskq_thread_t
));
1057 if (spl_taskq_thread_bind
) {
1058 last_used_cpu
= (last_used_cpu
+ 1) % num_online_cpus();
1059 kthread_bind(tqt
->tqt_thread
, last_used_cpu
);
1062 if (spl_taskq_thread_priority
)
1063 set_user_nice(tqt
->tqt_thread
, PRIO_TO_NICE(tq
->tq_pri
));
1065 wake_up_process(tqt
->tqt_thread
);
1071 taskq_create(const char *name
, int threads_arg
, pri_t pri
,
1072 int minalloc
, int maxalloc
, uint_t flags
)
1075 taskq_thread_t
*tqt
;
1076 int count
= 0, rc
= 0, i
;
1077 unsigned long irqflags
;
1078 int nthreads
= threads_arg
;
1080 ASSERT(name
!= NULL
);
1081 ASSERT(minalloc
>= 0);
1082 ASSERT(!(flags
& (TASKQ_CPR_SAFE
))); /* Unsupported */
1084 /* Scale the number of threads using nthreads as a percentage */
1085 if (flags
& TASKQ_THREADS_CPU_PCT
) {
1086 ASSERT(nthreads
<= 100);
1087 ASSERT(nthreads
>= 0);
1088 nthreads
= MIN(threads_arg
, 100);
1089 nthreads
= MAX(nthreads
, 0);
1090 nthreads
= MAX((num_online_cpus() * nthreads
) /100, 1);
1093 tq
= kmem_alloc(sizeof (*tq
), KM_PUSHPAGE
);
1097 tq
->tq_hp_support
= B_FALSE
;
1098 #ifdef HAVE_CPU_HOTPLUG
1099 if (flags
& TASKQ_THREADS_CPU_PCT
) {
1100 tq
->tq_hp_support
= B_TRUE
;
1101 if (cpuhp_state_add_instance_nocalls(spl_taskq_cpuhp_state
,
1102 &tq
->tq_hp_cb_node
) != 0) {
1103 kmem_free(tq
, sizeof (*tq
));
1109 spin_lock_init(&tq
->tq_lock
);
1110 INIT_LIST_HEAD(&tq
->tq_thread_list
);
1111 INIT_LIST_HEAD(&tq
->tq_active_list
);
1112 tq
->tq_name
= kmem_strdup(name
);
1114 tq
->tq_nthreads
= 0;
1116 tq
->tq_maxthreads
= nthreads
;
1117 tq
->tq_cpu_pct
= threads_arg
;
1119 tq
->tq_minalloc
= minalloc
;
1120 tq
->tq_maxalloc
= maxalloc
;
1122 tq
->tq_flags
= (flags
| TASKQ_ACTIVE
);
1123 tq
->tq_next_id
= TASKQID_INITIAL
;
1124 tq
->tq_lowest_id
= TASKQID_INITIAL
;
1125 tq
->lastshouldstop
= 0;
1126 INIT_LIST_HEAD(&tq
->tq_free_list
);
1127 INIT_LIST_HEAD(&tq
->tq_pend_list
);
1128 INIT_LIST_HEAD(&tq
->tq_prio_list
);
1129 INIT_LIST_HEAD(&tq
->tq_delay_list
);
1130 init_waitqueue_head(&tq
->tq_work_waitq
);
1131 init_waitqueue_head(&tq
->tq_wait_waitq
);
1132 tq
->tq_lock_class
= TQ_LOCK_GENERAL
;
1133 INIT_LIST_HEAD(&tq
->tq_taskqs
);
1135 if (flags
& TASKQ_PREPOPULATE
) {
1136 spin_lock_irqsave_nested(&tq
->tq_lock
, irqflags
,
1139 for (i
= 0; i
< minalloc
; i
++)
1140 task_done(tq
, task_alloc(tq
, TQ_PUSHPAGE
| TQ_NEW
,
1143 spin_unlock_irqrestore(&tq
->tq_lock
, irqflags
);
1146 if ((flags
& TASKQ_DYNAMIC
) && spl_taskq_thread_dynamic
)
1149 for (i
= 0; i
< nthreads
; i
++) {
1150 tqt
= taskq_thread_create(tq
);
1157 /* Wait for all threads to be started before potential destroy */
1158 wait_event(tq
->tq_wait_waitq
, tq
->tq_nthreads
== count
);
1160 * taskq_thread might have touched nspawn, but we don't want them to
1161 * because they're not dynamically spawned. So we reset it to 0
1169 down_write(&tq_list_sem
);
1170 tq
->tq_instance
= taskq_find_by_name(name
) + 1;
1171 list_add_tail(&tq
->tq_taskqs
, &tq_list
);
1172 up_write(&tq_list_sem
);
1177 EXPORT_SYMBOL(taskq_create
);
1180 taskq_destroy(taskq_t
*tq
)
1182 struct task_struct
*thread
;
1183 taskq_thread_t
*tqt
;
1185 unsigned long flags
;
1188 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
, tq
->tq_lock_class
);
1189 tq
->tq_flags
&= ~TASKQ_ACTIVE
;
1190 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
1192 #ifdef HAVE_CPU_HOTPLUG
1193 if (tq
->tq_hp_support
) {
1194 VERIFY0(cpuhp_state_remove_instance_nocalls(
1195 spl_taskq_cpuhp_state
, &tq
->tq_hp_cb_node
));
1199 * When TASKQ_ACTIVE is clear new tasks may not be added nor may
1200 * new worker threads be spawned for dynamic taskq.
1202 if (dynamic_taskq
!= NULL
)
1203 taskq_wait_outstanding(dynamic_taskq
, 0);
1207 /* remove taskq from global list used by the kstats */
1208 down_write(&tq_list_sem
);
1209 list_del(&tq
->tq_taskqs
);
1210 up_write(&tq_list_sem
);
1212 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
, tq
->tq_lock_class
);
1213 /* wait for spawning threads to insert themselves to the list */
1214 while (tq
->tq_nspawn
) {
1215 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
1216 schedule_timeout_interruptible(1);
1217 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
,
1222 * Signal each thread to exit and block until it does. Each thread
1223 * is responsible for removing itself from the list and freeing its
1224 * taskq_thread_t. This allows for idle threads to opt to remove
1225 * themselves from the taskq. They can be recreated as needed.
1227 while (!list_empty(&tq
->tq_thread_list
)) {
1228 tqt
= list_entry(tq
->tq_thread_list
.next
,
1229 taskq_thread_t
, tqt_thread_list
);
1230 thread
= tqt
->tqt_thread
;
1231 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
1233 kthread_stop(thread
);
1235 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
,
1239 while (!list_empty(&tq
->tq_free_list
)) {
1240 t
= list_entry(tq
->tq_free_list
.next
, taskq_ent_t
, tqent_list
);
1242 ASSERT(!(t
->tqent_flags
& TQENT_FLAG_PREALLOC
));
1244 list_del_init(&t
->tqent_list
);
1248 ASSERT0(tq
->tq_nthreads
);
1249 ASSERT0(tq
->tq_nalloc
);
1250 ASSERT0(tq
->tq_nspawn
);
1251 ASSERT(list_empty(&tq
->tq_thread_list
));
1252 ASSERT(list_empty(&tq
->tq_active_list
));
1253 ASSERT(list_empty(&tq
->tq_free_list
));
1254 ASSERT(list_empty(&tq
->tq_pend_list
));
1255 ASSERT(list_empty(&tq
->tq_prio_list
));
1256 ASSERT(list_empty(&tq
->tq_delay_list
));
1258 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
1260 kmem_strfree(tq
->tq_name
);
1261 kmem_free(tq
, sizeof (taskq_t
));
1263 EXPORT_SYMBOL(taskq_destroy
);
1265 static unsigned int spl_taskq_kick
= 0;
1269 * module_param_cb is introduced to take kernel_param_ops and
1270 * module_param_call is marked as obsolete. Also set and get operations
1271 * were changed to take a 'const struct kernel_param *'.
1274 #ifdef module_param_cb
1275 param_set_taskq_kick(const char *val
, const struct kernel_param
*kp
)
1277 param_set_taskq_kick(const char *val
, struct kernel_param
*kp
)
1283 unsigned long flags
;
1285 ret
= param_set_uint(val
, kp
);
1286 if (ret
< 0 || !spl_taskq_kick
)
1291 down_read(&tq_list_sem
);
1292 list_for_each_entry(tq
, &tq_list
, tq_taskqs
) {
1293 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
,
1295 /* Check if the first pending is older than 5 seconds */
1296 t
= taskq_next_ent(tq
);
1297 if (t
&& time_after(jiffies
, t
->tqent_birth
+ 5*HZ
)) {
1298 (void) taskq_thread_spawn(tq
);
1299 printk(KERN_INFO
"spl: Kicked taskq %s/%d\n",
1300 tq
->tq_name
, tq
->tq_instance
);
1302 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
1304 up_read(&tq_list_sem
);
1308 #ifdef module_param_cb
1309 static const struct kernel_param_ops param_ops_taskq_kick
= {
1310 .set
= param_set_taskq_kick
,
1311 .get
= param_get_uint
,
1313 module_param_cb(spl_taskq_kick
, ¶m_ops_taskq_kick
, &spl_taskq_kick
, 0644);
1315 module_param_call(spl_taskq_kick
, param_set_taskq_kick
, param_get_uint
,
1316 &spl_taskq_kick
, 0644);
1318 MODULE_PARM_DESC(spl_taskq_kick
,
1319 "Write nonzero to kick stuck taskqs to spawn more threads");
1321 #ifdef HAVE_CPU_HOTPLUG
1323 * This callback will be called exactly once for each core that comes online,
1324 * for each dynamic taskq. We attempt to expand taskqs that have
1325 * TASKQ_THREADS_CPU_PCT set. We need to redo the percentage calculation every
1326 * time, to correctly determine whether or not to add a thread.
1329 spl_taskq_expand(unsigned int cpu
, struct hlist_node
*node
)
1331 taskq_t
*tq
= list_entry(node
, taskq_t
, tq_hp_cb_node
);
1332 unsigned long flags
;
1336 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
, tq
->tq_lock_class
);
1338 if (!(tq
->tq_flags
& TASKQ_ACTIVE
)) {
1339 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
1343 ASSERT(tq
->tq_flags
& TASKQ_THREADS_CPU_PCT
);
1344 int nthreads
= MIN(tq
->tq_cpu_pct
, 100);
1345 nthreads
= MAX(((num_online_cpus() + 1) * nthreads
) / 100, 1);
1346 tq
->tq_maxthreads
= nthreads
;
1348 if (!((tq
->tq_flags
& TASKQ_DYNAMIC
) && spl_taskq_thread_dynamic
) &&
1349 tq
->tq_maxthreads
> tq
->tq_nthreads
) {
1350 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
1351 taskq_thread_t
*tqt
= taskq_thread_create(tq
);
1356 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
1361 * While we don't support offlining CPUs, it is possible that CPUs will fail
1362 * to online successfully. We do need to be able to handle this case
1366 spl_taskq_prepare_down(unsigned int cpu
, struct hlist_node
*node
)
1368 taskq_t
*tq
= list_entry(node
, taskq_t
, tq_hp_cb_node
);
1369 unsigned long flags
;
1372 spin_lock_irqsave_nested(&tq
->tq_lock
, flags
, tq
->tq_lock_class
);
1374 if (!(tq
->tq_flags
& TASKQ_ACTIVE
))
1377 ASSERT(tq
->tq_flags
& TASKQ_THREADS_CPU_PCT
);
1378 int nthreads
= MIN(tq
->tq_cpu_pct
, 100);
1379 nthreads
= MAX(((num_online_cpus()) * nthreads
) / 100, 1);
1380 tq
->tq_maxthreads
= nthreads
;
1382 if (!((tq
->tq_flags
& TASKQ_DYNAMIC
) && spl_taskq_thread_dynamic
) &&
1383 tq
->tq_maxthreads
< tq
->tq_nthreads
) {
1384 ASSERT3U(tq
->tq_maxthreads
, ==, tq
->tq_nthreads
- 1);
1385 taskq_thread_t
*tqt
= list_entry(tq
->tq_thread_list
.next
,
1386 taskq_thread_t
, tqt_thread_list
);
1387 struct task_struct
*thread
= tqt
->tqt_thread
;
1388 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
1390 kthread_stop(thread
);
1396 spin_unlock_irqrestore(&tq
->tq_lock
, flags
);
1402 spl_taskq_init(void)
1404 init_rwsem(&tq_list_sem
);
1405 tsd_create(&taskq_tsd
, NULL
);
1407 #ifdef HAVE_CPU_HOTPLUG
1408 spl_taskq_cpuhp_state
= cpuhp_setup_state_multi(CPUHP_AP_ONLINE_DYN
,
1409 "fs/spl_taskq:online", spl_taskq_expand
, spl_taskq_prepare_down
);
1412 system_taskq
= taskq_create("spl_system_taskq", MAX(boot_ncpus
, 64),
1413 maxclsyspri
, boot_ncpus
, INT_MAX
, TASKQ_PREPOPULATE
|TASKQ_DYNAMIC
);
1414 if (system_taskq
== NULL
)
1417 system_delay_taskq
= taskq_create("spl_delay_taskq", MAX(boot_ncpus
, 4),
1418 maxclsyspri
, boot_ncpus
, INT_MAX
, TASKQ_PREPOPULATE
|TASKQ_DYNAMIC
);
1419 if (system_delay_taskq
== NULL
) {
1420 #ifdef HAVE_CPU_HOTPLUG
1421 cpuhp_remove_multi_state(spl_taskq_cpuhp_state
);
1423 taskq_destroy(system_taskq
);
1427 dynamic_taskq
= taskq_create("spl_dynamic_taskq", 1,
1428 maxclsyspri
, boot_ncpus
, INT_MAX
, TASKQ_PREPOPULATE
);
1429 if (dynamic_taskq
== NULL
) {
1430 #ifdef HAVE_CPU_HOTPLUG
1431 cpuhp_remove_multi_state(spl_taskq_cpuhp_state
);
1433 taskq_destroy(system_taskq
);
1434 taskq_destroy(system_delay_taskq
);
1439 * This is used to annotate tq_lock, so
1440 * taskq_dispatch -> taskq_thread_spawn -> taskq_dispatch
1441 * does not trigger a lockdep warning re: possible recursive locking
1443 dynamic_taskq
->tq_lock_class
= TQ_LOCK_DYNAMIC
;
1449 spl_taskq_fini(void)
1451 taskq_destroy(dynamic_taskq
);
1452 dynamic_taskq
= NULL
;
1454 taskq_destroy(system_delay_taskq
);
1455 system_delay_taskq
= NULL
;
1457 taskq_destroy(system_taskq
);
1458 system_taskq
= NULL
;
1460 tsd_destroy(&taskq_tsd
);
1462 #ifdef HAVE_CPU_HOTPLUG
1463 cpuhp_remove_multi_state(spl_taskq_cpuhp_state
);
1464 spl_taskq_cpuhp_state
= 0;