Wait on the monotonic clock everywhere a deadline is waited for

Review 004's timer fix (103) counted the deadlines on the monotonic
clock, but the owner still slept until a wall-clock time: with the
clock set back an hour a node's timers would stop for that hour. So
would a wire's pause between tries, and a lone commit after a group
would wait an hour for the others to join it. The test's LD_PRELOAD
moves only what the process reads, so there each wait ended at once
instead, and the timers went on ticking while the owner spun a core.

core/wait.h makes conditions that wait on the monotonic clock
(pthread_condattr_setclock; macOS keeps the wall clock) and deadlines
on it. The owner's waits, the wire's pauses and the group commit's
gathering use them.

clock-back.so can now move the clock forward too (CLOCK_SHIFT_S), and
check.py server_clocks moves it an hour each way under a node with a
timer and a wire to a port no one listens on, and under a document
server whose three threads commit together, then one alone. Before:
back, 159 % CPU and 50,886 tries in 1.2 s; forward, 1 tick, 1 try, and
a lone commit that never ended. After: 59 ticks, 0 % CPU, 4 tries and a
lone commit of 2.6 ms, both ways.

6 files changed, 146 insertions, 30 deletions

diff --git a/core/wait.h b/core/wait.h
new file mode 100644
index 0000000..241eaef
--- /dev/null
+++ b/core/wait.h
@@ -0,0 +1,44 @@
+#ifndef PURE_WAIT_H
+#define PURE_WAIT_H
+/* Timed waits that count on the monotonic clock. A deadline read from the
+ * wall clock moves when the clock is set (by a person, or NTP): set back an
+ * hour, a wait begun before lasts an hour more; set forward, it ends at once.
+ * A condition made by pure_cond_monotonic waits until a time pure_wait_until
+ * gives, and neither happens. (macOS has no clock for conditions: there both
+ * keep to the wall clock.) */
+#include <pthread.h>
+#include <time.h>
+
+#ifdef __APPLE__
+#define PURE_WAIT_CLOCK CLOCK_REALTIME
+#else
+#define PURE_WAIT_CLOCK CLOCK_MONOTONIC
+#endif
+
+static inline int pure_cond_monotonic(pthread_cond_t *cond) {
+#ifdef __APPLE__
+    return pthread_cond_init(cond, NULL);
+#else
+    pthread_condattr_t attr;
+    int result = pthread_condattr_init(&attr);
+    if (result) return result;
+    result = pthread_condattr_setclock(&attr, CLOCK_MONOTONIC);
+    if (!result) result = pthread_cond_init(cond, &attr);
+    pthread_condattr_destroy(&attr);
+    return result;
+#endif
+}
+
+/* The time ns from now, on the clock pure_cond_monotonic's conditions wait on. */
+static inline struct timespec pure_wait_until(long long ns) {
+    struct timespec t;
+    clock_gettime(PURE_WAIT_CLOCK, &t);
+    t.tv_sec += (time_t)(ns / 1000000000);
+    t.tv_nsec += (long)(ns % 1000000000);
+    if (t.tv_nsec >= 1000000000) {
+        ++t.tv_sec;
+        t.tv_nsec -= 1000000000;
+    }
+    return t;
+}
+#endif
diff --git a/serve/deliver.c b/serve/deliver.c
index 68c5f14..511a2cf 100644
--- a/serve/deliver.c
+++ b/serve/deliver.c
@@ -1,5 +1,6 @@
 #define _POSIX_C_SOURCE 200809L
 #include "deliver.h"
+#include "wait.h"
 #include "seal.h"
 #include "message-json.h"
 #include "module-text.h"
@@ -222,7 +223,7 @@ PureWire *pure_wire_open(const PureModule *m, const char *output, const char *ur
     w->tls = tls;
     w->answer_input = -1;
     pthread_mutex_init(&w->lock, NULL);
-    pthread_cond_init(&w->changed, NULL);
+    pure_cond_monotonic(&w->changed);
     return w;
 }
 
@@ -738,11 +739,7 @@ static void *deliver(void *context) {
         }
         pause_ms = outcome != AGAIN ? LONGEST_PAUSE_MS : pause_ms ? 2 * pause_ms : FIRST_PAUSE_MS;
         if (pause_ms > LONGEST_PAUSE_MS) pause_ms = LONGEST_PAUSE_MS;
-        struct timespec until;
-        clock_gettime(CLOCK_REALTIME, &until);
-        until.tv_sec += pause_ms / 1000;
-        until.tv_nsec += (pause_ms % 1000) * 1000000;
-        if (until.tv_nsec >= 1000000000) { ++until.tv_sec; until.tv_nsec -= 1000000000; }
+        struct timespec until = pure_wait_until(pause_ms * 1000000LL);
         while (!w->stopping && pthread_cond_timedwait(&w->changed, &w->lock, &until) == 0) {}
     }
     pthread_mutex_unlock(&w->lock);
diff --git a/serve/serve.c b/serve/serve.c
index ab386ae..e140705 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -30,6 +30,7 @@
  * threads hand it their ready requests and answer what it hands back. Its
  * bursts take the requests of every thread, one save each. */
 #include "deliver.h"
+#include "wait.h"
 #include "link.h"
 #include "seal.h"
 #include "surface.h"
@@ -1989,9 +1990,7 @@ static void *run_owner(void *context) {
                 pthread_cond_wait(&owner->waiting, &owner->lock);
                 continue;
             }
-            struct timespec wake;
-            clock_gettime(CLOCK_REALTIME, &wake);
-            wake = later(wake, due * 1000000L);
+            struct timespec wake = pure_wait_until(due * 1000000LL);
             pthread_cond_timedwait(&owner->waiting, &owner->lock, &wake);
         }
         /* After a round that saved, its answers bring requests back from the
@@ -2009,7 +2008,7 @@ static void *run_owner(void *context) {
          * there gathering would only leave the owner idle). */
         uint64_t waited = owner->executor.save_waited, budget = waited < 1000000 ? waited : 1000000;
         struct timespec now, end;
-        clock_gettime(CLOCK_REALTIME, &now);
+        clock_gettime(PURE_WAIT_CLOCK, &now);
         end = later(now, (long)budget);
         while (gather && owner->count && !owner->stopping && before(now, end)) {
             size_t seen = owner->count;
@@ -2017,7 +2016,7 @@ static void *run_owner(void *context) {
             struct timespec quiet = later(now, budget < 100000 ? (long)budget : 100000);
             pthread_cond_timedwait(&owner->waiting, &owner->lock, returned && before(quiet, end) ? &quiet : &end);
             if (returned && owner->count == seen) break;
-            clock_gettime(CLOCK_REALTIME, &now);
+            clock_gettime(PURE_WAIT_CLOCK, &now);
         }
         if (!owner->count) {
             pthread_mutex_unlock(&owner->lock);
@@ -2678,7 +2677,7 @@ int main(int argc, char **argv) {
     }
     if (owned) {
         if (!start_executor(&owner.executor, &module, show, edit, store) || pthread_mutex_init(&owner.lock, NULL) ||
-            pthread_cond_init(&owner.waiting, NULL) || pthread_mutex_init(&owner.executor.published_lock, NULL))
+            pure_cond_monotonic(&owner.waiting) || pthread_mutex_init(&owner.executor.published_lock, NULL))
             return 1;
         owner.executor.reads_first = true;
         owner.executor.publishes = true;
diff --git a/store/store-tree.c b/store/store-tree.c
index 83f59e3..ad251d7 100644
--- a/store/store-tree.c
+++ b/store/store-tree.c
@@ -14,6 +14,7 @@
 #define _DARWIN_C_SOURCE
 #endif
 #include "store.h"
+#include "wait.h"
 #include "state.h"
 #include "digest.h"
 #include <errno.h>
@@ -1592,6 +1593,13 @@ void pure_store_group_counts(unsigned long *groups, unsigned long *requests) {
 
 enum { GATHER_NS = 2000000 };
 
+/* Once: joined waits on the monotonic clock (core/wait.h). */
+static pthread_once_t group_made = PTHREAD_ONCE_INIT;
+static void make_group(void) {
+    pthread_cond_destroy(&group.joined);
+    if (pure_cond_monotonic(&group.joined)) pthread_cond_init(&group.joined, NULL); /* then it does not gather */
+}
+
 /* Publish the queued requests of the store at path in one session. */
 static void commit_group(const char *path, Waiter *first) {
     size_t total = 0;
@@ -1695,6 +1703,7 @@ PureResult pure_draft_commit_all(PureDraft *const *drafts, size_t count, int64_t
     }
     if (!count) return PURE_OK;
     Waiter me = {.drafts = drafts, .count = count, .revisions = revisions};
+    pthread_once(&group_made, make_group);
     pthread_mutex_lock(&group.lock);
     if (group.tail) group.tail->next = &me; else group.head = &me;
     group.tail = &me;
@@ -1707,10 +1716,7 @@ PureResult pure_draft_commit_all(PureDraft *const *drafts, size_t count, int64_t
          * a moment (at most 2 ms) to join, or they would miss this group and
          * wait for the next one: a sync costs far more. One thread never waits. */
         if (group.expected > 1 && group.queued < group.expected) {
-            struct timespec deadline;
-            clock_gettime(CLOCK_REALTIME, &deadline);
-            deadline.tv_nsec += GATHER_NS;
-            if (deadline.tv_nsec >= 1000000000) { ++deadline.tv_sec; deadline.tv_nsec -= 1000000000; }
+            struct timespec deadline = pure_wait_until(GATHER_NS);
             while (group.queued < group.expected &&
                    pthread_cond_timedwait(&group.joined, &group.lock, &deadline) == 0) {}
         }
diff --git a/tests/check.py b/tests/check.py
index e15a996..379982d 100644
--- a/tests/check.py
+++ b/tests/check.py
@@ -2653,17 +2653,82 @@ def server_timers(aup, pure, work):
         stats = host.stop()
     assert stats[0]['ticks'] >= 10, stats
     seen['stateless'] = stats[0]['ticks']
-    # The wall clock set back an hour 0.2 s in (build/clock-back.so): the
-    # timers wait on the monotonic clock, so they go on ticking.
+    return seen
+
+
+def cpu_seconds(pid):
+    """The CPU time a process has used (user and system)."""
     import os
-    host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'heartbeat-back'],
-                    env={**os.environ, 'LD_PRELOAD': str(pure / 'build/clock-back.so')})
+    with open(f'/proc/{pid}/stat') as stat:
+        fields = stat.read().rsplit(')', 1)[1].split()
+    return (int(fields[11]) + int(fields[12])) / os.sysconf('SC_CLK_TCK')
+
+
+def server_clocks(aup, pure, work):
+    """The wall clock moved an hour while a node runs (build/clock-back.so
+    moves what the node reads of it, from its start): its timers, the pauses
+    of a wire whose receiver is down, the owner's waits and a store's group
+    commit all count on the monotonic clock. Were a wait's deadline read
+    from the wall clock, set back it would end at once, again and again (a
+    core spun, the wire tried thousands of times); set forward it would last
+    an hour (the timers stopped, the wire tried once, and a lone commit
+    after a group waited that long for the others to join it)."""
+    import os
+    import socket
+    import struct
+    import threading
+    import time
+    module, source = work / 'clocks.pmod', work / 'clocks.aup'
+    source.write_text(HEARTBEAT_SOURCE)
+    compile_module(aup, source, module)
+    with socket.socket() as probe:
+        probe.bind(('127.0.0.1', 0))
+        nobody = probe.getsockname()[1]
+    seen = {}
+    for shift in (-3600, 3600):
+        env = {**os.environ, 'LD_PRELOAD': str(pure / 'build/clock-back.so'), 'CLOCK_BACK_MS': '0',
+               'CLOCK_SHIFT_S': str(shift)}
+        host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / f'clocks{shift}',
+                         '--send', f'pulse=http://127.0.0.1:{nobody}/inputs/pulse'], env=env)
+        try:
+            time.sleep(0.2)
+            began, used = time.time(), cpu_seconds(host.process.pid)
+            time.sleep(1.0)
+            share = (cpu_seconds(host.process.pid) - used) / (time.time() - began)
+        finally:
+            stats = host.stop()[-1]
+        assert stats['ticks'] >= 30 and share < 0.25 and 3 <= stats['delivery_attempts'] <= 12, \
+            (shift, stats['ticks'], share, stats['delivery_attempts'])
+        seen[f'shift_{shift}'] = {'ticks': stats['ticks'], 'cpu_share': round(share, 3),
+                                  'wire_attempts': stats['delivery_attempts']}
+    editor = work / 'clocks-editor.pmod'
+    compile_module(aup, EDITOR, editor)
+    env = {**os.environ, 'LD_PRELOAD': str(pure / 'build/clock-back.so'), 'CLOCK_BACK_MS': '0', 'CLOCK_SHIFT_S': '3600'}
+    host = HttpHost([pure / 'build/pure-serve', '--module', editor, '--store', work / 'clocks-docs', '--threads', 3],
+                    env=env)
+    new = struct.pack('<II', 0, 0)  # an edit that makes the document: at 0, none removed, then its bytes
+    statuses, alone = [], 0.0
     try:
-        time.sleep(0.6)
+        for round_number in range(20):  # three at once, grouped; then one alone
+            together = threading.Barrier(3)
+
+            def put(k):
+                together.wait()
+                statuses.append(host.request('PUT', f'd{round_number}-{k}', new + b'x' * 100000, 0)[0])
+
+            threads = [threading.Thread(target=put, args=(k,)) for k in range(3)]
+            for thread in threads:
+                thread.start()
+            for thread in threads:
+                thread.join()
+            began = time.time()
+            statuses.append(host.request('PUT', f'alone-{round_number}', new + b'y', 0)[0])
+            alone = max(alone, time.time() - began)
     finally:
-        stats = host.stop()
-    assert stats[0]['ticks'] >= 20, stats
-    seen['clock_set_back'] = stats[0]['ticks']
+        stats = host.stop()[-1]
+    assert statuses == [200] * 80 and alone < 1 and stats['grouped_commits'] > stats['commit_groups'], \
+        (statuses, alone, stats)
+    seen['alone_after_groups_s'] = round(alone, 4)
     return seen
 
 
@@ -2846,6 +2911,7 @@ def main():
             server_app(aup, pure, work, range(args.seeds), threads=3)
         report['server_json_requests'] = server_json(aup, pure, work)
         report['server_timer_beats'] = server_timers(aup, pure, work)
+        report['server_clocks'] = server_clocks(aup, pure, work)
         report['server_plain_edit'] = server_plain_edit(aup, pure, work)
         report['server_links'] = server_links(aup, pure, work)
         report['server_links_udp'] = server_links_udp(aup, pure, work)
diff --git a/tests/clock-back.c b/tests/clock-back.c
index 50870ca..2ab83e0 100644
--- a/tests/clock-back.c
+++ b/tests/clock-back.c
@@ -1,8 +1,12 @@
 #define _GNU_SOURCE
-/* LD_PRELOAD: the wall clock (CLOCK_REALTIME) set back an hour from
- * CLOCK_BACK_MS milliseconds (200 unless given) after the process first
- * asks for it, as a person or NTP may set it. A program's timers must go on
- * all the same: they wait on the monotonic clock (tests/check.py). */
+/* LD_PRELOAD: the wall clock (CLOCK_REALTIME) set back an hour (moved
+ * CLOCK_SHIFT_S seconds: -3600 unless given) from CLOCK_BACK_MS milliseconds
+ * (200 unless given) after the process first asks for it, as a person or NTP
+ * may set it. Only what the process reads moves, so a wait until a deadline
+ * read from the wall clock ends at once (set back) or lasts an hour (set
+ * forward), as one would with the clock set the other way after it was read.
+ * A program's timers and waits must go on all the same: they count on the
+ * monotonic clock (tests/check.py). */
 #include <dlfcn.h>
 #include <stdlib.h>
 #include <time.h>
@@ -17,7 +21,7 @@ int clock_gettime(clockid_t id, struct timespec *t) {
     real(CLOCK_MONOTONIC, &now);
     long long ms = (long long)now.tv_sec * 1000 + now.tv_nsec / 1000000;
     if (began < 0) began = ms;
-    const char *after = getenv("CLOCK_BACK_MS");
-    if (ms - began >= (after ? atoll(after) : 200)) t->tv_sec -= 3600;
+    const char *after = getenv("CLOCK_BACK_MS"), *shift = getenv("CLOCK_SHIFT_S");
+    if (ms - began >= (after ? atoll(after) : 200)) t->tv_sec += shift ? atoll(shift) : -3600;
     return result;
 }