diff options
| -rw-r--r-- | docs/design/008-module-format.md | 21 | ||||
| -rw-r--r-- | prototypes/connections/Makefile | 6 | ||||
| -rw-r--r-- | prototypes/connections/server.c | 211 |
3 files changed, 230 insertions, 8 deletions
diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md index 4c5a9ac..2d30003 100644 --- a/docs/design/008-module-format.md +++ b/docs/design/008-module-format.md @@ -197,6 +197,25 @@ text·리스트가 남을 때만 위치를 유지한다. 실패하면 묶음 전체가 되돌려지고, 저장 중 프로세스가 죽으면 묶음 전체가 남거나 전혀 남지 않는다. 비용(Pi 5, SD의 ext4): 저장하는 이벤트 하나에 약 10.6 ms가 더 든다. 대부분 두 번의 `fsync`다. 300항목 리스트에서 인덱스 이동 한 번은 파일을 약 220 B 키운다(압축 전). -아직 런타임 코어(`runtime.c`)의 작업·예산 모델과 연결하지 않았다. 호스트가 직접 호출한다. 실행기 코드는 +## 런타임 코어 위의 HTTP 서버 + +`pure-connections-module --module 앱.pmod --store 경로`는 기존 문서 서버(`server.c`)와 같은 HTTP 계약을 모듈로 +제공한다. `GET /objects/키`는 모듈의 `show` 메시지, `PUT`(`If-Match`/`If-None-Match`)은 `edit` 메시지다. 요청은 +런타임 코어(`runtime.c`)를 그대로 거친다. 작업 슬롯 8개, 입력 바이트 예산, 코어가 소유한 ingress, 입력에서 빌린 +PUT 응답이 모두 기존과 같다. 효과는 비동기 인터페이스(`pure_runtime_take`/`pure_runtime_complete`)로 소유 스레드에서 +처리한다. + +- **poll 한 바퀴가 묶음 하나다.** 준비된 요청을 모두 꺼내 지연 저장 모드의 앱에 차례로 적용하고, `pure_app_save` + 한 번으로 저장한 **뒤에** 티켓을 완료하고 응답한다. 저장이 실패하면 그 묶음의 읽기까지 모두 실패한다. 저장되지 + 않은 상태를 본 응답은 나가지 않는다. 같은 키가 다시 오면 묶음을 먼저 끝낸다. 모듈 서버는 기다리는 연결을 한 번에 + 모두 받아 동시 요청이 같은 묶음에 들어가게 한다. +- GET 응답은 코어 예산에 계상되는 자원으로 복사한다. PUT 응답은 코어 계약대로 요청 본문에서 빌려야 하므로 `edit` + 핸들러는 메시지 필드만 응답할 수 있다(`reply edit.insert`). 시작할 때 확인한다. +- 검증(aup `make check`): `editor.aup`를 실행하는 모듈 서버와 `editor.pure`를 실행하는 기존 서버가 무작위 요청 + (새 키, 오래된 리비전, 범위 밖, 짧은 본문, 5,000 B 삽입)에 상태 코드·ETag·오류·본문까지 같게 답한다. 동시 편집 중 + `kill -9`로 죽여도 확인 응답을 받은 편집은 다시 시작한 서버에 모두 있다. Pi 5 SD(ext4)에서 동시 클라이언트 8개의 + PUT 240건: 기존 서버 2.14 s, 모듈 서버 0.59 s(묶음 61개). + +아직 모듈 앱을 런타임 코어의 효과 단위로 세분하지는 않았다(`runtime.c`의 `definition`은 앱 전체다). 호스트가 직접 호출한다. 실행기 코드는 `module.c` 33.1 KB, `module-text.c` 3.6 KB, `module-list.c` 5.7 KB, `module-save.c` 5.3 KB(`-O2` text)로 v1(7.4 KB)보다 크다. 줄이는 것이 남은 일이다. 항목 text를 읽을 때마다 복사하는 비용도 남아 있다. diff --git a/prototypes/connections/Makefile b/prototypes/connections/Makefile index e78fdbd..83af085 100644 --- a/prototypes/connections/Makefile +++ b/prototypes/connections/Makefile @@ -40,6 +40,10 @@ $(BUILD)/pure-connections-async: server.c $(HOST) flow.c program.c $(CORE) $(HEA $(BUILD)/pure-connections-async-gated: server.c $(HOST) flow.c program.c $(CORE) test-async-server-gate.c $(HEADERS) | $(BUILD) $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_RUNTIME_METRICS -DPURE_ASYNC_EFFECTS -DPURE_HTTP_ASYNC -DPURE_STORE_TEST server.c $(HOST) flow.c program.c $(CORE) test-async-server-gate.c $(LDFLAGS) -pthread $(LDLIBS) -o $@ +# The same HTTP host running an aup module (show and edit) with group commits. +$(BUILD)/pure-connections-module: server.c flow.c module-store.c $(MODULE) $(CORE) module-store.h $(MODULE_HEADERS) | $(BUILD) + $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_RUNTIME_METRICS -DPURE_ASYNC_EFFECTS -DPURE_MODULE_SERVER server.c flow.c module-store.c $(MODULE) $(CORE) $(LDFLAGS) -lm $(LDLIBS) -o $@ + $(BUILD)/pure-connections-fixed: server.c $(HOST) flow.c $(CORE) $(HEADERS) $(BUILD)/embedded-plan.h | $(BUILD) $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_RUNTIME_METRICS -DPURE_EMBEDDED_PROGRAM -I$(BUILD) server.c $(HOST) flow.c $(CORE) $(LDFLAGS) $(LDLIBS) -o $@ @@ -130,7 +134,7 @@ $(BUILD)/pure-run: pure-run.c module-store.c $(MODULE) module-store.h $(MODULE_H $(BUILD)/pure-edit-module: edit-module.c module-store.c $(MODULE) module-store.h $(MODULE_HEADERS) | $(BUILD) $(CC) $(CPPFLAGS) $(CFLAGS) edit-module.c module-store.c $(MODULE) $(LDFLAGS) -lm $(LDLIBS) -o $@ -check-module: $(BUILD)/test-module $(BUILD)/pure-run $(BUILD)/pure-edit-module +check-module: $(BUILD)/test-module $(BUILD)/pure-run $(BUILD)/pure-edit-module $(BUILD)/pure-connections-module $(BUILD)/test-module $(BUILD)/test-tree-store: test-tree-store.c store-tree.c $(HEADERS) | $(BUILD) diff --git a/prototypes/connections/server.c b/prototypes/connections/server.c index 505bb71..448a915 100644 --- a/prototypes/connections/server.c +++ b/prototypes/connections/server.c @@ -2,6 +2,9 @@ #include "store.h" #include "program.h" #include "store-port.h" +#ifdef PURE_MODULE_SERVER +#include "module-store.h" +#endif #ifdef PURE_EMBEDDED_PROGRAM #include "embedded-plan.h" #endif @@ -27,6 +30,14 @@ #if defined(PURE_HTTP_ASYNC) && !defined(PURE_ASYNC_EFFECTS) #error PURE_HTTP_ASYNC requires PURE_ASYNC_EFFECTS #endif +/* A module server runs an aup module instead of a document program: GET is + * its show message, PUT its edit message. Effects are taken on the owner + * thread, applied with deferred saving, saved once per burst and only then + * completed, so concurrent requests share one commit. */ +#if defined(PURE_MODULE_SERVER) && (!defined(PURE_ASYNC_EFFECTS) || defined(PURE_HTTP_ASYNC) || \ + defined(PURE_EMBEDDED_PROGRAM)) +#error PURE_MODULE_SERVER takes effects on the owner thread and has no program +#endif enum { SLOTS = 8, HEADER_LIMIT = 4096, CHUNK = 65536 }; typedef enum { READING_HEADER, READING_BODY, WRITING, WAITING_EFFECT } Phase; @@ -58,7 +69,12 @@ typedef struct { } Connection; struct Server { const char *path; -#ifndef PURE_EMBEDDED_PROGRAM +#ifdef PURE_MODULE_SERVER + PureModule module; + PureApp app; + uint32_t show, edit; + uint64_t saves, burst_events; /* saves is the number of bursts */ +#elif !defined(PURE_EMBEDDED_PROGRAM) PureProgram program; #endif PureRuntime runtime; @@ -252,8 +268,136 @@ static void finish_effects(Server *server) { } #endif +#ifdef PURE_MODULE_SERVER +static void release_copy(void *allocation, size_t size) { + (void)size; + free(allocation); +} + +static bool allocate_ingress(void *context, size_t size, PureResource *out) { + (void)context; + void *bytes = malloc(size); + if (!bytes) return false; + *out = (PureResource){.body = {bytes, size}, .allocation = bytes, .allocation_size = size, .release = release_copy}; + return true; +} + +typedef struct { unsigned char *bytes; size_t size; const unsigned char *first; size_t spans; } Reply; + +static bool measure_reply(void *context, PureSpan span) { + Reply *r = context; + if (!r->spans++) r->first = span.data; + r->size += span.size; + return true; +} + +static bool copy_reply(void *context, PureSpan span) { + Reply *r = context; + if (span.size) memcpy(r->bytes + r->size, span.data, span.size); + r->size += span.size; + return true; +} + +/* One burst of requests: applied in order, saved with one commit, then + * completed. If the save fails every request of the burst fails, reads too: + * they saw changes that were never stored. */ +typedef struct { + PureTicket ticket; + PureInput input; + PureResult status; + int64_t revision; + PureResource resource; + PureSpan reply; + char key[65]; +} Applied; + +static void apply_effect(Server *server, const PureEffect *effect, Applied *out) { + PureApp *app = &server->app; + *out = (Applied){.ticket = effect->ticket, .input = effect->input}; + memcpy(out->key, effect->key, strlen(effect->key) + 1); + bool read = effect->input == PURE_INPUT_READ; + out->status = pure_app_message(app, read ? server->show : server->edit, effect->key, + read ? -1 : effect->expected, read ? (PureSpan){0} : effect->value); + out->revision = app->revision; + if (out->status != PURE_OK) return; + ++server->burst_events; + Reply reply = {0}; + pure_app_reply(app, measure_reply, &reply); + if (!read) { + /* The module was checked at start: an edit replies part of its request. */ + out->reply = (PureSpan){reply.size ? reply.first : NULL, reply.size}; + return; + } + if (reply.size > effect->available) { out->status = PURE_BUDGET; return; } + if (!reply.size) return; + unsigned char *copy = malloc(reply.size); + if (!copy) { out->status = PURE_ALLOCATION; return; } + reply = (Reply){.bytes = copy}; + pure_app_reply(app, copy_reply, &reply); + out->resource = (PureResource){.body = {copy, reply.size}, .allocation = copy, + .allocation_size = reply.size, .release = release_copy}; +} + +static void finish_burst(Server *server, Applied *applied, size_t count) { + if (!count) return; + PureResult saved = pure_app_save(&server->app); + ++server->saves; + for (size_t i = 0; i < count; ++i) { + Applied *a = &applied[i]; + PureResult status = a->status == PURE_OK ? saved : a->status; + if (!pure_runtime_complete(a->ticket, status, a->input == PURE_INPUT_READ ? &a->resource : NULL, + status == PURE_OK ? a->reply : (PureSpan){0}, a->revision)) abort(); + for (size_t j = 0; j < SLOTS; ++j) { + Connection *c = &server->connections[j]; + if (c->fd < 0 || c->phase != WAITING_EFFECT || + c->ticket.slot != a->ticket.slot || c->ticket.generation != a->ticket.generation) continue; + PureOutcome result; + if (!pure_runtime_result(a->ticket, &result)) abort(); + c->body = NULL; + answer(c, result); + break; + } + } +} + +/* Every request that became ready since the last burst; a key seen again + * ends the burst first, so a document's revision moves once per save. */ +static void serve_bursts(Server *server) { + Applied applied[SLOTS]; + size_t count = 0; + PureEffect effect; + while (pure_runtime_take(&server->runtime, &effect) == PURE_TAKE_EFFECT) { + bool repeated = false; + for (size_t i = 0; i < count && !repeated; ++i) repeated = !strcmp(applied[i].key, effect.key); + if (repeated) { + finish_burst(server, applied, count); + count = 0; + } + apply_effect(server, &effect, &applied[count++]); + } + finish_burst(server, applied, count); +} + +/* PUT replies must be borrowed from the request, so an edit handler may + * reply only a field of its message. */ +static bool edit_replies_borrow(const PureModule *m, uint32_t edit) { + for (uint32_t h = 0; h < m->handlers; ++h) { + if (m->handler[h].event != 1 + 2 * edit) continue; + const PureBlock *code = &m->handler[h].code; + for (uint32_t pc = code->start; pc < code->start + code->length; ++pc) + if (m->code[pc].op == PURE_OP_REPLY && (pc == code->start || m->code[pc - 1].op != PURE_OP_FIELD)) + return false; + } + return true; +} +#endif + static void run_request(Connection *c) { if (c->put && !pure_runtime_ready(c->ticket)) abort(); +#ifdef PURE_MODULE_SERVER + c->phase = WAITING_EFFECT; /* the burst after this poll round serves it */ + return; +#endif #ifdef PURE_HTTP_ASYNC Server *server = c->server; PureEffect effect; @@ -411,7 +555,10 @@ static bool close_on_exec(int fd) { int main(int argc, char **argv) { Server server = {.budget = 4u * 1024u * 1024u, .timeout_ms = 10000}; unsigned port = 8767; -#ifndef PURE_EMBEDDED_PROGRAM +#ifdef PURE_MODULE_SERVER + const char *module_path = NULL; + bool explain = false; +#elif !defined(PURE_EMBEDDED_PROGRAM) const char *program_path = NULL; bool explain = argc == 3 && !strcmp(argv[1], "--check-program"); char program_error[160]; @@ -423,7 +570,9 @@ int main(int argc, char **argv) { if (i + 1 == argc) goto usage; uint64_t value; if (!strcmp(argv[i], "--store")) server.path = argv[i + 1]; -#ifndef PURE_EMBEDDED_PROGRAM +#ifdef PURE_MODULE_SERVER + else if (!strcmp(argv[i], "--module")) module_path = argv[i + 1]; +#elif !defined(PURE_EMBEDDED_PROGRAM) else if (!strcmp(argv[i], "--program")) program_path = argv[i + 1]; #endif else if (!strcmp(argv[i], "--port") && number(argv[i + 1], 65535, &value)) port = (unsigned)value; @@ -433,7 +582,30 @@ int main(int argc, char **argv) { else goto usage; } if (!server.path && !explain) goto usage; -#ifndef PURE_EMBEDDED_PROGRAM +#ifdef PURE_MODULE_SERVER + if (!module_path) goto usage; + { + static unsigned char bytes[1 << 17]; + char error[128]; + FILE *file = fopen(module_path, "rb"); + size_t size = file ? fread(bytes, 1, sizeof(bytes), file) : 0; + bool read_ok = file && !ferror(file) && size < sizeof(bytes); + if (file) fclose(file); + if (!read_ok || pure_module_load(bytes, size, &server.module, error, sizeof(error)) != PURE_OK) { + fprintf(stderr, "Module: %s\n", read_ok ? error : "cannot read"); return 2; + } + int show = pure_module_input(&server.module, "show"), edit = pure_module_input(&server.module, "edit"); + if (show < 0 || edit < 0 || server.module.input[show].kind != PURE_KIND_MESSAGE || !server.module.input[show].keyed || + server.module.input[edit].kind != PURE_KIND_MESSAGE || !server.module.input[edit].keyed) { + fputs("Module: needs keyed messages named show and edit\n", stderr); return 2; + } + if (!edit_replies_borrow(&server.module, (uint32_t)edit)) { + fputs("Module: edit may reply only a field of its message\n", stderr); return 2; + } + server.show = (uint32_t)show; + server.edit = (uint32_t)edit; + } +#elif !defined(PURE_EMBEDDED_PROGRAM) if (!program_path) goto usage; if (!pure_program_load(program_path, &server.program, program_error, sizeof(program_error))) { fprintf(stderr, "Program: %s\n", program_error); return 2; @@ -464,6 +636,13 @@ int main(int argc, char **argv) { return 0; } #endif +#ifdef PURE_MODULE_SERVER + const void *program = &server.app; + if (pure_store_init(server.path) != PURE_OK) { fputs("Cannot initialize byte store directory\n", stderr); return 1; } + if (pure_app_start(&server.app, &server.module, pure_tree_data(server.path)) != PURE_OK || + !pure_app_defer(&server.app, true)) { fputs("Cannot start the module on the store\n", stderr); return 1; } + PureHost host = {.allocate = allocate_ingress}; +#else const PureProgram *program = #ifdef PURE_EMBEDDED_PROGRAM &pure_embedded_program; @@ -472,6 +651,7 @@ int main(int argc, char **argv) { #endif if (pure_store_init(server.path) != PURE_OK) { fputs("Cannot initialize byte store directory\n", stderr); return 1; } PureHost host = pure_store_host(server.path); +#endif #ifdef PURE_HTTP_ASYNC server.effect_host = host; host.read = NULL; @@ -479,7 +659,7 @@ int main(int argc, char **argv) { #endif if (!pure_runtime_init(&server.runtime, server.work, SLOTS, program, host, server.budget)) return 1; -#ifdef PURE_HTTP_ASYNC +#if defined(PURE_HTTP_ASYNC) || defined(PURE_MODULE_SERVER) if (!pure_runtime_enable_async(&server.runtime)) abort(); #endif signal(SIGPIPE, SIG_IGN); @@ -589,8 +769,19 @@ int main(int argc, char **argv) { if (c->phase == WRITING) write_connection(c); else read_connection(c); } } +#ifdef PURE_MODULE_SERVER + serve_bursts(&server); +#endif + /* A module server takes every waiting connection at once, so their + * requests can share the next burst. */ +#ifdef PURE_MODULE_SERVER + for (bool more = true; more && !stopping && (polls[0].revents & POLLIN) && server.active < SLOTS;) { +#else if (!stopping && (polls[0].revents & POLLIN)) { + bool more; +#endif int fd = accept(listener, NULL, NULL); + more = fd >= 0; if (fd < 0) continue; if (!nonblocking(fd)) { close(fd); continue; } if (server.send_buffer) setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &server.send_buffer, sizeof(server.send_buffer)); @@ -601,6 +792,7 @@ int main(int argc, char **argv) { if (server.active > server.peak_active) server.peak_active = server.active; break; } + (void)more; } } close(listener); @@ -611,6 +803,10 @@ int main(int argc, char **argv) { #endif pure_runtime_stop(&server.runtime); for (size_t i = 0; i < SLOTS; ++i) close_connection(&server.connections[i]); +#ifdef PURE_MODULE_SERVER + fprintf(stderr, "{\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 "}\n", server.saves, server.burst_events); + pure_app_stop(&server.app); +#endif #ifdef PURE_HTTP_ASYNC pthread_mutex_lock(&server.job_mutex); server.worker_stop = true; @@ -635,7 +831,10 @@ int main(int argc, char **argv) { server.expired, server.runtime.releases, server.prefix_copy_bytes, server.runtime.borrowed_output_bytes, sizeof(server)); return 0; usage: -#ifdef PURE_EMBEDDED_PROGRAM +#ifdef PURE_MODULE_SERVER + fputs("pure-connections-module --module FILE.pmod --store DIRECTORY [--port 8767] [--budget BYTES] " + "[--timeout-ms 10000] [--send-buffer BYTES]\n", stderr); +#elif defined(PURE_EMBEDDED_PROGRAM) fputs("pure-connections-fixed --store DIRECTORY [--port 8767] [--budget BYTES] [--timeout-ms 10000] [--send-buffer BYTES]\n", stderr); #else fputs("pure-connections --program FILE --store DIRECTORY [--port 8767] [--budget BYTES] [--timeout-ms 10000] [--send-buffer BYTES]\n" |