Markk116/urus-benches
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bench: port urus memory store to gen_server; add spinning A/B harness

Browse Source 
- store.rs: MemStore now implements GenServer (one inbox, one hop/request),
  replacing the AnyReq + two-forwarder fan-in that cost two extra park/unpark
  hops per request and contaminated the spin signal. StoreHandle is now an
  enum {Memory(ServerRef<MemStore>) | Sqlite{reads,writes}} with one call().
  Response status/body semantics unchanged (S3 stays comparable).
- SQLite store left on raw actors (single-writer + reader pool); not ported
  (single-inbox would serialise S4 reads) and not part of the spin A/B. Kept
  behind the handle so --store=sqlite still builds.
- handlers.rs: 6 handlers collapse to state.store().call(StoreReq::..); the
  per-handler channel+send+recv boilerplate is gone. main.rs untouched.
- spin_ab_urus.sh + analyze_spin_ab_urus.py: A/B the spin policy on urus by
  checking out smarm b0c9685 (pre-spin) vs HEAD, S2/S3 on urus-mem, server
  cores 1/2/4/8 (2/4 gate-active, 1/8 inert), printing RPS speedup + latency
  deltas. Verified: builds release; mem+sqlite smoke pass; scripts dry-run +
  analyzer fixture-checked. Real sweep needs the multi-core box.
This commit is contained in:
sandbox-agent
2026-06-14 21:07:52 +00:00
parent 44ee8dcf4e
commit 56e490c205
4 changed files with 558 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files
analyze_spin_ab_urus.py
Executable
+142
View File
@@ -0,0 +1,142 @@
#!/usr/bin/env python3
"""Pair before/after urus-bench results and print an A/B comparison.
Layout produced by spin_ab_urus.sh:
<outdir>/before/<server>-<scenario>-c<cores>/result.json
<outdir>/after/<server>-<scenario>-c<cores>/result.json
We pair by the (server, scenario, cores) key and report sustained RPS
(after/before speedup) and p50/p99/p99.9 latency deltas. The spin gate is
active only at 2-4 schedulers, so rows at cores 2 and 4 are where an effect is
expected; cores 1 and 8 are controls (gate inert) and should stay flat.
"""
import argparse
import json
import os
import re
import sys
GATED = {2, 4} # core counts where the RFC 004 spin policy is active
def parse_latency_ms(s):
"""wrk2 prints '0.50ms' / '1.83ms' / '24.50ms' / '1.20s' / '850.00us'."""
if s is None:
return None
s = str(s).strip()
m = re.match(r"^([0-9.]+)\s*(us|ms|s)?$", s)
if not m:
return None
val = float(m.group(1))
unit = m.group(2) or "ms"
return {"us": val / 1000.0, "ms": val, "s": val * 1000.0}[unit]
def load(outdir, sub):
"""Return {(server, scenario, cores): result_dict} for one side."""
base = os.path.join(outdir, sub)
out = {}
if not os.path.isdir(base):
return out
for name in os.listdir(base):
rj = os.path.join(base, name, "result.json")
if not os.path.isfile(rj):
continue
m = re.match(r"^(.*)-(s\d)-c(\d+)$", name)
if not m:
continue
server, scenario, cores = m.group(1), m.group(2), int(m.group(3))
try:
with open(rj) as f:
out[(server, scenario, cores)] = json.load(f)
except (OSError, json.JSONDecodeError) as e:
print(f"warn: could not read {rj}: {e}", file=sys.stderr)
return out
def fnum(x):
try:
return float(x)
except (TypeError, ValueError):
return None
def main():
ap = argparse.ArgumentParser()
ap.add_argument("outdir")
ap.add_argument("--before", default="before")
ap.add_argument("--after", default="after")
args = ap.parse_args()
before = load(args.outdir, args.before)
after = load(args.outdir, args.after)
keys = sorted(set(before) | set(after))
if not keys:
print("no results found under", args.outdir, file=sys.stderr)
return 1
# --- RPS table ---------------------------------------------------------
hdr = f"{'server':<10} {'scen':<4} {'cores':>5} {'before rps':>12} {'after rps':>12} {'speedup':>8} gate"
print("\n== Sustained RPS (higher is better) ==")
print(hdr)
print("-" * len(hdr))
for k in keys:
srv, scn, cores = k
b = fnum(before.get(k, {}).get("sustained_rps"))
a = fnum(after.get(k, {}).get("sustained_rps"))
spd = f"{a / b:.2f}x" if (a and b) else "—"
gate = "ACTIVE" if cores in GATED else "inert"
bs = f"{b:,.0f}" if b is not None else "—"
as_ = f"{a:,.0f}" if a is not None else "—"
print(f"{srv:<10} {scn:<4} {cores:>5} {bs:>12} {as_:>12} {spd:>8} {gate}")
# --- latency tables ----------------------------------------------------
for pct in ("p50", "p99", "p999"):
print(f"\n== {pct} latency, ms (lower is better) ==")
h = f"{'server':<10} {'scen':<4} {'cores':>5} {'before':>10} {'after':>10} {'delta':>9} gate"
print(h)
print("-" * len(h))
for k in keys:
srv, scn, cores = k
b = parse_latency_ms(before.get(k, {}).get("latency", {}).get(pct))
a = parse_latency_ms(after.get(k, {}).get("latency", {}).get(pct))
if a is not None and b is not None:
delta = f"{(a - b):+.2f}"
else:
delta = "—"
gate = "ACTIVE" if cores in GATED else "inert"
bs = f"{b:.2f}" if b is not None else "—"
as_ = f"{a:.2f}" if a is not None else "—"
print(f"{srv:<10} {scn:<4} {cores:>5} {bs:>10} {as_:>10} {delta:>9} {gate}")
# --- error sanity (non-2xx should be ~0 on both sides) -----------------
noisy = []
for k in keys:
for side, d in (("before", before), ("after", after)):
n = d.get(k, {}).get("non_2xx", 0)
if isinstance(n, (int, float)) and n > 0:
noisy.append((side, k, n))
if noisy:
print("\n== WARNING: non-2xx responses (results may be unreliable) ==")
for side, (srv, scn, cores), n in noisy:
print(f" {side} {srv} {scn} c{cores}: {n} non-2xx")
print(
"\nReading it: spin is 'worth it' if s3 (and/or s2) RPS rises and/or "
"tail latency drops at cores 2/4 (gate ACTIVE), with cores 1/8 (inert) "
"staying flat. Flat-or-worse at 2/4 = not worth it on real traffic."
)
if "axum-mem" in {k[0] for k in keys}:
print(
"axum-mem is the control: it doesn't depend on smarm, so its "
"before/after should be ~identical. Any drift there means the run "
"environment moved, not the patch."
)
return 0
if __name__ == "__main__":
sys.exit(main())
spin_ab_urus.sh
Executable
+223
View File
@@ -0,0 +1,223 @@
#!/usr/bin/env bash
# spin_ab_urus.sh — A/B the RFC 004 spinning policy on the *realistic* urus
# workload (not the multi_scheduler microbench).
#
# It does exactly what the handoff asks: check out the pre-spin smarm commit,
# run the urus benches; check out the latest smarm commit, run them again;
# then print a before/after comparison.
#
# BEFORE = b0c9685 (pre-spin: no spinning code exists at all)
# AFTER = current smarm HEAD (spin defaults on + N=1 fix; e.g. 2f011f6)
#
# smarm is a *path dependency* of urus, which is a path dependency of
# urus-benches (all three are siblings). So we A/B by checking out smarm at
# each sha and rebuilding urus-benches; urus itself is held fixed — only the
# smarm under it changes between A and B. Each (commit, scenario, cores) run
# goes through the existing runner.sh (saturation probe -> warmup -> wrk2
# latency-honest measure) and lands a result.json; analyze_spin_ab_urus.py
# pairs them up at the end.
#
# Scenarios (per RECON §6.4): s2 (router+auth, no store) and s3 (memory store,
# the headline). Server is urus-mem for both. We do NOT touch s4/sqlite.
#
# Core sweep (per RECON §6.3): the spin gate is active at 2-4 schedulers and
# inert at 1 and >=5, so 2 and 4 should show the effect and 1 and 8 bracket it
# as controls. Server is pinned to cores 0..N-1; the load generator gets its
# own disjoint cores (LOADGEN_CPUS).
#
# Usage:
# ./spin_ab_urus.sh # full A/B, defaults
# ./spin_ab_urus.sh --quick # short timings (15s phases)
# ./spin_ab_urus.sh --dry-run # print the plan, run nothing
# CORES="2 4" SCENARIOS="s3" ./spin_ab_urus.sh
# BEFORE=b0c9685 AFTER=2f011f6 ./spin_ab_urus.sh
# ./spin_ab_urus.sh --with-axum-control # also run axum-mem (smarm-free
# # invariant; should be flat A vs B)
#
# Env knobs (all overridable):
# BEFORE / AFTER smarm commit-ishes (default b0c9685 / HEAD)
# SCENARIOS space-separated (default "s2 s3")
# CORES server core counts to sweep (default "1 2 4 8")
# LOADGEN_CPUS loadgen taskset range (default "12-19")
# SERVER memory server backend (default urus-mem)
# PROBE_SEC/WARMUP_SEC/MEASURE_SEC/SAT_RATIO/PREPOP/WRK_CONNS/WRK_THREADS
# passed straight through to runner.sh
# SMARM_DIR override smarm location (default: sibling ../smarm)
# OUTDIR results dir (default: results/spin_ab_urus_<ts>)
set -euo pipefail
# ---------------------------------------------------------------------------
# Locate the pieces
# ---------------------------------------------------------------------------
BENCH_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" # urus-benches
RUNNER="${BENCH_DIR}/runner.sh"
ANALYZER="${BENCH_DIR}/analyze_spin_ab_urus.py"
SMARM_DIR="${SMARM_DIR:-$(cd "${BENCH_DIR}/../smarm" 2>/dev/null && pwd || true)}"
if [[ ! -x "$RUNNER" ]]; then
echo "ERROR: runner.sh not found/executable at $RUNNER" >&2; exit 1
fi
if [[ -z "$SMARM_DIR" || ! -d "$SMARM_DIR/.git" ]]; then
echo "ERROR: smarm git repo not found (looked for sibling ../smarm). Set SMARM_DIR." >&2
exit 1
fi
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
BEFORE="${BEFORE:-b0c9685}"
# AFTER defaults to whatever smarm is currently on ("the latest").
AFTER_DEFAULT="$(git -C "$SMARM_DIR" rev-parse --short HEAD)"
AFTER="${AFTER:-$AFTER_DEFAULT}"
SCENARIOS="${SCENARIOS:-s2 s3}"
CORES="${CORES:-1 2 4 8}"
SERVER="${SERVER:-urus-mem}"
LOADGEN_CPUS="${LOADGEN_CPUS:-12-19}"
QUICK=0
DRY=0
WITH_AXUM=0
for arg in "$@"; do
case "$arg" in
--quick) QUICK=1 ;;
--dry-run|-n) DRY=1 ;;
--with-axum-control) WITH_AXUM=1 ;;
-h|--help) sed -n '2,/^set -euo/p' "$0" | sed 's/^# \{0,1\}//; $d'; exit 0 ;;
*) echo "unknown arg: $arg" >&2; exit 2 ;;
esac
done
if [[ "$QUICK" == "1" ]]; then
export PROBE_SEC="${PROBE_SEC:-15}"
export WARMUP_SEC="${WARMUP_SEC:-15}"
export MEASURE_SEC="${MEASURE_SEC:-15}"
fi
TS="$(date +%Y%m%d-%H%M%S)"
OUTDIR="${OUTDIR:-${BENCH_DIR}/results/spin_ab_urus_${TS}}"
NPROC="$(nproc)"
# Loadgen low core, for an overlap sanity check against the server range.
LG_LO="${LOADGEN_CPUS%%-*}"
SERVERS_TO_RUN=("$SERVER")
[[ "$WITH_AXUM" == "1" ]] && SERVERS_TO_RUN+=("axum-mem")
echo ">>> smarm: $SMARM_DIR"
echo ">>> BEFORE=$BEFORE AFTER=$AFTER"
echo ">>> servers: ${SERVERS_TO_RUN[*]}"
echo ">>> scenarios: $SCENARIOS"
echo ">>> cores: $CORES (loadgen on $LOADGEN_CPUS)"
echo ">>> machine: $NPROC cores"
echo ">>> outdir: $OUTDIR"
[[ "$QUICK" == "1" ]] && echo ">>> --quick: probe/warmup/measure = ${PROBE_SEC}/${WARMUP_SEC}/${MEASURE_SEC}s"
echo
# server core range for a given count: "0" for 1, else "0-(N-1)"
server_cpus() { (( $1 == 1 )) && echo "0" || echo "0-$(( $1 - 1 ))"; }
# ---------------------------------------------------------------------------
# Safety: refuse a dirty smarm tree; restore original ref on exit
# ---------------------------------------------------------------------------
if ! git -C "$SMARM_DIR" diff-index --quiet HEAD -- 2>/dev/null; then
echo "ERROR: smarm working tree is dirty. Commit/stash before A/B so checkouts are safe." >&2
git -C "$SMARM_DIR" status --short >&2
exit 1
fi
ORIG_REF="$(git -C "$SMARM_DIR" symbolic-ref --quiet --short HEAD || git -C "$SMARM_DIR" rev-parse HEAD)"
restore_smarm() {
echo
echo ">>> restoring smarm to $ORIG_REF"
git -C "$SMARM_DIR" checkout --quiet "$ORIG_REF" 2>/dev/null \
|| echo "WARN: could not restore $ORIG_REF — smarm is on $(git -C "$SMARM_DIR" rev-parse --short HEAD)" >&2
}
[[ "$DRY" == "0" ]] && trap restore_smarm EXIT
# Overlap sanity: server top core must be below the loadgen range.
MAX_SRV=0; for c in $CORES; do (( c > MAX_SRV )) && MAX_SRV=$c; done
if [[ "$LG_LO" =~ ^[0-9]+$ ]] && (( MAX_SRV - 1 >= LG_LO )); then
echo "WARN: server cores reach $((MAX_SRV-1)) which overlaps LOADGEN_CPUS=$LOADGEN_CPUS." >&2
echo " set LOADGEN_CPUS to a disjoint range (two-process discipline)." >&2
fi
if (( MAX_SRV > NPROC )); then
echo "WARN: sweep asks for up to ${MAX_SRV} server cores but machine has ${NPROC}." >&2
fi
mkdir -p "$OUTDIR"
# ---------------------------------------------------------------------------
# Run one commit across the (server x scenario x cores) sub-matrix
# ---------------------------------------------------------------------------
run_commit() {
local sha="$1" label="$2"
echo "============================================================"
echo ">>> $label: checking out smarm $sha and rebuilding urus-benches"
echo "============================================================"
if [[ "$DRY" == "1" ]]; then
echo " [dry] git -C $SMARM_DIR checkout $sha"
echo " [dry] (cd $BENCH_DIR && cargo build --release)"
else
git -C "$SMARM_DIR" checkout --quiet "$sha"
# urus is a path dep of urus-benches and smarm a path dep of urus, so a
# workspace release build picks up the swapped smarm. cargo refingerprints
# on the changed sources.
( cd "$BENCH_DIR" && cargo build --release ) 2>&1 | tail -3
fi
for srv in "${SERVERS_TO_RUN[@]}"; do
for scn in $SCENARIOS; do
for c in $CORES; do
local scpus; scpus="$(server_cpus "$c")"
local run_id="${label}__${srv}__${scn}__c${c}"
local dest="${OUTDIR}/${label}/${srv}-${scn}-c${c}"
echo ">>> $label $srv $scn server-cores=$c (taskset $scpus) loadgen=$LOADGEN_CPUS"
if [[ "$DRY" == "1" ]]; then
echo " [dry] SERVER_CPUS=$scpus LOADGEN_CPUS=$LOADGEN_CPUS $RUNNER $srv $scn $run_id"
continue
fi
mkdir -p "$dest"
if SERVER_CPUS="$scpus" LOADGEN_CPUS="$LOADGEN_CPUS" \
"$RUNNER" "$srv" "$scn" "$run_id" > "${dest}/runner.log" 2>&1; then
if [[ -d "${BENCH_DIR}/results/${run_id}" ]]; then
mv "${BENCH_DIR}/results/${run_id}/"* "$dest/" 2>/dev/null || true
rmdir "${BENCH_DIR}/results/${run_id}" 2>/dev/null || true
fi
local rps p99
rps=$(awk -F'[:,]' '/"sustained_rps"/ {gsub(/[ "]/,"",$2); print $2; exit}' "${dest}/result.json" 2>/dev/null || echo "?")
p99=$(awk -F'"' '/"p99"/ {print $4; exit}' "${dest}/result.json" 2>/dev/null || echo "?")
echo " -> sustained=${rps} rps p99=${p99}"
else
echo " -> FAILED (see ${dest}/runner.log)"
fi
done
done
done
}
run_commit "$BEFORE" "before"
run_commit "$AFTER" "after"
[[ "$DRY" == "1" ]] && { echo; echo ">>> dry run complete (nothing executed, smarm untouched)."; exit 0; }
# ---------------------------------------------------------------------------
# Analyze
# ---------------------------------------------------------------------------
echo
echo "============================================================"
echo ">>> A/B comparison"
echo "============================================================"
if command -v python3 >/dev/null && [[ -f "$ANALYZER" ]]; then
python3 "$ANALYZER" "$OUTDIR" --before before --after after | tee "${OUTDIR}/comparison.txt"
else
echo "(python3 or analyzer missing; raw result.json files are under $OUTDIR)"
fi
echo
echo ">>> DONE. results: $OUTDIR"
urus-server/src/handlers.rs
+24 -44
View File
@@ -1,10 +1,10 @@
//! Request handlers and the AppState that holds the store-handle factory.
//!
//! The trick: smarm requires `spawn` to be called from inside an actor.
//! The bootstrap thread that calls `serve_with` is NOT an actor — it's
//! the user's main(). So we can't spawn the store actor at startup; we
//! defer to the first request that asks for it. That request runs inside
//! a connection actor, which is a valid place to call spawn.
//! The trick: smarm requires `spawn` (and `gen_server::start`) to be called
//! from inside an actor. The bootstrap thread that calls `serve_with` is NOT
//! an actor — it's the user's main(). So we can't spawn the store actor at
//! startup; we defer to the first request that asks for it. That request runs
//! inside a connection actor, which is a valid place to spawn.
//!
//! AppState holds:
//! - a "store_init" factory that lazily spawns the store on first use
@@ -18,11 +18,10 @@
use std::sync::{Arc, OnceLock};
use smarm::channel;
use urus::{Conn, Next, Plug, Router};
use crate::middleware::LogRing;
use crate::store::{ReadReq, StoreHandle, WriteReq};
use crate::store::{StoreHandle, StoreReq};
// ---------------------------------------------------------------------------
// AppState
@@ -64,7 +63,7 @@ impl AppState {
}
// ---------------------------------------------------------------------------
// Tiny helper for writing JSON responses
// Tiny helpers
// ---------------------------------------------------------------------------
fn json(conn: Conn, status: u16, body: Vec<u8>) -> Conn {
@@ -81,13 +80,17 @@ fn text(conn: Conn, status: u16, body: &'static str) -> Conn {
fn parse_id(s: &str) -> Option<u64> { s.parse().ok() }
const STORE_UNAVAILABLE: &[u8] = b"{\"error\":\"store unavailable\"}";
// ---------------------------------------------------------------------------
// Handlers
// ---------------------------------------------------------------------------
//
// All handlers close over a clone of AppState. Each handler closure is a
// `Fn(Conn, Next) -> Conn`, which is exactly the Plug shape urus's
// blanket impl picks up.
// `Fn(Conn, Next) -> Conn`, which is exactly the Plug shape urus's blanket
// impl picks up. A request is now a single `store().call(...)`: the gen_server
// `call` builds the reply channel, sends, and parks for the reply, so the
// old per-handler channel + send + recv boilerplate is gone.
fn build_ping_handler() -> impl Plug {
// /ping is a literal endpoint: status 200, body "pong". No work.
@@ -96,13 +99,9 @@ fn build_ping_handler() -> impl Plug {
fn build_list_handler(state: AppState) -> impl Plug {
move |conn: Conn, _next: Next| {
let (tx, rx) = channel::<(u16, Vec<u8>)>();
if state.store().reads().send(ReadReq::List { reply: tx }).is_err() {
return json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec());
}
match rx.recv() {
match state.store().call(StoreReq::List) {
Ok((s, b)) => json(conn, s, b),
Err(_) => json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec()),
Err(_) => json(conn, 503, STORE_UNAVAILABLE.to_vec()),
}
}
}
@@ -113,31 +112,20 @@ fn build_get_one_handler(state: AppState) -> impl Plug {
Some(id) => id,
None => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()),
};
let (tx, rx) = channel::<(u16, Vec<u8>)>();
if state.store().reads().send(ReadReq::Get { id, reply: tx }).is_err() {
return json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec());
}
match rx.recv() {
match state.store().call(StoreReq::Get { id }) {
Ok((s, b)) => json(conn, s, b),
Err(_) => json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec()),
Err(_) => json(conn, 503, STORE_UNAVAILABLE.to_vec()),
}
}
}
fn build_create_handler(state: AppState) -> impl Plug {
move |conn: Conn, _next: Next| {
// Take the body out of the request side. `Body::into_bytes`
// would need to consume `conn`, which we still need; clone the
// bytes instead. (For a 100-byte JSON object this is fine; for
// a large upload we'd want to redesign.)
// Clone the body bytes out; we still need `conn` to build the reply.
let body = conn.body.as_bytes().to_vec();
let (tx, rx) = channel::<(u16, Vec<u8>)>();
if state.store().writes().send(WriteReq::Create { body, reply: tx }).is_err() {
return json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec());
}
match rx.recv() {
match state.store().call(StoreReq::Create { body }) {
Ok((s, b)) => json(conn, s, b),
Err(_) => json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec()),
Err(_) => json(conn, 503, STORE_UNAVAILABLE.to_vec()),
}
}
}
@@ -149,13 +137,9 @@ fn build_update_handler(state: AppState) -> impl Plug {
None => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()),
};
let body = conn.body.as_bytes().to_vec();
let (tx, rx) = channel::<(u16, Vec<u8>)>();
if state.store().writes().send(WriteReq::Update { id, body, reply: tx }).is_err() {
return json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec());
}
match rx.recv() {
match state.store().call(StoreReq::Update { id, body }) {
Ok((s, b)) => json(conn, s, b),
Err(_) => json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec()),
Err(_) => json(conn, 503, STORE_UNAVAILABLE.to_vec()),
}
}
}
@@ -166,13 +150,9 @@ fn build_delete_handler(state: AppState) -> impl Plug {
Some(id) => id,
None => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()),
};
let (tx, rx) = channel::<(u16, Vec<u8>)>();
if state.store().writes().send(WriteReq::Delete { id, reply: tx }).is_err() {
return json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec());
}
match rx.recv() {
match state.store().call(StoreReq::Delete { id }) {
Ok((s, b)) => json(conn, s, b),
Err(_) => json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec()),
Err(_) => json(conn, 503, STORE_UNAVAILABLE.to_vec()),
}
}
}
urus-server/src/store.rs
+169 -209
View File
@@ -1,43 +1,58 @@
//! Store actors for the urus-server bench.
//!
//! Two implementations, same protocol:
//! - Memory store: one actor owns a Vec<User>.
//! - SQLite store: one writer actor + a pool of reader actors. Reads
//! are dispatched through a single MPSC `Receiver` shared by all
//! readers — smarm's MPSC gives us "first available reader wins" for
//! free.
//! Two implementations behind one [`StoreHandle`]:
//! - **Memory store** (S1/S2/S3): one `gen_server` actor owning a
//! `Vec<User>`. Reads and writes share the single gen_server inbox, so a
//! request is exactly one park/unpark round-trip — the path the RFC 004
//! spinning patch perturbs. This is the scenario we actually want clean.
//! - **SQLite store** (S4): one writer actor + a pool of reader actors over
//! raw `channel` actors. NOT ported to gen_server (single-inbox would
//! serialise the reader pool and kill S4 read parallelism); left exactly
//! as it was. It is not part of the spinning A/B and is kept only so the
//! binary still builds with `--store=sqlite`.
//!
//! `StoreHandle` is what handlers hold. It's cheap-cloneable (it wraps a
//! pair of smarm Senders) and carries the routing logic for splitting
//! requests across the writer and the reader pool.
//! `StoreHandle` is what handlers hold. It's cheap-cloneable and exposes a
//! single `call(StoreReq) -> Reply` that both backends answer.
use smarm::{channel, Receiver, Sender};
use std::sync::Arc;
use smarm::{channel, GenServer, Receiver, Sender, ServerBuilder, ServerRef};
use std::path::Path;
use std::sync::Arc;
use rusqlite::{params, Connection, OpenFlags};
// ---------------------------------------------------------------------------
// Wire types — what the store returns over the reply channel.
// Wire types
// ---------------------------------------------------------------------------
//
// We send (status, body_bytes) so handlers can hand them straight to
// We return (status, body_bytes) so handlers can hand them straight to
// `Conn::put_body` without re-serialising. The store does the JSON work.
pub type Reply = (u16, Vec<u8>);
// Debug is intentionally not derived: `smarm::Sender<T>` doesn't implement
// Debug, and these enums carry one. We never print them.
/// The unified store request. One variant per route. Unlike the old
/// `ReadReq`/`WriteReq`, these carry *no* reply channel — gen_server's `call`
/// creates and threads the reply for us, so the request is just the payload.
pub enum StoreReq {
List,
Get { id: u64 },
Create { body: Vec<u8> },
Update { id: u64, body: Vec<u8> },
Delete { id: u64 },
}
// The SQLite backend still routes reads vs writes to two different actor
// sets, so it keeps the split request enums *with* their reply senders.
// These are now used ONLY by the SQLite store internals.
pub enum ReadReq {
List { reply: Sender<Reply> },
Get { id: u64, reply: Sender<Reply> },
List { reply: Sender<Reply> },
Get { id: u64, reply: Sender<Reply> },
}
pub enum WriteReq {
Create { body: Vec<u8>, reply: Sender<Reply> },
Update { id: u64, body: Vec<u8>, reply: Sender<Reply> },
Delete { id: u64, reply: Sender<Reply> },
Create { body: Vec<u8>, reply: Sender<Reply> },
Update { id: u64, body: Vec<u8>, reply: Sender<Reply> },
Delete { id: u64, reply: Sender<Reply> },
}
// ---------------------------------------------------------------------------
@@ -61,187 +76,149 @@ struct NewUser {
// Public handle
// ---------------------------------------------------------------------------
//
// Both implementations of the store give back a StoreHandle. Internally
// we always have a "read side" and a "write side". For the memory store
// they're the same actor (so the writer Sender is just a clone of the
// reader Sender wrapped in a small adapter); we keep the split anyway for
// API uniformity.
// Both backends produce a StoreHandle. Memory wraps a gen_server ServerRef;
// SQLite wraps the read/write Sender pair. `call` hides the difference so
// handlers issue one uniform request regardless of backend.
#[derive(Clone)]
pub struct StoreHandle {
reads: Sender<ReadReq>,
writes: Sender<WriteReq>,
pub enum StoreHandle {
Memory(ServerRef<MemStore>),
Sqlite { reads: Sender<ReadReq>, writes: Sender<WriteReq> },
}
impl StoreHandle {
pub fn reads(&self) -> &Sender<ReadReq> { &self.reads }
pub fn writes(&self) -> &Sender<WriteReq> { &self.writes }
}
// ---------------------------------------------------------------------------
// Memory store — one actor, one Vec.
// ---------------------------------------------------------------------------
//
// The memory store collapses reads and writes into one mailbox. We bridge
// the two-channel API by spawning a tiny *fan-in* actor whose job is to
// forward ReadReq | WriteReq into the single backing actor's mailbox.
//
// Actually, simpler: we make the backing actor accept an `enum AnyReq`
// and spawn two forwarder loops. That avoids changing the public API or
// inventing a `try_recv_either`. The cost is two extra hops per request,
// which is fine for the memory store — it's the *baseline*, not the
// target of the bench.
//
// Wait — that's two extra context switches *per request*, and the
// memory tier exists specifically to expose actor-message-pass costs.
// Tainting it with extra hops would make the bench lie. Let's do it
// properly: one actor, a `select`-shaped recv. smarm doesn't have a
// multi-channel select primitive (per the README it's a single-mailbox
// model), but we *can* have one actor own one channel that carries
// `enum Req { Read(...), Write(...) }` and expose the two Senders by
// mapping at the boundary. The mapping happens in a couple of cheap
// forwarder actors. For a *pure* in-memory benchmark we still pay one
// extra hop per request — fair across all paths.
//
// Decision: take the one extra hop for clean code. The actor-message
// cost we want to measure is "handler -> store -> handler", and the
// forwarder just makes the bookkeeping uniform. We document it.
enum AnyReq {
Read(ReadReq),
Write(WriteReq),
}
pub fn spawn_memory_store() -> StoreHandle {
let (any_tx, any_rx) = channel::<AnyReq>();
// The backing actor.
let core_tx = any_tx.clone();
let _ = core_tx; // keep alive via forwarders below
smarm::spawn(move || memory_store_loop(any_rx));
// Forwarder: ReadReq -> AnyReq::Read.
let (reads_tx, reads_rx) = channel::<ReadReq>();
{
let any_tx = any_tx.clone();
smarm::spawn(move || forward_reads(reads_rx, any_tx));
}
// Forwarder: WriteReq -> AnyReq::Write.
let (writes_tx, writes_rx) = channel::<WriteReq>();
{
let any_tx = any_tx.clone();
smarm::spawn(move || forward_writes(writes_rx, any_tx));
}
StoreHandle { reads: reads_tx, writes: writes_tx }
}
fn forward_reads(rx: Receiver<ReadReq>, tx: Sender<AnyReq>) {
while let Ok(r) = rx.recv() {
if tx.send(AnyReq::Read(r)).is_err() { return; }
}
}
fn forward_writes(rx: Receiver<WriteReq>, tx: Sender<AnyReq>) {
while let Ok(w) = rx.recv() {
if tx.send(AnyReq::Write(w)).is_err() { return; }
}
}
fn memory_store_loop(rx: Receiver<AnyReq>) {
let mut users: Vec<User> = Vec::with_capacity(1024);
let mut next_id: u64 = 1;
loop {
let req = match rx.recv() {
Ok(r) => r,
Err(_) => return, // all forwarders dropped
};
match req {
AnyReq::Read(ReadReq::List { reply }) => {
// Cap at 100 per the spec; reuse a Vec via collect.
let snapshot: Vec<&User> = users.iter().take(100).collect();
let body = serde_json::to_vec(&snapshot).unwrap_or_else(|_| b"[]".to_vec());
let _ = reply.send((200, body));
}
AnyReq::Read(ReadReq::Get { id, reply }) => {
match users.iter().find(|u| u.id == id) {
Some(u) => {
let body = serde_json::to_vec(u).unwrap();
let _ = reply.send((200, body));
}
None => {
let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec()));
}
}
}
AnyReq::Write(WriteReq::Create { body, reply }) => {
match serde_json::from_slice::<NewUser>(&body) {
Ok(nu) => {
let u = User { id: next_id, name: nu.name, email: nu.email };
next_id += 1;
users.push(u.clone());
// The bench cares about *steady-state* RPS; let
// the in-memory store cap itself so we don't
// grow to Vec<infinity>.
if users.len() > 100_000 {
users.drain(..50_000);
}
let _ = reply.send((201, serde_json::to_vec(&u).unwrap()));
}
Err(_) => {
let _ = reply.send((400, b"{\"error\":\"invalid body\"}".to_vec()));
}
}
}
AnyReq::Write(WriteReq::Update { id, body, reply }) => {
match serde_json::from_slice::<NewUser>(&body) {
Ok(nu) => match users.iter_mut().find(|u| u.id == id) {
Some(u) => {
u.name = nu.name;
u.email = nu.email;
let snap = u.clone();
let _ = reply.send((200, serde_json::to_vec(&snap).unwrap()));
}
None => {
let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec()));
}
},
Err(_) => {
let _ = reply.send((400, b"{\"error\":\"invalid body\"}".to_vec()));
}
}
}
AnyReq::Write(WriteReq::Delete { id, reply }) => {
let before = users.len();
users.retain(|u| u.id != id);
if users.len() < before {
let _ = reply.send((204, Vec::new()));
} else {
let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec()));
}
/// Issue a request and wait for the reply. `Err(())` means the store is
/// unreachable (server gone / channel closed); handlers turn that into a
/// 503. The memory path is a single gen_server `call`; the SQLite path
/// builds a one-shot reply channel and routes to the writer or a reader.
pub fn call(&self, req: StoreReq) -> Result<Reply, ()> {
match self {
StoreHandle::Memory(r) => r.call(req).map_err(|_| ()),
StoreHandle::Sqlite { reads, writes } => {
let (tx, rx) = channel::<Reply>();
// Map each backend's distinct SendError<T> to () so the arms
// unify, then `?` on the common Result<(), ()>.
let sent: Result<(), ()> = match req {
StoreReq::List => reads.send(ReadReq::List { reply: tx }).map_err(|_| ()),
StoreReq::Get { id } => reads.send(ReadReq::Get { id, reply: tx }).map_err(|_| ()),
StoreReq::Create { body } => writes.send(WriteReq::Create { body, reply: tx }).map_err(|_| ()),
StoreReq::Update { id, body } => writes.send(WriteReq::Update { id, body, reply: tx }).map_err(|_| ()),
StoreReq::Delete { id } => writes.send(WriteReq::Delete { id, reply: tx }).map_err(|_| ()),
};
sent?;
rx.recv().map_err(|_| ())
}
}
}
}
// ---------------------------------------------------------------------------
// SQLite store — one writer + N readers.
// Memory store — one gen_server, one Vec.
// ---------------------------------------------------------------------------
//
// Everything is a `Call`: the HTTP handler needs the reply to build the
// response, so there is no fire-and-forget path here (`type Cast = ()`,
// unused). One inbox, one hop per request — no forwarder actors, which is the
// whole point of the port: the old raw-actor store needed two forwarders to
// fake a read/write select over smarm's single mailbox, costing two extra
// park/unpark hops per request and contaminating exactly the cost the spin
// patch moves.
pub struct MemStore {
users: Vec<User>,
next_id: u64,
}
impl MemStore {
fn new() -> Self {
MemStore { users: Vec::with_capacity(1024), next_id: 1 }
}
}
impl GenServer for MemStore {
type Call = StoreReq;
type Reply = Reply;
type Cast = ();
type Info = ();
fn handle_call(&mut self, req: StoreReq) -> Reply {
match req {
StoreReq::List => {
// Cap at 100 per the spec.
let snapshot: Vec<&User> = self.users.iter().take(100).collect();
let body = serde_json::to_vec(&snapshot).unwrap_or_else(|_| b"[]".to_vec());
(200, body)
}
StoreReq::Get { id } => {
match self.users.iter().find(|u| u.id == id) {
Some(u) => (200, serde_json::to_vec(u).unwrap()),
None => (404, b"{\"error\":\"not found\"}".to_vec()),
}
}
StoreReq::Create { body } => {
match serde_json::from_slice::<NewUser>(&body) {
Ok(nu) => {
let u = User { id: self.next_id, name: nu.name, email: nu.email };
self.next_id += 1;
self.users.push(u.clone());
// Steady-state RPS bench: cap so we don't grow unbounded.
if self.users.len() > 100_000 {
self.users.drain(..50_000);
}
(201, serde_json::to_vec(&u).unwrap())
}
Err(_) => (400, b"{\"error\":\"invalid body\"}".to_vec()),
}
}
StoreReq::Update { id, body } => {
match serde_json::from_slice::<NewUser>(&body) {
Ok(nu) => match self.users.iter_mut().find(|u| u.id == id) {
Some(u) => {
u.name = nu.name;
u.email = nu.email;
let snap = u.clone();
(200, serde_json::to_vec(&snap).unwrap())
}
None => (404, b"{\"error\":\"not found\"}".to_vec()),
},
Err(_) => (400, b"{\"error\":\"invalid body\"}".to_vec()),
}
}
StoreReq::Delete { id } => {
let before = self.users.len();
self.users.retain(|u| u.id != id);
if self.users.len() < before {
(204, Vec::new())
} else {
(404, b"{\"error\":\"not found\"}".to_vec())
}
}
}
}
fn handle_cast(&mut self, _req: ()) {}
}
pub fn spawn_memory_store() -> StoreHandle {
// gen_server::start (via ServerBuilder) must run inside an actor — the
// OnceLock lazy-init in AppState::store() guarantees that, same as before.
StoreHandle::Memory(ServerBuilder::new(MemStore::new()).start())
}
// ---------------------------------------------------------------------------
// SQLite store — one writer + N readers. (UNCHANGED — not ported.)
// ---------------------------------------------------------------------------
//
// Per the spec §5.1:
// - One writer actor owns a single rusqlite::Connection (read-write).
// - N=4 reader actors share a single `Receiver<ReadReq>`. The MPSC
// semantics guarantee each request goes to exactly one reader; the
// scheduler picks whichever reader is parked.
// - N=4 reader actors each own a private channel; a dispatcher round-robins
// reads across them. (gen_server is single-inbox, so porting this would
// serialise reads through one actor and kill S4 read parallelism — hence
// it stays on raw actors. It is not part of the spinning A/B.)
//
// All connections open the same database file. WAL is set on the writer's
// connection (the journal mode is a database-wide setting, so any
// connection setting it sticks). busy_timeout is set on every conn so a
// momentary writer hold doesn't EBUSY the readers.
//
// Schema is created idempotently by the writer on first run.
// connection (the journal mode is a database-wide setting). busy_timeout is
// set on every conn so a momentary writer hold doesn't EBUSY the readers.
const SQLITE_READER_COUNT: usize = 4;
@@ -258,14 +235,7 @@ pub fn spawn_sqlite_store(db_path: &str) -> StoreHandle {
// The readers. smarm's channel is MPSC, so we can't share a single
// Receiver across N reader actors. Instead: one dispatcher owns the
// public `reads_rx`, and round-robins each request into one of N
// per-reader private channels. Cost: one extra context switch on
// every read. That's the price of fan-out on an MPSC runtime.
//
// Alternative considered: have handlers know about N senders and
// pick one themselves. Rejected: it bakes the pool size into the
// handler protocol and gives noisy load balancing under bursty
// traffic (any per-handler hashing is worse than a single dispatcher
// with strict round-robin).
// per-reader private channels.
let (reads_tx, reads_rx) = channel::<ReadReq>();
let mut per_reader_txs: Vec<Sender<ReadReq>> = Vec::with_capacity(SQLITE_READER_COUNT);
for _ in 0..SQLITE_READER_COUNT {
@@ -276,7 +246,7 @@ pub fn spawn_sqlite_store(db_path: &str) -> StoreHandle {
}
smarm::spawn(move || dispatch_reads(reads_rx, per_reader_txs));
StoreHandle { reads: reads_tx, writes: writes_tx }
StoreHandle::Sqlite { reads: reads_tx, writes: writes_tx }
}
fn open_writer(path: &str) -> Connection {
@@ -303,9 +273,7 @@ fn open_reader(path: &str) -> Connection {
OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
).expect("sqlite: open reader");
conn.pragma_update(None, "busy_timeout", 5000_i64).expect("sqlite: busy_timeout");
// We deliberately don't set journal_mode here — readers inherit the
// DB-wide WAL setting from the writer. Setting it on a read-only
// connection would just be a no-op or an error.
// Readers inherit the DB-wide WAL setting from the writer.
conn
}
@@ -316,8 +284,6 @@ fn sqlite_writer_loop(path: &str, rx: Receiver<WriteReq>) {
WriteReq::Create { body, reply } => {
match serde_json::from_slice::<NewUser>(&body) {
Ok(nu) => {
// BEGIN IMMEDIATE so the write lock is taken at
// statement start, not lazily.
let res = conn.execute(
"INSERT INTO users (name, email) VALUES (?1, ?2)",
params![nu.name, nu.email],
@@ -376,23 +342,17 @@ fn sqlite_writer_loop(path: &str, rx: Receiver<WriteReq>) {
}
fn dispatch_reads(rx: Receiver<ReadReq>, workers: Vec<Sender<ReadReq>>) {
// Strict round-robin. A free-list / readiness queue would distribute
// better under skew, but it requires bidirectional signalling we
// don't have a primitive for. Round-robin is fine here: SQLite reads
// are uniform in cost (single-row PK lookup or LIMIT 100), and any
// brief reader stall just shifts a request to the next reader on
// the next pass.
// Strict round-robin. SQLite reads are uniform in cost (single-row PK
// lookup or LIMIT 100), so a brief reader stall just shifts a request to
// the next reader on the next pass.
let n = workers.len();
let mut idx = 0usize;
while let Ok(req) = rx.recv() {
let target = &workers[idx % n];
idx = idx.wrapping_add(1);
if target.send(req).is_err() {
// Reader died; skip it. (smarm restarts panicking actors
// under a supervisor, but in this binary the readers are
// unsupervised — a panic on one of them would mean a dropped
// request now and that worker is permanently silent. Fine
// for the bench; for production it'd want a supervisor.)
// Reader died; skip it. Unsupervised in this binary — fine for the
// bench.
}
}
}