diff options
| -rw-r--r-- | CHECKPOINT.md | 2 | ||||
| -rw-r--r-- | docs/design/008-module-format.md | 4 | ||||
| -rw-r--r-- | docs/research/083-delivery-over-http.md | 24 | ||||
| -rw-r--r-- | prototypes/connections/deliver.c | 191 | ||||
| -rw-r--r-- | prototypes/connections/deliver.h | 21 | ||||
| -rw-r--r-- | prototypes/connections/serve.c | 74 |
6 files changed, 281 insertions, 35 deletions
diff --git a/CHECKPOINT.md b/CHECKPOINT.md index e7cb1c0..53c306f 100644 --- a/CHECKPOINT.md +++ b/CHECKPOINT.md @@ -18,7 +18,7 @@ | HTTP 서비스 | 보기가 GET, 입력이 POST다(JSON 본문과 그 안의 그룹 객체를 제자리에서 읽는다, 라우터 코드 없음). 앱의 `send`는 이벤트와 함께 저장되고, `--send 출력=URL`로 다른 앱이나 HTTP 끝점에 순서대로 묶어 전해지며, 받는 앱은 번호마다 한 번 적용한다(서버를 죽였다 살려도 정확히 한 번). 답은 저장 뒤에 나간다. 상태를 가진 앱은 소유자 스레드 하나가 입력을 적용하고, 연결 스레드는 마지막 저장의 스냅숏으로 보기를 답한다(쓰기가 동기화하는 동안에도 읽기는 기다리지 않는다) | 008 HTTP 호스트, [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [082](docs/research/082-json-in-place.md), [083](docs/research/083-delivery-over-http.md) | | 성능(Pi 5) | 문서 서버: 작은 GET 151.6k/s, 1 MiB GET 2,794/s(axum 1,372, drogon 708), 충돌 안전 PUT은 axum-atomic의 5.1–5.8배. 앱 서비스(aup 22줄 대 손으로 쓴 서비스): 내구성 있는 쓰기 약 76배(그 뒤 저장소 커밋을 한 구간으로 바꿔 64연결 21–26k/s, axum·drogon 내구 약 180/s), 20행 JSON GET 145k/s(axum 95k, drogon 144k), 쓰기와 섞인 읽기 149k/s·p99 0.6 ms(axum·drogon 내구 1.1–1.3k/s·p99 111–156 ms), 최대 RSS 3.0–4.1 MiB(axum 6.4, drogon 11.6) | [075](docs/research/075-http-host-against-axum.md), [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [081](docs/research/081-tail-records.md) | -남은 것: macOS 검증, `bool`·`i64`, 꼬리 늘리기·압축을 커밋 밖으로와 열린 부하에서의 저장·적용 겹치기(081 다음 결정), 응답을 이벤트로 돌려받기(083 다음 결정), 앱 라우트의 인증·TLS(문서 +남은 것: macOS 검증, `bool`·`i64`, 꼬리 늘리기·압축을 커밋 밖으로와 열린 부하에서의 저장·적용 겹치기(081 다음 결정), 밀린 outbox를 저장마다 다 훑지 않기(083 다음 결정), 앱 라우트의 인증·TLS(문서 피어의 TLS는 있다), aup·Vega의 배포 묶음. ## 공개할 작업물과 완료 증거 diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md index 96a77fa..3fd4d52 100644 --- a/docs/design/008-module-format.md +++ b/docs/design/008-module-format.md @@ -343,7 +343,9 @@ text·리스트가 남을 때만 위치를 유지한다. JSON(필드와 그룹)으로 그 출력의 줄에 넣고, 출력마다 전달 스레드가 순서대로 보낸다(`X-Pure-Source: 이름.출력`, `X-Pure-Sequence`, `Idempotency-Key`). 2xx 또는 Pure 받는 쪽의 거절이면 전해진 것이고, 소유자가 덜어 낸다. 그 밖에는 쉬었다가 다시 보낸다(버리지 않는다). 받는 쪽이 `X-Pure-Batch`를 알리면 밀린 것을 묶어 보낸다([083](../research/083-delivery-over-http.md)). - 출력이 있는 앱은 상태가 없어도 소유자를 둔다. + 출력이 있는 앱은 상태가 없어도 소유자를 둔다. `--reply 출력=입력`이면 각 보냄의 최종 응답(2xx, 408·429를 뺀 4xx)이 입력의 + 메시지 이벤트가 된다: 필드는 응답 JSON에서 이름으로(모르는 멤버는 건너뛴다), 없으면 보낸 메시지의 같은 이름·타입 필드에서, + u32 `http_status`는 상태 코드로. 응답은 보낸 번호로 한 번만 적용되고, 적용한 뒤에 보냄을 덜어 낸다. - **보기는 스냅숏에서.** 소유자는 저장이 성공할 때마다, 답을 돌려주기 전에 저장된 상태의 스냅숏을 게시한다. 연결 스레드는 라운드마다(돌려받은 답을 처리한 뒤) 최신 스냅숏을 잡아 자기 읽기용 앱으로 그 라운드의 보기를 모두 평가하고 답을 모아 보낸다. 입력은 그보다 먼저 소유자에게 넘겨, 소유자가 그동안 적용하고 저장한다. 그래서 쓰기의 diff --git a/docs/research/083-delivery-over-http.md b/docs/research/083-delivery-over-http.md index 5427590..829c666 100644 --- a/docs/research/083-delivery-over-http.md +++ b/docs/research/083-delivery-over-http.md @@ -41,6 +41,24 @@ - **TLS.** `https://` 받는 곳에는 OpenSSL로 TLS 1.2 이상을 쓴다. 인증서는 시스템 CA(또는 `--send-ca 파일`)로 검증하고, 호스트 이름(SNI와 `SSL_set1_host`, IP 주소면 그 주소)과 맞아야 한다. 맞지 않으면 기록하고 10초마다 다시 시도한다(보냄은 남는다). +## 응답을 이벤트로 + +결제처럼 답이 필요한 호출은 `--reply 출력=입력`으로 잇는다. 그 출력으로 보낸 요청의 최종 응답(2xx, 그리고 408·429를 뺀 +4xx)이 입력의 메시지 이벤트가 된다. + +- 필드는 응답 JSON 객체에서 이름으로 채우고, 모르는 멤버는 값째 건너뛴다(API는 많은 것을 돌려준다). 응답에 없는 필드는 + 보낸 메시지의 같은 이름·타입 필드에서 채운다. 그래서 요청의 `order`를 응답 이벤트에서 그대로 볼 수 있다(API가 되풀이하지 + 않아도). u32 필드 `http_status`는 응답의 상태 코드다. 어느 쪽에도 없는 필드가 있으면 기록하고 10초마다 다시 보낸다. +- 응답은 보낸 쪽 이름과 보낸 번호로 `pure_app_deliver`된다. 소유자는 응답을 적용한 뒤에 그 보냄을 덜어 내고, 둘은 한 저장에 + 담긴다. 그 사이에 죽으면 보냄이 남아 다시 보내지고(`Idempotency-Key`로 API가 같은 답을 준다), 응답은 번호로 한 번만 적용된다. +- 응답을 기다리는 출력은 한 요청에 하나씩 보낸다. 요청이 기다리는 라운드가 있으면 그 저장에 응답을 합치고, 쉴 때는 따로 + 저장한다. + +**찾은 것: 조각으로 쓰는 서버와 지연 ACK.** 처음에는 응답이 초당 24건쯤밖에 오지 않았다. 파이썬 `http.server`는 헤더와 +본문을 두 번에 나눠 쓰고(Nagle이 둘째 조각을 첫 조각의 ACK까지 붙잡는다), 우리 클라이언트는 ACK를 40 ms 늦췄다. 요청마다 +40 ms다. 읽을 때마다 `TCP_QUICKACK`으로 곧바로 ACK하자 같은 검사가 30초를 넘기던 것에서 1.8초가 됐다. 받는 서버가 어떻게 +쓰든 클라이언트가 막히지 않아야 하므로 클라이언트에서 푼다. + ## 장치가 말한 것 Raspberry Pi 5. 받는 서버는 CPU 0–1(연결 스레드 2), 보내는 서버는 CPU 2, 클라이언트(파이썬 16스레드)는 나머지. @@ -63,13 +81,15 @@ Raspberry Pi 5. 받는 서버는 CPU 0–1(연결 스레드 2), 보내는 서버 - aup `make check`의 `server_webhook`: 파이썬 HTTPS 서버(자체 인증서)를 제3자 웹훅으로 두고, 주문 다섯 건이 순서대로 출력의 JSON 모양과 헤더(`X-Pure-Source`, `Idempotency-Key`, 묶음 헤더 없음)로 도착한다. 처음에 503으로 답한 2번은 다시 온다. 그 인증서를 믿지 않는 보내는 쪽에서는 아무것도 도착하지 않고 기록이 남는다. ASan·UBSan 빌드로도 통과했다. +- aup `make check`의 `server_answers`: 파이썬 "결제 API"(1000 이하는 200 `succeeded`, 넘으면 402 `declined`, 같은 + `Idempotency-Key`에는 첫 답을 되풀이)에 출력을 잇고 응답을 `charged` 입력으로 받는다. 클라이언트 셋이 사는 동안 보내는 서버를 + `kill -9`로 죽였다 살린다. 끝에 앱의 결제·거절 수의 합이 저장된 구매 수와 API가 받은 키 수와 같고, 결제 수와 결제 합계가 API의 + 200 응답과 같다(각 3,774–3,979건, 네 번, ASan·UBSan 빌드로 한 번 더). - `test-module`의 링크 검사([078](078-app-connections.md), 008)가 런타임 쪽(이벤트의 보냄, 저장 실패, 재시작, 중복·거절)을 맡는다. ## 다음 결정 -- **응답을 이벤트로.** 지금은 보내고 끝이다(알림, 웹훅). 결제처럼 답이 필요한 호출은 응답을 앱의 입력으로 돌려주는 - 배선(`--send 출력=URL --reply 입력`)이 필요하다. - 저장이 끝날 때마다 outbox 전체를 훑어 새 보냄을 찾는다. 받는 쪽이 오래 꺼져 수만 건이 밀리면 저장마다 그만큼 훑는다. 마지막으로 넣은 자리에서 이어 훑게 해야 한다. - 덜어 낸 것은 다음 저장에 지워진다. 요청이 없는 앱은 저장하지 않으므로, 다시 시작하면 전한 것을 한 번 더 보낸다(받는 diff --git a/prototypes/connections/deliver.c b/prototypes/connections/deliver.c index c4136a2..0d3da64 100644 --- a/prototypes/connections/deliver.c +++ b/prototypes/connections/deliver.c @@ -1,5 +1,6 @@ #define _POSIX_C_SOURCE 200809L #include "deliver.h" +#include "message-json.h" #include "module-text.h" #include <errno.h> #include <netdb.h> @@ -18,10 +19,14 @@ #include <time.h> #include <unistd.h> -enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256 }; -enum { DELIVERED, AGAIN, REFUSED_BY_PEER, TLS_FAILED }; +enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256, + ANSWER_MOST = 1 << 20 }; +enum { DELIVERED, AGAIN, REFUSED_BY_PEER, TLS_FAILED, ANSWER_UNREAD }; -typedef struct { char *body; size_t size; uint64_t seq; } Pending; +typedef struct { char *body; size_t size; uint64_t seq; unsigned char *sent; size_t sent_size; } Pending; +/* An answer made into its input's message: a table of the fields' starts and + * sizes (u32le), then the payload from its prefix on. */ +typedef struct { uint64_t seq; unsigned char *bytes; size_t size, table; } Answer; struct PureWire { const PureModule *module; @@ -40,6 +45,11 @@ struct PureWire { int fd; /* the connection kept for the next send, or -1; changed under the lock */ bool tls; /* https: the connection is TLS */ SSL *ssl; /* its TLS session */ + int answer_input; /* the input answers become events of, or -1 */ + Answer *answers; /* made, for the owner to take in order */ + size_t answer_head, answer_count, answer_capacity; + Answer taken; /* the one the owner took last, kept until it takes the next */ + uint32_t placed[2 * PURE_MODULE_FIELDS]; }; /* ---- TLS: one client context, verifying the receiver's certificate against @@ -226,12 +236,42 @@ PureWire *pure_wire_open(const PureModule *m, const char *output, const char *ur memcpy(w->source, source, length + 1); w->fd = -1; w->tls = tls; + w->answer_input = -1; pthread_mutex_init(&w->lock, NULL); pthread_cond_init(&w->changed, NULL); return w; } uint32_t pure_wire_output(const PureWire *w) { return w->output; } + +bool pure_wire_reply(PureWire *w, uint32_t input, char *error, size_t size) { + const PureModule *m = w->module; + if (input >= m->inputs || m->input[input].kind != PURE_KIND_MESSAGE || m->input[input].keyed) { + snprintf(error, size, "an answer goes to a message input without a key"); + return false; + } + w->answer_input = (int)input; + return true; +} + +bool pure_wire_answer(PureWire *w, PureWireAnswer *out) { + free(w->taken.bytes); + w->taken = (Answer){0}; + pthread_mutex_lock(&w->lock); + bool some = w->answer_count > 0; + if (some) { + w->taken = w->answers[w->answer_head++]; + --w->answer_count; + w->delivered = w->taken.seq; /* trimmed after the owner applies it */ + } + pthread_mutex_unlock(&w->lock); + if (!some) return false; + const PureInputShape *in = &w->module->input[w->answer_input]; + for (size_t i = 0; i < 2 * (size_t)in->field_count; ++i) w->placed[i] = read_u32(w->taken.bytes + 4 * i); + *out = (PureWireAnswer){(uint32_t)w->answer_input, w->taken.seq, + {w->taken.bytes + w->taken.table, w->taken.size - w->taken.table}, w->placed, w->source}; + return true; +} uint64_t pure_wire_queued(const PureWire *w) { return w->queued; } uint64_t pure_wire_delivered(PureWire *w) { @@ -275,7 +315,14 @@ bool pure_wire_queue(PureWire *w, uint64_t seq, PureSpan payload) { w->capacity = capacity; } } - w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq}; + unsigned char *sent = NULL; + if (w->answer_input >= 0 && !(sent = malloc(payload.size ? payload.size : 1))) { + pthread_mutex_unlock(&w->lock); + free(t.bytes); + return false; + } + if (sent && payload.size) memcpy(sent, payload.data, payload.size); + w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq, sent, payload.size}; w->queued = seq; pthread_cond_signal(&w->changed); pthread_mutex_unlock(&w->lock); @@ -294,6 +341,13 @@ static bool fill(Reader *r) { if (w->ssl) n = SSL_read(w->ssl, r->bytes + r->end, (int)(sizeof(r->bytes) - r->end)); else do n = recv(w->fd, r->bytes + r->end, sizeof(r->bytes) - r->end, 0); while (n < 0 && errno == EINTR); if (n <= 0) return false; +#ifdef TCP_QUICKACK + /* Acknowledge at once: a server that writes an answer in pieces holds the + * next piece until the first is acknowledged (Nagle), which a delayed + * acknowledgement would stall for 40 ms. The flag does not stay set. */ + int one = 1; + setsockopt(w->fd, IPPROTO_TCP, TCP_QUICKACK, &one, sizeof(one)); +#endif r->end += (size_t)n; return true; } @@ -320,14 +374,17 @@ static bool read_line(Reader *r, char *line, size_t size) { } } -static bool skip(Reader *r, uint64_t size) { +/* size bytes of the body: kept in into (up to ANSWER_MOST), or skipped when it is NULL. */ +static bool take(Reader *r, uint64_t size, Text *into) { + if (into && size > ANSWER_MOST - into->size) return false; while (size) { if (r->at == r->end && !fill(r)) return false; size_t n = r->end - r->at < size ? r->end - r->at : (size_t)size; + if (into) put(into, r->bytes + r->at, n); r->at += n; size -= n; } - return true; + return !into || !into->failed; } static bool send_all(PureWire *w, const char *bytes, size_t size) { @@ -397,7 +454,7 @@ static bool refusal(const char *error) { /* The answer to one request: its status, whether a Pure receiver refused it, * and whether the connection may carry the next. */ -static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, unsigned *batch) { +static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, unsigned *batch, Text *body) { Reader r = {.wire = w}; char line[1024]; if (!read_line(&r, line, sizeof(line)) || strncmp(line, "HTTP/1.", 7) || strlen(line) < 12) return false; @@ -414,15 +471,18 @@ static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, uns if (!read_line(&r, line, sizeof(line))) return false; chunk = strtoull(line, NULL, 16); if (!chunk) break; - if (!skip(&r, chunk) || !read_line(&r, line, sizeof(line))) return false; + if (!take(&r, chunk, body) || !read_line(&r, line, sizeof(line))) return false; } while (read_line(&r, line, sizeof(line)) && *line) {} return true; } if (*status == 204 || *status == 304 || *status < 200) return true; - if (sized) return skip(&r, length); + if (sized) return take(&r, length, body); *keep = false; /* the body runs to the close */ - while (fill(&r)) r.at = r.end; + while (fill(&r)) { + if (body && !take(&r, r.end - r.at, body)) return false; + r.at = r.end; + } return true; } char *value = strchr(line, ':'); @@ -438,9 +498,64 @@ static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, uns return false; } +static void write_u32(unsigned char *p, uint32_t v) { for (unsigned i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> (8 * i)); } + +/* The answer (status, body) to the send p as a message of the answer input: + * fields from the body's JSON object by name, others skipped; a field the + * body lacks from the send's field of that name and type; a u32 field named + * http_status from the status. False when a field has none of these. */ +static bool form_answer(const PureWire *w, int status, const Text *body, const Pending *p, Answer *out) { + const PureModule *m = w->module; + const PureInputShape *in = &m->input[w->answer_input], *sent = &m->input[w->output]; + size_t table = 8 * (size_t)in->field_count, total = table + in->prefix + body->size + p->sent_size; + unsigned char *bytes = calloc(1, total ? total : 1); + if (!bytes) return false; + 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}; + 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(bytes); + return false; + } + 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_U32 && !strcmp(field->name, "http_status")) { + write_u32(prefix + field->offset, (uint32_t)status); + continue; + } + /* The send's field of the same name and type, and where its bytes are. */ + size_t start = sent->prefix, length = 0; + const PureField *from = NULL; + for (uint32_t g = 0; g < sent->field_count && !from; ++g) { + const PureField *other = &m->field[sent->field_start + g]; + bool variable = other->type == PURE_FIELD_BYTES || other->type == PURE_FIELD_TEXT; + if (variable) length = other->offset == PURE_FIELD_REST ? p->sent_size - start : read_u32(p->sent + other->offset); + if (!strcmp(other->name, field->name) && other->type == field->type) from = other; + else if (variable) start += length; + } + bool variable = field->type == PURE_FIELD_BYTES || field->type == PURE_FIELD_TEXT; + if (!from || (variable && (start > p->sent_size || length > p->sent_size - start))) { + free(bytes); + return false; + } + if (field->type == PURE_FIELD_U32 || field->type == PURE_FIELD_NUMBER) + memcpy(prefix + field->offset, p->sent + from->offset, field->type == PURE_FIELD_U32 ? 4 : 8); + else { + write_u32(bytes + 8 * f, (uint32_t)(copy + start - prefix)); + write_u32(bytes + 8 * f + 4, (uint32_t)length); + } + } + *out = (Answer){p->seq, bytes, total, table}; + return true; +} + /* One request for count sends from p (numbered in a row): one object, or - * an array of them for a receiver that takes batches. */ -static int post(PureWire *w, const Pending *p, unsigned count) { + * an array of them for a receiver that takes batches. A final answer to a + * send whose answers are wired is made into *answer. */ +static int post(PureWire *w, const Pending *p, unsigned count, Answer *answer) { size_t body = count > 1 ? count + 1 : 0; for (unsigned i = 0; i < count; ++i) body += p[i].size; char head[2048], batch[32] = ""; @@ -473,19 +588,30 @@ static int post(PureWire *w, const Pending *p, unsigned count) { int status; bool refused, keep, sent = send_all(w, head, (size_t)size); unsigned batch_size = 0; + Text body = {0}; for (unsigned i = 0; sent && i < count; ++i) sent = (!i || count == 1 || send_all(w, ",", 1)) && (i || count == 1 || send_all(w, "[", 1)) && send_all(w, p[i].body, p[i].size); if (sent && count > 1) sent = send_all(w, "]", 1); - if (!sent || !read_answer(w, &status, &refused, &keep, &batch_size)) { + if (!sent || !read_answer(w, &status, &refused, &keep, &batch_size, w->answer_input >= 0 ? &body : NULL)) { + free(body.bytes); drop_connection(w); if (fresh) return AGAIN; continue; } if (!keep) drop_connection(w); - if (batch_size) w->batch = batch_size; + if (batch_size && w->answer_input < 0) w->batch = batch_size; + if (status == 408 || status == 429 || status >= 500) { + free(body.bytes); + return AGAIN; + } + if (w->answer_input >= 0 && status >= 200 && status < 500) { /* a final answer: the app hears it */ + bool formed = form_answer(w, status, &body, p, answer); + free(body.bytes); + return formed ? DELIVERED : ANSWER_UNREAD; + } + free(body.bytes); if ((status >= 200 && status < 300) || (status >= 400 && status < 500 && refused)) return DELIVERED; - if (status == 408 || status == 429 || status >= 500) return AGAIN; return REFUSED_BY_PEER; } return AGAIN; @@ -507,13 +633,31 @@ static void *deliver(void *context) { memcpy(next, w->queue + w->head, count * sizeof(*next)); ++w->attempts; pthread_mutex_unlock(&w->lock); - int outcome = post(w, next, count); + Answer answer = {0}; + int outcome = post(w, next, count, &answer); pthread_mutex_lock(&w->lock); + if (outcome == DELIVERED && answer.bytes) { /* for the owner, who then counts it delivered */ + if (w->answer_head + w->answer_count == w->answer_capacity) { + if (w->answer_head) { + memmove(w->answers, w->answers + w->answer_head, w->answer_count * sizeof(*w->answers)); + w->answer_head = 0; + } else { + size_t capacity = w->answer_capacity ? 2 * w->answer_capacity : 16; + Answer *grown = realloc(w->answers, capacity * sizeof(*grown)); + if (!grown) { free(answer.bytes); outcome = AGAIN; } + else { w->answers = grown; w->answer_capacity = capacity; } + } + } + if (outcome == DELIVERED) w->answers[w->answer_head + w->answer_count++] = answer; + } if (outcome == DELIVERED) { - for (unsigned i = 0; i < count; ++i) free(w->queue[w->head + i].body); + for (unsigned i = 0; i < count; ++i) { + free(w->queue[w->head + i].body); + free(w->queue[w->head + i].sent); + } w->head += count; w->count -= count; - w->delivered = next[count - 1].seq; + if (!answer.bytes) w->delivered = next[count - 1].seq; w->deliveries += count; pause_ms = 0; pthread_mutex_unlock(&w->lock); @@ -521,9 +665,10 @@ static void *deliver(void *context) { pthread_mutex_lock(&w->lock); continue; } - if ((outcome == REFUSED_BY_PEER || outcome == TLS_FAILED) && reported != next[0].seq) { + if (outcome != AGAIN && reported != next[0].seq) { fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%llu,\"error\":\"%s\"}\n", w->source, (unsigned long long)next[0].seq, outcome == TLS_FAILED ? "the receiver's certificate is not trusted; trying again" : + outcome == ANSWER_UNREAD ? "its answer does not make the answer input's message; trying again" : "the receiver does not take it; trying again"); reported = next[0].seq; } @@ -557,7 +702,13 @@ void pure_wire_close(PureWire *w) { if (w->started) pthread_join(w->thread, NULL); if (w->ssl) SSL_free(w->ssl); if (w->fd >= 0) close(w->fd); - for (size_t i = 0; i < w->count; ++i) free(w->queue[w->head + i].body); + for (size_t i = 0; i < w->count; ++i) { + free(w->queue[w->head + i].body); + free(w->queue[w->head + i].sent); + } + for (size_t i = 0; i < w->answer_count; ++i) free(w->answers[w->answer_head + i].bytes); + free(w->answers); + free(w->taken.bytes); free(w->queue); pthread_cond_destroy(&w->changed); pthread_mutex_destroy(&w->lock); diff --git a/prototypes/connections/deliver.h b/prototypes/connections/deliver.h index 13048fc..2d1da29 100644 --- a/prototypes/connections/deliver.h +++ b/prototypes/connections/deliver.h @@ -34,6 +34,27 @@ uint64_t pure_wire_delivered(PureWire *); bool pure_wire_start(PureWire *, void (*notify)(void *), void *context); /* Stop (a delivery under way is finished or abandoned) and free. */ void pure_wire_close(PureWire *); +/* Answers as events: the final answer to each send (a 2xx, or a 4xx but 408 + * and 429) becomes a message event of input (a message without a key), in + * the order sent: its fields come from the answer's JSON object by name + * (other members are skipped), a field the answer lacks from the send's + * field of that name and type (as a request's id), and a u32 field named + * http_status is the answer's status. An answer that makes no message is + * logged and the send tried again every 10 s. Answers go one per request. */ +bool pure_wire_reply(PureWire *, uint32_t input, char *error, size_t size); +/* The owner's side: the next answer in order, or false. Taking it counts + * its send delivered (the owner applies the answer, then trims); what out + * points to is the wire's until the next call. The answer is applied with + * pure_app_deliver from the wire's source with the send's number, so an + * answer that comes again after a restart is applied once. */ +typedef struct { + uint32_t input; + uint64_t seq; + PureSpan payload; + const uint32_t *placed; + const char *source; +} PureWireAnswer; +bool pure_wire_answer(PureWire *, PureWireAnswer *out); /* Counts for a host's report. */ void pure_wire_counts(PureWire *, uint64_t *delivered, uint64_t *attempts); #endif diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c index 88a9a6f..60771e1 100644 --- a/prototypes/connections/serve.c +++ b/prototypes/connections/serve.c @@ -150,7 +150,8 @@ typedef struct { /* The owner's: outputs wired to HTTP receivers, and how far each was trimmed. */ PureWire *wires[PURE_MODULE_INPUTS]; unsigned wire_count; - uint64_t trimmed[PURE_MODULE_INPUTS]; + uint64_t trimmed[PURE_MODULE_INPUTS], answers; + bool held[PURE_MODULE_INPUTS]; /* an answer failed to apply: the wire's sends stay until a restart */ } Executor; /* A request made ready: its effect, what it names, and the thread that answers it. */ @@ -1051,14 +1052,32 @@ static void after_save(Executor *ex) { if (ex->wire_count) pure_app_outbox(&ex->app, queue_send, &q); } -/* What the wires delivered leaves the outbox (with the next save). */ -static void trim_delivered(Executor *ex) { +/* The answers the wires made become events, in order; then what the wires + * delivered leaves the outbox (with the next save). True when an answer was + * applied (the caller saves). */ +static bool take_deliveries(Executor *ex) { + bool answered = false; for (unsigned i = 0; i < ex->wire_count; ++i) { + PureWireAnswer answer; + while (pure_wire_answer(ex->wires[i], &answer)) { + ex->app.now = wall_ms(); + PureResult result = pure_app_deliver(&ex->app, answer.input, answer.source, answer.seq, answer.payload, + answer.placed); + if (result != PURE_OK && ex->app.fault == PURE_FAULT_NONE) { + ex->held[i] = true; /* not applied and not refused: sent again after a restart */ + fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%" PRIu64 ",\"error\":\"its answer was not applied\"}\n", + answer.source, answer.seq); + } + pure_app_finish(&ex->app); + answered = true; + ++ex->answers; + } uint64_t delivered = pure_wire_delivered(ex->wires[i]); - if (delivered <= ex->trimmed[i]) continue; + if (delivered <= ex->trimmed[i] || ex->held[i]) continue; pure_app_trim(&ex->app, pure_wire_output(ex->wires[i]), delivered); ex->trimmed[i] = delivered; } + return answered; } /* One save for a burst: its requests are durable with it or fail with it. */ @@ -1290,8 +1309,14 @@ static void *run_owner(void *context) { owner->delivered = false; pthread_mutex_unlock(&owner->lock); long due = run_timers(&owner->executor); - if (delivered) trim_delivered(&owner->executor); + bool answered = delivered && take_deliveries(&owner->executor); pthread_mutex_lock(&owner->lock); + /* Answers are saved with the next round, or now if none waits. */ + if (answered && !owner->count && !owner->stopping) { + pthread_mutex_unlock(&owner->lock); + if (pure_app_save(&owner->executor.app) == PURE_OK) after_save(&owner->executor); + pthread_mutex_lock(&owner->lock); + } if (owner->count || owner->stopping) break; if (owner->delivered) continue; if (due < 0) { @@ -1604,6 +1629,30 @@ static bool open_wires(Executor *ex, const PureModule *m, const char *const *sen return true; } +/* --reply OUTPUT=INPUT: the answers to that output's sends become INPUT's events. */ +static bool wire_replies(Executor *ex, const PureModule *m, const char *const *replies, unsigned count) { + for (unsigned i = 0; i < count; ++i) { + const char *equals = strchr(replies[i], '='); + char output[PURE_MODULE_NAME], error[160] = ""; + size_t length = equals ? (size_t)(equals - replies[i]) : 0; + int input = equals ? pure_module_input(m, equals + 1) : -1; + PureWire *wire = NULL; + if (length && length < sizeof(output)) { + memcpy(output, replies[i], length); + output[length] = 0; + int index = pure_module_input(m, output); + for (unsigned k = 0; k < ex->wire_count && index >= 0; ++k) + if (pure_wire_output(ex->wires[k]) == (uint32_t)index) wire = ex->wires[k]; + } + if (!wire || input < 0 || !pure_wire_reply(wire, (uint32_t)input, error, sizeof(error))) { + fprintf(stderr, "--reply %s: %s\n", replies[i], !wire ? "takes OUTPUT=INPUT of an output given --send" : + input < 0 ? "no such input" : error); + return false; + } + } + return true; +} + /* --threads auto: a connection thread for each CPU this process may run on, * less the one the owner of an app with states takes. */ static unsigned auto_threads(bool owner) { @@ -1617,8 +1666,8 @@ static unsigned auto_threads(bool owner) { int main(int argc, char **argv) { static PureModule module; static Owner owner; - const char *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS]; - unsigned send_count = 0; + const char *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS], *replies[PURE_MODULE_INPUTS]; + unsigned send_count = 0, reply_count = 0; unsigned port = 8767, threads = 1; size_t budget = 64u << 20, capacity = 1024; uint64_t timeout_ms = 10000; @@ -1630,6 +1679,7 @@ int main(int argc, char **argv) { else if (!strcmp(argv[i], "--send") && send_count < PURE_MODULE_INPUTS) sends[send_count++] = argv[i + 1]; else if (!strcmp(argv[i], "--name")) name = argv[i + 1]; else if (!strcmp(argv[i], "--send-ca")) pure_wire_trust(argv[i + 1]); + else if (!strcmp(argv[i], "--reply") && reply_count < PURE_MODULE_INPUTS) replies[reply_count++] = 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) capacity = (size_t)value; @@ -1671,7 +1721,8 @@ int main(int argc, char **argv) { return 1; owner.executor.reads_first = true; owner.executor.publishes = true; - if (!open_wires(&owner.executor, &module, sends, send_count, name)) return 2; + if (!open_wires(&owner.executor, &module, sends, send_count, name) || + !wire_replies(&owner.executor, &module, replies, reply_count)) return 2; after_save(&owner.executor); /* the snapshot, and the sends stored before */ for (unsigned i = 0; i < owner.executor.wire_count; ++i) if (!pure_wire_start(owner.executor.wires[i], wire_delivered, &owner)) { fputs("Cannot start a wire\n", stderr); return 1; } @@ -1731,15 +1782,16 @@ int main(int argc, char **argv) { fprintf(stderr, "{\"active\":%zu,\"retained_payload_bytes\":%zu,\"accepted\":%" PRIu64 ",\"requests\":%" PRIu64 ",\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 ",\"ticks\":%" PRIu64 ",\"expired\":%" PRIu64 ",\"threads\":%u,\"owner\":%s,\"snapshots\":%" PRIu64 ",\"views_from_snapshots\":%" PRIu64 - ",\"delivered\":%" PRIu64 ",\"delivery_attempts\":%" PRIu64 ",\"commit_groups\":%lu,\"grouped_commits\":%lu}\n", + ",\"delivered\":%" PRIu64 ",\"delivery_attempts\":%" PRIu64 ",\"answers\":%" PRIu64 + ",\"commit_groups\":%lu,\"grouped_commits\":%lu}\n", active, retained, accepted, requests, bursts, events, ticks, expired, threads, owned ? "true" : "false", - owner.executor.publications, shown, delivered, attempts, groups, grouped); + owner.executor.publications, shown, delivered, attempts, owner.executor.answers, groups, grouped); free(hosts); free(thread); return 0; usage: fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] " "[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=http[s]://host:port/path ...] [--name NAME] " - "[--send-ca FILE]\n", stderr); + "[--send-ca FILE] [--reply OUTPUT=INPUT ...]\n", stderr); return 2; } |