From 768317d789a01d722fe7455d792fed11e6579b5c Mon Sep 17 00:00:00 2001 From: eunhhu Date: Sat, 26 Sep 2026 16:56:08 +0100 Subject: Benchmark an app served by its views and inputs run.py --suite app serves the scoreboard of scores.aup (22 lines of aup) with pure-serve against the same scoreboard written by hand in axum and drogon: POST /inputs/score?points=N with a name as the body, GET /views/total and a page of 20 rows as JSON. Durable, the baselines append each score to a log and fdatasync it before answering (replayed at start); the load generator learns the three requests. --- benchmarks/http/axum-baseline/src/main.rs | 129 ++++++++++++++++++++++++++++-- 1 file changed, 124 insertions(+), 5 deletions(-) (limited to 'benchmarks/http/axum-baseline') diff --git a/benchmarks/http/axum-baseline/src/main.rs b/benchmarks/http/axum-baseline/src/main.rs index 6684bfc..4ffee8c 100644 --- a/benchmarks/http/axum-baseline/src/main.rs +++ b/benchmarks/http/axum-baseline/src/main.rs @@ -3,6 +3,15 @@ //! a PUT body of u32le start, u32le remove, then the bytes to insert; the //! reply is the inserted bytes; ETag is the revision). //! +//! The scoreboard app of the app benchmarks, as Pure serves an aup app's views +//! and inputs: POST /inputs/score?points=N with a name as the body adds N to +//! that name's points and replies the name; GET /views/total is the sum of +//! the points; GET /views/leaders?from=F&count=C is a page of the board in name +//! order as JSON ({"rows":N,"from":F,"items":[{"name":..,"points":..}]}). +//! Durable (either flag), each score is appended to DIR/scores.log and +//! fdatasynced before the answer, and the log is replayed at start: an +//! append-only log is crash safe, a torn last record was never acknowledged. +//! //! axum-baseline --port P [--durable DIR | --atomic DIR] //! Without either, documents live in memory. --durable writes the document to //! DIR/key and fdatasyncs before answering, as a straightforward durable @@ -11,13 +20,17 @@ //! the directory: the usual crash-safe replacement. use axum::{ body::Bytes, - extract::{Path, State}, - http::{HeaderMap, HeaderValue, StatusCode}, + extract::{Path, RawQuery, State}, + http::{header, HeaderMap, HeaderValue, StatusCode}, response::{IntoResponse, Response}, - routing::get, + routing::{get, post}, Router, }; -use std::{collections::HashMap, path::PathBuf, sync::Arc}; +use std::{ + collections::{BTreeMap, HashMap}, + path::PathBuf, + sync::Arc, +}; use tokio::sync::Mutex; #[derive(Default)] @@ -30,6 +43,13 @@ struct Service { documents: Mutex>>>, durable: Option, atomic: bool, + board: Mutex, +} + +/// The scoreboard: points by name, in name (byte) order, and the log. +struct Board { + points: BTreeMap, + log: Option>, } fn valid_key(key: &str) -> bool { @@ -120,6 +140,101 @@ async fn edit( tagged(StatusCode::OK, document.revision, "none", insert) } +/// The value of a name=value pair of a query, parsed. +fn query_value(query: &Option, name: &str) -> Option { + query.as_deref()?.split('&').find_map(|pair| pair.strip_prefix(name)?.strip_prefix('=')?.parse().ok()) +} + +fn text(status: StatusCode, error: &'static str, body: String) -> Response { + let mut response = (status, body).into_response(); + response.headers_mut().insert("X-Pure-Error", HeaderValue::from_static(error)); + response +} + +async fn score(State(service): State>, RawQuery(query): RawQuery, body: Bytes) -> Response { + let (Some(points), Ok(name)) = (query_value::(&query, "points"), std::str::from_utf8(&body)) else { + return tagged(StatusCode::BAD_REQUEST, 0, "request", Bytes::new()); + }; + let mut board = service.board.lock().await; + if let Some(log) = board.log.clone() { + let mut record = Vec::with_capacity(8 + name.len()); + record.extend_from_slice(&(name.len() as u32).to_le_bytes()); + record.extend_from_slice(name.as_bytes()); + record.extend_from_slice(&points.to_le_bytes()); + let written = tokio::task::spawn_blocking(move || -> std::io::Result<()> { + use std::io::Write; + (&*log).write_all(&record)?; + log.sync_data() + }) + .await; + if !matches!(written, Ok(Ok(()))) { + return tagged(StatusCode::INTERNAL_SERVER_ERROR, 0, "storage", Bytes::new()); + } + } + *board.points.entry(name.to_string()).or_default() += points as u64; + tagged(StatusCode::OK, 0, "none", body) +} + +async fn total(State(service): State>) -> Response { + let board = service.board.lock().await; + text(StatusCode::OK, "none", board.points.values().sum::().to_string()) +} + +fn json_text(out: &mut String, text: &str) { + out.push('"'); + for c in text.chars() { + match c { + '"' => out.push_str("\\\""), + '\\' => out.push_str("\\\\"), + c if (c as u32) < 0x20 => out.push_str(&format!("\\u{:04x}", c as u32)), + c => out.push(c), + } + } + out.push('"'); +} + +async fn leaders(State(service): State>, RawQuery(query): RawQuery) -> Response { + let from = query_value::(&query, "from").unwrap_or(0); + let count = query_value::(&query, "count").unwrap_or(usize::MAX); + let board = service.board.lock().await; + let rows = board.points.len(); + let from = from.min(rows); + let mut out = format!("{{\"rows\":{rows},\"from\":{from},\"items\":["); + for (i, (name, points)) in board.points.iter().skip(from).take(count).enumerate() { + out.push_str(if i > 0 { ",{\"name\":" } else { "{\"name\":" }); + json_text(&mut out, name); + out.push_str(&format!(",\"points\":{points}}}")); + } + out.push_str("]}"); + let mut response = text(StatusCode::OK, "none", out); + response.headers_mut().insert(header::CONTENT_TYPE, HeaderValue::from_static("application/json")); + response +} + +/// Replay the scoreboard's log up to its last whole record, and keep it open for appending. +fn open_board(directory: &Option) -> Board { + let mut points = BTreeMap::new(); + let Some(directory) = directory else { return Board { points, log: None } }; + let path = directory.join("scores.log"); + let bytes = std::fs::read(&path).unwrap_or_default(); + let mut at = 0; + while at + 4 <= bytes.len() { + let length = u32::from_le_bytes(bytes[at..at + 4].try_into().unwrap()) as usize; + if at + 8 + length > bytes.len() { + break; + } + let name = String::from_utf8_lossy(&bytes[at + 4..at + 4 + length]).into_owned(); + let added = u32::from_le_bytes(bytes[at + 4 + length..at + 8 + length].try_into().unwrap()); + *points.entry(name).or_insert(0u64) += added as u64; + at += 8 + length; + } + let log = std::fs::OpenOptions::new().create(true).write(true).open(&path).expect("scores.log"); + log.set_len(at as u64).expect("trim the log"); + use std::io::Seek; + (&log).seek(std::io::SeekFrom::End(0)).expect("seek"); + Board { points, log: Some(Arc::new(log)) } +} + #[tokio::main] async fn main() { let arguments: Vec = std::env::args().collect(); @@ -142,9 +257,13 @@ async fn main() { if let Some(directory) = &durable { std::fs::create_dir_all(directory).expect("durable directory"); } - let service = Arc::new(Service { documents: Mutex::new(HashMap::new()), durable, atomic }); + let board = Mutex::new(open_board(&durable)); + let service = Arc::new(Service { documents: Mutex::new(HashMap::new()), durable, atomic, board }); let app = Router::new() .route("/objects/{key}", get(show).put(edit)) + .route("/inputs/score", post(score)) + .route("/views/total", get(total)) + .route("/views/leaders", get(leaders)) .with_state(service); let listener = tokio::net::TcpListener::bind(("127.0.0.1", port)).await.expect("bind"); println!("{{\"port\":{}}}", listener.local_addr().unwrap().port()); -- cgit v1.2.3