Run a surface for people inside the node (design 010)

pure-serve --surface LIBRARY [-- ARGUMENTS] maps a library and runs its
pure_surface_run in a thread of its own (serve/surface.h). The surface
reads the app as of the last save where it is, a published snapshot in a
reader app, as a connection thread reads views; an eventfd tells it of
each new snapshot; and its events (a key pressed, text typed) are
requests without a host in the owner's queue, applied and saved with the
round, their results handed back to the surface's waiter. So one node
serves programs over HTTP and people on its display with one owner of
its states, and nothing between the states and the screen. pure-serve
exports its pure_* functions for the surface to call, one copy of the
app model in the process; the table carries sizeof(PureApp) and
sizeof(PureModule) so a surface built for another layout refuses.

The check builds a probe surface that types, presses and reads: its
events show in HTTP views, an HTTP input shows in it, wrong inputs are
refused, it stops with the server, and a module without states cannot
have one. Without a surface the request path measured the same (small
GETs 140.9k vs 139.9k, app totals 126.3k vs 126.3k over eight rounds);
the exported symbols cost 11 KB of the file and 24 KiB idle.

2 files changed, 211 insertions, 11 deletions

diff --git a/serve/serve.c b/serve/serve.c
index 0e60e2a..6467683 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -31,12 +31,14 @@
  * bursts take the requests of every thread, one save each. */
 #include "deliver.h"
 #include "seal.h"
+#include "surface.h"
 #include "message-json.h"
 #include "module-store.h"
 #include "module-text.h"
 #include "runtime.h"
 #include "store.h"
 #include <arpa/inet.h>
+#include <dlfcn.h>
 #include <errno.h>
 #include <fcntl.h>
 #include <inttypes.h>
@@ -139,8 +141,12 @@ typedef struct Connection {
  * it. A read answered from the store lends its extents: they point into a
  * store mapping, which the resource holds and the runtime charges and
  * releases with the ticket. */
+/* A surface's event, waiting for its save: done, and how it went. */
+typedef struct { PureResult status; bool done; } Waiter;
+
 typedef struct {
-    Host *host;
+    Host *host;       /* NULL for a surface's event */
+    Waiter *waiter;   /* and then its waiter */
     PureTicket ticket;
     PureInput input;
     PureResult status;
@@ -178,6 +184,7 @@ typedef struct {
     pthread_mutex_t published_lock;
     PureSnapshot *published;
     uint64_t published_changes, publications;
+    struct PureSurfaceHost *surface; /* told of each snapshot published */
     /* The owner's: outputs wired to HTTP receivers, and how far each was trimmed. */
     PureWire *wires[PURE_MODULE_INPUTS];
     unsigned wire_count;
@@ -185,14 +192,15 @@ typedef struct {
     bool held[PURE_MODULE_INPUTS]; /* an answer failed to apply: the wire's sends stay until a restart */
 } Executor;
 
-/* A request made ready: its effect, what it names, and the thread that answers it. */
-typedef struct { Host *host; PureEffect effect; Route route; } Job;
+/* A request made ready: its effect, what it names, and the thread that
+ * answers it (or, for a surface's event, the waiter it wakes). */
+typedef struct { Host *host; PureEffect effect; Route route; Waiter *waiter; } Job;
 
 /* The owner of an app with states served by several connection threads:
  * their requests wait here, and it runs them in bursts, one save each
  * however many threads they came from, handing each thread back its
- * applied requests to answer. */
-typedef struct {
+ * applied requests to answer (and a surface its events). */
+typedef struct Owner {
     Executor executor;
     pthread_mutex_t lock;
     pthread_cond_t waiting;
@@ -1144,7 +1152,8 @@ 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;
-    *out = (Applied){.host = job->host, .ticket = effect->ticket, .input = effect->input, .route = job->route.kind};
+    *out = (Applied){.host = job->host, .waiter = job->waiter, .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) {
         out->status = show_view(app, &job->route, effect->available, out);
@@ -1185,6 +1194,24 @@ static void apply_effect(Executor *ex, const Job *job, Applied *out) {
                                    .release = release_copy};
 }
 
+/* ---- A surface in the node (serve/surface.h): one, run in its own thread,
+ * reading published snapshots and giving events to the owner. */
+struct PureSurfaceHost {
+    struct Owner *owner;
+    pthread_mutex_t lock;
+    pthread_cond_t applied; /* a waiter's event was applied and saved (or not) */
+    PureSurface table;
+    PureSurfaceRun run;
+    int argc, result;
+    char **argv;
+    pthread_t thread;
+};
+
+static void surface_changed(PureSurfaceHost *s) {
+    uint64_t one = 1;
+    if (write(s->table.changes, &one, sizeof(one)) != sizeof(one)) { /* full: it is readable already */ }
+}
+
 /* After a save, before its answers go: from now on views show what it
  * saved, so a read made after a write's answer sees the write. A failed save
  * returns the app to the last save's states, which the snapshot still shows. */
@@ -1198,6 +1225,7 @@ static void publish(Executor *ex) {
     pure_snapshot_release(old);
     ex->published_changes = ex->app.changes;
     ++ex->publications;
+    if (ex->surface) surface_changed(ex->surface);
 }
 
 /* The saved sends of wired outputs not queued yet go to their wires, in
@@ -1378,6 +1406,16 @@ static void answer_burst(void *context, Applied *applied, size_t count) {
 /* 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;
+    PureSurfaceHost *surface = owner->executor.surface;
+    bool events = false;
+    for (size_t i = 0; i < count; ++i) events = events || !applied[i].host;
+    if (events) { /* a surface's: its waiter wakes */
+        pthread_mutex_lock(&surface->lock);
+        for (size_t i = 0; i < count; ++i)
+            if (!applied[i].host) applied[i].waiter->status = applied[i].status, applied[i].waiter->done = true;
+        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;
@@ -1432,7 +1470,7 @@ static void serve_bursts(Host *host) {
         size_t count = 0;
         PureEffect effect;
         while (pure_runtime_take(&host->runtime, &effect) == PURE_TAKE_EFFECT)
-            host->jobs[count++] = (Job){host, effect, host->routes[effect.ticket.slot]};
+            host->jobs[count++] = (Job){host, effect, host->routes[effect.ticket.slot], NULL};
         if (!count) return;
         if (!host->owner) {
             execute(&host->local, host->jobs, count, answer_burst, NULL);
@@ -1560,6 +1598,99 @@ static void *run_owner(void *context) {
     return NULL;
 }
 
+/* The surface's table (serve/surface.h). */
+static PureSnapshot *surface_read(PureSurfaceHost *s, PureApp *reader) {
+    Executor *ex = &s->owner->executor;
+    pthread_mutex_lock(&ex->published_lock);
+    PureSnapshot *snapshot = ex->published;
+    if (snapshot) pure_snapshot_hold(snapshot);
+    pthread_mutex_unlock(&ex->published_lock);
+    if (snapshot) pure_app_read(reader, snapshot);
+    return snapshot;
+}
+
+static void surface_release(PureSurfaceHost *s, PureSnapshot *snapshot) {
+    (void)s;
+    pure_snapshot_release(snapshot);
+}
+
+/* An event queued at the owner as a request is, waited for until its save. */
+static PureResult surface_event(PureSurfaceHost *s, uint32_t input, PureSpan text) {
+    Owner *owner = s->owner;
+    const PureModule *m = owner->executor.module;
+    if (input >= m->inputs || (m->input[input].kind != PURE_KIND_KEY && m->input[input].kind != PURE_KIND_TEXT) ||
+        (text.size && !text.data))
+        return PURE_INVALID;
+    if (text.size > PURE_OBJECT_LIMIT) return PURE_LIMIT;
+    unsigned char *value = malloc(4 + text.size); /* as a request's ingress: its size, then its bytes */
+    if (!value) return PURE_ALLOCATION;
+    put_u32(value, (uint32_t)text.size);
+    if (text.size) memcpy(value + 4, text.data, text.size);
+    Waiter waiter = {PURE_INVALID, false};
+    Job job = {.route = {.kind = ROUTE_INPUT, .target = input}, .waiter = &waiter,
+               .effect = {.input = PURE_INPUT_WRITE, .key = "", .expected = -1, .value = {value, 4 + text.size}}};
+    pthread_mutex_lock(&owner->lock);
+    bool queued = !owner->stopping && grow(&owner->jobs, &owner->capacity, owner->count + 1, sizeof(*owner->jobs));
+    if (queued) {
+        owner->jobs[owner->count++] = job;
+        ++owner->arrivals;
+        pthread_cond_signal(&owner->waiting);
+    }
+    pthread_mutex_unlock(&owner->lock);
+    pthread_mutex_lock(&s->lock);
+    while (queued && !waiter.done) pthread_cond_wait(&s->applied, &s->lock);
+    pthread_mutex_unlock(&s->lock);
+    free(value);
+    return queued ? waiter.status : PURE_INVALID;
+}
+
+static PureResult surface_press(PureSurfaceHost *s, uint32_t input) {
+    return input < s->owner->executor.module->inputs &&
+                   s->owner->executor.module->input[input].kind == PURE_KIND_KEY
+               ? surface_event(s, input, (PureSpan){0})
+               : PURE_INVALID;
+}
+
+static PureResult surface_type(PureSurfaceHost *s, uint32_t input, PureSpan text) {
+    return input < s->owner->executor.module->inputs &&
+                   s->owner->executor.module->input[input].kind == PURE_KIND_TEXT
+               ? surface_event(s, input, text)
+               : PURE_INVALID;
+}
+
+static bool surface_stopping(PureSurfaceHost *s) {
+    (void)s;
+    return atomic_load(&stopping);
+}
+
+static void *run_surface(void *context) {
+    PureSurfaceHost *s = context;
+    s->result = s->run(&s->table, s->argc, s->argv);
+    fprintf(stderr, "{\"surface\":\"%s\",\"returned\":%d}\n", s->argv[0], s->result);
+    return NULL;
+}
+
+/* --surface LIBRARY [-- ARGUMENTS]: mapped now, run in its own thread. */
+static bool start_surface(PureSurfaceHost *s, Owner *owner, int argc, char **argv) {
+    void *library = dlopen(argv[0], RTLD_NOW | RTLD_LOCAL);
+    PureSurfaceRun run = library ? (PureSurfaceRun)(uintptr_t)dlsym(library, "pure_surface_run") : NULL;
+    if (!run) {
+        fprintf(stderr, "--surface %s: %s\n", argv[0], library ? "no pure_surface_run in it" : dlerror());
+        return false;
+    }
+    *s = (PureSurfaceHost){.owner = owner, .run = run, .argc = argc, .argv = argv};
+    s->table = (PureSurface){PURE_SURFACE_VERSION, sizeof(PureApp), sizeof(PureModule), owner->executor.module, s,
+                             eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC), surface_read, surface_release, surface_press,
+                             surface_type, surface_stopping};
+    if (s->table.changes < 0 || pthread_mutex_init(&s->lock, NULL) || pthread_cond_init(&s->applied, NULL)) return false;
+    owner->executor.surface = s; /* before the thread: its first snapshot may come any time */
+    if (pthread_create(&s->thread, NULL, run_surface, s)) {
+        owner->executor.surface = NULL;
+        return false;
+    }
+    return true;
+}
+
 /* A write's reply is borrowed from its request, so the handler of an input
  * served by PUT or POST may reply only a field of its message. */
 static bool replies_borrow(const PureModule *m, uint32_t input) {
@@ -1909,17 +2040,27 @@ static unsigned auto_threads(bool owner) {
 int main(int argc, char **argv) {
     static PureModule module;
     static Owner owner;
-    const char *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS], *replies[PURE_MODULE_INPUTS];
+    static PureSurfaceHost surface;
+    /* What follows -- is the surface's: its library first, then its arguments. */
+    int options = argc, surface_argc = 0;
+    char *surface_argv[64];
+    for (int i = 1; i < argc; ++i)
+        if (!strcmp(argv[i], "--")) {
+            options = i;
+            break;
+        }
+    const char *surface_path = NULL, *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS], *replies[PURE_MODULE_INPUTS];
     const char *seals[PURE_MODULE_INPUTS], *named[PURE_MODULE_INPUTS], *key_path = NULL;
     unsigned send_count = 0, reply_count = 0, seal_count = 0, named_count = 0;
     unsigned port = 8767, threads = 1;
     size_t budget = 64u << 20, capacity = 1024;
     uint64_t timeout_ms = 10000;
-    for (int i = 1; i < argc; i += 2) {
+    for (int i = 1; i < options; i += 2) {
         uint64_t value;
         if (!strcmp(argv[i], "--sealed-only")) { sealed_only = true; --i; continue; }
-        if (i + 1 == argc) goto usage;
+        if (i + 1 == options) goto usage;
         if (!strcmp(argv[i], "--module")) module_path = argv[i + 1];
+        else if (!strcmp(argv[i], "--surface")) surface_path = argv[i + 1];
         else if (!strcmp(argv[i], "--key")) key_path = argv[i + 1];
         else if (!strcmp(argv[i], "--seal") && seal_count < PURE_MODULE_INPUTS) seals[seal_count++] = argv[i + 1];
         else if (!strcmp(argv[i], "--store")) store = argv[i + 1];
@@ -1936,6 +2077,10 @@ int main(int argc, char **argv) {
         else if (!strcmp(argv[i], "--timeout-ms") && number(argv[i + 1], 3600000, &value) && value) timeout_ms = value;
         else goto usage;
     }
+    if (surface_path) surface_argv[surface_argc++] = (char *)surface_path;
+    for (int i = options + 1; i < argc && surface_argc < 63; ++i) surface_argv[surface_argc++] = argv[i];
+    surface_argv[surface_argc] = NULL;
+    if (options < argc && !surface_path) goto usage;
     uint32_t show, edit;
     bool documents;
     if (!module_path || !store || !load_module(&module, module_path, &documents, &show, &edit)) goto usage;
@@ -1998,10 +2143,17 @@ int main(int argc, char **argv) {
     if (*public_key) printf("{\"port\":%u,\"threads\":%u,\"public_key\":\"%s\"}\n", port, threads, public_key);
     else printf("{\"port\":%u,\"threads\":%u}\n", port, threads);
     fflush(stdout);
+    /* A surface reads the owner's snapshots and gives it events. */
+    if (surface_path && !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;
     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);
+    if (surface_path) { /* stopped before the owner, which may still apply its last event */
+        surface_changed(&surface);
+        pthread_join(surface.thread, NULL);
+    }
     if (owned) {
         pthread_mutex_lock(&owner.lock);
         owner.stopping = true;
@@ -2052,6 +2204,6 @@ usage:
     fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] "
           "[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=http[s]://host:port/path ...] [--name NAME] "
           "[--send-ca FILE] [--reply OUTPUT=INPUT ...] [--seal OUTPUT=KEY ...] [--key FILE] [--sealed-only] "
-          "[--sender NAME=KEY ...]\n", stderr);
+          "[--sender NAME=KEY ...] [--surface LIBRARY [-- ARGUMENTS]]\n", stderr);
     return 2;
 }
diff --git a/serve/surface.h b/serve/surface.h
new file mode 100644
index 0000000..a79bf05
--- /dev/null
+++ b/serve/surface.h
@@ -0,0 +1,48 @@
+#ifndef PURE_SURFACE_H
+#define PURE_SURFACE_H
+#include "module.h"
+
+/* A surface shows an app to people (Vega's display, say): a library that
+ * pure-serve maps and runs in a thread of its own beside its connection
+ * threads (--surface LIBRARY [-- ARGUMENTS]). It reads the app as of the
+ * last save, as a connection thread reads views, and gives it events that
+ * the owner applies with the requests, one save a round. So a node is one
+ * process with one owner of its states, whether people or programs use it,
+ * and a surface reads states where they are, with no HTTP or JSON between.
+ *
+ * The library exports
+ *
+ *     int pure_surface_run(const PureSurface *, int argc, char **argv);
+ *
+ * and calls the app model (core/module.h) as pure-serve exports it: one
+ * copy of that code in the process, the host's. argv[0] is the library.
+ * pure_surface_run returns when stopping says so (the changes descriptor
+ * becomes readable then too), or when it is done; the server serves on. */
+enum { PURE_SURFACE_VERSION = 1 };
+
+typedef struct PureSurfaceHost PureSurfaceHost;
+typedef struct {
+    unsigned version;
+    size_t app_size, module_size; /* sizeof(PureApp), sizeof(PureModule): a surface built for another layout refuses */
+    const PureModule *module;
+    PureSurfaceHost *host;
+    /* An eventfd, readable after each save that changed what views show and
+     * when the server stops: poll it with the surface's own descriptors, and
+     * read its 8 bytes to clear it. */
+    int changes;
+    /* The app as of the last save, shown in reader (a zeroed app the first
+     * time, the same one after): its views, rows and texts read as
+     * core/module.h says while the snapshot is held. Give it back with
+     * release. NULL before the first save. */
+    PureSnapshot *(*read)(PureSurfaceHost *, PureApp *reader);
+    void (*release)(PureSurfaceHost *, PureSnapshot *);
+    /* An event, applied by the owner with the next save: a key input pressed
+     * and released, or text typed into a text input. Returns once it was
+     * applied and saved (PURE_OK), or why it was not. */
+    PureResult (*press)(PureSurfaceHost *, uint32_t input);
+    PureResult (*type)(PureSurfaceHost *, uint32_t input, PureSpan text);
+    bool (*stopping)(PureSurfaceHost *);
+} PureSurface;
+
+typedef int (*PureSurfaceRun)(const PureSurface *, int argc, char **argv);
+#endif