From 6fda961e835527de2aac7d91e8a84fb7ae710327 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Sun, 4 Oct 2026 18:30:43 +0100 Subject: Compute views with a body on view workers, off the connection threads, asked and pushed (020) - pure-serve --view-workers N (a CPU each; 0: on the connection threads): a GET of a view with a body (or a table with such a column) goes to the workers, which show it from the app's newest snapshot on readers of their own and give the answer back as an owner does (eventfd). The connection thread reads and writes its other connections meanwhile. - A snapshot of another module than the request was resolved with, or none (the app stopped), answers 503 stopped. Past an app's queue=N of them, 503 busy. - A push group of such a view hands one evaluation at a time; a newer snapshot meanwhile is evaluated after it; a group whose subscribers all left stays until its evaluation is back. - check.py server_view_workers: 8 askers of a long view while its app changes module (another view in its place) and then goes: every answer the old module's, the new one's, 503 or 404; streams closed at once leave nothing; --view-workers 0 as before. --- docs/design/020-computing-faster.md | 11 ++ serve/serve.c | 353 +++++++++++++++++++++++++++--------- tests/check.py | 110 +++++++++++ 3 files changed, 393 insertions(+), 81 deletions(-) diff --git a/docs/design/020-computing-faster.md b/docs/design/020-computing-faster.md index 14f3a57..bb0ebbe 100644 --- a/docs/design/020-computing-faster.md +++ b/docs/design/020-computing-faster.md @@ -69,6 +69,17 @@ - 해 보고 뺀 것: 잎의 울타리(키로 찾은 잎의 다음 잎의 첫 키, 손가락이 키 범위를 더 넓게 알게) — 185.5 대 185.5 ms. 손가락 넷 (위). +### 2단계에서 정한 것 + +- 보기의 스레드는 넘겨받은 일을 할 때 그 앱의 **가장 새 스냅숏을 스스로 읽는다**(연결 스레드가 그 라운드에 읽은 것을 쥐고 넘기지 + 않는다). 답은 그 라운드의 것보다 새것이거나 같으니, 쓰기의 답을 받은 뒤의 `GET`이 그 쓰기를 본다는 것은 그대로다. 스냅숏의 + 모듈이 요청이 풀린 모듈과 다르면(앱이 모듈을 바꿨다) 계산하지 않고 503 `stopped`다. 앱이 멈춰 스냅숏이 없어도 그렇다. +- 넘긴 일의 한도는 앱의 `queue=N`(`--apps`의 줄)이다. 그 앱의 보기 가운데 보기의 스레드에 있는 것이 N이면 503 `busy`. 한 앱의 + 호스트처럼 `queue`가 없으면 한도는 연결 수다(연결 하나에 요청 하나가 답을 기다린다). 미는 무리는 한 번에 계산 하나만 넘긴다. +- 표는 그 열 가운데 하나라도 본문이 있으면 넘긴다. 식 하나의 보기는 연결 스레드가 계산한다(지금처럼). +- 스레드의 이름은 `pure-view`, 멈출 때의 통계에 `view_workers`와 `views_by_workers`가 있다. `--view-workers 0`이면 전처럼 연결 + 스레드가 계산한다(견주는 쪽). + ## 단계 1. 키로 찾기(소속 표, 손가락). diff --git a/serve/serve.c b/serve/serve.c index f4eea46..bb339cf 100644 --- a/serve/serve.c +++ b/serve/serve.c @@ -208,6 +208,8 @@ typedef struct { const char *refused; /* refused by the host (503 and this name), not applied */ uint64_t heard; /* then the source's last number heard after them */ bool gap; /* and whether they stopped at a gap */ + struct Group *group; /* a push group's evaluation, by a view worker (docs/design/020) */ + uint64_t changes; /* and the snapshot it was of */ } Applied; /* The last save's snapshot an owner published, the connection threads' @@ -428,7 +430,8 @@ struct App { uint32_t show, edit; _Atomic size_t memory; /* its budgets (0: none): what its states hold (read by its views' readers too), */ _Atomic uint64_t work; /* what an event or a view's evaluation may run */ - size_t queue; /* the requests that may wait for its turn */ + size_t queue; /* the requests that may wait for its turn, and its views' with the view workers */ + _Atomic size_t viewing; bool running; /* started (the main thread's word); else why says why */ char why[256]; unsigned home; /* the connection thread its connections move to (a host of apps) */ @@ -463,7 +466,7 @@ struct Host { unsigned *touched; /* the apps read in this round */ Applied *views; /* a round's views shown here, answered together */ size_t views_capacity; - uint64_t shown; /* views shown from snapshots */ + uint64_t shown, pooled; /* views shown from snapshots; of them, by the view workers */ Route *routes; /* by work slot */ struct Given { char *text; size_t capacity; } *given; /* a view's arguments as asked, by work slot */ PureRuntime runtime; @@ -2239,6 +2242,21 @@ static void answer_burst(void *context, Applied *applied, size_t count) { for (size_t i = 0; i < count; ++i) answer_applied(applied[i].host, &applied[i]); } +/* A connection thread's requests among those applied, given back to it (an + * owner's or a view worker's), and it woken. */ +static void give_back(Host *host, const Applied *applied, size_t count) { + size_t mine = 0; + for (size_t i = 0; i < count; ++i) mine += applied[i].host == host; + if (!mine) return; + pthread_mutex_lock(&host->lock); + if (!grow(&host->done, &host->done_capacity, host->done_count + mine, sizeof(*host->done))) abort(); + for (size_t i = 0; i < count; ++i) + if (applied[i].host == host) host->done[host->done_count++] = applied[i]; + pthread_mutex_unlock(&host->lock); + uint64_t one = 1; + if (write(host->wake, &one, sizeof(one)) != sizeof(one)) abort(); +} + /* On the owner: give each connection thread its applied requests, and wake it. */ static void hand_back(void *context, Applied *applied, size_t count) { Owner *owner = context; @@ -2262,22 +2280,12 @@ static void hand_back(void *context, Applied *applied, size_t count) { pthread_cond_broadcast(&surface->applied); pthread_mutex_unlock(&surface->lock); } - for (unsigned h = 0; h < owner->host_count; ++h) { - Host *host = &owner->hosts[h]; - size_t mine = 0; - for (size_t i = 0; i < count; ++i) mine += applied[i].host == host; - if (!mine) continue; - pthread_mutex_lock(&host->lock); - if (!grow(&host->done, &host->done_capacity, host->done_count + mine, sizeof(*host->done))) abort(); - for (size_t i = 0; i < count; ++i) - if (applied[i].host == host) host->done[host->done_count++] = applied[i]; - pthread_mutex_unlock(&host->lock); - uint64_t one = 1; - if (write(host->wake, &one, sizeof(one)) != sizeof(one)) abort(); - } + for (unsigned h = 0; h < owner->host_count; ++h) give_back(&owner->hosts[h], applied, count); } -/* On a connection thread: answer what the owner handed back. */ +static void group_evaluated(Host *host, Applied *a); + +/* On a connection thread: answer what the owner and the view workers handed back. */ static void answer_handed(Host *host) { uint64_t signals; if (read(host->wake, &signals, sizeof(signals)) < 0 && errno != EAGAIN) abort(); @@ -2290,26 +2298,133 @@ static void answer_handed(Host *host) { pthread_mutex_unlock(&host->lock); host->answering = done; host->answering_capacity = capacity; - for (size_t i = 0; i < count; ++i) answer_applied(host, &done[i]); + for (size_t i = 0; i < count; ++i) + if (done[i].group) group_evaluated(host, &done[i]); + else answer_applied(host, &done[i]); host->pending -= count; } /* The owner's newest snapshot, held, shown by this thread's reader; NULL * when there is none. Taken for each round, after the round's answers from * the owner: those were handed back after their saves' snapshots. */ -static PureSnapshot *read_published(Host *host, App *app) { +static PureSnapshot *read_newest(PureApp *reader, App *app) { Published *published = &app->owner.published; pthread_mutex_lock(&published->lock); PureSnapshot *snapshot = published->snapshot; if (snapshot) pure_snapshot_hold(snapshot); pthread_mutex_unlock(&published->lock); - PureApp *reader = &host->readers[app - apps]; if (snapshot) pure_app_read(reader, snapshot); reader->work_budget = atomic_load_explicit(&app->work, memory_order_relaxed); reader->memory = atomic_load_explicit(&app->memory, memory_order_relaxed); /* what a view's body may add to its own lists */ return snapshot; } +static PureSnapshot *read_published(Host *host, App *app) { return read_newest(&host->readers[app - apps], app); } + +/* ---- Views that compute, off the connection threads (docs/design/020's + * stage 2). A view with a body may run long: its GET, and a push group's + * evaluation of it, go to the view workers, which show it from its app's + * newest snapshot on readers of their own and give the answer back to the + * thread it came from as an owner does. That thread reads and writes its + * other connections meanwhile. A view of one expression stays on it. */ +typedef struct ViewJob { + Host *host; + Route route; /* in the module it was resolved with: shown only from a snapshot of that one */ + PureTicket ticket; /* a GET's, */ + PureInput input; + size_t available; + struct Group *group; /* or a push group's evaluation */ + struct ViewJob *next; +} ViewJob; +static struct { + pthread_mutex_t lock; + pthread_cond_t ready; + ViewJob *first, *last; + bool stopping; +} views_waiting = {PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER, NULL, NULL, false}; +static unsigned view_workers = UINT_MAX; /* --view-workers (UINT_MAX: a CPU each; 0: none, on the connection threads) */ +static pthread_t *view_threads; +static unsigned view_thread_count; +static unsigned online_cpus(void); + +/* Whether a view may run long: it has a body, or it is a table one of whose columns has. */ +static bool computes(const PureModule *m, uint32_t view) { + if (m->view_body[view]) return true; + if (m->view_type[view] != PURE_VIEW_TABLE) return false; + for (uint32_t column = 0; column < m->views; ++column) + if (m->view_table[column] == view + 1 && m->view_body[column]) return true; + return false; +} + +/* A view for the workers to show: it comes back to host. */ +static void hand_view(Host *host, const Route *route, PureTicket ticket, PureInput input, size_t available, + struct Group *group) { + ViewJob *job = malloc(sizeof(*job)); + if (!job) abort(); + *job = (ViewJob){host, *route, ticket, input, available, group, NULL}; + ++host->pending; + pthread_mutex_lock(&views_waiting.lock); + if (views_waiting.last) views_waiting.last->next = job; + else views_waiting.first = job; + views_waiting.last = job; + pthread_cond_signal(&views_waiting.ready); + pthread_mutex_unlock(&views_waiting.lock); +} + +static void *run_view_worker(void *unused) { + (void)unused; + PureApp *readers = calloc(app_slots, sizeof(*readers)); /* by app place, as a connection thread's */ + if (!readers) abort(); + for (;;) { + pthread_mutex_lock(&views_waiting.lock); + while (!views_waiting.first && !views_waiting.stopping) pthread_cond_wait(&views_waiting.ready, &views_waiting.lock); + ViewJob *job = views_waiting.first; + if (job && !(views_waiting.first = job->next)) views_waiting.last = NULL; + pthread_mutex_unlock(&views_waiting.lock); + if (!job) break; + App *app = job->route.app; + PureApp *reader = &readers[app - apps]; + PureSnapshot *snapshot = read_newest(reader, app); + Applied a = {.host = job->host, .ticket = job->ticket, .input = job->input, .route = ROUTE_VIEW, .group = job->group}; + if (!snapshot || reader->module != job->route.module) { /* its app stopped, or runs another module now */ + a.status = PURE_CLOSED; + a.refused = "stopped"; + } else { + a.status = show_view(reader, &job->route, job->available, &a); + a.changes = reader->changes; + if (job->group && a.status != PURE_OK && a.fault == PURE_FAULT_NONE) a.fault = pure_app_view_fault(reader); + } + if (snapshot) pure_snapshot_release(snapshot); /* an answer holds what it lends */ + if (!job->group) atomic_fetch_sub_explicit(&app->viewing, 1, memory_order_relaxed); + give_back(job->host, &a, 1); + free(job); + } + for (unsigned i = 0, count = places(); i < count; ++i) pure_app_stop(&readers[i]); + free(readers); + return NULL; +} + +/* Made before the connection threads, stopped after them (they wait for what they handed). */ +static bool start_view_workers(void) { + unsigned count = view_workers == UINT_MAX ? online_cpus() : view_workers; + if (!count) return true; + if (!(view_threads = calloc(count, sizeof(*view_threads)))) return false; + for (; view_thread_count < count; ++view_thread_count) { + if (pthread_create(&view_threads[view_thread_count], NULL, run_view_worker, NULL)) return false; + pthread_setname_np(view_threads[view_thread_count], "pure-view"); /* their CPU, seen apart */ + } + return true; +} + +static void stop_view_workers(void) { + pthread_mutex_lock(&views_waiting.lock); + views_waiting.stopping = true; + pthread_cond_broadcast(&views_waiting.ready); + pthread_mutex_unlock(&views_waiting.lock); + for (unsigned i = 0; i < view_thread_count; ++i) pthread_join(view_threads[i], NULL); + free(view_threads); +} + /* ---- Pushing views (docs/design/019). A view's GET that asks for an event * stream makes its connection a subscriber: its socket, and the group of * this thread's subscribers to that view with those arguments. A pass, at @@ -2343,6 +2458,10 @@ typedef struct Group { size_t value_size; char *event; size_t event_size; + /* A view worker evaluates it (a view with a body): for the module asked + * of; again: a newer snapshot came meanwhile. It stays until it is back. */ + bool evaluating, again; + const PureModule *asked; Subscriber *first; struct Group *next; } Group; @@ -2421,6 +2540,13 @@ static bool subscriber_flush(Host *host, Subscriber *s) { return true; } +static void free_group(Group *g) { + free(g->text); + free(g->value); + free(g->event); + free(g); +} + static void unsubscribe(Host *host, Subscriber *s) { Group *g = s->group; for (Subscriber **at = &g->first; *at; at = &(*at)->next) @@ -2435,10 +2561,7 @@ static void unsubscribe(Host *host, Subscriber *s) { if (g->first) return; for (Group **at = &host->groups; *at; at = &(*at)->next) if (*at == g) { *at = g->next; break; } - free(g->text); - free(g->value); - free(g->event); - free(g); + if (!g->evaluating) free_group(g); /* else when its evaluation is back */ } static void close_group(Host *host, Group *g) { @@ -2470,10 +2593,75 @@ static void subscriber_event(Host *host, Subscriber *s, uint32_t events) { if (gone || !subscriber_flush(host, s)) unsubscribe(host, s); } -/* A group's value from the snapshot its app's reader shows: as the view's - * GET answers it, or why it has none (event none). An event made when it - * differs from the last; then each subscriber gets it, and a fresh one the - * group's value at least. */ +/* A group's value, as the view's GET answers it (a), or why it has none + * (event none), from the snapshot of changes. An event made when it differs + * from the last, for each subscriber. */ +static void group_value(Group *g, Applied *a, uint64_t changes) { + Output value = {0}; + bool failed = a->status != PURE_OK; + if (failed) { /* why it has none */ + const char *why = a->status == PURE_NOT_FOUND && a->fault == PURE_FAULT_NONE ? "no_value" : pure_fault_name(a->fault); + put_bytes(&value, why, strlen(why)); + } else if (a->extents) { + for (size_t i = 0; i < a->extent_count; ++i) put_bytes(&value, a->extents[i].iov_base, a->extents[i].iov_len); + } else put_bytes(&value, a->resource.body.data, a->resource.body.size); + if (value.failed) abort(); + bool same = g->computed && failed == g->failed && value.size == g->value_size && + (!value.size || !memcmp(value.bytes, g->value, value.size)); + g->changes = changes; + g->computed = true; + if (!same) { + free(g->value); + g->value = (char *)value.bytes, g->value_size = value.size, g->failed = failed; + value.bytes = NULL; + Output event = {0}; /* id, an event's name if it has none, and a data line per line */ + char id[32]; + int length = snprintf(id, sizeof(id), "id: %" PRIu64 "\n", g->changes); + put_bytes(&event, id, (size_t)length); + if (failed) put_bytes(&event, "event: none\n", 12); + for (size_t at = 0;;) { /* a data line per line of the value */ + const char *end = at < g->value_size ? memchr(g->value + at, '\n', g->value_size - at) : NULL; + size_t line = end ? (size_t)(end - (g->value + at)) : g->value_size - at; + put_bytes(&event, "data: ", 6); + if (line) put_bytes(&event, g->value + at, line); + put_bytes(&event, "\n", 1); + if (!end) break; + at += line + 1; + } + put_bytes(&event, "\n", 1); + if (event.failed) abort(); + free(g->event); + g->event = (char *)event.bytes, g->event_size = event.size; + for (Subscriber *s = g->first; s; s = s->next) s->fresh = true; /* each gets it */ + } + free(value.bytes); +} + +static void let_go_applied(Applied *a) { + if (a->resource.release) a->resource.release(a->resource.allocation, a->resource.allocation_size); + free(a->extents); +} + +/* Each subscriber that has not had the group's value gets it (none before + * its first evaluation is back). False when the group went with the last. */ +static bool group_send(Host *host, Group *g) { + for (Subscriber *s = g->first, *after; g->event && s; s = after) { + after = s->next; + if (!s->fresh) continue; + s->fresh = false; + if (!subscriber_send(host, s, g->event, g->event_size)) { + bool last = !s->next && g->first == s; + unsubscribe(host, s); + if (last) return false; + } + } + return true; +} + +/* A group's value from the snapshot its app's reader shows, when it is not + * the one it has: here, or by a view worker for a view with a body (one at + * a time; a newer snapshot meanwhile is evaluated after). Then each + * subscriber gets it, and a fresh one the group's value at least. */ static void push_group(Host *host, Group *g) { App *app = g->app; if (atomic_load(&app->gone)) { close_group(host, g); return; } @@ -2508,59 +2696,43 @@ static void push_group(Host *host, Group *g) { g->computed = false; } if (!g->computed || reader->changes != g->changes) { - Applied a = {0}; - PureResult status = show_view(reader, &g->route, PURE_OBJECT_LIMIT, &a); - Output value = {0}; - bool failed = status != PURE_OK; - if (failed) { /* why it has none */ - PureFault fault = a.fault != PURE_FAULT_NONE ? a.fault : pure_app_view_fault(reader); - const char *why = status == PURE_NOT_FOUND && fault == PURE_FAULT_NONE ? "no_value" : pure_fault_name(fault); - put_bytes(&value, why, strlen(why)); - } else if (a.extents) { - for (size_t i = 0; i < a.extent_count; ++i) put_bytes(&value, a.extents[i].iov_base, a.extents[i].iov_len); - } else put_bytes(&value, a.resource.body.data, a.resource.body.size); - if (a.resource.release) a.resource.release(a.resource.allocation, a.resource.allocation_size); - free(a.extents); - if (value.failed) abort(); - bool same = g->computed && failed == g->failed && value.size == g->value_size && - (!value.size || !memcmp(value.bytes, g->value, value.size)); - g->changes = reader->changes; - g->computed = true; - if (!same) { - free(g->value); - g->value = (char *)value.bytes, g->value_size = value.size, g->failed = failed; - value.bytes = NULL; - Output event = {0}; /* id, an event's name if it has none, and a data line per line */ - char id[32]; - int length = snprintf(id, sizeof(id), "id: %" PRIu64 "\n", g->changes); - put_bytes(&event, id, (size_t)length); - if (failed) put_bytes(&event, "event: none\n", 12); - for (size_t at = 0;;) { /* a data line per line of the value */ - const char *end = at < g->value_size ? memchr(g->value + at, '\n', g->value_size - at) : NULL; - size_t line = end ? (size_t)(end - (g->value + at)) : g->value_size - at; - put_bytes(&event, "data: ", 6); - if (line) put_bytes(&event, g->value + at, line); - put_bytes(&event, "\n", 1); - if (!end) break; - at += line + 1; + if (view_thread_count && computes(g->route.module, g->route.target)) { + if (g->evaluating) g->again = true; + else { + g->evaluating = true; + g->asked = g->route.module; + hand_view(host, &g->route, (PureTicket){0}, PURE_INPUT_READ, PURE_OBJECT_LIMIT, g); } - put_bytes(&event, "\n", 1); - if (event.failed) abort(); - free(g->event); - g->event = (char *)event.bytes, g->event_size = event.size; - for (Subscriber *s = g->first; s; s = s->next) s->fresh = true; /* each gets it */ + } else { + Applied a = {0}; + a.status = show_view(reader, &g->route, PURE_OBJECT_LIMIT, &a); + if (a.status != PURE_OK && a.fault == PURE_FAULT_NONE) a.fault = pure_app_view_fault(reader); + group_value(g, &a, reader->changes); + let_go_applied(&a); } - free(value.bytes); } pure_snapshot_release(snapshot); - for (Subscriber *s = g->first, *after; s; s = after) { - after = s->next; - if (!s->fresh) continue; - s->fresh = false; - if (!subscriber_send(host, s, g->event, g->event_size)) { - if (!s->next && g->first == s) { unsubscribe(host, s); return; } - unsubscribe(host, s); - } + group_send(host, g); +} + +/* A view worker's evaluation of a group, back on its thread: its value, + * unless the app stopped or changed module meanwhile (the next look says + * which); then the newer snapshot's, if one came. */ +static void group_evaluated(Host *host, Applied *a) { + Group *g = a->group; + g->evaluating = false; + if (!g->first) { /* its subscribers all left meanwhile */ + let_go_applied(a); + free_group(g); + return; + } + if (a->refused || g->asked != g->route.module) g->again = true; + else group_value(g, a, a->changes); + let_go_applied(a); + if (!group_send(host, g)) return; + if (g->again && !atomic_load(&stopping)) { + g->again = false; + push_group(host, g); } } @@ -2815,6 +2987,19 @@ static void serve_bursts(Host *host) { applied->refused = "stopped"; continue; } + if (job->route.kind == ROUTE_VIEW && view_thread_count && computes(job->route.module, job->route.target)) { + App *app = job->route.app; /* a view with a body: the workers show it, past its app's queue 503 busy */ + if (app->queue && atomic_load_explicit(&app->viewing, memory_order_relaxed) >= app->queue) { + applied->status = PURE_BUSY; + applied->refused = "busy"; + continue; + } + atomic_fetch_add_explicit(&app->viewing, 1, memory_order_relaxed); + hand_view(host, &job->route, job->effect.ticket, job->effect.input, job->effect.available, NULL); + --answers; /* answered when it is back */ + ++shown, ++host->pooled; + continue; + } applied->status = job->route.kind == ROUTE_VIEW ? show_view(&host->readers[a], &job->route, job->effect.available, applied) : registering ? register_query(job) : run_query(job, host->snapshots[a], applied); @@ -4563,6 +4748,7 @@ int main(int argc, char **argv) { else if (!strcmp(argv[i], "--timeout-ms") && number(argv[i + 1], 3600000, &value) && value) timeout_ms = value; else if (!strcmp(argv[i], "--subscribers") && number(argv[i + 1], 1u << 22, &value)) subscribers_most = (size_t)value; else if (!strcmp(argv[i], "--push-every-ms") && number(argv[i + 1], 60000, &value)) push_every_ms = value; + else if (!strcmp(argv[i], "--view-workers") && number(argv[i + 1], 256, &value)) view_workers = (unsigned)value; else goto usage; } if (surface_path) surface_argv[surface_argc++] = (char *)surface_path; @@ -4750,10 +4936,12 @@ int main(int argc, char **argv) { /* A surface reads the owner's snapshots and gives it events. */ if (surface_path && (named_apps || !apps[0].owned)) { fputs("--surface needs an app with states\n", stderr); return 2; } if (surface_path && !start_surface(&surface, owner, surface_argc, surface_argv)) return 1; + if (apps[0].owned && !start_view_workers()) { fputs("Cannot start a thread\n", stderr); return 1; } for (unsigned i = 1; i < threads; ++i) if (pthread_create(&thread[i], NULL, run_shard, &hosts[i])) { fputs("Cannot start a thread\n", stderr); return 1; } run_shard(&hosts[0]); for (unsigned i = 1; i < threads; ++i) pthread_join(thread[i], NULL); + stop_view_workers(); /* the connection threads had back all they handed */ if (surface_path) { /* stopped before the owner, which may still apply its last event */ surface_changed(&surface); pthread_join(surface.thread, NULL); @@ -4780,10 +4968,10 @@ int main(int argc, char **argv) { size_t active = 0, retained = 0, state_bytes = 0; uint64_t unauthorized = 0, forbidden = 0, subscribed = 0, unsubscribed = 0, pushes = 0, pushed = 0, replaced = 0; size_t subscribers = 0; - uint64_t moved = 0, handed = 0, accepted = 0, requests = 0, bursts = 0, events = 0, expired = 0, ticks = 0, shown = 0, checkpoints = 0, + uint64_t moved = 0, handed = 0, accepted = 0, requests = 0, bursts = 0, events = 0, expired = 0, ticks = 0, shown = 0, pooled = 0, checkpoints = 0, memory_refused = 0, windows_full = 0, publications = 0, answers = 0, delivered = 0, attempts = 0; for (unsigned i = 0; i < threads; ++i) { - shown += hosts[i].shown; + shown += hosts[i].shown, pooled += hosts[i].pooled; moved += hosts[i].moved_away; active += hosts[i].active; retained += hosts[i].runtime.retained; @@ -4823,6 +5011,7 @@ int main(int argc, char **argv) { fprintf(stderr, "{\"active\":%zu,\"retained_payload_bytes\":%zu,\"accepted\":%" PRIu64 ",\"requests\":%" PRIu64 ",\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 ",\"ticks\":%" PRIu64 ",\"expired\":%" PRIu64 ",\"threads\":%u,\"owner\":%s,\"snapshots\":%" PRIu64 ",\"views_from_snapshots\":%" PRIu64 + ",\"view_workers\":%u,\"views_by_workers\":%" PRIu64 ",\"delivered\":%" PRIu64 ",\"delivery_attempts\":%" PRIu64 ",\"answers\":%" PRIu64 ",\"commit_groups\":%lu,\"grouped_commits\":%lu,\"checkpoints\":%" PRIu64 ",\"link_packets\":%" PRIu64 ",\"link_records\":%" PRIu64 ",\"link_resent\":%" PRIu64 @@ -4835,7 +5024,7 @@ int main(int argc, char **argv) { ",\"unauthorized\":%" PRIu64 ",\"forbidden\":%" PRIu64 ",\"subscribers\":%zu,\"subscribed\":%" PRIu64 ",\"pushes\":%" PRIu64 ",\"pushed\":%" PRIu64 ",\"push_replaced\":%" PRIu64, active, retained, accepted, requests, bursts, events, ticks, expired, threads, apps[0].owned ? "true" : "false", - publications, shown, delivered, attempts, answers, groups, grouped, + publications, shown, view_thread_count, pooled, delivered, attempts, answers, groups, grouped, checkpoints, sent.packets, sent.records, sent.resent, sent.lost, sent.acknowledged, sent.stored, sent.refused, heard.packets, heard.received, heard.deliveries, heard.woken, heard.remembered, heard.swept, heard.active, heard.keyed, heard.secrets, heard.epochs, heard.refused, state_bytes, memory_refused, @@ -4860,12 +5049,12 @@ int main(int argc, char **argv) { return 0; usage: fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] " - "[--connections 1024] [--timeout-ms 10000] [--subscribers 65536] [--push-every-ms 10] [--send OUTPUT=URL ...] [--name NAME] " + "[--connections 1024] [--timeout-ms 10000] [--subscribers 65536] [--push-every-ms 10] [--view-workers N] [--send OUTPUT=URL ...] [--name NAME] " "[--send-ca FILE] [--reply OUTPUT=INPUT ...] [--seal OUTPUT=KEY ...] [--key FILE] [--sealed-only] " "[--sender NAME=KEY ...] [--link-port PORT] [--senders FILE] [--link-idle-ms 30000] [--memory BYTES] [--work N] " "[--window 4194304] [--tokens FILE] [--surface LIBRARY [-- ARGUMENTS]]\n" "pure-serve --apps FILE [--port 8767] [--threads 1|N|auto] [--workers N] [--app-slots 256] [--budget BYTES] [--connections 1024] " - "[--timeout-ms 10000] [--subscribers 65536] [--push-every-ms 10] [--key FILE] [--sealed-only] [--sender NAME=KEY ...] [--link-port PORT] [--senders FILE] " + "[--timeout-ms 10000] [--subscribers 65536] [--push-every-ms 10] [--view-workers N] [--key FILE] [--sealed-only] [--sender NAME=KEY ...] [--link-port PORT] [--senders FILE] " "[--send-ca FILE] [--window BYTES]\n" " --apps FILE: a host of apps (docs/design/017-apps-on-one-host.md), a line each:\n" " NAME MODULE.pmod STORE [memory=BYTES] [work=N] [queue=N] [send=OUTPUT=URL ...] [seal=OUTPUT=KEY ...]\n" @@ -4879,6 +5068,8 @@ usage: " a delivery sealed by a key --sender names needs none (docs/design/017-apps-on-one-host.md, 3b)\n" "pure-serve --make-token read|write|input=NAME[,NAME]: a new token and its line for a tokens file\n" " --workers N: the threads the apps' owners take turns on (1 for one app; for apps, one a CPU at most)\n" + " --view-workers N: the threads that show views with a body, asked and pushed, off the connection threads\n" + " (a CPU each; 0: on the connection threads, docs/design/020); past an app's queue=N, 503 busy\n" " --send OUTPUT=pure://host:port/INPUT: to a Pure node's --link-port, the way between nodes (UDP, no\n" " connection, docs/design/012-link-protocol.md; its key by --seal OUTPUT=KEY)\n" " --send OUTPUT=http[s]://host:port/path: to anything else, over HTTP\n" diff --git a/tests/check.py b/tests/check.py index f599fb6..7d5648b 100644 --- a/tests/check.py +++ b/tests/check.py @@ -4493,6 +4493,115 @@ def server_view_bodies(aup, pure, work): return {'answers': asked[0], 'connections': 64} +SLOW_SOURCE = """state n = 0 +input bump = text +on bump { + n = n + 1 +} +view slow { + let acc = 0 + for i in 0 .. 3000000 { + acc = acc + 1 + } + return acc + n +} +""" + + +def server_view_workers(aup, pure, work): + """Views with a body on the view workers (docs/design/020's stage 2): + while 8 connections ask a view that loops 3,000,000 times, its app's + module changes (SIGHUP) to one with another view in its place and the + view tenfold, then the app is removed: every answer is the old module's, + the new one's, 503 stopped or 404 (gone), none mixed. Streams of it + opened and closed at once leave nothing behind. With --view-workers 0 + the connection threads show it, as before.""" + import select + import signal + import threading + module, tenfold = work / 'slow.pmod', work / 'slow-tenfold.pmod' + (work / 'slow.aup').write_text(SLOW_SOURCE) + compile_module(aup, work / 'slow.aup', module) + (work / 'slow-tenfold.aup').write_text(SLOW_SOURCE.replace('view slow {', 'view other = 7\nview slow {') + .replace('return acc + n', 'return acc + n * 10')) + compile_module(aup, work / 'slow-tenfold.aup', tenfold) + conf = work / 'slow.conf' + + def write(app_module, name='s'): + conf.write_text(f"{name} {app_module} {work / ('slow-' + name)}\n") + + def reading(): + host.process.send_signal(signal.SIGHUP) + ready, _, _ = select.select([host.process.stdout], [], [], 30) + assert ready, 'no reading' + return json.loads(host.process.stdout.readline()) + + write(module) + host = HttpHost([pure / 'build/pure-serve', '--apps', conf, '--threads', 1, '--view-workers', 2]) + answers, wrong, stop = {}, [], threading.Event() + try: + assert http(host.port, 'POST', '/s/inputs/bump', b'x')[0] == 200 + assert http(host.port, 'GET', '/s/views/slow')[3] == b'3000001' + for _ in range(10): # a stream that goes before its value is back + stream = EventStream(host.port, '/s/views/slow') + assert stream.head()[0] == 200 + stream.close() + stream = EventStream(host.port, '/s/views/slow') + assert stream.next()[2] == '3000001' + + def ask(): + import http.client + c = http.client.HTTPConnection('127.0.0.1', host.port, timeout=30) + while not stop.is_set(): + try: + c.request('GET', '/s/views/slow') + response = c.getresponse() + got = (response.status, response.getheader('X-Pure-Error') if response.status != 200 else 'none', + response.read()) + except (ConnectionError, http.client.HTTPException): + c.close() + c = http.client.HTTPConnection('127.0.0.1', host.port, timeout=30) + continue + answers[got] = answers.get(got, 0) + 1 + if got not in ((200, 'none', b'3000001'), (200, 'none', b'3000010'), (503, 'stopped', b'')) and got[0] != 404: + wrong.append(got) + c.close() + + threads = [threading.Thread(target=ask) for _ in range(8)] + for t in threads: + t.start() + time.sleep(0.5) + write(tenfold) + assert reading()['apps'][0]['did'] == 'changed' + assert stream.next(timeout=10)[2] == '3000010' # the stream, on the new module's view + time.sleep(0.5) + write(tenfold, 'q') # s removed + reading() + assert stream.ends(timeout=10), 'a stream of an app gone still runs' + time.sleep(0.3) + stop.set() + for t in threads: + t.join() + stream.close() + finally: + stop.set() + stats = host.stop() + old, new = answers.get((200, 'none', b'3000001'), 0), answers.get((200, 'none', b'3000010'), 0) + assert not wrong and old and new, (wrong[:5], answers) + assert stats[0]['view_workers'] == 2 and stats[0]['views_by_workers'] >= old + new and stats[0]['subscribers'] == 0, stats[0] + by_workers = stats[0]['views_by_workers'] + write(module) + host = HttpHost([pure / 'build/pure-serve', '--apps', conf, '--view-workers', 0]) + try: + assert http(host.port, 'POST', '/s/inputs/bump', b'x')[0] == 200 + assert http(host.port, 'GET', '/s/views/slow')[3] == b'3000002' # its store kept the first bump + finally: + stats = host.stop() + assert stats[0]['view_workers'] == 0 and stats[0]['views_by_workers'] == 0, stats[0] + return {'old': old, 'new': new, 'stopped': answers.get((503, 'stopped', b''), 0), + 'gone': sum(v for k, v in answers.items() if k[0] == 404), 'by_workers': by_workers} + + PUSH_SOURCE = """record Row { name: text n = 0 @@ -4977,6 +5086,7 @@ def main(): report['server_runaway'] = server_runaway(aup, pure, work) report['server_view_bodies'] = server_view_bodies(aup, pure, work) report['server_push'] = server_push(aup, pure, work) + report['server_view_workers'] = server_view_workers(aup, pure, work) print(json.dumps(report)) -- cgit v1.2.3