subr_taskqueue.c revision 208715
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 208715 2010-06-01 16:04:01Z zml $"); 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; 4867551Sjhb 4961033Sdfrstruct taskqueue { 5061033Sdfr STAILQ_HEAD(, task) tq_queue; 5161033Sdfr const char *tq_name; 5261033Sdfr taskqueue_enqueue_fn tq_enqueue; 5361033Sdfr void *tq_context; 54208715Szml struct task *tq_running; 5585521Sjhb struct mtx tq_mutex; 56178015Ssam struct thread **tq_threads; 57178015Ssam int tq_tcount; 58180588Skmacy int tq_spin; 59154333Sscottl int tq_flags; 6061033Sdfr}; 6161033Sdfr 62154333Sscottl#define TQ_FLAGS_ACTIVE (1 << 0) 63177621Sscottl#define TQ_FLAGS_BLOCKED (1 << 1) 64177621Sscottl#define TQ_FLAGS_PENDING (1 << 2) 65154333Sscottl 66180588Skmacystatic __inline void 67180588SkmacyTQ_LOCK(struct taskqueue *tq) 68180588Skmacy{ 69180588Skmacy if (tq->tq_spin) 70180588Skmacy mtx_lock_spin(&tq->tq_mutex); 71180588Skmacy else 72180588Skmacy mtx_lock(&tq->tq_mutex); 73180588Skmacy} 74154167Sscottl 75180588Skmacystatic __inline void 76180588SkmacyTQ_UNLOCK(struct taskqueue *tq) 77180588Skmacy{ 78180588Skmacy if (tq->tq_spin) 79180588Skmacy mtx_unlock_spin(&tq->tq_mutex); 80180588Skmacy else 81180588Skmacy mtx_unlock(&tq->tq_mutex); 82180588Skmacy} 83154167Sscottl 84154167Sscottlstatic __inline int 85154167SscottlTQ_SLEEP(struct taskqueue *tq, void *p, struct mtx *m, int pri, const char *wm, 86154167Sscottl int t) 87154167Sscottl{ 88180588Skmacy if (tq->tq_spin) 89154167Sscottl return (msleep_spin(p, m, wm, t)); 90154167Sscottl return (msleep(p, m, pri, wm, t)); 91154167Sscottl} 92154167Sscottl 93154167Sscottlstatic struct taskqueue * 94154167Sscottl_taskqueue_create(const char *name, int mflags, 95145729Ssam taskqueue_enqueue_fn enqueue, void *context, 96154167Sscottl int mtxflags, const char *mtxname) 9761033Sdfr{ 9861033Sdfr struct taskqueue *queue; 99180588Skmacy 10085521Sjhb queue = malloc(sizeof(struct taskqueue), M_TASKQUEUE, mflags | M_ZERO); 10161033Sdfr if (!queue) 102188058Simp return NULL; 103180588Skmacy 10461033Sdfr STAILQ_INIT(&queue->tq_queue); 10561033Sdfr queue->tq_name = name; 10661033Sdfr queue->tq_enqueue = enqueue; 10761033Sdfr queue->tq_context = context; 108180588Skmacy queue->tq_spin = (mtxflags & MTX_SPIN) != 0; 109180588Skmacy queue->tq_flags |= TQ_FLAGS_ACTIVE; 110154167Sscottl mtx_init(&queue->tq_mutex, mtxname, NULL, mtxflags); 11161033Sdfr 11261033Sdfr return queue; 11361033Sdfr} 11461033Sdfr 115154167Sscottlstruct taskqueue * 116154167Sscottltaskqueue_create(const char *name, int mflags, 117154333Sscottl taskqueue_enqueue_fn enqueue, void *context) 118154167Sscottl{ 119154333Sscottl return _taskqueue_create(name, mflags, enqueue, context, 120154167Sscottl MTX_DEF, "taskqueue"); 121154167Sscottl} 122154167Sscottl 123145729Ssam/* 124145729Ssam * Signal a taskqueue thread to terminate. 125145729Ssam */ 126145729Ssamstatic void 127178015Ssamtaskqueue_terminate(struct thread **pp, struct taskqueue *tq) 128145729Ssam{ 129145729Ssam 130178015Ssam while (tq->tq_tcount > 0) { 131154333Sscottl wakeup(tq); 132154333Sscottl TQ_SLEEP(tq, pp, &tq->tq_mutex, PWAIT, "taskqueue_destroy", 0); 133145729Ssam } 134145729Ssam} 135145729Ssam 13661033Sdfrvoid 13761033Sdfrtaskqueue_free(struct taskqueue *queue) 13861033Sdfr{ 13985521Sjhb 140154167Sscottl TQ_LOCK(queue); 141154333Sscottl queue->tq_flags &= ~TQ_FLAGS_ACTIVE; 142131246Sjhb taskqueue_run(queue); 143178015Ssam taskqueue_terminate(queue->tq_threads, queue); 14485521Sjhb mtx_destroy(&queue->tq_mutex); 145178015Ssam free(queue->tq_threads, M_TASKQUEUE); 14661033Sdfr free(queue, M_TASKQUEUE); 14761033Sdfr} 14861033Sdfr 14961033Sdfrint 15061033Sdfrtaskqueue_enqueue(struct taskqueue *queue, struct task *task) 15161033Sdfr{ 15261033Sdfr struct task *ins; 15361033Sdfr struct task *prev; 15461033Sdfr 155154167Sscottl TQ_LOCK(queue); 15685560Sjhb 15761033Sdfr /* 15861033Sdfr * Count multiple enqueues. 15961033Sdfr */ 160180588Skmacy if (task->ta_pending) { 16161033Sdfr task->ta_pending++; 162154167Sscottl TQ_UNLOCK(queue); 16361033Sdfr return 0; 16461033Sdfr } 16561033Sdfr 16661033Sdfr /* 16761033Sdfr * Optimise the case when all tasks have the same priority. 16861033Sdfr */ 16964199Shsu prev = STAILQ_LAST(&queue->tq_queue, task, ta_link); 17061033Sdfr if (!prev || prev->ta_priority >= task->ta_priority) { 17161033Sdfr STAILQ_INSERT_TAIL(&queue->tq_queue, task, ta_link); 17261033Sdfr } else { 173188058Simp prev = NULL; 17461033Sdfr for (ins = STAILQ_FIRST(&queue->tq_queue); ins; 17561033Sdfr prev = ins, ins = STAILQ_NEXT(ins, ta_link)) 17661033Sdfr if (ins->ta_priority < task->ta_priority) 17761033Sdfr break; 17861033Sdfr 17961033Sdfr if (prev) 18061033Sdfr STAILQ_INSERT_AFTER(&queue->tq_queue, prev, task, ta_link); 18161033Sdfr else 18261033Sdfr STAILQ_INSERT_HEAD(&queue->tq_queue, task, ta_link); 18361033Sdfr } 18461033Sdfr 18561033Sdfr task->ta_pending = 1; 186180588Skmacy if ((queue->tq_flags & TQ_FLAGS_BLOCKED) == 0) 187177621Sscottl queue->tq_enqueue(queue->tq_context); 188180588Skmacy else 189177621Sscottl queue->tq_flags |= TQ_FLAGS_PENDING; 19085560Sjhb 191154167Sscottl TQ_UNLOCK(queue); 19285560Sjhb 19361033Sdfr return 0; 19461033Sdfr} 19561033Sdfr 19661033Sdfrvoid 197177621Sscottltaskqueue_block(struct taskqueue *queue) 198177621Sscottl{ 199177621Sscottl 200177621Sscottl TQ_LOCK(queue); 201177621Sscottl queue->tq_flags |= TQ_FLAGS_BLOCKED; 202177621Sscottl TQ_UNLOCK(queue); 203177621Sscottl} 204177621Sscottl 205177621Sscottlvoid 206177621Sscottltaskqueue_unblock(struct taskqueue *queue) 207177621Sscottl{ 208177621Sscottl 209177621Sscottl TQ_LOCK(queue); 210177621Sscottl queue->tq_flags &= ~TQ_FLAGS_BLOCKED; 211177621Sscottl if (queue->tq_flags & TQ_FLAGS_PENDING) { 212177621Sscottl queue->tq_flags &= ~TQ_FLAGS_PENDING; 213177621Sscottl queue->tq_enqueue(queue->tq_context); 214177621Sscottl } 215177621Sscottl TQ_UNLOCK(queue); 216177621Sscottl} 217177621Sscottl 218177621Sscottlvoid 21961033Sdfrtaskqueue_run(struct taskqueue *queue) 22061033Sdfr{ 22161033Sdfr struct task *task; 222131246Sjhb int owned, pending; 22361033Sdfr 224131246Sjhb owned = mtx_owned(&queue->tq_mutex); 225131246Sjhb if (!owned) 226154167Sscottl TQ_LOCK(queue); 22761033Sdfr while (STAILQ_FIRST(&queue->tq_queue)) { 22861033Sdfr /* 22961033Sdfr * Carefully remove the first task from the queue and 23061033Sdfr * zero its pending count. 23161033Sdfr */ 23261033Sdfr task = STAILQ_FIRST(&queue->tq_queue); 23361033Sdfr STAILQ_REMOVE_HEAD(&queue->tq_queue, ta_link); 23461033Sdfr pending = task->ta_pending; 23561033Sdfr task->ta_pending = 0; 236208715Szml queue->tq_running = task; 237154167Sscottl TQ_UNLOCK(queue); 23861033Sdfr 23985560Sjhb task->ta_func(task->ta_context, pending); 24061033Sdfr 241154167Sscottl TQ_LOCK(queue); 242208715Szml queue->tq_running = NULL; 243208715Szml wakeup(task); 24461033Sdfr } 245131246Sjhb 246131246Sjhb /* 247131246Sjhb * For compatibility, unlock on return if the queue was not locked 248131246Sjhb * on entry, although this opens a race window. 249131246Sjhb */ 250131246Sjhb if (!owned) 251154167Sscottl TQ_UNLOCK(queue); 25261033Sdfr} 25361033Sdfr 254136131Simpvoid 255136131Simptaskqueue_drain(struct taskqueue *queue, struct task *task) 256136131Simp{ 257180588Skmacy if (queue->tq_spin) { /* XXX */ 258154167Sscottl mtx_lock_spin(&queue->tq_mutex); 259208715Szml while (task->ta_pending != 0 || task == queue->tq_running) 260154167Sscottl msleep_spin(task, &queue->tq_mutex, "-", 0); 261154167Sscottl mtx_unlock_spin(&queue->tq_mutex); 262154167Sscottl } else { 263154167Sscottl WITNESS_WARN(WARN_GIANTOK | WARN_SLEEPOK, NULL, __func__); 264145729Ssam 265154167Sscottl mtx_lock(&queue->tq_mutex); 266208715Szml while (task->ta_pending != 0 || task == queue->tq_running) 267154167Sscottl msleep(task, &queue->tq_mutex, PWAIT, "-", 0); 268154167Sscottl mtx_unlock(&queue->tq_mutex); 269154167Sscottl } 270136131Simp} 271136131Simp 27261033Sdfrstatic void 27361033Sdfrtaskqueue_swi_enqueue(void *context) 27461033Sdfr{ 27588900Sjhb swi_sched(taskqueue_ih, 0); 27661033Sdfr} 27761033Sdfr 27861033Sdfrstatic void 27967551Sjhbtaskqueue_swi_run(void *dummy) 28061033Sdfr{ 28161033Sdfr taskqueue_run(taskqueue_swi); 28261033Sdfr} 28361033Sdfr 284111528Sscottlstatic void 285111528Sscottltaskqueue_swi_giant_enqueue(void *context) 286111528Sscottl{ 287111528Sscottl swi_sched(taskqueue_giant_ih, 0); 288111528Sscottl} 289111528Sscottl 290111528Sscottlstatic void 291111528Sscottltaskqueue_swi_giant_run(void *dummy) 292111528Sscottl{ 293111528Sscottl taskqueue_run(taskqueue_swi_giant); 294111528Sscottl} 295111528Sscottl 296154333Sscottlint 297154333Sscottltaskqueue_start_threads(struct taskqueue **tqp, int count, int pri, 298154333Sscottl const char *name, ...) 299154333Sscottl{ 300154333Sscottl va_list ap; 301178015Ssam struct thread *td; 302154333Sscottl struct taskqueue *tq; 303178015Ssam int i, error; 304198411Sjhb char ktname[MAXCOMLEN + 1]; 305154333Sscottl 306154333Sscottl if (count <= 0) 307154333Sscottl return (EINVAL); 308178015Ssam 309154333Sscottl tq = *tqp; 310154333Sscottl 311154333Sscottl va_start(ap, name); 312198411Sjhb vsnprintf(ktname, sizeof(ktname), name, ap); 313154333Sscottl va_end(ap); 314154333Sscottl 315178015Ssam tq->tq_threads = malloc(sizeof(struct thread *) * count, M_TASKQUEUE, 316157314Ssam M_NOWAIT | M_ZERO); 317178015Ssam if (tq->tq_threads == NULL) { 318157314Ssam printf("%s: no memory for %s threads\n", __func__, ktname); 319157314Ssam return (ENOMEM); 320157314Ssam } 321157314Ssam 322154333Sscottl for (i = 0; i < count; i++) { 323154333Sscottl if (count == 1) 324178015Ssam error = kthread_add(taskqueue_thread_loop, tqp, NULL, 325178015Ssam &tq->tq_threads[i], RFSTOPPED, 0, ktname); 326154333Sscottl else 327178015Ssam error = kthread_add(taskqueue_thread_loop, tqp, NULL, 328178015Ssam &tq->tq_threads[i], RFSTOPPED, 0, 329178015Ssam "%s_%d", ktname, i); 330158904Ssam if (error) { 331157314Ssam /* should be ok to continue, taskqueue_free will dtrt */ 332178015Ssam printf("%s: kthread_add(%s): error %d", __func__, 333178015Ssam ktname, error); 334178015Ssam tq->tq_threads[i] = NULL; /* paranoid */ 335158904Ssam } else 336178015Ssam tq->tq_tcount++; 337154333Sscottl } 338158904Ssam for (i = 0; i < count; i++) { 339178015Ssam if (tq->tq_threads[i] == NULL) 340158904Ssam continue; 341178015Ssam td = tq->tq_threads[i]; 342170307Sjeff thread_lock(td); 343158904Ssam sched_prio(td, pri); 344166188Sjeff sched_add(td, SRQ_BORING); 345170307Sjeff thread_unlock(td); 346158904Ssam } 347154333Sscottl 348154333Sscottl return (0); 349154333Sscottl} 350154333Sscottl 351133305Sjmgvoid 352133305Sjmgtaskqueue_thread_loop(void *arg) 353119708Sken{ 354133305Sjmg struct taskqueue **tqp, *tq; 355131246Sjhb 356133305Sjmg tqp = arg; 357133305Sjmg tq = *tqp; 358154167Sscottl TQ_LOCK(tq); 359188548Sthompsa while ((tq->tq_flags & TQ_FLAGS_ACTIVE) != 0) { 360133305Sjmg taskqueue_run(tq); 361196293Spjd /* 362196293Spjd * Because taskqueue_run() can drop tq_mutex, we need to 363196293Spjd * check if the TQ_FLAGS_ACTIVE flag wasn't removed in the 364196293Spjd * meantime, which means we missed a wakeup. 365196293Spjd */ 366196293Spjd if ((tq->tq_flags & TQ_FLAGS_ACTIVE) == 0) 367196293Spjd break; 368157815Sjhb TQ_SLEEP(tq, tq, &tq->tq_mutex, 0, "-", 0); 369188592Sthompsa } 370145729Ssam 371145729Ssam /* rendezvous with thread that asked us to terminate */ 372178015Ssam tq->tq_tcount--; 373178015Ssam wakeup_one(tq->tq_threads); 374154167Sscottl TQ_UNLOCK(tq); 375178123Sjhb kthread_exit(); 376119708Sken} 377119708Sken 378133305Sjmgvoid 379119708Skentaskqueue_thread_enqueue(void *context) 380119708Sken{ 381133305Sjmg struct taskqueue **tqp, *tq; 382131246Sjhb 383133305Sjmg tqp = context; 384133305Sjmg tq = *tqp; 385133305Sjmg 386133305Sjmg mtx_assert(&tq->tq_mutex, MA_OWNED); 387145729Ssam wakeup_one(tq); 388119708Sken} 389119708Sken 390188058SimpTASKQUEUE_DEFINE(swi, taskqueue_swi_enqueue, NULL, 391111528Sscottl swi_add(NULL, "task queue", taskqueue_swi_run, NULL, SWI_TQ, 392111528Sscottl INTR_MPSAFE, &taskqueue_ih)); 393111528Sscottl 394188058SimpTASKQUEUE_DEFINE(swi_giant, taskqueue_swi_giant_enqueue, NULL, 395151656Sjhb swi_add(NULL, "Giant taskq", taskqueue_swi_giant_run, 396111528Sscottl NULL, SWI_TQ_GIANT, 0, &taskqueue_giant_ih)); 397119708Sken 398133305SjmgTASKQUEUE_DEFINE_THREAD(thread); 399119789Ssam 400154167Sscottlstruct taskqueue * 401154167Sscottltaskqueue_create_fast(const char *name, int mflags, 402154333Sscottl taskqueue_enqueue_fn enqueue, void *context) 403119789Ssam{ 404154333Sscottl return _taskqueue_create(name, mflags, enqueue, context, 405154167Sscottl MTX_SPIN, "fast_taskqueue"); 406119789Ssam} 407119789Ssam 408154167Sscottl/* NB: for backwards compatibility */ 409154167Sscottlint 410154167Sscottltaskqueue_enqueue_fast(struct taskqueue *queue, struct task *task) 411119789Ssam{ 412154167Sscottl return taskqueue_enqueue(queue, task); 413119789Ssam} 414119789Ssam 415119789Ssamstatic void *taskqueue_fast_ih; 416119789Ssam 417119789Ssamstatic void 418154167Sscottltaskqueue_fast_enqueue(void *context) 419119789Ssam{ 420119789Ssam swi_sched(taskqueue_fast_ih, 0); 421119789Ssam} 422119789Ssam 423119789Ssamstatic void 424119789Ssamtaskqueue_fast_run(void *dummy) 425119789Ssam{ 426154167Sscottl taskqueue_run(taskqueue_fast); 427119789Ssam} 428119789Ssam 429188058SimpTASKQUEUE_FAST_DEFINE(fast, taskqueue_fast_enqueue, NULL, 430154167Sscottl swi_add(NULL, "Fast task queue", taskqueue_fast_run, NULL, 431154167Sscottl SWI_TQ_FAST, INTR_MPSAFE, &taskqueue_fast_ih)); 432196295Spjd 433196295Spjdint 434196295Spjdtaskqueue_member(struct taskqueue *queue, struct thread *td) 435196295Spjd{ 436196295Spjd int i, j, ret = 0; 437196295Spjd 438196295Spjd TQ_LOCK(queue); 439196295Spjd for (i = 0, j = 0; ; i++) { 440196295Spjd if (queue->tq_threads[i] == NULL) 441196295Spjd continue; 442196295Spjd if (queue->tq_threads[i] == td) { 443196295Spjd ret = 1; 444196295Spjd break; 445196295Spjd } 446196295Spjd if (++j >= queue->tq_tcount) 447196295Spjd break; 448196295Spjd } 449196295Spjd TQ_UNLOCK(queue); 450196295Spjd return (ret); 451196295Spjd} 452