Merge pull request #6238 from pauluap/break_dispatch_flag

Remove windup behavior from break_dispatch

events/equeue/equeue.c

@@ -73,7 +73,7 @@ int equeue_create_inplace(equeue_t *q, size_t size, void *buffer) {
     q->queue = 0;
     q->tick = equeue_tick();
     q->generation = 0;
-    q->breaks = 0;
+    q->break_requested = false;
 
     q->background.active = false;
     q->background.update = 0;
@@ -366,7 +366,7 @@ void equeue_cancel(equeue_t *q, int id) {
 
 void equeue_break(equeue_t *q) {
     equeue_mutex_lock(&q->queuelock);
-    q->breaks++;
+    q->break_requested = true;
     equeue_mutex_unlock(&q->queuelock);
     equeue_sema_signal(&q->eventsema);
 }
@@ -418,6 +418,7 @@ void equeue_dispatch(equeue_t *q, int ms) {
                     q->background.active = true;
                     equeue_mutex_unlock(&q->queuelock);
                 }
+                q->break_requested = false;
                 return;
             }
         }
@@ -436,10 +437,10 @@ void equeue_dispatch(equeue_t *q, int ms) {
         equeue_sema_wait(&q->eventsema, deadline);
 
         // check if we were notified to break out of dispatch
-        if (q->breaks) {
+        if (q->break_requested) {
             equeue_mutex_lock(&q->queuelock);
-            if (q->breaks > 0) {
-                q->breaks--;
+            if (q->break_requested) {
+                q->break_requested = false;
                 equeue_mutex_unlock(&q->queuelock);
                 return;
             }

events/equeue/equeue.h

@@ -58,7 +58,7 @@ struct equeue_event {
 typedef struct equeue {
     struct equeue_event *queue;
     unsigned tick;
-    unsigned breaks;
+    bool break_requested;
     uint8_t generation;
 
     unsigned char *buffer;

events/equeue/tests/tests.c

@@ -391,6 +391,26 @@ void break_test(void) {
     equeue_destroy(&q);
 }
 
+void break_no_windup_test(void) {
+    equeue_t q;
+    int err = equeue_create(&q, 2048);
+    test_assert(!err);
+
+    int count = 0;
+    equeue_call_every(&q, 0, simple_func, &count);
+
+    equeue_break(&q);
+    equeue_break(&q);
+    equeue_dispatch(&q, -1);
+    test_assert(count == 1);
+
+    count = 0;
+    equeue_dispatch(&q, 55);
+    test_assert(count > 1);
+
+    equeue_destroy(&q);
+}
+
 void period_test(void) {
     equeue_t q;
     int err = equeue_create(&q, 2048);
@@ -687,6 +707,43 @@ void multithreaded_barrage_test(int N) {
     equeue_destroy(&q);
 }
 
+struct count_and_queue
+{
+    int p;
+    equeue_t* q;
+};
+
+void simple_breaker(void *p) {
+    struct count_and_queue* caq = (struct count_and_queue*)p;
+    equeue_break(caq->q);
+    usleep(10000);
+    caq->p++;
+}
+
+void break_request_cleared_on_timeout(void) {
+    equeue_t q;
+    int err = equeue_create(&q, 2048);
+    test_assert(!err);
+
+    struct count_and_queue pq;
+    pq.p = 0;
+    pq.q = &q;
+
+    int id = equeue_call_every(&q, 10, simple_breaker, &pq);
+
+    equeue_dispatch(&q, 10);
+    test_assert(pq.p == 1);
+
+    equeue_cancel(&q, id);
+
+    int count = 0;
+    equeue_call_every(&q, 10, simple_func, &count);
+
+    equeue_dispatch(&q, 55);
+    test_assert(count > 1);
+
+    equeue_destroy(&q);
+}
 
 int main() {
     printf("beginning tests...\n");
@@ -702,6 +759,7 @@ int main() {
     test_run(cancel_unnecessarily_test);
     test_run(loop_protect_test);
     test_run(break_test);
+    test_run(break_no_windup_test);
     test_run(period_test);
     test_run(nested_test);
     test_run(sloth_test);
@@ -712,6 +770,7 @@ int main() {
     test_run(simple_barrage_test, 20);
     test_run(fragmenting_barrage_test, 20);
     test_run(multithreaded_barrage_test, 20);
+    test_run(break_request_cleared_on_timeout);
 
     printf("done!\n");
     return test_failure;