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

baseline

Browse Source
This commit is contained in:
Markk116
2026-05-26 23:22:32 +02:00
parent 842fca2b72
commit 730de334e0
17 changed files with 5380 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
urus-server/Cargo.toml
+35
View File
@@ -0,0 +1,35 @@
[package]
name = "urus-server"
version = "0.1.0"
edition = "2021"
rust-version = "1.95"
[dependencies]
# Path deps point at the user's local checkouts; adjust if your tree
# differs. urus's own Cargo.toml has `smarm = { path = "../smarm" }`, so
# cargo resolves smarm transitively as long as it's a sibling of urus.
urus = { path = "../../urus" }
smarm = { path = "../../smarm" }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
# Constant-time compare for the bearer-token check. Tiny dep, no transitive
# bloat, and the right thing to use.
subtle = "2"
# In-memory ring buffer for the logger plug. Bounded MPSC over a fixed
# array — no allocation on the hot path.
crossbeam-queue = "0.3"
# Embedded SQLite. The `bundled` feature compiles SQLite from source so we
# don't depend on the system library — important for reproducibility on
# build hosts that may or may not have libsqlite3-dev.
rusqlite = { version = "0.32", features = ["bundled"] }
# Tiny argv parser so we don't pull in clap for a 3-flag CLI.
# We hand-roll it instead. No dep here.
[[bin]]
name = "urus-server"
path = "src/main.rs"
urus-server/src/handlers.rs
+192
View File
@@ -0,0 +1,192 @@
//! 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.
//!
//! AppState holds:
//! - a "store_init" factory that lazily spawns the store on first use
//! and caches the resulting StoreHandle
//! - the shared log ring (not yet used here; handlers don't log
//! directly, the logger plug does)
//!
//! Concurrency: the first connection actor to ask wins the init race;
//! the rest see the already-initialised handle. We use OnceLock for this
//! — std primitive, no extra dep.
use std::sync::{Arc, OnceLock};
use smarm::channel;
use urus::{Conn, Next, Plug, Router};
use crate::middleware::LogRing;
use crate::store::{ReadReq, StoreHandle, WriteReq};
// ---------------------------------------------------------------------------
// AppState
// ---------------------------------------------------------------------------
#[derive(Clone)]
pub struct AppState {
inner: Arc<AppStateInner>,
}
struct AppStateInner {
store_handle: OnceLock<StoreHandle>,
store_init: Arc<dyn Fn() -> StoreHandle + Send + Sync + 'static>,
// Kept for parity with the spec even though handlers don't read it.
#[allow(dead_code)]
log_ring: LogRing,
}
impl AppState {
pub fn new(
store_init: Arc<dyn Fn() -> StoreHandle + Send + Sync + 'static>,
log_ring: LogRing,
) -> Self {
Self {
inner: Arc::new(AppStateInner {
store_handle: OnceLock::new(),
store_init,
log_ring,
}),
}
}
/// Lazy store accessor. Safe to call from inside any connection
/// actor; safe to call concurrently — `OnceLock::get_or_init`
/// guarantees the init closure runs exactly once.
pub fn store(&self) -> &StoreHandle {
self.inner.store_handle.get_or_init(|| (self.inner.store_init)())
}
}
// ---------------------------------------------------------------------------
// Tiny helper for writing JSON responses
// ---------------------------------------------------------------------------
fn json(conn: Conn, status: u16, body: Vec<u8>) -> Conn {
conn.put_status(status)
.put_header("content-type", "application/json")
.put_body(body)
}
fn text(conn: Conn, status: u16, body: &'static str) -> Conn {
conn.put_status(status)
.put_header("content-type", "text/plain; charset=utf-8")
.put_body(body)
}
fn parse_id(s: &str) -> Option<u64> { s.parse().ok() }
// ---------------------------------------------------------------------------
// 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 build_ping_handler() -> impl Plug {
// /ping is a literal endpoint: status 200, body "pong". No work.
move |conn: Conn, _next: Next| text(conn, 200, "pong")
}
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() {
Ok((s, b)) => json(conn, s, b),
Err(_) => json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec()),
}
}
}
fn build_get_one_handler(state: AppState) -> impl Plug {
move |conn: Conn, _next: Next| {
let id = match conn.params.get("id").and_then(parse_id) {
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() {
Ok((s, b)) => json(conn, s, b),
Err(_) => json(conn, 503, b"{\"error\":\"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.)
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() {
Ok((s, b)) => json(conn, s, b),
Err(_) => json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec()),
}
}
}
fn build_update_handler(state: AppState) -> impl Plug {
move |conn: Conn, _next: Next| {
let id = match conn.params.get("id").and_then(parse_id) {
Some(id) => id,
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() {
Ok((s, b)) => json(conn, s, b),
Err(_) => json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec()),
}
}
}
fn build_delete_handler(state: AppState) -> impl Plug {
move |conn: Conn, _next: Next| {
let id = match conn.params.get("id").and_then(parse_id) {
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() {
Ok((s, b)) => json(conn, s, b),
Err(_) => json(conn, 503, b"{\"error\":\"store unavailable\"}".to_vec()),
}
}
}
// ---------------------------------------------------------------------------
// Router assembly
// ---------------------------------------------------------------------------
pub fn build_router(state: AppState) -> Router {
Router::new()
.get( "/ping", build_ping_handler())
.get( "/api/v1/users", build_list_handler(state.clone()))
.post( "/api/v1/users", build_create_handler(state.clone()))
.get( "/api/v1/users/:id", build_get_one_handler(state.clone()))
.put( "/api/v1/users/:id", build_update_handler(state.clone()))
.delete("/api/v1/users/:id", build_delete_handler(state))
}
urus-server/src/main.rs
+194
View File
@@ -0,0 +1,194 @@
//! urus-server — the urus side of the urus / axum / cowboy benchmark.
//!
//! Exposes the routes from `urus-bench-spec.md` §3 with the middleware
//! stack from §3.1–§3.4. Backing store is switchable at startup:
//!
//! --store=memory in-memory actor (S1, S2, S3)
//! --store=sqlite single-writer + reader-pool over rusqlite (S4)
//!
//! Other flags:
//! --addr=HOST:PORT default 127.0.0.1:8080
//! --token=TOKEN default "test-token-aaaaaaaaaaaaaaaaaaaa"
//! --db-path=PATH SQLite file path (sqlite store only)
//! --no-auth skip the auth plug (used for S1 /ping check)
//!
//! The `/ping` route is mounted *outside* the auth plug so the S1 scenario
//! does not pay an auth cost it doesn't need. S2/S3/S4 routes are inside it.
mod handlers;
mod middleware;
mod store;
use std::process::ExitCode;
use std::sync::Arc;
use urus::{serve_with, Config, Conn, Next, Pipeline};
use handlers::AppState;
use middleware::{auth_plug, logger_plug, request_id_plug, LogRing};
use store::{spawn_memory_store, spawn_sqlite_store, StoreHandle};
// ---------------------------------------------------------------------------
// CLI
// ---------------------------------------------------------------------------
#[derive(Clone, Debug)]
struct Cli {
addr: String,
store: StoreKind,
token: String,
db_path: String,
no_auth: bool,
}
#[derive(Clone, Copy, Debug)]
enum StoreKind { Memory, Sqlite }
impl Cli {
fn parse() -> Result<Self, String> {
// Hand-rolled because clap would dominate compile time for three
// flags. Each arg is `--name=value` or `--name` (no positionals).
let mut cli = Cli {
addr: "127.0.0.1:8080".into(),
store: StoreKind::Memory,
token: "test-token-aaaaaaaaaaaaaaaaaaaa".into(),
db_path: "/tmp/urus-bench.sqlite".into(),
no_auth: false,
};
for arg in std::env::args().skip(1) {
let (k, v) = match arg.split_once('=') {
Some((k, v)) => (k, Some(v.to_string())),
None => (arg.as_str(), None),
};
match (k, v) {
("--addr", Some(v)) => cli.addr = v,
("--token", Some(v)) => cli.token = v,
("--db-path", Some(v)) => cli.db_path = v,
("--store", Some(v)) => cli.store = match v.as_str() {
"memory" => StoreKind::Memory,
"sqlite" => StoreKind::Sqlite,
other => return Err(format!("unknown store: {other}")),
},
("--no-auth", None) => cli.no_auth = true,
("--help" | "-h", _) => {
print_usage();
std::process::exit(0);
}
(k, _) => return Err(format!("unknown flag: {k}")),
}
}
Ok(cli)
}
}
fn print_usage() {
eprintln!(
"urus-server [--addr=HOST:PORT] [--store=memory|sqlite] \
[--token=TOKEN] [--db-path=PATH] [--no-auth]"
);
}
// ---------------------------------------------------------------------------
// main
// ---------------------------------------------------------------------------
fn main() -> ExitCode {
let cli = match Cli::parse() {
Ok(c) => c,
Err(e) => { eprintln!("urus-server: {e}"); print_usage(); return ExitCode::from(2); }
};
let addr = match cli.addr.parse() {
Ok(a) => a,
Err(e) => { eprintln!("urus-server: bad --addr: {e}"); return ExitCode::from(2); }
};
// The log ring is a bounded MPSC-like buffer with a fixed capacity. It
// is shared by every connection actor (via Arc) and drained on Ctrl-C
// — but we don't wire shutdown in v1, so it's just an in-memory sink.
// Bounded means the bench can run for an hour without RSS growth.
let log_ring = LogRing::with_capacity(64 * 1024);
// Token is wrapped in Arc<[u8]> so each plug clone is a refcount bump.
// The constant-time compare reads it as bytes.
let token: Arc<[u8]> = cli.token.as_bytes().to_vec().into();
// The pipeline is shape:
// logger -> request_id -> [auth ->] router
//
// /ping is mounted as a literal at the top of the router so it shares
// the middleware chain with everything else; the bench spec says S1 is
// a "naked" hello world, but in this implementation /ping still goes
// through the logger and request_id plugs. That keeps the impl simple
// and lets us measure the middleware cost as the *S2 minus S1* delta,
// which is what the spec actually asks for (§2).
//
// --no-auth turns off auth for cases where the load generator can't
// easily set headers (it shouldn't be needed for wrk2 though).
let auth_token = token.clone();
let auth = move |conn: Conn, next: Next| auth_plug(conn, next, &auth_token);
// The store handle is what handlers use to talk to the store actor.
// It's a wrapper around a smarm Sender; cloning is cheap (refcount).
//
// We can't actually *spawn* the store actor here (smarm requires that
// spawn be called from inside an actor, not from the bootstrap
// thread). The OnceLock in handlers::AppState defers that until the
// first request comes in. We just pass the construction parameters
// through here.
//
// For SQLite we use the configured DB path; for memory we don't
// need anything.
let store_init: Arc<dyn Fn() -> StoreHandle + Send + Sync + 'static> =
match cli.store {
StoreKind::Memory => Arc::new(|| spawn_memory_store()),
StoreKind::Sqlite => {
let path = cli.db_path.clone();
Arc::new(move || spawn_sqlite_store(&path))
}
};
let state = AppState::new(store_init, log_ring.clone());
// Build the router. Handlers close over `state` (cheap Arc clone).
// Note that handlers themselves use the AppState to resolve the
// store on first call.
let router = handlers::build_router(state.clone());
// Assemble the pipeline.
//
// The closures below take `Conn, Next` with explicit type annotations.
// This is necessary because `Next<'a>` is generic over a lifetime;
// without the annotation rustc binds the closure to *one specific*
// lifetime and then it doesn't satisfy `Plug`'s `Fn(Conn, Next) -> Conn`
// higher-ranked bound. With `Next` written out, the closure parses as
// `for<'a> Fn(Conn, Next<'a>) -> Conn`, which is what Plug wants.
let log_plug = {
let ring = log_ring.clone();
move |conn: Conn, next: Next| logger_plug(conn, next, &ring)
};
let mut pipeline = Pipeline::new()
.plug(log_plug)
.plug(request_id_plug);
if !cli.no_auth {
pipeline = pipeline.plug(auth);
}
let pipeline = pipeline.plug(router);
// Serve. We use the default Config and let urus's heuristic pick
// listener pool size = num CPUs.
let cfg = Config::new(addr);
eprintln!(
"urus-server: store={:?} addr={} auth={}",
cli.store, cli.addr, !cli.no_auth
);
if let Err(e) = serve_with(cfg, pipeline) {
eprintln!("urus-server: serve failed: {e}");
return ExitCode::from(1);
}
ExitCode::SUCCESS
}
urus-server/src/middleware.rs
+234
View File
@@ -0,0 +1,234 @@
//! Middleware plugs for the bench.
//!
//! Per the spec §3:
//! - Logger writes to a bounded ring (no stdout — that would dominate).
//! - Request ID is generated from a thread-local fast RNG (SplitMix64).
//! - Auth is constant-time bearer-token compare.
//!
//! Each plug is a `Fn(Conn, Next) -> Conn` closure; urus's blanket impl
//! turns them into Plug values when handed to Pipeline::plug.
use std::cell::Cell;
use std::sync::Arc;
use std::time::Instant;
use crossbeam_queue::ArrayQueue;
use subtle::ConstantTimeEq;
use urus::{Conn, Next};
// ---------------------------------------------------------------------------
// Log ring
// ---------------------------------------------------------------------------
//
// Bounded MPSC over a fixed-capacity array. We push from many connection
// actors and never read during the bench (drain on shutdown only). If the
// ring fills, new entries are dropped — the bench is the priority, not
// the log. This matches what we said in spec §1.5: warm-up + 60s measured
// at high RPS would overflow any unbounded buffer.
#[derive(Debug, Clone)]
pub struct LogEntry {
pub method: String,
pub path: String,
pub status: u16,
pub elapsed_us: u64,
}
#[derive(Clone)]
pub struct LogRing {
inner: Arc<ArrayQueue<LogEntry>>,
}
impl LogRing {
pub fn with_capacity(n: usize) -> Self {
Self { inner: Arc::new(ArrayQueue::new(n)) }
}
/// Push an entry. Drops the entry on overflow (returns Err in
/// crossbeam's API — we discard).
pub fn push(&self, e: LogEntry) {
let _ = self.inner.push(e);
}
/// For inspection after a run.
#[allow(dead_code)]
pub fn drain(&self) -> Vec<LogEntry> {
let mut out = Vec::with_capacity(self.inner.len());
while let Some(e) = self.inner.pop() { out.push(e); }
out
}
}
// ---------------------------------------------------------------------------
// Logger plug
// ---------------------------------------------------------------------------
//
// Wraps `next`. Captures method/path *before* the inner plugs run (the
// router may move them), then captures status + elapsed after.
pub fn logger_plug(conn: Conn, next: Next, ring: &LogRing) -> Conn {
let start = Instant::now();
let method = conn.method.as_str().to_string();
let path = conn.path.clone();
let conn = next.run(conn);
let elapsed_us = start.elapsed().as_micros() as u64;
ring.push(LogEntry {
method,
path,
status: conn.status.unwrap_or(0),
elapsed_us,
});
conn
}
// ---------------------------------------------------------------------------
// Request ID — SplitMix64 in a thread-local, base32-encoded.
// ---------------------------------------------------------------------------
//
// We avoid global RNG / OS getrandom on the hot path. Each scheduler OS
// thread gets its own seed (Cell + thread_local) and advances it with the
// SplitMix64 step. 12 bytes of output is encoded with the RFC 4648 base32
// alphabet (no padding, no lowercase — the header is opaque to the load
// generator).
//
// Crockford-style alphabet would be friendlier to humans but base32-hex
// is fine here.
thread_local! {
static RNG_STATE: Cell<u64> = const { Cell::new(0) };
}
fn splitmix64(s: u64) -> (u64, u64) {
// Standard SplitMix64. One 64-bit step gives a uniformly distributed
// output and the next state.
let new_state = s.wrapping_add(0x9E3779B97F4A7C15);
let mut z = new_state;
z = (z ^ (z >> 30)).wrapping_mul(0xBF58476D1CE4E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D049BB133111EB);
(new_state, z ^ (z >> 31))
}
fn next_random_bytes(out: &mut [u8; 12]) {
RNG_STATE.with(|cell| {
let mut s = cell.get();
if s == 0 {
// Lazy first-time seed. Mixing pointer + thread id + clock
// is sufficient for a bench-local RNG; this isn't crypto.
let p = cell as *const _ as usize as u64;
let t = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0);
s = p ^ t ^ 0xA5A5_A5A5_A5A5_A5A5;
if s == 0 { s = 1; }
}
let (s1, r1) = splitmix64(s);
let (s2, r2) = splitmix64(s1);
cell.set(s2);
out[0..8].copy_from_slice(&r1.to_le_bytes());
out[8..12].copy_from_slice(&r2.to_le_bytes()[..4]);
});
}
// Base32 (RFC 4648) without padding. 12 bytes = 96 bits -> ceil(96/5) = 20
// output chars exactly. No padding needed.
fn b32_encode(input: &[u8; 12]) -> String {
static ALPHABET: &[u8; 32] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ234567";
let mut out = String::with_capacity(20);
let mut buf: u64 = 0;
let mut bits: u32 = 0;
for &b in input {
buf = (buf << 8) | (b as u64);
bits += 8;
while bits >= 5 {
bits -= 5;
let idx = ((buf >> bits) & 0x1F) as usize;
out.push(ALPHABET[idx] as char);
}
}
if bits > 0 {
let idx = ((buf << (5 - bits)) & 0x1F) as usize;
out.push(ALPHABET[idx] as char);
}
out
}
pub fn request_id_plug(conn: Conn, next: Next) -> Conn {
let mut raw = [0u8; 12];
next_random_bytes(&mut raw);
let id = b32_encode(&raw);
// Set on the response. We could also stash it in `conn.assigns` for
// handlers to read, but the bench doesn't need that.
let conn = next.run(conn);
conn.put_header("x-request-id", id)
}
// ---------------------------------------------------------------------------
// Auth plug
// ---------------------------------------------------------------------------
//
// Read `Authorization: Bearer <token>`; constant-time compare against the
// configured token. Mismatch -> 401 + halt.
//
// The token is borrowed (&[u8]) so the closure that wraps this can hold
// it via Arc<[u8]> and pass a slice in. No alloc per request.
pub fn auth_plug(conn: Conn, next: Next, expected: &[u8]) -> Conn {
// /ping is the S1 "naked" endpoint per the bench spec — it must not
// pay the auth cost. We exempt it by path. This is a deliberate
// exception, not a generic feature; the rest of the pipeline (logger,
// request_id) still applies to /ping, which is what the spec asks for
// (S1 measures the framework floor, not "no middleware whatsoever").
if conn.path == "/ping" {
return next.run(conn);
}
let header = conn.headers.get("authorization");
let presented: Option<&[u8]> = header.and_then(|h| {
// strip the "Bearer " prefix (case-insensitive per RFC 6750, but
// wrk2 sends the canonical form so an exact prefix check is fine)
h.strip_prefix("Bearer ").map(|s| s.as_bytes())
});
let ok = match presented {
Some(p) if p.len() == expected.len() =>
bool::from(p.ct_eq(expected)),
_ => false,
};
if ok {
next.run(conn)
} else {
conn.put_status(401)
.put_header("content-type", "application/json")
.put_body(&b"{\"error\":\"unauthorized\"}"[..])
.halt()
}
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn b32_round_trip_lengths() {
let mut raw = [0u8; 12];
for i in 0..12u8 { raw[i as usize] = i; }
let s = b32_encode(&raw);
assert_eq!(s.len(), 20);
assert!(s.chars().all(|c| c.is_ascii_uppercase() || c.is_ascii_digit()));
}
#[test]
fn rng_advances() {
let mut a = [0u8; 12];
let mut b = [0u8; 12];
next_random_bytes(&mut a);
next_random_bytes(&mut b);
assert_ne!(a, b, "consecutive calls must produce different bytes");
}
}
urus-server/src/store.rs
+444
View File
@@ -0,0 +1,444 @@
//! 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.
//!
//! `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.
use smarm::{channel, Receiver, Sender};
use std::sync::Arc;
use std::path::Path;
use rusqlite::{params, Connection, OpenFlags};
// ---------------------------------------------------------------------------
// Wire types — what the store returns over the reply channel.
// ---------------------------------------------------------------------------
//
// We send (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.
pub enum ReadReq {
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> },
}
// ---------------------------------------------------------------------------
// User model
// ---------------------------------------------------------------------------
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
struct User {
id: u64,
name: String,
email: String,
}
#[derive(serde::Deserialize)]
struct NewUser {
name: String,
email: String,
}
// ---------------------------------------------------------------------------
// 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.
#[derive(Clone)]
pub struct StoreHandle {
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()));
}
}
}
}
}
// ---------------------------------------------------------------------------
// SQLite store — one writer + N readers.
// ---------------------------------------------------------------------------
//
// 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.
//
// 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.
const SQLITE_READER_COUNT: usize = 4;
pub fn spawn_sqlite_store(db_path: &str) -> StoreHandle {
let path: Arc<str> = Arc::from(db_path);
// The writer actor.
let (writes_tx, writes_rx) = channel::<WriteReq>();
{
let path = path.clone();
smarm::spawn(move || sqlite_writer_loop(&path, writes_rx));
}
// 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).
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 {
let (tx, rx) = channel::<ReadReq>();
per_reader_txs.push(tx);
let path = path.clone();
smarm::spawn(move || sqlite_reader_loop(&path, rx));
}
smarm::spawn(move || dispatch_reads(reads_rx, per_reader_txs));
StoreHandle { reads: reads_tx, writes: writes_tx }
}
fn open_writer(path: &str) -> Connection {
let conn = Connection::open(path).expect("sqlite: open writer");
// WAL is the whole reason this design works: writers don't block
// readers, readers don't block writers, only writers block other
// writers (and there's only one).
conn.pragma_update(None, "journal_mode", "WAL").expect("sqlite: WAL");
conn.pragma_update(None, "synchronous", "NORMAL").expect("sqlite: synchronous");
conn.pragma_update(None, "busy_timeout", 5000_i64).expect("sqlite: busy_timeout");
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT NOT NULL
);"
).expect("sqlite: schema");
conn
}
fn open_reader(path: &str) -> Connection {
let conn = Connection::open_with_flags(
Path::new(path),
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.
conn
}
fn sqlite_writer_loop(path: &str, rx: Receiver<WriteReq>) {
let conn = open_writer(path);
while let Ok(req) = rx.recv() {
match req {
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],
);
match res {
Ok(_) => {
let id = conn.last_insert_rowid() as u64;
let u = User { id, name: nu.name, email: nu.email };
let _ = reply.send((201, serde_json::to_vec(&u).unwrap()));
}
Err(e) => {
eprintln!("sqlite: insert failed: {e}");
let _ = reply.send((500, b"{\"error\":\"db\"}".to_vec()));
}
}
}
Err(_) => {
let _ = reply.send((400, b"{\"error\":\"invalid body\"}".to_vec()));
}
}
}
WriteReq::Update { id, body, reply } => {
match serde_json::from_slice::<NewUser>(&body) {
Ok(nu) => {
let res = conn.execute(
"UPDATE users SET name=?1, email=?2 WHERE id=?3",
params![nu.name, nu.email, id as i64],
);
match res {
Ok(0) => { let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec())); }
Ok(_) => {
let u = User { id, name: nu.name, email: nu.email };
let _ = reply.send((200, serde_json::to_vec(&u).unwrap()));
}
Err(_) => { let _ = reply.send((500, b"{\"error\":\"db\"}".to_vec())); }
}
}
Err(_) => {
let _ = reply.send((400, b"{\"error\":\"invalid body\"}".to_vec()));
}
}
}
WriteReq::Delete { id, reply } => {
let res = conn.execute(
"DELETE FROM users WHERE id=?1",
params![id as i64],
);
match res {
Ok(0) => { let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec())); }
Ok(_) => { let _ = reply.send((204, Vec::new())); }
Err(_) => { let _ = reply.send((500, b"{\"error\":\"db\"}".to_vec())); }
}
}
}
}
}
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.
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.)
}
}
}
fn sqlite_reader_loop(path: &str, rx: Receiver<ReadReq>) {
let conn = open_reader(path);
let mut stmt_get = conn.prepare("SELECT id, name, email FROM users WHERE id=?1")
.expect("sqlite: prepare get");
let mut stmt_list = conn.prepare("SELECT id, name, email FROM users LIMIT 100")
.expect("sqlite: prepare list");
while let Ok(req) = rx.recv() {
match req {
ReadReq::Get { id, reply } => {
let row = stmt_get.query_row(params![id as i64], |r| {
Ok(User {
id: r.get::<_, i64>(0)? as u64,
name: r.get(1)?,
email: r.get(2)?,
})
});
match row {
Ok(u) => { let _ = reply.send((200, serde_json::to_vec(&u).unwrap())); }
Err(rusqlite::Error::QueryReturnedNoRows) =>
{ let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec())); }
Err(_) =>
{ let _ = reply.send((500, b"{\"error\":\"db\"}".to_vec())); }
}
}
ReadReq::List { reply } => {
let rows = stmt_list.query_map([], |r| {
Ok(User {
id: r.get::<_, i64>(0)? as u64,
name: r.get(1)?,
email: r.get(2)?,
})
});
match rows {
Ok(iter) => {
let users: Vec<User> = iter.filter_map(|r| r.ok()).collect();
let body = serde_json::to_vec(&users).unwrap_or_else(|_| b"[]".to_vec());
let _ = reply.send((200, body));
}
Err(_) => { let _ = reply.send((500, b"{\"error\":\"db\"}".to_vec())); }
}
}
}
}
}