diff --git a/analyze_spin_ab_urus.py b/analyze_spin_ab_urus.py new file mode 100755 index 0000000..f01dfd5 --- /dev/null +++ b/analyze_spin_ab_urus.py @@ -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: + + /before/--c/result.json + /after/--c/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()) diff --git a/spin_ab_urus.sh b/spin_ab_urus.sh new file mode 100755 index 0000000..4a8515c --- /dev/null +++ b/spin_ab_urus.sh @@ -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_) + +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" diff --git a/urus-server/src/handlers.rs b/urus-server/src/handlers.rs index 4d454b1..fdf4801 100644 --- a/urus-server/src/handlers.rs +++ b/urus-server/src/handlers.rs @@ -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) -> Conn { @@ -81,13 +80,17 @@ fn text(conn: Conn, status: u16, body: &'static str) -> Conn { fn parse_id(s: &str) -> Option { 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)>(); - 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)>(); - 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)>(); - 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)>(); - 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)>(); - 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()), } } } diff --git a/urus-server/src/store.rs b/urus-server/src/store.rs index 7c18d51..e780df4 100644 --- a/urus-server/src/store.rs +++ b/urus-server/src/store.rs @@ -1,43 +1,58 @@ //! Store actors for the urus-server bench. //! -//! Two implementations, same protocol: -//! - Memory store: one actor owns a Vec. -//! - 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`. 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); -// Debug is intentionally not derived: `smarm::Sender` 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 }, + Update { id: u64, body: Vec }, + 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 }, - Get { id: u64, reply: Sender }, + List { reply: Sender }, + Get { id: u64, reply: Sender }, } pub enum WriteReq { - Create { body: Vec, reply: Sender }, - Update { id: u64, body: Vec, reply: Sender }, - Delete { id: u64, reply: Sender }, + Create { body: Vec, reply: Sender }, + Update { id: u64, body: Vec, reply: Sender }, + Delete { id: u64, reply: Sender }, } // --------------------------------------------------------------------------- @@ -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, - writes: Sender, +pub enum StoreHandle { + Memory(ServerRef), + Sqlite { reads: Sender, writes: Sender }, } impl StoreHandle { - pub fn reads(&self) -> &Sender { &self.reads } - pub fn writes(&self) -> &Sender { &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::(); - - // 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::(); - { - 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::(); - { - 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, tx: Sender) { - while let Ok(r) = rx.recv() { - if tx.send(AnyReq::Read(r)).is_err() { return; } - } -} - -fn forward_writes(rx: Receiver, tx: Sender) { - while let Ok(w) = rx.recv() { - if tx.send(AnyReq::Write(w)).is_err() { return; } - } -} - -fn memory_store_loop(rx: Receiver) { - let mut users: Vec = 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::(&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. - 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::(&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 { + match self { + StoreHandle::Memory(r) => r.call(req).map_err(|_| ()), + StoreHandle::Sqlite { reads, writes } => { + let (tx, rx) = channel::(); + // Map each backend's distinct SendError 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, + 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::(&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::(&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`. 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::(); let mut per_reader_txs: Vec> = 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::Create { body, reply } => { match serde_json::from_slice::(&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) { } fn dispatch_reads(rx: Receiver, workers: Vec>) { - // 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. } } }