aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--docs/research/075-http-host-against-axum.md22
-rw-r--r--prototypes/connections/serve.c121
-rw-r--r--prototypes/connections/state.h10
-rw-r--r--prototypes/connections/store-tree.c18
4 files changed, 141 insertions, 30 deletions
diff --git a/docs/research/075-http-host-against-axum.md b/docs/research/075-http-host-against-axum.md
index 8630fb3..ab26a37 100644
--- a/docs/research/075-http-host-against-axum.md
+++ b/docs/research/075-http-host-against-axum.md
@@ -102,3 +102,25 @@ PUT은 연결마다 자기 키 앞에 16 B를 넣고 받은 ETag로 다음 `If-M
ThreadSanitizer는 이 커널(47비트 가상 주소)을 지원하지 않아 쓰지 못했다. 스레드 동시 커밋은 ASan 빌드의 저장소
검사(스레드 넷이 각자와 공유 키에 동시 커밋)와 aup의 3샤드 서버 검사로 확인한다.
+
+## 3단계: 샤드 간 그룹 커밋과 무복사 읽기
+
+- **그룹 커밋.** 같은 때 커밋하는 스레드들은 한 커밋을 나눠 쓴다. 먼저 온 스레드가 리더가 되어 줄 선 요청을 한
+ head로 공개하고 나머지는 결과를 기다린다. 요청 단위의 원자성은 그대로다(낡은 draft가 있는 요청이나, 같은 그룹의
+ 앞 요청이 바꾼 키를 옛 리비전으로 고치려는 요청은 빠진다). 여러 스레드가 함께 커밋해 왔다면 리더는 최근 그룹
+ 크기만큼 모일 때까지 최대 2 ms 기다린다. 공개를 멈춘 사이 커밋 다섯 건이 정확히 공개 두 번이 되는지 검사한다.
+ 세 샤드, 쓰는 연결 8개에서 묶음 1,234개가 공개 421번으로 나갔다(1스레드와 처리량이 같아졌다: 327/s 대 348/s).
+- **무복사 읽기.** 저장된 문서를 읽는 응답은 복사하지 않고 저장소 매핑의 익스텐트를 `writev`로 빌려준다. 매핑
+ 참조(`pure_draft_mapping`)를 응답이 끝날 때까지 쥐므로, 보내는 도중 문서가 바뀌거나 파일이 압축으로 교체돼도
+ 읽은 버전 그대로 나간다(aup 검사: 1 MiB를 느린 소켓으로 보내는 동안 다섯 번 바꾼다).
+
+| 작업 | pure-serve 1스레드 | pure-serve 3샤드 | axum 메모리 | axum 충돌 안전 |
+|---|---:|---:|---:|---:|
+| GET 64 B, keep-alive | **127,802** (p99 544) | **151,620** (p99 608) | 125,690 (p99 768) | 126,903 |
+| GET 1 MiB, keep-alive | **2,761** | 1,884 | 1,380 | 1,446 |
+
+1 MiB 읽기는 axum의 약 2배(초당 약 2.9 GB)이고, 이 속도에서는 코어 하나로 받는 부하 생성기가 병목에 가깝다.
+
+**저장 보장 쓰기 측정의 함정.** 저장 장치는 SD 카드 하나뿐이고, 동기 쓰기가 몇 분 이어지면 내부 정리로 크게 느려진다
+(같은 설정이 순서에 따라 연결 8개에서 350/s → 17/s, p99 4.4초). 앞의 표들의 쓰기 수치는 순서 영향을 받았다. 쓰기 비교는
+설정 순서를 섞어 여러 번 돌리고, 매번 장치의 fsync 지연을 함께 재서 보고해야 한다(다음 작업).
diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c
index 349bacf..b6d6b26 100644
--- a/prototypes/connections/serve.c
+++ b/prototypes/connections/serve.c
@@ -24,6 +24,7 @@
#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
+#include <limits.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <pthread.h>
@@ -58,12 +59,15 @@ typedef struct Connection {
size_t body_size, received, in_used;
unsigned char *body;
PureTicket ticket;
- PureTransfer transfer;
+ struct iovec *out, small[4]; /* the answer: its head, then its body's extents */
+ size_t out_count, out_at;
+ PureMapping *mapping; /* keeps store extents the answer borrows mapped */
char head[320];
char in[HEADER_LIMIT]; /* the header being read, then pipelined bytes */
} Connection;
-/* One request applied in the current burst. */
+/* One request applied in the current burst. A read answered from the store
+ * lends its extents: extents point into mapping, which the answer keeps. */
typedef struct {
Connection *connection;
PureTicket ticket;
@@ -72,6 +76,9 @@ typedef struct {
int64_t revision;
PureResource resource;
PureSpan reply;
+ struct iovec *extents;
+ size_t extent_count, extent_size;
+ PureMapping *mapping;
char key[65];
} Applied;
@@ -123,12 +130,22 @@ static void release_ticket(void *owner) {
c->body = NULL;
}
+/* Everything an answer borrowed goes back: the request's payload and ticket,
+ * the store extents. */
+static void finish_output(Connection *c) {
+ release_ticket(c);
+ pure_mapping_release(c->mapping);
+ c->mapping = NULL;
+ if (c->out != c->small) free(c->out);
+ c->out = NULL;
+ c->out_count = c->out_at = 0;
+}
+
/* Events of this round may still name c: it is reused after the round. */
static void close_connection(Connection *c) {
if (c->phase == CLOSED) return;
Host *host = c->host;
- if (c->transfer.active) pure_transfer_abort(&c->transfer);
- else release_ticket(c);
+ finish_output(c);
close(c->fd); /* also leaves the epoll set */
c->phase = CLOSED;
host->table[c->index] = NULL;
@@ -137,17 +154,30 @@ static void close_connection(Connection *c) {
--host->active;
}
-static void respond(Connection *c, int status, const char *reason, const char *error, int64_t revision,
- PureSpan body) {
+/* Answer with a body of `count` extents (`size` bytes in all). The answer
+ * takes the mapping reference; 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, PureMapping *mapping) {
char tag[48] = "";
if (revision > 0) snprintf(tag, sizeof(tag), "ETag: \"%" PRId64 "\"\r\n", revision);
int length = snprintf(c->head, sizeof(c->head),
"HTTP/1.1 %d %s\r\nContent-Length: %zu\r\n%sContent-Type: application/octet-stream\r\n"
"Cache-Control: no-store\r\nX-Pure-Error: %s\r\n%s\r\n",
- status, reason, body.size, c->keep_alive ? "" : "Connection: close\r\n", error, tag);
+ status, reason, size, c->keep_alive ? "" : "Connection: close\r\n", error, tag);
+ c->mapping = mapping;
+ c->out = count < sizeof(c->small) / sizeof(*c->small) ? c->small : malloc((count + 1) * sizeof(*c->out));
c->phase = WRITING;
- pure_transfer_init(&c->transfer, (PureSpan){(const unsigned char *)c->head, (size_t)length}, body, c,
- release_ticket);
+ if (!c->out) { c->keep_alive = false; c->out = c->small; count = 0; } /* cannot send it: the close tells */
+ c->out[0] = (struct iovec){c->head, (size_t)length};
+ if (count) memcpy(c->out + 1, extents, count * sizeof(*extents));
+ c->out_count = count + 1;
+ c->out_at = 0;
+}
+
+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, NULL);
}
/* A request the host cannot read further: answer and close. */
@@ -325,25 +355,24 @@ static void drive(Connection *c) {
}
if (c->phase == WRITING) {
if (!c->writable) return;
- PureTransfer *t = &c->transfer;
- struct iovec parts[2];
- int count = 0;
- for (size_t part = t->part; part < 2; ++part) {
- size_t offset = part == t->part ? t->offset : 0;
- if (t->spans[part].size > offset)
- parts[count++] = (struct iovec){(void *)(t->spans[part].data + offset), t->spans[part].size - offset};
- }
- ssize_t n = count ? writev(c->fd, parts, count) : 0;
+ size_t left = c->out_count - c->out_at;
+ ssize_t n = writev(c->fd, c->out + c->out_at, (int)(left < IOV_MAX ? left : IOV_MAX));
if (n < 0 && errno == EAGAIN) { c->writable = false; return; }
if (n < 0 && errno == EINTR) continue;
if (n < 0) { close_connection(c); return; }
- for (size_t left = (size_t)n; left && t->active;) {
- PureSpan next = pure_transfer_next(t, left);
- if (!pure_transfer_consume(t, next.size)) abort();
- left -= next.size;
+ for (size_t sent = (size_t)n; c->out_at < c->out_count;) {
+ struct iovec *v = &c->out[c->out_at];
+ if (sent < v->iov_len) {
+ v->iov_base = (char *)v->iov_base + sent;
+ v->iov_len -= sent;
+ break;
+ }
+ sent -= v->iov_len;
+ ++c->out_at;
}
- if (t->active) continue;
+ if (c->out_at < c->out_count) continue;
/* Answered: the next request may already be here. */
+ finish_output(c);
c->deadline = now_ms() + c->host->timeout_ms;
if (!c->keep_alive) { close_connection(c); return; }
c->phase = READING_HEAD;
@@ -377,6 +406,17 @@ static bool measure_reply(void *context, PureSpan span) {
return true;
}
+/* The reply's extents, if the store mapping holds every one. */
+typedef struct { struct iovec *extents; size_t count; const PureMapping *mapping; bool held; } Gather;
+
+static bool gather_extents(void *context, PureSpan span) {
+ Gather *g = context;
+ if (!span.size) return true;
+ g->held = g->held && pure_mapping_holds(g->mapping, span);
+ if (g->held) g->extents[g->count++] = (struct iovec){(void *)span.data, span.size};
+ return g->held;
+}
+
static bool copy_reply(void *context, PureSpan span) {
Reply *r = context;
if (span.size) memcpy(r->bytes + r->size, span.data, span.size);
@@ -405,6 +445,20 @@ static void apply_effect(Host *host, const PureEffect *effect, Applied *out) {
}
if (reply.size > effect->available) { out->status = PURE_BUDGET; return; }
if (!reply.size) return;
+ /* A stored document is lent, extent by extent, not copied. */
+ PureMapping *mapping = app->document ? pure_draft_mapping(app->document) : NULL;
+ if (mapping) {
+ Gather gather = {malloc(reply.spans * sizeof(struct iovec)), 0, mapping, true};
+ if (gather.extents && pure_app_reply(app, gather_extents, &gather) == PURE_OK && gather.held) {
+ out->extents = gather.extents;
+ out->extent_count = gather.count;
+ out->extent_size = reply.size;
+ out->mapping = mapping;
+ return;
+ }
+ free(gather.extents);
+ pure_mapping_release(mapping);
+ }
unsigned char *copy = malloc(reply.size);
if (!copy) { out->status = PURE_ALLOCATION; return; }
reply = (Reply){.bytes = copy};
@@ -423,10 +477,19 @@ static void finish_burst(Host *host, size_t count) {
if (!pure_runtime_complete(a->ticket, status, a->input == PURE_INPUT_READ ? &a->resource : NULL,
status == PURE_OK ? a->reply : (PureSpan){0}, a->revision)) abort();
Connection *c = a->connection;
- if (!c || c->phase != WAITING || c->ticket.generation != a->ticket.generation) continue;
- PureOutcome result;
- if (!pure_runtime_result(a->ticket, &result)) abort();
- answer(c, result);
+ bool waiting = c && c->phase == WAITING && c->ticket.generation == a->ticket.generation;
+ if (waiting && status == PURE_OK && a->mapping) {
+ respond_extents(c, 200, "OK", "none", a->revision, a->extents, a->extent_count, a->extent_size, a->mapping);
+ a->mapping = NULL;
+ }
+ pure_mapping_release(a->mapping);
+ free(a->extents);
+ if (!waiting) continue;
+ if (c->phase == WAITING) {
+ PureOutcome result;
+ if (!pure_runtime_result(a->ticket, &result)) abort();
+ answer(c, result);
+ }
drive(c); /* most answers fit the socket buffer at once */
}
}
@@ -515,7 +578,9 @@ static void accept_all(Host *host) {
c->readable = c->writable = c->keep_alive = true;
c->deadline = now_ms() + host->timeout_ms;
c->ticket = (PureTicket){0};
- c->transfer = (PureTransfer){0};
+ c->out = NULL;
+ c->out_count = c->out_at = 0;
+ c->mapping = NULL;
c->body = NULL;
c->in_used = 0;
struct epoll_event event = {.events = EPOLLIN | EPOLLOUT | EPOLLRDHUP | EPOLLET, .data.ptr = c};
diff --git a/prototypes/connections/state.h b/prototypes/connections/state.h
index 55474c0..0945475 100644
--- a/prototypes/connections/state.h
+++ b/prototypes/connections/state.h
@@ -30,6 +30,16 @@ PureResult pure_store_open(const char *path, PureStore **out);
void pure_store_close(PureStore *);
PureResult pure_store_draft(PureStore *, const char *key, PureDraft **out);
PureResult pure_store_state(PureStore *, const char *name, PureDraft **out);
+/* Stored extents can outlive their draft: pure_draft_mapping takes a
+ * reference to the mapping a draft opened through a handle reads its stored
+ * nodes from (NULL for other drafts). Spans that mapping holds stay readable
+ * until the reference is released, after the draft closes and after the
+ * store moves on. Spans of a draft's unsaved edits are not in it. Use it on
+ * the handle's owner thread. */
+typedef struct PureMapping PureMapping;
+PureMapping *pure_draft_mapping(const PureDraft *);
+bool pure_mapping_holds(const PureMapping *, PureSpan);
+void pure_mapping_release(PureMapping *);
/* Pinned, drafts open at the version found when pinning, without checking
* the file each time (a host pins for one burst of requests); commits made
* through the handle still move it on. */
diff --git a/prototypes/connections/store-tree.c b/prototypes/connections/store-tree.c
index 5e15304..3163265 100644
--- a/prototypes/connections/store-tree.c
+++ b/prototypes/connections/store-tree.c
@@ -53,8 +53,9 @@ typedef struct {
size_t size; /* leaf bytes, child count, or next free slot + 1 */
unsigned height; /* FREE for a reclaimed slot */
} Node;
-/* One mapped version of the file, shared by the drafts of a store handle. */
-typedef struct { const unsigned char *map; uint64_t end; unsigned refs; } Mapping;
+/* One mapped version of the file, shared by the drafts of a store handle
+ * and by whoever still sends extents from it. */
+typedef struct PureMapping { const unsigned char *map; uint64_t end; unsigned refs; } Mapping;
typedef struct {
const unsigned char *map;
uint64_t end;
@@ -1141,6 +1142,19 @@ void pure_store_close(PureStore *store) {
free(store);
}
+PureMapping *pure_draft_mapping(const PureDraft *draft) {
+ Mapping *mapping = draft->tree.shared;
+ if (mapping) ++mapping->refs;
+ return mapping;
+}
+
+bool pure_mapping_holds(const PureMapping *mapping, PureSpan span) {
+ uintptr_t base = (uintptr_t)mapping->map, at = (uintptr_t)span.data;
+ return !span.size || (span.data && at >= base && at - base <= mapping->end && span.size <= mapping->end - (at - base));
+}
+
+void pure_mapping_release(PureMapping *mapping) { release_mapping(mapping); }
+
PureResult pure_store_pin(PureStore *store, bool pinned) {
store->pinned = false;
PureResult result = pinned ? refresh(store) : PURE_OK;
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 11:24:33 +0000