diff options
| -rw-r--r-- | Makefile | 3 | ||||
| -rw-r--r-- | README.md | 2 | ||||
| -rw-r--r-- | docs/design/009-layout.md | 2 | ||||
| -rw-r--r-- | docs/design/010-surfaces.md | 62 | ||||
| -rw-r--r-- | serve/serve.c | 174 | ||||
| -rw-r--r-- | serve/surface.h | 48 | ||||
| -rw-r--r-- | tests/check.py | 87 | ||||
| -rw-r--r-- | tests/surface-probe.c | 85 |
8 files changed, 449 insertions, 14 deletions
@@ -38,9 +38,10 @@ $(BUILD)/libpure.a: $(OBJECTS) # The runtime's tickets and budgets, with effects completed off the owner thread. # It links after the library, last as it always was, so code keeps its place. +# The app model's functions are exported for a surface it maps (serve/surface.h). $(BUILD)/pure-serve: $(SERVE) core/runtime.c $(BUILD)/libpure.a $(HEADERS) $(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -Iserve -DPURE_ASYNC_EFFECTS $(SERVE) $(BUILD)/libpure.a core/runtime.c \ - $(LDFLAGS) -lm -pthread $(LDLIBS) -o $@ + $(LDFLAGS) -Wl,--export-dynamic-symbol='pure_*' -lm -pthread $(LDLIBS) -o $@ $(BUILD)/pure-crypto.so: serve/crypto-openssl.c $(HEADERS) $(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -Iserve -fPIC -shared serve/crypto-openssl.c $(LDFLAGS) -lssl -lcrypto \ @@ -23,7 +23,7 @@ The source follows the principle below, a small core with the rest selected arou |---|---|---| | `core/` | execution, ownership, limits and completion (`runtime.c`); the app model: modules verified before they run, events as transactions, states as drafts (`module*.c`); what the core asks of storage (`state.h`, `store.h`) | `build/libpure.a` (with the store) | | `store/` | the tree store: one file, a commit is one contiguous write and one sync (formats `PURETRE3`, `PUREREC1`) | in `libpure.a` | -| `serve/` | `pure-serve`: the HTTP host, outbound delivery, JSON read in place; OpenSSL's part (TLS to https receivers, sealed sends) apart, mapped only by a server that uses one | `build/pure-serve`, `build/pure-crypto.so` | +| `serve/` | `pure-serve`: the HTTP host, outbound delivery, JSON read in place, surfaces for people run in the node ([010](docs/design/010-surfaces.md)); OpenSSL's part (TLS to https receivers, sealed sends) apart, mapped only by a server that uses one | `build/pure-serve`, `build/pure-crypto.so` | | `cli/` | command-line hosts | `build/pure-run`, `build/pure-edit-module` | | `prototypes/` | the earlier document runtime (000-077), built against `core/` and `store/` | `make -C prototypes/connections` | | `tests/` | the hosts checked end to end, with apps built by aup (the sibling checkout) | `make check` | diff --git a/docs/design/009-layout.md b/docs/design/009-layout.md index d985003..de5563d 100644 --- a/docs/design/009-layout.md +++ b/docs/design/009-layout.md @@ -29,7 +29,7 @@ |---|---|---| | `core/` | 실행·소유권·한도·완료(`runtime.c`, `flow.c`), 앱 모델(`module*.c`: 실행 전에 검증한 모듈, 트랜잭션인 이벤트, draft인 상태), 코어가 저장소에 요구하는 것(`state.h`, `store.h`) | `build/libpure.a`(저장소와 함께) | | `store/` | 트리 저장소. 한 파일, 커밋은 한 연속 쓰기와 동기화 한 번 | `libpure.a` 안 | -| `serve/` | `pure-serve`: HTTP 호스트, 나가는 전달, 제자리 JSON. OpenSSL을 쓰는 부분(https 수신자로의 TLS, 봉인한 전달)은 따로 두어 쓰는 서버만 매핑한다 | `build/pure-serve`, `build/pure-crypto.so` | +| `serve/` | `pure-serve`: HTTP 호스트, 나가는 전달, 제자리 JSON, 노드 안에서 도는 사람을 위한 표면([010](010-surfaces.md)). OpenSSL을 쓰는 부분(https 수신자로의 TLS, 봉인한 전달)은 따로 두어 쓰는 서버만 매핑한다 | `build/pure-serve`, `build/pure-crypto.so` | | `cli/` | 명령행 호스트 | `build/pure-run`, `build/pure-edit-module` | | `prototypes/` | 이전 문서 런타임. `core/`와 `store/`를 가져다 빌드한다 | `make -C prototypes/connections` | diff --git a/docs/design/010-surfaces.md b/docs/design/010-surfaces.md new file mode 100644 index 0000000..1c5100f --- /dev/null +++ b/docs/design/010-surfaces.md @@ -0,0 +1,62 @@ +# 010 — 표면: 노드 안에서 사람에게 보이는 앱 + +2026-09-27. 메신저를 만들려면 한 기기에서 앱이 네트워크(봉인한 전달, 088)와 화면(Vega)을 함께 가져야 한다. 그 둘을 어떻게 +잇는지 정한 것. + +## 문제 + +- pure-serve는 앱의 상태를 가진 소유자 스레드 하나와 연결 스레드들로 된 노드다. 보기는 GET, 입력은 POST이고, 전달선은 이 + 노드의 출력을 다른 노드로 봉인해 보낸다. 사람을 위한 표면은 없다. +- Vega의 표면(터미널, 픽셀, KMS 디스플레이)은 앱을 제 프로세스 안에서 돌린다(`PureApp`을 직접). 네트워크가 없다. +- 두 프로세스가 같은 저장소를 쓰면 안 된다. 저장소는 쓰는 이가 하나다. + +길은 둘이었다. + +1. **원격 표면.** Vega가 노드의 HTTP API를 쓰는 클라이언트가 된다. 화면마다 보기를 GET으로 받고(JSON), 키와 글자를 + POST하고, 바뀐 것은 긴 폴링으로 안다. 화면이 다른 기기에 있어도 된다. 그러나 상태와 화면 사이에 HTTP와 JSON이 들어서고, + Vega의 모든 읽기를 원격 판으로 한 번 더 써야 한다. +2. **노드 안의 표면.** pure-serve가 표면 라이브러리를 매핑해 제 스레드에서 돌린다. 표면은 연결 스레드가 보기를 읽는 것과 + 같이 마지막 저장의 스냅숏(080)을 제자리에서 읽고, 이벤트는 요청과 같은 줄로 소유자에게 준다. + +로드맵은 Vega가 "뷰 상태 변경 바이트를 화면과 입출력 버퍼로 직통 바인딩"하기를 바란다. 2가 그것이다. 상태의 주인이 +하나이고, 사람과 프로그램이 같은 노드를 쓰며, 새 호스트가 생기지 않는다. JVM이 JavaFX를 싣는 것과 같은 관계다. 런타임이 +라이브러리를 싣고, 런타임은 화면을 모른다(009의 선). 1이 필요해지면(다른 기기의 화면) 그때 HTTP 쪽에 더한다. + +## 계약 + +`serve/surface.h`. 표면 라이브러리는 `pure_surface_run(const PureSurface *, argc, argv)`을 내보낸다. pure-serve는 +`--surface LIBRARY [-- ARGUMENTS]`로 그것을 매핑하고, 소유자를 띄운 뒤 제 스레드에서 부른다. 표가 주는 것은 이렇다. + +- `read(host, reader)`: 마지막 저장의 스냅숏을 붙잡아 표면의 리더 앱에 보인다. 보기, 행, 텍스트는 `core/module.h`의 함수 + 그대로 읽는다. `release`로 돌려준다. +- `changes`: eventfd. 보기가 바뀌는 저장(스냅숏 게시)마다, 그리고 서버가 멈출 때 읽을 수 있게 된다. 표면은 제 입력 + 장치와 함께 poll한다. 새 저장을 기다리며 도는 일이 없다. +- `press(host, input)`, `type(host, input, text)`: 키 입력을 눌렀다 떼고, 텍스트 입력에 글자를 넣는다. 소유자가 요청과 + 같은 버스트에서 적용하고 저장한 뒤에 돌아온다. 다른 종류의 입력이나 없는 입력은 `PURE_INVALID`다. +- `stopping(host)`: 멈출 때다. pure-serve는 연결 스레드가 멈춘 뒤, 소유자를 멈추기 전에 표면을 기다린다. 표면이 기다리는 + 마지막 이벤트를 소유자가 아직 적용할 수 있다. + +앱 모델의 코드는 프로세스에 한 벌이다. pure-serve가 제 `pure_*` 함수를 동적 심볼로 내보내고(`--export-dynamic-symbol`), +표면은 그것을 부른다. 표면이 제 사본을 싣지 않으므로 판이 어긋날 일이 없다. 표에는 `sizeof(PureApp)`, +`sizeof(PureModule)`도 있어서, 다른 배치로 빌드된 표면은 스스로 거절한다. + +표면의 이벤트는 호스트가 없는 요청(`Job.host == NULL`)이다. 소유자의 버스트 규칙(한 라운드에 저장 한 번, 문서 키는 +버스트에 한 번), 게시, 전달선이 모두 요청과 같다. 결과는 `hand_back`이 표면의 기다리는 이에게 준다. 상태가 없는 모듈에는 +소유자가 없으므로 표면을 둘 수 없다. + +## 비용 + +표면을 쓰지 않는 서버에는 게시할 때 분기 하나, `Job`과 `Applied`에 포인터 하나다. 전과 번갈아 쟀다. 64 B GET은 140.9k 대 +139.9k/s, 앱 합계 GET은 여덟 라운드 중앙값 126.3k 대 126.3k, 20행 페이지는 96.1k 대 95.3k로 같다. 내보낸 심볼 157개로 +실행 파일의 읽기 전용 부분이 11 KB, 유휴 RSS가 24 KiB 늘었다. + +## 검사 + +`tests/check.py`의 `server_surface`는 검사용 표면(`tests/surface-probe.c`)을 빌드해 싣는다. 그 표면은 텍스트 세 줄을 +넣고 키를 누르며 저장마다 보기를 적는다. 검사는 이것들을 본다. + +- 표면의 이벤트가 HTTP의 보기에 보인다. +- HTTP로 넣은 입력이 `changes`로 표면에 보인다. +- 다른 종류의 입력과 없는 입력은 거절된다(8, `PURE_INVALID`). +- 표면은 서버와 함께 0으로 끝난다. +- 상태가 없는 모듈은 표면을 거절한다(종료 2). 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 diff --git a/tests/check.py b/tests/check.py index 76a23e3..9ff771a 100644 --- a/tests/check.py +++ b/tests/check.py @@ -937,6 +937,92 @@ def server_sealed(aup, pure, work): relay.shutdown() +SURFACE_SOURCE = """record Task { + title: text + done = 0 +} +state tasks: list of Task +state at: index of tasks +input typed = text +input toggle = key Space +on typed { + tasks.append(Task { title: typed }) + at = tasks.last +} +on toggle.down { at.done = !at.done } +view count = tasks.count +view current = if tasks.count > 0 { at.title } else { "(empty)" } +view finished = at.done +""" + + +def server_surface(aup, pure, work): + """A surface in the node (--surface, serve/surface.h): a library that + pure-serve maps and runs beside its connection threads types text and + presses a key as a person at the node's display would. The owner applies + its events with the requests, one save a round, and it reads each save's + views where they are: its events show in HTTP views, and an HTTP input + shows in it through the changes descriptor. An event for an input of + another kind, or for none, is refused; it stops with the server; and a + module without states cannot have one.""" + import select + import signal + import time + probe = work / 'surface-probe.so' + subprocess.run(['cc', '-O2', '-std=c17', '-Wall', '-Wextra', '-Werror', '-shared', '-fPIC', f'-I{pure / "core"}', + f'-I{pure / "serve"}', str(pure / 'tests' / 'surface-probe.c'), '-o', str(probe)], check=True) + source, module = work / 'surface.aup', work / 'surface.pmod' + source.write_text(SURFACE_SOURCE) + compile_module(aup, source, module) + out = work / 'surface.jsonl' + process = subprocess.Popen([str(a) for a in (pure / 'build/pure-serve', '--module', module, '--store', + work / 'surface-store', '--port', '0', '--surface', probe, '--', out, + 'typed', 'toggle', 'one,two,three', 'count', 'current', 'finished')], + stdout=subprocess.PIPE, stderr=subprocess.PIPE) + + def lines(): + return [json.loads(line) for line in out.read_text().splitlines()] if out.exists() else [] + + def wait_for(condition): + deadline = time.time() + 20 + while not condition() and time.time() < deadline: + time.sleep(0.05) + return condition() + + try: + ready, _, _ = select.select([process.stdout], [], [], 10) + assert ready, 'host did not start' + port = json.loads(process.stdout.readline())['port'] + assert wait_for(lambda: any(line['event'].startswith('wrong kind') for line in lines())), lines() + assert http(port, 'GET', '/views/count')[3] == b'3' and http(port, 'GET', '/views/current')[3] == b'three' + assert http(port, 'POST', '/inputs/typed', b'four')[0] == 200 + assert wait_for(lambda: any(line['event'] == 'changed' and line['views']['count'] == 4 for line in lines())) + finally: + process.send_signal(signal.SIGTERM) + _, error = process.communicate(timeout=30) + assert process.returncode == 0 and b'"returned":0' in error, error[-500:] + got = lines() + views = [line['views'] for line in got] + assert [line['event'] for line in got[:6]] == ['start', 'typed one: 0', 'typed two: 0', 'typed three: 0', 'pressed: 0', + 'wrong kind: 8, no input: 8'], got[:6] + assert views[:6] == [{'count': 0, 'current': '(empty)', 'finished': None}, + {'count': 1, 'current': 'one', 'finished': 0}, {'count': 2, 'current': 'two', 'finished': 0}, + {'count': 3, 'current': 'three', 'finished': 0}, {'count': 3, 'current': 'three', 'finished': 1}, + {'count': 3, 'current': 'three', 'finished': 1}], views[:6] + assert views[-1] == {'count': 4, 'current': 'four', 'finished': 0}, views[-1] + changes = [line['changes'] for line in got] + assert changes[1:5] == sorted(set(changes[1:5])) and changes[-1] > changes[4], changes + # A module without states has no owner to apply a surface's events. + stateless = work / 'surface-echo.aup' + stateless.write_text('input shout = text\non shout { }\n') + compile_module(aup, stateless, stateless.with_suffix('.pmod')) + refused = subprocess.run([str(a) for a in (pure / 'build/pure-serve', '--module', stateless.with_suffix('.pmod'), + '--store', work / 'surface-echo', '--port', '0', '--surface', probe, '--', + out, 'shout', 'shout', 'x')], capture_output=True, timeout=30) + assert refused.returncode == 2 and b'needs an app with states' in refused.stderr, refused.stderr + return {'events': 5, 'lines': len(got)} + + def server_webhook(aup, pure, work): """An output wired to a third-party HTTPS endpoint (a Python server with a certificate of its own): five orders arrive in order as JSON with the @@ -1396,6 +1482,7 @@ def main(): report['server_links'] = server_links(aup, pure, work) report['server_webhook'] = server_webhook(aup, pure, work) report['server_sealed'] = server_sealed(aup, pure, work) + report['server_surface'] = server_surface(aup, pure, work) report['server_answers'] = server_answers(aup, pure, work) print(json.dumps(report)) diff --git a/tests/surface-probe.c b/tests/surface-probe.c new file mode 100644 index 0000000..51feb95 --- /dev/null +++ b/tests/surface-probe.c @@ -0,0 +1,85 @@ +#define _GNU_SOURCE +/* A surface for tests/check.py (serve/surface.h): as a person at the node's + * display would, it types each line of TEXTS into the text input and then + * presses the key input; after each it writes what the views show, and then, + * until the server stops, what they show after each save the node made for + * others (HTTP requests). One JSON line each into OUTPUT. + * + * pure-serve ... --surface surface-probe.so -- OUTPUT TEXT_INPUT KEY_INPUT TEXTS VIEW... */ +#include "surface.h" +#include <poll.h> +#include <stdio.h> +#include <string.h> +#include <unistd.h> + +static int named(const char (*names)[PURE_MODULE_NAME], unsigned count, const char *name) { + for (unsigned i = 0; i < count; ++i) + if (!strcmp(names[i], name)) return (int)i; + return -1; +} + +static bool put_text(void *context, PureSpan span) { + fwrite(span.data, 1, span.size, context); + return true; +} + +/* {"event":...,"changes":N,"views":{"name":value,...}} as the last save shows them. */ +static void show(const PureSurface *s, PureApp *reader, FILE *out, const char *event, char **views, int count) { + PureSnapshot *snapshot = s->read(s->host, reader); + fprintf(out, "{\"event\":\"%s\",\"changes\":%llu,\"views\":{", event, + snapshot ? (unsigned long long)reader->changes : 0ull); + for (int i = 0; i < count; ++i) { + int view = named(s->module->view_names, s->module->views, views[i]); + fprintf(out, "%s\"%s\":", i ? "," : "", views[i]); + double value; + if (!snapshot || view < 0) fputs("null", out); + else if (s->module->view_type[view] == PURE_VIEW_TEXT) { + fputc('"', out); + pure_app_view_text(reader, (uint32_t)view, put_text, out); + fputc('"', out); + } else if (pure_app_view(reader, (uint32_t)view, &value)) fprintf(out, "%g", value); + else fputs("null", out); + } + fputs("}}\n", out); + fflush(out); + if (snapshot) s->release(s->host, snapshot); +} + +int pure_surface_run(const PureSurface *s, int argc, char **argv) { + if (s->version != PURE_SURFACE_VERSION || s->app_size != sizeof(PureApp) || s->module_size != sizeof(PureModule)) + return 2; + if (argc < 5) return 2; + FILE *out = fopen(argv[1], "w"); + int text = named(s->module->input_names, s->module->inputs, argv[2]); + int key = named(s->module->input_names, s->module->inputs, argv[3]); + if (!out || text < 0 || key < 0) return 2; + static PureApp reader; /* zeroed the first time, as read wants */ + char **views = argv + 5; + int view_count = argc - 5; + show(s, &reader, out, "start", views, view_count); + char texts[256]; + snprintf(texts, sizeof(texts), "%s", argv[4]); + for (char *line = strtok(texts, ","); line; line = strtok(NULL, ",")) { + PureResult typed = s->type(s->host, (uint32_t)text, (PureSpan){(const unsigned char *)line, strlen(line)}); + char event[300]; + snprintf(event, sizeof(event), "typed %s: %d", line, (int)typed); + show(s, &reader, out, event, views, view_count); + } + char event[64]; + snprintf(event, sizeof(event), "pressed: %d", (int)s->press(s->host, (uint32_t)key)); + show(s, &reader, out, event, views, view_count); + /* Refusals: an input of another kind, and one past the module's. */ + snprintf(event, sizeof(event), "wrong kind: %d, no input: %d", (int)s->press(s->host, (uint32_t)text), + (int)s->type(s->host, s->module->inputs, (PureSpan){0})); + show(s, &reader, out, event, views, view_count); + while (!s->stopping(s->host)) { + struct pollfd wait = {s->changes, POLLIN, 0}; + if (poll(&wait, 1, 1000) <= 0) continue; + uint64_t times; + if (read(s->changes, ×, sizeof(times)) != sizeof(times) || s->stopping(s->host)) continue; + show(s, &reader, out, "changed", views, view_count); + } + pure_app_stop(&reader); + fclose(out); + return 0; +} |