Tick a served app's timers and give its events the clock

pure-serve sets the time an event sees to the wall clock when the event
is applied, and the thread that runs an app with states (the connection
thread alone, or the owner) ticks its timers on time: epoll waits no
longer than the next tick, the owner's condition wait is timed to it, and
the ticks due at once share one save, so a restarted server goes on from
the stored ticks. The statistics count ticks.

2 files changed, 64 insertions, 9 deletions

diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md
index c8dcc0c..1039358 100644
--- a/docs/design/008-module-format.md
+++ b/docs/design/008-module-format.md
@@ -264,6 +264,9 @@ text·리스트가 남을 때만 위치를 유지한다.
   모르거나 두 번 오거나(`json_field`), 타입이 다르거나(`json_type`), JSON이 아니면(`json`) 400이고 요청은 실행되지 않는다.
 - **라우트는 작업 슬롯에 둔다.** 요청이 가리키는 것(문서, 보기, 입력)과 메시지 필드는 받을 때 풀어 슬롯별 표와 ingress에
   둔다. 연결이 묶음 전에 끊겨도 받아들인 이벤트는 실행된다. 런타임 코어는 그대로다(키는 문서 키 또는 보기·입력 이름).
+- **시간.** 요청의 이벤트는 적용되는 순간의 벽시계를 `now`로 본다. 앱을 돌리는 스레드(한 스레드면 그 스레드, 아니면
+  소유자)가 상태를 가진 앱의 타이머를 제때 틱하고, 틱들을 저장 한 번으로 저장한다. 그래서 다시 띄운 서버는 저장된
+  상태에서 이어서 틱한다. 대기는 다음 틱까지다(epoll의 대기 시간, 소유자는 조건 변수의 시한).
 - **POST 응답은 요청에서 빌린다.** 코어가 쓰기 결과를 요청 바이트에서 빌리게 하므로 HTTP로 받는 메시지의 핸들러는
   메시지 필드만 응답할 수 있다(`reply score.name`). 시작할 때 확인한다. 상태에서 계산한 값은 보기로 읽는다.
 
diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c
index 2cb69c1..a60bf44 100644
--- a/prototypes/connections/serve.c
+++ b/prototypes/connections/serve.c
@@ -128,6 +128,8 @@ typedef struct {
     Applied *applied; /* a round's requests, grown to fit */
     size_t applied_capacity;
     bool wrote; /* the last round saved an event or an edit */
+    double next[PURE_MODULE_INPUTS]; /* when each timer ticks next (wall clock, ms) */
+    uint64_t ticks;
 } Executor;
 
 /* A request made ready: its effect, what it names, and the thread that answers it. */
@@ -193,6 +195,13 @@ static uint64_t now_ms(void) {
     return (uint64_t)t.tv_sec * 1000 + (uint64_t)t.tv_nsec / 1000000;
 }
 
+/* The time events see: the wall clock, in milliseconds since 1970 (UTC). */
+static double wall_ms(void) {
+    struct timespec t;
+    clock_gettime(CLOCK_REALTIME, &t);
+    return floor((double)t.tv_sec * 1000 + (double)t.tv_nsec / 1e6);
+}
+
 static bool number(const char *text, uint64_t maximum, uint64_t *value) {
     if (!*text) return false;
     uint64_t n = 0;
@@ -973,6 +982,7 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect
 static void apply_effect(Executor *ex, const Job *job, Applied *out) {
     PureApp *app = &ex->app;
     const PureEffect *effect = &job->effect;
+    app->now = wall_ms();
     *out = (Applied){.host = job->host, .ticket = effect->ticket, .input = effect->input, .route = job->route.kind};
     memcpy(out->key, effect->key, strlen(effect->key) + 1);
     if (job->route.kind == ROUTE_VIEW) {
@@ -1063,6 +1073,29 @@ static void execute(Executor *ex, const Job *jobs, size_t count, void (*finish)(
     if (pinned) pure_store_pin(ex->store, false);
 }
 
+/* Tick every timer that is due, with one save for them (an app with states
+ * only: a timer can change nothing else). The milliseconds until the next
+ * is due, or -1 without timers. Missed ticks are not made up. */
+static long run_timers(Executor *ex) {
+    const PureModule *m = ex->module;
+    if (!m->states) return -1;
+    double now = wall_ms(), soonest = -1;
+    bool ticked = false;
+    for (uint32_t i = 0; i < m->inputs; ++i) {
+        if (m->input[i].kind != PURE_KIND_TIMER) continue;
+        if (!ex->next[i]) ex->next[i] = now + m->input[i].period; /* the first period starts now */
+        if (ex->next[i] <= now) {
+            ex->app.now = now;
+            if (pure_app_tick(&ex->app, i) == PURE_OK) ++ex->ticks;
+            ticked = true;
+            ex->next[i] = now + m->input[i].period;
+        }
+        if (soonest < 0 || ex->next[i] - now < soonest) soonest = ex->next[i] - now;
+    }
+    if (ticked) pure_app_save(&ex->app); /* a failed save undoes the ticks */
+    return soonest < 0 ? -1 : (long)ceil(soonest);
+}
+
 /* On the request's connection thread: complete its ticket and answer it if
  * its connection still waits for it. */
 static void answer_applied(Host *host, Applied *a) {
@@ -1150,8 +1183,9 @@ static void serve_bursts(Host *host) {
 /* The owner's thread: take every waiting request of every connection thread
  * as one round, until they have all stopped. */
 static struct timespec later(struct timespec t, long nanoseconds) {
-    t.tv_nsec += nanoseconds;
-    while (t.tv_nsec >= 1000000000) { ++t.tv_sec; t.tv_nsec -= 1000000000; }
+    t.tv_sec += nanoseconds / 1000000000;
+    t.tv_nsec += nanoseconds % 1000000000;
+    if (t.tv_nsec >= 1000000000) { ++t.tv_sec; t.tv_nsec -= 1000000000; }
     return t;
 }
 
@@ -1166,7 +1200,21 @@ static void *run_owner(void *context) {
     bool gather = false;
     for (;;) {
         pthread_mutex_lock(&owner->lock);
-        while (!owner->count && !owner->stopping) pthread_cond_wait(&owner->waiting, &owner->lock);
+        /* Wait for requests, or until the next timer is due. */
+        for (;;) {
+            pthread_mutex_unlock(&owner->lock);
+            long due = run_timers(&owner->executor);
+            pthread_mutex_lock(&owner->lock);
+            if (owner->count || owner->stopping) break;
+            if (due < 0) {
+                pthread_cond_wait(&owner->waiting, &owner->lock);
+                continue;
+            }
+            struct timespec wake;
+            clock_gettime(CLOCK_REALTIME, &wake);
+            wake = later(wake, due * 1000000L);
+            pthread_cond_timedwait(&owner->waiting, &owner->lock, &wake);
+        }
         /* After a round that saved, its answers bring requests back from the
          * connection threads: gather them until none has come for 100 µs (a
          * millisecond at most), so they share one save as they would on a
@@ -1362,7 +1410,9 @@ static void *run_shard(void *context) {
     struct epoll_event events[EVENTS];
     uint64_t next_sweep = now_ms() + 1000;
     while (!atomic_load(&stopping)) {
-        int ready = epoll_wait(host->epoll, events, EVENTS, 1000);
+        /* A thread that runs the app ticks its timers on time. */
+        long due = host->owner ? -1 : run_timers(&host->local);
+        int ready = epoll_wait(host->epoll, events, EVENTS, due >= 0 && due < 1000 ? (int)due : 1000);
         for (int e = 0; e < ready; ++e) {
             Connection *c = events[e].data.ptr;
             if (!c) { accept_all(host); continue; }
@@ -1483,7 +1533,8 @@ int main(int argc, char **argv) {
         pthread_join(owner_thread, NULL);
     }
     size_t active = 0, retained = 0;
-    uint64_t accepted = 0, requests = 0, bursts = owner.executor.bursts, events = owner.executor.burst_events, expired = 0;
+    uint64_t accepted = 0, requests = 0, bursts = owner.executor.bursts, events = owner.executor.burst_events, expired = 0,
+             ticks = owner.executor.ticks;
     for (unsigned i = 0; i < threads; ++i) {
         active += hosts[i].active;
         retained += hosts[i].runtime.retained;
@@ -1491,6 +1542,7 @@ int main(int argc, char **argv) {
         requests += hosts[i].requests;
         bursts += hosts[i].local.bursts;
         events += hosts[i].local.burst_events;
+        ticks += hosts[i].local.ticks;
         expired += hosts[i].expired;
         stop_shard(&hosts[i]);
     }
@@ -1501,10 +1553,10 @@ int main(int argc, char **argv) {
     unsigned long groups, grouped;
     pure_store_group_counts(&groups, &grouped);
     fprintf(stderr, "{\"active\":%zu,\"retained_payload_bytes\":%zu,\"accepted\":%" PRIu64 ",\"requests\":%" PRIu64
-            ",\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 ",\"expired\":%" PRIu64 ",\"threads\":%u"
-            ",\"owner\":%s,\"commit_groups\":%lu,\"grouped_commits\":%lu}\n",
-            active, retained, accepted, requests, bursts, events, expired, threads, owned ? "true" : "false", groups,
-            grouped);
+            ",\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 ",\"ticks\":%" PRIu64 ",\"expired\":%" PRIu64
+            ",\"threads\":%u,\"owner\":%s,\"commit_groups\":%lu,\"grouped_commits\":%lu}\n",
+            active, retained, accepted, requests, bursts, events, ticks, expired, threads, owned ? "true" : "false",
+            groups, grouped);
     free(hosts);
     free(thread);
     return 0;