Wake each thread at most once a push, end streams only of stopped apps, and show a subscriber costs writes nothing (019)

1 files changed, 14 insertions, 4 deletions

diff --git a/serve/serve.c b/serve/serve.c
index 4e62d3d..f4eea46 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -2476,7 +2476,7 @@ static void subscriber_event(Host *host, Subscriber *s, uint32_t events) {
  * group's value at least. */
 static void push_group(Host *host, Group *g) {
     App *app = g->app;
-    if (!atomic_load(&app->live) || atomic_load(&app->gone)) { close_group(host, g); return; }
+    if (atomic_load(&app->gone)) { close_group(host, g); return; }
     const Tokens *tokens = atomic_load_explicit(&app->tokens, memory_order_acquire);
     for (Subscriber *s = g->first, *after; s; s = after) { /* a tokens file changed: who may still read */
         after = s->next;
@@ -2486,7 +2486,12 @@ static void push_group(Host *host, Group *g) {
             unsubscribe(host, s);
         } else s->checked = tokens;
     }
+    /* Quiet (taking no request: its module changing, or stopping) it waits
+     * for the next word; stopped (quiet and its snapshot let go) its streams end. */
+    bool quiet = !atomic_load(&app->live);
     PureSnapshot *snapshot = read_published(host, app);
+    if (quiet && !snapshot) { close_group(host, g); return; }
+    if (quiet) { pure_snapshot_release(snapshot); return; }
     if (!snapshot) return; /* none yet: the next publication wakes this */
     PureApp *reader = &host->readers[app - apps];
     if (reader->module != g->route.module) { /* the app changed module: the view of that name, if it is the same kind */
@@ -3429,10 +3434,15 @@ static void *run_shard(void *context) {
             drive(c);
         }
         serve_bursts(host);
-        bool told = atomic_exchange_explicit(&host->push_due, false, memory_order_acq_rel);
-        if (host->groups && (told || host->push_waiting)) { /* at most one pass every --push-every-ms */
+        /* At most one pass every --push-every-ms; the word stays given until
+         * one runs, so owners write to this thread at most once a pass. */
+        bool told = atomic_load_explicit(&host->push_due, memory_order_acquire);
+        if (host->groups && (told || host->push_waiting)) {
             host->push_waiting = host->now - host->pushed_at < push_every_ms;
-            if (!host->push_waiting) push_pass(host);
+            if (!host->push_waiting) {
+                atomic_store_explicit(&host->push_due, false, memory_order_release); /* a word given meanwhile runs the next */
+                push_pass(host);
+            }
         }
         uint64_t now = host->now;
         if (now >= next_sweep) {