Added the ability to background an event queue onto a single-shot timer

The equeue_background function requires only a user-provided update
function, which is called to indicate when the queue should be
dispatched.

The equeue_background function adds a new level of composability to the
event queue library. Allowing equeues to be easily backgrounded on
hardware timers or even be combined with existing event loops.

Author: geky

equeue.c

@@ -39,6 +39,10 @@ int equeue_create_inplace(equeue_t *q, size_t size, void *buffer) {
     q->generation = 0;
     q->breaks = 0;
 
+    q->background.active = false;
+    q->background.update = 0;
+    q->background.timer = 0;
+
     int err;
     err = equeue_sema_create(&q->eventsema);
     if (err < 0) {
@@ -68,6 +72,10 @@ void equeue_destroy(equeue_t *q) {
         }
     }
 
+    if (q->background.update) {
+        q->background.update(q->background.timer, -1);
+    }
+
     equeue_mutex_destroy(&q->memlock);
     equeue_mutex_destroy(&q->queuelock);
     equeue_sema_destroy(&q->eventsema);
@@ -199,6 +207,11 @@ static int equeue_enqueue(equeue_t *q, struct equeue_event *e, unsigned ms) {
     *p = e;
     e->ref = p;
 
+    if ((q->background.update && q->background.active) &&
+        (q->queue == e && !e->sibling)) {
+        q->background.update(q->background.timer, ms);
+    }
+
     equeue_mutex_unlock(&q->queuelock);
 
     return id;
@@ -312,6 +325,7 @@ void equeue_break(equeue_t *q) {
 void equeue_dispatch(equeue_t *q, int ms) {
     unsigned tick = equeue_tick();
     unsigned timeout = tick + ms;
+    q->background.active = false;
 
     while (1) {
         // collect all the available events and next deadline
@@ -344,6 +358,16 @@ void equeue_dispatch(equeue_t *q, int ms) {
         if (ms >= 0) {
             deadline = equeue_tickdiff(timeout, tick);
             if (deadline <= 0) {
+                // update background timer if necessary
+                if (q->background.update) {
+                    equeue_mutex_lock(&q->queuelock);
+                    if (q->background.update && q->queue) {
+                        q->background.update(q->background.timer,
+                                equeue_tickdiff(q->queue->target, tick));
+                    }
+                    q->background.active = true;
+                    equeue_mutex_unlock(&q->queuelock);
+                }
                 return;
             }
         }
@@ -441,3 +465,22 @@ int equeue_call_every(equeue_t *q, int ms, void (*cb)(void*), void *data) {
     e->data = data;
     return equeue_post(q, ecallback_dispatch, e);
 }
+
+// backgrounding
+void equeue_background(equeue_t *q,
+        void (*update)(void *timer, int ms), void *timer) {
+    equeue_mutex_lock(&q->queuelock);
+    if (q->background.update) {
+        q->background.update(q->background.timer, -1);
+    }
+
+    q->background.update = update;
+    q->background.timer = timer;
+
+    if (q->background.update && q->queue) {
+        q->background.update(q->background.timer,
+                equeue_tickdiff(q->queue->target, equeue_tick()));
+    }
+    q->background.active = true;
+    equeue_mutex_unlock(&q->queuelock);
+}

equeue.h

@@ -58,6 +58,12 @@ typedef struct equeue {
         unsigned char *data;
     } slab;
 
+    struct equeue_background {
+        bool active;
+        void (*update)(void *timer, int ms);
+        void *timer;
+    } background;
+
     equeue_sema_t eventsema;
     equeue_mutex_t queuelock;
     equeue_mutex_t memlock;
@@ -136,6 +142,14 @@ int equeue_post(equeue_t *queue, void (*cb)(void *), void *event);
 // stop a currently executing event
 void equeue_cancel(equeue_t *queue, int event);
 
+// Background an event queue onto a single-shot timer
+//
+// The provided update function will be called to indicate when the queue
+// should be dispatched. A negative timeout will be passed to the update
+// function when the timer is no longer needed.
+void equeue_background(equeue_t *queue,
+        void (*update)(void *timer, int ms), void *timer);
+
 
 #ifdef __cplusplus
 }

tests/tests.c

@@ -479,6 +479,37 @@ void multithread_test(void) {
     equeue_destroy(&q);
 }
 
+void background_func(void *p, int ms) {
+    *(unsigned *)p = ms;
+}
+
+void background_test(void) {
+    equeue_t q;
+    int err = equeue_create(&q, 2048);
+    test_assert(!err);
+
+    int id = equeue_call_in(&q, 20, pass_func, 0);
+    test_assert(id);
+
+    unsigned ms;
+    equeue_background(&q, background_func, &ms);
+    test_assert(ms == 20);
+
+    id = equeue_call_in(&q, 10, pass_func, 0);
+    test_assert(id);
+    test_assert(ms == 10);
+
+    id = equeue_call(&q, pass_func, 0);
+    test_assert(id);
+    test_assert(ms == 0);
+
+    equeue_dispatch(&q, 0);
+    test_assert(ms == 10);
+
+    equeue_destroy(&q);
+    test_assert(ms == -1);
+}
+
 // Barrage tests
 void simple_barrage_test(int N) {
     equeue_t q;
@@ -589,6 +620,7 @@ int main() {
     test_run(period_test);
     test_run(nested_test);
     test_run(sloth_test);
+    test_run(background_test);
     test_run(multithread_test);
     test_run(simple_barrage_test, 20);
     test_run(fragmenting_barrage_test, 20);