From 7771461a93f993425f97d4e05c91e64d17179e60 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Sat, 26 Sep 2026 05:40:20 +0100 Subject: Serve modules over persistent connections pure-serve replaces the module mode of server.c, which goes back to what it was. It keeps the document contract, the runtime core (a work slot per connection, the payload budget, core-owned ingress, borrowed PUT replies) and the group commits, and adds what server.c lacked: HTTP/1.1 persistent connections with in-order pipelining, an edge-triggered epoll owner thread, up to 1,024 connections allocated as they arrive and reused, one writev per answer, and documents opened through a store handle. On the Pi 5 with one thread it serves 106k small keep-alive reads per second (axum on three cores: 126k) at the same p99, 1,741 1 MiB reads (axum: 1,370), and 1,126 durable writes from 64 connections (crash-safe axum: 372), in 7.3 MiB against axum's 17-20 MiB. Note 075 has the table. --- benchmarks/http/README.md | 2 +- benchmarks/http/run.py | 4 +- docs/design/008-module-format.md | 37 +- docs/research/075-http-host-against-axum.md | 23 + prototypes/connections/Makefile | 8 +- prototypes/connections/serve.c | 631 ++++++++++++++++++++++++++++ prototypes/connections/server.c | 211 +--------- 7 files changed, 686 insertions(+), 230 deletions(-) create mode 100644 prototypes/connections/serve.c diff --git a/benchmarks/http/README.md b/benchmarks/http/README.md index 3fc5740..3c6eb4f 100644 --- a/benchmarks/http/README.md +++ b/benchmarks/http/README.md @@ -5,7 +5,7 @@ Pure의 문서 서버들과 같은 계약의 axum 서비스를 같은 기기· ```sh cc -O2 -std=c17 -o load load.c (cd axum-baseline && cargo build --release) -make -C ../../prototypes/connections build/pure-connections-module +make -C ../../prototypes/connections build/pure-serve make -C ../../prototypes/connections BUILD=build/tree STORE=store-tree.c build/tree/pure-connections aup build ../../../aup/examples/editor.aup -o editor.pmod python3 run.py --module editor.pmod --directory /var/tmp # 저장소를 둘 장치에서 diff --git a/benchmarks/http/run.py b/benchmarks/http/run.py index 8202440..96f4bc0 100644 --- a/benchmarks/http/run.py +++ b/benchmarks/http/run.py @@ -70,8 +70,8 @@ def main(): servers = { 'pure-program': lambda d: [PURE / 'build/tree/pure-connections', '--program', PURE / 'examples/editor.pure', '--store', d, '--port', '0', '--budget', str(args.budget)], - 'pure-module': lambda d: [PURE / 'build/pure-connections-module', '--module', args.module, '--store', d, - '--port', '0', '--budget', str(args.budget)], + 'pure-serve': lambda d: [PURE / 'build/pure-serve', '--module', args.module, '--store', d, '--port', '0', + '--budget', str(args.budget)], 'axum-memory': lambda d: [axum, '--port', '0'], 'axum-durable': lambda d: [axum, '--port', '0', '--durable', d], 'axum-atomic': lambda d: [axum, '--port', '0', '--atomic', d], diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md index 2d30003..2407980 100644 --- a/docs/design/008-module-format.md +++ b/docs/design/008-module-format.md @@ -197,24 +197,25 @@ text·리스트가 남을 때만 위치를 유지한다. 실패하면 묶음 전체가 되돌려지고, 저장 중 프로세스가 죽으면 묶음 전체가 남거나 전혀 남지 않는다. 비용(Pi 5, SD의 ext4): 저장하는 이벤트 하나에 약 10.6 ms가 더 든다. 대부분 두 번의 `fsync`다. 300항목 리스트에서 인덱스 이동 한 번은 파일을 약 220 B 키운다(압축 전). -## 런타임 코어 위의 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개). +## HTTP 호스트 (`pure-serve`) + +`pure-serve --module 앱.pmod --store 경로`는 기존 문서 서버(`server.c`)와 같은 HTTP 계약을 모듈로 제공한다. +`GET /objects/키`는 모듈의 `show` 메시지, `PUT`(`If-Match`/`If-None-Match`)은 `edit` 메시지다. + +- **연결.** HTTP/1.1 지속 연결과 파이프라이닝(응답은 순서대로), edge-triggered epoll 소유 스레드 하나, 기다리는 + 연결을 한 번에 받는다. 연결 구조체는 받을 때 할당하고 닫으면 재사용하므로 메모리는 설정한 최대치가 아니라 + 열린 연결 수를 따른다(기본 최대 1,024). 응답은 머리말과 본문을 `writev` 한 번으로 보낸다. +- **런타임 코어.** 요청은 `runtime.c`를 그대로 거친다(연결마다 작업 슬롯, 입력 바이트 예산, 코어가 소유한 + ingress, 요청에서 빌린 PUT 응답). 효과는 비동기 인터페이스(`pure_runtime_take`/`complete`)로 소유 스레드에서 처리한다. +- **이벤트 한 바퀴가 묶음 하나다.** 준비된 요청을 모두 지연 저장 모드의 앱에 차례로 적용하고, 저장 한 번 **뒤에** + 응답한다. 저장이 실패하면 묶음의 읽기까지 모두 실패한다. 같은 키가 다시 오거나 문서 수가 한 저장의 한도를 넘으려 + 하면 묶음을 먼저 끝낸다. 그래서 묶음 안의 모든 요청은 자기 묶음의 저장으로 저장되거나 그것과 함께 실패한다. +- **저장소 핸들.** 문서와 상태를 저장소 핸들(`pure_store_data`)로 열어 파일 매핑 하나를 공유한다. +- GET 응답은 예산에 계상되는 자원으로 복사한다. PUT 응답은 요청 본문에서 빌려야 하므로 `edit` 핸들러는 메시지 + 필드만 응답할 수 있다(`reply edit.insert`). 시작할 때 확인한다. + +성능과 비교는 [연구 075](../research/075-http-host-against-axum.md). 검증(aup `make check`): 무작위 요청에 기존 서버와 +상태 코드·ETag·오류·본문이 같고, 동시 편집이 커밋을 나누어 쓰며, `kill -9` 뒤에도 확인 응답한 편집이 모두 남는다. 아직 모듈 앱을 런타임 코어의 효과 단위로 세분하지는 않았다(`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)보다 크다. 줄이는 것이 diff --git a/docs/research/075-http-host-against-axum.md b/docs/research/075-http-host-against-axum.md index 96814d3..308363e 100644 --- a/docs/research/075-http-host-against-axum.md +++ b/docs/research/075-http-host-against-axum.md @@ -52,3 +52,26 @@ PUT은 연결마다 자기 키 앞에 16 B를 넣고 받은 ETag로 다음 `If-M 4. 모듈 서버의 무복사 GET. 중단 기준: 어떤 단계든 메모리가 axum보다 커지거나, 저장 보장(확인 응답 = 저장됨, `kill -9` 검사)이 깨지면 그 방식은 버린다. + +## 1단계 결과: `pure-serve` (소유 스레드 하나) + +새 호스트 [`serve.c`](../../prototypes/connections/serve.c): 지속 연결·파이프라이닝, edge-triggered epoll, 연결 +최대 1,024(받을 때 할당), `writev` 응답, 같은 런타임 코어와 묶음 커밋. 같은 조건에서 다시 쟀다. + +`strace -c`로 보니 GET 하나마다 저장소 draft를 여닫느라 시스템 호출이 9번(파일·디렉터리 열기, `fstat` 두 번, +전체 `mmap`/`munmap`, head 읽기)이었다. 저장소 핸들(`pure_store_open`)이 매핑을 공유하고 열 때 파일이 바뀌었는지만 +본다(2번). + +| 작업 | 이전 Pure | pure-serve, 핸들 전 | **pure-serve** | axum 메모리(3코어) | axum 충돌 안전 | +|---|---:|---:|---:|---:|---:| +| GET 64 B, keep-alive | 23,892 | 57,971 | **106,447** (p99 736) | 126,160 (p99 768) | 126,865 | +| GET 64 B, 매번 새 연결 | 23,852 | 22,912 | 23,315 | 21,947 | 22,703 | +| GET 1 MiB, keep-alive | 1,197 | 1,537 | **1,741** | 1,370 | 1,354 | +| PUT 저장 보장, 연결 64 | 191 | 1,081 | **1,126** | – | 372 | +| PUT 저장 보장, 연결 8 | 171 | 333 | **336** | – | 63 | + +최대 RSS: pure-serve 7.3 MiB, axum 17.4–19.6 MiB. + +- 코어 하나로 axum(코어 3개)의 84%(작은 keep-alive GET), p99는 같다. 큰 읽기는 27% 빠르다. +- 저장 보장 쓰기는 충돌 안전 axum의 3.0–5.3배, 안전하지 않은 axum(429/s)보다도 빠르다. +- 남은 차이는 코어 수다. 다음은 2단계(코어마다 소유 스레드)다. diff --git a/prototypes/connections/Makefile b/prototypes/connections/Makefile index 83af085..b683230 100644 --- a/prototypes/connections/Makefile +++ b/prototypes/connections/Makefile @@ -40,9 +40,9 @@ $(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 $@ +# The HTTP host for aup modules: persistent connections, epoll, group commits. +$(BUILD)/pure-serve: serve.c flow.c module-store.c $(MODULE) $(CORE) module-store.h $(MODULE_HEADERS) | $(BUILD) + $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_ASYNC_EFFECTS serve.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 $@ @@ -134,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 $(BUILD)/pure-connections-module +check-module: $(BUILD)/test-module $(BUILD)/pure-run $(BUILD)/pure-edit-module $(BUILD)/pure-serve $(BUILD)/test-module $(BUILD)/test-tree-store: test-tree-store.c store-tree.c $(HEADERS) | $(BUILD) diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c new file mode 100644 index 0000000..39f4f33 --- /dev/null +++ b/prototypes/connections/serve.c @@ -0,0 +1,631 @@ +#define _GNU_SOURCE +/* pure-serve: the HTTP host for aup modules (docs/research/075). + * + * The document contract of server.c (GET and PUT /objects/KEY, If-Match or + * If-None-Match: *, ETag = revision, X-Pure-Error) served by a module: GET is + * its show message and PUT its edit message. HTTP/1.1 connections persist + * and may pipeline; one epoll owner thread accepts, reads and writes without + * blocking. Requests go through the runtime core (a work slot per + * connection, the payload budget, ingress owned by the core, PUT replies + * borrowed from the request). After each round of events every ready + * request is applied to the module with deferred saving, one save stores + * the burst, and only then are its requests answered: an answer is durable, + * and a failed save fails the whole burst, reads included. */ +#include "module-store.h" +#include "runtime.h" +#include "store.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +enum { HEADER_LIMIT = 4096, EVENTS = 256 }; +typedef enum { CLOSED, READING_HEAD, READING_BODY, WAITING, WRITING } Phase; +typedef struct Host Host; + +/* Allocated when accepted and reused after closing: memory follows the + * connections that are open, not the configured maximum. */ +typedef struct Connection { + Host *host; + struct Connection *next; /* on the spare or the closed list */ + size_t index; /* in the host's table */ + int fd; + Phase phase; + bool readable, writable, keep_alive, put; + uint64_t deadline; + char key[65]; + int64_t expected; + size_t body_size, received, in_used; + unsigned char *body; + PureTicket ticket; + PureTransfer transfer; + char head[320]; + char in[HEADER_LIMIT]; /* the header being read, then pipelined bytes */ +} Connection; + +/* One request applied in the current burst. */ +typedef struct { + Connection *connection; + PureTicket ticket; + PureInput input; + PureResult status; + int64_t revision; + PureResource resource; + PureSpan reply; + char key[65]; +} Applied; + +struct Host { + PureModule module; + PureApp app; + uint32_t show, edit; + PureRuntime runtime; + PureWork *work; + Connection **table; /* open connections by index */ + Connection **by_slot; /* the connection waiting on a work slot */ + Connection *spare, *closed; + Applied *applied; + size_t capacity, active; + int epoll, listener; + uint64_t timeout_ms; + uint64_t accepted, requests, bursts, burst_events, expired; +}; + +static volatile sig_atomic_t stopping; +static void stop(int signal_number) { (void)signal_number; stopping = 1; } + +static uint64_t now_ms(void) { + struct timespec t; + clock_gettime(CLOCK_MONOTONIC, &t); + return (uint64_t)t.tv_sec * 1000 + (uint64_t)t.tv_nsec / 1000000; +} + +static bool number(const char *text, uint64_t maximum, uint64_t *value) { + if (!*text) return false; + uint64_t n = 0; + for (; *text; ++text) { + if (*text < '0' || *text > '9') return false; + unsigned digit = (unsigned)(*text - '0'); + if (n > (maximum - digit) / 10) return false; + n = n * 10 + digit; + } + *value = n; + return true; +} + +/* ---- Connections ------------------------------------------------------ */ + +static void release_ticket(void *owner) { + Connection *c = owner; + if (c->ticket.runtime && !pure_runtime_release(c->ticket)) abort(); + c->ticket = (PureTicket){0}; + c->body = NULL; +} + +/* Events of this round may still name c: it is reused after the round. */ +static void close_connection(Connection *c) { + if (c->phase == CLOSED) return; + Host *host = c->host; + if (c->transfer.active) pure_transfer_abort(&c->transfer); + else release_ticket(c); + close(c->fd); /* also leaves the epoll set */ + c->phase = CLOSED; + host->table[c->index] = NULL; + c->next = host->closed; + host->closed = c; + --host->active; +} + +static void respond(Connection *c, int status, const char *reason, const char *error, int64_t revision, + PureSpan body) { + char tag[48] = ""; + if (revision > 0) snprintf(tag, sizeof(tag), "ETag: \"%" PRId64 "\"\r\n", revision); + int length = snprintf(c->head, sizeof(c->head), + "HTTP/1.1 %d %s\r\nContent-Length: %zu\r\n%sContent-Type: application/octet-stream\r\n" + "Cache-Control: no-store\r\nX-Pure-Error: %s\r\n%s\r\n", + status, reason, body.size, c->keep_alive ? "" : "Connection: close\r\n", error, tag); + c->phase = WRITING; + pure_transfer_init(&c->transfer, (PureSpan){(const unsigned char *)c->head, (size_t)length}, body, c, + release_ticket); +} + +/* A request the host cannot read further: answer and close. */ +static void refuse(Connection *c, int status, const char *reason, const char *error) { + release_ticket(c); + c->keep_alive = false; + respond(c, status, reason, error, 0, (PureSpan){0}); +} + +static void answer(Connection *c, PureOutcome result) { + switch (result.status) { + case PURE_OK: respond(c, 200, "OK", "none", result.revision, result.body); return; + case PURE_CONFLICT: respond(c, 412, "Precondition Failed", "conflict", result.revision, (PureSpan){0}); return; + case PURE_NOT_FOUND: respond(c, 404, "Not Found", "not_found", 0, (PureSpan){0}); break; + case PURE_BUSY: respond(c, 503, "Service Unavailable", "store_busy", 0, (PureSpan){0}); break; + case PURE_BUDGET: respond(c, 503, "Service Unavailable", "payload_budget", 0, (PureSpan){0}); break; + case PURE_ALLOCATION: respond(c, 503, "Service Unavailable", "allocation", 0, (PureSpan){0}); break; + case PURE_CLOSED: respond(c, 503, "Service Unavailable", "stopped", 0, (PureSpan){0}); break; + case PURE_DENIED: respond(c, 403, "Forbidden", "denied", 0, (PureSpan){0}); break; + case PURE_LIMIT: respond(c, 413, "Content Too Large", "limit", 0, (PureSpan){0}); break; + case PURE_UNCERTAIN: respond(c, 500, "Internal Server Error", "durability_unknown", 0, (PureSpan){0}); break; + default: respond(c, 500, "Internal Server Error", "storage", 0, (PureSpan){0}); break; + } + /* A failed request carries no body: the payload can go now. */ + if (!result.body.size) release_ticket(c); +} + +/* Parse one request head of `size` bytes (through the blank line): the same + * bounded HTTP/1.1 subset as server.c, plus Connection. */ +static bool parse_head(Connection *c, char *head, size_t size) { + for (size_t i = 0; i < size; ++i) { + unsigned char b = (unsigned char)head[i]; + if ((b < 32 && b != '\r' && b != '\n' && b != '\t') || b == 127) return false; + } + head[size - 2] = 0; /* keep the last CRLF of the last header */ + char *line = head, *next = strstr(line, "\r\n"); + if (!next) return false; + *next = 0; + char *path = strchr(line, ' '); + if (!path) return false; + *path++ = 0; + c->put = !strcmp(line, "PUT"); + if (!c->put && strcmp(line, "GET")) return false; + char *version = strchr(path, ' '); + if (!version) return false; + *version++ = 0; + if (strcmp(version, "HTTP/1.1") || strncmp(path, "/objects/", 9)) return false; + const char *key = path + 9; + size_t key_size = strlen(key); + if (!key_size || key_size > 64) return false; + for (size_t i = 0; i < key_size; ++i) { + char b = key[i]; + if (!((b >= 'a' && b <= 'z') || (b >= 'A' && b <= 'Z') || (b >= '0' && b <= '9') || b == '-' || b == '_')) + return false; + } + memcpy(c->key, key, key_size + 1); + bool length_seen = false, condition_seen = false, host_seen = false; + c->body_size = 0; + c->expected = -1; + c->keep_alive = true; + line = next + 2; + while (*line) { + next = strstr(line, "\r\n"); + if (!next) return false; + *next = 0; + char *value = strchr(line, ':'); + if (!value || value == line) return false; + *value++ = 0; + for (char *p = line; *p; ++p) + if (!((*p >= 'a' && *p <= 'z') || (*p >= 'A' && *p <= 'Z') || (*p >= '0' && *p <= '9') || *p == '-')) + return false; + while (*value == ' ' || *value == '\t') ++value; + char *tail = value + strlen(value); + while (tail > value && (tail[-1] == ' ' || tail[-1] == '\t')) *--tail = 0; + uint64_t n; + if (!strcasecmp(line, "Content-Length")) { + if (length_seen || !number(value, SIZE_MAX, &n)) return false; + length_seen = true; + c->body_size = (size_t)n; + } else if (!strcasecmp(line, "Host")) { + if (host_seen || !*value) return false; + host_seen = true; + } else if (!strcasecmp(line, "If-Match")) { + if (condition_seen || tail - value < 3 || *value != '"' || tail[-1] != '"') return false; + tail[-1] = 0; + if (!number(value + 1, INT64_MAX, &n) || !n) return false; + c->expected = (int64_t)n; + condition_seen = true; + } else if (!strcasecmp(line, "If-None-Match")) { + if (condition_seen || strcmp(value, "*")) return false; + c->expected = 0; + condition_seen = true; + } else if (!strcasecmp(line, "Connection")) { + if (!strcasecmp(value, "close")) c->keep_alive = false; + else if (strcasecmp(value, "keep-alive")) return false; + } else if (!strcasecmp(line, "Transfer-Encoding") || !strcasecmp(line, "Expect")) return false; + line = next + 2; + } + return host_seen && (c->put ? length_seen && condition_seen : !condition_seen && c->body_size == 0); +} + +/* A whole head is in `in`: admit the request and take any body bytes that + * came with it; later bytes stay for the next request. */ +static void start_request(Connection *c, size_t head_size) { + Host *host = c->host; + char head[HEADER_LIMIT]; + memcpy(head, c->in, head_size); + size_t extra = c->in_used - head_size; + memmove(c->in, c->in + head_size, extra); + c->in_used = extra; + if (!parse_head(c, head, head_size)) { refuse(c, 400, "Bad Request", "request"); return; } + if (c->body_size > PURE_OBJECT_LIMIT) { refuse(c, 413, "Content Too Large", "limit"); return; } + ++host->requests; + PureResult admitted = pure_runtime_accept(&host->runtime, c->put ? PURE_INPUT_WRITE : PURE_INPUT_READ, c->key, + c->expected, c->body_size, &c->ticket); + if (admitted != PURE_OK) { + c->ticket = (PureTicket){0}; + /* An unread body would desynchronize the stream. */ + if (c->body_size) c->keep_alive = false; + answer(c, (PureOutcome){.status = admitted}); + return; + } + host->by_slot[c->ticket.slot] = c; + if (!c->put) { c->phase = WAITING; return; } + c->body = pure_runtime_ingress(c->ticket, NULL); + size_t take = c->in_used < c->body_size ? c->in_used : c->body_size; + if (take) memcpy(c->body, c->in, take); + memmove(c->in, c->in + take, c->in_used - take); + c->in_used -= take; + c->received = take; + c->phase = READING_BODY; + if (c->received == c->body_size) { + if (!pure_runtime_ready(c->ticket)) abort(); + c->phase = WAITING; + } +} + +static size_t head_end(const Connection *c) { + for (size_t i = 3; i < c->in_used; ++i) + if (!memcmp(c->in + i - 3, "\r\n\r\n", 4)) return i + 1; + return 0; +} + +/* Advance a connection as far as its socket allows without blocking. */ +static void drive(Connection *c) { + for (;;) { + if (c->phase == READING_HEAD) { + size_t end = head_end(c); + if (end) { start_request(c, end); continue; } + if (c->in_used == HEADER_LIMIT) { refuse(c, 431, "Request Header Fields Too Large", "headers"); continue; } + if (!c->readable) return; + ssize_t n = recv(c->fd, c->in + c->in_used, HEADER_LIMIT - c->in_used, 0); + if (n > 0) { c->in_used += (size_t)n; c->deadline = now_ms() + c->host->timeout_ms; continue; } + if (n < 0 && errno == EAGAIN) { c->readable = false; return; } + if (n < 0 && errno == EINTR) continue; + close_connection(c); + return; + } + if (c->phase == READING_BODY) { + if (!c->readable) return; + ssize_t n = recv(c->fd, c->body + c->received, c->body_size - c->received, 0); + if (n > 0) { + c->received += (size_t)n; + c->deadline = now_ms() + c->host->timeout_ms; + if (c->received == c->body_size) { + if (!pure_runtime_ready(c->ticket)) abort(); + c->phase = WAITING; + } + continue; + } + if (n < 0 && errno == EAGAIN) { c->readable = false; return; } + if (n < 0 && errno == EINTR) continue; + close_connection(c); + return; + } + if (c->phase == WRITING) { + if (!c->writable) return; + PureTransfer *t = &c->transfer; + struct iovec parts[2]; + int count = 0; + for (size_t part = t->part; part < 2; ++part) { + size_t offset = part == t->part ? t->offset : 0; + if (t->spans[part].size > offset) + parts[count++] = (struct iovec){(void *)(t->spans[part].data + offset), t->spans[part].size - offset}; + } + ssize_t n = count ? writev(c->fd, parts, count) : 0; + if (n < 0 && errno == EAGAIN) { c->writable = false; return; } + if (n < 0 && errno == EINTR) continue; + if (n < 0) { close_connection(c); return; } + for (size_t left = (size_t)n; left && t->active;) { + PureSpan next = pure_transfer_next(t, left); + if (!pure_transfer_consume(t, next.size)) abort(); + left -= next.size; + } + if (t->active) continue; + /* Answered: the next request may already be here. */ + c->deadline = now_ms() + c->host->timeout_ms; + if (!c->keep_alive) { close_connection(c); return; } + c->phase = READING_HEAD; + continue; + } + return; /* WAITING: the next burst answers */ + } +} + +/* ---- Bursts ----------------------------------------------------------- */ + +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; +} + +static void apply_effect(Host *host, const PureEffect *effect, Applied *out) { + PureApp *app = &host->app; + *out = (Applied){.ticket = effect->ticket, .input = effect->input}; + memcpy(out->key, effect->key, strlen(effect->key) + 1); + Connection *c = host->by_slot[effect->ticket.slot]; + if (c && c->phase == WAITING && c->ticket.generation == effect->ticket.generation) out->connection = c; + bool read = effect->input == PURE_INPUT_READ; + out->status = pure_app_message(app, read ? host->show : host->edit, effect->key, + read ? -1 : effect->expected, read ? (PureSpan){0} : effect->value); + out->revision = app->revision; + if (out->status != PURE_OK) return; + ++host->burst_events; + Reply reply = {0}; + pure_app_reply(app, measure_reply, &reply); + if (!read) { + /* Checked at start: an edit replies a field 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(Host *host, size_t count) { + if (!count) return; + PureResult saved = pure_app_save(&host->app); + ++host->bursts; + for (size_t i = 0; i < count; ++i) { + Applied *a = &host->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(); + Connection *c = a->connection; + if (!c || c->phase != WAITING || c->ticket.generation != a->ticket.generation) continue; + PureOutcome result; + if (!pure_runtime_result(a->ticket, &result)) abort(); + answer(c, result); + drive(c); /* most answers fit the socket buffer at once */ + } +} + +/* Every request made ready since the last burst. A burst ends before a key + * repeats or before its documents would outgrow one save, so every request + * in it is stored by its own save or fails with it. */ +static void serve_bursts(Host *host) { + size_t count = 0, keys = 0; + PureEffect effect; + while (pure_runtime_take(&host->runtime, &effect) == PURE_TAKE_EFFECT) { + bool repeated = false; + for (size_t i = 0; i < count && !repeated; ++i) repeated = !strcmp(host->applied[i].key, effect.key); + if (repeated || keys == PURE_APP_PENDING) { + finish_burst(host, count); + count = keys = 0; + } + apply_effect(host, &effect, &host->applied[count++]); + ++keys; + } + finish_burst(host, count); +} + +/* PUT replies are 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; +} + +/* ---- Main ------------------------------------------------------------- */ + +static bool load_module(Host *host, const char *path) { + static unsigned char bytes[1 << 17]; + char error[128]; + FILE *file = fopen(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, &host->module, error, sizeof(error)) != PURE_OK) { + fprintf(stderr, "Module: %s\n", read_ok ? error : "cannot read"); + return false; + } + const PureModule *m = &host->module; + int show = pure_module_input(m, "show"), edit = pure_module_input(m, "edit"); + if (show < 0 || edit < 0 || m->input[show].kind != PURE_KIND_MESSAGE || !m->input[show].keyed || + m->input[edit].kind != PURE_KIND_MESSAGE || !m->input[edit].keyed) { + fputs("Module: needs keyed messages named show and edit\n", stderr); + return false; + } + if (!edit_replies_borrow(m, (uint32_t)edit)) { + fputs("Module: edit may reply only a field of its message\n", stderr); + return false; + } + host->show = (uint32_t)show; + host->edit = (uint32_t)edit; + return true; +} + +static void accept_all(Host *host) { + for (;;) { + int fd = accept4(host->listener, NULL, NULL, SOCK_NONBLOCK | SOCK_CLOEXEC); + if (fd < 0) return; + if (host->active == host->capacity) { close(fd); continue; } /* full: the peer sees the close */ + Connection *c = host->spare; + if (c) host->spare = c->next; + else if (!(c = malloc(sizeof(*c)))) { close(fd); continue; } + size_t index = 0; + while (host->table[index]) ++index; /* a free index exists: active < capacity */ + int one = 1; + setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)); + c->host = host; + c->index = index; + c->fd = fd; + c->phase = READING_HEAD; + c->readable = c->writable = c->keep_alive = true; + c->deadline = now_ms() + host->timeout_ms; + c->ticket = (PureTicket){0}; + c->transfer = (PureTransfer){0}; + c->body = NULL; + c->in_used = 0; + struct epoll_event event = {.events = EPOLLIN | EPOLLOUT | EPOLLRDHUP | EPOLLET, .data.ptr = c}; + if (epoll_ctl(host->epoll, EPOLL_CTL_ADD, fd, &event)) { + close(fd); + c->next = host->spare; + host->spare = c; + continue; + } + host->table[index] = c; + ++host->active; + ++host->accepted; + drive(c); + } +} + +int main(int argc, char **argv) { + static Host host; + const char *module_path = NULL, *store = NULL; + unsigned port = 8767; + size_t budget = 64u << 20; + host.capacity = 1024; + host.timeout_ms = 10000; + for (int i = 1; i < argc; i += 2) { + uint64_t value; + if (i + 1 == argc) goto usage; + if (!strcmp(argv[i], "--module")) module_path = argv[i + 1]; + else if (!strcmp(argv[i], "--store")) store = argv[i + 1]; + else if (!strcmp(argv[i], "--port") && number(argv[i + 1], 65535, &value)) port = (unsigned)value; + else if (!strcmp(argv[i], "--budget") && number(argv[i + 1], SIZE_MAX, &value) && value) budget = (size_t)value; + else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) host.capacity = (size_t)value; + else if (!strcmp(argv[i], "--timeout-ms") && number(argv[i + 1], 3600000, &value) && value) host.timeout_ms = value; + else goto usage; + } + if (!module_path || !store || !load_module(&host, module_path)) goto usage; + if (pure_store_init(store) != PURE_OK) { fputs("Cannot initialize the store directory\n", stderr); return 1; } + PureStore *handle; + if (pure_store_open(store, &handle) != PURE_OK) { fputs("Cannot open the store\n", stderr); return 1; } + if (pure_app_start(&host.app, &host.module, pure_store_data(handle)) != PURE_OK || !pure_app_defer(&host.app, true)) { + fputs("Cannot start the module on the store\n", stderr); + return 1; + } + host.work = calloc(host.capacity, sizeof(*host.work)); + host.table = calloc(host.capacity, sizeof(*host.table)); + host.by_slot = calloc(host.capacity, sizeof(*host.by_slot)); + host.applied = calloc(host.capacity, sizeof(*host.applied)); + if (!host.work || !host.table || !host.by_slot || !host.applied) { fputs("Cannot allocate connections\n", stderr); return 1; } + if (!pure_runtime_init(&host.runtime, host.work, host.capacity, &host.app, + (PureHost){.allocate = allocate_ingress}, budget) || + !pure_runtime_enable_async(&host.runtime)) abort(); + + signal(SIGPIPE, SIG_IGN); + struct sigaction stop_action = {.sa_handler = stop}; + sigemptyset(&stop_action.sa_mask); + sigaction(SIGINT, &stop_action, NULL); + sigaction(SIGTERM, &stop_action, NULL); + host.listener = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0); + int one = 1; + setsockopt(host.listener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)); + struct sockaddr_in address = {.sin_family = AF_INET, .sin_port = htons((uint16_t)port), + .sin_addr.s_addr = htonl(INADDR_LOOPBACK)}; + socklen_t address_size = sizeof(address); + if (host.listener < 0 || bind(host.listener, (struct sockaddr *)&address, sizeof(address)) || + listen(host.listener, 4096) || getsockname(host.listener, (struct sockaddr *)&address, &address_size)) { + perror("listen"); + return 1; + } + host.epoll = epoll_create1(EPOLL_CLOEXEC); + struct epoll_event listen_event = {.events = EPOLLIN, .data.ptr = NULL}; + if (host.epoll < 0 || epoll_ctl(host.epoll, EPOLL_CTL_ADD, host.listener, &listen_event)) { perror("epoll"); return 1; } + printf("{\"port\":%u}\n", ntohs(address.sin_port)); + fflush(stdout); + + struct epoll_event events[EVENTS]; + uint64_t next_sweep = now_ms() + 1000; + while (!stopping) { + int ready = epoll_wait(host.epoll, events, EVENTS, 1000); + for (int e = 0; e < ready; ++e) { + Connection *c = events[e].data.ptr; + if (!c) { accept_all(&host); continue; } + if (c->phase == CLOSED) continue; + if (events[e].events & EPOLLERR) { close_connection(c); continue; } + if (events[e].events & (EPOLLIN | EPOLLRDHUP | EPOLLHUP)) c->readable = true; + if (events[e].events & EPOLLOUT) c->writable = true; + drive(c); + } + serve_bursts(&host); + uint64_t now = now_ms(); + if (now >= next_sweep) { + next_sweep = now + 1000; + for (size_t i = 0; i < host.capacity; ++i) { + Connection *c = host.table[i]; + if (c && c->phase != WAITING && now >= c->deadline) { ++host.expired; close_connection(c); } + } + } + /* The round is over: closed connections can be reused. */ + while (host.closed) { + Connection *c = host.closed; + host.closed = c->next; + c->next = host.spare; + host.spare = c; + } + } + pure_runtime_stop(&host.runtime); + serve_bursts(&host); + for (size_t i = 0; i < host.capacity; ++i) if (host.table[i]) close_connection(host.table[i]); + fprintf(stderr, "{\"active\":%zu,\"retained_payload_bytes\":%zu,\"accepted\":%" PRIu64 ",\"requests\":%" PRIu64 + ",\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 ",\"expired\":%" PRIu64 "}\n", + host.active, host.runtime.retained, host.accepted, host.requests, host.bursts, host.burst_events, + host.expired); + pure_app_stop(&host.app); + pure_store_close(handle); + close(host.listener); + close(host.epoll); + for (Connection **list = (Connection *[]){host.spare, host.closed}, **end = list + 2; list < end; ++list) + for (Connection *c = *list, *next; c; c = next) { next = c->next; free(c); } + free(host.work); + free(host.table); + free(host.by_slot); + free(host.applied); + return 0; +usage: + fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--budget BYTES] [--connections 1024] " + "[--timeout-ms 10000]\n", stderr); + return 2; +} diff --git a/prototypes/connections/server.c b/prototypes/connections/server.c index 448a915..505bb71 100644 --- a/prototypes/connections/server.c +++ b/prototypes/connections/server.c @@ -2,9 +2,6 @@ #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 @@ -30,14 +27,6 @@ #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; @@ -69,12 +58,7 @@ typedef struct { } Connection; struct Server { const char *path; -#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) +#ifndef PURE_EMBEDDED_PROGRAM PureProgram program; #endif PureRuntime runtime; @@ -268,136 +252,8 @@ 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; @@ -555,10 +411,7 @@ 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; -#ifdef PURE_MODULE_SERVER - const char *module_path = NULL; - bool explain = false; -#elif !defined(PURE_EMBEDDED_PROGRAM) +#ifndef PURE_EMBEDDED_PROGRAM const char *program_path = NULL; bool explain = argc == 3 && !strcmp(argv[1], "--check-program"); char program_error[160]; @@ -570,9 +423,7 @@ 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]; -#ifdef PURE_MODULE_SERVER - else if (!strcmp(argv[i], "--module")) module_path = argv[i + 1]; -#elif !defined(PURE_EMBEDDED_PROGRAM) +#ifndef 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; @@ -582,30 +433,7 @@ int main(int argc, char **argv) { else goto usage; } if (!server.path && !explain) goto usage; -#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) +#ifndef 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; @@ -636,13 +464,6 @@ 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; @@ -651,7 +472,6 @@ 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; @@ -659,7 +479,7 @@ int main(int argc, char **argv) { #endif if (!pure_runtime_init(&server.runtime, server.work, SLOTS, program, host, server.budget)) return 1; -#if defined(PURE_HTTP_ASYNC) || defined(PURE_MODULE_SERVER) +#ifdef PURE_HTTP_ASYNC if (!pure_runtime_enable_async(&server.runtime)) abort(); #endif signal(SIGPIPE, SIG_IGN); @@ -769,19 +589,8 @@ 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)); @@ -792,7 +601,6 @@ int main(int argc, char **argv) { if (server.active > server.peak_active) server.peak_active = server.active; break; } - (void)more; } } close(listener); @@ -803,10 +611,6 @@ 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; @@ -831,10 +635,7 @@ 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_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) +#ifdef 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" -- cgit v1.2.3