README: a barrier built from pt_wait() and pt_broadcast(), with a test

README.md

@@ -548,7 +548,7 @@ If **A** calls **B**, and then calls **C** after **B** returns, **B** and **C**
 
 There is no associated mutex because the scheduler is non-preemptive. `pthread_cond_wait()` needs one to make "test the predicate" and "suspend" a single indivisible step. Protothreads can't interleave (except explicitly), so that step is already indivisible. (But this problem can exist with interrupts; see [Lost wakeups](#lost-wakeups)).
 
-**It is universal.** Every common blocking synchronization primitive can be built on top of it. The semaphores in `protothread_sem.h`, the reader-writer lock in `protothread_lock.h`, the sleeps in `protothread_timer.h`, and the queues in `protothread_queue.h` are all ordinary protothread code using `pt_wait()` and `pt_broadcast()`, with no privileged access to the scheduler. If you need a barrier or a latch, you write it the same way, in your own code, without patching the library.
+**It is universal.** Every common blocking synchronization primitive can be built on top of it. The semaphores in `protothread_sem.h`, the reader-writer lock in `protothread_lock.h`, the sleeps in `protothread_timer.h`, and the queues in `protothread_queue.h` are all ordinary protothread code using `pt_wait()` and `pt_broadcast()`, with no privileged access to the scheduler. If you need a barrier or a latch, you write it the same way, in your own code, without patching the library (see [Example - a barrier](#example---a-barrier)).
 
 **The mental model stays simple.** A protothread waits for a condition to be *true*, not for an event to *happen*. In electrical engineering, this is level-triggered versus edge-triggered. Level-triggered means a condition can be re-tested at any time. An event (edge-triggered) can be missed. So the simplest correct way to wait is to spin on the predicate:
 
@@ -584,6 +584,34 @@ In a large, mission-critical system, consider calling `pt_broadcast_all()` perio
 
 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.
 
+### Example - a barrier ###
+
+A barrier holds protothreads until `n` of them have arrived, then lets them all go. With `n` of 2 it's a rendezvous. Two protothreads meet, and neither continues until the other gets there. They can exchange data this way, too. Each writes its part before arriving and reads the other's part after. If they meet again, alternate between two sets of slots by round. Otherwise one may write its next part before the other has read this one.
+
+```c
+typedef struct {
+    int n;              /* how many must arrive */
+    int arrived;
+    unsigned int gen;   /* incremented each time it opens */
+} barrier_t;
+
+/* arrive, in a protothread function whose context has a gen */
+if (++b->arrived < b->n) {
+    c->gen = b->gen;
+    while (b->gen == c->gen) {
+        pt_wait(c, &b->gen);
+    }
+} else {
+    b->arrived = 0;
+    b->gen++;
+    pt_broadcast(pt_get_pt(c), &b->gen);
+}
+```
+
+The waiters test `gen` rather than `arrived`, and here's why. Say `n` is 3. A and B arrive and wait. C arrives last, opens the barrier, and keeps running. A and B are now ready, but they won't run until C yields or waits. Before then, C may come back around and arrive for the next round, and so may D. Now `arrived` is 2. If A and B tested `arrived`, they'd see a round in progress, go back to waiting, and could wait forever. Instead they find that `gen` has changed since they arrived, and they go on, which is correct.
+
+`gen` is only compared for equality, so it's fine for it to wrap around. It would have to advance 2^32 times before A and B run, and it can't. The run list is first-in, first-out, so A and B run ahead of any thread readied after them.
+
 ## Deterministic execution ##
 
 When multiple POSIX threads are ready to run, the scheduler runs them in a random, or at least unpredictable, order, even if there's a single processor. After choosing a thread to run, it may be preempted at any time, with execution resuming on another thread (timeslicing). In contrast, an application built on protothreads, like event-driven software, runs deterministically if all other events are deterministic too. This allows one to write pseudo-random tests that can reliably reproduce bugs. You start the test with a randomly chosen random number generator seed (based on wall-clock time, for example). If a bug is found during the run, you can start the test again with the same seed, perhaps with more tracing or new assertions to catch the problem earlier. The test follows the same sequence of states, so it reproduces the bug. This often lets you verify a proposed fix too. That doesn't work if the fix changes the execution sequence enough to invalidate the seed, but sometimes you can "timebomb" the fix, making it active only just before the failure.

TODO.md

@@ -121,8 +121,9 @@ Timers (`protothread_timer.h`) and message queues (`protothread_queue.h`, in
 
     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.
+  * **Countdown latch.** Tiny; it's Go's `WaitGroup`, which suits the
+    fan-out/fan-in shape protothreads fall into naturally. (A barrier is shown
+    in README.md, "Example - a barrier".)
   * **Rendezvous.** Go's unbuffered channel, as its own object rather than a
     queue of capacity 0 (doc/queue.md says why queues can't be). The item
     stays in the sender's context, so there's no buffer. The sender posts a

protothread_test.c

@@ -401,6 +401,132 @@ test_pc_big(void)
 
 #undef N
 
+/* barrier ********************************************************************/
+
+/* The barrier from README.md, as a nested protothread function. With n of 2
+ * it's a rendezvous.
+ */
+typedef struct {
+    int n ;                 /* how many must arrive */
+    int arrived ;
+    unsigned int gen ;      /* incremented each time it opens */
+} barrier_t ;
+
+typedef struct barrier_arrive_context_s {
+    pt_func_t pt_func ;
+    unsigned int gen ;
+} barrier_arrive_context_t ;
+
+static pt_t
+barrier_arrive(barrier_arrive_context_t * const c, barrier_t * const b)
+{
+    pt_resume(c) ;
+
+    if (++b->arrived < b->n) {
+        c->gen = b->gen ;
+        while (b->gen == c->gen) {
+            pt_wait(c, &b->gen) ;
+        }
+    } else {
+        b->arrived = 0 ;
+        b->gen++ ;
+        pt_broadcast(pt_get_pt(c), &b->gen) ;
+    }
+    return PT_DONE ;
+}
+
+#define BARRIER_N 3
+#define BARRIER_ROUNDS 5
+
+typedef struct barrier_context_s {
+    pt_thread_t pt_thread ;
+    pt_func_t pt_func ;
+    barrier_arrive_context_t arrive ;
+    barrier_t * b ;
+    int * round ;           /* every thread's current round */
+    int id ;
+    int r ;
+    int i ;
+} barrier_context_t ;
+
+static pt_t
+barrier_thr(env_t const env)
+{
+    barrier_context_t * const c = env ;
+    pt_resume(c) ;
+
+    for (c->r = 0; c->r < BARRIER_ROUNDS; c->r++) {
+        c->round[c->id] = c->r ;
+        /* stagger, so the last to arrive differs from round to round */
+        for (c->i = 0; c->i < (c->id + c->r) % BARRIER_N; c->i++) {
+            pt_yield(c) ;
+        }
+        pt_call(c, barrier_arrive, &c->arrive, c->b) ;
+        /* everyone has reached this round, and nobody is past the next */
+        for (int i = 0; i < BARRIER_N; i++) {
+            check(c->round[i] == c->r || c->round[i] == c->r + 1) ;
+        }
+    }
+    return PT_DONE ;
+}
+
+/* a rendezvous exchanging a value each round */
+typedef struct exchange_context_s {
+    pt_thread_t pt_thread ;
+    pt_func_t pt_func ;
+    barrier_arrive_context_t arrive ;
+    barrier_t * b ;
+    int (* slot)[2] ;       /* two sets, so writing round r+1 can't clobber r */
+    int id ;
+    int r ;
+} exchange_context_t ;
+
+static pt_t
+exchange_thr(env_t const env)
+{
+    exchange_context_t * const c = env ;
+    pt_resume(c) ;
+
+    for (c->r = 0; c->r < BARRIER_ROUNDS; c->r++) {
+        c->slot[c->r % 2][c->id] = 100 * c->id + c->r ;
+        pt_call(c, barrier_arrive, &c->arrive, c->b) ;
+        check(c->slot[c->r % 2][!c->id] == 100 * !c->id + c->r) ;
+    }
+    return PT_DONE ;
+}
+
+static void
+test_barrier(void)
+{
+    protothread_t const pt = protothread_create() ;
+    barrier_t b = { .n = BARRIER_N } ;
+    int round[BARRIER_N] ;
+    barrier_context_t c[BARRIER_N] ;
+    barrier_t b2 = { .n = 2 } ;
+    int slot[2][2] ;
+    exchange_context_t x[2] ;
+
+    for (int i = 0; i < BARRIER_N; i++) {
+        c[i] = (barrier_context_t){ .b = &b, .round = round, .id = i } ;
+        pt_create(pt, &c[i].pt_thread, barrier_thr, &c[i]) ;
+    }
+    for (int i = 0; i < 2; i++) {
+        x[i] = (exchange_context_t){ .b = &b2, .slot = slot, .id = i } ;
+        pt_create(pt, &x[i].pt_thread, exchange_thr, &x[i]) ;
+    }
+    while (protothread_run(pt)) {
+    }
+    for (int i = 0; i < BARRIER_N; i++) {
+        check(!pt_is_alive(&c[i].pt_thread)) ;
+    }
+    for (int i = 0; i < 2; i++) {
+        check(!pt_is_alive(&x[i].pt_thread)) ;
+    }
+    check(b.gen == BARRIER_ROUNDS && b.arrived == 0) ;
+    check(b2.gen == BARRIER_ROUNDS && b2.arrived == 0) ;
+    protothread_free(pt) ;
+}
+
 /* recursive ******************************************************************/
 
 /* count to 4096 the hard way */
@@ -2453,6 +2579,7 @@ main()
     test_broadcast_all() ;
     test_pc() ;
     test_pc_big() ;
+    test_barrier() ;
     test_recursive() ;
     test_sem() ;
     test_lock() ;