]>
git.proxmox.com Git - mirror_zfs.git/blob - module/spl/spl-taskq.c
1 /*****************************************************************************\
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.
9 * For details, see <http://github.com/behlendorf/spl/>.
11 * The SPL is free software; you can redistribute it and/or modify it
12 * under the terms of the GNU General Public License as published by the
13 * Free Software Foundation; either version 2 of the License, or (at your
14 * option) any later version.
16 * The SPL is distributed in the hope that it will be useful, but WITHOUT
17 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
18 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
21 * You should have received a copy of the GNU General Public License along
22 * with the SPL. If not, see <http://www.gnu.org/licenses/>.
23 *****************************************************************************
24 * Solaris Porting Layer (SPL) Task Queue Implementation.
25 \*****************************************************************************/
27 #include <sys/taskq.h>
29 #include <spl-debug.h>
31 #ifdef SS_DEBUG_SUBSYS
32 #undef SS_DEBUG_SUBSYS
35 #define SS_DEBUG_SUBSYS SS_TASKQ
37 /* Global system-wide dynamic task queue available for all consumers */
38 taskq_t
*system_taskq
;
39 EXPORT_SYMBOL(system_taskq
);
41 typedef struct taskq_ent
{
42 spinlock_t tqent_lock
;
43 struct list_head tqent_list
;
45 task_func_t
*tqent_func
;
50 * NOTE: Must be called with tq->tq_lock held, returns a list_t which
51 * is not attached to the free, work, or pending taskq lists.
54 task_alloc(taskq_t
*tq
, uint_t flags
)
61 ASSERT(flags
& (TQ_SLEEP
| TQ_NOSLEEP
)); /* One set */
62 ASSERT(!((flags
& TQ_SLEEP
) && (flags
& TQ_NOSLEEP
))); /* Not both */
63 ASSERT(spin_is_locked(&tq
->tq_lock
));
65 /* Acquire taskq_ent_t's from free list if available */
66 if (!list_empty(&tq
->tq_free_list
) && !(flags
& TQ_NEW
)) {
67 t
= list_entry(tq
->tq_free_list
.next
, taskq_ent_t
, tqent_list
);
68 list_del_init(&t
->tqent_list
);
72 /* Free list is empty and memory allocations are prohibited */
73 if (flags
& TQ_NOALLOC
)
76 /* Hit maximum taskq_ent_t pool size */
77 if (tq
->tq_nalloc
>= tq
->tq_maxalloc
) {
78 if (flags
& TQ_NOSLEEP
)
82 * Sleep periodically polling the free list for an available
83 * taskq_ent_t. Dispatching with TQ_SLEEP should always succeed
84 * but we cannot block forever waiting for an taskq_entq_t to
85 * show up in the free list, otherwise a deadlock can happen.
87 * Therefore, we need to allocate a new task even if the number
88 * of allocated tasks is above tq->tq_maxalloc, but we still
89 * end up delaying the task allocation by one second, thereby
90 * throttling the task dispatch rate.
92 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
93 schedule_timeout(HZ
/ 100);
94 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
96 SGOTO(retry
, count
++);
99 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
100 t
= kmem_alloc(sizeof(taskq_ent_t
), flags
& (TQ_SLEEP
| TQ_NOSLEEP
));
101 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
104 spin_lock_init(&t
->tqent_lock
);
105 INIT_LIST_HEAD(&t
->tqent_list
);
107 t
->tqent_func
= NULL
;
116 * NOTE: Must be called with tq->tq_lock held, expects the taskq_ent_t
117 * to already be removed from the free, work, or pending taskq lists.
120 task_free(taskq_t
*tq
, taskq_ent_t
*t
)
126 ASSERT(spin_is_locked(&tq
->tq_lock
));
127 ASSERT(list_empty(&t
->tqent_list
));
129 kmem_free(t
, sizeof(taskq_ent_t
));
136 * NOTE: Must be called with tq->tq_lock held, either destroys the
137 * taskq_ent_t if too many exist or moves it to the free list for later use.
140 task_done(taskq_t
*tq
, taskq_ent_t
*t
)
145 ASSERT(spin_is_locked(&tq
->tq_lock
));
147 list_del_init(&t
->tqent_list
);
149 if (tq
->tq_nalloc
<= tq
->tq_minalloc
) {
151 t
->tqent_func
= NULL
;
153 list_add_tail(&t
->tqent_list
, &tq
->tq_free_list
);
162 * As tasks are submitted to the task queue they are assigned a
163 * monotonically increasing taskqid and added to the tail of the pending
164 * list. As worker threads become available the tasks are removed from
165 * the head of the pending or priority list, giving preference to the
166 * priority list. The tasks are then added to the work list, preserving
167 * the ordering by taskqid. Finally, as tasks complete they are removed
168 * from the work list. This means that the pending and work lists are
169 * always kept sorted by taskqid. Thus the lowest outstanding
170 * incomplete taskqid can be determined simply by checking the min
171 * taskqid for each head item on the pending, priority, and work list.
172 * This value is stored in tq->tq_lowest_id and only updated to the new
173 * lowest id when the previous lowest id completes. All taskqids lower
174 * than tq->tq_lowest_id must have completed. It is also possible
175 * larger taskqid's have completed because they may be processed in
176 * parallel by several worker threads. However, this is not a problem
177 * because the behavior of taskq_wait_id() is to block until all
178 * previously submitted taskqid's have completed.
180 * XXX: Taskqid_t wrapping is not handled. However, taskqid_t's are
181 * 64-bit values so even if a taskq is processing 2^24 (16,777,216)
182 * taskqid_ts per second it will still take 2^40 seconds, 34,865 years,
183 * before the wrap occurs. I can live with that for now.
186 taskq_wait_check(taskq_t
*tq
, taskqid_t id
)
190 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
191 rc
= (id
< tq
->tq_lowest_id
);
192 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
198 __taskq_wait_id(taskq_t
*tq
, taskqid_t id
)
203 wait_event(tq
->tq_wait_waitq
, taskq_wait_check(tq
, id
));
207 EXPORT_SYMBOL(__taskq_wait_id
);
210 __taskq_wait(taskq_t
*tq
)
216 /* Wait for the largest outstanding taskqid */
217 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
218 id
= tq
->tq_next_id
- 1;
219 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
221 __taskq_wait_id(tq
, id
);
226 EXPORT_SYMBOL(__taskq_wait
);
229 __taskq_member(taskq_t
*tq
, void *t
)
237 for (i
= 0; i
< tq
->tq_nthreads
; i
++)
238 if (tq
->tq_threads
[i
] == (struct task_struct
*)t
)
243 EXPORT_SYMBOL(__taskq_member
);
246 __taskq_dispatch(taskq_t
*tq
, task_func_t func
, void *arg
, uint_t flags
)
255 /* Solaris assumes TQ_SLEEP if not passed explicitly */
256 if (!(flags
& (TQ_SLEEP
| TQ_NOSLEEP
)))
259 if (unlikely(in_atomic() && (flags
& TQ_SLEEP
)))
260 PANIC("May schedule while atomic: %s/0x%08x/%d\n",
261 current
->comm
, preempt_count(), current
->pid
);
263 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
265 /* Taskq being destroyed and all tasks drained */
266 if (!(tq
->tq_flags
& TQ_ACTIVE
))
269 /* Do not queue the task unless there is idle thread for it */
270 ASSERT(tq
->tq_nactive
<= tq
->tq_nthreads
);
271 if ((flags
& TQ_NOQUEUE
) && (tq
->tq_nactive
== tq
->tq_nthreads
))
274 if ((t
= task_alloc(tq
, flags
)) == NULL
)
277 spin_lock(&t
->tqent_lock
);
279 /* Queue to the priority list instead of the pending list */
280 if (flags
& TQ_FRONT
)
281 list_add_tail(&t
->tqent_list
, &tq
->tq_prio_list
);
283 list_add_tail(&t
->tqent_list
, &tq
->tq_pend_list
);
285 t
->tqent_id
= rc
= tq
->tq_next_id
;
287 t
->tqent_func
= func
;
289 spin_unlock(&t
->tqent_lock
);
291 wake_up(&tq
->tq_work_waitq
);
293 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
296 EXPORT_SYMBOL(__taskq_dispatch
);
298 * Returns the lowest incomplete taskqid_t. The taskqid_t may
299 * be queued on the pending list, on the priority list, or on
300 * the work list currently being handled, but it is not 100%
304 taskq_lowest_id(taskq_t
*tq
)
306 taskqid_t lowest_id
= tq
->tq_next_id
;
311 ASSERT(spin_is_locked(&tq
->tq_lock
));
313 if (!list_empty(&tq
->tq_pend_list
)) {
314 t
= list_entry(tq
->tq_pend_list
.next
, taskq_ent_t
, tqent_list
);
315 lowest_id
= MIN(lowest_id
, t
->tqent_id
);
318 if (!list_empty(&tq
->tq_prio_list
)) {
319 t
= list_entry(tq
->tq_prio_list
.next
, taskq_ent_t
, tqent_list
);
320 lowest_id
= MIN(lowest_id
, t
->tqent_id
);
323 if (!list_empty(&tq
->tq_work_list
)) {
324 t
= list_entry(tq
->tq_work_list
.next
, taskq_ent_t
, tqent_list
);
325 lowest_id
= MIN(lowest_id
, t
->tqent_id
);
332 * Insert a task into a list keeping the list sorted by increasing
336 taskq_insert_in_order(taskq_t
*tq
, taskq_ent_t
*t
)
344 ASSERT(spin_is_locked(&tq
->tq_lock
));
346 list_for_each_prev(l
, &tq
->tq_work_list
) {
347 w
= list_entry(l
, taskq_ent_t
, tqent_list
);
348 if (w
->tqent_id
< t
->tqent_id
) {
349 list_add(&t
->tqent_list
, l
);
353 if (l
== &tq
->tq_work_list
)
354 list_add(&t
->tqent_list
, &tq
->tq_work_list
);
360 taskq_thread(void *args
)
362 DECLARE_WAITQUEUE(wait
, current
);
367 struct list_head
*pend_list
;
371 current
->flags
|= PF_NOFREEZE
;
373 /* Disable the direct memory reclaim path */
374 if (tq
->tq_flags
& TASKQ_NORECLAIM
)
375 current
->flags
|= PF_MEMALLOC
;
377 sigfillset(&blocked
);
378 sigprocmask(SIG_BLOCK
, &blocked
, NULL
);
379 flush_signals(current
);
381 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
383 wake_up(&tq
->tq_wait_waitq
);
384 set_current_state(TASK_INTERRUPTIBLE
);
386 while (!kthread_should_stop()) {
388 add_wait_queue(&tq
->tq_work_waitq
, &wait
);
389 if (list_empty(&tq
->tq_pend_list
) &&
390 list_empty(&tq
->tq_prio_list
)) {
391 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
393 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
395 __set_current_state(TASK_RUNNING
);
398 remove_wait_queue(&tq
->tq_work_waitq
, &wait
);
400 if (!list_empty(&tq
->tq_prio_list
))
401 pend_list
= &tq
->tq_prio_list
;
402 else if (!list_empty(&tq
->tq_pend_list
))
403 pend_list
= &tq
->tq_pend_list
;
408 t
= list_entry(pend_list
->next
, taskq_ent_t
, tqent_list
);
409 list_del_init(&t
->tqent_list
);
410 taskq_insert_in_order(tq
, t
);
412 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
414 /* Perform the requested task */
415 t
->tqent_func(t
->tqent_arg
);
417 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
422 /* When the current lowest outstanding taskqid is
423 * done calculate the new lowest outstanding id */
424 if (tq
->tq_lowest_id
== id
) {
425 tq
->tq_lowest_id
= taskq_lowest_id(tq
);
426 ASSERT(tq
->tq_lowest_id
> id
);
429 wake_up_all(&tq
->tq_wait_waitq
);
432 set_current_state(TASK_INTERRUPTIBLE
);
436 __set_current_state(TASK_RUNNING
);
438 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
444 __taskq_create(const char *name
, int nthreads
, pri_t pri
,
445 int minalloc
, int maxalloc
, uint_t flags
)
448 struct task_struct
*t
;
449 int rc
= 0, i
, j
= 0;
452 ASSERT(name
!= NULL
);
453 ASSERT(pri
<= maxclsyspri
);
454 ASSERT(minalloc
>= 0);
455 ASSERT(maxalloc
<= INT_MAX
);
456 ASSERT(!(flags
& (TASKQ_CPR_SAFE
| TASKQ_DYNAMIC
))); /* Unsupported */
458 /* Scale the number of threads using nthreads as a percentage */
459 if (flags
& TASKQ_THREADS_CPU_PCT
) {
460 ASSERT(nthreads
<= 100);
461 ASSERT(nthreads
>= 0);
462 nthreads
= MIN(nthreads
, 100);
463 nthreads
= MAX(nthreads
, 0);
464 nthreads
= MAX((num_online_cpus() * nthreads
) / 100, 1);
467 tq
= kmem_alloc(sizeof(*tq
), KM_SLEEP
);
471 tq
->tq_threads
= kmem_alloc(nthreads
* sizeof(t
), KM_SLEEP
);
472 if (tq
->tq_threads
== NULL
) {
473 kmem_free(tq
, sizeof(*tq
));
477 spin_lock_init(&tq
->tq_lock
);
478 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
483 tq
->tq_minalloc
= minalloc
;
484 tq
->tq_maxalloc
= maxalloc
;
486 tq
->tq_flags
= (flags
| TQ_ACTIVE
);
488 tq
->tq_lowest_id
= 1;
489 INIT_LIST_HEAD(&tq
->tq_free_list
);
490 INIT_LIST_HEAD(&tq
->tq_work_list
);
491 INIT_LIST_HEAD(&tq
->tq_pend_list
);
492 INIT_LIST_HEAD(&tq
->tq_prio_list
);
493 init_waitqueue_head(&tq
->tq_work_waitq
);
494 init_waitqueue_head(&tq
->tq_wait_waitq
);
496 if (flags
& TASKQ_PREPOPULATE
)
497 for (i
= 0; i
< minalloc
; i
++)
498 task_done(tq
, task_alloc(tq
, TQ_SLEEP
| TQ_NEW
));
500 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
502 for (i
= 0; i
< nthreads
; i
++) {
503 t
= kthread_create(taskq_thread
, tq
, "%s/%d", name
, i
);
505 tq
->tq_threads
[i
] = t
;
506 kthread_bind(t
, i
% num_online_cpus());
507 set_user_nice(t
, PRIO_TO_NICE(pri
));
511 tq
->tq_threads
[i
] = NULL
;
516 /* Wait for all threads to be started before potential destroy */
517 wait_event(tq
->tq_wait_waitq
, tq
->tq_nthreads
== j
);
526 EXPORT_SYMBOL(__taskq_create
);
529 __taskq_destroy(taskq_t
*tq
)
536 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
537 tq
->tq_flags
&= ~TQ_ACTIVE
;
538 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
540 /* TQ_ACTIVE cleared prevents new tasks being added to pending */
543 nthreads
= tq
->tq_nthreads
;
544 for (i
= 0; i
< nthreads
; i
++)
545 if (tq
->tq_threads
[i
])
546 kthread_stop(tq
->tq_threads
[i
]);
548 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
550 while (!list_empty(&tq
->tq_free_list
)) {
551 t
= list_entry(tq
->tq_free_list
.next
, taskq_ent_t
, tqent_list
);
552 list_del_init(&t
->tqent_list
);
556 ASSERT(tq
->tq_nthreads
== 0);
557 ASSERT(tq
->tq_nalloc
== 0);
558 ASSERT(list_empty(&tq
->tq_free_list
));
559 ASSERT(list_empty(&tq
->tq_work_list
));
560 ASSERT(list_empty(&tq
->tq_pend_list
));
561 ASSERT(list_empty(&tq
->tq_prio_list
));
563 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
564 kmem_free(tq
->tq_threads
, nthreads
* sizeof(taskq_ent_t
*));
565 kmem_free(tq
, sizeof(taskq_t
));
569 EXPORT_SYMBOL(__taskq_destroy
);
576 /* Solaris creates a dynamic taskq of up to 64 threads, however in
577 * a Linux environment 1 thread per-core is usually about right */
578 system_taskq
= taskq_create("spl_system_taskq", num_online_cpus(),
579 minclsyspri
, 4, 512, TASKQ_PREPOPULATE
);
580 if (system_taskq
== NULL
)
590 taskq_destroy(system_taskq
);