module/spl/spl-taskq.c

   1 /*
   2  *  This file is part of the SPL: Solaris Porting Layer.
   3  *
   4  *  Copyright (c) 2008 Lawrence Livermore National Security, LLC.
   5  *  Produced at Lawrence Livermore National Laboratory
   6  *  Written by:
   7  *          Brian Behlendorf <behlendorf1@llnl.gov>,
   8  *          Herb Wartens <wartens2@llnl.gov>,
   9  *          Jim Garlick <garlick@llnl.gov>
  10  *  UCRL-CODE-235197
  11  *
  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.
  16  *
  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
  20  *  for more details.
  21  *
  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.
  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 /* 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.
  50  */
  51 static spl_task_t *
  52 task_alloc(taskq_t *tq, uint_t flags)
  53 {
  54         spl_task_t *t;
  55         int count = 0;
  56         ENTRY;
  57
  58         ASSERT(tq);
  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));
  62 retry:
  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);
  67                 RETURN(t);
  68         }
  69
  70         /* Free list is empty and memory allocs are prohibited */
  71         if (flags & TQ_NOALLOC)
  72                 RETURN(NULL);
  73
  74         /* Hit maximum spl_task_t pool size */
  75         if (tq->tq_nalloc >= tq->tq_maxalloc) {
  76                 if (flags & TQ_NOSLEEP)
  77                         RETURN(NULL);
  78
  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);
  86                         if (count < 100)
  87                                 GOTO(retry, count++);
  88
  89                         RETURN(NULL);
  90                 }
  91
  92                 /* Unreachable, TQ_SLEEP xor TQ_NOSLEEP */
  93                 SBUG();
  94         }
  95
  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);
  99
 100         if (t) {
 101                 spin_lock_init(&t->t_lock);
 102                 INIT_LIST_HEAD(&t->t_list);
 103                 t->t_id = 0;
 104                 t->t_func = NULL;
 105                 t->t_arg = NULL;
 106                 tq->tq_nalloc++;
 107         }
 108
 109         RETURN(t);
 110 }
 111
 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.
 114  */
 115 static void
 116 task_free(taskq_t *tq, spl_task_t *t)
 117 {
 118         ENTRY;
 119
 120         ASSERT(tq);
 121         ASSERT(t);
 122         ASSERT(spin_is_locked(&tq->tq_lock));
 123         ASSERT(list_empty(&t->t_list));
 124
 125         kmem_free(t, sizeof(spl_task_t));
 126         tq->tq_nalloc--;
 127
 128         EXIT;
 129 }
 130
 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.
 133  */
 134 static void
 135 task_done(taskq_t *tq, spl_task_t *t)
 136 {
 137         ENTRY;
 138         ASSERT(tq);
 139         ASSERT(t);
 140         ASSERT(spin_is_locked(&tq->tq_lock));
 141
 142         list_del_init(&t->t_list);
 143
 144         if (tq->tq_nalloc <= tq->tq_minalloc) {
 145                 t->t_id = 0;
 146                 t->t_func = NULL;
 147                 t->t_arg = NULL;
 148                 list_add_tail(&t->t_list, &tq->tq_free_list);
 149         } else {
 150                 task_free(tq, t);
 151         }
 152
 153         EXIT;
 154 }
 155
 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.
 163  */
 164 static int
 165 taskq_wait_check(taskq_t *tq, taskqid_t id)
 166 {
 167         RETURN(tq->tq_lowest_id >= id);
 168 }
 169
 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. */
 173 void
 174 __taskq_wait_id(taskq_t *tq, taskqid_t id)
 175 {
 176         ENTRY;
 177         ASSERT(tq);
 178
 179         wait_event(tq->tq_wait_waitq, taskq_wait_check(tq, id));
 180
 181         EXIT;
 182 }
 183 EXPORT_SYMBOL(__taskq_wait_id);
 184
 185 void
 186 __taskq_wait(taskq_t *tq)
 187 {
 188         taskqid_t id;
 189         ENTRY;
 190         ASSERT(tq);
 191
 192         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 193         id = tq->tq_next_id;
 194         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 195
 196         __taskq_wait_id(tq, id);
 197
 198         EXIT;
 199
 200 }
 201 EXPORT_SYMBOL(__taskq_wait);
 202
 203 int
 204 __taskq_member(taskq_t *tq, void *t)
 205 {
 206         int i;
 207         ENTRY;
 208
 209         ASSERT(tq);
 210         ASSERT(t);
 211
 212         for (i = 0; i < tq->tq_nthreads; i++)
 213                 if (tq->tq_threads[i] == (struct task_struct *)t)
 214                         RETURN(1);
 215
 216         RETURN(0);
 217 }
 218 EXPORT_SYMBOL(__taskq_member);
 219
 220 taskqid_t
 221 __taskq_dispatch(taskq_t *tq, task_func_t func, void *arg, uint_t flags)
 222 {
 223         spl_task_t *t;
 224         taskqid_t rc = 0;
 225         ENTRY;
 226
 227         ASSERT(tq);
 228         ASSERT(func);
 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);
 232                 SBUG();
 233         }
 234
 235         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 236
 237         /* Taskq being destroyed and all tasks drained */
 238         if (!(tq->tq_flags & TQ_ACTIVE))
 239                 GOTO(out, rc = 0);
 240
 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))
 244                 GOTO(out, rc = 0);
 245
 246         if ((t = task_alloc(tq, flags)) == NULL)
 247                 GOTO(out, rc = 0);
 248
 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;
 252         tq->tq_next_id++;
 253         t->t_func = func;
 254         t->t_arg = arg;
 255         spin_unlock(&t->t_lock);
 256
 257         wake_up(&tq->tq_work_waitq);
 258 out:
 259         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 260         RETURN(rc);
 261 }
 262 EXPORT_SYMBOL(__taskq_dispatch);
 263
 264 /* NOTE: Must be called with tq->tq_lock held */
 265 static taskqid_t
 266 taskq_lowest_id(taskq_t *tq)
 267 {
 268         taskqid_t lowest_id = ~0;
 269         spl_task_t *t;
 270         ENTRY;
 271
 272         ASSERT(tq);
 273         ASSERT(spin_is_locked(&tq->tq_lock));
 274
 275         list_for_each_entry(t, &tq->tq_pend_list, t_list)
 276                 if (t->t_id < lowest_id)
 277                         lowest_id = t->t_id;
 278
 279         list_for_each_entry(t, &tq->tq_work_list, t_list)
 280                 if (t->t_id < lowest_id)
 281                         lowest_id = t->t_id;
 282
 283         RETURN(lowest_id);
 284 }
 285
 286 static int
 287 taskq_thread(void *args)
 288 {
 289         DECLARE_WAITQUEUE(wait, current);
 290         sigset_t blocked;
 291         taskqid_t id;
 292         taskq_t *tq = args;
 293         spl_task_t *t;
 294         ENTRY;
 295
 296         ASSERT(tq);
 297         current->flags |= PF_NOFREEZE;
 298
 299         sigfillset(&blocked);
 300         sigprocmask(SIG_BLOCK, &blocked, NULL);
 301         flush_signals(current);
 302
 303         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 304         tq->tq_nthreads++;
 305         wake_up(&tq->tq_wait_waitq);
 306         set_current_state(TASK_INTERRUPTIBLE);
 307
 308         while (!kthread_should_stop()) {
 309
 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);
 313                         schedule();
 314                         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 315                 } else {
 316                         __set_current_state(TASK_RUNNING);
 317                 }
 318
 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);
 324                         tq->tq_nactive++;
 325                         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 326
 327                         /* Perform the requested task */
 328                         t->t_func(t->t_arg);
 329
 330                         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 331                         tq->tq_nactive--;
 332                         id = t->t_id;
 333                         task_done(tq, t);
 334
 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);
 339                         }
 340
 341                         wake_up_all(&tq->tq_wait_waitq);
 342                 }
 343
 344                 set_current_state(TASK_INTERRUPTIBLE);
 345
 346         }
 347
 348         __set_current_state(TASK_RUNNING);
 349         tq->tq_nthreads--;
 350         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 351
 352         RETURN(0);
 353 }
 354
 355 taskq_t *
 356 __taskq_create(const char *name, int nthreads, pri_t pri,
 357                int minalloc, int maxalloc, uint_t flags)
 358 {
 359         taskq_t *tq;
 360         struct task_struct *t;
 361         int rc = 0, i, j = 0;
 362         ENTRY;
 363
 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 */
 369
 370         tq = kmem_alloc(sizeof(*tq), KM_SLEEP);
 371         if (tq == NULL)
 372                 RETURN(NULL);
 373
 374         tq->tq_threads = kmem_alloc(nthreads * sizeof(t), KM_SLEEP);
 375         if (tq->tq_threads == NULL) {
 376                 kmem_free(tq, sizeof(*tq));
 377                 RETURN(NULL);
 378         }
 379
 380         spin_lock_init(&tq->tq_lock);
 381         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 382         tq->tq_name      = name;
 383         tq->tq_nactive   = 0;
 384         tq->tq_nthreads  = 0;
 385         tq->tq_pri       = pri;
 386         tq->tq_minalloc  = minalloc;
 387         tq->tq_maxalloc  = maxalloc;
 388         tq->tq_nalloc    = 0;
 389         tq->tq_flags     = (flags | TQ_ACTIVE);
 390         tq->tq_next_id   = 1;
 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);
 397
 398         if (flags & TASKQ_PREPOPULATE)
 399                 for (i = 0; i < minalloc; i++)
 400                         task_done(tq, task_alloc(tq, TQ_SLEEP | TQ_NEW));
 401
 402         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 403
 404         for (i = 0; i < nthreads; i++) {
 405                 t = kthread_create(taskq_thread, tq, "%s/%d", name, i);
 406                 if (t) {
 407                         tq->tq_threads[i] = t;
 408                         kthread_bind(t, i % num_online_cpus());
 409                         set_user_nice(t, PRIO_TO_NICE(pri));
 410                         wake_up_process(t);
 411                         j++;
 412                 } else {
 413                         tq->tq_threads[i] = NULL;
 414                         rc = 1;
 415                 }
 416         }
 417
 418         /* Wait for all threads to be started before potential destroy */
 419         wait_event(tq->tq_wait_waitq, tq->tq_nthreads == j);
 420
 421         if (rc) {
 422                 __taskq_destroy(tq);
 423                 tq = NULL;
 424         }
 425
 426         RETURN(tq);
 427 }
 428 EXPORT_SYMBOL(__taskq_create);
 429
 430 void
 431 __taskq_destroy(taskq_t *tq)
 432 {
 433         spl_task_t *t;
 434         int i, nthreads;
 435         ENTRY;
 436
 437         ASSERT(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);
 441
 442         /* TQ_ACTIVE cleared prevents new tasks being added to pending */
 443         __taskq_wait(tq);
 444
 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]);
 449
 450         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 451
 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);
 455                 task_free(tq, t);
 456         }
 457
 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));
 463
 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));
 467
 468         EXIT;
 469 }
 470 EXPORT_SYMBOL(__taskq_destroy);
 471
 472 int
 473 spl_taskq_init(void)
 474 {
 475         ENTRY;
 476
 477         system_taskq = taskq_create("system_taskq", 64, minclsyspri, 4, 512,
 478                                     TASKQ_PREPOPULATE);
 479         if (system_taskq == NULL)
 480                 RETURN(1);
 481
 482         RETURN(0);
 483 }
 484
 485 void
 486 spl_taskq_fini(void)
 487 {
 488         ENTRY;
 489         taskq_destroy(system_taskq);
 490         EXIT;
 491 }