Fix taskq dynamic spawning

[mirror_spl.git] / module / spl / spl-taskq.c

diff --git a/module/spl/spl-taskq.c b/module/spl/spl-taskq.c
index 44799de1d19360191236c5845a738e001548b7d0..2c2e3ad465abc2fad764816edb53723892595db6 100644 (file)
--- a/module/spl/spl-taskq.c
+++ b/module/spl/spl-taskq.c
@@ -36,6 +36,11 @@ int spl_taskq_thread_dynamic = 1;
 module_param(spl_taskq_thread_dynamic, int, 0644);
 MODULE_PARM_DESC(spl_taskq_thread_dynamic, "Allow dynamic taskq threads");
 
+int spl_taskq_thread_priority = 1;
+module_param(spl_taskq_thread_priority, int, 0644);
+MODULE_PARM_DESC(spl_taskq_thread_priority,
+    "Allow non-default priority for taskq threads");
+
 int spl_taskq_thread_sequential = 4;
 module_param(spl_taskq_thread_sequential, int, 0644);
 MODULE_PARM_DESC(spl_taskq_thread_sequential,
@@ -443,8 +448,8 @@ taskq_wait(taskq_t *tq)
 }
 EXPORT_SYMBOL(taskq_wait);
 
-int
-taskq_member(taskq_t *tq, void *t)
+static int
+taskq_member_impl(taskq_t *tq, void *t)
 {
        struct list_head *l;
        taskq_thread_t *tqt;
@@ -452,8 +457,8 @@ taskq_member(taskq_t *tq, void *t)
 
        ASSERT(tq);
        ASSERT(t);
+       ASSERT(spin_is_locked(&tq->tq_lock));
 
-       spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
        list_for_each(l, &tq->tq_thread_list) {
                tqt = list_entry(l, taskq_thread_t, tqt_thread_list);
                if (tqt->tqt_thread == (struct task_struct *)t) {
@@ -461,6 +466,16 @@ taskq_member(taskq_t *tq, void *t)
                        break;
                }
        }
+       return (found);
+}
+
+int
+taskq_member(taskq_t *tq, void *t)
+{
+       int found;
+
+       spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
+       found = taskq_member_impl(tq, t);
        spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 
        return (found);
@@ -523,6 +538,8 @@ taskq_cancel_id(taskq_t *tq, taskqid_t id)
 }
 EXPORT_SYMBOL(taskq_cancel_id);
 
+static int taskq_thread_spawn(taskq_t *tq);
+
 taskqid_t
 taskq_dispatch(taskq_t *tq, task_func_t func, void *arg, uint_t flags)
 {
@@ -569,6 +586,10 @@ taskq_dispatch(taskq_t *tq, task_func_t func, void *arg, uint_t flags)
 
        wake_up(&tq->tq_work_waitq);
 out:
+       /* Spawn additional taskq threads if required. */
+       if (tq->tq_nactive == tq->tq_nthreads)
+               (void) taskq_thread_spawn(tq);
+
        spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
        return (rc);
 }
@@ -612,6 +633,9 @@ taskq_dispatch_delay(taskq_t *tq, task_func_t func, void *arg,
 
        spin_unlock(&t->tqent_lock);
 out:
+       /* Spawn additional taskq threads if required. */
+       if (tq->tq_nactive == tq->tq_nthreads)
+               (void) taskq_thread_spawn(tq);
        spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
        return (rc);
 }
@@ -656,6 +680,9 @@ taskq_dispatch_ent(taskq_t *tq, task_func_t func, void *arg, uint_t flags,
 
        wake_up(&tq->tq_work_waitq);
 out:
+       /* Spawn additional taskq threads if required. */
+       if (tq->tq_nactive == tq->tq_nthreads)
+               (void) taskq_thread_spawn(tq);
        spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 }
 EXPORT_SYMBOL(taskq_dispatch_ent);
@@ -726,15 +753,14 @@ taskq_thread_spawn_task(void *arg)
  * which is also a dynamic taskq cannot be safely used for this.
  */
 static int
-taskq_thread_spawn(taskq_t *tq, int seq_tasks)
+taskq_thread_spawn(taskq_t *tq)
 {
        int spawning = 0;
 
        if (!(tq->tq_flags & TASKQ_DYNAMIC))
                return (0);
 
-       if ((seq_tasks > spl_taskq_thread_sequential) &&
-           (tq->tq_nthreads + tq->tq_nspawn < tq->tq_maxthreads) &&
+       if ((tq->tq_nthreads + tq->tq_nspawn < tq->tq_maxthreads) &&
            (tq->tq_flags & TASKQ_ACTIVE)) {
                spawning = (++tq->tq_nspawn);
                taskq_dispatch(dynamic_taskq, taskq_thread_spawn_task,
@@ -787,6 +813,10 @@ taskq_thread(void *args)
        tq = tqt->tqt_tq;
        current->flags |= PF_NOFREEZE;
 
+       #if defined(PF_MEMALLOC_NOIO)
+       (void) memalloc_noio_save();
+       #endif
+
        sigfillset(&blocked);
        sigprocmask(SIG_BLOCK, &blocked, NULL);
        flush_signals(current);
@@ -864,7 +894,8 @@ taskq_thread(void *args)
                        }
 
                        /* Spawn additional taskq threads if required. */
-                       if (taskq_thread_spawn(tq, ++seq_tasks))
+                       if ((++seq_tasks) > spl_taskq_thread_sequential &&
+                           taskq_thread_spawn(tq))
                                seq_tasks = 0;
 
                        tqt->tqt_id = 0;
@@ -913,7 +944,9 @@ taskq_thread_create(taskq_t *tq)
                kthread_bind(tqt->tqt_thread, last_used_cpu);
        }
 
-       set_user_nice(tqt->tqt_thread, PRIO_TO_NICE(tq->tq_pri));
+       if (spl_taskq_thread_priority)
+               set_user_nice(tqt->tqt_thread, PRIO_TO_NICE(tq->tq_pri));
+
        wake_up_process(tqt->tqt_thread);
 
        return (tqt);
@@ -928,7 +961,6 @@ taskq_create(const char *name, int nthreads, pri_t pri,
        int count = 0, rc = 0, i;
 
        ASSERT(name != NULL);
-       ASSERT(pri <= maxclsyspri);
        ASSERT(minalloc >= 0);
        ASSERT(maxalloc <= INT_MAX);
        ASSERT(!(flags & (TASKQ_CPR_SAFE))); /* Unsupported */
@@ -1070,12 +1102,12 @@ int
 spl_taskq_init(void)
 {
        system_taskq = taskq_create("spl_system_taskq", MAX(boot_ncpus, 64),
-           minclsyspri, boot_ncpus, INT_MAX, TASKQ_PREPOPULATE|TASKQ_DYNAMIC);
+           maxclsyspri, boot_ncpus, INT_MAX, TASKQ_PREPOPULATE|TASKQ_DYNAMIC);
        if (system_taskq == NULL)
                return (1);
 
        dynamic_taskq = taskq_create("spl_dynamic_taskq", 1,
-           minclsyspri, boot_ncpus, INT_MAX, TASKQ_PREPOPULATE);
+           maxclsyspri, boot_ncpus, INT_MAX, TASKQ_PREPOPULATE);
        if (dynamic_taskq == NULL) {
                taskq_destroy(system_taskq);
                return (1);