diff options
| author | eunhhu <[email protected]> | 2026-09-26 16:50:03 +0100 |
|---|---|---|
| committer | eunhhu <[email protected]> | 2026-09-26 16:50:03 +0100 |
| commit | 6177f07292682cf5dbf5e91ce101ce116c858e67 (patch) | |
| tree | 98b8c2c543c665c3d445672dd57737dfba2095a1 | |
| parent | b6061fa10436bad78d7601a36712101dfe873643 (diff) | |
| download | pure-6177f07292682cf5dbf5e91ce101ce116c858e67.tar.gz pure-6177f07292682cf5dbf5e91ce101ce116c858e67.zip | |
Serve an app's views and inputs over HTTP
pure-serve answers GET /views/NAME with a view's value (a number or text
as text/plain, a table as JSON rows, a page of them with ?from=&count=,
each cell computed only for the rows asked) and POST /inputs/NAME with an
event: a text input's text is the body, a key input is pressed, a message
takes its u32 fields from the query and its bytes field from the body and
answers its handler's reply. A failed event answers with its fault (404
not_found, 409 exists, 422 the others). No routes are written: the app's
views and inputs are its interface. Documents stay served when the module
has show and edit.
The route of a request is kept per work slot, so an accepted event runs
even if its connection closes before the burst; views and app events join
any burst, whose one save comes before its answers.
| -rw-r--r-- | docs/design/008-module-format.md | 27 | ||||
| -rw-r--r-- | prototypes/connections/serve.c | 394 |
2 files changed, 369 insertions, 52 deletions
diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md index 38eeb07..f12dcbe 100644 --- a/docs/design/008-module-format.md +++ b/docs/design/008-module-format.md @@ -241,8 +241,22 @@ text·리스트가 남을 때만 위치를 유지한다. ## HTTP 호스트 (`pure-serve`) -`pure-serve --module 앱.pmod --store 경로`는 기존 문서 서버(`server.c`)와 같은 HTTP 계약을 모듈로 제공한다. -`GET /objects/키`는 모듈의 `show` 메시지, `PUT`(`If-Match`/`If-None-Match`)은 `edit` 메시지다. +`pure-serve --module 앱.pmod --store 경로`는 앱의 **보기와 입력이 곧 HTTP 인터페이스**다. 라우터 코드가 없다. + +| 요청 | 뜻 | +|---|---| +| `GET /views/이름` | 보기의 값. 수는 가장 짧은 왕복 10진수, text는 UTF-8 그대로(`text/plain`) | +| `GET /views/표?from=N&count=M` | 표의 행: `{"rows":전체,"from":N,"items":[{열: 값}…]}`(`application/json`). 요청한 행의 칸만 계산한다. 값이 없는 칸은 `null` | +| `POST /inputs/이름` | 이벤트. text 입력은 본문이 text, 키 입력은 누름(down 뒤 up), 메시지는 u32 필드를 쿼리(`?points=3`)로, bytes 필드를 본문으로 받는다. 응답 본문은 핸들러의 `reply`(메시지 필드) | +| `GET`·`PUT /objects/키` | 모듈에 키별 메시지 `show`·`edit`가 있으면 기존 문서 서버(`server.c`)와 같은 계약. `GET`은 `show`, `PUT`(`If-Match`/`If-None-Match`)은 `edit` | + +- **실패는 실패 종류로 답한다.** 이벤트가 실패하면 `X-Pure-Error`에 종류를 싣는다. `not_found` 404, `exists` 409, + `limit`·`divide`·`clamp`·`non_finite`·`encoding` 422, UTF-8이 아닌 text 400. 값이 없는 보기는 404 `no_value`, 없는 이름은 + 404 `no_view`·`no_input`, 모자라거나 모르는 쿼리 필드는 400 `missing_field`·`query`다. +- **라우트는 작업 슬롯에 둔다.** 요청이 가리키는 것(문서, 보기, 입력)과 메시지 필드는 받을 때 풀어 슬롯별 표와 ingress에 + 둔다. 연결이 묶음 전에 끊겨도 받아들인 이벤트는 실행된다. 런타임 코어는 그대로다(키는 문서 키 또는 보기·입력 이름). +- **POST 응답은 요청에서 빌린다.** 코어가 쓰기 결과를 요청 바이트에서 빌리게 하므로 HTTP로 받는 메시지의 핸들러는 + 메시지 필드만 응답할 수 있다(`reply score.name`). 시작할 때 확인한다. 상태에서 계산한 값은 보기로 읽는다. - **연결.** HTTP/1.1 지속 연결과 파이프라이닝(응답은 순서대로), edge-triggered epoll 소유 스레드 하나, 기다리는 연결을 한 번에 받는다. 연결 구조체는 받을 때 할당하고 닫으면 재사용하므로 메모리는 설정한 최대치가 아니라 @@ -251,8 +265,10 @@ text·리스트가 남을 때만 위치를 유지한다. - **런타임 코어.** 요청은 `runtime.c`를 그대로 거친다(연결마다 작업 슬롯, 입력 바이트 예산, 코어가 소유한 ingress, 요청에서 빌린 PUT 응답). 효과는 비동기 인터페이스(`pure_runtime_take`/`complete`)로 소유 스레드에서 처리한다. - **이벤트 한 바퀴가 묶음 하나다.** 준비된 요청을 모두 지연 저장 모드의 앱에 차례로 적용하고, 저장 한 번 **뒤에** - 응답한다. 저장이 실패하면 묶음의 읽기까지 모두 실패한다. 같은 키가 다시 오거나 문서 수가 한 저장의 한도를 넘으려 - 하면 묶음을 먼저 끝낸다. 그래서 묶음 안의 모든 요청은 자기 묶음의 저장으로 저장되거나 그것과 함께 실패한다. + 응답한다. 저장이 실패하면 묶음의 읽기까지 모두 실패한다. 같은 문서 키가 다시 오거나 문서 수가 한 저장의 한도를 + 넘으려 하면 묶음을 먼저 끝낸다. 보기와 앱 상태의 이벤트는 어느 묶음에나 들어간다. 그래서 묶음 안의 모든 요청은 + 자기 묶음의 저장으로 저장되거나 그것과 함께 실패하고, 보기는 저장될 상태만 내보낸다. 한 연결의 요청은 앞 요청에 + 답한 뒤에 읽으므로 쓴 것을 곧 읽는다. - **저장소 핸들.** 문서와 상태를 저장소 핸들(`pure_store_data`)로 열어 파일 매핑 하나를 공유하고, 묶음마다 한 번만 저장소가 바뀌었는지 본다. - **샤드.** `--threads N`은 스레드마다 같은 포트의 리스너(`SO_REUSEPORT`), epoll, 런타임 코어, 앱, 저장소 핸들을 둔다. @@ -266,6 +282,9 @@ text·리스트가 남을 때만 위치를 유지한다. 성능과 비교는 [연구 075](../research/075-http-host-against-axum.md). 검증(aup `make check`): 무작위 요청에 기존 서버와 상태 코드·ETag·오류·본문이 같고, 동시 편집이 커밋을 나누어 쓰며, `kill -9` 뒤에도 확인 응답한 편집이 모두 남는다. +앱 라우트는 키 있는 리스트 앱에 무작위 POST(쿼리 필드와 본문을 가진 메시지, text, 키)와 GET(수, text, 표의 쪽)을 +800번 보내 상태 코드·오류·본문·JSON 행이 파이썬 모델과 같고, 멈췄다 다시 띄워도, 여섯 클라이언트가 보내는 중에 +`kill -9`로 죽여도 확인 응답한 이벤트(약 4,200개)가 모두 남는다. 아직 모듈 앱을 런타임 코어의 효과 단위로 세분하지는 않았다(`runtime.c`의 `definition`은 앱 전체다). 호스트가 직접 호출한다. diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c index 65ca1fb..76f00e1 100644 --- a/prototypes/connections/serve.c +++ b/prototypes/connections/serve.c @@ -1,9 +1,18 @@ #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 + * An app's views and inputs are its HTTP interface, with no routes to write: + * GET /views/NAME answers a view (a number or text as text/plain, a table as + * JSON rows, ?from=N&count=M for a page of them) and POST /inputs/NAME + * delivers an event (a text input's text is the body; a key input is + * pressed, down then up; a message takes its u32 fields from the query and + * its bytes field from the body, and answers what its handler replies, a + * field of the message). A failed event answers with its fault. + * + * A module with keyed messages named show and edit also serves the document + * contract of server.c (GET and PUT /objects/KEY, If-Match or If-None-Match: + * *, ETag = revision, X-Pure-Error): 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 @@ -18,6 +27,7 @@ * whose commits take turns and whose revisions settle concurrent edits of a * key. A module with states (not only keyed documents) runs on one thread. */ #include "module-store.h" +#include "module-text.h" #include "runtime.h" #include "store.h" #include <arpa/inet.h> @@ -44,6 +54,12 @@ enum { HEADER_LIMIT = 4096, EVENTS = 256 }; typedef enum { CLOSED, READING_HEAD, READING_BODY, WAITING, WRITING } Phase; typedef struct Host Host; +/* What a request names: a document, a view (a table's rows from, count) or + * an input. Kept per work slot, so an accepted event runs even if its + * connection is gone before the burst. */ +typedef enum { ROUTE_NONE, ROUTE_OBJECT, ROUTE_VIEW, ROUTE_INPUT } RouteKind; +typedef struct { RouteKind kind; uint32_t target; size_t from, count; } Route; + /* Allocated when accepted and reused after closing: memory follows the * connections that are open, not the configured maximum. */ typedef struct Connection { @@ -52,9 +68,12 @@ typedef struct Connection { size_t index; /* in the host's table */ int fd; Phase phase; - bool readable, writable, keep_alive, put; + bool readable, writable, keep_alive, put, post; uint64_t deadline; - char key[65]; + char key[65]; /* the document's key, or the view's or input's name */ + RouteKind route; + unsigned char fields[4 * PURE_MODULE_FIELDS]; /* a message's u32 fields, from the query */ + size_t field_bytes; int64_t expected; size_t body_size, received, in_used; unsigned char *body; @@ -79,6 +98,9 @@ typedef struct { struct iovec *extents; size_t extent_count; char key[65]; + RouteKind route; + PureFault fault; /* a failed event's */ + const char *type; /* a view's content type */ } Applied; struct Host { @@ -86,6 +108,8 @@ struct Host { PureStore *store; PureApp app; uint32_t show, edit; + bool documents; /* the module serves /objects: it has show and edit */ + Route *routes; /* by work slot */ PureRuntime runtime; PureWork *work; Connection **table; /* open connections by index */ @@ -153,14 +177,14 @@ static void close_connection(Connection *c) { /* Answer with a body of `count` extents (`size` bytes in all); the ticket * keeps them alive. One writev sends as much as the socket takes. */ -static void respond_extents(Connection *c, int status, const char *reason, const char *error, int64_t revision, - const struct iovec *extents, size_t count, size_t size) { +static void respond_typed(Connection *c, int status, const char *reason, const char *error, int64_t revision, + const char *type, const struct iovec *extents, size_t count, size_t size) { 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" + "HTTP/1.1 %d %s\r\nContent-Length: %zu\r\n%sContent-Type: %s\r\n" "Cache-Control: no-store\r\nX-Pure-Error: %s\r\n%s\r\n", - status, reason, size, c->keep_alive ? "" : "Connection: close\r\n", error, tag); + status, reason, size, c->keep_alive ? "" : "Connection: close\r\n", type, error, tag); c->out = count < sizeof(c->small) / sizeof(*c->small) ? c->small : malloc((count + 1) * sizeof(*c->out)); c->phase = WRITING; if (!c->out) { c->keep_alive = false; c->out = c->small; count = 0; } /* cannot send it: the close tells */ @@ -170,12 +194,23 @@ static void respond_extents(Connection *c, int status, const char *reason, const c->out_at = 0; } +static void respond_extents(Connection *c, int status, const char *reason, const char *error, int64_t revision, + const struct iovec *extents, size_t count, size_t size) { + respond_typed(c, status, reason, error, revision, "application/octet-stream", extents, count, size); +} + static void respond(Connection *c, int status, const char *reason, const char *error, int64_t revision, PureSpan body) { struct iovec extent = {(void *)body.data, body.size}; respond_extents(c, status, reason, error, revision, &extent, body.size ? 1 : 0, body.size); } +/* A request refused before it has a ticket; a body it came with is not read. */ +static void reject(Connection *c, int status, const char *reason, const char *error) { + if (c->body_size) c->keep_alive = false; + respond(c, status, reason, error, 0, (PureSpan){0}); +} + /* 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); @@ -201,9 +236,38 @@ static void answer(Connection *c, PureOutcome result) { if (!result.body.size) release_ticket(c); } +static const char *const fault_names[] = {"none", "divide", "non_finite", "clamp", "limit", "not_found", "encoding", + "exists"}; + +/* An app route's answer: a view's value, or what an event replied. A failed + * event names its fault: 404 not_found, 409 exists, 422 for the others; + * a view without a value is 404 no_value. */ +static void answer_app(Connection *c, PureOutcome result, bool view, PureFault fault, const char *type) { + const char *named = fault != PURE_FAULT_NONE ? fault_names[fault] : NULL; + if (result.status == PURE_OK) { + struct iovec extent = {(void *)result.body.data, result.body.size}; + respond_typed(c, 200, "OK", "none", 0, type ? type : "application/octet-stream", &extent, + result.body.size ? 1 : 0, result.body.size); + } else if (result.status == PURE_NOT_FOUND) { + respond(c, 404, "Not Found", view ? "no_value" : named ? named : "not_found", 0, (PureSpan){0}); + } else if (result.status == PURE_CONFLICT) { + respond(c, 409, "Conflict", named ? named : "conflict", 0, (PureSpan){0}); + } else if (named && (result.status == PURE_LIMIT || result.status == PURE_INVALID)) { + respond(c, 422, "Unprocessable Content", named, 0, (PureSpan){0}); + } else if (result.status == PURE_INVALID) { + respond(c, 400, "Bad Request", "invalid", 0, (PureSpan){0}); /* such as a text that is not UTF-8 */ + } else { + answer(c, result); + return; + } + 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) { + * bounded HTTP/1.1 subset as server.c, plus Connection, POST and the app + * routes. A query stays in head for start_request; an unknown path is + * answered 404 once its head is read. */ +static bool parse_head(Connection *c, char *head, size_t size, char **query) { 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; @@ -216,20 +280,33 @@ static bool parse_head(Connection *c, char *head, size_t size) { if (!path) return false; *path++ = 0; c->put = !strcmp(line, "PUT"); - if (!c->put && strcmp(line, "GET")) return false; + c->post = !strcmp(line, "POST"); + if (!c->put && !c->post && 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; + if (strcmp(version, "HTTP/1.1")) return false; + if ((*query = strchr(path, '?'))) *(*query)++ = 0; + static const struct { const char *prefix; RouteKind kind; } routes[] = { + {"/objects/", ROUTE_OBJECT}, {"/views/", ROUTE_VIEW}, {"/inputs/", ROUTE_INPUT}}; + const char *key = ""; + c->route = ROUTE_NONE; + for (size_t i = 0; i < sizeof(routes) / sizeof(*routes) && c->route == ROUTE_NONE; ++i) + if (!strncmp(path, routes[i].prefix, strlen(routes[i].prefix))) { + c->route = routes[i].kind; + key = path + strlen(routes[i].prefix); + } size_t key_size = strlen(key); - if (!key_size || key_size > 64) return false; + if (c->route != ROUTE_NONE && (!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); + /* Each route has its method: documents GET and PUT, views GET, inputs POST. */ + if ((c->route == ROUTE_OBJECT && (c->post || *query)) || (c->route == ROUTE_VIEW && (c->put || c->post)) || + (c->route == ROUTE_INPUT && !c->post)) return false; bool length_seen = false, condition_seen = false, host_seen = false; c->body_size = 0; c->expected = -1; @@ -272,23 +349,84 @@ static bool parse_head(Connection *c, char *head, size_t size) { } 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); + if (c->put) return host_seen && length_seen && condition_seen; + if (c->post) return host_seen && !condition_seen; /* no Content-Length: no body */ + return host_seen && !condition_seen && c->body_size == 0; +} + +/* What an app route names in the module, with a message's u32 fields taken + * from the query into c->fields. NULL, or the error to answer (404 for a + * name the module lacks, 400 otherwise). */ +static const char *resolve(const Host *host, Connection *c, char *query, Route *route, int *status) { + const PureModule *m = host->module; + *status = 404; + c->field_bytes = 0; + if (route->kind == ROUTE_NONE || (route->kind == ROUTE_OBJECT && !host->documents)) return "route"; + if (route->kind == ROUTE_OBJECT) return NULL; + int found = route->kind == ROUTE_VIEW ? pure_module_view(m, c->key) : pure_module_input(m, c->key); + if (found < 0) return route->kind == ROUTE_VIEW ? "no_view" : "no_input"; + *status = 400; + route->target = (uint32_t)found; + route->from = 0; + route->count = SIZE_MAX; + const PureInputShape *shape = route->kind == ROUTE_INPUT ? &m->input[found] : NULL; + bool message = shape && shape->kind == PURE_KIND_MESSAGE; + if (message && shape->keyed) return "keyed"; + if (shape) { + const PureField *last = message && shape->field_count ? &m->field[shape->field_start + shape->field_count - 1] : NULL; + if (c->body_size && shape->kind != PURE_KIND_TEXT && !(last && last->type == PURE_FIELD_BYTES)) return "body"; + c->field_bytes = message ? shape->prefix : 0; + } + /* name=value pairs: a table's from and count, a message's u32 fields. */ + bool given[PURE_MODULE_FIELDS] = {false}; + for (char *pair = query; pair && *pair;) { + char *end = strchr(pair, '&'), *value = strchr(pair, '='); + if (end) *end = 0; + if (!value || (end && value > end)) return "query"; + *value++ = 0; + uint64_t n; + if (!number(value, route->kind == ROUTE_VIEW ? SIZE_MAX : UINT32_MAX, &n)) return "query"; + if (route->kind == ROUTE_VIEW) { + bool from = !strcmp(pair, "from"); + if (m->view_type[found] != PURE_VIEW_TABLE || (!from && strcmp(pair, "count"))) return "query"; + if (from) route->from = (size_t)n; + else route->count = (size_t)n; + } else { + uint32_t f = 0; + while (f < (message ? shape->field_count : 0) && strcmp(m->field[shape->field_start + f].name, pair)) ++f; + const PureField *field = message && f < shape->field_count ? &m->field[shape->field_start + f] : NULL; + if (!field || field->type != PURE_FIELD_U32 || given[f]) return "query"; + given[f] = true; + for (unsigned i = 0; i < 4; ++i) c->fields[field->offset + i] = (unsigned char)(n >> (8 * i)); + } + pair = end ? end + 1 : NULL; + } + for (uint32_t f = 0; message && f < shape->field_count; ++f) + if (m->field[shape->field_start + f].type == PURE_FIELD_U32 && !given[f]) return "missing_field"; + return NULL; } /* 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]; + char head[HEADER_LIMIT], *query = NULL; 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 (!parse_head(c, head, head_size, &query)) { 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); + Route route = {.kind = c->route}; + int status; + const char *why = resolve(host, c, query, &route, &status); + if (why) { reject(c, status, status == 404 ? "Not Found" : "Bad Request", why); return; } + /* A document's PUT and an input's event carry a payload: the message's fields, then the body. */ + bool write = route.kind == ROUTE_INPUT || c->put; + PureResult admitted = pure_runtime_accept(&host->runtime, write ? PURE_INPUT_WRITE : PURE_INPUT_READ, c->key, + route.kind == ROUTE_INPUT ? 0 : c->expected, + write ? c->field_bytes + c->body_size : 0, &c->ticket); if (admitted != PURE_OK) { c->ticket = (PureTicket){0}; /* An unread body would desynchronize the stream. */ @@ -297,8 +435,11 @@ static void start_request(Connection *c, size_t head_size) { return; } host->by_slot[c->ticket.slot] = c; - if (!c->put) { c->phase = WAITING; return; } - c->body = pure_runtime_ingress(c->ticket, NULL); + host->routes[c->ticket.slot] = route; + if (!write) { c->phase = WAITING; return; } + unsigned char *ingress = pure_runtime_ingress(c->ticket, NULL); + memcpy(ingress, c->fields, c->field_bytes); + c->body = ingress + c->field_bytes; 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); @@ -393,7 +534,7 @@ static void release_loan(void *allocation, size_t size) { static bool allocate_ingress(void *context, size_t size, PureResource *out) { (void)context; - void *bytes = malloc(size); + void *bytes = malloc(size ? size : 1); /* a key press or an empty text has no payload */ if (!bytes) return false; *out = (PureResource){.body = {bytes, size}, .allocation = bytes, .allocation_size = size, .release = release_copy}; return true; @@ -426,12 +567,154 @@ static bool copy_reply(void *context, PureSpan span) { return true; } +/* ---- App routes -------------------------------------------------------- */ + +typedef struct { unsigned char *bytes; size_t size, capacity; bool failed; } Output; + +static bool put_bytes(Output *o, const void *bytes, size_t size) { + if (o->failed) return false; + if (size > o->capacity - o->size) { + size_t capacity = o->capacity ? o->capacity : 256; + while (capacity - o->size < size) { + if (capacity > SIZE_MAX / 2) { o->failed = true; return false; } + capacity *= 2; + } + unsigned char *grown = realloc(o->bytes, capacity); + if (!grown) { o->failed = true; return false; } + o->bytes = grown; + o->capacity = capacity; + } + if (size) memcpy(o->bytes + o->size, bytes, size); + o->size += size; + return true; +} + +static bool put_text(Output *o, const char *text) { return put_bytes(o, text, strlen(text)); } + +static bool put_span(void *context, PureSpan span) { return put_bytes(context, span.data, span.size); } + +/* A JSON string's contents: quotes, backslashes and control bytes escaped, + * other bytes (UTF-8, which Pure keeps valid) as they are. */ +static bool put_json_span(void *context, PureSpan span) { + Output *o = context; + const unsigned char *p = span.data; + size_t plain = 0; + for (size_t i = 0; i < span.size; ++i) { + unsigned char b = p[i]; + if (b >= 0x20 && b != '"' && b != '\\') continue; + char escape[8]; + int length = b == '"' || b == '\\' ? snprintf(escape, sizeof(escape), "\\%c", b) : + snprintf(escape, sizeof(escape), "\\u%04x", b); + put_bytes(o, p + plain, i - plain); + put_bytes(o, escape, (size_t)length); + plain = i + 1; + } + return put_bytes(o, p + plain, span.size - plain); +} + +static void put_number(Output *o, double value) { + char digits[32]; + put_bytes(o, digits, pure_format_number(value, digits)); +} + +/* A view's value: a number in Pure's shortest form or a text, as text; a + * table as {"rows":N,"from":F,"items":[{column: value}...]} for the rows + * asked, each cell computed only for them (a cell without a value is null). */ +static PureResult show_view(PureApp *app, const Route *route, size_t available, PureResource *out, const char **type) { + const PureModule *m = app->module; + uint32_t view = route->target; + Output o = {0}; + PureResult status = PURE_OK; + double value; + if (m->view_type[view] == PURE_VIEW_NUM) { + *type = "text/plain; charset=utf-8"; + if (pure_app_view(app, view, &value)) put_number(&o, value); + else status = PURE_NOT_FOUND; + } else if (m->view_type[view] == PURE_VIEW_TEXT) { + *type = "text/plain; charset=utf-8"; + status = pure_app_view_text(app, view, put_span, &o); + } else { + *type = "application/json"; + size_t rows = pure_app_rows(app, view), from = route->from < rows ? route->from : rows; + size_t end = from + (route->count < rows - from ? route->count : rows - from); + char head[80]; + snprintf(head, sizeof(head), "{\"rows\":%zu,\"from\":%zu,\"items\":[", rows, from); + put_text(&o, head); + for (size_t row = from; row < end && !o.failed; ++row) { + put_text(&o, row > from ? ",{" : "{"); + bool first = true; + for (uint32_t column = 0; column < m->views; ++column) { + if (m->view_table[column] != view + 1) continue; + put_text(&o, first ? "\"" : ",\""); + put_text(&o, m->view_names[column]); + put_text(&o, "\":"); + first = false; + if (m->view_type[column] == PURE_VIEW_NUM) { + if (pure_app_cell(app, column, row, &value)) put_number(&o, value); + else put_text(&o, "null"); + continue; + } + size_t mark = o.size; + put_text(&o, "\""); + if (pure_app_cell_text(app, column, row, put_json_span, &o) == PURE_OK) put_text(&o, "\""); + else if (!o.failed) { + o.size = mark; + put_text(&o, "null"); + } + } + put_text(&o, "}"); + } + put_text(&o, "]}"); + } + if (status == PURE_OK && !o.bytes && !(o.bytes = malloc(1))) o.failed = true; /* an empty value has a buffer too */ + if (status == PURE_OK && o.failed) status = PURE_ALLOCATION; + if (status == PURE_OK && o.size > available) status = PURE_BUDGET; + if (status != PURE_OK) { + free(o.bytes); + return status; + } + *out = (PureResource){.body = {o.bytes, o.size}, .allocation = o.bytes, .allocation_size = o.size, + .release = release_copy}; + return PURE_OK; +} + +/* An input's event: a text, a key pressed (down, then up), or a message + * whose reply (a field of it) is borrowed from the request. */ +static PureResult apply_input(PureApp *app, const Route *route, const PureEffect *effect, Applied *out) { + const PureInputShape *shape = &app->module->input[route->target]; + PureResult status; + if (shape->kind == PURE_KIND_TEXT) { + status = pure_app_text(app, route->target, effect->value); + } else if (shape->kind == PURE_KIND_KEY) { + status = pure_app_input(app, route->target, PURE_EDGE_DOWN); + if (status == PURE_OK) status = pure_app_input(app, route->target, PURE_EDGE_UP); + } else { + status = pure_app_message(app, route->target, NULL, -1, effect->value); + Reply reply = {0}; + if (status == PURE_OK) pure_app_reply(app, measure_reply, &reply); + out->reply = (PureSpan){reply.size ? reply.first : NULL, reply.size}; + } + out->fault = app->fault; + return status; +} + 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; + const Route *route = &host->routes[effect->ticket.slot]; + out->route = route->kind; + if (route->kind == ROUTE_VIEW) { + out->status = show_view(app, route, effect->available, &out->resource, &out->type); + return; + } + if (route->kind == ROUTE_INPUT) { + out->status = apply_input(app, route, effect, out); + if (out->status == PURE_OK) ++host->burst_events; + return; + } 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); @@ -485,6 +768,7 @@ static void finish_burst(Host *host, size_t count) { /* A lent answer's body is its extents (the runtime charged them). */ if (result.status == PURE_OK && a->extents) respond_extents(c, 200, "OK", "none", result.revision, a->extents, a->extent_count, result.body.size); + else if (a->route != ROUTE_OBJECT) answer_app(c, result, a->route == ROUTE_VIEW, a->fault, a->type); else answer(c, result); drive(c); /* most answers fit the socket buffer at once */ } @@ -492,9 +776,10 @@ static void finish_burst(Host *host, size_t count) { } } -/* 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. */ +/* Every request made ready since the last burst. A burst ends before a + * document's 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. Views and + * events of the app's states join any burst. */ static void serve_bursts(Host *host) { size_t count = 0, keys = 0; PureEffect effect; @@ -503,24 +788,25 @@ static void serve_bursts(Host *host) { bool pinned = false; while (pure_runtime_take(&host->runtime, &effect) == PURE_TAKE_EFFECT) { if (!pinned) pinned = pure_store_pin(host->store, true) == PURE_OK; - 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) { + bool document = host->routes[effect.ticket.slot].kind == ROUTE_OBJECT, repeated = false; + for (size_t i = 0; i < count && document && !repeated; ++i) + repeated = host->applied[i].route == ROUTE_OBJECT && !strcmp(host->applied[i].key, effect.key); + if (repeated || (document && keys == PURE_APP_PENDING)) { finish_burst(host, count); count = keys = 0; } apply_effect(host, &effect, &host->applied[count++]); - ++keys; + keys += document; } finish_burst(host, count); if (pinned) pure_store_pin(host->store, false); } -/* 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) { +/* A write's reply is borrowed from its request, so the handler of an input + * served by PUT or POST may reply only a field of its message. */ +static bool replies_borrow(const PureModule *m, uint32_t input) { for (uint32_t h = 0; h < m->handlers; ++h) { - if (m->handler[h].event != 1 + 2 * edit) continue; + if (m->handler[h].event != 1 + 2 * input) 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)) @@ -531,7 +817,7 @@ static bool edit_replies_borrow(const PureModule *m, uint32_t edit) { /* ---- Main ------------------------------------------------------------- */ -static bool load_module(PureModule *m, const char *path, uint32_t *show_out, uint32_t *edit_out) { +static bool load_module(PureModule *m, const char *path, bool *documents, uint32_t *show_out, uint32_t *edit_out) { static unsigned char bytes[1 << 17]; char error[128]; FILE *file = fopen(path, "rb"); @@ -542,18 +828,23 @@ static bool load_module(PureModule *m, const char *path, uint32_t *show_out, uin fprintf(stderr, "Module: %s\n", read_ok ? error : "cannot read"); return false; } + /* Documents are served when the module has show and edit, as keyed messages. */ 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); + *documents = show >= 0 || edit >= 0; + if (*documents && (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: documents need 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; + for (uint32_t i = 0; i < m->inputs; ++i) { + bool served = (*documents && i == (uint32_t)edit) || (m->input[i].kind == PURE_KIND_MESSAGE && !m->input[i].keyed); + if (served && !replies_borrow(m, i)) { + fprintf(stderr, "Module: %s may reply only a field of its message\n", m->input_names[i]); + return false; + } } - *show_out = (uint32_t)show; - *edit_out = (uint32_t)edit; + *show_out = (uint32_t)(show >= 0 ? show : 0); + *edit_out = (uint32_t)(edit >= 0 ? edit : 0); return true; } @@ -596,9 +887,10 @@ static void accept_all(Host *host) { /* One shard: its own listener on the shared port, epoll set, runtime domain, * app and store handle. */ -static bool start_shard(Host *host, const PureModule *module, uint32_t show, uint32_t edit, const char *store, - size_t budget, unsigned *port) { +static bool start_shard(Host *host, const PureModule *module, bool documents, uint32_t show, uint32_t edit, + const char *store, size_t budget, unsigned *port) { host->module = module; + host->documents = documents; host->show = show; host->edit = edit; if (pure_store_open(store, &host->store) != PURE_OK) { fputs("Cannot open the store\n", stderr); return false; } @@ -610,7 +902,11 @@ static bool start_shard(Host *host, const PureModule *module, uint32_t show, uin 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 false; } + host->routes = calloc(host->capacity, sizeof(*host->routes)); + if (!host->work || !host->table || !host->by_slot || !host->applied || !host->routes) { + fputs("Cannot allocate connections\n", stderr); + return false; + } if (!pure_runtime_init(&host->runtime, host->work, host->capacity, &host->app, (PureHost){.allocate = allocate_ingress}, budget) || !pure_runtime_enable_async(&host->runtime)) return false; @@ -690,6 +986,7 @@ static void stop_shard(Host *host) { free(host->table); free(host->by_slot); free(host->applied); + free(host->routes); } int main(int argc, char **argv) { @@ -711,7 +1008,8 @@ int main(int argc, char **argv) { else goto usage; } uint32_t show, edit; - if (!module_path || !store || !load_module(&module, module_path, &show, &edit)) goto usage; + bool documents; + if (!module_path || !store || !load_module(&module, module_path, &documents, &show, &edit)) goto usage; if (threads > 1 && module.states) { fputs("Module: its states have one owner, so it runs with --threads 1\n", stderr); return 2; @@ -729,7 +1027,7 @@ int main(int argc, char **argv) { hosts[i].capacity = capacity; hosts[i].timeout_ms = timeout_ms; hosts[i].listener = hosts[i].epoll = -1; - if (!start_shard(&hosts[i], &module, show, edit, store, budget, &port)) return 1; + if (!start_shard(&hosts[i], &module, documents, show, edit, store, budget, &port)) return 1; } printf("{\"port\":%u,\"threads\":%u}\n", port, threads); fflush(stdout); |
