Move callout execution to top-half
MODIFY
include/callout.h
include/callout.h
@@ -16,7 +16,8 @@ typedef struct callout {
unsigned c_index; /* index of bucket this callout is assigned to */
} callout_t;
-#define CALLOUT_ACTIVE 0x0001 /* callout is currently being executed */
+/* callout has been delegated to callout thread and will be executed soon */
+#define CALLOUT_ACTIVE 0x0001
#define CALLOUT_PENDING 0x0002 /* callout is waiting for timeout */
/*
@@ -33,13 +34,16 @@ void callout_setup_relative(callout_t *handle, systime_t time, timeout_t fn,
void *arg);
/*
- * Delete a callout from the queue without handling it.
+ * Cancel a callout if it is currently pending.
+ *
+ * \return True if the callout was pending and has been stopped, false if the
+ * callout has already been delegated to callout thread or executed.
*/
-void callout_stop(callout_t *handle);
+bool callout_stop(callout_t *handle);
/*
- * Process all callouts that happened since last time.
- * A callout is removed from the queue when handled.
+ * Process all callouts that happened since last time and delegate them to
+ * callout thread.
*/
void callout_process(systime_t now);

MODIFY
sys/callout.c
sys/callout.c
#include <spinlock.h>
#include <sysinit.h>
#include <sleepq.h>
+#include <thread.h>
+#include <sched.h>
#include <interrupt.h>
/* Note: If the difference in time between ticks is greater than the number of
@@ -40,6 +42,32 @@ static inline callout_list_t *ci_list(int i) {
return &ci.heads[i];
}
+static callout_list_t delegated;
+
+static void callout_thread(void *arg) {
+ while (true) {
+ callout_t *elem;
+
+ WITH_INTR_DISABLED {
+ while (TAILQ_EMPTY(&delegated)) {
+ sleepq_wait(&delegated, NULL);
+ }
+
+ elem = TAILQ_FIRST(&delegated);
+ TAILQ_REMOVE(&delegated, elem, c_link);
+ }
+
+ assert(callout_is_active(elem));
+ assert(!callout_is_pending(elem));
+
+ /* Execute callout's function. */
+ elem->c_func(elem->c_arg);
+ callout_clear_active(elem);
+ /* Wake threads that wait for execution of this callout in callout_drain. */
+ sleepq_broadcast(elem);
+ }
+}
+
static void callout_init(void) {
bzero(&ci, sizeof(ci));
@@ -47,12 +75,19 @@ static void callout_init(void) {
for (int i = 0; i < CALLOUT_BUCKETS; i++)
TAILQ_INIT(ci_list(i));
+
+ TAILQ_INIT(&delegated);
+
+ thread_t *td =
+ thread_create("callout-thread", callout_thread, NULL, prio_kthread(0));
+ sched_add(td);
}
static void _callout_setup(callout_t *handle, systime_t time, timeout_t fn,
void *arg) {
assert(spin_owned(&ci.lock));
assert(!callout_is_pending(handle));
+ assert(!callout_is_active(handle));
int index = time % CALLOUT_BUCKETS;
@@ -81,7 +116,7 @@ void callout_setup_relative(callout_t *handle, systime_t time, timeout_t fn,
_callout_setup(handle, now + time, fn, arg);
}
-void callout_stop(callout_t *handle) {
+bool callout_stop(callout_t *handle) {
SCOPED_SPIN_LOCK(&ci.lock);
klog("Remove callout {%p} at %ld.", handle, handle->c_time);
@@ -89,12 +124,15 @@ void callout_stop(callout_t *handle) {
if (callout_is_pending(handle)) {
callout_clear_pending(handle);
TAILQ_REMOVE(ci_list(handle->c_index), handle, c_link);
+ return true;
}
+
+ return false;
}
/*
- * Handle all timeouted callouts from queues between last position and current
- * position.
+ * Process all timeouted callouts from queues between last position and current
+ * position and delegate them to callout thread.
*/
void callout_process(systime_t time) {
unsigned int last_bucket;
@@ -108,8 +146,6 @@ void callout_process(systime_t time) {
last_bucket = time % CALLOUT_BUCKETS;
}
- callout_list_t detached;
-
/* We are in kernel's bottom half. */
assert(intr_disabled());
@@ -117,28 +153,17 @@ void callout_process(systime_t time) {
callout_list_t *head = ci_list(current_bucket);
callout_t *elem, *next;
- TAILQ_INIT(&detached);
-
- /* Detach triggered callouts from the queue. */
+ /* Detach triggered callouts from ci_list queue. */
TAILQ_FOREACH_SAFE(elem, head, c_link, next) {
if (elem->c_time <= time) {
callout_set_active(elem);
callout_clear_pending(elem);
TAILQ_REMOVE(head, elem, c_link);
- TAILQ_INSERT_TAIL(&detached, elem, c_link);
+ /* Attach elem to callout thread's queue. */
+ TAILQ_INSERT_TAIL(&delegated, elem, c_link);
}
}
- /* Now safely call them one by one. */
- TAILQ_FOREACH_SAFE(elem, &detached, c_link, next) {
- TAILQ_REMOVE(&detached, elem, c_link);
- elem->c_func(elem->c_arg);
- /* Wake threads that wait for execution of this callout in function
- * callout_drain. */
- sleepq_broadcast(elem);
- callout_clear_active(elem);
- }
-
if (current_bucket == last_bucket)
break;
@@ -146,16 +171,16 @@ void callout_process(systime_t time) {
}
ci.last = time;
+
+ /* Wake callout thread. */
+ if (!TAILQ_EMPTY(&delegated)) {
+ sleepq_signal(&delegated);
+ }
}
bool callout_drain(callout_t *handle) {
- /* A callout may be in active state only in callout_process,
- * which is called in bottom half (with interrupts disabled),
- * so we can't notice it. */
- assert(!callout_is_active(handle));
-
WITH_INTR_DISABLED {
- if (callout_is_pending(handle)) {
+ if (callout_is_pending(handle) || callout_is_active(handle)) {
sleepq_wait(handle, NULL);
return true;
}
@@ -164,4 +189,4 @@ bool callout_drain(callout_t *handle) {
return false;
}
-SYSINIT_ADD(callout, callout_init, NODEPS);
+SYSINIT_ADD(callout, callout_init, DEPS("sched"));

MODIFY
tests/callout.c
tests/callout.c
#include <ktest.h>
#include <interrupt.h>
-static void periodic_callout(void *arg) {
- callout_t *callout = arg;
- callout_setup_relative(callout, 1, periodic_callout, arg);
-}
-
-static int test_callout_sync(void) {
- const int K = 50;
- const int N = 7;
-
- callout_t callout[N];
- bzero(callout, sizeof(callout_t) * N);
-
- for (int j = 0; j < K; j++) {
- for (int i = 0; i < N; i++)
- callout_setup_relative(&callout[i], 1, periodic_callout, &callout[i]);
-
- for (int i = 0; i < N; i++)
- callout_stop(&callout[i]);
- }
-
- return KTEST_SUCCESS;
-}
-
static int counter;
/* This test verifies whether callouts work at all. */
@@ -59,12 +36,9 @@ static int order[ORDER_N] = {2, 5, 4, 6, 9, 0, 8, 1, 3, 7};
static int current;
static void callout_ordered(void *arg) {
- /* There is no race condition here as callouts run in bottom half (with
- * interrupts disabled). */
- assert(intr_disabled());
-
int ord = (int)arg;
assert(current == ord);
+ /* There is no race condition here since callouts don't run concurrently. */
current++;
}
@@ -119,7 +93,6 @@ static int test_callout_drain(void) {
return KTEST_SUCCESS;
}
-KTEST_ADD(callout_sync, test_callout_sync, 0);
KTEST_ADD(callout_simple, test_callout_simple, 0);
KTEST_ADD(callout_order, test_callout_order, 0);
KTEST_ADD(callout_stop, test_callout_stop, 0);

MODIFY
tests/sleepq.c
tests/sleepq.c
static volatile int wakeups;
-static void test_thread(void *p) {
- while (wakeups < (int)p) {
+static void test_thread(void *expected) {
+ while (wakeups < (int)expected) {
WITH_NO_PREEMPTION {
wakeups++;
}
@@ -16,23 +16,22 @@ static void test_thread(void *p) {
}
}
-static void periodic_callout(void *arg) {
- callout_t *callout = arg;
- callout_setup_relative(callout, 1, periodic_callout, callout);
+static void wake_threads_up(void *arg) {
sleepq_broadcast(&test_thread);
}
static int test_sleepq_sync(void) {
- const int K = 5;
- const int N = 20;
+ const int K = 5; /* number of test threads */
+ const int N = 20; /* number of expected wakeups */
wakeups = 0;
- callout_t callout;
- bzero(&callout, sizeof(callout_t));
+ callout_t callout[N];
+ bzero(callout, sizeof(callout_t) * N);
thread_t *td[K];
- callout_setup_relative(&callout, 1, periodic_callout, &callout);
+ for (int i = 0; i < N; i++)
+ callout_setup_relative(&callout[i], i + 1, wake_threads_up, NULL);
for (int i = 0; i < K; i++) {
td[i] =
@@ -43,7 +42,8 @@ static int test_sleepq_sync(void) {
for (int i = 0; i < K; i++)
thread_join(td[i]);
- callout_stop(&callout);
+ for (int i = 0; i < N; i++)
+ callout_drain(&callout[i]);
return KTEST_SUCCESS;
}

Move callout execution to top-half (#485)