Add functions and macros and refactor code to make it easier to manage
fast taskqueues. The following have been added:
TASKQUEUE_FAST_DEFINE() - create a global task queue.
an arbitrary execution context.
TASKQUEUE_FAST_DEFINE_THREAD() - create a global taskqueue that uses a
dedicated kthread.
taskqueue_create_fast() - create a local/private taskqueue.
These are all complimentary of the standard taskqueue functions. They are
primarily useful for fast interrupt handlers that can only use spinlock for
synchronization.
I personally think that the taskqueue API is starting to get too narrow and
hairy, but fixing it will require a major redesign on the API. Such a
redesign would be good but would break compatibility with FreeBSD 6.x, so
it really isn't desirable at this time.
Submitted by: sam
This commit is contained in:
scottl
parent
81ed190221
commit
706bc421be
@ -55,10 +55,38 @@ struct taskqueue {
|
||||
struct task *tq_running;
|
||||
struct mtx tq_mutex;
|
||||
struct proc **tq_pproc;
|
||||
int tq_spin;
|
||||
};
|
||||
|
||||
static __inline void
|
||||
TQ_LOCK(struct taskqueue *tq)
|
||||
{
|
||||
if (tq->tq_spin)
|
||||
mtx_lock_spin(&tq->tq_mutex);
|
||||
else
|
||||
mtx_lock(&tq->tq_mutex);
|
||||
}
|
||||
|
||||
static __inline void
|
||||
TQ_UNLOCK(struct taskqueue *tq)
|
||||
{
|
||||
if (tq->tq_spin)
|
||||
mtx_unlock_spin(&tq->tq_mutex);
|
||||
else
|
||||
mtx_unlock(&tq->tq_mutex);
|
||||
}
|
||||
|
||||
static void init_taskqueue_list(void *data);
|
||||
|
||||
static __inline int
|
||||
TQ_SLEEP(struct taskqueue *tq, void *p, struct mtx *m, int pri, const char *wm,
|
||||
int t)
|
||||
{
|
||||
if (tq->tq_spin)
|
||||
return (msleep_spin(p, m, wm, t));
|
||||
return (msleep(p, m, pri, wm, t));
|
||||
}
|
||||
|
||||
static void
|
||||
init_taskqueue_list(void *data __unused)
|
||||
{
|
||||
@ -69,10 +97,11 @@ init_taskqueue_list(void *data __unused)
|
||||
SYSINIT(taskqueue_list, SI_SUB_INTRINSIC, SI_ORDER_ANY, init_taskqueue_list,
|
||||
NULL);
|
||||
|
||||
struct taskqueue *
|
||||
taskqueue_create(const char *name, int mflags,
|
||||
static struct taskqueue *
|
||||
_taskqueue_create(const char *name, int mflags,
|
||||
taskqueue_enqueue_fn enqueue, void *context,
|
||||
struct proc **pp)
|
||||
struct proc **pp,
|
||||
int mtxflags, const char *mtxname)
|
||||
{
|
||||
struct taskqueue *queue;
|
||||
|
||||
@ -85,7 +114,8 @@ taskqueue_create(const char *name, int mflags,
|
||||
queue->tq_enqueue = enqueue;
|
||||
queue->tq_context = context;
|
||||
queue->tq_pproc = pp;
|
||||
mtx_init(&queue->tq_mutex, "taskqueue", NULL, MTX_DEF);
|
||||
queue->tq_spin = (mtxflags & MTX_SPIN) != 0;
|
||||
mtx_init(&queue->tq_mutex, mtxname, NULL, mtxflags);
|
||||
|
||||
mtx_lock(&taskqueue_queues_mutex);
|
||||
STAILQ_INSERT_TAIL(&taskqueue_queues, queue, tq_link);
|
||||
@ -94,6 +124,15 @@ taskqueue_create(const char *name, int mflags,
|
||||
return queue;
|
||||
}
|
||||
|
||||
struct taskqueue *
|
||||
taskqueue_create(const char *name, int mflags,
|
||||
taskqueue_enqueue_fn enqueue, void *context,
|
||||
struct proc **pp)
|
||||
{
|
||||
return _taskqueue_create(name, mflags, enqueue, context, pp,
|
||||
MTX_DEF, "taskqueue");
|
||||
}
|
||||
|
||||
/*
|
||||
* Signal a taskqueue thread to terminate.
|
||||
*/
|
||||
@ -106,11 +145,11 @@ taskqueue_terminate(struct proc **pp, struct taskqueue *tq)
|
||||
*pp = NULL;
|
||||
if (p) {
|
||||
wakeup_one(tq);
|
||||
PROC_LOCK(p); /* NB: insure we don't miss wakeup */
|
||||
mtx_unlock(&tq->tq_mutex); /* let taskqueue thread run */
|
||||
msleep(p, &p->p_mtx, PWAIT, "taskqueue_destroy", 0);
|
||||
PROC_LOCK(p); /* NB: insure we don't miss wakeup */
|
||||
TQ_UNLOCK(tq); /* let taskqueue thread run */
|
||||
TQ_SLEEP(tq, p, &p->p_mtx, PWAIT, "taskqueue_destroy", 0);
|
||||
PROC_UNLOCK(p);
|
||||
mtx_lock(&tq->tq_mutex);
|
||||
TQ_LOCK(tq);
|
||||
}
|
||||
}
|
||||
|
||||
@ -122,7 +161,7 @@ taskqueue_free(struct taskqueue *queue)
|
||||
STAILQ_REMOVE(&taskqueue_queues, queue, taskqueue, tq_link);
|
||||
mtx_unlock(&taskqueue_queues_mutex);
|
||||
|
||||
mtx_lock(&queue->tq_mutex);
|
||||
TQ_LOCK(queue);
|
||||
taskqueue_run(queue);
|
||||
taskqueue_terminate(queue->tq_pproc, queue);
|
||||
mtx_destroy(&queue->tq_mutex);
|
||||
@ -140,7 +179,7 @@ taskqueue_find(const char *name)
|
||||
mtx_lock(&taskqueue_queues_mutex);
|
||||
STAILQ_FOREACH(queue, &taskqueue_queues, tq_link) {
|
||||
if (strcmp(queue->tq_name, name) == 0) {
|
||||
mtx_lock(&queue->tq_mutex);
|
||||
TQ_LOCK(queue);
|
||||
mtx_unlock(&taskqueue_queues_mutex);
|
||||
return queue;
|
||||
}
|
||||
@ -155,14 +194,14 @@ taskqueue_enqueue(struct taskqueue *queue, struct task *task)
|
||||
struct task *ins;
|
||||
struct task *prev;
|
||||
|
||||
mtx_lock(&queue->tq_mutex);
|
||||
TQ_LOCK(queue);
|
||||
|
||||
/*
|
||||
* Count multiple enqueues.
|
||||
*/
|
||||
if (task->ta_pending) {
|
||||
task->ta_pending++;
|
||||
mtx_unlock(&queue->tq_mutex);
|
||||
TQ_UNLOCK(queue);
|
||||
return 0;
|
||||
}
|
||||
|
||||
@ -188,7 +227,7 @@ taskqueue_enqueue(struct taskqueue *queue, struct task *task)
|
||||
task->ta_pending = 1;
|
||||
queue->tq_enqueue(queue->tq_context);
|
||||
|
||||
mtx_unlock(&queue->tq_mutex);
|
||||
TQ_UNLOCK(queue);
|
||||
|
||||
return 0;
|
||||
}
|
||||
@ -201,7 +240,7 @@ taskqueue_run(struct taskqueue *queue)
|
||||
|
||||
owned = mtx_owned(&queue->tq_mutex);
|
||||
if (!owned)
|
||||
mtx_lock(&queue->tq_mutex);
|
||||
TQ_LOCK(queue);
|
||||
while (STAILQ_FIRST(&queue->tq_queue)) {
|
||||
/*
|
||||
* Carefully remove the first task from the queue and
|
||||
@ -212,11 +251,11 @@ taskqueue_run(struct taskqueue *queue)
|
||||
pending = task->ta_pending;
|
||||
task->ta_pending = 0;
|
||||
queue->tq_running = task;
|
||||
mtx_unlock(&queue->tq_mutex);
|
||||
TQ_UNLOCK(queue);
|
||||
|
||||
task->ta_func(task->ta_context, pending);
|
||||
|
||||
mtx_lock(&queue->tq_mutex);
|
||||
TQ_LOCK(queue);
|
||||
queue->tq_running = NULL;
|
||||
wakeup(task);
|
||||
}
|
||||
@ -226,18 +265,25 @@ taskqueue_run(struct taskqueue *queue)
|
||||
* on entry, although this opens a race window.
|
||||
*/
|
||||
if (!owned)
|
||||
mtx_unlock(&queue->tq_mutex);
|
||||
TQ_UNLOCK(queue);
|
||||
}
|
||||
|
||||
void
|
||||
taskqueue_drain(struct taskqueue *queue, struct task *task)
|
||||
{
|
||||
WITNESS_WARN(WARN_GIANTOK | WARN_SLEEPOK, NULL, "taskqueue_drain");
|
||||
if (queue->tq_spin) { /* XXX */
|
||||
mtx_lock_spin(&queue->tq_mutex);
|
||||
while (task->ta_pending != 0 || task == queue->tq_running)
|
||||
msleep_spin(task, &queue->tq_mutex, "-", 0);
|
||||
mtx_unlock_spin(&queue->tq_mutex);
|
||||
} else {
|
||||
WITNESS_WARN(WARN_GIANTOK | WARN_SLEEPOK, NULL, __func__);
|
||||
|
||||
mtx_lock(&queue->tq_mutex);
|
||||
while (task->ta_pending != 0 || task == queue->tq_running)
|
||||
msleep(task, &queue->tq_mutex, PWAIT, "-", 0);
|
||||
mtx_unlock(&queue->tq_mutex);
|
||||
mtx_lock(&queue->tq_mutex);
|
||||
while (task->ta_pending != 0 || task == queue->tq_running)
|
||||
msleep(task, &queue->tq_mutex, PWAIT, "-", 0);
|
||||
mtx_unlock(&queue->tq_mutex);
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
@ -271,15 +317,15 @@ taskqueue_thread_loop(void *arg)
|
||||
|
||||
tqp = arg;
|
||||
tq = *tqp;
|
||||
mtx_lock(&tq->tq_mutex);
|
||||
TQ_LOCK(tq);
|
||||
do {
|
||||
taskqueue_run(tq);
|
||||
msleep(tq, &tq->tq_mutex, PWAIT, "-", 0);
|
||||
TQ_SLEEP(tq, tq, &tq->tq_mutex, curthread->td_priority, "-", 0);
|
||||
} while (*tq->tq_pproc != NULL);
|
||||
|
||||
/* rendezvous with thread that asked us to terminate */
|
||||
wakeup_one(tq);
|
||||
mtx_unlock(&tq->tq_mutex);
|
||||
TQ_UNLOCK(tq);
|
||||
kthread_exit(0);
|
||||
}
|
||||
|
||||
@ -305,80 +351,26 @@ TASKQUEUE_DEFINE(swi_giant, taskqueue_swi_giant_enqueue, 0,
|
||||
|
||||
TASKQUEUE_DEFINE_THREAD(thread);
|
||||
|
||||
struct taskqueue *
|
||||
taskqueue_create_fast(const char *name, int mflags,
|
||||
taskqueue_enqueue_fn enqueue, void *context,
|
||||
struct proc **pp)
|
||||
{
|
||||
return _taskqueue_create(name, mflags, enqueue, context, pp,
|
||||
MTX_SPIN, "fast_taskqueue");
|
||||
}
|
||||
|
||||
/* NB: for backwards compatibility */
|
||||
int
|
||||
taskqueue_enqueue_fast(struct taskqueue *queue, struct task *task)
|
||||
{
|
||||
struct task *ins;
|
||||
struct task *prev;
|
||||
|
||||
mtx_lock_spin(&queue->tq_mutex);
|
||||
|
||||
/*
|
||||
* Count multiple enqueues.
|
||||
*/
|
||||
if (task->ta_pending) {
|
||||
task->ta_pending++;
|
||||
mtx_unlock_spin(&queue->tq_mutex);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/*
|
||||
* Optimise the case when all tasks have the same priority.
|
||||
*/
|
||||
prev = STAILQ_LAST(&queue->tq_queue, task, ta_link);
|
||||
if (!prev || prev->ta_priority >= task->ta_priority) {
|
||||
STAILQ_INSERT_TAIL(&queue->tq_queue, task, ta_link);
|
||||
} else {
|
||||
prev = 0;
|
||||
for (ins = STAILQ_FIRST(&queue->tq_queue); ins;
|
||||
prev = ins, ins = STAILQ_NEXT(ins, ta_link))
|
||||
if (ins->ta_priority < task->ta_priority)
|
||||
break;
|
||||
|
||||
if (prev)
|
||||
STAILQ_INSERT_AFTER(&queue->tq_queue, prev, task, ta_link);
|
||||
else
|
||||
STAILQ_INSERT_HEAD(&queue->tq_queue, task, ta_link);
|
||||
}
|
||||
|
||||
task->ta_pending = 1;
|
||||
queue->tq_enqueue(queue->tq_context);
|
||||
|
||||
mtx_unlock_spin(&queue->tq_mutex);
|
||||
|
||||
return 0;
|
||||
return taskqueue_enqueue(queue, task);
|
||||
}
|
||||
|
||||
static void
|
||||
taskqueue_run_fast(struct taskqueue *queue)
|
||||
{
|
||||
struct task *task;
|
||||
int pending;
|
||||
|
||||
mtx_lock_spin(&queue->tq_mutex);
|
||||
while (STAILQ_FIRST(&queue->tq_queue)) {
|
||||
/*
|
||||
* Carefully remove the first task from the queue and
|
||||
* zero its pending count.
|
||||
*/
|
||||
task = STAILQ_FIRST(&queue->tq_queue);
|
||||
STAILQ_REMOVE_HEAD(&queue->tq_queue, ta_link);
|
||||
pending = task->ta_pending;
|
||||
task->ta_pending = 0;
|
||||
mtx_unlock_spin(&queue->tq_mutex);
|
||||
|
||||
task->ta_func(task->ta_context, pending);
|
||||
|
||||
mtx_lock_spin(&queue->tq_mutex);
|
||||
}
|
||||
mtx_unlock_spin(&queue->tq_mutex);
|
||||
}
|
||||
|
||||
struct taskqueue *taskqueue_fast;
|
||||
static void *taskqueue_fast_ih;
|
||||
|
||||
static void
|
||||
taskqueue_fast_schedule(void *context)
|
||||
taskqueue_fast_enqueue(void *context)
|
||||
{
|
||||
swi_sched(taskqueue_fast_ih, 0);
|
||||
}
|
||||
@ -386,31 +378,9 @@ taskqueue_fast_schedule(void *context)
|
||||
static void
|
||||
taskqueue_fast_run(void *dummy)
|
||||
{
|
||||
taskqueue_run_fast(taskqueue_fast);
|
||||
taskqueue_run(taskqueue_fast);
|
||||
}
|
||||
|
||||
static void
|
||||
taskqueue_define_fast(void *arg)
|
||||
{
|
||||
|
||||
taskqueue_fast = malloc(sizeof(struct taskqueue), M_TASKQUEUE,
|
||||
M_NOWAIT | M_ZERO);
|
||||
if (!taskqueue_fast) {
|
||||
printf("%s: Unable to allocate fast task queue!\n", __func__);
|
||||
return;
|
||||
}
|
||||
|
||||
STAILQ_INIT(&taskqueue_fast->tq_queue);
|
||||
taskqueue_fast->tq_name = "fast";
|
||||
taskqueue_fast->tq_enqueue = taskqueue_fast_schedule;
|
||||
mtx_init(&taskqueue_fast->tq_mutex, "taskqueue_fast", NULL, MTX_SPIN);
|
||||
|
||||
mtx_lock(&taskqueue_queues_mutex);
|
||||
STAILQ_INSERT_TAIL(&taskqueue_queues, taskqueue_fast, tq_link);
|
||||
mtx_unlock(&taskqueue_queues_mutex);
|
||||
|
||||
swi_add(NULL, "Fast taskq", taskqueue_fast_run,
|
||||
NULL, SWI_TQ_FAST, INTR_MPSAFE, &taskqueue_fast_ih);
|
||||
}
|
||||
SYSINIT(taskqueue_fast, SI_SUB_CONFIGURE, SI_ORDER_SECOND,
|
||||
taskqueue_define_fast, NULL);
|
||||
TASKQUEUE_FAST_DEFINE(fast, taskqueue_fast_enqueue, 0,
|
||||
swi_add(NULL, "Fast task queue", taskqueue_fast_run, NULL,
|
||||
SWI_TQ_FAST, INTR_MPSAFE, &taskqueue_fast_ih));
|
||||
|
||||
@ -80,7 +80,7 @@ void taskqueue_thread_enqueue(void *context);
|
||||
extern struct taskqueue *taskqueue_##name
|
||||
|
||||
/*
|
||||
* Define and initialise a taskqueue.
|
||||
* Define and initialise a global taskqueue that uses sleep mutexes.
|
||||
*/
|
||||
#define TASKQUEUE_DEFINE(name, enqueue, context, init) \
|
||||
\
|
||||
@ -105,6 +105,33 @@ TASKQUEUE_DEFINE(name, taskqueue_thread_enqueue, &taskqueue_##name, \
|
||||
kthread_create(taskqueue_thread_loop, &taskqueue_##name, \
|
||||
&taskqueue_##name##_proc, 0, 0, #name " taskq"))
|
||||
|
||||
/*
|
||||
* Define and initialise a global taskqueue that uses spin mutexes.
|
||||
*/
|
||||
#define TASKQUEUE_FAST_DEFINE(name, enqueue, context, init) \
|
||||
\
|
||||
struct taskqueue *taskqueue_##name; \
|
||||
\
|
||||
static void \
|
||||
taskqueue_define_##name(void *arg) \
|
||||
{ \
|
||||
static struct proc *taskqueue_##name##_proc; \
|
||||
taskqueue_##name = \
|
||||
taskqueue_create_fast(#name, M_NOWAIT, (enqueue), (context),\
|
||||
&taskqueue_##name##_proc); \
|
||||
init; \
|
||||
} \
|
||||
\
|
||||
SYSINIT(taskqueue_##name, SI_SUB_CONFIGURE, SI_ORDER_SECOND, \
|
||||
taskqueue_define_##name, NULL) \
|
||||
\
|
||||
struct __hack
|
||||
#define TASKQUEUE_FAST_DEFINE_THREAD(name) \
|
||||
TASKQUEUE_FAST_DEFINE(name, taskqueue_thread_enqueue, \
|
||||
&taskqueue_##name, kthread_create(taskqueue_thread_loop, \
|
||||
&taskqueue_##name, &taskqueue_##name##_proc, 0, 0, \
|
||||
#name " fast taskq"))
|
||||
|
||||
/*
|
||||
* These queues are serviced by software interrupt handlers. To enqueue
|
||||
* a task, call taskqueue_enqueue(taskqueue_swi, &task) or
|
||||
@ -127,5 +154,8 @@ TASKQUEUE_DECLARE(thread);
|
||||
*/
|
||||
TASKQUEUE_DECLARE(fast);
|
||||
int taskqueue_enqueue_fast(struct taskqueue *queue, struct task *task);
|
||||
struct taskqueue *taskqueue_create_fast(const char *name, int mflags,
|
||||
taskqueue_enqueue_fn enqueue,
|
||||
void *context, struct proc **);
|
||||
|
||||
#endif /* !_SYS_TASKQUEUE_H_ */
|
||||
|
||||
Loading…
Reference in New Issue
Block a user