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
  30 #ifdef DEBUG_SUBSYSTEM
  31 #undef DEBUG_SUBSYSTEM
  32 #endif
  33
  34 #define DEBUG_SUBSYSTEM S_TASKQ
  35
  36 /* Global system-wide dynamic task queue available for all consumers */
  37 taskq_t *system_taskq;
  38 EXPORT_SYMBOL(system_taskq);
  39
  40 typedef struct spl_task {
  41         spinlock_t              t_lock;
  42         struct list_head        t_list;
  43         taskqid_t               t_id;
  44         task_func_t             *t_func;
  45         void                    *t_arg;
  46 } spl_task_t;
  47
  48 /*
  49  * NOTE: Must be called with tq->tq_lock held, returns a list_t which
  50  * is not attached to the free, work, or pending taskq lists.
  51  */
  52 static spl_task_t *
  53 task_alloc(taskq_t *tq, uint_t flags)
  54 {
  55         spl_task_t *t;
  56         int count = 0;
  57         ENTRY;
  58
  59         ASSERT(tq);
  60         ASSERT(flags & (TQ_SLEEP | TQ_NOSLEEP));               /* One set */
  61         ASSERT(!((flags & TQ_SLEEP) && (flags & TQ_NOSLEEP))); /* Not both */
  62         ASSERT(spin_is_locked(&tq->tq_lock));
  63 retry:
  64         /* Acquire spl_task_t's from free list if available */
  65         if (!list_empty(&tq->tq_free_list) && !(flags & TQ_NEW)) {
  66                 t = list_entry(tq->tq_free_list.next, spl_task_t, t_list);
  67                 list_del_init(&t->t_list);
  68                 RETURN(t);
  69         }
  70
  71         /* Free list is empty and memory allocations are prohibited */
  72         if (flags & TQ_NOALLOC)
  73                 RETURN(NULL);
  74
  75         /* Hit maximum spl_task_t pool size */
  76         if (tq->tq_nalloc >= tq->tq_maxalloc) {
  77                 if (flags & TQ_NOSLEEP)
  78                         RETURN(NULL);
  79
  80                 /* Sleep periodically polling the free list for an available
  81                  * spl_task_t.  If a full second passes and we have not found
  82                  * one gives up and return a NULL to the caller. */
  83                 if (flags & TQ_SLEEP) {
  84                         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
  85                         schedule_timeout(HZ / 100);
  86                         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
  87                         if (count < 100)
  88                                 GOTO(retry, count++);
  89
  90                         RETURN(NULL);
  91                 }
  92
  93                 /* Unreachable, TQ_SLEEP or TQ_NOSLEEP */
  94                 SBUG();
  95         }
  96
  97         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
  98         t = kmem_alloc(sizeof(spl_task_t), flags & (TQ_SLEEP | TQ_NOSLEEP));
  99         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 100
 101         if (t) {
 102                 spin_lock_init(&t->t_lock);
 103                 INIT_LIST_HEAD(&t->t_list);
 104                 t->t_id = 0;
 105                 t->t_func = NULL;
 106                 t->t_arg = NULL;
 107                 tq->tq_nalloc++;
 108         }
 109
 110         RETURN(t);
 111 }
 112
 113 /*
 114  * NOTE: Must be called with tq->tq_lock held, expects the spl_task_t
 115  * to already be removed from the free, work, or pending taskq lists.
 116  */
 117 static void
 118 task_free(taskq_t *tq, spl_task_t *t)
 119 {
 120         ENTRY;
 121
 122         ASSERT(tq);
 123         ASSERT(t);
 124         ASSERT(spin_is_locked(&tq->tq_lock));
 125         ASSERT(list_empty(&t->t_list));
 126
 127         kmem_free(t, sizeof(spl_task_t));
 128         tq->tq_nalloc--;
 129
 130         EXIT;
 131 }
 132
 133 /*
 134  * NOTE: Must be called with tq->tq_lock held, either destroys the
 135  * spl_task_t if too many exist or moves it to the free list for later use.
 136  */
 137 static void
 138 task_done(taskq_t *tq, spl_task_t *t)
 139 {
 140         ENTRY;
 141         ASSERT(tq);
 142         ASSERT(t);
 143         ASSERT(spin_is_locked(&tq->tq_lock));
 144
 145         list_del_init(&t->t_list);
 146
 147         if (tq->tq_nalloc <= tq->tq_minalloc) {
 148                 t->t_id = 0;
 149                 t->t_func = NULL;
 150                 t->t_arg = NULL;
 151                 list_add_tail(&t->t_list, &tq->tq_free_list);
 152         } else {
 153                 task_free(tq, t);
 154         }
 155
 156         EXIT;
 157 }
 158
 159 /*
 160  * As tasks are submitted to the task queue they are assigned a
 161  * monotonically increasing taskqid and added to the tail of the pending
 162  * list.  As worker threads become available the tasks are removed from
 163  * the head of the pending or priority list, giving preference to the
 164  * priority list.  The tasks are then added to the work list, preserving
 165  * the ordering by taskqid.  Finally, as tasks complete they are removed
 166  * from the work list.  This means that the pending and work lists are
 167  * always kept sorted by taskqid.  Thus the lowest outstanding
 168  * incomplete taskqid can be determined simply by checking the min
 169  * taskqid for each head item on the pending, priority, and work list.
 170  * This value is stored in tq->tq_lowest_id and only updated to the new
 171  * lowest id when the previous lowest id completes.  All taskqids lower
 172  * than tq->tq_lowest_id must have completed.  It is also possible
 173  * larger taskqid's have completed because they may be processed in
 174  * parallel by several worker threads.  However, this is not a problem
 175  * because the behavior of taskq_wait_id() is to block until all
 176  * previously submitted taskqid's have completed.
 177  *
 178  * XXX: Taskqid_t wrapping is not handled.  However, taskqid_t's are
 179  * 64-bit values so even if a taskq is processing 2^24 (16,777,216)
 180  * taskqid_ts per second it will still take 2^40 seconds, 34,865 years,
 181  * before the wrap occurs.  I can live with that for now.
 182  */
 183 static int
 184 taskq_wait_check(taskq_t *tq, taskqid_t id)
 185 {
 186         int rc;
 187
 188         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 189         rc = (id < tq->tq_lowest_id);
 190         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 191
 192         RETURN(rc);
 193 }
 194
 195 void
 196 __taskq_wait_id(taskq_t *tq, taskqid_t id)
 197 {
 198         ENTRY;
 199         ASSERT(tq);
 200
 201         wait_event(tq->tq_wait_waitq, taskq_wait_check(tq, id));
 202
 203         EXIT;
 204 }
 205 EXPORT_SYMBOL(__taskq_wait_id);
 206
 207 void
 208 __taskq_wait(taskq_t *tq)
 209 {
 210         taskqid_t id;
 211         ENTRY;
 212         ASSERT(tq);
 213
 214         /* Wait for the largest outstanding taskqid */
 215         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 216         id = tq->tq_next_id - 1;
 217         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 218
 219         __taskq_wait_id(tq, id);
 220
 221         EXIT;
 222
 223 }
 224 EXPORT_SYMBOL(__taskq_wait);
 225
 226 int
 227 __taskq_member(taskq_t *tq, void *t)
 228 {
 229         int i;
 230         ENTRY;
 231
 232         ASSERT(tq);
 233         ASSERT(t);
 234
 235         for (i = 0; i < tq->tq_nthreads; i++)
 236                 if (tq->tq_threads[i] == (struct task_struct *)t)
 237                         RETURN(1);
 238
 239         RETURN(0);
 240 }
 241 EXPORT_SYMBOL(__taskq_member);
 242
 243 taskqid_t
 244 __taskq_dispatch(taskq_t *tq, task_func_t func, void *arg, uint_t flags)
 245 {
 246         spl_task_t *t;
 247         taskqid_t rc = 0;
 248         ENTRY;
 249
 250         ASSERT(tq);
 251         ASSERT(func);
 252
 253         /* Solaris assumes TQ_SLEEP if not passed explicitly */
 254         if (!(flags & (TQ_SLEEP | TQ_NOSLEEP)))
 255                 flags |= TQ_SLEEP;
 256
 257         if (unlikely(in_atomic() && (flags & TQ_SLEEP))) {
 258                 CERROR("May schedule while atomic: %s/0x%08x/%d\n",
 259                        current->comm, preempt_count(), current->pid);
 260                 SBUG();
 261         }
 262
 263         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 264
 265         /* Taskq being destroyed and all tasks drained */
 266         if (!(tq->tq_flags & TQ_ACTIVE))
 267                 GOTO(out, rc = 0);
 268
 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))
 272                 GOTO(out, rc = 0);
 273
 274         if ((t = task_alloc(tq, flags)) == NULL)
 275                 GOTO(out, rc = 0);
 276
 277         spin_lock(&t->t_lock);
 278
 279         /* Queue to the priority list instead of the pending list */
 280         if (flags & TQ_FRONT)
 281                 list_add_tail(&t->t_list, &tq->tq_prio_list);
 282         else
 283                 list_add_tail(&t->t_list, &tq->tq_pend_list);
 284
 285         t->t_id = rc = tq->tq_next_id;
 286         tq->tq_next_id++;
 287         t->t_func = func;
 288         t->t_arg = arg;
 289         spin_unlock(&t->t_lock);
 290
 291         wake_up(&tq->tq_work_waitq);
 292 out:
 293         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 294         RETURN(rc);
 295 }
 296 EXPORT_SYMBOL(__taskq_dispatch);
 297
 298 /*
 299  * Returns the lowest incomplete taskqid_t.  The taskqid_t may
 300  * be queued on the pending list, on the priority list,  or on
 301  * the work list currently being handled, but it is not 100%
 302  * complete yet.
 303  */
 304 static taskqid_t
 305 taskq_lowest_id(taskq_t *tq)
 306 {
 307         taskqid_t lowest_id = tq->tq_next_id;
 308         spl_task_t *t;
 309         ENTRY;
 310
 311         ASSERT(tq);
 312         ASSERT(spin_is_locked(&tq->tq_lock));
 313
 314         if (!list_empty(&tq->tq_pend_list)) {
 315                 t = list_entry(tq->tq_pend_list.next, spl_task_t, t_list);
 316                 lowest_id = MIN(lowest_id, t->t_id);
 317         }
 318
 319         if (!list_empty(&tq->tq_prio_list)) {
 320                 t = list_entry(tq->tq_prio_list.next, spl_task_t, t_list);
 321                 lowest_id = MIN(lowest_id, t->t_id);
 322         }
 323
 324         if (!list_empty(&tq->tq_work_list)) {
 325                 t = list_entry(tq->tq_work_list.next, spl_task_t, t_list);
 326                 lowest_id = MIN(lowest_id, t->t_id);
 327         }
 328
 329         RETURN(lowest_id);
 330 }
 331
 332 /*
 333  * Insert a task into a list keeping the list sorted by increasing
 334  * taskqid.
 335  */
 336 static void
 337 taskq_insert_in_order(taskq_t *tq, spl_task_t *t)
 338 {
 339         spl_task_t *w;
 340         struct list_head *l;
 341
 342         ENTRY;
 343         ASSERT(tq);
 344         ASSERT(t);
 345         ASSERT(spin_is_locked(&tq->tq_lock));
 346
 347         list_for_each_prev(l, &tq->tq_work_list) {
 348                 w = list_entry(l, spl_task_t, t_list);
 349                 if (w->t_id < t->t_id) {
 350                         list_add(&t->t_list, l);
 351                         break;
 352                 }
 353         }
 354         if (l == &tq->tq_work_list)
 355                 list_add(&t->t_list, &tq->tq_work_list);
 356
 357         EXIT;
 358 }
 359
 360 static int
 361 taskq_thread(void *args)
 362 {
 363         DECLARE_WAITQUEUE(wait, current);
 364         sigset_t blocked;
 365         taskqid_t id;
 366         taskq_t *tq = args;
 367         spl_task_t *t;
 368         struct list_head *pend_list;
 369         ENTRY;
 370
 371         ASSERT(tq);
 372         current->flags |= PF_NOFREEZE;
 373
 374         sigfillset(&blocked);
 375         sigprocmask(SIG_BLOCK, &blocked, NULL);
 376         flush_signals(current);
 377
 378         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 379         tq->tq_nthreads++;
 380         wake_up(&tq->tq_wait_waitq);
 381         set_current_state(TASK_INTERRUPTIBLE);
 382
 383         while (!kthread_should_stop()) {
 384
 385                 add_wait_queue(&tq->tq_work_waitq, &wait);
 386                 if (list_empty(&tq->tq_pend_list) &&
 387                     list_empty(&tq->tq_prio_list)) {
 388                         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 389                         schedule();
 390                         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 391                 } else {
 392                         __set_current_state(TASK_RUNNING);
 393                 }
 394
 395                 remove_wait_queue(&tq->tq_work_waitq, &wait);
 396
 397                 if (!list_empty(&tq->tq_prio_list))
 398                         pend_list = &tq->tq_prio_list;
 399                 else if (!list_empty(&tq->tq_pend_list))
 400                         pend_list = &tq->tq_pend_list;
 401                 else
 402                         pend_list = NULL;
 403
 404                 if (pend_list) {
 405                         t = list_entry(pend_list->next, spl_task_t, t_list);
 406                         list_del_init(&t->t_list);
 407                         taskq_insert_in_order(tq, t);
 408                         tq->tq_nactive++;
 409                         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 410
 411                         /* Perform the requested task */
 412                         t->t_func(t->t_arg);
 413
 414                         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 415                         tq->tq_nactive--;
 416                         id = t->t_id;
 417                         task_done(tq, t);
 418
 419                         /* When the current lowest outstanding taskqid is
 420                          * done calculate the new lowest outstanding id */
 421                         if (tq->tq_lowest_id == id) {
 422                                 tq->tq_lowest_id = taskq_lowest_id(tq);
 423                                 ASSERT(tq->tq_lowest_id > id);
 424                         }
 425
 426                         wake_up_all(&tq->tq_wait_waitq);
 427                 }
 428
 429                 set_current_state(TASK_INTERRUPTIBLE);
 430
 431         }
 432
 433         __set_current_state(TASK_RUNNING);
 434         tq->tq_nthreads--;
 435         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 436
 437         RETURN(0);
 438 }
 439
 440 taskq_t *
 441 __taskq_create(const char *name, int nthreads, pri_t pri,
 442                int minalloc, int maxalloc, uint_t flags)
 443 {
 444         taskq_t *tq;
 445         struct task_struct *t;
 446         int rc = 0, i, j = 0;
 447         ENTRY;
 448
 449         ASSERT(name != NULL);
 450         ASSERT(pri <= maxclsyspri);
 451         ASSERT(minalloc >= 0);
 452         ASSERT(maxalloc <= INT_MAX);
 453         ASSERT(!(flags & (TASKQ_CPR_SAFE | TASKQ_DYNAMIC))); /* Unsupported */
 454
 455         /* Scale the number of threads using nthreads as a percentage */
 456         if (flags & TASKQ_THREADS_CPU_PCT) {
 457                 ASSERT(nthreads <= 100);
 458                 ASSERT(nthreads >= 0);
 459                 nthreads = MIN(nthreads, 100);
 460                 nthreads = MAX(nthreads, 0);
 461                 nthreads = MAX((num_online_cpus() * nthreads) / 100, 1);
 462         }
 463
 464         tq = kmem_alloc(sizeof(*tq), KM_SLEEP);
 465         if (tq == NULL)
 466                 RETURN(NULL);
 467
 468         tq->tq_threads = kmem_alloc(nthreads * sizeof(t), KM_SLEEP);
 469         if (tq->tq_threads == NULL) {
 470                 kmem_free(tq, sizeof(*tq));
 471                 RETURN(NULL);
 472         }
 473
 474         spin_lock_init(&tq->tq_lock);
 475         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 476         tq->tq_name      = name;
 477         tq->tq_nactive   = 0;
 478         tq->tq_nthreads  = 0;
 479         tq->tq_pri       = pri;
 480         tq->tq_minalloc  = minalloc;
 481         tq->tq_maxalloc  = maxalloc;
 482         tq->tq_nalloc    = 0;
 483         tq->tq_flags     = (flags | TQ_ACTIVE);
 484         tq->tq_next_id   = 1;
 485         tq->tq_lowest_id = 1;
 486         INIT_LIST_HEAD(&tq->tq_free_list);
 487         INIT_LIST_HEAD(&tq->tq_work_list);
 488         INIT_LIST_HEAD(&tq->tq_pend_list);
 489         INIT_LIST_HEAD(&tq->tq_prio_list);
 490         init_waitqueue_head(&tq->tq_work_waitq);
 491         init_waitqueue_head(&tq->tq_wait_waitq);
 492
 493         if (flags & TASKQ_PREPOPULATE)
 494                 for (i = 0; i < minalloc; i++)
 495                         task_done(tq, task_alloc(tq, TQ_SLEEP | TQ_NEW));
 496
 497         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 498
 499         for (i = 0; i < nthreads; i++) {
 500                 t = kthread_create(taskq_thread, tq, "%s/%d", name, i);
 501                 if (t) {
 502                         tq->tq_threads[i] = t;
 503                         kthread_bind(t, i % num_online_cpus());
 504                         set_user_nice(t, PRIO_TO_NICE(pri));
 505                         wake_up_process(t);
 506                         j++;
 507                 } else {
 508                         tq->tq_threads[i] = NULL;
 509                         rc = 1;
 510                 }
 511         }
 512
 513         /* Wait for all threads to be started before potential destroy */
 514         wait_event(tq->tq_wait_waitq, tq->tq_nthreads == j);
 515
 516         if (rc) {
 517                 __taskq_destroy(tq);
 518                 tq = NULL;
 519         }
 520
 521         RETURN(tq);
 522 }
 523 EXPORT_SYMBOL(__taskq_create);
 524
 525 void
 526 __taskq_destroy(taskq_t *tq)
 527 {
 528         spl_task_t *t;
 529         int i, nthreads;
 530         ENTRY;
 531
 532         ASSERT(tq);
 533         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 534         tq->tq_flags &= ~TQ_ACTIVE;
 535         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 536
 537         /* TQ_ACTIVE cleared prevents new tasks being added to pending */
 538         __taskq_wait(tq);
 539
 540         nthreads = tq->tq_nthreads;
 541         for (i = 0; i < nthreads; i++)
 542                 if (tq->tq_threads[i])
 543                         kthread_stop(tq->tq_threads[i]);
 544
 545         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 546
 547         while (!list_empty(&tq->tq_free_list)) {
 548                 t = list_entry(tq->tq_free_list.next, spl_task_t, t_list);
 549                 list_del_init(&t->t_list);
 550                 task_free(tq, t);
 551         }
 552
 553         ASSERT(tq->tq_nthreads == 0);
 554         ASSERT(tq->tq_nalloc == 0);
 555         ASSERT(list_empty(&tq->tq_free_list));
 556         ASSERT(list_empty(&tq->tq_work_list));
 557         ASSERT(list_empty(&tq->tq_pend_list));
 558         ASSERT(list_empty(&tq->tq_prio_list));
 559
 560         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 561         kmem_free(tq->tq_threads, nthreads * sizeof(spl_task_t *));
 562         kmem_free(tq, sizeof(taskq_t));
 563
 564         EXIT;
 565 }
 566 EXPORT_SYMBOL(__taskq_destroy);
 567
 568 int
 569 spl_taskq_init(void)
 570 {
 571         ENTRY;
 572
 573         /* Solaris creates a dynamic taskq of up to 64 threads, however in
 574          * a Linux environment 1 thread per-core is usually about right */
 575         system_taskq = taskq_create("spl_system_taskq", num_online_cpus(),
 576                                     minclsyspri, 4, 512, TASKQ_PREPOPULATE);
 577         if (system_taskq == NULL)
 578                 RETURN(1);
 579
 580         RETURN(0);
 581 }
 582
 583 void
 584 spl_taskq_fini(void)
 585 {
 586         ENTRY;
 587         taskq_destroy(system_taskq);
 588         EXIT;
 589 }