sys/kern/subr_taskqueue.c

   1 /*-
   2  * Copyright (c) 2000 Doug Rabson
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions
   7  * are met:
   8  * 1. Redistributions of source code must retain the above copyright
   9  *    notice, this list of conditions and the following disclaimer.
  10  * 2. Redistributions in binary form must reproduce the above copyright
  11  *    notice, this list of conditions and the following disclaimer in the
  12  *    documentation and/or other materials provided with the distribution.
  13  *
  14  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
  15  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  16  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  17  * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
  18  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  19  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  20  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  21  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  22  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  23  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  24  * SUCH DAMAGE.
  25  *
  26  *       $FreeBSD: src/sys/kern/subr_taskqueue.c,v 1.1.2.3 2003/09/10 00:40:39 ken Exp $
  27  */
  28
  29 #include <sys/param.h>
  30 #include <sys/queue.h>
  31 #include <sys/systm.h>
  32 #include <sys/kernel.h>
  33 #include <sys/taskqueue.h>
  34 #include <sys/interrupt.h>
  35 #include <sys/lock.h>
  36 #include <sys/malloc.h>
  37 #include <sys/kthread.h>
  38 #include <sys/thread2.h>
  39 #include <sys/spinlock.h>
  40 #include <sys/spinlock2.h>
  41 #include <sys/serialize.h>
  42 #include <sys/proc.h>
  43 #include <machine/varargs.h>
  44
  45 MALLOC_DEFINE(M_TASKQUEUE, "taskqueue", "Task Queues");
  46
  47 static STAILQ_HEAD(taskqueue_list, taskqueue) taskqueue_queues;
  48 static struct lock      taskqueue_queues_lock;
  49
  50 struct taskqueue {
  51         STAILQ_ENTRY(taskqueue) tq_link;
  52         STAILQ_HEAD(, task)     tq_queue;
  53         const char              *tq_name;
  54         taskqueue_enqueue_fn    tq_enqueue;
  55         void                    *tq_context;
  56
  57         struct task             *tq_running;
  58         struct spinlock         tq_lock;
  59         struct thread           **tq_threads;
  60         int                     tq_tcount;
  61         int                     tq_flags;
  62 };
  63
  64 #define TQ_FLAGS_ACTIVE         (1 << 0)
  65 #define TQ_FLAGS_BLOCKED        (1 << 1)
  66 #define TQ_FLAGS_PENDING        (1 << 2)
  67
  68 static void taskqueue_run(struct taskqueue *queue, int lock_held);
  69
  70 static __inline void
  71 TQ_LOCK_INIT(struct taskqueue *tq)
  72 {
  73         spin_init(&tq->tq_lock);
  74 }
  75
  76 static __inline void
  77 TQ_LOCK_UNINIT(struct taskqueue *tq)
  78 {
  79         spin_uninit(&tq->tq_lock);
  80 }
  81
  82 static __inline void
  83 TQ_LOCK(struct taskqueue *tq)
  84 {
  85         spin_lock(&tq->tq_lock);
  86 }
  87
  88 static __inline void
  89 TQ_UNLOCK(struct taskqueue *tq)
  90 {
  91         spin_unlock(&tq->tq_lock);
  92 }
  93
  94 static __inline void
  95 TQ_SLEEP(struct taskqueue *tq, void *ident, const char *wmesg)
  96 {
  97         ssleep(ident, &tq->tq_lock, 0, wmesg, 0);
  98 }
  99
 100 struct taskqueue *
 101 taskqueue_create(const char *name, int mflags,
 102                  taskqueue_enqueue_fn enqueue, void *context)
 103 {
 104         struct taskqueue *queue;
 105
 106         queue = kmalloc(sizeof(*queue), M_TASKQUEUE, mflags | M_ZERO);
 107         if (!queue)
 108                 return NULL;
 109         STAILQ_INIT(&queue->tq_queue);
 110         queue->tq_name = name;
 111         queue->tq_enqueue = enqueue;
 112         queue->tq_context = context;
 113         queue->tq_flags |= TQ_FLAGS_ACTIVE;
 114         TQ_LOCK_INIT(queue);
 115
 116         lockmgr(&taskqueue_queues_lock, LK_EXCLUSIVE);
 117         STAILQ_INSERT_TAIL(&taskqueue_queues, queue, tq_link);
 118         lockmgr(&taskqueue_queues_lock, LK_RELEASE);
 119
 120         return queue;
 121 }
 122
 123 static void
 124 taskqueue_terminate(struct thread **pp, struct taskqueue *tq)
 125 {
 126         while(tq->tq_tcount > 0) {
 127                 wakeup(tq);
 128                 TQ_SLEEP(tq, pp, "taskqueue_terminate");
 129         }
 130 }
 131
 132 void
 133 taskqueue_free(struct taskqueue *queue)
 134 {
 135         TQ_LOCK(queue);
 136         queue->tq_flags &= ~TQ_FLAGS_ACTIVE;
 137         taskqueue_run(queue, 1);
 138         taskqueue_terminate(queue->tq_threads, queue);
 139         TQ_UNLOCK(queue);
 140
 141         lockmgr(&taskqueue_queues_lock, LK_EXCLUSIVE);
 142         STAILQ_REMOVE(&taskqueue_queues, queue, taskqueue, tq_link);
 143         lockmgr(&taskqueue_queues_lock, LK_RELEASE);
 144
 145         TQ_LOCK_UNINIT(queue);
 146
 147         kfree(queue, M_TASKQUEUE);
 148 }
 149
 150 struct taskqueue *
 151 taskqueue_find(const char *name)
 152 {
 153         struct taskqueue *queue;
 154
 155         lockmgr(&taskqueue_queues_lock, LK_EXCLUSIVE);
 156         STAILQ_FOREACH(queue, &taskqueue_queues, tq_link) {
 157                 if (!strcmp(queue->tq_name, name)) {
 158                         lockmgr(&taskqueue_queues_lock, LK_RELEASE);
 159                         return queue;
 160                 }
 161         }
 162         lockmgr(&taskqueue_queues_lock, LK_RELEASE);
 163         return NULL;
 164 }
 165
 166 /*
 167  * NOTE!  If using the per-cpu taskqueues ``taskqueue_thread[mycpuid]'',
 168  * be sure NOT TO SHARE the ``task'' between CPUs.  TASKS ARE NOT LOCKED.
 169  * So either use a throwaway task which will only be enqueued once, or
 170  * use one task per CPU!
 171  */
 172 int
 173 taskqueue_enqueue(struct taskqueue *queue, struct task *task)
 174 {
 175         struct task *ins;
 176         struct task *prev;
 177
 178         TQ_LOCK(queue);
 179
 180         /*
 181          * Don't allow new tasks on a queue which is being freed.
 182          */
 183         if ((queue->tq_flags & TQ_FLAGS_ACTIVE) == 0) {
 184                 TQ_UNLOCK(queue);
 185                 return EPIPE;
 186         }
 187
 188         /*
 189          * Count multiple enqueues.
 190          */
 191         if (task->ta_pending) {
 192                 task->ta_pending++;
 193                 TQ_UNLOCK(queue);
 194                 return 0;
 195         }
 196
 197         /*
 198          * Optimise the case when all tasks have the same priority.
 199          */
 200         prev = STAILQ_LAST(&queue->tq_queue, task, ta_link);
 201         if (!prev || prev->ta_priority >= task->ta_priority) {
 202                 STAILQ_INSERT_TAIL(&queue->tq_queue, task, ta_link);
 203         } else {
 204                 prev = NULL;
 205                 for (ins = STAILQ_FIRST(&queue->tq_queue); ins;
 206                      prev = ins, ins = STAILQ_NEXT(ins, ta_link))
 207                         if (ins->ta_priority < task->ta_priority)
 208                                 break;
 209
 210                 if (prev)
 211                         STAILQ_INSERT_AFTER(&queue->tq_queue, prev, task, ta_link);
 212                 else
 213                         STAILQ_INSERT_HEAD(&queue->tq_queue, task, ta_link);
 214         }
 215
 216         task->ta_pending = 1;
 217         if ((queue->tq_flags & TQ_FLAGS_BLOCKED) == 0) {
 218                 if (queue->tq_enqueue)
 219                         queue->tq_enqueue(queue->tq_context);
 220         } else {
 221                 queue->tq_flags |= TQ_FLAGS_PENDING;
 222         }
 223
 224         TQ_UNLOCK(queue);
 225
 226         return 0;
 227 }
 228
 229 void
 230 taskqueue_block(struct taskqueue *queue)
 231 {
 232         TQ_LOCK(queue);
 233         queue->tq_flags |= TQ_FLAGS_BLOCKED;
 234         TQ_UNLOCK(queue);
 235 }
 236
 237 void
 238 taskqueue_unblock(struct taskqueue *queue)
 239 {
 240         TQ_LOCK(queue);
 241         queue->tq_flags &= ~TQ_FLAGS_BLOCKED;
 242         if (queue->tq_flags & TQ_FLAGS_PENDING) {
 243                 queue->tq_flags &= ~TQ_FLAGS_PENDING;
 244                 if (queue->tq_enqueue)
 245                         queue->tq_enqueue(queue->tq_context);
 246         }
 247         TQ_UNLOCK(queue);
 248 }
 249
 250 void
 251 taskqueue_run(struct taskqueue *queue, int lock_held)
 252 {
 253         struct task *task;
 254         int pending;
 255
 256         if (lock_held == 0)
 257                 TQ_LOCK(queue);
 258         while (STAILQ_FIRST(&queue->tq_queue)) {
 259                 /*
 260                  * Carefully remove the first task from the queue and
 261                  * zero its pending count.
 262                  */
 263                 task = STAILQ_FIRST(&queue->tq_queue);
 264                 STAILQ_REMOVE_HEAD(&queue->tq_queue, ta_link);
 265                 pending = task->ta_pending;
 266                 task->ta_pending = 0;
 267                 queue->tq_running = task;
 268                 TQ_UNLOCK(queue);
 269
 270                 task->ta_func(task->ta_context, pending);
 271
 272                 TQ_LOCK(queue);
 273                 queue->tq_running = NULL;
 274                 wakeup(task);
 275         }
 276         if (lock_held == 0)
 277                 TQ_UNLOCK(queue);
 278 }
 279
 280 void
 281 taskqueue_drain(struct taskqueue *queue, struct task *task)
 282 {
 283         TQ_LOCK(queue);
 284         while (task->ta_pending != 0 || task == queue->tq_running)
 285                 TQ_SLEEP(queue, task, "-");
 286         TQ_UNLOCK(queue);
 287 }
 288
 289 static void
 290 taskqueue_swi_enqueue(void *context)
 291 {
 292         setsofttq();
 293 }
 294
 295 static void
 296 taskqueue_swi_run(void *arg, void *frame)
 297 {
 298         taskqueue_run(taskqueue_swi, 0);
 299 }
 300
 301 static void
 302 taskqueue_swi_mp_run(void *arg, void *frame)
 303 {
 304         taskqueue_run(taskqueue_swi_mp, 0);
 305 }
 306
 307 int
 308 taskqueue_start_threads(struct taskqueue **tqp, int count, int pri, int ncpu,
 309                         const char *fmt, ...)
 310 {
 311         __va_list ap;
 312         struct thread *td;
 313         struct taskqueue *tq;
 314         int i, error, cpu;
 315         char ktname[MAXCOMLEN];
 316
 317         if (count <= 0)
 318                 return EINVAL;
 319
 320         tq = *tqp;
 321         cpu = ncpu;
 322
 323         __va_start(ap, fmt);
 324         kvsnprintf(ktname, MAXCOMLEN, fmt, ap);
 325         __va_end(ap);
 326
 327         tq->tq_threads = kmalloc(sizeof(struct thread *) * count, M_TASKQUEUE,
 328             M_WAITOK | M_ZERO);
 329
 330         for (i = 0; i < count; i++) {
 331                 /*
 332                  * If no specific cpu was specified and more than one thread
 333                  * is to be created, we distribute the threads amongst all
 334                  * cpus.
 335                  */
 336                 if ((ncpu <= -1) && (count > 1))
 337                         cpu = i%ncpus;
 338
 339                 if (count == 1) {
 340                         error = lwkt_create(taskqueue_thread_loop, tqp,
 341                                             &tq->tq_threads[i], NULL,
 342                                             TDF_STOPREQ, cpu,
 343                                             "%s", ktname);
 344                 } else {
 345                         error = lwkt_create(taskqueue_thread_loop, tqp,
 346                                             &tq->tq_threads[i], NULL,
 347                                             TDF_STOPREQ, cpu,
 348                                             "%s_%d", ktname, i);
 349                 }
 350                 if (error) {
 351                         kprintf("%s: lwkt_create(%s): error %d", __func__,
 352                             ktname, error);
 353                         tq->tq_threads[i] = NULL;
 354                 } else {
 355                         td = tq->tq_threads[i];
 356                         lwkt_setpri_initial(td, pri);
 357                         lwkt_schedule(td);
 358                         tq->tq_tcount++;
 359                 }
 360         }
 361
 362         return 0;
 363 }
 364
 365 void
 366 taskqueue_thread_loop(void *arg)
 367 {
 368         struct taskqueue **tqp, *tq;
 369
 370         tqp = arg;
 371         tq = *tqp;
 372         TQ_LOCK(tq);
 373         while ((tq->tq_flags & TQ_FLAGS_ACTIVE) != 0) {
 374                 taskqueue_run(tq, 1);
 375                 TQ_SLEEP(tq, tq, "tqthr");
 376         }
 377
 378         /* rendezvous with thread that asked us to terminate */
 379         tq->tq_tcount--;
 380         wakeup_one(tq->tq_threads);
 381         TQ_UNLOCK(tq);
 382         lwkt_exit();
 383 }
 384
 385 void
 386 taskqueue_thread_enqueue(void *context)
 387 {
 388         struct taskqueue **tqp, *tq;
 389
 390         tqp = context;
 391         tq = *tqp;
 392
 393         wakeup_one(tq);
 394 }
 395
 396 TASKQUEUE_DEFINE(swi, taskqueue_swi_enqueue, 0,
 397          register_swi(SWI_TQ, taskqueue_swi_run, NULL, "swi_taskq", NULL));
 398 /*
 399  * XXX: possibly use a different SWI_TQ_MP or so.
 400  * related: sys/interrupt.h
 401  * related: platform/XXX/isa/ipl_funcs.c
 402  */
 403 TASKQUEUE_DEFINE(swi_mp, taskqueue_swi_enqueue, 0,
 404          register_swi(SWI_TQ, taskqueue_swi_mp_run, NULL, "swi_mp_taskq", NULL));
 405
 406 struct taskqueue *taskqueue_thread[MAXCPU];
 407
 408 static void
 409 taskqueue_init(void)
 410 {
 411         int cpu;
 412
 413         lockinit(&taskqueue_queues_lock, "tqqueues", 0, 0);
 414         STAILQ_INIT(&taskqueue_queues);
 415
 416         for (cpu = 0; cpu < ncpus; cpu++) {
 417                 taskqueue_thread[cpu] = taskqueue_create("thread", M_INTWAIT,
 418                     taskqueue_thread_enqueue, &taskqueue_thread[cpu]);
 419                 taskqueue_start_threads(&taskqueue_thread[cpu], 1,
 420                     TDPRI_KERN_DAEMON, cpu, "taskq_cpu %d", cpu);
 421         }
 422 }
 423
 424 SYSINIT(taskqueueinit, SI_SUB_PRE_DRIVERS, SI_ORDER_ANY, taskqueue_init, NULL);