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>.
   6  *  UCRL-CODE-235197
   7  *
   8  *  This file is part of the SPL, Solaris Porting Layer.
   9  *  For details, see <http://github.com/behlendorf/spl/>.
  10  *
  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.
  15  *
  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
  19  *  for more details.
  20  *
  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 \*****************************************************************************/
  26
  27 #include <sys/taskq.h>
  28 #include <sys/kmem.h>
  29 #include <spl-debug.h>
  30
  31 #ifdef SS_DEBUG_SUBSYS
  32 #undef SS_DEBUG_SUBSYS
  33 #endif
  34
  35 #define SS_DEBUG_SUBSYS SS_TASKQ
  36
  37 /* Global system-wide dynamic task queue available for all consumers */
  38 taskq_t *system_taskq;
  39 EXPORT_SYMBOL(system_taskq);
  40
  41 /*
  42  * NOTE: Must be called with tq->tq_lock held, returns a list_t which
  43  * is not attached to the free, work, or pending taskq lists.
  44  */
  45 static taskq_ent_t *
  46 task_alloc(taskq_t *tq, uint_t flags)
  47 {
  48         taskq_ent_t *t;
  49         int count = 0;
  50         SENTRY;
  51
  52         ASSERT(tq);
  53         ASSERT(flags & (TQ_SLEEP | TQ_NOSLEEP));               /* One set */
  54         ASSERT(!((flags & TQ_SLEEP) && (flags & TQ_NOSLEEP))); /* Not both */
  55         ASSERT(spin_is_locked(&tq->tq_lock));
  56 retry:
  57         /* Acquire taskq_ent_t's from free list if available */
  58         if (!list_empty(&tq->tq_free_list) && !(flags & TQ_NEW)) {
  59                 t = list_entry(tq->tq_free_list.next, taskq_ent_t, tqent_list);
  60
  61                 ASSERT(!(t->tqent_flags & TQENT_FLAG_PREALLOC));
  62
  63                 list_del_init(&t->tqent_list);
  64                 SRETURN(t);
  65         }
  66
  67         /* Free list is empty and memory allocations are prohibited */
  68         if (flags & TQ_NOALLOC)
  69                 SRETURN(NULL);
  70
  71         /* Hit maximum taskq_ent_t pool size */
  72         if (tq->tq_nalloc >= tq->tq_maxalloc) {
  73                 if (flags & TQ_NOSLEEP)
  74                         SRETURN(NULL);
  75
  76                 /*
  77                  * Sleep periodically polling the free list for an available
  78                  * taskq_ent_t. Dispatching with TQ_SLEEP should always succeed
  79                  * but we cannot block forever waiting for an taskq_entq_t to
  80                  * show up in the free list, otherwise a deadlock can happen.
  81                  *
  82                  * Therefore, we need to allocate a new task even if the number
  83                  * of allocated tasks is above tq->tq_maxalloc, but we still
  84                  * end up delaying the task allocation by one second, thereby
  85                  * throttling the task dispatch rate.
  86                  */
  87                  spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
  88                  schedule_timeout(HZ / 100);
  89                  spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
  90                  if (count < 100)
  91                         SGOTO(retry, count++);
  92         }
  93
  94         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
  95         t = kmem_alloc(sizeof(taskq_ent_t), flags & (TQ_SLEEP | TQ_NOSLEEP));
  96         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
  97
  98         if (t) {
  99                 taskq_init_ent(t);
 100                 tq->tq_nalloc++;
 101         }
 102
 103         SRETURN(t);
 104 }
 105
 106 /*
 107  * NOTE: Must be called with tq->tq_lock held, expects the taskq_ent_t
 108  * to already be removed from the free, work, or pending taskq lists.
 109  */
 110 static void
 111 task_free(taskq_t *tq, taskq_ent_t *t)
 112 {
 113         SENTRY;
 114
 115         ASSERT(tq);
 116         ASSERT(t);
 117         ASSERT(spin_is_locked(&tq->tq_lock));
 118         ASSERT(list_empty(&t->tqent_list));
 119
 120         kmem_free(t, sizeof(taskq_ent_t));
 121         tq->tq_nalloc--;
 122
 123         SEXIT;
 124 }
 125
 126 /*
 127  * NOTE: Must be called with tq->tq_lock held, either destroys the
 128  * taskq_ent_t if too many exist or moves it to the free list for later use.
 129  */
 130 static void
 131 task_done(taskq_t *tq, taskq_ent_t *t)
 132 {
 133         SENTRY;
 134         ASSERT(tq);
 135         ASSERT(t);
 136         ASSERT(spin_is_locked(&tq->tq_lock));
 137
 138         list_del_init(&t->tqent_list);
 139
 140         if (tq->tq_nalloc <= tq->tq_minalloc) {
 141                 t->tqent_id = 0;
 142                 t->tqent_func = NULL;
 143                 t->tqent_arg = NULL;
 144                 t->tqent_flags = 0;
 145
 146                 list_add_tail(&t->tqent_list, &tq->tq_free_list);
 147         } else {
 148                 task_free(tq, t);
 149         }
 150
 151         SEXIT;
 152 }
 153
 154 /*
 155  * As tasks are submitted to the task queue they are assigned a
 156  * monotonically increasing taskqid and added to the tail of the pending
 157  * list.  As worker threads become available the tasks are removed from
 158  * the head of the pending or priority list, giving preference to the
 159  * priority list.  The tasks are then added to the work list, preserving
 160  * the ordering by taskqid.  Finally, as tasks complete they are removed
 161  * from the work list.  This means that the pending and work lists are
 162  * always kept sorted by taskqid.  Thus the lowest outstanding
 163  * incomplete taskqid can be determined simply by checking the min
 164  * taskqid for each head item on the pending, priority, and work list.
 165  * This value is stored in tq->tq_lowest_id and only updated to the new
 166  * lowest id when the previous lowest id completes.  All taskqids lower
 167  * than tq->tq_lowest_id must have completed.  It is also possible
 168  * larger taskqid's have completed because they may be processed in
 169  * parallel by several worker threads.  However, this is not a problem
 170  * because the behavior of taskq_wait_id() is to block until all
 171  * previously submitted taskqid's have completed.
 172  *
 173  * XXX: Taskqid_t wrapping is not handled.  However, taskqid_t's are
 174  * 64-bit values so even if a taskq is processing 2^24 (16,777,216)
 175  * taskqid_ts per second it will still take 2^40 seconds, 34,865 years,
 176  * before the wrap occurs.  I can live with that for now.
 177  */
 178 static int
 179 taskq_wait_check(taskq_t *tq, taskqid_t id)
 180 {
 181         int rc;
 182
 183         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 184         rc = (id < tq->tq_lowest_id);
 185         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 186
 187         SRETURN(rc);
 188 }
 189
 190 void
 191 __taskq_wait_id(taskq_t *tq, taskqid_t id)
 192 {
 193         SENTRY;
 194         ASSERT(tq);
 195
 196         wait_event(tq->tq_wait_waitq, taskq_wait_check(tq, id));
 197
 198         SEXIT;
 199 }
 200 EXPORT_SYMBOL(__taskq_wait_id);
 201
 202 void
 203 __taskq_wait(taskq_t *tq)
 204 {
 205         taskqid_t id;
 206         SENTRY;
 207         ASSERT(tq);
 208
 209         /* Wait for the largest outstanding taskqid */
 210         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 211         id = tq->tq_next_id - 1;
 212         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 213
 214         __taskq_wait_id(tq, id);
 215
 216         SEXIT;
 217
 218 }
 219 EXPORT_SYMBOL(__taskq_wait);
 220
 221 int
 222 __taskq_member(taskq_t *tq, void *t)
 223 {
 224         struct list_head *l;
 225         taskq_thread_t *tqt;
 226         SENTRY;
 227
 228         ASSERT(tq);
 229         ASSERT(t);
 230
 231         list_for_each(l, &tq->tq_thread_list) {
 232                 tqt = list_entry(l, taskq_thread_t, tqt_thread_list);
 233                 if (tqt->tqt_thread == (struct task_struct *)t)
 234                         SRETURN(1);
 235         }
 236
 237         SRETURN(0);
 238 }
 239 EXPORT_SYMBOL(__taskq_member);
 240
 241 taskqid_t
 242 __taskq_dispatch(taskq_t *tq, task_func_t func, void *arg, uint_t flags)
 243 {
 244         taskq_ent_t *t;
 245         taskqid_t rc = 0;
 246         SENTRY;
 247
 248         ASSERT(tq);
 249         ASSERT(func);
 250
 251         /* Solaris assumes TQ_SLEEP if not passed explicitly */
 252         if (!(flags & (TQ_SLEEP | TQ_NOSLEEP)))
 253                 flags |= TQ_SLEEP;
 254
 255         if (unlikely(in_atomic() && (flags & TQ_SLEEP)))
 256                 PANIC("May schedule while atomic: %s/0x%08x/%d\n",
 257                     current->comm, preempt_count(), current->pid);
 258
 259         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 260
 261         /* Taskq being destroyed and all tasks drained */
 262         if (!(tq->tq_flags & TQ_ACTIVE))
 263                 SGOTO(out, rc = 0);
 264
 265         /* Do not queue the task unless there is idle thread for it */
 266         ASSERT(tq->tq_nactive <= tq->tq_nthreads);
 267         if ((flags & TQ_NOQUEUE) && (tq->tq_nactive == tq->tq_nthreads))
 268                 SGOTO(out, rc = 0);
 269
 270         if ((t = task_alloc(tq, flags)) == NULL)
 271                 SGOTO(out, rc = 0);
 272
 273         spin_lock(&t->tqent_lock);
 274
 275         /* Queue to the priority list instead of the pending list */
 276         if (flags & TQ_FRONT)
 277                 list_add_tail(&t->tqent_list, &tq->tq_prio_list);
 278         else
 279                 list_add_tail(&t->tqent_list, &tq->tq_pend_list);
 280
 281         t->tqent_id = rc = tq->tq_next_id;
 282         tq->tq_next_id++;
 283         t->tqent_func = func;
 284         t->tqent_arg = arg;
 285
 286         ASSERT(!(t->tqent_flags & TQENT_FLAG_PREALLOC));
 287
 288         spin_unlock(&t->tqent_lock);
 289
 290         wake_up(&tq->tq_work_waitq);
 291 out:
 292         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 293         SRETURN(rc);
 294 }
 295 EXPORT_SYMBOL(__taskq_dispatch);
 296
 297 void
 298 __taskq_dispatch_ent(taskq_t *tq, task_func_t func, void *arg, uint_t flags,
 299    taskq_ent_t *t)
 300 {
 301         SENTRY;
 302
 303         ASSERT(tq);
 304         ASSERT(func);
 305         ASSERT(!(tq->tq_flags & TASKQ_DYNAMIC));
 306
 307         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 308
 309         /* Taskq being destroyed and all tasks drained */
 310         if (!(tq->tq_flags & TQ_ACTIVE)) {
 311                 t->tqent_id = 0;
 312                 goto out;
 313         }
 314
 315         spin_lock(&t->tqent_lock);
 316
 317         /*
 318          * Mark it as a prealloc'd task.  This is important
 319          * to ensure that we don't free it later.
 320          */
 321         t->tqent_flags |= TQENT_FLAG_PREALLOC;
 322
 323         /* Queue to the priority list instead of the pending list */
 324         if (flags & TQ_FRONT)
 325                 list_add_tail(&t->tqent_list, &tq->tq_prio_list);
 326         else
 327                 list_add_tail(&t->tqent_list, &tq->tq_pend_list);
 328
 329         t->tqent_id = tq->tq_next_id;
 330         tq->tq_next_id++;
 331         t->tqent_func = func;
 332         t->tqent_arg = arg;
 333
 334         spin_unlock(&t->tqent_lock);
 335
 336         wake_up(&tq->tq_work_waitq);
 337 out:
 338         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 339         SEXIT;
 340 }
 341 EXPORT_SYMBOL(__taskq_dispatch_ent);
 342
 343 int
 344 __taskq_empty_ent(taskq_ent_t *t)
 345 {
 346         return list_empty(&t->tqent_list);
 347 }
 348 EXPORT_SYMBOL(__taskq_empty_ent);
 349
 350 void
 351 __taskq_init_ent(taskq_ent_t *t)
 352 {
 353         spin_lock_init(&t->tqent_lock);
 354         INIT_LIST_HEAD(&t->tqent_list);
 355         t->tqent_id = 0;
 356         t->tqent_func = NULL;
 357         t->tqent_arg = NULL;
 358         t->tqent_flags = 0;
 359 }
 360 EXPORT_SYMBOL(__taskq_init_ent);
 361
 362 /*
 363  * Returns the lowest incomplete taskqid_t.  The taskqid_t may
 364  * be queued on the pending list, on the priority list,  or on
 365  * the work list currently being handled, but it is not 100%
 366  * complete yet.
 367  */
 368 static taskqid_t
 369 taskq_lowest_id(taskq_t *tq)
 370 {
 371         taskqid_t lowest_id = tq->tq_next_id;
 372         taskq_ent_t *t;
 373         taskq_thread_t *tqt;
 374         SENTRY;
 375
 376         ASSERT(tq);
 377         ASSERT(spin_is_locked(&tq->tq_lock));
 378
 379         if (!list_empty(&tq->tq_pend_list)) {
 380                 t = list_entry(tq->tq_pend_list.next, taskq_ent_t, tqent_list);
 381                 lowest_id = MIN(lowest_id, t->tqent_id);
 382         }
 383
 384         if (!list_empty(&tq->tq_prio_list)) {
 385                 t = list_entry(tq->tq_prio_list.next, taskq_ent_t, tqent_list);
 386                 lowest_id = MIN(lowest_id, t->tqent_id);
 387         }
 388
 389         if (!list_empty(&tq->tq_active_list)) {
 390                 tqt = list_entry(tq->tq_active_list.next, taskq_thread_t,
 391                                  tqt_active_list);
 392                 ASSERT(tqt->tqt_id != 0);
 393                 lowest_id = MIN(lowest_id, tqt->tqt_id);
 394         }
 395
 396         SRETURN(lowest_id);
 397 }
 398
 399 /*
 400  * Insert a task into a list keeping the list sorted by increasing
 401  * taskqid.
 402  */
 403 static void
 404 taskq_insert_in_order(taskq_t *tq, taskq_thread_t *tqt)
 405 {
 406         taskq_thread_t *w;
 407         struct list_head *l;
 408
 409         SENTRY;
 410         ASSERT(tq);
 411         ASSERT(tqt);
 412         ASSERT(spin_is_locked(&tq->tq_lock));
 413
 414         list_for_each_prev(l, &tq->tq_active_list) {
 415                 w = list_entry(l, taskq_thread_t, tqt_active_list);
 416                 if (w->tqt_id < tqt->tqt_id) {
 417                         list_add(&tqt->tqt_active_list, l);
 418                         break;
 419                 }
 420         }
 421         if (l == &tq->tq_active_list)
 422                 list_add(&tqt->tqt_active_list, &tq->tq_active_list);
 423
 424         SEXIT;
 425 }
 426
 427 static int
 428 taskq_thread(void *args)
 429 {
 430         DECLARE_WAITQUEUE(wait, current);
 431         sigset_t blocked;
 432         taskq_thread_t *tqt = args;
 433         taskq_t *tq;
 434         taskq_ent_t *t;
 435         struct list_head *pend_list;
 436         SENTRY;
 437
 438         ASSERT(tqt);
 439         tq = tqt->tqt_tq;
 440         current->flags |= PF_NOFREEZE;
 441
 442         /* Disable the direct memory reclaim path */
 443         if (tq->tq_flags & TASKQ_NORECLAIM)
 444                 current->flags |= PF_MEMALLOC;
 445
 446         sigfillset(&blocked);
 447         sigprocmask(SIG_BLOCK, &blocked, NULL);
 448         flush_signals(current);
 449
 450         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 451         tq->tq_nthreads++;
 452         wake_up(&tq->tq_wait_waitq);
 453         set_current_state(TASK_INTERRUPTIBLE);
 454
 455         while (!kthread_should_stop()) {
 456
 457                 add_wait_queue(&tq->tq_work_waitq, &wait);
 458                 if (list_empty(&tq->tq_pend_list) &&
 459                     list_empty(&tq->tq_prio_list)) {
 460                         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 461                         schedule();
 462                         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 463                 } else {
 464                         __set_current_state(TASK_RUNNING);
 465                 }
 466
 467                 remove_wait_queue(&tq->tq_work_waitq, &wait);
 468
 469                 if (!list_empty(&tq->tq_prio_list))
 470                         pend_list = &tq->tq_prio_list;
 471                 else if (!list_empty(&tq->tq_pend_list))
 472                         pend_list = &tq->tq_pend_list;
 473                 else
 474                         pend_list = NULL;
 475
 476                 if (pend_list) {
 477                         t = list_entry(pend_list->next, taskq_ent_t, tqent_list);
 478                         list_del_init(&t->tqent_list);
 479
 480                         /* In order to support recursively dispatching a
 481                          * preallocated taskq_ent_t, tqent_id must be
 482                          * stored prior to executing tqent_func. */
 483                         tqt->tqt_id = t->tqent_id;
 484
 485                         /* We must store a copy of the flags prior to
 486                          * servicing the task (servicing a prealloc'd task
 487                          * returns the ownership of the tqent back to
 488                          * the caller of taskq_dispatch). Thus,
 489                          * tqent_flags _may_ change within the call. */
 490                         tqt->tqt_flags = t->tqent_flags;
 491
 492                         taskq_insert_in_order(tq, tqt);
 493                         tq->tq_nactive++;
 494                         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 495
 496                         /* Perform the requested task */
 497                         t->tqent_func(t->tqent_arg);
 498
 499                         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 500                         tq->tq_nactive--;
 501                         list_del_init(&tqt->tqt_active_list);
 502
 503                         /* For prealloc'd tasks, we don't free anything. */
 504                         if ((tq->tq_flags & TASKQ_DYNAMIC) ||
 505                             !(tqt->tqt_flags & TQENT_FLAG_PREALLOC))
 506                                 task_done(tq, t);
 507
 508                         /* When the current lowest outstanding taskqid is
 509                          * done calculate the new lowest outstanding id */
 510                         if (tq->tq_lowest_id == tqt->tqt_id) {
 511                                 tq->tq_lowest_id = taskq_lowest_id(tq);
 512                                 ASSERT3S(tq->tq_lowest_id, >, tqt->tqt_id);
 513                         }
 514
 515                         tqt->tqt_id = 0;
 516                         tqt->tqt_flags = 0;
 517                         wake_up_all(&tq->tq_wait_waitq);
 518                 }
 519
 520                 set_current_state(TASK_INTERRUPTIBLE);
 521
 522         }
 523
 524         __set_current_state(TASK_RUNNING);
 525         tq->tq_nthreads--;
 526         list_del_init(&tqt->tqt_thread_list);
 527         kmem_free(tqt, sizeof(taskq_thread_t));
 528
 529         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 530
 531         SRETURN(0);
 532 }
 533
 534 taskq_t *
 535 __taskq_create(const char *name, int nthreads, pri_t pri,
 536                int minalloc, int maxalloc, uint_t flags)
 537 {
 538         taskq_t *tq;
 539         taskq_thread_t *tqt;
 540         int rc = 0, i, j = 0;
 541         SENTRY;
 542
 543         ASSERT(name != NULL);
 544         ASSERT(pri <= maxclsyspri);
 545         ASSERT(minalloc >= 0);
 546         ASSERT(maxalloc <= INT_MAX);
 547         ASSERT(!(flags & (TASKQ_CPR_SAFE | TASKQ_DYNAMIC))); /* Unsupported */
 548
 549         /* Scale the number of threads using nthreads as a percentage */
 550         if (flags & TASKQ_THREADS_CPU_PCT) {
 551                 ASSERT(nthreads <= 100);
 552                 ASSERT(nthreads >= 0);
 553                 nthreads = MIN(nthreads, 100);
 554                 nthreads = MAX(nthreads, 0);
 555                 nthreads = MAX((num_online_cpus() * nthreads) / 100, 1);
 556         }
 557
 558         tq = kmem_alloc(sizeof(*tq), KM_SLEEP);
 559         if (tq == NULL)
 560                 SRETURN(NULL);
 561
 562         spin_lock_init(&tq->tq_lock);
 563         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 564         INIT_LIST_HEAD(&tq->tq_thread_list);
 565         INIT_LIST_HEAD(&tq->tq_active_list);
 566         tq->tq_name      = name;
 567         tq->tq_nactive   = 0;
 568         tq->tq_nthreads  = 0;
 569         tq->tq_pri       = pri;
 570         tq->tq_minalloc  = minalloc;
 571         tq->tq_maxalloc  = maxalloc;
 572         tq->tq_nalloc    = 0;
 573         tq->tq_flags     = (flags | TQ_ACTIVE);
 574         tq->tq_next_id   = 1;
 575         tq->tq_lowest_id = 1;
 576         INIT_LIST_HEAD(&tq->tq_free_list);
 577         INIT_LIST_HEAD(&tq->tq_pend_list);
 578         INIT_LIST_HEAD(&tq->tq_prio_list);
 579         init_waitqueue_head(&tq->tq_work_waitq);
 580         init_waitqueue_head(&tq->tq_wait_waitq);
 581
 582         if (flags & TASKQ_PREPOPULATE)
 583                 for (i = 0; i < minalloc; i++)
 584                         task_done(tq, task_alloc(tq, TQ_SLEEP | TQ_NEW));
 585
 586         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 587
 588         for (i = 0; i < nthreads; i++) {
 589                 tqt = kmem_alloc(sizeof(*tqt), KM_SLEEP);
 590                 INIT_LIST_HEAD(&tqt->tqt_thread_list);
 591                 INIT_LIST_HEAD(&tqt->tqt_active_list);
 592                 tqt->tqt_tq = tq;
 593                 tqt->tqt_id = 0;
 594
 595                 tqt->tqt_thread = kthread_create(taskq_thread, tqt,
 596                                                  "%s/%d", name, i);
 597                 if (tqt->tqt_thread) {
 598                         list_add(&tqt->tqt_thread_list, &tq->tq_thread_list);
 599                         kthread_bind(tqt->tqt_thread, i % num_online_cpus());
 600                         set_user_nice(tqt->tqt_thread, PRIO_TO_NICE(pri));
 601                         wake_up_process(tqt->tqt_thread);
 602                         j++;
 603                 } else {
 604                         kmem_free(tqt, sizeof(taskq_thread_t));
 605                         rc = 1;
 606                 }
 607         }
 608
 609         /* Wait for all threads to be started before potential destroy */
 610         wait_event(tq->tq_wait_waitq, tq->tq_nthreads == j);
 611
 612         if (rc) {
 613                 __taskq_destroy(tq);
 614                 tq = NULL;
 615         }
 616
 617         SRETURN(tq);
 618 }
 619 EXPORT_SYMBOL(__taskq_create);
 620
 621 void
 622 __taskq_destroy(taskq_t *tq)
 623 {
 624         struct task_struct *thread;
 625         taskq_thread_t *tqt;
 626         taskq_ent_t *t;
 627         SENTRY;
 628
 629         ASSERT(tq);
 630         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 631         tq->tq_flags &= ~TQ_ACTIVE;
 632         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 633
 634         /* TQ_ACTIVE cleared prevents new tasks being added to pending */
 635         __taskq_wait(tq);
 636
 637         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 638
 639         /*
 640          * Signal each thread to exit and block until it does.  Each thread
 641          * is responsible for removing itself from the list and freeing its
 642          * taskq_thread_t.  This allows for idle threads to opt to remove
 643          * themselves from the taskq.  They can be recreated as needed.
 644          */
 645         while (!list_empty(&tq->tq_thread_list)) {
 646                 tqt = list_entry(tq->tq_thread_list.next,
 647                                  taskq_thread_t, tqt_thread_list);
 648                 thread = tqt->tqt_thread;
 649                 spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 650
 651                 kthread_stop(thread);
 652
 653                 spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 654         }
 655
 656         while (!list_empty(&tq->tq_free_list)) {
 657                 t = list_entry(tq->tq_free_list.next, taskq_ent_t, tqent_list);
 658
 659                 ASSERT(!(t->tqent_flags & TQENT_FLAG_PREALLOC));
 660
 661                 list_del_init(&t->tqent_list);
 662                 task_free(tq, t);
 663         }
 664
 665         ASSERT(tq->tq_nthreads == 0);
 666         ASSERT(tq->tq_nalloc == 0);
 667         ASSERT(list_empty(&tq->tq_thread_list));
 668         ASSERT(list_empty(&tq->tq_active_list));
 669         ASSERT(list_empty(&tq->tq_free_list));
 670         ASSERT(list_empty(&tq->tq_pend_list));
 671         ASSERT(list_empty(&tq->tq_prio_list));
 672
 673         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 674
 675         kmem_free(tq, sizeof(taskq_t));
 676
 677         SEXIT;
 678 }
 679 EXPORT_SYMBOL(__taskq_destroy);
 680
 681 int
 682 spl_taskq_init(void)
 683 {
 684         SENTRY;
 685
 686         /* Solaris creates a dynamic taskq of up to 64 threads, however in
 687          * a Linux environment 1 thread per-core is usually about right */
 688         system_taskq = taskq_create("spl_system_taskq", num_online_cpus(),
 689                                     minclsyspri, 4, 512, TASKQ_PREPOPULATE);
 690         if (system_taskq == NULL)
 691                 SRETURN(1);
 692
 693         SRETURN(0);
 694 }
 695
 696 void
 697 spl_taskq_fini(void)
 698 {
 699         SENTRY;
 700         taskq_destroy(system_taskq);
 701         SEXIT;
 702 }