subr_taskqueue.c revision 170307
161033Sdfr/*- 261033Sdfr * Copyright (c) 2000 Doug Rabson 361033Sdfr * All rights reserved. 461033Sdfr * 561033Sdfr * Redistribution and use in source and binary forms, with or without 661033Sdfr * modification, are permitted provided that the following conditions 761033Sdfr * are met: 861033Sdfr * 1. Redistributions of source code must retain the above copyright 961033Sdfr * notice, this list of conditions and the following disclaimer. 1061033Sdfr * 2. Redistributions in binary form must reproduce the above copyright 1161033Sdfr * notice, this list of conditions and the following disclaimer in the 1261033Sdfr * documentation and/or other materials provided with the distribution. 1361033Sdfr * 1461033Sdfr * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND 1561033Sdfr * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 1661033Sdfr * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 1761033Sdfr * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE 1861033Sdfr * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 1961033Sdfr * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS 2061033Sdfr * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) 2161033Sdfr * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 2261033Sdfr * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY 2361033Sdfr * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF 2461033Sdfr * SUCH DAMAGE. 2561033Sdfr */ 2661033Sdfr 27116182Sobrien#include <sys/cdefs.h> 28116182Sobrien__FBSDID("$FreeBSD: head/sys/kern/subr_taskqueue.c 170307 2007-06-05 00:00:57Z jeff $"); 29116182Sobrien 3061033Sdfr#include <sys/param.h> 3185521Sjhb#include <sys/systm.h> 3265822Sjhb#include <sys/bus.h> 3385560Sjhb#include <sys/interrupt.h> 3461033Sdfr#include <sys/kernel.h> 35123614Sjhb#include <sys/kthread.h> 3685521Sjhb#include <sys/lock.h> 3761033Sdfr#include <sys/malloc.h> 3885521Sjhb#include <sys/mutex.h> 39145729Ssam#include <sys/proc.h> 40154333Sscottl#include <sys/sched.h> 4185521Sjhb#include <sys/taskqueue.h> 42119708Sken#include <sys/unistd.h> 43154333Sscottl#include <machine/stdarg.h> 4461033Sdfr 4569774Sphkstatic MALLOC_DEFINE(M_TASKQUEUE, "taskqueue", "Task Queues"); 46123614Sjhbstatic void *taskqueue_giant_ih; 47123614Sjhbstatic void *taskqueue_ih; 4861033Sdfrstatic STAILQ_HEAD(taskqueue_list, taskqueue) taskqueue_queues; 4985521Sjhbstatic struct mtx taskqueue_queues_mutex; 5067551Sjhb 5161033Sdfrstruct taskqueue { 5261033Sdfr STAILQ_ENTRY(taskqueue) tq_link; 5361033Sdfr STAILQ_HEAD(, task) tq_queue; 5461033Sdfr const char *tq_name; 5561033Sdfr taskqueue_enqueue_fn tq_enqueue; 5661033Sdfr void *tq_context; 57145473Ssam struct task *tq_running; 5885521Sjhb struct mtx tq_mutex; 59145729Ssam struct proc **tq_pproc; 60154333Sscottl int tq_pcount; 61154167Sscottl int tq_spin; 62154333Sscottl int tq_flags; 6361033Sdfr}; 6461033Sdfr 65154333Sscottl#define TQ_FLAGS_ACTIVE (1 << 0) 66154333Sscottl 67154167Sscottlstatic __inline void 68154167SscottlTQ_LOCK(struct taskqueue *tq) 69154167Sscottl{ 70154167Sscottl if (tq->tq_spin) 71154167Sscottl mtx_lock_spin(&tq->tq_mutex); 72154167Sscottl else 73154167Sscottl mtx_lock(&tq->tq_mutex); 74154167Sscottl} 75154167Sscottl 76154167Sscottlstatic __inline void 77154167SscottlTQ_UNLOCK(struct taskqueue *tq) 78154167Sscottl{ 79154167Sscottl if (tq->tq_spin) 80154167Sscottl mtx_unlock_spin(&tq->tq_mutex); 81154167Sscottl else 82154167Sscottl mtx_unlock(&tq->tq_mutex); 83154167Sscottl} 84154167Sscottl 8585521Sjhbstatic void init_taskqueue_list(void *data); 8685521Sjhb 87154167Sscottlstatic __inline int 88154167SscottlTQ_SLEEP(struct taskqueue *tq, void *p, struct mtx *m, int pri, const char *wm, 89154167Sscottl int t) 90154167Sscottl{ 91154167Sscottl if (tq->tq_spin) 92154167Sscottl return (msleep_spin(p, m, wm, t)); 93154167Sscottl return (msleep(p, m, pri, wm, t)); 94154167Sscottl} 95154167Sscottl 9685521Sjhbstatic void 9785521Sjhbinit_taskqueue_list(void *data __unused) 9885521Sjhb{ 9985521Sjhb 10093818Sjhb mtx_init(&taskqueue_queues_mutex, "taskqueue list", NULL, MTX_DEF); 10185521Sjhb STAILQ_INIT(&taskqueue_queues); 10285521Sjhb} 10385521SjhbSYSINIT(taskqueue_list, SI_SUB_INTRINSIC, SI_ORDER_ANY, init_taskqueue_list, 10485521Sjhb NULL); 10585521Sjhb 106154167Sscottlstatic struct taskqueue * 107154167Sscottl_taskqueue_create(const char *name, int mflags, 108145729Ssam taskqueue_enqueue_fn enqueue, void *context, 109154167Sscottl int mtxflags, const char *mtxname) 11061033Sdfr{ 11161033Sdfr struct taskqueue *queue; 11261033Sdfr 11385521Sjhb queue = malloc(sizeof(struct taskqueue), M_TASKQUEUE, mflags | M_ZERO); 11461033Sdfr if (!queue) 11561033Sdfr return 0; 11685521Sjhb 11761033Sdfr STAILQ_INIT(&queue->tq_queue); 11861033Sdfr queue->tq_name = name; 11961033Sdfr queue->tq_enqueue = enqueue; 12061033Sdfr queue->tq_context = context; 121154167Sscottl queue->tq_spin = (mtxflags & MTX_SPIN) != 0; 122154333Sscottl queue->tq_flags |= TQ_FLAGS_ACTIVE; 123154167Sscottl mtx_init(&queue->tq_mutex, mtxname, NULL, mtxflags); 12461033Sdfr 12585521Sjhb mtx_lock(&taskqueue_queues_mutex); 12661033Sdfr STAILQ_INSERT_TAIL(&taskqueue_queues, queue, tq_link); 12785521Sjhb mtx_unlock(&taskqueue_queues_mutex); 12861033Sdfr 12961033Sdfr return queue; 13061033Sdfr} 13161033Sdfr 132154167Sscottlstruct taskqueue * 133154167Sscottltaskqueue_create(const char *name, int mflags, 134154333Sscottl taskqueue_enqueue_fn enqueue, void *context) 135154167Sscottl{ 136154333Sscottl return _taskqueue_create(name, mflags, enqueue, context, 137154167Sscottl MTX_DEF, "taskqueue"); 138154167Sscottl} 139154167Sscottl 140145729Ssam/* 141145729Ssam * Signal a taskqueue thread to terminate. 142145729Ssam */ 143145729Ssamstatic void 144145729Ssamtaskqueue_terminate(struct proc **pp, struct taskqueue *tq) 145145729Ssam{ 146145729Ssam 147154333Sscottl while (tq->tq_pcount > 0) { 148154333Sscottl wakeup(tq); 149154333Sscottl TQ_SLEEP(tq, pp, &tq->tq_mutex, PWAIT, "taskqueue_destroy", 0); 150145729Ssam } 151145729Ssam} 152145729Ssam 15361033Sdfrvoid 15461033Sdfrtaskqueue_free(struct taskqueue *queue) 15561033Sdfr{ 15685521Sjhb 15785521Sjhb mtx_lock(&taskqueue_queues_mutex); 15861033Sdfr STAILQ_REMOVE(&taskqueue_queues, queue, taskqueue, tq_link); 15985521Sjhb mtx_unlock(&taskqueue_queues_mutex); 16061033Sdfr 161154167Sscottl TQ_LOCK(queue); 162154333Sscottl queue->tq_flags &= ~TQ_FLAGS_ACTIVE; 163131246Sjhb taskqueue_run(queue); 164145729Ssam taskqueue_terminate(queue->tq_pproc, queue); 16585521Sjhb mtx_destroy(&queue->tq_mutex); 166154333Sscottl free(queue->tq_pproc, M_TASKQUEUE); 16761033Sdfr free(queue, M_TASKQUEUE); 16861033Sdfr} 16961033Sdfr 17085521Sjhb/* 17185521Sjhb * Returns with the taskqueue locked. 17285521Sjhb */ 17361033Sdfrstruct taskqueue * 17461033Sdfrtaskqueue_find(const char *name) 17561033Sdfr{ 17661033Sdfr struct taskqueue *queue; 17761033Sdfr 17885521Sjhb mtx_lock(&taskqueue_queues_mutex); 17985521Sjhb STAILQ_FOREACH(queue, &taskqueue_queues, tq_link) { 180123614Sjhb if (strcmp(queue->tq_name, name) == 0) { 181154167Sscottl TQ_LOCK(queue); 18285521Sjhb mtx_unlock(&taskqueue_queues_mutex); 18361033Sdfr return queue; 18461033Sdfr } 18585521Sjhb } 18685521Sjhb mtx_unlock(&taskqueue_queues_mutex); 187123614Sjhb return NULL; 18861033Sdfr} 18961033Sdfr 19061033Sdfrint 19161033Sdfrtaskqueue_enqueue(struct taskqueue *queue, struct task *task) 19261033Sdfr{ 19361033Sdfr struct task *ins; 19461033Sdfr struct task *prev; 19561033Sdfr 196154167Sscottl TQ_LOCK(queue); 19785560Sjhb 19861033Sdfr /* 19961033Sdfr * Count multiple enqueues. 20061033Sdfr */ 20161033Sdfr if (task->ta_pending) { 20261033Sdfr task->ta_pending++; 203154167Sscottl TQ_UNLOCK(queue); 20461033Sdfr return 0; 20561033Sdfr } 20661033Sdfr 20761033Sdfr /* 20861033Sdfr * Optimise the case when all tasks have the same priority. 20961033Sdfr */ 21064199Shsu prev = STAILQ_LAST(&queue->tq_queue, task, ta_link); 21161033Sdfr if (!prev || prev->ta_priority >= task->ta_priority) { 21261033Sdfr STAILQ_INSERT_TAIL(&queue->tq_queue, task, ta_link); 21361033Sdfr } else { 21461033Sdfr prev = 0; 21561033Sdfr for (ins = STAILQ_FIRST(&queue->tq_queue); ins; 21661033Sdfr prev = ins, ins = STAILQ_NEXT(ins, ta_link)) 21761033Sdfr if (ins->ta_priority < task->ta_priority) 21861033Sdfr break; 21961033Sdfr 22061033Sdfr if (prev) 22161033Sdfr STAILQ_INSERT_AFTER(&queue->tq_queue, prev, task, ta_link); 22261033Sdfr else 22361033Sdfr STAILQ_INSERT_HEAD(&queue->tq_queue, task, ta_link); 22461033Sdfr } 22561033Sdfr 22661033Sdfr task->ta_pending = 1; 227145729Ssam queue->tq_enqueue(queue->tq_context); 22885560Sjhb 229154167Sscottl TQ_UNLOCK(queue); 23085560Sjhb 23161033Sdfr return 0; 23261033Sdfr} 23361033Sdfr 23461033Sdfrvoid 23561033Sdfrtaskqueue_run(struct taskqueue *queue) 23661033Sdfr{ 23761033Sdfr struct task *task; 238131246Sjhb int owned, pending; 23961033Sdfr 240131246Sjhb owned = mtx_owned(&queue->tq_mutex); 241131246Sjhb if (!owned) 242154167Sscottl TQ_LOCK(queue); 24361033Sdfr while (STAILQ_FIRST(&queue->tq_queue)) { 24461033Sdfr /* 24561033Sdfr * Carefully remove the first task from the queue and 24661033Sdfr * zero its pending count. 24761033Sdfr */ 24861033Sdfr task = STAILQ_FIRST(&queue->tq_queue); 24961033Sdfr STAILQ_REMOVE_HEAD(&queue->tq_queue, ta_link); 25061033Sdfr pending = task->ta_pending; 25161033Sdfr task->ta_pending = 0; 252145473Ssam queue->tq_running = task; 253154167Sscottl TQ_UNLOCK(queue); 25461033Sdfr 25585560Sjhb task->ta_func(task->ta_context, pending); 25661033Sdfr 257154167Sscottl TQ_LOCK(queue); 258145473Ssam queue->tq_running = NULL; 259136131Simp wakeup(task); 26061033Sdfr } 261131246Sjhb 262131246Sjhb /* 263131246Sjhb * For compatibility, unlock on return if the queue was not locked 264131246Sjhb * on entry, although this opens a race window. 265131246Sjhb */ 266131246Sjhb if (!owned) 267154167Sscottl TQ_UNLOCK(queue); 26861033Sdfr} 26961033Sdfr 270136131Simpvoid 271136131Simptaskqueue_drain(struct taskqueue *queue, struct task *task) 272136131Simp{ 273154167Sscottl if (queue->tq_spin) { /* XXX */ 274154167Sscottl mtx_lock_spin(&queue->tq_mutex); 275154167Sscottl while (task->ta_pending != 0 || task == queue->tq_running) 276154167Sscottl msleep_spin(task, &queue->tq_mutex, "-", 0); 277154167Sscottl mtx_unlock_spin(&queue->tq_mutex); 278154167Sscottl } else { 279154167Sscottl WITNESS_WARN(WARN_GIANTOK | WARN_SLEEPOK, NULL, __func__); 280145729Ssam 281154167Sscottl mtx_lock(&queue->tq_mutex); 282154167Sscottl while (task->ta_pending != 0 || task == queue->tq_running) 283154167Sscottl msleep(task, &queue->tq_mutex, PWAIT, "-", 0); 284154167Sscottl mtx_unlock(&queue->tq_mutex); 285154167Sscottl } 286136131Simp} 287136131Simp 28861033Sdfrstatic void 28961033Sdfrtaskqueue_swi_enqueue(void *context) 29061033Sdfr{ 29188900Sjhb swi_sched(taskqueue_ih, 0); 29261033Sdfr} 29361033Sdfr 29461033Sdfrstatic void 29567551Sjhbtaskqueue_swi_run(void *dummy) 29661033Sdfr{ 29761033Sdfr taskqueue_run(taskqueue_swi); 29861033Sdfr} 29961033Sdfr 300111528Sscottlstatic void 301111528Sscottltaskqueue_swi_giant_enqueue(void *context) 302111528Sscottl{ 303111528Sscottl swi_sched(taskqueue_giant_ih, 0); 304111528Sscottl} 305111528Sscottl 306111528Sscottlstatic void 307111528Sscottltaskqueue_swi_giant_run(void *dummy) 308111528Sscottl{ 309111528Sscottl taskqueue_run(taskqueue_swi_giant); 310111528Sscottl} 311111528Sscottl 312154333Sscottlint 313154333Sscottltaskqueue_start_threads(struct taskqueue **tqp, int count, int pri, 314154333Sscottl const char *name, ...) 315154333Sscottl{ 316154333Sscottl va_list ap; 317154333Sscottl struct taskqueue *tq; 318158904Ssam struct thread *td; 319154333Sscottl char ktname[MAXCOMLEN]; 320157314Ssam int i, error; 321154333Sscottl 322154333Sscottl if (count <= 0) 323154333Sscottl return (EINVAL); 324154333Sscottl tq = *tqp; 325154333Sscottl 326154333Sscottl va_start(ap, name); 327154333Sscottl vsnprintf(ktname, MAXCOMLEN, name, ap); 328154333Sscottl va_end(ap); 329154333Sscottl 330157314Ssam tq->tq_pproc = malloc(sizeof(struct proc *) * count, M_TASKQUEUE, 331157314Ssam M_NOWAIT | M_ZERO); 332157314Ssam if (tq->tq_pproc == NULL) { 333157314Ssam printf("%s: no memory for %s threads\n", __func__, ktname); 334157314Ssam return (ENOMEM); 335157314Ssam } 336157314Ssam 337154333Sscottl for (i = 0; i < count; i++) { 338154333Sscottl if (count == 1) 339157314Ssam error = kthread_create(taskqueue_thread_loop, tqp, 340158904Ssam &tq->tq_pproc[i], RFSTOPPED, 0, ktname); 341154333Sscottl else 342157314Ssam error = kthread_create(taskqueue_thread_loop, tqp, 343158904Ssam &tq->tq_pproc[i], RFSTOPPED, 0, "%s_%d", ktname, i); 344158904Ssam if (error) { 345157314Ssam /* should be ok to continue, taskqueue_free will dtrt */ 346157314Ssam printf("%s: kthread_create(%s): error %d", 347157314Ssam __func__, ktname, error); 348158904Ssam tq->tq_pproc[i] = NULL; /* paranoid */ 349158904Ssam } else 350158904Ssam tq->tq_pcount++; 351154333Sscottl } 352158904Ssam for (i = 0; i < count; i++) { 353158904Ssam if (tq->tq_pproc[i] == NULL) 354158904Ssam continue; 355158904Ssam td = FIRST_THREAD_IN_PROC(tq->tq_pproc[i]); 356170307Sjeff thread_lock(td); 357158904Ssam sched_prio(td, pri); 358166188Sjeff sched_add(td, SRQ_BORING); 359170307Sjeff thread_unlock(td); 360158904Ssam } 361154333Sscottl 362154333Sscottl return (0); 363154333Sscottl} 364154333Sscottl 365133305Sjmgvoid 366133305Sjmgtaskqueue_thread_loop(void *arg) 367119708Sken{ 368133305Sjmg struct taskqueue **tqp, *tq; 369131246Sjhb 370133305Sjmg tqp = arg; 371133305Sjmg tq = *tqp; 372154167Sscottl TQ_LOCK(tq); 373145729Ssam do { 374133305Sjmg taskqueue_run(tq); 375157815Sjhb TQ_SLEEP(tq, tq, &tq->tq_mutex, 0, "-", 0); 376154333Sscottl } while ((tq->tq_flags & TQ_FLAGS_ACTIVE) != 0); 377145729Ssam 378145729Ssam /* rendezvous with thread that asked us to terminate */ 379154333Sscottl tq->tq_pcount--; 380154333Sscottl wakeup_one(tq->tq_pproc); 381154167Sscottl TQ_UNLOCK(tq); 382145729Ssam kthread_exit(0); 383119708Sken} 384119708Sken 385133305Sjmgvoid 386119708Skentaskqueue_thread_enqueue(void *context) 387119708Sken{ 388133305Sjmg struct taskqueue **tqp, *tq; 389131246Sjhb 390133305Sjmg tqp = context; 391133305Sjmg tq = *tqp; 392133305Sjmg 393133305Sjmg mtx_assert(&tq->tq_mutex, MA_OWNED); 394145729Ssam wakeup_one(tq); 395119708Sken} 396119708Sken 39761033SdfrTASKQUEUE_DEFINE(swi, taskqueue_swi_enqueue, 0, 398111528Sscottl swi_add(NULL, "task queue", taskqueue_swi_run, NULL, SWI_TQ, 399111528Sscottl INTR_MPSAFE, &taskqueue_ih)); 400111528Sscottl 401111528SscottlTASKQUEUE_DEFINE(swi_giant, taskqueue_swi_giant_enqueue, 0, 402151656Sjhb swi_add(NULL, "Giant taskq", taskqueue_swi_giant_run, 403111528Sscottl NULL, SWI_TQ_GIANT, 0, &taskqueue_giant_ih)); 404119708Sken 405133305SjmgTASKQUEUE_DEFINE_THREAD(thread); 406119789Ssam 407154167Sscottlstruct taskqueue * 408154167Sscottltaskqueue_create_fast(const char *name, int mflags, 409154333Sscottl taskqueue_enqueue_fn enqueue, void *context) 410119789Ssam{ 411154333Sscottl return _taskqueue_create(name, mflags, enqueue, context, 412154167Sscottl MTX_SPIN, "fast_taskqueue"); 413119789Ssam} 414119789Ssam 415154167Sscottl/* NB: for backwards compatibility */ 416154167Sscottlint 417154167Sscottltaskqueue_enqueue_fast(struct taskqueue *queue, struct task *task) 418119789Ssam{ 419154167Sscottl return taskqueue_enqueue(queue, task); 420119789Ssam} 421119789Ssam 422119789Ssamstatic void *taskqueue_fast_ih; 423119789Ssam 424119789Ssamstatic void 425154167Sscottltaskqueue_fast_enqueue(void *context) 426119789Ssam{ 427119789Ssam swi_sched(taskqueue_fast_ih, 0); 428119789Ssam} 429119789Ssam 430119789Ssamstatic void 431119789Ssamtaskqueue_fast_run(void *dummy) 432119789Ssam{ 433154167Sscottl taskqueue_run(taskqueue_fast); 434119789Ssam} 435119789Ssam 436154167SscottlTASKQUEUE_FAST_DEFINE(fast, taskqueue_fast_enqueue, 0, 437154167Sscottl swi_add(NULL, "Fast task queue", taskqueue_fast_run, NULL, 438154167Sscottl SWI_TQ_FAST, INTR_MPSAFE, &taskqueue_fast_ih)); 439