diff options
Diffstat (limited to 'prototypes/connections')
| -rw-r--r-- | prototypes/connections/Makefile | 8 | ||||
| -rw-r--r-- | prototypes/connections/serve.c | 631 | ||||
| -rw-r--r-- | prototypes/connections/server.c | 211 |
3 files changed, 641 insertions, 209 deletions
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" |