diff options
| -rw-r--r-- | core/runtime.c | 29 | ||||
| -rw-r--r-- | core/runtime.h | 4 | ||||
| -rw-r--r-- | docs/design/008-module-format.md | 34 | ||||
| -rw-r--r-- | serve/deliver.c | 24 | ||||
| -rw-r--r-- | serve/message-json.c | 139 | ||||
| -rw-r--r-- | serve/message-json.h | 8 | ||||
| -rw-r--r-- | serve/serve.c | 55 | ||||
| -rw-r--r-- | tests/check.py | 158 |
8 files changed, 421 insertions, 30 deletions
diff --git a/core/runtime.c b/core/runtime.c index 4a857c7..19ff215 100644 --- a/core/runtime.c +++ b/core/runtime.c @@ -201,6 +201,35 @@ unsigned char *pure_runtime_ingress(PureTicket ticket, size_t *size) { return (unsigned char *)w->resource.body.data; } +PureResult pure_runtime_grow(PureTicket ticket, size_t size) { + PureWork *w = lookup(ticket, true); + PureRuntime *r = ticket.runtime; + if (!w || r->running || w->state != RECEIVING || w->input != PURE_INPUT_WRITE || w->parent.runtime) + return PURE_INVALID; + size_t had = w->resource.body.size; + if (size <= had) return PURE_OK; + if (size - had > r->budget - r->retained) { + METRIC(++r->budget_rejections); + return PURE_BUDGET; + } + if (!r->host.allocate) return PURE_ALLOCATION; + PureResource grown = {0}; + r->running = true; + bool ok = r->host.allocate(r->host.context, size, &grown); + r->running = false; + if (!ok || !grown.body.data || grown.body.size != size || !grown.release) { + dispose(r, &grown); + return PURE_ALLOCATION; + } + if (had) memcpy((unsigned char *)grown.body.data, w->resource.body.data, had); + dispose(r, &w->resource); + w->resource = grown; + w->charged += size - had; + r->retained += size - had; + METRIC(if (r->retained > r->peak_retained) r->peak_retained = r->retained); + return PURE_OK; +} + bool pure_runtime_step(PureRuntime *r) { if (!r || r->running || r->ready_head == NONE) return false; #ifdef PURE_ASYNC_EFFECTS diff --git a/core/runtime.h b/core/runtime.h index 70f5d08..fd19525 100644 --- a/core/runtime.h +++ b/core/runtime.h @@ -110,6 +110,10 @@ bool pure_runtime_bind(PureRuntime *, PureWork *, size_t capacity); PureResult pure_runtime_accept(PureRuntime *, PureInput, const char *key, int64_t expected, size_t input_size, PureTicket *out); unsigned char *pure_runtime_ingress(PureTicket, size_t *size); +/* A WRITE input still being filled grows to size with its bytes kept (the + * ingress moves: take it again), charged to the budget as its admission was; + * BUDGET or ALLOCATION leave it as it was. */ +PureResult pure_runtime_grow(PureTicket, size_t size); bool pure_runtime_ready(PureTicket); bool pure_runtime_step(PureRuntime *); /* execute at most one ready job */ bool pure_runtime_result(PureTicket, PureOutcome *out); diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md index b70b8f0..4b9855c 100644 --- a/docs/design/008-module-format.md +++ b/docs/design/008-module-format.md @@ -13,8 +13,9 @@ v3은 **text 상태와 커서**, text 상수, text 입력, text 보기를 더했 v4는 **레코드**, 레코드의 **리스트** 상태, 리스트의 **인덱스** 상태를 더했다. v5는 값을 가진 채 합류하는 분기(조건식)와, 리스트 항목마다 행을 내는 **표 보기**, 한 번 계산한 값을 여러 번 쓰게 하는 스택 복사(PICK·SLIDE, 툴체인의 함수 인라인용)를 더한다. 이전 판은 받지 않는다(생산자는 aup 하나다). -v5에는 뒤에 집계(FOLD), 걸러진 표, 마크, 그리고 **키 있는 리스트**와 text 비교가 더해졌다. 모두 전의 v5 모듈이 그대로 -유효하도록 더한 것이다(새 명령과 이전에 0이던 자리의 값). +v5에는 뒤에 집계(FOLD), 걸러진 표, 마크, 그리고 **키 있는 리스트**와 text 비교가 더해졌다. 그 뒤 핸들러에서 리스트를 +도는 **걷기**(EACH)와 **메시지의 행**(레코드의 리스트인 메시지 필드와, 그 행을 담는 행 목록 상태)이 더해졌다. 모두 전의 +v5 모듈이 그대로 유효하도록 더한 것이다(새 명령과 이전에 0이던 자리의 값). ## 바이트 배치 @@ -27,10 +28,10 @@ v5에는 뒤에 집계(FOLD), 걸러진 표, 마크, 그리고 **키 있는 리� | 16 | u32 명령 수, u32 문자열 바이트 수 | | 24 | u16 필드, u16 컬렉션, u16 상수, u16 레코드 | | 32 | u32 상수 블롭 바이트 수, u16 레코드 필드, u16 0 | -| 40 | 상태 × 16 B: u32 이름, u32 타입(0 수, 1 text, 2 커서, 3 리스트, 4 인덱스), 8 B 값: 수는 f64 초기값, text는 u32 초기 상수, 커서는 u32 대상 text 상태와 u32 플래그(1이면 마크: 그 자리에 넣은 글 앞에 남는다), 리스트는 u32 레코드와 u32 키(0이면 없음, 아니면 키 필드 + 1), 인덱스는 u32 대상 리스트 상태 | +| 40 | 상태 × 16 B: u32 이름, u32 타입(0 수, 1 text, 2 커서, 3 리스트, 4 인덱스), 8 B 값: 수는 f64 초기값, text는 u32 초기 상수, 커서는 u32 대상 text 상태와 u32 플래그(1이면 마크: 그 자리에 넣은 글 앞에 남는다), 리스트는 u32 레코드와 u32 키(0이면 없음, 아니면 키 필드 + 1, 256이면 메시지의 행 목록: 이름은 그 필드의 경로 `입력.필드`이고 키가 없다), 인덱스는 u32 대상 리스트 상태 | | | 컬렉션 × 12 B: u32 이름, u32 1(바이트), u32 1(키별) | | | 입력 × 16 B: u32 이름, u8 종류(1 키, 2 메시지, 3 text, 4 타이머, 5 출력: 앱이 보내는 메시지, 모양은 메시지와 같고 핸들러가 없다), u8 키별 여부, u16 0, u32 키 이름(키 입력)·주기(타이머, 1 ms–24시간을 ms로) 또는 0, u16 첫 필드, u16 필드 수 | -| | 필드 × 8 B: u32 이름(식별자, 또는 그룹 안의 필드면 점으로 이은 경로 `user.name`), u32 타입(1 u32le, 2 바이트, 3 수(binary64), 4 text). 페이로드는 접두부와 가변 필드의 바이트다. 접두부에는 선언 순서로 고정 필드의 값과 가변 필드(바이트·text)의 u32 길이가 들어가는데, 마지막 가변 필드는 길이 없이 나머지를 차지한다. 가변 필드의 바이트는 접두부 뒤에 선언 순서로 이어진다(가변 필드가 하나면 예전의 "바이트 꼬리"와 같다) | +| | 필드 × 8 B: u32 이름(식별자, 또는 그룹 안의 필드면 점으로 이은 경로 `user.name`), u32 타입(1 u32le, 2 바이트, 3 수(binary64), 4 text, 5 행: 비트 8–23에 그 행이 채우는 행 목록 상태). 페이로드는 접두부와 가변 필드의 바이트다. 접두부에는 선언 순서로 고정 필드의 값과 가변 필드(바이트·text)의 u32 길이가 들어가는데, 마지막 가변 필드는 길이 없이 나머지를 차지한다. 가변 필드의 바이트는 접두부 뒤에 선언 순서로 이어진다(가변 필드가 하나면 예전의 "바이트 꼬리"와 같다) | | | 상수 × 8 B: u32 블롭 오프셋, u32 길이 (UTF-8) | | | 레코드 × 8 B: u32 이름, u16 첫 레코드 필드, u16 필드 수(1–16) | | | 레코드 필드 × 16 B: u32 이름, u32 타입(0 수, 1 text), 8 B 기본값: 수는 f64, text는 u32 상수 | @@ -106,6 +107,8 @@ v5에는 뒤에 집계(FOLD), 걸러진 표, 마크, 그리고 **키 있는 리� | 70 | FLOOR | n → n | 내림 | | 71 | MOD | n n → n | 나눗셈을 내림한 나머지. 결과는 나누는 수의 부호를 따른다(0도). 0으로 나누면 `divide` | | 72 | SEND | f… → | 출력 `인자`의 필드 값을 필드 순서대로 받아(u32·수는 n, 바이트는 b, text는 t) 메시지 페이로드(접두부와 가변 필드)로 만들고 이벤트의 일부로 보낸다. u32 필드의 값이 0–2³²−1의 정수가 아니면 `limit`. 핸들러 전용 | +| 73 | OUTER_ROW | → i | 열 안의 집계 블록에서 그 열의 행(집계의 항목은 ROW). 걷기 안의 집계에서는 걷는 항목 | +| 74 | EACH | | 걷기. `인자`의 비트 8–15는 리스트 상태, 16–31은 뒤따르는 블록의 길이 L. 그 L개 명령이 리스트가 시작할 때 가진 항목마다 차례로 핸들러의 코드로 돈다(`ROW`가 항목). 블록은 상태, 그 리스트의 항목(필드 바꾸기, 지우기), 다른 리스트를 바꿀 수 있지만 걷는 리스트에 넣을 수는 없다. 걷는 항목이나 그 앞의 항목을 지우면 걷기가 따라가 다음 항목을 건너뛰지 않는다. 핸들러 전용 | `f`는 레코드 필드의 타입(수 또는 text)이고, `k`는 그 리스트의 키 필드 타입이다. 수 연산 결과가 유한하지 않으면 실패한다. 실패하면 그 이벤트의 모든 핸들러가 한 일이 취소된다. @@ -139,6 +142,12 @@ v5에는 뒤에 집계(FOLD), 걸러진 표, 마크, 그리고 **키 있는 리� 리스트만 가리키고 키 필드 타입의 값을 받는다. FIND의 결과는 그 리스트의 인덱스다. - 걸러진 표의 필터 블록은 그 리스트의 열로 검증한다. 읽기만 하고 수 하나를 남기며, 안에 집계를 둘 수 없다. 타입 다음 바이트는 표에만 쓸 수 있다. +- 걷기(EACH)의 블록은 그 핸들러의 코드로 따로 검증하되 `ROW`(걷는 리스트의 항목)를 읽을 수 있다. 값을 남기지 않고, + 걷는 리스트에 넣지(LIST_APPEND·LIST_INSERT) 못하며, 안에 다시 걷기를 둘 수 없고(비용 상한), 응답(REPLY)하지 않는다(응답은 + 이벤트에 하나다). 안의 집계는 걷는 항목을 OUTER_ROW로 읽는다. 둘레의 블록은 EACH와 그 블록을 명령 하나로 보고, 앞의 분기가 + 블록 안으로 들어오지 못한다. 걷는 리스트는 줄기만 하므로 한 걷기는 시작할 때의 항목 수 × 블록 길이를 넘지 않는다. +- 행 필드(타입 5)는 메시지 입력에만 있고(출력은 행을 보내지 않는다), 가리키는 상태는 행 목록이며 그 이름이 `입력 이름.필드 + 경로`와 같다. 그래서 행 목록 하나는 필드 하나가 채운다. FIELD는 행 필드를 읽지 않는다(행은 리스트로 읽는다). - 표는 리스트 상태를 가리키고 열은 앞에 나온 표를 가리킨다. 표 밖의 이름은 한 이름공간이고, 열 이름은 표 안에서만 유일하다. @@ -158,6 +167,18 @@ v5에는 뒤에 집계(FOLD), 걸러진 표, 마크, 그리고 **키 있는 리� 이벤트 동안 리스트 변경은 되돌리기 기록에 쌓이고, 실패하면 거꾸로 되돌리며 성공하면 지운 항목과 바뀐 text를 해제한다. 항목의 text 필드를 읽으면 이벤트 공간에 복사한다. +**메시지의 행.** 행 필드는 가변 필드이고, 바이트는 그 행 목록의 저장 형식(항목마다 필드 순서대로 수는 binary64, text는 +u32 길이와 바이트)이다. 메시지가 도착하면 핸들러가 돌기 전에 행 목록을 그 행들로 채우고, 유한하지 않은 수, UTF-8이 아닌 +text, 끝이 맞지 않는 바이트는 메시지를 `limit`으로 실패시킨다. 행 목록은 모든 이벤트가 끝날 때(성공이든 실패든) 비워지고, +저장되지 않으며, 그 변경은 되돌리기 기록에 쌓이지 않는다(비울 것이므로). 그래서 핸들러는 메시지의 행을 보통의 리스트로 +읽고(걷기, 집계, `count`), 한 이벤트 안의 임시 표로 쓸 수도 있다. 한 메시지의 모든 행은 한 트랜잭션이다. 행 하나가 실패하면 +(없는 키, `require`) 어느 행도 적용되지 않는다. + +**걷기.** EACH는 걷는 항목의 위치를 인덱스처럼 들고 블록을 항목마다 돌린다. 블록이 걷는 리스트에서 항목을 지우면(LIST_REMOVE) +그 위치가 걷는 항목 앞이면 한 칸 당기고, 걷는 항목 자신이면 다음 차례에 같은 위치(곧 다음 항목)를 돈다. 블록은 걷는 리스트에 +넣을 수 없으므로 걷기는 시작할 때 있던 항목 가운데 차례가 올 때까지 남은 것을 한 번씩, 순서대로 돈다. 블록이 실패하면 이벤트가 +실패하고 모든 변경이 되돌려진다. SQL의 `UPDATE … WHERE`·`DELETE … WHERE`와 한 요청의 여러 행 넣기가 한 이벤트가 된다. + **키 있는 리스트**는 항목을 키 순서(수는 값, text는 바이트)로 두고 키마다 항목 하나만 둔다. 맵·집합·정렬된 맵을 따로 두지 않는다. 리스트 하나가 순서의 규칙만 달리 가진다. 그래서 표 보기, 걸러진 표, 집계, 인덱스, 저장, 모듈 교체가 모두 그대로다. 넣기(LIST_APPEND)는 반씩 나눠 찾은 자리에 넣고, 같은 키가 있으면 이벤트가 `exists`로 실패한다(호스트에는 @@ -304,6 +325,11 @@ text·리스트가 남을 때만 위치를 유지한다. 쿼리와 본문에 함께 오거나(`json_field`), 타입이 다르거나(필드 자리에 그룹, 그룹 자리에 값, 배열, `true`·`false`·`null`: `json_type`), JSON이 아니면(`json`) 400이고 요청은 실행되지 않는다. 256 KiB 본문의 요청이 CPU 0.25 ms다 ([082](../research/082-json-in-place.md)). +- **행은 객체의 배열이다.** 행 필드의 값은 레코드 필드를 이름으로 가진 객체들의 배열이다(수는 수 필드, 문자열은 text 필드, + 빠진 필드는 기본값). 연결 스레드가 행을 저장 형식으로 옮겨(수는 JSON보다 길 수 있다) 요청의 입력을 그만큼 늘리고 + (`pure_runtime_grow`: 늘린 만큼 예산에 계상, 모자라면 503 `payload_budget`) 본문 뒤에 둔다. 가변 필드가 행 필드 하나뿐인 + 메시지는 JSON이 아닌 본문을 그 행의 저장 형식 그대로 받는다(대량 넣기: 3만 행 JSON 한 요청이 초당 약 24만 행). `--reply`의 + 응답 JSON도 같은 방식으로 행을 준다(없으면 행 없음). 행이 든 메시지는 전달 묶음(`X-Pure-Count`)에 들어가지 않는다. - **라우트는 작업 슬롯에 둔다.** 요청이 가리키는 것(문서, 보기, 입력)과 메시지 필드는 받을 때 풀어 슬롯별 표와 ingress에 둔다. 연결이 묶음 전에 끊겨도 받아들인 이벤트는 실행된다. 런타임 코어는 그대로다(키는 문서 키 또는 보기·입력 이름). - **시간.** 요청의 이벤트는 적용되는 순간의 벽시계를 `now`로 본다. 앱을 돌리는 스레드(한 스레드면 그 스레드, 아니면 diff --git a/serve/deliver.c b/serve/deliver.c index 36b39f4..05e7f21 100644 --- a/serve/deliver.c +++ b/serve/deliver.c @@ -482,15 +482,37 @@ static bool form_answer(const PureWire *w, int status, const Text *body, const P unsigned char *prefix = bytes + table, *text = prefix + in->prefix, *copy = text + body->size; if (body->size) memcpy(text, body->bytes, body->size); if (p->sent_size) memcpy(copy, p->sent, p->sent_size); - PureJsonMessage message = {.module = m, .shape = in, .payload = prefix, .table = bytes, .lenient = true}; + /* Rows the answer gives go after the send's copy. */ + PureJsonMessage message = {.module = m, .shape = in, .payload = prefix, .table = bytes, .lenient = true, + .rows_base = in->prefix + body->size + p->sent_size}; size_t at = pure_json_space(text, 0, body->size); if (at < body->size && text[at] == '{' && pure_json_message(&message, text, body->size, &at)) { + free(message.rows); free(bytes); return false; } + if (message.rows_size) { + unsigned char *grown = realloc(bytes, total + message.rows_size); + if (!grown) { + free(message.rows); + free(bytes); + return false; + } + memcpy(grown + total, message.rows, message.rows_size); + bytes = grown; + prefix = bytes + table; + copy = prefix + in->prefix + body->size; + total += message.rows_size; + } + free(message.rows); for (uint32_t f = 0; f < in->field_count; ++f) { if (message.seen[f]) continue; const PureField *field = &m->field[in->field_start + f]; + if (field->type == PURE_FIELD_ROWS) { /* none given: none */ + write_u32(bytes + 8 * f, 0); + write_u32(bytes + 8 * f + 4, 0); + continue; + } if (field->type == PURE_FIELD_U32 && !strcmp(field->name, "http_status")) { write_u32(prefix + field->offset, (uint32_t)status); continue; diff --git a/serve/message-json.c b/serve/message-json.c index 66c812d..8c288ce 100644 --- a/serve/message-json.c +++ b/serve/message-json.c @@ -42,22 +42,33 @@ static size_t json_number(const unsigned char *p, size_t size, bool *integer) { return i; } -/* A fixed field's value from its text (a query value or a JSON number) into - * the prefix: a u32 in range, or a finite number. */ -bool pure_json_fixed(const PureField *field, const unsigned char *text, size_t size, unsigned char *prefix) { +/* A JSON number's finite value. */ +static bool json_double(const unsigned char *text, size_t size, double *value) { bool integer; if (!size || json_number(text, size, &integer) != size || size > 40) return false; char digits[48]; memcpy(digits, text, size); digits[size] = 0; + *value = strtod(digits, NULL); + return isfinite(*value); +} + +/* A fixed field's value from its text (a query value or a JSON number) into + * the prefix: a u32 in range, or a finite number. */ +bool pure_json_fixed(const PureField *field, const unsigned char *text, size_t size, unsigned char *prefix) { + bool integer; + if (!size || json_number(text, size, &integer) != size || size > 40) return false; if (field->type == PURE_FIELD_U32) { + char digits[48]; uint64_t n; + memcpy(digits, text, size); + digits[size] = 0; if (!integer || !decimal(digits, UINT32_MAX, &n)) return false; write_u32(prefix + field->offset, (uint32_t)n); return true; } - double value = strtod(digits, NULL); - if (!isfinite(value)) return false; + double value; + if (!json_double(text, size, &value)) return false; memcpy(prefix + field->offset, &value, 8); return true; } @@ -203,6 +214,116 @@ static uint32_t field_named(const PureJsonMessage *b, const char *path, size_t p return b->shape->field_count; } +/* Room for more bytes of rows. */ +static bool rows_room(PureJsonMessage *b, size_t more) { + if (more <= b->rows_capacity - b->rows_size) return true; + size_t capacity = b->rows_capacity ? b->rows_capacity : 4096; + while (capacity - b->rows_size < more) { + if (capacity > SIZE_MAX / 2) return false; + capacity *= 2; + } + unsigned char *grown = realloc(b->rows, capacity); + if (!grown) return false; + b->rows = grown; + b->rows_capacity = capacity; + return true; +} + +/* A rows field's array (at its opening bracket): each an object of its + * record's fields by name, a number or a string, a field left out taking its + * default; encoded one after another as the list's stored form (a number's + * binary64 bits, a text's u32le length and bytes). */ +static const char *rows_of(PureJsonMessage *b, unsigned char *json, size_t size, size_t *at, uint32_t f, unsigned depth) { + const PureModule *m = b->module; + const PureField *field = &m->field[b->shape->field_start + f]; + const PureRecord *record = &m->record[m->state_ref[field->rows]]; + size_t start = b->rows_size; + if (*at == size || json[(*at)++] != '[') return "json_type"; + *at = pure_json_space(json, *at, size); + for (bool first = true; *at < size && json[*at] != ']'; first = false) { + if (!first) { + if (json[(*at)++] != ',') return "json"; + *at = pure_json_space(json, *at, size); + } + if (*at == size || json[(*at)++] != '{') return "json_type"; + double numbers[PURE_RECORD_FIELDS]; + PureSpan texts[PURE_RECORD_FIELDS]; + bool given[PURE_RECORD_FIELDS] = {false}; + *at = pure_json_space(json, *at, size); + for (bool member = true; *at < size && json[*at] != '}'; member = false) { + if (!member) { + if (json[(*at)++] != ',') return "json"; + *at = pure_json_space(json, *at, size); + } + size_t key_end = *at < size && json[*at] == '"' ? json_string_end(json + *at + 1, size - *at - 1) : 0; + if (!key_end) return "json"; + const unsigned char *key = json + *at + 1; + size_t key_size = key_end - 1; + *at = pure_json_space(json, *at + 1 + key_end, size); + if (*at == size || json[(*at)++] != ':') return "json"; + *at = pure_json_space(json, *at, size); + unsigned k = 0; + while (k < record->count && (strlen(m->record_field_names[record->first + k]) != key_size || + memcmp(m->record_field_names[record->first + k], key, key_size))) + ++k; + if (k == record->count) { + if (!b->lenient) return "json_field"; + if (!skip_value(json, size, at, depth + 1)) return "json"; + *at = pure_json_space(json, *at, size); + continue; + } + if (given[k]) return "json_field"; + given[k] = true; + bool text = m->record_field_text[record->first + k]; + if (*at < size && json[*at] == '"') { + size_t end = json_string_end(json + *at + 1, size - *at - 1), written; + if (!end) return "json"; + if (!text) return "json_type"; + unsigned char *value = json + *at + 1; + if (!memchr(value, '\\', end - 1)) written = end - 1; + else if (!json_unescape(value, end - 1, value, &written)) return "json"; + if (!pure_utf8_valid((PureSpan){value, written})) return "json"; + texts[k] = (PureSpan){value, written}; + *at = pure_json_space(json, *at + 1 + end, size); + continue; + } + bool integer; + size_t length = json_number(json + *at, size - *at, &integer); + if (!length) return *at < size && (json[*at] == '[' || json[*at] == '{' || json[*at] == 't' || json[*at] == 'f' || + json[*at] == 'n') ? "json_type" : "json"; + if (text || !json_double(json + *at, length, &numbers[k])) return "json_type"; + *at = pure_json_space(json, *at + length, size); + } + if (*at == size) return "json"; + ++*at; + for (unsigned k = 0; k < record->count; ++k) { + uint32_t at_field = record->first + k; + if (m->record_field_text[at_field]) { + uint32_t constant = (uint32_t)m->record_field_default[at_field]; + PureSpan text = given[k] ? texts[k] + : (PureSpan){m->constant_data + m->constant_offset[constant], m->constant_length[constant]}; + if (!rows_room(b, 4 + text.size)) return "memory"; + write_u32(b->rows + b->rows_size, (uint32_t)text.size); + if (text.size) memcpy(b->rows + b->rows_size + 4, text.data, text.size); + b->rows_size += 4 + text.size; + } else { + double value = given[k] ? numbers[k] : m->record_field_default[at_field]; + uint64_t bits; + memcpy(&bits, &value, 8); + if (!rows_room(b, 8)) return "memory"; + for (unsigned i = 0; i < 8; ++i) b->rows[b->rows_size + i] = (unsigned char)(bits >> (8 * i)); + b->rows_size += 8; + } + } + } + if (*at == size) return "json"; + ++*at; + if (b->rows_base + b->rows_size > UINT32_MAX >> 2) return "limit"; + write_u32(b->table + 8 * f, (uint32_t)(b->rows_base + start)); + write_u32(b->table + 8 * f + 4, (uint32_t)(b->rows_size - start)); + return NULL; +} + /* An object's members from json[*at] (its opening brace), in the group named * path: a field's value, or a group's object. Strings stay where they are * (an escaped one is unescaped over itself, which never makes it longer). @@ -226,6 +347,14 @@ static const char *members(PureJsonMessage *b, unsigned char *json, size_t size, *at = pure_json_space(json, *at, size); bool group; uint32_t f = field_named(b, path, path_size, key, key_size, &group); + if (f < b->shape->field_count && m->field[b->shape->field_start + f].type == PURE_FIELD_ROWS) { + if (b->seen[f] || (b->given && b->given[f])) return "json_field"; + b->seen[f] = true; + const char *why = rows_of(b, json, size, at, f, depth); + if (why) return why; + *at = pure_json_space(json, *at, size); + continue; + } if (f == b->shape->field_count && !group && b->lenient) { /* not the app's: skipped */ if (!skip_value(json, size, at, depth)) return "json"; *at = pure_json_space(json, *at, size); diff --git a/serve/message-json.h b/serve/message-json.h index 61211d6..c5332b4 100644 --- a/serve/message-json.h +++ b/serve/message-json.h @@ -14,10 +14,16 @@ typedef struct { const bool *given; /* fields given elsewhere (a request's query), or NULL */ bool lenient; /* a member naming no field is skipped, value and all (an API's answer) */ bool seen[PURE_MODULE_FIELDS]; + /* A rows field's array of objects is encoded in the stored form of its + * list into rows (grown with malloc: the caller frees it), which the + * caller places from rows_base on in the payload, as the table says. */ + unsigned char *rows; + size_t rows_size, rows_capacity, rows_base; } PureJsonMessage; /* The object at json[*at] (its opening brace), *at moved past it: NULL, or - * what is wrong ("json", "json_field": unknown or given twice, "json_type"). */ + * what is wrong ("json", "json_field": unknown or given twice, "json_type", + * "limit": rows past what a payload holds, "memory"). */ const char *pure_json_message(PureJsonMessage *, unsigned char *json, size_t size, size_t *at); size_t pure_json_space(const unsigned char *json, size_t at, size_t size); /* A fixed field's value from its text (a query value or a JSON number) into diff --git a/serve/serve.c b/serve/serve.c index 588feec..7fb3768 100644 --- a/serve/serve.c +++ b/serve/serve.c @@ -643,12 +643,14 @@ static const char *resolve(const Host *host, Connection *c, char *query, Route * if (*c->source && !message) return "delivery"; if (!host->owner) c->source[0] = 0; /* an app without states or outputs applies it as a message */ if (shape) { - /* A body is a text input's text, a message's one variable field, or a - * JSON object of a message's fields (two variable fields need one). */ + /* A body is a text input's text, a message's one variable field (rows + * in their list's stored form), or a JSON object of a message's fields + * (two variable fields need one). */ unsigned variables = 0; for (uint32_t f = 0; message && f < shape->field_count; ++f) variables += m->field[shape->field_start + f].type == PURE_FIELD_BYTES || - m->field[shape->field_start + f].type == PURE_FIELD_TEXT; + m->field[shape->field_start + f].type == PURE_FIELD_TEXT || + m->field[shape->field_start + f].type == PURE_FIELD_ROWS; if (c->json && !message) return "json"; if (!c->json && (variables > 1 || (c->body_size && shape->kind == PURE_KIND_KEY) || (c->body_size && message && !variables))) return variables > 1 ? "json" : "body"; @@ -706,31 +708,45 @@ enum { DELIVERY = 80 }; static const char *convert(Connection *c) { const PureInputShape *shape = &c->host->module->input[c->input]; uint32_t delivered = get_u32(c->ingress) & DELIVERED; - size_t table = 8 * (size_t)shape->field_count; + size_t table = 8 * (size_t)shape->field_count, payload = c->batch * c->field_bytes + c->body_size; unsigned char *tables = c->ingress + 4 + (delivered ? DELIVERY : 0), *prefixes = tables + c->batch * table; unsigned char *json = c->body; size_t size = c->body_size, at = pure_json_space(json, 0, size); + /* A message's rows go after the body: they are encoded, then placed. */ + PureJsonMessage b = {.module = c->host->module, .shape = shape, .given = c->given, .rows_base = payload}; + const char *why = NULL; /* A batch is an array of its messages' objects. */ if (c->batch > 1) { if (at == size || json[at++] != '[') return "json"; at = pure_json_space(json, at, size); } - for (unsigned i = 0; i < c->batch; ++i) { - if (i) { - if (at == size || json[at++] != ',') return "json"; - at = pure_json_space(json, at, size); - } - PureJsonMessage b = {.module = c->host->module, .shape = shape, .table = tables + i * table, - .payload = prefixes + i * c->field_bytes, .given = c->given}; - const char *why = pure_json_message(&b, json, size, &at); - if (why) return why; + for (unsigned i = 0; i < c->batch && !why; ++i) { + if (i && (at == size || json[at++] != ',')) why = "json"; + at = pure_json_space(json, at, size); + b.table = tables + i * table; + b.payload = prefixes + i * c->field_bytes; + memset(b.seen, 0, sizeof(b.seen)); + if (!why) why = pure_json_message(&b, json, size, &at); at = pure_json_space(json, at, size); - for (uint32_t f = 0; f < shape->field_count; ++f) - if (!b.seen[f] && !c->given[f]) return "missing_field"; + for (uint32_t f = 0; !why && f < shape->field_count; ++f) + if (!b.seen[f] && !c->given[f]) why = "missing_field"; + if (!why && c->batch > 1 && b.rows_size) why = "delivery"; /* an output sends no rows */ + } + if (!why && c->batch > 1 && (at == size || json[at++] != ']')) why = "json"; + if (!why && pure_json_space(json, at, size) != size) why = "json"; + if (!why && b.rows_size) { + size_t body = (size_t)(c->body - c->ingress); + PureResult grown = pure_runtime_grow(c->ticket, body + c->body_size + b.rows_size); + if (grown != PURE_OK) why = grown == PURE_BUDGET ? "payload_budget" : "memory"; + else { + c->ingress = pure_runtime_ingress(c->ticket, NULL); + c->body = c->ingress + body; + memcpy(c->body + c->body_size, b.rows, b.rows_size); + } } - if (c->batch > 1 && (at == size || json[at++] != ']')) return "json"; - if (pure_json_space(json, at, size) != size) return "json"; - put_u32(c->ingress, (uint32_t)(c->batch * c->field_bytes + c->body_size) | PLACED | delivered); + free(b.rows); + if (why) return why; + put_u32(c->ingress, (uint32_t)(payload + b.rows_size) | PLACED | delivered); return NULL; } @@ -784,7 +800,8 @@ static void body_done(Connection *c) { const char *why = c->route == ROUTE_INPUT && c->json ? convert(c) : NULL; if (why) { release_ticket(c); /* never made ready: it never runs */ - respond(c, 400, "Bad Request", why, 0, (PureSpan){0}); + if (!strcmp(why, "payload_budget") || !strcmp(why, "memory")) respond(c, 503, "Service Unavailable", why, 0, (PureSpan){0}); + else respond(c, 400, "Bad Request", why, 0, (PureSpan){0}); return; } if (!pure_runtime_ready(c->ticket)) abort(); diff --git a/tests/check.py b/tests/check.py index d262b8e..cc33947 100644 --- a/tests/check.py +++ b/tests/check.py @@ -1023,6 +1023,163 @@ def server_surface(aup, pure, work): return {'events': 5, 'lines': len(got)} +SHOP = ''' +record Stock { + item = 0 + left = 0 +} +record Line { + item = 0 + qty = 1 + note: text +} +record Entry { + item = 0 + qty = 0 + note: text +} +state stock: list of Stock by item +state log: list of Entry +state orders = 0 + +input restock = message { + rows: list of Stock +} +input order = message { + customer: u32 + lines: list of Line +} +input clean = message { + below: number +} + +on restock { + for r in restock.rows { + stock.add(Stock { item: r.item, left: r.left }) + } +} +on order { + for l in order.lines { + stock[l.item].left = stock[l.item].left - l.qty + require stock[l.item].left >= 0 + log.append(Entry { item: l.item, qty: l.qty, note: l.note }) + } + orders = orders + order.lines.count +} +on clean { + for s in stock where s.left < clean.below { + stock.remove(s) + } +} +view items = stock.count +view left = sum(s.left for s in stock) +view logged = log.count +view total = sum(e.qty for e in log) +view lines = orders +view notes = each e in log { + item = e.item + note = e.note +} +''' + + +def server_rows(aup, pure, work): + """A message's rows (a list of a record: a JSON array of objects, or the + list's stored form as the body) walked with for, all or nothing: 30,000 + stock rows in one request, then 20,000 as bytes, then random orders whose + rows take stock (an item not stocked or not enough of it fails the whole + order, none of its rows applied), a walk that removes, and a restart that + keeps every state and no rows. Checked against a model.""" + import http.client + import random + import struct + import time + source, module = work / 'shop.aup', work / 'shop.pmod' + source.write_text(SHOP) + compile_module(aup, source, module) + store = work / 'shop-store' + host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', store]) + + def call(method, path, body=None, kind='application/json'): + connection = http.client.HTTPConnection('127.0.0.1', host.port, timeout=60) + try: + connection.request(method, path, body, {'Content-Type': kind} if body is not None else {}) + answer = connection.getresponse() + return answer.status, answer.getheader('X-Pure-Error'), answer.read() + finally: + connection.close() + + def number(view): + status, _, data = call('GET', f'/views/{view}') + assert status == 200, (view, status) + return float(data) + + stock, log = {}, [] + try: + began = time.perf_counter() + rows = [{'item': i, 'left': i % 7} for i in range(30000)] + assert call('POST', '/inputs/restock', json.dumps({'rows': rows}).encode())[:2] == (200, 'none') + seconds = time.perf_counter() - began + stock.update({i: i % 7 for i in range(30000)}) + packed = b''.join(struct.pack('<dd', i, 5) for i in range(30000, 50000)) + assert call('POST', '/inputs/restock', packed, 'application/octet-stream')[:2] == (200, 'none') + stock.update({i: 5 for i in range(30000, 50000)}) + assert number('items') == 50000 and number('left') == sum(stock.values()) + # All or nothing: a row whose item is there already fails the whole message. + assert call('POST', '/inputs/restock', json.dumps({'rows': [{'item': 60000}, {'item': 7}]}).encode())[:2] == \ + (409, 'exists') + assert call('POST', '/inputs/restock', packed[:40], 'application/octet-stream')[:2] == (422, 'limit') + assert call('POST', '/inputs/restock', json.dumps({'rows': [{'item': 1, 'lft': 2}]}).encode())[:2] == \ + (400, 'json_field') + assert call('POST', '/inputs/restock', json.dumps({'rows': [{'item': 'one'}]}).encode())[:2] == (400, 'json_type') + assert number('items') == 50000 + rng = random.Random(21) + notes = ['', 'fast', '빨리', 'a "quoted" \\ note', 'tab\there', '😀'] + applied = refused = 0 + for _ in range(400): + lines = [{'item': rng.randrange(50010), 'qty': rng.randrange(4), 'note': rng.choice(notes)} + for _ in range(rng.randrange(6))] + if lines and rng.random() < 0.3: + del lines[0]['qty'] # a field left out takes its default (1) + model = dict(stock) + ok = True + for line in lines: + qty = line.get('qty', 1) + if line['item'] not in model or model[line['item']] - qty < 0: + ok = False + break + model[line['item']] -= qty + status, error, _ = call('POST', '/inputs/order', json.dumps({'customer': 3, 'lines': lines}).encode()) + assert (status == 200) == ok, (status, error, lines) + if ok: + stock = model + log.extend((line['item'], line.get('qty', 1), line['note']) for line in lines) + applied += 1 + else: + refused += 1 + assert number('logged') == len(log) and number('total') == sum(q for _, q, _ in log) + assert number('left') == sum(stock.values()) and number('lines') == len(log) + shown = json.loads(call('GET', f'/views/notes?from={max(0, len(log) - 50)}')[2])['items'] + assert [(r['item'], r['note']) for r in shown] == [(i, n) for i, _, n in log[-50:]] + assert call('POST', '/inputs/clean', json.dumps({'below': 2}).encode())[:2] == (200, 'none') + stock = {item: left for item, left in stock.items() if left >= 2} + assert number('items') == len(stock) and number('left') == sum(stock.values()) + finally: + host.stop() + # Restarted: the states as they were; rows are never kept, and come again. + host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', store]) + try: + assert number('items') == len(stock) and number('logged') == len(log) and number('lines') == len(log) + assert call('POST', '/inputs/restock', json.dumps({'rows': [{'item': 90000, 'left': 1}]}).encode())[:2] == \ + (200, 'none') + assert number('items') == len(stock) + 1 + finally: + host.stop() + assert applied > 50 and refused > 50 + return {'bulk_rows_json': 30000, 'bulk_rows_bytes': 20000, 'bulk_rows_per_second': round(30000 / seconds), + 'orders_applied': applied, 'orders_refused': refused, 'logged': len(log)} + + def server_queries(aup, pure, work): """Queries as verified programs over a snapshot (PUT and GET /queries/NAME): aup query compiles a bank's declarations with views of its own; the node @@ -1608,6 +1765,7 @@ def main(): report['server_sealed'] = server_sealed(aup, pure, work) report['server_surface'] = server_surface(aup, pure, work) report['server_queries'] = server_queries(aup, pure, work) + report['server_rows'] = server_rows(aup, pure, work) report['server_answers'] = server_answers(aup, pure, work) print(json.dumps(report)) |