subr_taskqueue.c revision 210380
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 210380 2010-07-22 17:23:43Z mdf $"); 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 66210377Smdfstatic void taskqueue_run(struct taskqueue *, struct task **); 67210377Smdf 68180588Skmacystatic __inline void 69180588SkmacyTQ_LOCK(struct taskqueue *tq) 70180588Skmacy{ 71180588Skmacy if (tq->tq_spin) 72180588Skmacy mtx_lock_spin(&tq->tq_mutex); 73180588Skmacy else 74180588Skmacy mtx_lock(&tq->tq_mutex); 75180588Skmacy} 76154167Sscottl 77180588Skmacystatic __inline void 78180588SkmacyTQ_UNLOCK(struct taskqueue *tq) 79180588Skmacy{ 80180588Skmacy if (tq->tq_spin) 81180588Skmacy mtx_unlock_spin(&tq->tq_mutex); 82180588Skmacy else 83180588Skmacy mtx_unlock(&tq->tq_mutex); 84180588Skmacy} 85154167Sscottl 86154167Sscottlstatic __inline int 87154167SscottlTQ_SLEEP(struct taskqueue *tq, void *p, struct mtx *m, int pri, const char *wm, 88154167Sscottl int t) 89154167Sscottl{ 90180588Skmacy if (tq->tq_spin) 91154167Sscottl return (msleep_spin(p, m, wm, t)); 92154167Sscottl return (msleep(p, m, pri, wm, t)); 93154167Sscottl} 94154167Sscottl 95154167Sscottlstatic struct taskqueue * 96154167Sscottl_taskqueue_create(const char *name, int mflags, 97145729Ssam taskqueue_enqueue_fn enqueue, void *context, 98154167Sscottl int mtxflags, const char *mtxname) 9961033Sdfr{ 10061033Sdfr struct taskqueue *queue; 101180588Skmacy 10285521Sjhb queue = malloc(sizeof(struct taskqueue), M_TASKQUEUE, mflags | M_ZERO); 10361033Sdfr if (!queue) 104188058Simp return NULL; 105180588Skmacy 10661033Sdfr STAILQ_INIT(&queue->tq_queue); 10761033Sdfr queue->tq_name = name; 10861033Sdfr queue->tq_enqueue = enqueue; 10961033Sdfr queue->tq_context = context; 110180588Skmacy queue->tq_spin = (mtxflags & MTX_SPIN) != 0; 111180588Skmacy queue->tq_flags |= TQ_FLAGS_ACTIVE; 112154167Sscottl mtx_init(&queue->tq_mutex, mtxname, NULL, mtxflags); 11361033Sdfr 11461033Sdfr return queue; 11561033Sdfr} 11661033Sdfr 117154167Sscottlstruct taskqueue * 118154167Sscottltaskqueue_create(const char *name, int mflags, 119154333Sscottl taskqueue_enqueue_fn enqueue, void *context) 120154167Sscottl{ 121154333Sscottl return _taskqueue_create(name, mflags, enqueue, context, 122154167Sscottl MTX_DEF, "taskqueue"); 123154167Sscottl} 124154167Sscottl 125145729Ssam/* 126145729Ssam * Signal a taskqueue thread to terminate. 127145729Ssam */ 128145729Ssamstatic void 129178015Ssamtaskqueue_terminate(struct thread **pp, struct taskqueue *tq) 130145729Ssam{ 131145729Ssam 132178015Ssam while (tq->tq_tcount > 0) { 133154333Sscottl wakeup(tq); 134154333Sscottl TQ_SLEEP(tq, pp, &tq->tq_mutex, PWAIT, "taskqueue_destroy", 0); 135145729Ssam } 136145729Ssam} 137145729Ssam 13861033Sdfrvoid 13961033Sdfrtaskqueue_free(struct taskqueue *queue) 14061033Sdfr{ 14185521Sjhb 142154167Sscottl TQ_LOCK(queue); 143154333Sscottl queue->tq_flags &= ~TQ_FLAGS_ACTIVE; 144210377Smdf taskqueue_run(queue, &queue->tq_running); 145178015Ssam taskqueue_terminate(queue->tq_threads, queue); 14685521Sjhb mtx_destroy(&queue->tq_mutex); 147178015Ssam free(queue->tq_threads, M_TASKQUEUE); 14861033Sdfr free(queue, M_TASKQUEUE); 14961033Sdfr} 15061033Sdfr 15161033Sdfrint 15261033Sdfrtaskqueue_enqueue(struct taskqueue *queue, struct task *task) 15361033Sdfr{ 15461033Sdfr struct task *ins; 15561033Sdfr struct task *prev; 15661033Sdfr 157154167Sscottl TQ_LOCK(queue); 15885560Sjhb 15961033Sdfr /* 16061033Sdfr * Count multiple enqueues. 16161033Sdfr */ 162180588Skmacy if (task->ta_pending) { 16361033Sdfr task->ta_pending++; 164154167Sscottl TQ_UNLOCK(queue); 16561033Sdfr return 0; 16661033Sdfr } 16761033Sdfr 16861033Sdfr /* 16961033Sdfr * Optimise the case when all tasks have the same priority. 17061033Sdfr */ 17164199Shsu prev = STAILQ_LAST(&queue->tq_queue, task, ta_link); 17261033Sdfr if (!prev || prev->ta_priority >= task->ta_priority) { 17361033Sdfr STAILQ_INSERT_TAIL(&queue->tq_queue, task, ta_link); 17461033Sdfr } else { 175188058Simp prev = NULL; 17661033Sdfr for (ins = STAILQ_FIRST(&queue->tq_queue); ins; 17761033Sdfr prev = ins, ins = STAILQ_NEXT(ins, ta_link)) 17861033Sdfr if (ins->ta_priority < task->ta_priority) 17961033Sdfr break; 18061033Sdfr 18161033Sdfr if (prev) 18261033Sdfr STAILQ_INSERT_AFTER(&queue->tq_queue, prev, task, ta_link); 18361033Sdfr else 18461033Sdfr STAILQ_INSERT_HEAD(&queue->tq_queue, task, ta_link); 18561033Sdfr } 18661033Sdfr 18761033Sdfr task->ta_pending = 1; 188180588Skmacy if ((queue->tq_flags & TQ_FLAGS_BLOCKED) == 0) 189177621Sscottl queue->tq_enqueue(queue->tq_context); 190180588Skmacy else 191177621Sscottl queue->tq_flags |= TQ_FLAGS_PENDING; 19285560Sjhb 193154167Sscottl TQ_UNLOCK(queue); 19485560Sjhb 19561033Sdfr return 0; 19661033Sdfr} 19761033Sdfr 19861033Sdfrvoid 199177621Sscottltaskqueue_block(struct taskqueue *queue) 200177621Sscottl{ 201177621Sscottl 202177621Sscottl TQ_LOCK(queue); 203177621Sscottl queue->tq_flags |= TQ_FLAGS_BLOCKED; 204177621Sscottl TQ_UNLOCK(queue); 205177621Sscottl} 206177621Sscottl 207177621Sscottlvoid 208177621Sscottltaskqueue_unblock(struct taskqueue *queue) 209177621Sscottl{ 210177621Sscottl 211177621Sscottl TQ_LOCK(queue); 212177621Sscottl queue->tq_flags &= ~TQ_FLAGS_BLOCKED; 213177621Sscottl if (queue->tq_flags & TQ_FLAGS_PENDING) { 214177621Sscottl queue->tq_flags &= ~TQ_FLAGS_PENDING; 215177621Sscottl queue->tq_enqueue(queue->tq_context); 216177621Sscottl } 217177621Sscottl TQ_UNLOCK(queue); 218177621Sscottl} 219177621Sscottl 220210377Smdfstatic void 221210377Smdftaskqueue_run(struct taskqueue *queue, struct task **tpp) 22261033Sdfr{ 223210380Smdf struct task *task; 224210377Smdf int pending; 22561033Sdfr 226210377Smdf mtx_assert(&queue->tq_mutex, MA_OWNED); 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; 236210377Smdf task->ta_running = tpp; 237210377Smdf *tpp = task; 238154167Sscottl TQ_UNLOCK(queue); 23961033Sdfr 24085560Sjhb task->ta_func(task->ta_context, pending); 24161033Sdfr 242154167Sscottl TQ_LOCK(queue); 243210377Smdf *tpp = NULL; 244208715Szml wakeup(task); 24561033Sdfr } 24661033Sdfr} 24761033Sdfr 248136131Simpvoid 249136131Simptaskqueue_drain(struct taskqueue *queue, struct task *task) 250136131Simp{ 251180588Skmacy if (queue->tq_spin) { /* XXX */ 252154167Sscottl mtx_lock_spin(&queue->tq_mutex); 253210377Smdf while (task->ta_pending != 0 || 254210377Smdf (task->ta_running != NULL && task == *task->ta_running)) { 255154167Sscottl msleep_spin(task, &queue->tq_mutex, "-", 0); 256210377Smdf } 257154167Sscottl mtx_unlock_spin(&queue->tq_mutex); 258154167Sscottl } else { 259154167Sscottl WITNESS_WARN(WARN_GIANTOK | WARN_SLEEPOK, NULL, __func__); 260145729Ssam 261154167Sscottl mtx_lock(&queue->tq_mutex); 262210377Smdf while (task->ta_pending != 0 || 263210377Smdf (task->ta_running != NULL && task == *task->ta_running)) { 264154167Sscottl msleep(task, &queue->tq_mutex, PWAIT, "-", 0); 265210377Smdf } 266154167Sscottl mtx_unlock(&queue->tq_mutex); 267154167Sscottl } 268136131Simp} 269136131Simp 27061033Sdfrstatic void 27161033Sdfrtaskqueue_swi_enqueue(void *context) 27261033Sdfr{ 27388900Sjhb swi_sched(taskqueue_ih, 0); 27461033Sdfr} 27561033Sdfr 27661033Sdfrstatic void 27767551Sjhbtaskqueue_swi_run(void *dummy) 27861033Sdfr{ 279210377Smdf TQ_LOCK(taskqueue_swi); 280210377Smdf taskqueue_run(taskqueue_swi, &taskqueue_swi->tq_running); 281210377Smdf TQ_UNLOCK(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{ 293210377Smdf TQ_LOCK(taskqueue_swi_giant); 294210377Smdf taskqueue_run(taskqueue_swi_giant, &taskqueue_swi_giant->tq_running); 295210377Smdf TQ_UNLOCK(taskqueue_swi_giant); 296111528Sscottl} 297111528Sscottl 298154333Sscottlint 299154333Sscottltaskqueue_start_threads(struct taskqueue **tqp, int count, int pri, 300154333Sscottl const char *name, ...) 301154333Sscottl{ 302154333Sscottl va_list ap; 303178015Ssam struct thread *td; 304154333Sscottl struct taskqueue *tq; 305178015Ssam int i, error; 306198411Sjhb char ktname[MAXCOMLEN + 1]; 307154333Sscottl 308154333Sscottl if (count <= 0) 309154333Sscottl return (EINVAL); 310178015Ssam 311154333Sscottl tq = *tqp; 312154333Sscottl 313154333Sscottl va_start(ap, name); 314198411Sjhb vsnprintf(ktname, sizeof(ktname), name, ap); 315154333Sscottl va_end(ap); 316154333Sscottl 317178015Ssam tq->tq_threads = malloc(sizeof(struct thread *) * count, M_TASKQUEUE, 318157314Ssam M_NOWAIT | M_ZERO); 319178015Ssam if (tq->tq_threads == NULL) { 320157314Ssam printf("%s: no memory for %s threads\n", __func__, ktname); 321157314Ssam return (ENOMEM); 322157314Ssam } 323157314Ssam 324154333Sscottl for (i = 0; i < count; i++) { 325154333Sscottl if (count == 1) 326178015Ssam error = kthread_add(taskqueue_thread_loop, tqp, NULL, 327209062Savg &tq->tq_threads[i], RFSTOPPED, 0, "%s", ktname); 328154333Sscottl else 329178015Ssam error = kthread_add(taskqueue_thread_loop, tqp, NULL, 330178015Ssam &tq->tq_threads[i], RFSTOPPED, 0, 331178015Ssam "%s_%d", ktname, i); 332158904Ssam if (error) { 333157314Ssam /* should be ok to continue, taskqueue_free will dtrt */ 334178015Ssam printf("%s: kthread_add(%s): error %d", __func__, 335178015Ssam ktname, error); 336178015Ssam tq->tq_threads[i] = NULL; /* paranoid */ 337158904Ssam } else 338178015Ssam tq->tq_tcount++; 339154333Sscottl } 340158904Ssam for (i = 0; i < count; i++) { 341178015Ssam if (tq->tq_threads[i] == NULL) 342158904Ssam continue; 343178015Ssam td = tq->tq_threads[i]; 344170307Sjeff thread_lock(td); 345158904Ssam sched_prio(td, pri); 346166188Sjeff sched_add(td, SRQ_BORING); 347170307Sjeff thread_unlock(td); 348158904Ssam } 349154333Sscottl 350154333Sscottl return (0); 351154333Sscottl} 352154333Sscottl 353133305Sjmgvoid 354133305Sjmgtaskqueue_thread_loop(void *arg) 355119708Sken{ 356133305Sjmg struct taskqueue **tqp, *tq; 357210377Smdf struct task *running; 358131246Sjhb 359210377Smdf /* 360210377Smdf * The kernel stack space is globaly addressable, and it would 361210377Smdf * be an error to ask whether a task is running after the 362210377Smdf * taskqueue has been released. So it is safe to have the 363210377Smdf * task point back to an address in the taskqueue's stack to 364210377Smdf * determine if the task is running. 365210377Smdf */ 366210377Smdf running = NULL; 367210377Smdf 368133305Sjmg tqp = arg; 369133305Sjmg tq = *tqp; 370154167Sscottl TQ_LOCK(tq); 371188548Sthompsa while ((tq->tq_flags & TQ_FLAGS_ACTIVE) != 0) { 372210377Smdf taskqueue_run(tq, &running); 373196293Spjd /* 374196293Spjd * Because taskqueue_run() can drop tq_mutex, we need to 375196293Spjd * check if the TQ_FLAGS_ACTIVE flag wasn't removed in the 376196293Spjd * meantime, which means we missed a wakeup. 377196293Spjd */ 378196293Spjd if ((tq->tq_flags & TQ_FLAGS_ACTIVE) == 0) 379196293Spjd break; 380157815Sjhb TQ_SLEEP(tq, tq, &tq->tq_mutex, 0, "-", 0); 381188592Sthompsa } 382145729Ssam 383145729Ssam /* rendezvous with thread that asked us to terminate */ 384178015Ssam tq->tq_tcount--; 385178015Ssam wakeup_one(tq->tq_threads); 386154167Sscottl TQ_UNLOCK(tq); 387178123Sjhb kthread_exit(); 388119708Sken} 389119708Sken 390133305Sjmgvoid 391119708Skentaskqueue_thread_enqueue(void *context) 392119708Sken{ 393133305Sjmg struct taskqueue **tqp, *tq; 394131246Sjhb 395133305Sjmg tqp = context; 396133305Sjmg tq = *tqp; 397133305Sjmg 398133305Sjmg mtx_assert(&tq->tq_mutex, MA_OWNED); 399145729Ssam wakeup_one(tq); 400119708Sken} 401119708Sken 402188058SimpTASKQUEUE_DEFINE(swi, taskqueue_swi_enqueue, NULL, 403111528Sscottl swi_add(NULL, "task queue", taskqueue_swi_run, NULL, SWI_TQ, 404111528Sscottl INTR_MPSAFE, &taskqueue_ih)); 405111528Sscottl 406188058SimpTASKQUEUE_DEFINE(swi_giant, taskqueue_swi_giant_enqueue, NULL, 407151656Sjhb swi_add(NULL, "Giant taskq", taskqueue_swi_giant_run, 408111528Sscottl NULL, SWI_TQ_GIANT, 0, &taskqueue_giant_ih)); 409119708Sken 410133305SjmgTASKQUEUE_DEFINE_THREAD(thread); 411119789Ssam 412154167Sscottlstruct taskqueue * 413154167Sscottltaskqueue_create_fast(const char *name, int mflags, 414154333Sscottl taskqueue_enqueue_fn enqueue, void *context) 415119789Ssam{ 416154333Sscottl return _taskqueue_create(name, mflags, enqueue, context, 417154167Sscottl MTX_SPIN, "fast_taskqueue"); 418119789Ssam} 419119789Ssam 420154167Sscottl/* NB: for backwards compatibility */ 421154167Sscottlint 422154167Sscottltaskqueue_enqueue_fast(struct taskqueue *queue, struct task *task) 423119789Ssam{ 424154167Sscottl return taskqueue_enqueue(queue, task); 425119789Ssam} 426119789Ssam 427119789Ssamstatic void *taskqueue_fast_ih; 428119789Ssam 429119789Ssamstatic void 430154167Sscottltaskqueue_fast_enqueue(void *context) 431119789Ssam{ 432119789Ssam swi_sched(taskqueue_fast_ih, 0); 433119789Ssam} 434119789Ssam 435119789Ssamstatic void 436119789Ssamtaskqueue_fast_run(void *dummy) 437119789Ssam{ 438210377Smdf TQ_LOCK(taskqueue_fast); 439210377Smdf taskqueue_run(taskqueue_fast, &taskqueue_fast->tq_running); 440210377Smdf TQ_UNLOCK(taskqueue_fast); 441119789Ssam} 442119789Ssam 443188058SimpTASKQUEUE_FAST_DEFINE(fast, taskqueue_fast_enqueue, NULL, 444154167Sscottl swi_add(NULL, "Fast task queue", taskqueue_fast_run, NULL, 445154167Sscottl SWI_TQ_FAST, INTR_MPSAFE, &taskqueue_fast_ih)); 446196295Spjd 447196295Spjdint 448196295Spjdtaskqueue_member(struct taskqueue *queue, struct thread *td) 449196295Spjd{ 450196295Spjd int i, j, ret = 0; 451196295Spjd 452196295Spjd TQ_LOCK(queue); 453196295Spjd for (i = 0, j = 0; ; i++) { 454196295Spjd if (queue->tq_threads[i] == NULL) 455196295Spjd continue; 456196295Spjd if (queue->tq_threads[i] == td) { 457196295Spjd ret = 1; 458196295Spjd break; 459196295Spjd } 460196295Spjd if (++j >= queue->tq_tcount) 461196295Spjd break; 462196295Spjd } 463196295Spjd TQ_UNLOCK(queue); 464196295Spjd return (ret); 465196295Spjd} 466