From 5372d03b5968796515966e2cfc89628d6448f0c1 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Sun, 27 Sep 2026 08:24:08 +0100 Subject: Look for new sends newest first, only as far back as they go After each save the owner walked the whole outbox to find the sends not yet queued, so a receiver long down made every save walk the backlog. pure_app_outbox_newest visits saved sends newest first, and the owner stops once every wired output reaches the number it queued last, then queues what it found in order. --- CHECKPOINT.md | 2 +- docs/research/083-delivery-over-http.md | 4 +-- prototypes/connections/module.c | 7 ++++++ prototypes/connections/module.h | 2 ++ prototypes/connections/serve.c | 43 +++++++++++++++++++++++++-------- 5 files changed, 45 insertions(+), 13 deletions(-) diff --git a/CHECKPOINT.md b/CHECKPOINT.md index 53c306f..62bd853 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 다음 결정), 밀린 outbox를 저장마다 다 훑지 않기(083 다음 결정), 앱 라우트의 인증·TLS(문서 +남은 것: macOS 검증, `bool`·`i64`, 꼬리 늘리기·압축을 커밋 밖으로와 열린 부하에서의 저장·적용 겹치기(081 다음 결정), 응답을 기다리는 출력의 묶음 전달(083), 앱 라우트의 인증·TLS(문서 피어의 TLS는 있다), aup·Vega의 배포 묶음. ## 공개할 작업물과 완료 증거 diff --git a/docs/research/083-delivery-over-http.md b/docs/research/083-delivery-over-http.md index 829c666..eedc5f8 100644 --- a/docs/research/083-delivery-over-http.md +++ b/docs/research/083-delivery-over-http.md @@ -90,7 +90,7 @@ Raspberry Pi 5. 받는 서버는 CPU 0–1(연결 스레드 2), 보내는 서버 ## 다음 결정 -- 저장이 끝날 때마다 outbox 전체를 훑어 새 보냄을 찾는다. 받는 쪽이 오래 꺼져 수만 건이 밀리면 저장마다 그만큼 훑는다. - 마지막으로 넣은 자리에서 이어 훑게 해야 한다. +- 저장이 끝날 때마다 outbox를 가장 새것부터 거꾸로 훑어, 배선된 출력마다 이미 줄에 넣은 번호에 닿으면 멈춘다 + (`pure_app_outbox_newest`). 받는 쪽이 오래 꺼져 수만 건이 밀려도 저장마다 새 보냄만큼만 본다. - 덜어 낸 것은 다음 저장에 지워진다. 요청이 없는 앱은 저장하지 않으므로, 다시 시작하면 전한 것을 한 번 더 보낸다(받는 쪽이 걸러 낸다). diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c index 5cb7f11..697e918 100644 --- a/prototypes/connections/module.c +++ b/prototypes/connections/module.c @@ -1668,6 +1668,13 @@ void pure_app_outbox(const PureApp *app, PureSentVisit visit, void *context) { } } +void pure_app_outbox_newest(const PureApp *app, PureSentVisit visit, void *context) { + for (size_t i = app->sent_durable; i-- > 0;) { + const struct PureSent *sent = &app->sent[i]; + if (!sent->trimmed && !visit(context, sent->output, sent->seq, (PureSpan){sent->payload, sent->size})) return; + } +} + void pure_app_trim(PureApp *app, uint32_t output, uint64_t through) { for (size_t i = 0; i < app->sent_durable; ++i) if (app->sent[i].output == output && app->sent[i].seq <= through && !app->sent[i].trimmed) { diff --git a/prototypes/connections/module.h b/prototypes/connections/module.h index 3441c03..f62326b 100644 --- a/prototypes/connections/module.h +++ b/prototypes/connections/module.h @@ -309,6 +309,8 @@ PureResult pure_app_replace(PureApp *, const PureModule *); * and trims what its receiver has saved; a receiver applies each once. */ typedef bool (*PureSentVisit)(void *context, uint32_t output, uint64_t seq, PureSpan payload); void pure_app_outbox(const PureApp *, PureSentVisit visit, void *context); +/* The same, newest first (a host looking only for what is new stops early). */ +void pure_app_outbox_newest(const PureApp *, PureSentVisit visit, void *context); /* Forget output's saved sends numbered up to through (saved with the next save). */ void pure_app_trim(PureApp *, uint32_t output, uint64_t through); /* Deliver a message event from source (a sender's name, 1 to 64 printable diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c index 60771e1..2ea7adb 100644 --- a/prototypes/connections/serve.c +++ b/prototypes/connections/serve.c @@ -1033,23 +1033,46 @@ static void publish(Executor *ex) { ++ex->publications; } -/* The saved sends of wired outputs not queued yet go to their wires, in order; - * a wire that cannot take one takes the rest at a later save. */ -typedef struct { Executor *executor; bool blocked[PURE_MODULE_INPUTS]; } Queueing; - -static bool queue_send(void *context, uint32_t output, uint64_t seq, PureSpan payload) { +/* The saved sends of wired outputs not queued yet go to their wires, in + * order; a wire that cannot take one takes the rest at a later save. The + * outbox is looked at newest first, only as far back as the new sends go, + * however many older ones wait for their receivers. */ +typedef struct { uint32_t wire; uint64_t seq; PureSpan payload; } Fresh; +typedef struct { Executor *executor; Fresh *fresh; size_t count, capacity; unsigned done; bool failed; } Queueing; + +static bool find_fresh(void *context, uint32_t output, uint64_t seq, PureSpan payload) { Queueing *q = context; Executor *ex = q->executor; - for (unsigned i = 0; i < ex->wire_count; ++i) - if (pure_wire_output(ex->wires[i]) == output && !q->blocked[i] && seq > pure_wire_queued(ex->wires[i])) - q->blocked[i] = !pure_wire_queue(ex->wires[i], seq, payload); - return true; + for (unsigned i = 0; i < ex->wire_count; ++i) { + if (pure_wire_output(ex->wires[i]) != output || (q->done >> i & 1)) continue; + if (seq <= pure_wire_queued(ex->wires[i])) { + q->done |= 1u << i; /* older ones were queued before */ + break; + } + if (q->count == q->capacity) { + size_t capacity = q->capacity ? 2 * q->capacity : 64; + Fresh *grown = realloc(q->fresh, capacity * sizeof(*grown)); + if (!grown) { q->failed = true; return false; } + q->fresh = grown; + q->capacity = capacity; + } + q->fresh[q->count++] = (Fresh){i, seq, payload}; + break; + } + return q->done != (ex->wire_count >= 32 ? UINT32_MAX : (1u << ex->wire_count) - 1); } static void after_save(Executor *ex) { publish(ex); + if (!ex->wire_count) return; Queueing q = {.executor = ex}; - if (ex->wire_count) pure_app_outbox(&ex->app, queue_send, &q); + pure_app_outbox_newest(&ex->app, find_fresh, &q); + bool blocked[PURE_MODULE_INPUTS] = {false}; + for (size_t k = q.failed ? 0 : q.count; k-- > 0;) { + const Fresh *f = &q.fresh[k]; + if (!blocked[f->wire]) blocked[f->wire] = !pure_wire_queue(ex->wires[f->wire], f->seq, f->payload); + } + free(q.fresh); } /* The answers the wires made become events, in order; then what the wires -- cgit v1.2.3