diff options
| -rw-r--r-- | serve/serve.c | 651 | ||||
| -rw-r--r-- | tests/check.py | 253 |
2 files changed, 762 insertions, 142 deletions
diff --git a/serve/serve.c b/serve/serve.c index 3b713ef..e53d023 100644 --- a/serve/serve.c +++ b/serve/serve.c @@ -375,6 +375,12 @@ typedef struct Owner { * with --module has no name (its routes start /views/ and so on); each of * --apps has its own, the first segment of its routes. */ enum { APP_NAME = 32 }; +/* What an app did, added up over its starts (its executor's are its present run's). */ +typedef struct { + uint64_t bursts, events, ticks, memory_refused, windows_full, publications, answers, handed, checkpoints, delivered, + attempts; + PureLinkCounts sent; +} Counts; struct App { char name[APP_NAME + 1]; /* never changed once the app is in its place: threads find it by name */ PureModule *module; /* the module it runs, or ran or was given last (NULL: none loaded) */ @@ -384,18 +390,22 @@ struct App { * and snapshots may still point to it. */ _Atomic(const PureModule *) live; _Atomic bool gone; + char *module_path; /* its module's file as its line gives it (a host of apps), and that file's digest */ + uint64_t digest; const char *store; bool documents; /* the module serves /objects: it has show and edit */ bool owned; /* an owner runs it (states or outputs, or any of --apps); else the connection threads */ uint32_t show, edit; size_t memory; /* its budgets (0: none): what its states hold, */ _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 */ 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) */ /* Its wiring in --apps (send=, seal=, reply=), as --send, --seal and --reply give the one app's. */ const char *sends[PURE_MODULE_INPUTS], *seals[PURE_MODULE_INPUTS], *replies[PURE_MODULE_INPUTS]; unsigned send_count, seal_count, reply_count; + Counts past; /* its runs' before the present one */ Owner owner; }; static App *apps; @@ -409,6 +419,7 @@ static unsigned places(void) { return atomic_load_explicit(&app_count, memory_or static struct Retired { PureModule *module; struct Retired *next; } *retired_modules; static bool named_apps; /* --apps: a route's first segment names its app */ static Host *all_hosts; /* the connection threads, for moving a connection to its app's home */ +static unsigned host_count; static bool moved_home(Connection *c, size_t head_size); static void schedule(Owner *owner); @@ -1778,8 +1789,11 @@ static bool local_bind(Local *l, const PureModule *r, char *error, size_t size) * receiver that does not run holds it (LOCAL_RETRY_MS), and one that runs * another module now has its input found again first. */ static void local_hand(Executor *ex, Local *l, uint64_t now) { + pthread_mutex_lock(&l->from->lock); + bool leaving = l->from->stopping; /* its app stops: no batch more */ + pthread_mutex_unlock(&l->from->lock); pthread_mutex_lock(&l->lock); - bool waits = l->out || !l->count || (l->retry && now < l->retry); + bool waits = leaving || l->out || !l->count || (l->retry && now < l->retry); pthread_mutex_unlock(&l->lock); if (waits) return; Owner *to = l->to; @@ -1854,6 +1868,7 @@ static void local_hand(Executor *ex, Local *l, uint64_t now) { * again (after LOCAL_RETRY_MS when it had no room). The sender's turn * comes either way. */ static void local_done(Local *l, uint64_t heard, bool stored, bool full) { + Owner *from = l->from; /* read first: with out false the sender may stop, and let l go */ pthread_mutex_lock(&l->lock); l->out = false; l->out_bytes = 0; @@ -1863,7 +1878,6 @@ static void local_done(Local *l, uint64_t heard, bool stored, bool full) { if (full) l->retry = now_ms() + LOCAL_RETRY_MS; } pthread_mutex_unlock(&l->lock); - Owner *from = l->from; pthread_mutex_lock(&from->lock); from->delivered = true; pthread_cond_signal(&from->waiting); @@ -3369,14 +3383,22 @@ static bool wire_replies(Executor *ex, const PureModule *m, const char *const *r * less the one the owner of an app with states takes. */ /* An app's owner, its executor started on the app's store, its snapshot * published; false (why says why) when the app cannot start. */ +/* An owner's locks, made once, with its app's place (other apps' outlets + * lock them before the app starts, or when it never does). */ +static bool make_owner(Owner *owner) { + if (!owner->made && (pthread_mutex_init(&owner->lock, NULL) || pure_cond_monotonic(&owner->waiting) || + pthread_mutex_init(&owner->published.lock, NULL))) + return false; + owner->made = true; + return true; +} + static bool start_owner(App *app, Host *hosts, unsigned threads) { Owner *owner = &app->owner; - if (!owner->made && (pthread_mutex_init(&owner->lock, NULL) || pure_cond_monotonic(&owner->waiting) || - pthread_mutex_init(&owner->published.lock, NULL))) { + if (!make_owner(owner)) { snprintf(app->why, sizeof(app->why), "Cannot make the owner's locks"); return false; } - owner->made = true; if (!start_executor(&owner->executor, app)) return false; owner->executor.reads_first = true; owner->executor.published = &owner->published; @@ -3385,79 +3407,191 @@ static bool start_owner(App *app, Host *hosts, unsigned threads) { return true; } -/* --apps FILE: an app a line, NAME MODULE STORE and its budgets - * (memory=BYTES work=N queue=N), # beginning a comment. An app whose - * module does not load is kept, stopped, with why; false (and why on - * stderr) when the file cannot be read or a line is not an app. */ -static bool read_apps(const char *path) { +/* A line of --apps: an app as the file gives it, its strings its own. */ +typedef struct { + char name[APP_NAME + 1]; + char *module, *store; + uint64_t digest; /* of the module file's bytes when it was read (0: it could not be) */ + size_t memory, queue; + uint64_t work; + char *sends[PURE_MODULE_INPUTS], *seals[PURE_MODULE_INPUTS], *replies[PURE_MODULE_INPUTS]; + unsigned send_count, seal_count, reply_count; + bool kept; /* its app runs as the line says: a reading leaves it be */ +} Line; + +static void let_go_line(Line *line) { + free(line->module); + free(line->store); + for (unsigned i = 0; i < line->send_count; ++i) free(line->sends[i]); + for (unsigned i = 0; i < line->seal_count; ++i) free(line->seals[i]); + for (unsigned i = 0; i < line->reply_count; ++i) free(line->replies[i]); +} + +/* A file's bytes' digest (FNV-1a, 64 bits); 0 when it cannot be read. */ +static uint64_t file_digest(const char *path) { + FILE *file = fopen(path, "rb"); + if (!file) return 0; + uint64_t digest = 14695981039346656037u; + unsigned char bytes[16384]; + for (size_t n; (n = fread(bytes, 1, sizeof(bytes), file));) + for (size_t i = 0; i < n; ++i) digest = (digest ^ bytes[i]) * 1099511628211u; + bool read = !ferror(file); + fclose(file); + return read && digest ? digest : 1; +} + +/* --apps FILE: an app a line, NAME MODULE STORE, its budgets (memory=BYTES + * work=N queue=N) and its wiring (send=, seal=, reply=), # beginning a + * comment. False, why in error, when the file cannot be read, a line is not + * an app, or none is. */ +static bool parse_apps(const char *path, Line **out, unsigned *count, char *error, size_t size) { FILE *file = fopen(path, "r"); if (!file) { - fprintf(stderr, "--apps %s: %s\n", path, strerror(errno)); + snprintf(error, size, "--apps %s: %s", path, strerror(errno)); return false; } - char line[4096]; - unsigned at = 0; + Line *lines = NULL; + unsigned n = 0, at = 0; + char text[4096]; bool ok = true; - while (ok && fgets(line, sizeof(line), file)) { + while (ok && fgets(text, sizeof(text), file)) { ++at; - char *hash = strchr(line, '#'), *words[3 + 3 + 3 * PURE_MODULE_INPUTS], *rest = NULL; + char *hash = strchr(text, '#'), *words[3 + 3 + 3 * PURE_MODULE_INPUTS], *rest = NULL; if (hash) *hash = 0; - unsigned count = 0; - for (char *word = strtok_r(line, " \t\r\n", &rest); word && count < sizeof(words) / sizeof(*words); + unsigned found = 0; + for (char *word = strtok_r(text, " \t\r\n", &rest); word && found < sizeof(words) / sizeof(*words); word = strtok_r(NULL, " \t\r\n", &rest)) - words[count++] = word; - if (!count) continue; - App *grown = count >= 3 ? realloc(apps, (app_count + 1) * sizeof(*apps)) : NULL; + words[found++] = word; + if (!found) continue; + Line *grown = found >= 3 ? realloc(lines, (n + 1) * sizeof(*lines)) : NULL; if (!grown) { - fprintf(stderr, "--apps %s:%u: NAME MODULE STORE [memory=BYTES] [work=N] [queue=N] [send=OUTPUT=URL ...]\n", - path, at); + snprintf(error, size, "--apps %s:%u: NAME MODULE STORE [memory=BYTES] [work=N] [queue=N] [send=OUTPUT=URL ...]", + path, at); ok = false; break; } - apps = grown; - App *app = &apps[app_count]; - memset(app, 0, sizeof(*app)); + lines = grown; + Line *line = &lines[n]; + memset(line, 0, sizeof(*line)); size_t length = strlen(words[0]); bool named = length <= APP_NAME; for (size_t i = 0; named && i < length; ++i) { char c = words[0][i]; named = (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '_' || c == '-'; } - for (unsigned i = 0; named && i < app_count; ++i) named = strcmp(apps[i].name, words[0]) != 0; + for (unsigned i = 0; named && i < n; ++i) named = strcmp(lines[i].name, words[0]) != 0; if (!named) { - fprintf(stderr, "--apps %s:%u: a name is 1-32 of a-z 0-9 _ -, each app's its own\n", path, at); + snprintf(error, size, "--apps %s:%u: a name is 1-32 of a-z 0-9 _ -, each app's its own", path, at); ok = false; break; } - memcpy(app->name, words[0], length + 1); - for (unsigned w = 3; ok && w < count; ++w) { + memcpy(line->name, words[0], length + 1); + ok = (line->module = strdup(words[1])) && (line->store = strdup(words[2])); + for (unsigned w = 3; ok && w < found; ++w) { char *value = strchr(words[w], '='); - uint64_t n; + uint64_t number_given; if (value) *value++ = 0; - if (value && !strcmp(words[w], "memory") && number(value, SIZE_MAX, &n)) app->memory = (size_t)n; - else if (value && !strcmp(words[w], "work") && number(value, UINT64_MAX, &n)) app->work = n; - else if (value && !strcmp(words[w], "queue") && number(value, SIZE_MAX, &n)) app->owner.queue = (size_t)n; - else if (value && !strcmp(words[w], "send") && app->send_count < PURE_MODULE_INPUTS) - ok = (app->sends[app->send_count++] = strdup(value)) != NULL; - else if (value && !strcmp(words[w], "seal") && app->seal_count < PURE_MODULE_INPUTS) - ok = (app->seals[app->seal_count++] = strdup(value)) != NULL; - else if (value && !strcmp(words[w], "reply") && app->reply_count < PURE_MODULE_INPUTS) - ok = (app->replies[app->reply_count++] = strdup(value)) != NULL; + if (value && !strcmp(words[w], "memory") && number(value, SIZE_MAX, &number_given)) line->memory = (size_t)number_given; + else if (value && !strcmp(words[w], "work") && number(value, UINT64_MAX, &number_given)) line->work = number_given; + else if (value && !strcmp(words[w], "queue") && number(value, SIZE_MAX, &number_given)) line->queue = (size_t)number_given; + else if (value && !strcmp(words[w], "send") && line->send_count < PURE_MODULE_INPUTS) + ok = (line->sends[line->send_count++] = strdup(value)) != NULL; + else if (value && !strcmp(words[w], "seal") && line->seal_count < PURE_MODULE_INPUTS) + ok = (line->seals[line->seal_count++] = strdup(value)) != NULL; + else if (value && !strcmp(words[w], "reply") && line->reply_count < PURE_MODULE_INPUTS) + ok = (line->replies[line->reply_count++] = strdup(value)) != NULL; else { - fprintf(stderr, "--apps %s:%u: an app takes memory=BYTES, work=N, queue=N, send=OUTPUT=URL, " - "seal=OUTPUT=KEY and reply=OUTPUT=INPUT\n", path, at); + snprintf(error, size, "--apps %s:%u: an app takes memory=BYTES, work=N, queue=N, send=OUTPUT=URL, " + "seal=OUTPUT=KEY and reply=OUTPUT=INPUT", path, at); ok = false; } } - app->store = strdup(words[2]); - app->owned = true; /* every app of a host of apps takes turns on the workers */ - if (!app->store) ok = false; - else ++app_count; - if (ok && load_module(app, words[1])) app->running = true; /* until its store says otherwise; else stopped, with why */ + if (!ok && !*error) snprintf(error, size, "--apps %s:%u: out of memory", path, at); + if (!ok) { + let_go_line(line); + break; + } + line->digest = file_digest(line->module); + ++n; } fclose(file); - if (ok && !app_count) fprintf(stderr, "--apps %s: no app\n", path); - return ok && app_count; + if (ok && !n) { + snprintf(error, size, "--apps %s: no app", path); + ok = false; + } + if (!ok) { + for (unsigned i = 0; i < n; ++i) let_go_line(&lines[i]); + free(lines); + return false; + } + *out = lines; + *count = n; + return true; +} + +/* An app takes its line: its module's file, store, budgets and wiring (the + * strings move to it; those it had go). Its module is not read here. */ +static void take_line(App *app, Line *line) { + free(app->module_path); + free((char *)app->store); + for (unsigned i = 0; i < app->send_count; ++i) free((char *)app->sends[i]); + for (unsigned i = 0; i < app->seal_count; ++i) free((char *)app->seals[i]); + for (unsigned i = 0; i < app->reply_count; ++i) free((char *)app->replies[i]); + app->module_path = line->module; + app->digest = line->digest; + app->store = line->store; + app->memory = line->memory; + atomic_store(&app->work, line->work); + app->queue = line->queue; + memcpy(app->sends, line->sends, sizeof(app->sends)); + memcpy(app->seals, line->seals, sizeof(app->seals)); + memcpy(app->replies, line->replies, sizeof(app->replies)); + app->send_count = line->send_count, app->seal_count = line->seal_count, app->reply_count = line->reply_count; + *line = (Line){.kept = line->kept}; + memcpy(line->name, app->name, sizeof(line->name)); +} + +static bool same_strings(const char *const *a, unsigned a_count, char *const *b, unsigned b_count) { + for (unsigned i = 0; a_count == b_count && i < a_count; ++i) + if (strcmp(a[i], b[i])) return false; + return a_count == b_count; +} + +/* Whether an app runs as a line says: the same module (its file's bytes + * too), store, budgets and wiring. */ +static bool same_line(const App *app, const Line *line) { + return app->module_path && !strcmp(app->module_path, line->module) && app->digest == line->digest && + !strcmp(app->store, line->store) && app->memory == line->memory && atomic_load(&app->work) == line->work && + app->queue == line->queue && same_strings(app->sends, app->send_count, line->sends, line->send_count) && + same_strings(app->seals, app->seal_count, line->seals, line->seal_count) && + same_strings(app->replies, app->reply_count, line->replies, line->reply_count); +} + +/* A host of apps' first reading: each app in a place, room left for apps + * started later (slots), its module read (an app whose module does not + * load is kept, stopped, with why). False, why on stderr, when the file is + * not one of apps. */ +static bool read_apps(const char *path, unsigned slots) { + Line *lines; + unsigned count; + char error[512]; + if (!parse_apps(path, &lines, &count, error, sizeof(error))) { + fprintf(stderr, "%s\n", error); + return false; + } + app_slots = slots > count ? slots : count; + if (!(apps = calloc(app_slots, sizeof(*apps)))) return false; + for (unsigned i = 0; i < count; ++i) { + App *app = &apps[i]; + memcpy(app->name, lines[i].name, sizeof(app->name)); + app->owned = true; /* every app of a host of apps takes turns on the workers */ + if (!make_owner(&app->owner)) return false; + take_line(app, &lines[i]); + load_module(app, app->module_path); /* else stopped, with why */ + } + atomic_store(&app_count, count); + free(lines); + return true; } /* A JSON string's quotes and escapes around text. */ @@ -3481,6 +3615,272 @@ static unsigned auto_threads(bool owner) { return owner && cpus > 1 ? cpus - 1 : cpus; } + +/* ---- Apps that start and stop while the host runs (docs/design/017's stage 3) */ + +/* An executor's counts, once it stopped, added to what its app did. */ +static void fold_run(const Executor *ex, Counts *c) { + c->bursts += ex->bursts, c->events += ex->burst_events, c->ticks += ex->ticks; + c->memory_refused += ex->memory_refused, c->windows_full += ex->windows_full, c->publications += ex->publications; + c->answers += ex->answers, c->handed += ex->handed, c->checkpoints += ex->checkpoints; +} + +/* An executor's outlets closed, their counts added: what they did not + * deliver stays in the outbox, and goes again when the app starts again. */ +static void close_outlets(Executor *ex, Counts *c) { + for (unsigned i = 0; i < ex->wire_count; ++i) { + uint64_t delivered, attempts; + pure_wire_counts(ex->wires[i], &delivered, &attempts); + c->delivered += delivered, c->attempts += attempts; + pure_wire_close(ex->wires[i]); + } + for (unsigned i = 0; i < ex->link_count; ++i) { + PureLinkCounts k; + pure_link_counts(ex->links[i], &k); + c->sent.packets += k.packets, c->sent.records += k.records, c->sent.resent += k.resent, c->sent.lost += k.lost; + c->sent.acknowledged += k.acknowledged, c->sent.received += k.received, c->sent.stored += k.stored; + c->sent.refused += k.refused; + pure_link_close(ex->links[i]); + } + for (unsigned i = 0; i < ex->local_count; ++i) { + free(ex->locals[i]->batch); + pthread_mutex_destroy(&ex->locals[i]->lock); + free(ex->locals[i]); + } + ex->wire_count = ex->link_count = ex->local_count = 0; +} + +/* The workers: an app's owner takes its turns on them, none its own + * thread. One for the one app of a host; for apps, as many as run, at most + * a CPU each (or --workers): made as apps start, never let go. The main + * thread's, and the control's once the host runs. */ +static pthread_t *workers_made; +static unsigned worker_count, worker_room, workers_asked; +static bool more_workers(void) { + unsigned owners = 0, cpus = online_cpus(); + for (unsigned a = 0, count = places(); a < count; ++a) owners += apps[a].running && apps[a].owned; + unsigned want = !owners ? 0 : workers_asked ? workers_asked : !named_apps ? 1 : owners < cpus ? owners : cpus; + for (; worker_count < want; ++worker_count) { + if (worker_count == worker_room) { + unsigned room = worker_room ? 2 * worker_room : 4; + pthread_t *grown = realloc(workers_made, room * sizeof(*grown)); + if (!grown) return false; + workers_made = grown, worker_room = room; + } + if (pthread_create(&workers_made[worker_count], NULL, run_worker, NULL)) return false; + pthread_setname_np(workers_made[worker_count], "pure-owner"); /* their CPU, seen apart */ + } + return true; +} + +/* An app of a host of apps starts in its place: its store (none another + * running app has), its owner on it, its outlets and its snapshot; then it + * takes turns and requests, and the link streams to its inputs find them + * again. False, why in app->why, when it cannot: it stays stopped. */ +static bool start_app(App *app) { + struct stat mine, theirs; + app->why[0] = 0; + if (pure_store_init(app->store) != PURE_OK) snprintf(app->why, sizeof(app->why), "Cannot initialize the store directory"); + else if (stat(app->store, &mine)) snprintf(app->why, sizeof(app->why), "Cannot read the store directory"); + for (unsigned b = 0, count = places(); !app->why[0] && b < count; ++b) + if (&apps[b] != app && apps[b].running && !stat(apps[b].store, &theirs) && theirs.st_dev == mine.st_dev && + theirs.st_ino == mine.st_ino) + snprintf(app->why, sizeof(app->why), "The store is %s's", apps[b].name); + if (app->why[0]) return false; + Owner *o = &app->owner; + Executor *ex = &o->executor; + *ex = (Executor){0}; + if (!start_owner(app, all_hosts, host_count)) return false; + o->queue = app->queue; + bool wired = open_wires(app, app->name) && wire_replies(ex, app->module, app->replies, app->reply_count) && + wire_seals(ex, app->module, app->seals, app->seal_count, seal_key ? own_public : NULL); + if (!wired) snprintf(app->why, sizeof(app->why), "Its sends are wired wrong (said above)"); + pthread_mutex_lock(&o->lock); /* it takes requests resolved with its module */ + o->module = wired ? app->module : NULL; + o->stopping = !wired; + pthread_mutex_unlock(&o->lock); + if (wired) after_save(ex); /* the snapshot, and the sends stored before */ + char error[256] = ""; + for (unsigned i = 0; wired && i < ex->wire_count; ++i) + if (!pure_wire_start(ex->wires[i], wire_delivered, o)) snprintf(error, sizeof(error), "Cannot start a wire"); + for (unsigned i = 0; wired && !*error && i < ex->link_count; ++i) + pure_link_start(ex->links[i], wire_delivered, o, error, sizeof(error)); + if (!wired || *error) { + if (*error) snprintf(app->why, sizeof(app->why), "%s", error); + pthread_mutex_lock(&o->lock); + o->module = NULL; + o->stopping = true; + pthread_mutex_unlock(&o->lock); + close_outlets(ex, &app->past); + stop_executor(ex); + fold_run(ex, &app->past); + return false; + } + pthread_mutex_lock(&turns.lock); + o->live = true; + pthread_mutex_unlock(&turns.lock); + atomic_store(&app->gone, false); + atomic_store_explicit(&app->live, app->module, memory_order_release); + app->running = true; + if (link_listener) { + char prefix[APP_NAME + 2]; + snprintf(prefix, sizeof(prefix), "%s/", app->name); + pure_link_listener_rebind(link_listener, prefix); + } + schedule(o); /* a first turn: its timers begin */ + return true; +} + +/* An app stops: no request taken from now on (its routes answer 503 + * stopped, link records and other apps' batches wait at their senders), + * those taken applied, saved and answered, its batches with other apps + * back, then its outlets closed (what they did not deliver stays in its + * outbox) and its store closed. Its owner takes no turn after. */ +static void stop_app(App *app) { + if (!app->running) return; + app->running = false; + atomic_store_explicit(&app->live, NULL, memory_order_release); + Owner *o = &app->owner; + Executor *ex = &o->executor; + pthread_mutex_lock(&o->lock); + o->stopping = true; + o->module = NULL; + pthread_cond_signal(&o->waiting); /* it may be gathering */ + pthread_mutex_unlock(&o->lock); + for (bool done = false; !done;) { + pthread_mutex_lock(&o->lock); + bool taken = o->count > 0; + pthread_mutex_unlock(&o->lock); + bool out = false; + for (unsigned i = 0; i < ex->local_count; ++i) { + pthread_mutex_lock(&ex->locals[i]->lock); + out = out || ex->locals[i]->out; + pthread_mutex_unlock(&ex->locals[i]->lock); + } + pthread_mutex_lock(&turns.lock); + done = !taken && !out && !o->queued && !o->running; + if (done) o->live = false; + pthread_mutex_unlock(&turns.lock); + if (!done) nanosleep(&(struct timespec){0, 1000000}, NULL); + } + close_outlets(ex, &app->past); + stop_executor(ex); + fold_run(ex, &app->past); + o->gather = false; + o->due = 0; +} + +/* A reading of --apps on SIGHUP: the host made as the file says. Apps whose + * lines are gone stop and their routes go (404 no_app); those whose lines + * changed, and those that do not run (they could not start, or a save of + * theirs is uncertain), start (again) from their stores; new ones start in + * places of their own; the others are left be. A line out for each + * reading: each app, whether it runs (and why not), and what was done. */ +static unsigned readings; +static void read_again(const char *path) { + Line *lines = NULL; + unsigned count = 0, before = places(); + char error[512]; + ++readings; + const char **did = calloc(app_slots, sizeof(*did)); + if (!did || !parse_apps(path, &lines, &count, error, sizeof(error))) { + if (!did) snprintf(error, sizeof(error), "out of memory"); + printf("{\"reading\":%u,\"error\":", readings); + print_json(stdout, error); + printf("}\n"); + fflush(stdout); + free(did); + return; + } + /* First what goes or changes stops: a store an app lets go another may take. */ + for (unsigned a = 0; a < before; ++a) { + App *app = &apps[a]; + if (atomic_load(&app->gone)) continue; + Line *line = NULL; + for (unsigned i = 0; i < count && !line; ++i) + if (!strcmp(lines[i].name, app->name)) line = &lines[i]; + if (!line) { + stop_app(app); + atomic_store(&app->gone, true); + did[a] = "removed"; + continue; + } + bool uncertain = app->running && __atomic_load_n(&app->owner.executor.app.uncertain, __ATOMIC_RELAXED); + if (app->running && !uncertain && same_line(app, line)) { + line->kept = true; + did[a] = "kept"; + continue; + } + did[a] = app->running ? "restarted" : "started"; + stop_app(app); + } + /* Each new name a place, before any starts: an app may send to one + * written after it. */ + for (unsigned i = 0; i < count; ++i) { + unsigned a = 0, taken = places(); + while (a < taken && strcmp(apps[a].name, lines[i].name)) ++a; + if (a < taken || taken == app_slots || !make_owner(&apps[a].owner)) continue; + App *app = &apps[a]; + memcpy(app->name, lines[i].name, sizeof(app->name)); + app->owned = true; + app->home = a % host_count; + atomic_store(&app->gone, false); + atomic_store_explicit(&app_count, taken + 1, memory_order_release); /* its name is there now */ + did[a] = "started"; + } + for (unsigned i = 0; i < count; ++i) { + Line *line = &lines[i]; + unsigned a = 0, taken = places(); + while (a < taken && strcmp(apps[a].name, line->name)) ++a; + if (line->kept) continue; + if (a == taken) { + fprintf(stderr, "{\"app\":\"%s\",\"error\":\"no place for it (--app-slots %u)\"}\n", line->name, app_slots); + continue; + } + App *app = &apps[a]; + bool same_module = app->module && app->module_path && !strcmp(app->module_path, line->module) && + app->digest == line->digest && line->digest; + atomic_store(&app->gone, false); /* in the file again: stopped, not gone, until it starts */ + if (!did[a]) did[a] = "started"; + take_line(app, line); + app->why[0] = 0; + if (same_module || load_module(app, app->module_path)) start_app(app); /* else stopped, with why */ + } + more_workers(); + printf("{\"reading\":%u,\"apps\":[", readings); + bool first = true; + for (unsigned a = 0, taken = places(); a < taken; ++a) { + App *app = &apps[a]; + if (atomic_load(&app->gone) && !did[a]) continue; + printf("%s{\"name\":", first ? "" : ","); + first = false; + print_json(stdout, app->name); + printf(",\"running\":%s", app->running ? "true" : "false"); + if (did[a]) printf(",\"did\":\"%s\"", did[a]); + if (!app->running && !atomic_load(&app->gone)) { + printf(",\"why\":"); + print_json(stdout, app->why); + } + printf("}"); + } + printf("]}\n"); + fflush(stdout); + for (unsigned i = 0; i < count; ++i) let_go_line(&lines[i]); + free(lines); + free(did); +} + +/* The thread that reads --apps again: on SIGHUP, which every other thread + * blocks, until the host stops (it is sent one more then). */ +static void *run_apps(void *context) { + const char *path = context; + sigset_t hangup; + sigemptyset(&hangup); + sigaddset(&hangup, SIGHUP); + for (int signal_number; !sigwait(&hangup, &signal_number) && !atomic_load(&stopping);) read_again(path); + return NULL; +} + int main(int argc, char **argv) { static PureSurfaceHost surface; /* What follows -- is the surface's: its library first, then its arguments. */ @@ -3539,14 +3939,12 @@ int main(int argc, char **argv) { "--surface are for a host of one app (--module)\n", stderr); return 2; } - if (!read_apps(apps_path)) return 2; - /* Each app in a place of its own, room left for apps started later. */ - app_slots = slots > app_count ? slots : app_count; - App *placed = calloc(app_slots, sizeof(*placed)); - if (!placed) return 1; - memcpy(placed, apps, app_count * sizeof(*apps)); - free(apps); - apps = placed; + /* SIGHUP is the reading thread's (run_apps): blocked before any thread is made, as they all inherit it. */ + sigset_t hangup; + sigemptyset(&hangup); + sigaddset(&hangup, SIGHUP); + pthread_sigmask(SIG_BLOCK, &hangup, NULL); + if (!read_apps(apps_path, slots)) return 2; named_apps = true; } else { if (!module_path || !store || !(apps = calloc(1, sizeof(*apps)))) goto usage; @@ -3583,12 +3981,14 @@ int main(int argc, char **argv) { pthread_t *thread = calloc(threads, sizeof(*thread)); if (!hosts || !thread || pure_cond_monotonic(&turns.wake)) return 1; all_hosts = hosts; + host_count = threads; for (unsigned a = 0; a < app_count; ++a) apps[a].home = a % threads; /* each app's connections meet on one thread */ /* Each app's store, and its owner: an app that cannot start stops alone * (in a host of apps; the one app of a host stops the host). Two apps * never share a store. */ - unsigned owners = 0; - for (unsigned a = 0; a < app_count; ++a) { + for (unsigned a = 0; named_apps && a < app_count; ++a) + if (apps[a].module) start_app(&apps[a]); /* else stopped, with why */ + for (unsigned a = 0; !named_apps && a < app_count; ++a) { App *app = &apps[a]; if (!app->running) continue; struct stat mine, theirs; @@ -3602,11 +4002,10 @@ int main(int argc, char **argv) { shared = true; } app->running = !app->why[0] && (!app->owned || start_owner(app, hosts, threads)); - if (!app->running && !named_apps) { + if (!app->running) { fprintf(stderr, "%s\n", app->why); return 1; } - owners += app->running && app->owned; } /* The sender's name: --name, or the module file's name without .pmod. */ char stem[PURE_SOURCE + 1]; @@ -3624,27 +4023,13 @@ int main(int argc, char **argv) { memcpy(apps[0].replies, replies, reply_count * sizeof(*replies)); apps[0].send_count = send_count, apps[0].seal_count = seal_count, apps[0].reply_count = reply_count; } - /* Each app's wiring, its sends' source APP.OUTPUT (the one app's - * --name.OUTPUT): an app of --apps wired wrong stops alone. */ - for (unsigned a = 0; a < app_count; ++a) { - App *app = &apps[a]; - Executor *ex = &app->owner.executor; - if (!app->running || !app->owned) continue; - if (open_wires(app, named_apps ? app->name : name) && - wire_replies(ex, app->module, app->replies, app->reply_count) && - wire_seals(ex, app->module, app->seals, app->seal_count, seal_key ? own_public : NULL)) - continue; - if (!named_apps) return 2; - snprintf(app->why, sizeof(app->why), "Its sends are wired wrong (said above)"); - for (unsigned i = 0; i < ex->wire_count; ++i) pure_wire_close(ex->wires[i]); - for (unsigned i = 0; i < ex->link_count; ++i) pure_link_close(ex->links[i]); - for (unsigned i = 0; i < ex->local_count; ++i) free(ex->locals[i]); - ex->wire_count = ex->link_count = ex->local_count = 0; - stop_executor(ex); - app->running = false; - --owners; - } - for (unsigned a = 0; a < app_count; ++a) { + /* The one app's wiring, its sends' source --name.OUTPUT (an app of + * --apps wired itself as it started, APP.OUTPUT). */ + if (!named_apps && apps[0].owned && + (!open_wires(&apps[0], name) || !wire_replies(&owner->executor, apps[0].module, apps[0].replies, apps[0].reply_count) || + !wire_seals(&owner->executor, apps[0].module, apps[0].seals, apps[0].seal_count, seal_key ? own_public : NULL))) + return 2; + for (unsigned a = 0; !named_apps && a < app_count; ++a) { App *app = &apps[a]; if (!app->running) continue; if (app->owned) { /* it takes requests resolved with its module, and turns */ @@ -3654,7 +4039,7 @@ int main(int argc, char **argv) { } atomic_store(&app->live, app->module); } - for (unsigned a = 0; a < app_count; ++a) { + for (unsigned a = 0; !named_apps && a < app_count; ++a) { Owner *o = &apps[a].owner; if (!apps[a].running || !apps[a].owned) continue; for (unsigned i = 0; i < o->executor.wire_count; ++i) @@ -3673,19 +4058,9 @@ int main(int argc, char **argv) { hosts[i].listener = hosts[i].epoll = -1; if (!start_shard(&hosts[i], budget, &port)) return 1; } - /* The workers: an app's owner takes its turns on them, none its own - * thread. One for the one app of a host; for apps, as many as run, at - * most a CPU each. */ - if (!workers) workers = named_apps ? (owners < online_cpus() ? owners : online_cpus()) : 1; - if (!workers) workers = 1; - pthread_t *worker = calloc(workers, sizeof(*worker)); - if (!worker) return 1; - for (unsigned i = 0; owners && i < workers; ++i) { - if (pthread_create(&worker[i], NULL, run_worker, NULL)) { fputs("Cannot start a worker\n", stderr); return 1; } - pthread_setname_np(worker[i], "pure-owner"); /* their CPU, seen apart */ - } - for (unsigned a = 0; a < app_count; ++a) - if (apps[a].running && apps[a].owned) schedule(&apps[a].owner); /* a first turn: its timers begin */ + workers_asked = workers; + if (!more_workers()) { fputs("Cannot start a worker\n", stderr); return 1; } + if (!named_apps && apps[0].owned) schedule(owner); /* a first turn: its timers begin (an app of --apps had its) */ if (link_port) { /* records from the senders named, over UDP */ char error[256]; if ((!named_apps && !apps[0].owned) || !seal_key) { @@ -3708,7 +4083,7 @@ int main(int argc, char **argv) { } } if (named_apps) { /* each app, and why one stopped */ - printf("{\"port\":%u,\"threads\":%u,\"workers\":%u,\"apps\":[", port, threads, owners ? workers : 0); + printf("{\"port\":%u,\"threads\":%u,\"workers\":%u,\"apps\":[", port, threads, worker_count); for (unsigned a = 0; a < app_count; ++a) { printf("%s{\"name\":", a ? "," : ""); print_json(stdout, apps[a].name); @@ -3727,6 +4102,10 @@ int main(int argc, char **argv) { else printf("\"}\n"); } fflush(stdout); + /* A host of apps reads its file again on SIGHUP, on a thread of its own. */ + pthread_t reading; + if (named_apps && pthread_create(&reading, NULL, run_apps, (void *)apps_path)) { fputs("Cannot start a thread\n", stderr); return 1; } + if (named_apps) pthread_setname_np(reading, "pure-apps"); /* 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; @@ -3738,7 +4117,11 @@ int main(int argc, char **argv) { surface_changed(&surface); pthread_join(surface.thread, NULL); } - for (unsigned a = 0; a < app_count; ++a) { /* no gathering from now on */ + if (named_apps) { /* a reading under way ends first */ + pthread_kill(reading, SIGHUP); + pthread_join(reading, NULL); + } + for (unsigned a = 0, count = places(); a < count; ++a) { /* no gathering from now on */ Owner *o = &apps[a].owner; if (!apps[a].running || !apps[a].owned) continue; pthread_mutex_lock(&o->lock); @@ -3750,7 +4133,7 @@ int main(int argc, char **argv) { turns.stopping = true; pthread_cond_broadcast(&turns.wake); pthread_mutex_unlock(&turns.lock); - for (unsigned i = 0; owners && i < workers; ++i) pthread_join(worker[i], NULL); + for (unsigned i = 0; i < worker_count; ++i) pthread_join(workers_made[i], NULL); PureLinkCounts sent = {0}, heard = {0}; /* what links sent, and what the listener took */ pure_link_listener_close(link_listener, &heard); /* after the owner, which tells it what it stored */ size_t active = 0, retained = 0, state_bytes = 0; @@ -3770,42 +4153,24 @@ int main(int argc, char **argv) { expired += hosts[i].expired; stop_shard(&hosts[i]); } - for (unsigned a = 0; a < app_count; ++a) { - Executor *ex = &apps[a].owner.executor; - if (!apps[a].running || !apps[a].owned) continue; - bursts += ex->bursts; - events += ex->burst_events; - ticks += ex->ticks; - memory_refused += ex->memory_refused; - windows_full += ex->windows_full; - publications += ex->publications; - answers += ex->answers; - for (unsigned i = 0; i < ex->wire_count; ++i) { - uint64_t d, t; - pure_wire_counts(ex->wires[i], &d, &t); - delivered += d; - attempts += t; - pure_wire_close(ex->wires[i]); /* what was not delivered stays in the outbox */ - } - for (unsigned i = 0; i < ex->link_count; ++i) { - PureLinkCounts c; - pure_link_counts(ex->links[i], &c); - sent.packets += c.packets, sent.records += c.records, sent.resent += c.resent, sent.lost += c.lost; - sent.acknowledged += c.acknowledged, sent.received += c.received, sent.stored += c.stored; - sent.refused += c.refused; - pure_link_close(ex->links[i]); /* what was not stored stays in the outbox */ - } - for (unsigned i = 0; i < ex->local_count; ++i) { - free(ex->locals[i]->batch); - pthread_mutex_destroy(&ex->locals[i]->lock); - free(ex->locals[i]); + for (unsigned a = 0, count = places(); a < count; ++a) { /* each app's runs, the present one stopped now */ + App *app = &apps[a]; + Counts *c = &app->past; + if (app->running && app->owned) { + Executor *ex = &app->owner.executor; + state_bytes += pure_app_state_bytes(&ex->app); + close_outlets(ex, c); /* what was not delivered stays in the outbox */ + stop_executor(ex); + fold_run(ex, c); } - handed += ex->handed; - state_bytes += pure_app_state_bytes(&ex->app); - stop_executor(ex); - checkpoints += ex->checkpoints; - free(apps[a].owner.jobs); - free(apps[a].owner.spare); + free(app->owner.jobs); + free(app->owner.spare); + bursts += c->bursts, events += c->events, ticks += c->ticks, memory_refused += c->memory_refused; + windows_full += c->windows_full, publications += c->publications, answers += c->answers, handed += c->handed; + checkpoints += c->checkpoints, delivered += c->delivered, attempts += c->attempts; + sent.packets += c->sent.packets, sent.records += c->sent.records, sent.resent += c->sent.resent; + sent.lost += c->sent.lost, sent.acknowledged += c->sent.acknowledged, sent.received += c->sent.received; + sent.stored += c->sent.stored, sent.refused += c->sent.refused; } unsigned long groups, grouped; pure_store_group_counts(&groups, &grouped); @@ -3829,21 +4194,21 @@ int main(int argc, char **argv) { windows_full); if (named_apps) { /* what each app did */ fprintf(stderr, ",\"workers\":%u,\"moved_home\":%" PRIu64 ",\"handed\":%" PRIu64 ",\"apps\":[", - owners ? workers : 0, moved, handed); - for (unsigned a = 0; a < app_count; ++a) { - const Owner *o = &apps[a].owner; + worker_count, moved, handed); + for (unsigned a = 0, count = places(); a < count; ++a) { + const App *app = &apps[a]; fprintf(stderr, "%s{\"name\":", a ? "," : ""); - print_json(stderr, apps[a].name); + print_json(stderr, app->name); fprintf(stderr, ",\"running\":%s,\"turns\":%" PRIu64 ",\"bursts\":%" PRIu64 ",\"events\":%" PRIu64 - ",\"busy\":%" PRIu64 ",\"memory_refused\":%" PRIu64 "}", apps[a].running ? "true" : "false", - o->turns, o->executor.bursts, o->executor.burst_events, o->busy, o->executor.memory_refused); + ",\"busy\":%" PRIu64 ",\"memory_refused\":%" PRIu64 "}", app->running ? "true" : "false", + app->owner.turns, app->past.bursts, app->past.events, app->owner.busy, app->past.memory_refused); } fputc(']', stderr); } fputs("}\n", stderr); free(hosts); free(thread); - free(worker); + free(workers_made); return 0; usage: fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] " @@ -3851,13 +4216,15 @@ usage: "[--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] [--surface LIBRARY [-- ARGUMENTS]]\n" - "pure-serve --apps FILE [--port 8767] [--threads 1|N|auto] [--workers N] [--budget BYTES] [--connections 1024] " + "pure-serve --apps FILE [--port 8767] [--threads 1|N|auto] [--workers N] [--app-slots 256] [--budget BYTES] [--connections 1024] " "[--timeout-ms 10000] [--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" " [reply=OUTPUT=INPUT ...]; its routes start /NAME/, its sends' sources are NAME.OUTPUT, and a\n" - " --link-port takes its inputs as NAME/INPUT\n" + " --link-port takes its inputs as NAME/INPUT. On SIGHUP the host reads FILE again: the apps of lines\n" + " gone stop, those of new or changed lines start (again), from their stores; a line out says what it did\n" + " --app-slots N: the apps a host of apps holds at most, those started later included\n" " --workers N: the threads the apps' owners take turns on (1 for one app; for apps, one a CPU at most)\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" diff --git a/tests/check.py b/tests/check.py index 91fa9b9..3876bf8 100644 --- a/tests/check.py +++ b/tests/check.py @@ -3980,6 +3980,258 @@ def server_apps_links(aup, pure, work): return {'sent': sent, 'killed_at': killed_at, 'handed': near_stats['handed'], 'small_held': held} +LIFE_SOURCE = """state n = 0 +state hits = 0 +input add = message { step: number } +on add { + n = n + add.step + hits = hits + 1 +} +view total = n +view count = hits +""" +# Its next version: another input before add (add's number moves), and a state more. +LIFE_NEXT = """state twice = 0 +state n = 0 +state hits = 0 +input reset = message { to: number } +input add = message { step: number } +on reset { + n = reset.to +} +on add { + n = n + add.step + hits = hits + 1 + twice = twice + 2 * add.step +} +view total = n +view count = hits +view doubled = twice +""" +LIFE_SENDER = """state n = 0 +input put = message { k: u32 } +output here = message { k: u32 } +output far = message { k: u32 } +on put { + n = n + 1 + send here { k: put.k } + send far { k: put.k } +} +view count = n +""" +# A taker's next version: an input before take, which takes a field more (the sends by name). +LIFE_TAKER_NEXT = LINKS_TAKER.replace('input take = message { k: u32 }', """state pokes = 0 +input poke = message { x: u32 } +input take = message { + note: text + k: u32 +} +on poke { + pokes = pokes + poke.x +}""") + + +def server_apps_lifecycle(aup, pure, work): + """Apps start, stop and change on a running host (docs/design/017, + stage 3): on SIGHUP it reads its --apps file again and says what it did + on a line. An app added answers; an app removed goes (404 no_app) and + another process opens its store; apps change module while requests, + another app's local sends and another node's link records come to them, + the new modules' inputs at other numbers and taking the sends by name: + a write answered 200 is applied once and one answered 503 stopped not at + all, every send arrives once and in order, and the app left alone + answers every request. An app that could not open its store starts at + the next reading once it can; a file with a wrong line changes nothing; + after a kill -9 the host starts as the file says, the states as saved.""" + import os + import signal + import select + import socket + import threading + import time + for name, text in (('life', LIFE_SOURCE), ('life-next', LIFE_NEXT), ('life-sender', LIFE_SENDER), + ('life-taker', LINKS_TAKER), ('life-taker-next', LIFE_TAKER_NEXT)): + (work / f'{name}.aup').write_text(text) + compile_module(aup, work / f'{name}.aup', work / f'{name}.pmod') + serve = pure / 'build/pure-serve' + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe: + probe.bind(('127.0.0.1', 0)) + link_port = probe.getsockname()[1] + keys = {} + for side in ('host', 'far'): + (work / f'life-{side}-key.conf').write_text(f"x {work / 'life.pmod'} {work / f'life-keys-{side}'}\n") + host = HttpHost([serve, '--apps', work / f'life-{side}-key.conf', '--key', work / f'life-{side}.key']) + keys[side] = host.info['public_key'] + host.stop() + module = {name: work / f'{name}.pmod' for name in ('life', 'life-next', 'life-sender', 'life-taker', 'life-taker-next')} + store = {name: work / f'life-store-{name}' for name in ('a', 'b', 'c', 'src', 'dst', 'dstl', 'e', 'srcf')} + (work / 'life-ro').mkdir() + store['e'] = work / 'life-ro' / 'e' + lines = {'a': f"a {module['life']} {store['a']}", 'b': f"b {module['life']} {store['b']}", + 'src': f"src {module['life-sender']} {store['src']} send=here=dst/take", + 'dst': f"dst {module['life-taker']} {store['dst']}", 'dstl': f"dstl {module['life-taker']} {store['dstl']}"} + conf = work / 'life.conf' + + def write(names): + conf.write_text(''.join(lines[name] + '\n' for name in names)) + + write(['a', 'b', 'src', 'dst', 'dstl']) + args = [serve, '--apps', conf, '--key', work / 'life-host.key', '--link-port', link_port, '--threads', 2, + '--sender', f"srcf.far={keys['far']}"] + host = HttpHost(args) + (work / 'life-far.conf').write_text(f"srcf {module['life-sender']} {store['srcf']} " + f"send=far=pure://127.0.0.1:{link_port}/dstl/take seal=far={keys['host']}\n") + far = HttpHost([serve, '--apps', work / 'life-far.conf', '--key', work / 'life-far.key']) + + def reading(names=None): + """The host reads its file again: what it said.""" + if names is not None: + write(names) + host.process.send_signal(signal.SIGHUP) + ready, _, _ = select.select([host.process.stdout], [], [], 30) + assert ready, 'no reading' + said = json.loads(host.process.stdout.readline()) + return said, {app['name']: app for app in said.get('apps', [])} + + def number(port, path): + status, why, _, body = http(port, 'GET', path) + assert status == 200, (path, status, why) + return int(body) if body not in (b'true', b'false') else body == b'true' + + sent = 600 + try: + # Added: it answers. Removed: its routes go, and its store is let go. + assert http(host.port, 'POST', '/a/inputs/add?step=5')[0] == 200 + assert http(host.port, 'POST', '/b/inputs/add?step=7')[0] == 200 + lines['c'] = f"c {module['life']} {store['c']}" + _, did = reading(['a', 'b', 'src', 'dst', 'dstl', 'c']) + assert did['c'] == {'name': 'c', 'running': True, 'did': 'started'} and did['a']['did'] == 'kept', did + assert http(host.port, 'POST', '/c/inputs/add?step=2')[0] == 200 and number(host.port, '/c/views/total') == 2 + _, did = reading(['a', 'src', 'dst', 'dstl', 'c']) + assert did['b'] == {'name': 'b', 'running': False, 'did': 'removed'}, did + assert http(host.port, 'GET', '/b/views/total')[:2] == (404, 'no_app') + other = HttpHost([serve, '--module', module['life'], '--store', store['b']]) + try: + assert number(other.port, '/views/total') == 7 + finally: + other.stop() + # Changed under load: a, dst (another app's local sends) and dstl (another node's link records) take + # modules whose inputs are at other numbers, the takers' by name; then the senders start again, src + # (its local outlet) and, on the far node, srcf (its link); c, left alone, answers every request. Each + # change comes while the sends go (as far as each putter got: more than none, less than all). + answered, refused, done = [0], [0], threading.Event() + progress, trouble = {'src': 0, 'srcf': 0}, [] + + def adder(): + while not done.is_set(): + status, why, _, _ = http(host.port, 'POST', '/a/inputs/add?step=1') + if status == 200: + answered[0] += 1 + elif (status, why) == (503, 'stopped'): + refused[0] += 1 + else: + trouble.append(('a', status, why)) + + def putter(port, app): + for k in range(1, sent + 1): + status, why, _, _ = http(port, 'POST', f'/{app}/inputs/put?k={k}') + while (status, why) == (503, 'stopped'): # its app starting again: not applied, sent again + time.sleep(0.005) + status, why, _, _ = http(port, 'POST', f'/{app}/inputs/put?k={k}') + if status != 200: + trouble.append((app, k, status, why)) + return + progress[app] = k + time.sleep(0.002) + + def watcher(): + while not done.is_set(): + status, why, _, _ = http(host.port, 'GET', '/c/views/total') + if status != 200: + trouble.append(('c', status, why)) + + def once(app, k): + deadline = time.time() + 30 + while progress[app] < k and time.time() < deadline and not trouble: + time.sleep(0.005) + return dict(progress), answered[0] + + workers = [threading.Thread(target=adder), threading.Thread(target=putter, args=(host.port, 'src')), + threading.Thread(target=putter, args=(far.port, 'srcf')), threading.Thread(target=watcher)] + for thread in workers: + thread.start() + during = [] + during.append(once('src', 100)) + lines['a'] = f"a {module['life-next']} {store['a']}" + lines['dst'] = f"dst {module['life-taker-next']} {store['dst']}" + lines['dstl'] = f"dstl {module['life-taker-next']} {store['dstl']}" + _, did = reading(['a', 'src', 'dst', 'dstl', 'c']) + assert [did[name]['did'] for name in ('a', 'dst', 'dstl', 'c', 'src')] == \ + ['restarted', 'restarted', 'restarted', 'kept', 'kept'], did + during.append(once('src', 200)) + lines['dst'] = f"dst {module['life-taker']} {store['dst']}" # and back: take's number moves again + _, did = reading(['a', 'src', 'dst', 'dstl', 'c']) + assert did['dst']['did'] == 'restarted' and did['a']['did'] == 'kept', did + during.append(once('src', 300)) + lines['src'] += ' queue=100000' # the sender starts again (its module as it was), its outbox going on + _, did = reading(['a', 'src', 'dst', 'dstl', 'c']) + assert did['src']['did'] == 'restarted' and did['dst']['did'] == 'kept', did + during.append(once('srcf', 400)) + (work / 'life-far.conf').write_text((work / 'life-far.conf').read_text().replace('\n', ' queue=100000\n')) + far.process.send_signal(signal.SIGHUP) # the far node's sender starts again, its link with it + ready, _, _ = select.select([far.process.stdout], [], [], 30) + assert ready and json.loads(far.process.stdout.readline())['apps'][0]['did'] == 'restarted' + for thread in workers[1:3]: + thread.join() + done.set() + for thread in (workers[0], workers[3]): + thread.join() + assert not trouble, trouble[:5] + for put, adds in during: # each change came while the sends went + assert 0 < put['src'] < sent and 0 < put['srcf'] < sent and adds > 0, during + whole = [sent, sent * (sent + 1) // 2, True] + deadline = time.time() + 30 + while time.time() < deadline and ([number(host.port, f'/dst/views/{v}') for v in ('count', 'sum', 'in_order')] != whole or + [number(host.port, f'/dstl/views/{v}') for v in ('count', 'sum', 'in_order')] != whole): + time.sleep(0.05) + for app in ('dst', 'dstl'): + got = [number(host.port, f'/{app}/views/{v}') for v in ('count', 'sum', 'in_order')] + assert got == whole, (app, got) + total = 5 + answered[0] + assert number(host.port, '/a/views/total') == total and number(host.port, '/a/views/count') == 1 + answered[0], \ + (number(host.port, '/a/views/total'), answered[0], refused[0]) + # An app whose store cannot be made yet stops alone; once it can, the next reading starts it. + lines['e'] = f"e {module['life']} {store['e']}" + os.chmod(work / 'life-ro', 0o500) + try: + _, did = reading(['a', 'src', 'dst', 'dstl', 'c', 'e']) + finally: + os.chmod(work / 'life-ro', 0o700) + assert not did['e']['running'] and 'store' in did['e']['why'] and did['c']['did'] == 'kept', did + _, did = reading() + assert did['e'] == {'name': 'e', 'running': True, 'did': 'started'}, did + assert http(host.port, 'POST', '/e/inputs/add?step=3')[0] == 200 + # A file with a wrong line changes nothing. + good = conf.read_text() + conf.write_text(good + 'two words\n') # NAME MODULE STORE at least + said, _ = reading() + assert 'error' in said and number(host.port, '/a/views/total') == total, said + conf.write_text(good) + # After a kill -9 the host starts as the file says, its states as saved. + host.process.send_signal(signal.SIGKILL) + host.process.wait() + host = HttpHost(args) + assert all(app['running'] for app in host.info['apps']), host.info + assert sorted(app['name'] for app in host.info['apps']) == ['a', 'c', 'dst', 'dstl', 'e', 'src'], host.info + assert number(host.port, '/a/views/total') == total and number(host.port, '/e/views/total') == 3 + assert [number(host.port, f'/dst/views/{v}') for v in ('count', 'sum', 'in_order')] == whole + finally: + host.stop() + far.stop() + return {'readings': 8, 'far_readings': 1, 'sent': sent, 'answered_while_changing': answered[0], + 'refused_while_changing': refused[0], 'puts_at_changes': [put['src'] for put, _ in during]} + + def main(): global EDITOR parser = argparse.ArgumentParser() @@ -4031,6 +4283,7 @@ def main(): report['server_work'] = server_work(aup, pure, work) report['server_apps'] = server_apps(aup, pure, work) report['server_apps_links'] = server_apps_links(aup, pure, work) + report['server_apps_lifecycle'] = server_apps_lifecycle(aup, pure, work) print(json.dumps(report)) |