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.

2 files changed, 8 insertions, 12 deletions

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;