Implement the delayed task execution extension to the taskqueue

mechanism. The caller may specify a timeout in ticks after which the
task will be scheduled.

Sponsored by:	The FreeBSD Foundation
Reviewed by:	jeff, jhb
MFC after:	1 month

1 files changed, 115 insertions, 13 deletions

diff --git a/sys/kern/subr_taskqueue.c b/sys/kern/subr_taskqueue.c
index 20b6a0b..4c45899 100644
--- a/sys/kern/subr_taskqueue.c
+++ b/sys/kern/subr_taskqueue.c
@@ -61,12 +61,15 @@ struct taskqueue {
 	int			tq_tcount;
 	int			tq_spin;
 	int			tq_flags;
+	int			tq_callouts;
 };
 
 #define	TQ_FLAGS_ACTIVE		(1 << 0)
 #define	TQ_FLAGS_BLOCKED	(1 << 1)
 #define	TQ_FLAGS_PENDING	(1 << 2)
 
+#define	DT_CALLOUT_ARMED	(1 << 0)
+
 #define	TQ_LOCK(tq)							\
 	do {								\
 		if ((tq)->tq_spin)					\
@@ -83,6 +86,17 @@ struct taskqueue {
 			mtx_unlock(&(tq)->tq_mutex);			\
 	} while (0)
 
+void
+_timeout_task_init(struct taskqueue *queue, struct timeout_task *timeout_task,
+    int priority, task_fn_t func, void *context)
+{
+
+	TASK_INIT(&timeout_task->t, priority, func, context);
+	callout_init_mtx(&timeout_task->c, &queue->tq_mutex, 0);
+	timeout_task->q = queue;
+	timeout_task->f = 0;
+}
+
 static __inline int
 TQ_SLEEP(struct taskqueue *tq, void *p, struct mtx *m, int pri, const char *wm,
     int t)
@@ -129,7 +143,7 @@ static void
 taskqueue_terminate(struct thread **pp, struct taskqueue *tq)
 {
 
-	while (tq->tq_tcount > 0) {
+	while (tq->tq_tcount > 0 || tq->tq_callouts > 0) {
 		wakeup(tq);
 		TQ_SLEEP(tq, pp, &tq->tq_mutex, PWAIT, "taskqueue_destroy", 0);
 	}
@@ -143,26 +157,24 @@ taskqueue_free(struct taskqueue *queue)
 	queue->tq_flags &= ~TQ_FLAGS_ACTIVE;
 	taskqueue_terminate(queue->tq_threads, queue);
 	KASSERT(TAILQ_EMPTY(&queue->tq_active), ("Tasks still running?"));
+	KASSERT(queue->tq_callouts == 0, ("Armed timeout tasks"));
 	mtx_destroy(&queue->tq_mutex);
 	free(queue->tq_threads, M_TASKQUEUE);
 	free(queue, M_TASKQUEUE);
 }
 
-int
-taskqueue_enqueue(struct taskqueue *queue, struct task *task)
+static int
+taskqueue_enqueue_locked(struct taskqueue *queue, struct task *task)
 {
 	struct task *ins;
 	struct task *prev;
 
-	TQ_LOCK(queue);
-
 	/*
 	 * Count multiple enqueues.
 	 */
 	if (task->ta_pending) {
 		task->ta_pending++;
-		TQ_UNLOCK(queue);
-		return 0;
+		return (0);
 	}
 
 	/*
@@ -190,9 +202,60 @@ taskqueue_enqueue(struct taskqueue *queue, struct task *task)
 	else
 		queue->tq_flags |= TQ_FLAGS_PENDING;
 
+	return (0);
+}
+int
+taskqueue_enqueue(struct taskqueue *queue, struct task *task)
+{
+	int res;
+
+	TQ_LOCK(queue);
+	res = taskqueue_enqueue_locked(queue, task);
 	TQ_UNLOCK(queue);
 
-	return 0;
+	return (res);
+}
+
+static void
+taskqueue_timeout_func(void *arg)
+{
+	struct taskqueue *queue;
+	struct timeout_task *timeout_task;
+
+	timeout_task = arg;
+	queue = timeout_task->q;
+	KASSERT((timeout_task->f & DT_CALLOUT_ARMED) != 0, ("Stray timeout"));
+	timeout_task->f &= ~DT_CALLOUT_ARMED;
+	queue->tq_callouts--;
+	taskqueue_enqueue_locked(timeout_task->q, &timeout_task->t);
+}
+
+int
+taskqueue_enqueue_timeout(struct taskqueue *queue,
+    struct timeout_task *timeout_task, int ticks)
+{
+	int res;
+
+	TQ_LOCK(queue);
+	KASSERT(timeout_task->q == NULL || timeout_task->q == queue,
+	    ("Migrated queue"));
+	KASSERT(!queue->tq_spin, ("Timeout for spin-queue"));
+	timeout_task->q = queue;
+	res = timeout_task->t.ta_pending;
+	if (ticks == 0) {
+		taskqueue_enqueue_locked(queue, &timeout_task->t);
+	} else {
+		if ((timeout_task->f & DT_CALLOUT_ARMED) != 0) {
+			res++;
+		} else {
+			queue->tq_callouts++;
+			timeout_task->f |= DT_CALLOUT_ARMED;
+		}
+		callout_reset(&timeout_task->c, ticks, taskqueue_timeout_func,
+		    timeout_task);
+	}
+	TQ_UNLOCK(queue);
+	return (res);
 }
 
 void
@@ -271,6 +334,19 @@ task_is_running(struct taskqueue *queue, struct task *task)
 	return (0);
 }
 
+static int
+taskqueue_cancel_locked(struct taskqueue *queue, struct task *task,
+    u_int *pendp)
+{
+
+	if (task->ta_pending > 0)
+		STAILQ_REMOVE(&queue->tq_queue, task, task, ta_link);
+	if (pendp != NULL)
+		*pendp = task->ta_pending;
+	task->ta_pending = 0;
+	return (task_is_running(queue, task) ? EBUSY : 0);
+}
+
 int
 taskqueue_cancel(struct taskqueue *queue, struct task *task, u_int *pendp)
 {
@@ -278,14 +354,31 @@ taskqueue_cancel(struct taskqueue *queue, struct task *task, u_int *pendp)
 	int error;
 
 	TQ_LOCK(queue);
-	if ((pending = task->ta_pending) > 0)
-		STAILQ_REMOVE(&queue->tq_queue, task, task, ta_link);
-	task->ta_pending = 0;
-	error = task_is_running(queue, task) ? EBUSY : 0;
+	pending = task->ta_pending;
+	error = taskqueue_cancel_locked(queue, task, pendp);
+	TQ_UNLOCK(queue);
+
+	return (error);
+}
+
+int
+taskqueue_cancel_timeout(struct taskqueue *queue,
+    struct timeout_task *timeout_task, u_int *pendp)
+{
+	u_int pending, pending1;
+	int error;
+
+	TQ_LOCK(queue);
+	pending = !!callout_stop(&timeout_task->c);
+	error = taskqueue_cancel_locked(queue, &timeout_task->t, &pending1);
+	if ((timeout_task->f & DT_CALLOUT_ARMED) != 0) {
+		timeout_task->f &= ~DT_CALLOUT_ARMED;
+		queue->tq_callouts--;
+	}
 	TQ_UNLOCK(queue);
 
 	if (pendp != NULL)
-		*pendp = pending;
+		*pendp = pending + pending1;
 	return (error);
 }
 
@@ -302,6 +395,15 @@ taskqueue_drain(struct taskqueue *queue, struct task *task)
 	TQ_UNLOCK(queue);
 }
 
+void
+taskqueue_drain_timeout(struct taskqueue *queue,
+    struct timeout_task *timeout_task)
+{
+
+	callout_drain(&timeout_task->c);
+	taskqueue_drain(queue, &timeout_task->t);
+}
+
 static void
 taskqueue_swi_enqueue(void *context)
 {