diff options
| -rw-r--r-- | docs/design/008-module-format.md | 3 | ||||
| -rw-r--r-- | prototypes/connections/serve.c | 70 |
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; |