]>
git.proxmox.com Git - mirror_spl-debian.git/blob - module/spl/spl-taskq.c
2 * This file is part of the SPL: Solaris Porting Layer.
4 * Copyright (c) 2008 Lawrence Livermore National Security, LLC.
5 * Produced at Lawrence Livermore National Laboratory
7 * Brian Behlendorf <behlendorf1@llnl.gov>,
8 * Herb Wartens <wartens2@llnl.gov>,
9 * Jim Garlick <garlick@llnl.gov>
12 * This is free software; you can redistribute it and/or modify it
13 * under the terms of the GNU General Public License as published by
14 * the Free Software Foundation; either version 2 of the License, or
15 * (at your option) any later version.
17 * This is distributed in the hope that it will be useful, but WITHOUT
18 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
19 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
22 * You should have received a copy of the GNU General Public License along
23 * with this program; if not, write to the Free Software Foundation, Inc.,
24 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
27 #include <sys/taskq.h>
30 #ifdef DEBUG_SUBSYSTEM
31 #undef DEBUG_SUBSYSTEM
34 #define DEBUG_SUBSYSTEM S_TASKQ
36 /* Global system-wide dynamic task queue available for all consumers */
37 taskq_t
*system_taskq
;
38 EXPORT_SYMBOL(system_taskq
);
40 typedef struct spl_task
{
42 struct list_head t_list
;
48 /* NOTE: Must be called with tq->tq_lock held, returns a list_t which
49 * is not attached to the free, work, or pending taskq lists.
52 task_alloc(taskq_t
*tq
, uint_t flags
)
59 ASSERT(flags
& (TQ_SLEEP
| TQ_NOSLEEP
)); /* One set */
60 ASSERT(!((flags
& TQ_SLEEP
) && (flags
& TQ_NOSLEEP
))); /* Not both */
61 ASSERT(spin_is_locked(&tq
->tq_lock
));
63 /* Aquire spl_task_t's from free list if available */
64 if (!list_empty(&tq
->tq_free_list
) && !(flags
& TQ_NEW
)) {
65 t
= list_entry(tq
->tq_free_list
.next
, spl_task_t
, t_list
);
66 list_del_init(&t
->t_list
);
70 /* Free list is empty and memory allocs are prohibited */
71 if (flags
& TQ_NOALLOC
)
74 /* Hit maximum spl_task_t pool size */
75 if (tq
->tq_nalloc
>= tq
->tq_maxalloc
) {
76 if (flags
& TQ_NOSLEEP
)
79 /* Sleep periodically polling the free list for an available
80 * spl_task_t. If a full second passes and we have not found
81 * one gives up and return a NULL to the caller. */
82 if (flags
& TQ_SLEEP
) {
83 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
84 schedule_timeout(HZ
/ 100);
85 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
92 /* Unreachable, TQ_SLEEP xor TQ_NOSLEEP */
96 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
97 t
= kmem_alloc(sizeof(spl_task_t
), flags
& (TQ_SLEEP
| TQ_NOSLEEP
));
98 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
101 spin_lock_init(&t
->t_lock
);
102 INIT_LIST_HEAD(&t
->t_list
);
112 /* NOTE: Must be called with tq->tq_lock held, expectes the spl_task_t
113 * to already be removed from the free, work, or pending taskq lists.
116 task_free(taskq_t
*tq
, spl_task_t
*t
)
122 ASSERT(spin_is_locked(&tq
->tq_lock
));
123 ASSERT(list_empty(&t
->t_list
));
125 kmem_free(t
, sizeof(spl_task_t
));
131 /* NOTE: Must be called with tq->tq_lock held, either destroyes the
132 * spl_task_t if too many exist or moves it to the free list for later use.
135 task_done(taskq_t
*tq
, spl_task_t
*t
)
140 ASSERT(spin_is_locked(&tq
->tq_lock
));
142 list_del_init(&t
->t_list
);
144 if (tq
->tq_nalloc
<= tq
->tq_minalloc
) {
148 list_add_tail(&t
->t_list
, &tq
->tq_free_list
);
156 /* Taskqid's are handed out in a monotonically increasing fashion per
157 * taskq_t. We don't handle taskqid wrapping yet, but fortuntely it isi
158 * a 64-bit value so this is probably never going to happen. The lowest
159 * pending taskqid is stored in the taskq_t to make it easy for any
160 * taskq_wait()'ers to know if the tasks they're waiting for have
161 * completed. Unfortunately, tq_task_lowest is kept up to date is
162 * a pretty brain dead way, something more clever should be done.
165 taskq_wait_check(taskq_t
*tq
, taskqid_t id
)
167 RETURN(tq
->tq_lowest_id
>= id
);
170 /* Expected to wait for all previously scheduled tasks to complete. We do
171 * not need to wait for tasked scheduled after this call to complete. In
172 * otherwords we do not need to drain the entire taskq. */
174 __taskq_wait_id(taskq_t
*tq
, taskqid_t id
)
179 wait_event(tq
->tq_wait_waitq
, taskq_wait_check(tq
, id
));
183 EXPORT_SYMBOL(__taskq_wait_id
);
186 __taskq_wait(taskq_t
*tq
)
192 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
194 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
196 __taskq_wait_id(tq
, id
);
201 EXPORT_SYMBOL(__taskq_wait
);
204 __taskq_member(taskq_t
*tq
, void *t
)
212 for (i
= 0; i
< tq
->tq_nthreads
; i
++)
213 if (tq
->tq_threads
[i
] == (struct task_struct
*)t
)
218 EXPORT_SYMBOL(__taskq_member
);
221 __taskq_dispatch(taskq_t
*tq
, task_func_t func
, void *arg
, uint_t flags
)
229 if (unlikely(in_atomic() && (flags
& TQ_SLEEP
))) {
230 CERROR("May schedule while atomic: %s/0x%08x/%d\n",
231 current
->comm
, preempt_count(), current
->pid
);
235 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
237 /* Taskq being destroyed and all tasks drained */
238 if (!(tq
->tq_flags
& TQ_ACTIVE
))
241 /* Do not queue the task unless there is idle thread for it */
242 ASSERT(tq
->tq_nactive
<= tq
->tq_nthreads
);
243 if ((flags
& TQ_NOQUEUE
) && (tq
->tq_nactive
== tq
->tq_nthreads
))
246 if ((t
= task_alloc(tq
, flags
)) == NULL
)
249 spin_lock(&t
->t_lock
);
250 list_add_tail(&t
->t_list
, &tq
->tq_pend_list
);
251 t
->t_id
= rc
= tq
->tq_next_id
;
255 spin_unlock(&t
->t_lock
);
257 wake_up(&tq
->tq_work_waitq
);
259 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
262 EXPORT_SYMBOL(__taskq_dispatch
);
264 /* NOTE: Must be called with tq->tq_lock held */
266 taskq_lowest_id(taskq_t
*tq
)
268 taskqid_t lowest_id
= ~0;
273 ASSERT(spin_is_locked(&tq
->tq_lock
));
275 list_for_each_entry(t
, &tq
->tq_pend_list
, t_list
)
276 if (t
->t_id
< lowest_id
)
279 list_for_each_entry(t
, &tq
->tq_work_list
, t_list
)
280 if (t
->t_id
< lowest_id
)
287 taskq_thread(void *args
)
289 DECLARE_WAITQUEUE(wait
, current
);
297 current
->flags
|= PF_NOFREEZE
;
299 sigfillset(&blocked
);
300 sigprocmask(SIG_BLOCK
, &blocked
, NULL
);
301 flush_signals(current
);
303 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
305 wake_up(&tq
->tq_wait_waitq
);
306 set_current_state(TASK_INTERRUPTIBLE
);
308 while (!kthread_should_stop()) {
310 add_wait_queue(&tq
->tq_work_waitq
, &wait
);
311 if (list_empty(&tq
->tq_pend_list
)) {
312 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
314 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
316 __set_current_state(TASK_RUNNING
);
319 remove_wait_queue(&tq
->tq_work_waitq
, &wait
);
320 if (!list_empty(&tq
->tq_pend_list
)) {
321 t
= list_entry(tq
->tq_pend_list
.next
, spl_task_t
, t_list
);
322 list_del_init(&t
->t_list
);
323 list_add_tail(&t
->t_list
, &tq
->tq_work_list
);
325 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
327 /* Perform the requested task */
330 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
335 /* Update the lowest remaining taskqid yet to run */
336 if (tq
->tq_lowest_id
== id
) {
337 tq
->tq_lowest_id
= taskq_lowest_id(tq
);
338 ASSERT(tq
->tq_lowest_id
> id
);
341 wake_up_all(&tq
->tq_wait_waitq
);
344 set_current_state(TASK_INTERRUPTIBLE
);
348 __set_current_state(TASK_RUNNING
);
350 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
356 __taskq_create(const char *name
, int nthreads
, pri_t pri
,
357 int minalloc
, int maxalloc
, uint_t flags
)
360 struct task_struct
*t
;
361 int rc
= 0, i
, j
= 0;
364 ASSERT(name
!= NULL
);
365 ASSERT(pri
<= maxclsyspri
);
366 ASSERT(minalloc
>= 0);
367 ASSERT(maxalloc
<= INT_MAX
);
368 ASSERT(!(flags
& (TASKQ_CPR_SAFE
| TASKQ_DYNAMIC
))); /* Unsupported */
370 tq
= kmem_alloc(sizeof(*tq
), KM_SLEEP
);
374 tq
->tq_threads
= kmem_alloc(nthreads
* sizeof(t
), KM_SLEEP
);
375 if (tq
->tq_threads
== NULL
) {
376 kmem_free(tq
, sizeof(*tq
));
380 spin_lock_init(&tq
->tq_lock
);
381 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
386 tq
->tq_minalloc
= minalloc
;
387 tq
->tq_maxalloc
= maxalloc
;
389 tq
->tq_flags
= (flags
| TQ_ACTIVE
);
391 tq
->tq_lowest_id
= 1;
392 INIT_LIST_HEAD(&tq
->tq_free_list
);
393 INIT_LIST_HEAD(&tq
->tq_work_list
);
394 INIT_LIST_HEAD(&tq
->tq_pend_list
);
395 init_waitqueue_head(&tq
->tq_work_waitq
);
396 init_waitqueue_head(&tq
->tq_wait_waitq
);
398 if (flags
& TASKQ_PREPOPULATE
)
399 for (i
= 0; i
< minalloc
; i
++)
400 task_done(tq
, task_alloc(tq
, TQ_SLEEP
| TQ_NEW
));
402 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
404 for (i
= 0; i
< nthreads
; i
++) {
405 t
= kthread_create(taskq_thread
, tq
, "%s/%d", name
, i
);
407 tq
->tq_threads
[i
] = t
;
408 kthread_bind(t
, i
% num_online_cpus());
409 set_user_nice(t
, PRIO_TO_NICE(pri
));
413 tq
->tq_threads
[i
] = NULL
;
418 /* Wait for all threads to be started before potential destroy */
419 wait_event(tq
->tq_wait_waitq
, tq
->tq_nthreads
== j
);
428 EXPORT_SYMBOL(__taskq_create
);
431 __taskq_destroy(taskq_t
*tq
)
438 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
439 tq
->tq_flags
&= ~TQ_ACTIVE
;
440 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
442 /* TQ_ACTIVE cleared prevents new tasks being added to pending */
445 nthreads
= tq
->tq_nthreads
;
446 for (i
= 0; i
< nthreads
; i
++)
447 if (tq
->tq_threads
[i
])
448 kthread_stop(tq
->tq_threads
[i
]);
450 spin_lock_irqsave(&tq
->tq_lock
, tq
->tq_lock_flags
);
452 while (!list_empty(&tq
->tq_free_list
)) {
453 t
= list_entry(tq
->tq_free_list
.next
, spl_task_t
, t_list
);
454 list_del_init(&t
->t_list
);
458 ASSERT(tq
->tq_nthreads
== 0);
459 ASSERT(tq
->tq_nalloc
== 0);
460 ASSERT(list_empty(&tq
->tq_free_list
));
461 ASSERT(list_empty(&tq
->tq_work_list
));
462 ASSERT(list_empty(&tq
->tq_pend_list
));
464 spin_unlock_irqrestore(&tq
->tq_lock
, tq
->tq_lock_flags
);
465 kmem_free(tq
->tq_threads
, nthreads
* sizeof(spl_task_t
*));
466 kmem_free(tq
, sizeof(taskq_t
));
470 EXPORT_SYMBOL(__taskq_destroy
);
477 system_taskq
= taskq_create("system_taskq", 64, minclsyspri
, 4, 512,
479 if (system_taskq
== NULL
)
489 taskq_destroy(system_taskq
);