From d2adae5deacc1545cd5dbdfcc5b80697c91dac09 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Sat, 26 Sep 2026 22:03:08 +0100 Subject: Answer a round's reads before its writes on the owner With durable writes going on, a read waited for the save of the burst it landed in. A round's requests come from different connections, one request each at a time, so nothing orders them: the owner now applies a round's reads first and answers them at once, seeing only saved states, then its writes in bursts. 8 writers and 56 readers of the scoreboard, owner mode: reads 7,100-7,700 to 16,100-16,600 a second, their p99 55 to 13 ms, writes 1,000-1,100 to 2,300 a second. A single thread keeps the order: it waits out each save itself, and reads answered first come back first and split its next writes over more saves (the owner gathers after a save instead); there reads are 12,800-13,400 a second as before. --- prototypes/connections/serve.c | 36 ++++++++++++++++++++++++++---------- 1 file changed, 26 insertions(+), 10 deletions(-) diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c index 5e11f67..82c93cb 100644 --- a/prototypes/connections/serve.c +++ b/prototypes/connections/serve.c @@ -127,7 +127,8 @@ typedef struct { uint64_t bursts, burst_events; Applied *applied; /* a round's requests, grown to fit */ size_t applied_capacity; - bool wrote; /* the last round saved an event or an edit */ + bool wrote; /* the last round saved an event or an edit */ + bool reads_first; /* the owner's: a round's reads before its writes */ double next[PURE_MODULE_INPUTS]; /* when each timer ticks next (wall clock, ms) */ uint64_t ticks; } Executor; @@ -1068,7 +1069,16 @@ static void save_burst(Executor *ex, Applied *applied, size_t count) { /* Run a round's requests in bursts. A burst ends before a document's key * repeats or before its documents would outgrow one save, so every request * is stored by its burst's save or fails with it; views and events of the - * app's states join any burst. Each saved burst goes to finish. */ + * app's states join any burst. Each group of answers goes to finish. + * + * The owner takes a round's reads first and answers them at once: a + * round's requests come from different connections (each has one request + * at a time), so nothing orders them, and placed before the writes the + * reads see only saved states (a round starts from a save). With writes + * going on this halves the reads' wait and doubles both rates. A single + * thread does not: it waits out each save itself, and reads answered first + * come back first, splitting its next writes over more saves (the owner + * gathers after a save instead). */ static void execute(Executor *ex, const Job *jobs, size_t count, void (*finish)(void *, Applied *, size_t), void *context) { if (!count) return; @@ -1077,22 +1087,27 @@ static void execute(Executor *ex, const Job *jobs, size_t count, void (*finish)( /* One look at the store per round: a request made after another shard's * answer was sent is read after that answer's commit. */ bool pinned = pure_store_pin(ex->store, true) == PURE_OK; - size_t start = 0, keys = 0; + size_t reads = 0; + for (size_t i = 0; ex->reads_first && i < count; ++i) + if (jobs[i].effect.input == PURE_INPUT_READ) apply_effect(ex, &jobs[i], &applied[reads++]); + if (reads) finish(context, applied, reads); + size_t n = reads, start = reads, keys = 0; for (size_t i = 0; i < count; ++i) { + if (ex->reads_first && jobs[i].effect.input == PURE_INPUT_READ) continue; bool document = jobs[i].route.kind == ROUTE_OBJECT, repeated = false; - for (size_t k = start; k < i && document && !repeated; ++k) + for (size_t k = start; k < n && document && !repeated; ++k) repeated = applied[k].route == ROUTE_OBJECT && !strcmp(applied[k].key, jobs[i].effect.key); if (repeated || (document && keys == PURE_APP_PENDING)) { - save_burst(ex, applied + start, i - start); - finish(context, applied + start, i - start); - start = i; + save_burst(ex, applied + start, n - start); + finish(context, applied + start, n - start); + start = n; keys = 0; } - apply_effect(ex, &jobs[i], &applied[i]); + apply_effect(ex, &jobs[i], &applied[n++]); keys += document; } - save_burst(ex, applied + start, count - start); - finish(context, applied + start, count - start); + save_burst(ex, applied + start, n - start); + finish(context, applied + start, n - start); if (pinned) pure_store_pin(ex->store, false); } @@ -1527,6 +1542,7 @@ int main(int argc, char **argv) { if (owned) { if (!start_executor(&owner.executor, &module, show, edit, store) || pthread_mutex_init(&owner.lock, NULL) || pthread_cond_init(&owner.waiting, NULL)) return 1; + owner.executor.reads_first = true; owner.hosts = hosts; owner.host_count = threads; } -- cgit v1.2.3