aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--benchmarks/http/README.md2
-rw-r--r--benchmarks/http/run.py4
-rw-r--r--docs/design/008-module-format.md37
-rw-r--r--docs/research/075-http-host-against-axum.md23
-rw-r--r--prototypes/connections/Makefile8
-rw-r--r--prototypes/connections/serve.c631
-rw-r--r--prototypes/connections/server.c211
7 files changed, 686 insertions, 230 deletions
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 <arpa/inet.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <inttypes.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <signal.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <strings.h>
+#include <sys/epoll.h>
+#include <sys/socket.h>
+#include <sys/uio.h>
+#include <time.h>
+#include <unistd.h>
+
+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"
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 17:02:28 +0000