Add pt_broadcast_all()

CHANGELOG.md

@@ -32,6 +32,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `at()`, `remove()`, and `pt_queue_wait_new()` let a receiver take one kind
   of message while others wait, which with a tagged union stands in for Go's
   `select`.
+- `pt_broadcast_all(s)`, which wakes every waiting protothread, whatever its
+  channel. Each one re-tests its condition, so it's always safe, and calling
+  it periodically turns a lost wakeup from a hang into a delay.
 
 ### Changed
 

README.md

@@ -60,7 +60,7 @@ cmake -S . -B build && cmake --build build && ./build/pttest
 
 ### The API ###
 
-The core is twenty-one entries in the following two tables, all in `protothread.h`. The `protothread_` prefix is used for operations on the overall protothread object; the `pt_` prefix is for operations involving specific protothreads. Everything else in `protothread.h` is internal and has a `pt_i_` or `PT_I_` prefix. Anything *without* those prefixes is API you can rely on, and anything with one may change in a later release.
+The core is twenty-two entries in the following two tables, all in `protothread.h`. The `protothread_` prefix is used for operations on the overall protothread object; the `pt_` prefix is for operations involving specific protothreads. Everything else in `protothread.h` is internal and has a `pt_i_` or `PT_I_` prefix. Anything *without* those prefixes is API you can rely on, and anything with one may change in a later release.
 
 **The scheduler**
 
@@ -74,6 +74,7 @@ The core is twenty-one entries in the following two tables, all in `protothread.
 | `protothread_set_ready_function(s, f, env)` | called when the run list becomes non-empty |
 | `pt_create(s, thread, func, env)` | create a protothread and make it ready |
 | `pt_broadcast(s, channel)` | make every waiter on `channel` ready; the one to use by default |
+| `pt_broadcast_all(s)` | make every waiter on every channel ready |
 | `pt_signal(s, channel)` | make only the oldest waiter on `channel` ready |
 | `pt_kill(thread)` | unschedule one; true if it was still scheduled |
 | `pt_is_alive(thread)` | has it not yet exited or been killed? never blocks |
@@ -333,13 +334,13 @@ and `protothread_init()` on static storage, a program that uses protothreads (an
 
 **By default this library is not interrupt-safe.** The scheduler's lists are updated with several stores that are not atomic with respect to an interrupt handler. If an interrupt occurs in the middle of one, a protothread can be silently and permanently orphaned. It's removed from its wait list, never placed on the run list, and no future signal can reach it. No assertion fires.
 
-So by default, `pt_signal()`, `pt_broadcast()`, and `pt_kill()` must be called only from thread context, never from an interrupt handler.
+So by default, `pt_signal()`, `pt_broadcast()`, `pt_broadcast_all()`, and `pt_kill()` must be called only from thread context, never from an interrupt handler.
 
 Defining three macros ([how](#defining-the-critical-section-macros)) eliminates that restriction, for `pt_timer_run()` too. It does **not** change it for `protothread_run()`, which must never be called from an interrupt handler. It runs your thread code, and an interrupt handler that re-enters it would run a second protothread on top of the one already running. These are the restrictions:
 
 | | from an interrupt handler |
 |---|---|
-| `pt_signal()`, `pt_broadcast()` | lists stay intact once `PT_CRITICAL_*` are defined |
+| `pt_signal()`, `pt_broadcast()`, `pt_broadcast_all()` | lists stay intact once `PT_CRITICAL_*` are defined |
 | `pt_kill()`, `pt_timer_run()` | lists stay intact once `PT_CRITICAL_*` are defined |
 | `protothread_run()` | **never safe** |
 | `pt_wait()`, `pt_yield()`, `pt_call()` | never -- these only run inside a protothread |
@@ -578,7 +579,9 @@ The debug (testing) define `PT_SIGNAL_WAKES_ALL` checks the other direction. If
 
 The analogy has one limitation. A real spin loop can never miss anything, because it re-tests continuously. `pt_wait()`'s delay ends only when somebody signals, so a missing signal causes a hang. A missing signal could be a plain logic error, or it could come from a timing window (see [Lost wakeups](#lost-wakeups)).
 
-In a large, mission-critical system, consider calling `pt_broadcast()` on _every_ channel periodically. A missed-wakeup bug then causes a delay instead of a hang.
+In a large, mission-critical system, consider calling `pt_broadcast_all()` periodically. It wakes every waiting protothread, so a missed-wakeup bug causes a delay instead of a hang.
+
+It can also be part of an orderly shutdown process. Set a flag that every wait loop also tests, call `pt_broadcast_all()`, and call `protothread_run()` until it returns false. Each protothread wakes, sees the flag, releases what it holds, and returns. This only works for waits you write yourself. A protothread blocked inside a library call, such as a semaphore acquire or a queue receive, re-tests that call's condition and goes back to waiting.
 
 ## Deterministic execution ##
 
@@ -770,7 +773,7 @@ static inline uint32_t pt_critical_enter(void) {
 #define PT_CRITICAL_EXIT(s) __set_PRIMASK(s)
 ```
 
-The critical sections are short and O(1), except that `pt_signal()`, `pt_broadcast()`, and `pt_kill()` walk one wait list, and `pt_sleep()` and `pt_timer_cancel()` walk the timer list. If interrupt latency is critical, keep `PT_NWAIT` large enough that wait lists stay short, or have the interrupt handler set a flag that the main loop turns into a `pt_signal()` at a safe point.
+The critical sections are short and O(1), except that `pt_signal()`, `pt_broadcast()`, and `pt_kill()` walk one wait list, `pt_broadcast_all()` walks each one in its own critical section, and `pt_sleep()` and `pt_timer_cancel()` walk the timer list. If interrupt latency is critical, keep `PT_NWAIT` large enough that wait lists stay short, or have the interrupt handler set a flag that the main loop turns into a `pt_signal()` at a safe point.
 
 ## Compiler requirements ##
 
@@ -864,6 +867,10 @@ What matters is overlap, not which thread. Before the scheduler first runs there
 >
 > Send a signal to the given channel, which wakes up (schedules) all threads waiting on the channel to run in the same order they blocked. If there are no threads waiting, this call has no effect. The signal isn't queued (a channel has no memory). These threads queue behind all ready threads. Analogous to [POSIX pthread\_cond\_broadcast()](http://www.opengroup.org/onlinepubs/009695399/functions/pthread_cond_broadcast.html).
 
+> `void pt_broadcast_all(protothread_t)`
+>
+> Wake every waiting thread, whatever its channel. Each one re-tests its condition, and most go straight back to waiting, so this is always safe. Calling it periodically turns a lost wakeup from a hang into a delay (see [Wait channels](#wait-channels)). It costs time proportional to `PT_NWAIT` plus the number of waiters.
+
 > `void pt_signal(protothread_t, void *channel)`
 >
 > Same as `pt_broadcast()` but wakes up only one (the oldest) waiting thread. That's safe only under conditions that are easy to break, so prefer `pt_broadcast()` (see [Wait channels](#wait-channels)). Analogous to [POSIX pthread\_cond\_signal()](http://www.opengroup.org/onlinepubs/009695399/functions/pthread_cond_signal.html).

TODO.md

@@ -121,11 +121,8 @@ Timers are done (`protothread_timer.h`). Remaining, most useful first:
     argument for building the detector first and the recovery second, and for
     the log line naming the wait site rather than counting sweeps.
 
-    Next step: `pt_broadcast_all(s)`, which makes every waiter on every
-    channel ready. It's the sweep without the timing, so the caller decides
-    when (every N ticks of its own clock) and no channel needs to be known.
-    The end of the README's "Wait channels" section suggests doing this and
-    should point at it once it exists.
+    The recovery half now exists as `pt_broadcast_all(s)`, the sweep without
+    the timing, so the caller decides when. What's left is the detector.
   * **Barrier and countdown latch.** Both tiny; the latch is Go's `WaitGroup`,
     which suits the fan-out/fan-in shape protothreads fall into naturally.
 

protothread.h

@@ -45,6 +45,7 @@
  *   pt_create(s, thread, func, env)  create a protothread and make it ready
  *   pt_signal(s, channel)            make the oldest waiter on channel ready
  *   pt_broadcast(s, channel)         make every waiter on channel ready
+ *   pt_broadcast_all(s)              make every waiter on every channel ready
  *   pt_kill(thread)                  unschedule one; true if it was scheduled
  *   pt_is_alive(thread)              has it not yet exited or been killed?
  *
@@ -126,9 +127,9 @@ typedef void * env_t ;
  *
  * The scheduler's lists are updated with several stores that are not atomic
  * with respect to an interrupt handler. By default this library is NOT
- * interrupt-safe: pt_signal(), pt_broadcast() and pt_kill() must be called
- * from thread context only. Calling them from an interrupt handler can
- * silently and permanently orphan a protothread.
+ * interrupt-safe: pt_signal(), pt_broadcast(), pt_broadcast_all() and
+ * pt_kill() must be called from thread context only. Calling them from an
+ * interrupt handler can silently and permanently orphan a protothread.
  *
  * Define these to disable and restore interrupts to make those calls safe.
  * They nest, so EXIT must restore the saved state rather than
@@ -141,8 +142,9 @@ typedef void * env_t ;
  *   #define PT_CRITICAL_EXIT(s) __set_PRIMASK(s)
  *
  * The critical sections are short and O(1), except that pt_signal(),
- * pt_broadcast() and pt_kill() walk one wait list, and pt_sleep() and
- * pt_timer_cancel() walk the timer list.
+ * pt_broadcast() and pt_kill() walk one wait list, pt_broadcast_all() walks
+ * each in its own critical section, and pt_sleep() and pt_timer_cancel()
+ * walk the timer list.
  */
 #ifndef PT_CRITICAL_T
 #define PT_CRITICAL_T int
@@ -740,6 +742,33 @@ pt_broadcast(protothread_t const s, void * const channel)
     pt_i_wake(s, channel, false) ;
 }
 
+/* Make every waiting thread ready, whatever its channel; each re-tests */
+static inline void
+pt_broadcast_all(protothread_t const s)
+{
+    bool_t notify = false ;
+    int i ;
+
+    for (i = 0; i < PT_NWAIT; i++) {
+        pt_thread_t ** const wq = &s->wait[i] ;
+        /* only a protothread can add a waiter, so an empty bucket stays so */
+        if (*wq == NULL) {
+            continue ;
+        }
+        /* one bucket per critical section, oldest waiter first */
+        const pt_i_critical_t saved = PT_CRITICAL_ENTER() ;
+        while (*wq) {
+            if (pt_i_link_ready(s, pt_i_unlink_oldest(wq))) {
+                notify = true ;
+            }
+        }
+        PT_CRITICAL_EXIT(saved) ;
+    }
+    if (notify) {
+        pt_i_notify_ready(s) ;
+    }
+}
+
 /* This is used to prevent a thread from scheduling again. This can be
  * very dangerous if the thread in question isn't written to expect this
  * operation. Any cleanup is the caller's, when this returns true.

protothread_test.c

@@ -849,6 +849,115 @@ test_ready(void)
 
 /******************************************************************************/
 
+/* pt_broadcast_all() recovers a lost wakeup, and every other kind of waiter
+ * re-tests and goes back to waiting.
+ */
+typedef struct ball_context_s {
+    pt_thread_t pt_thread ;
+    pt_func_t pt_func ;
+    pt_sem_env_t sem_env ;
+    pt_timer_env_t timer_env ;
+    struct ball_global_s * g ;
+    int wakes ;
+} ball_context_t ;
+
+typedef struct ball_global_s {
+    ball_context_t lost, sem, join, sleep ;
+    pt_timers_t timers ;
+    unsigned int sem_value ;
+    int flag ;
+} ball_global_t ;
+
+static pt_t
+ball_lost_thr(env_t const env)
+{
+    ball_context_t * const c = env ;
+    pt_resume(c) ;
+    while (!c->g->flag) {
+        pt_wait(c, &c->g->flag) ;
+        c->wakes++ ;
+    }
+    return PT_DONE ;
+}
+
+static pt_t
+ball_sem_thr(env_t const env)
+{
+    ball_context_t * const c = env ;
+    pt_resume(c) ;
+    pt_sem_acquire(c, &c->sem_env, &c->g->sem_value) ;
+    return PT_DONE ;
+}
+
+static pt_t
+ball_join_thr(env_t const env)
+{
+    ball_context_t * const c = env ;
+    pt_resume(c) ;
+    pt_join(c, &c->g->sem.pt_thread) ;
+    return PT_DONE ;
+}
+
+static pt_t
+ball_sleep_thr(env_t const env)
+{
+    ball_context_t * const c = env ;
+    pt_resume(c) ;
+    pt_sleep(c, &c->timer_env, &c->g->timers, 10) ;
+    return PT_DONE ;
+}
+
+static void
+test_broadcast_all(void)
+{
+    protothread_t const pt = protothread_create() ;
+    ball_global_t g ;
+    int i ;
+
+    memset(&g, 0, sizeof(g)) ;
+    pt_timers_init(&g.timers, 0) ;
+    g.lost.g = g.sem.g = g.join.g = g.sleep.g = &g ;
+
+    /* nothing waiting is a no-op */
+    pt_broadcast_all(pt) ;
+    check(!protothread_run(pt)) ;
+
+    pt_create(pt, &g.lost.pt_thread, ball_lost_thr, &g.lost) ;
+    pt_create(pt, &g.sem.pt_thread, ball_sem_thr, &g.sem) ;
+    pt_create(pt, &g.join.pt_thread, ball_join_thr, &g.join) ;
+    pt_create(pt, &g.sleep.pt_thread, ball_sleep_thr, &g.sleep) ;
+    while (protothread_run(pt)) ;
+
+    /* the condition becomes true, but nobody signals */
+    g.flag = 1 ;
+    protothread_set_ready_function(pt, set_ready, &ready) ;
+    for (i = 0; i < 2; i++) {
+        pt_broadcast_all(pt) ;
+        check(ready) ;          /* notified once, not once per waiter */
+        ready = false ;
+        while (protothread_run(pt)) ;
+        check(!pt_is_alive(&g.lost.pt_thread)) ;
+        check(g.lost.wakes == 1) ;
+        check(pt_is_alive(&g.sem.pt_thread)) ;
+        check(pt_is_alive(&g.join.pt_thread)) ;
+        check(pt_is_alive(&g.sleep.pt_thread)) ;
+    }
+    protothread_set_ready_function(pt, NULL, NULL) ;
+
+    /* the real events still work */
+    g.sem_value = 1 ;
+    pt_broadcast(pt, &g.sem_value) ;
+    pt_timer_run(pt, &g.timers, 10) ;
+    while (protothread_run(pt)) ;
+    check(!pt_is_alive(&g.sem.pt_thread)) ;
+    check(!pt_is_alive(&g.join.pt_thread)) ;
+    check(!pt_is_alive(&g.sleep.pt_thread)) ;
+    check(g.sem_value == 0) ;
+    protothread_free(pt) ;
+}
+
+/******************************************************************************/
+
 typedef struct kill_context_s {
     pt_thread_t pt_thread ;
     pt_func_t pt_func ;
@@ -2223,6 +2332,7 @@ main()
     test_yield() ;
     test_wait() ;
     test_broadcast() ;
+    test_broadcast_all() ;
     test_pc() ;
     test_pc_big() ;
     test_recursive() ;