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         /* Acquire 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 allocations 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 or 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, expects 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 destroys 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 fortunately it is
 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         int rc;
 168
 169         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 170         rc = (id < tq->tq_lowest_id);
 171         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 172
 173         RETURN(rc);
 174 }
 175
 176 /* Expected to wait for all previously scheduled tasks to complete.  We do
 177  * not need to wait for tasked scheduled after this call to complete.  In
 178  * other words we do not need to drain the entire taskq. */
 179 void
 180 __taskq_wait_id(taskq_t *tq, taskqid_t id)
 181 {
 182         ENTRY;
 183         ASSERT(tq);
 184
 185         wait_event(tq->tq_wait_waitq, taskq_wait_check(tq, id));
 186
 187         EXIT;
 188 }
 189 EXPORT_SYMBOL(__taskq_wait_id);
 190
 191 void
 192 __taskq_wait(taskq_t *tq)
 193 {
 194         taskqid_t id;
 195         ENTRY;
 196         ASSERT(tq);
 197
 198         /* Wait for the largest outstanding taskqid */
 199         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 200         id = tq->tq_next_id - 1;
 201         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 202
 203         __taskq_wait_id(tq, id);
 204
 205         EXIT;
 206
 207 }
 208 EXPORT_SYMBOL(__taskq_wait);
 209
 210 int
 211 __taskq_member(taskq_t *tq, void *t)
 212 {
 213         int i;
 214         ENTRY;
 215
 216         ASSERT(tq);
 217         ASSERT(t);
 218
 219         for (i = 0; i < tq->tq_nthreads; i++)
 220                 if (tq->tq_threads[i] == (struct task_struct *)t)
 221                         RETURN(1);
 222
 223         RETURN(0);
 224 }
 225 EXPORT_SYMBOL(__taskq_member);
 226
 227 taskqid_t
 228 __taskq_dispatch(taskq_t *tq, task_func_t func, void *arg, uint_t flags)
 229 {
 230         spl_task_t *t;
 231         taskqid_t rc = 0;
 232         ENTRY;
 233
 234         ASSERT(tq);
 235         ASSERT(func);
 236         if (unlikely(in_atomic() && (flags & TQ_SLEEP))) {
 237                 CERROR("May schedule while atomic: %s/0x%08x/%d\n",
 238                        current->comm, preempt_count(), current->pid);
 239                 SBUG();
 240         }
 241
 242         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 243
 244         /* Taskq being destroyed and all tasks drained */
 245         if (!(tq->tq_flags & TQ_ACTIVE))
 246                 GOTO(out, rc = 0);
 247
 248         /* Do not queue the task unless there is idle thread for it */
 249         ASSERT(tq->tq_nactive <= tq->tq_nthreads);
 250         if ((flags & TQ_NOQUEUE) && (tq->tq_nactive == tq->tq_nthreads))
 251                 GOTO(out, rc = 0);
 252
 253         if ((t = task_alloc(tq, flags)) == NULL)
 254                 GOTO(out, rc = 0);
 255
 256         spin_lock(&t->t_lock);
 257         list_add_tail(&t->t_list, &tq->tq_pend_list);
 258         t->t_id = rc = tq->tq_next_id;
 259         tq->tq_next_id++;
 260         t->t_func = func;
 261         t->t_arg = arg;
 262         spin_unlock(&t->t_lock);
 263
 264         wake_up(&tq->tq_work_waitq);
 265 out:
 266         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 267         RETURN(rc);
 268 }
 269 EXPORT_SYMBOL(__taskq_dispatch);
 270
 271 /* NOTE: Must be called with tq->tq_lock held */
 272 static taskqid_t
 273 taskq_lowest_id(taskq_t *tq)
 274 {
 275         taskqid_t lowest_id = tq->tq_next_id;
 276         spl_task_t *t;
 277         ENTRY;
 278
 279         ASSERT(tq);
 280         ASSERT(spin_is_locked(&tq->tq_lock));
 281
 282         list_for_each_entry(t, &tq->tq_pend_list, t_list)
 283                 if (t->t_id < lowest_id)
 284                         lowest_id = t->t_id;
 285
 286         list_for_each_entry(t, &tq->tq_work_list, t_list)
 287                 if (t->t_id < lowest_id)
 288                         lowest_id = t->t_id;
 289
 290         RETURN(lowest_id);
 291 }
 292
 293 static int
 294 taskq_thread(void *args)
 295 {
 296         DECLARE_WAITQUEUE(wait, current);
 297         sigset_t blocked;
 298         taskqid_t id;
 299         taskq_t *tq = args;
 300         spl_task_t *t;
 301         ENTRY;
 302
 303         ASSERT(tq);
 304         current->flags |= PF_NOFREEZE;
 305
 306         sigfillset(&blocked);
 307         sigprocmask(SIG_BLOCK, &blocked, NULL);
 308         flush_signals(current);
 309
 310         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 311         tq->tq_nthreads++;
 312         wake_up(&tq->tq_wait_waitq);
 313         set_current_state(TASK_INTERRUPTIBLE);
 314
 315         while (!kthread_should_stop()) {
 316
 317                 add_wait_queue(&tq->tq_work_waitq, &wait);
 318                 if (list_empty(&tq->tq_pend_list)) {
 319                         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 320                         schedule();
 321                         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 322                 } else {
 323                         __set_current_state(TASK_RUNNING);
 324                 }
 325
 326                 remove_wait_queue(&tq->tq_work_waitq, &wait);
 327                 if (!list_empty(&tq->tq_pend_list)) {
 328                         t = list_entry(tq->tq_pend_list.next,spl_task_t,t_list);
 329                         list_del_init(&t->t_list);
 330                         list_add_tail(&t->t_list, &tq->tq_work_list);
 331                         tq->tq_nactive++;
 332                         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 333
 334                         /* Perform the requested task */
 335                         t->t_func(t->t_arg);
 336
 337                         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 338                         tq->tq_nactive--;
 339                         id = t->t_id;
 340                         task_done(tq, t);
 341
 342                         /* When the current lowest outstanding taskqid is
 343                          * done calculate the new lowest outstanding id */
 344                         if (tq->tq_lowest_id == id) {
 345                                 tq->tq_lowest_id = taskq_lowest_id(tq);
 346                                 ASSERT(tq->tq_lowest_id > id);
 347                         }
 348
 349                         wake_up_all(&tq->tq_wait_waitq);
 350                 }
 351
 352                 set_current_state(TASK_INTERRUPTIBLE);
 353
 354         }
 355
 356         __set_current_state(TASK_RUNNING);
 357         tq->tq_nthreads--;
 358         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 359
 360         RETURN(0);
 361 }
 362
 363 taskq_t *
 364 __taskq_create(const char *name, int nthreads, pri_t pri,
 365                int minalloc, int maxalloc, uint_t flags)
 366 {
 367         taskq_t *tq;
 368         struct task_struct *t;
 369         int rc = 0, i, j = 0;
 370         ENTRY;
 371
 372         ASSERT(name != NULL);
 373         ASSERT(pri <= maxclsyspri);
 374         ASSERT(minalloc >= 0);
 375         ASSERT(maxalloc <= INT_MAX);
 376         ASSERT(!(flags & (TASKQ_CPR_SAFE | TASKQ_DYNAMIC))); /* Unsupported */
 377
 378         tq = kmem_alloc(sizeof(*tq), KM_SLEEP);
 379         if (tq == NULL)
 380                 RETURN(NULL);
 381
 382         tq->tq_threads = kmem_alloc(nthreads * sizeof(t), KM_SLEEP);
 383         if (tq->tq_threads == NULL) {
 384                 kmem_free(tq, sizeof(*tq));
 385                 RETURN(NULL);
 386         }
 387
 388         spin_lock_init(&tq->tq_lock);
 389         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 390         tq->tq_name      = name;
 391         tq->tq_nactive   = 0;
 392         tq->tq_nthreads  = 0;
 393         tq->tq_pri       = pri;
 394         tq->tq_minalloc  = minalloc;
 395         tq->tq_maxalloc  = maxalloc;
 396         tq->tq_nalloc    = 0;
 397         tq->tq_flags     = (flags | TQ_ACTIVE);
 398         tq->tq_next_id   = 1;
 399         tq->tq_lowest_id = 1;
 400         INIT_LIST_HEAD(&tq->tq_free_list);
 401         INIT_LIST_HEAD(&tq->tq_work_list);
 402         INIT_LIST_HEAD(&tq->tq_pend_list);
 403         init_waitqueue_head(&tq->tq_work_waitq);
 404         init_waitqueue_head(&tq->tq_wait_waitq);
 405
 406         if (flags & TASKQ_PREPOPULATE)
 407                 for (i = 0; i < minalloc; i++)
 408                         task_done(tq, task_alloc(tq, TQ_SLEEP | TQ_NEW));
 409
 410         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 411
 412         for (i = 0; i < nthreads; i++) {
 413                 t = kthread_create(taskq_thread, tq, "%s/%d", name, i);
 414                 if (t) {
 415                         tq->tq_threads[i] = t;
 416                         kthread_bind(t, i % num_online_cpus());
 417                         set_user_nice(t, PRIO_TO_NICE(pri));
 418                         wake_up_process(t);
 419                         j++;
 420                 } else {
 421                         tq->tq_threads[i] = NULL;
 422                         rc = 1;
 423                 }
 424         }
 425
 426         /* Wait for all threads to be started before potential destroy */
 427         wait_event(tq->tq_wait_waitq, tq->tq_nthreads == j);
 428
 429         if (rc) {
 430                 __taskq_destroy(tq);
 431                 tq = NULL;
 432         }
 433
 434         RETURN(tq);
 435 }
 436 EXPORT_SYMBOL(__taskq_create);
 437
 438 void
 439 __taskq_destroy(taskq_t *tq)
 440 {
 441         spl_task_t *t;
 442         int i, nthreads;
 443         ENTRY;
 444
 445         ASSERT(tq);
 446         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 447         tq->tq_flags &= ~TQ_ACTIVE;
 448         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 449
 450         /* TQ_ACTIVE cleared prevents new tasks being added to pending */
 451         __taskq_wait(tq);
 452
 453         nthreads = tq->tq_nthreads;
 454         for (i = 0; i < nthreads; i++)
 455                 if (tq->tq_threads[i])
 456                         kthread_stop(tq->tq_threads[i]);
 457
 458         spin_lock_irqsave(&tq->tq_lock, tq->tq_lock_flags);
 459
 460         while (!list_empty(&tq->tq_free_list)) {
 461                 t = list_entry(tq->tq_free_list.next, spl_task_t, t_list);
 462                 list_del_init(&t->t_list);
 463                 task_free(tq, t);
 464         }
 465
 466         ASSERT(tq->tq_nthreads == 0);
 467         ASSERT(tq->tq_nalloc == 0);
 468         ASSERT(list_empty(&tq->tq_free_list));
 469         ASSERT(list_empty(&tq->tq_work_list));
 470         ASSERT(list_empty(&tq->tq_pend_list));
 471
 472         spin_unlock_irqrestore(&tq->tq_lock, tq->tq_lock_flags);
 473         kmem_free(tq->tq_threads, nthreads * sizeof(spl_task_t *));
 474         kmem_free(tq, sizeof(taskq_t));
 475
 476         EXIT;
 477 }
 478 EXPORT_SYMBOL(__taskq_destroy);
 479
 480 int
 481 spl_taskq_init(void)
 482 {
 483         ENTRY;
 484
 485         /* Solaris creates a dynamic taskq of up to 64 threads, however in
 486          * a Linux environment 1 thread per-core is usually about right */
 487         system_taskq = taskq_create("spl_system_taskq", num_online_cpus(),
 488                                     minclsyspri, 4, 512, TASKQ_PREPOPULATE);
 489         if (system_taskq == NULL)
 490                 RETURN(1);
 491
 492         RETURN(0);
 493 }
 494
 495 void
 496 spl_taskq_fini(void)
 497 {
 498         ENTRY;
 499         taskq_destroy(system_taskq);
 500         EXIT;
 501 }