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. --- prototypes/connections/Makefile | 8 +- prototypes/connections/serve.c | 631 ++++++++++++++++++++++++++++++++++++++++ prototypes/connections/server.c | 211 +------------- 3 files changed, 641 insertions(+), 209 deletions(-) create mode 100644 prototypes/connections/serve.c (limited to 'prototypes/connections') 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