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 ++++++++++++++++++++++++++++-- benchmarks/http/drogon-baseline/main.cc | 112 ++++++++++++++++++++++++++ benchmarks/http/load.c | 26 +++++- benchmarks/http/run.py | 32 +++++++- benchmarks/http/scores.aup | 22 +++++ 5 files changed, 310 insertions(+), 11 deletions(-) create mode 100644 benchmarks/http/scores.aup (limited to 'benchmarks/http') 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()); diff --git a/benchmarks/http/drogon-baseline/main.cc b/benchmarks/http/drogon-baseline/main.cc index 3151084..631f2be 100644 --- a/benchmarks/http/drogon-baseline/main.cc +++ b/benchmarks/http/drogon-baseline/main.cc @@ -3,6 +3,12 @@ // If-Match / If-None-Match: *, 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 (see axum-baseline): POST +// /inputs/score?points=N adds N to the name in the body, GET /views/total and +// GET /views/leaders?from=F&count=C read it. Durable (either flag), each score +// is appended to DIR/scores.log and fdatasynced before the answer, blocking +// the handler's loop as the document handlers do; the log is replayed at start. +// // drogon-baseline --port P [--durable DIR | --atomic DIR] // --durable writes the document and fdatasyncs before answering (not crash // safe); --atomic writes key.tmp, fsyncs it, renames it over the key and @@ -17,6 +23,7 @@ #include #include #include +#include #include #include #include @@ -122,6 +129,94 @@ static HttpResponsePtr edit(const HttpRequestPtr &request, const std::string &ke return tagged(k200OK, doc->revision, "none", std::string(insert)); } +static std::mutex board_lock; +static std::map board; /* points by name, in name (byte) order */ +static int board_log = -1; + +/* Replay the log up to its last whole record, and keep it open for appending. */ +static void open_board() { + if (durable.empty()) return; + board_log = open((durable + "/scores.log").c_str(), O_RDWR | O_CREAT, 0644); + std::string bytes; + char chunk[65536]; + for (ssize_t n; (n = read(board_log, chunk, sizeof(chunk))) > 0;) bytes.append(chunk, (size_t)n); + size_t at = 0; + while (at + 4 <= bytes.size()) { + uint32_t length, points; + memcpy(&length, bytes.data() + at, 4); + if (at + 8 + length > bytes.size()) break; + memcpy(&points, bytes.data() + at + 4 + length, 4); + board[bytes.substr(at + 4, length)] += points; + at += 8 + length; + } + if (ftruncate(board_log, (off_t)at) || lseek(board_log, 0, SEEK_END) < 0) abort(); +} + +static HttpResponsePtr score(const HttpRequestPtr &request) { + const std::string &given = request->getParameter("points"); + char *end = nullptr; + unsigned long points = strtoul(given.c_str(), &end, 10); + std::string name(request->body()); + if (given.empty() || !end || *end || points > UINT32_MAX) return tagged(k400BadRequest, 0, "request", ""); + std::lock_guard guard(board_lock); + if (board_log >= 0) { + uint32_t length = (uint32_t)name.size(), added = (uint32_t)points; + std::string record(reinterpret_cast(&length), 4); + record += name; + record.append(reinterpret_cast(&added), 4); + if (!write_all(board_log, record) || fdatasync(board_log)) return tagged(k500InternalServerError, 0, "storage", ""); + } + board[name] += points; + return tagged(k200OK, 0, "none", std::move(name)); +} + +static HttpResponsePtr text(std::string body, ContentType type) { + auto response = HttpResponse::newHttpResponse(); + response->setContentTypeCode(type); + response->addHeader("X-Pure-Error", "none"); + response->setBody(std::move(body)); + return response; +} + +static HttpResponsePtr total() { + std::lock_guard guard(board_lock); + uint64_t sum = 0; + for (auto &entry : board) sum += entry.second; + return text(std::to_string(sum), CT_TEXT_PLAIN); +} + +static void json_text(std::string &out, const std::string &text) { + out += '"'; + for (unsigned char c : text) { + if (c == '"' || c == '\\') out += '\\', out += (char)c; + else if (c < 0x20) { + char escape[8]; + snprintf(escape, sizeof(escape), "\\u%04x", c); + out += escape; + } else out += (char)c; + } + out += '"'; +} + +static HttpResponsePtr leaders(const HttpRequestPtr &request) { + const std::string &from_text = request->getParameter("from"), &count_text = request->getParameter("count"); + size_t from = from_text.empty() ? 0 : strtoull(from_text.c_str(), nullptr, 10); + size_t count = count_text.empty() ? SIZE_MAX : strtoull(count_text.c_str(), nullptr, 10); + std::lock_guard guard(board_lock); + size_t rows = board.size(); + if (from > rows) from = rows; + std::string out = "{\"rows\":" + std::to_string(rows) + ",\"from\":" + std::to_string(from) + ",\"items\":["; + auto entry = board.begin(); + std::advance(entry, from); + for (size_t i = 0; entry != board.end() && i < count; ++entry, ++i) { + out += i ? ",{\"name\":" : "{\"name\":"; + json_text(out, entry->first); + out += ",\"points\":" + std::to_string(entry->second) + "}"; + } + out += "]}"; + return text(std::move(out), CT_APPLICATION_JSON); +} + /* drogon listens once per event loop (SO_REUSEPORT): port 0 would give each * loop a port of its own, so a free port is chosen first. */ static uint16_t free_port() { @@ -150,6 +245,23 @@ int main(int argc, char **argv) { } } if (!durable.empty()) mkdir(durable.c_str(), 0755); + open_board(); + app().registerHandler( + "/inputs/score", + [](const HttpRequestPtr &request, std::function &&callback) { + callback(score(request)); + }, + {Post}); + app().registerHandler( + "/views/total", + [](const HttpRequestPtr &, std::function &&callback) { callback(total()); }, + {Get}); + app().registerHandler( + "/views/leaders", + [](const HttpRequestPtr &request, std::function &&callback) { + callback(leaders(request)); + }, + {Get}); app().registerHandler( "/objects/{key}", [](const HttpRequestPtr &request, std::function &&callback, diff --git a/benchmarks/http/load.c b/benchmarks/http/load.c index a54ec1c..fd074e8 100644 --- a/benchmarks/http/load.c +++ b/benchmarks/http/load.c @@ -2,11 +2,14 @@ /* HTTP/1.1 load for Pure's benchmarks: N connections on one epoll thread, * each sending one request at a time and measuring its latency. * - * load PORT CONNECTIONS SECONDS get|put KEYS [keep-alive|close] [BODY] + * load PORT CONNECTIONS SECONDS get|put|score|total|page KEYS [keep-alive|close] [BODY] * get: GET /objects/kI for I in [0, KEYS). * put: connection C edits its own key kC (C mod KEYS), inserting BODY bytes * at 0 with If-Match on the revision it last saw (If-None-Match: * at * first); a 412 takes the revision from the ETag and tries again. + * The scoreboard app: score posts a point for kC (POST /inputs/score?points=1, + * the name as the body), total reads GET /views/total and page GET + * /views/leaders?from=0&count=20. * Prints one JSON line: requests/s, statuses, latency percentiles (µs). */ #include #include @@ -64,12 +67,26 @@ static uint64_t percentile(uint64_t total, double p) { } static int port, connections_count, keys; +static enum { GET, PUT, SCORE, TOTAL, PAGE } method; static bool put, keep_alive; static char *body; static size_t body_bytes; static int epoll_fd; static void prepare(Connection *c) { + const char *closing = keep_alive ? "" : "Connection: close\r\n"; + if (method >= SCORE) { + char name[16]; + int length = snprintf(name, sizeof(name), "k%d", c->key); + c->request_size = (size_t)(method == SCORE ? + snprintf(c->request, sizeof(c->request), "POST /inputs/score?points=1 HTTP/1.1\r\nHost: localhost\r\n" + "Content-Length: %d\r\n%s\r\n%s", length, closing, name) : + snprintf(c->request, sizeof(c->request), "GET /views/%s HTTP/1.1\r\nHost: localhost\r\n%s\r\n", + method == TOTAL ? "total" : "leaders?from=0&count=20", closing)); + c->sent = c->received = c->body_sent = c->body_size = 0; + c->started = now_us(); + return; + } char condition[64] = ""; if (put) { if (c->revision) snprintf(condition, sizeof(condition), "If-Match: \"%llu\"\r\n", (unsigned long long)c->revision); @@ -123,13 +140,16 @@ static bool complete(Connection *c, int *status, bool *closing, size_t *length) int main(int argc, char **argv) { if (argc < 6) { - fputs("load PORT CONNECTIONS SECONDS get|put KEYS [keep-alive|close] [BODY]\n", stderr); + fputs("load PORT CONNECTIONS SECONDS get|put|score|total|page KEYS [keep-alive|close] [BODY]\n", stderr); return 2; } port = atoi(argv[1]); connections_count = atoi(argv[2]); double seconds = atof(argv[3]); - put = !strcmp(argv[4], "put"); + static const char *const methods[] = {"get", "put", "score", "total", "page"}; + for (method = GET; method <= PAGE && strcmp(argv[4], methods[method]); ++method) {} + if (method > PAGE) { fprintf(stderr, "load: unknown method %s\n", argv[4]); return 2; } + put = method == PUT; keys = atoi(argv[5]); keep_alive = argc > 6 && !strcmp(argv[6], "keep-alive"); body_bytes = argc > 7 ? strtoul(argv[7], NULL, 10) : 16; diff --git a/benchmarks/http/run.py b/benchmarks/http/run.py index 91356f9..a09b4ef 100644 --- a/benchmarks/http/run.py +++ b/benchmarks/http/run.py @@ -16,6 +16,10 @@ result several times over. Each round therefore runs the servers in a rotated order, each on a fresh store, and every write workload first times the device (an in-place 4 KiB write + fdatasync); the table shows that probe next to each result. + +--suite app serves the scoreboard of scores.aup (built by aup) with pure-serve, +against the same scoreboard written by hand in axum and drogon: durable score +posts, a sum and a page of 20 rows as JSON. """ import argparse import json @@ -70,6 +74,15 @@ def put(port, key, body, expected): assert response.status == 200, (key, response.status) +def post_score(port, name): + connection = http.client.HTTPConnection('127.0.0.1', port, timeout=30) + connection.request('POST', '/inputs/score?points=1', name.encode(), {'Content-Length': str(len(name))}) + response = connection.getresponse() + response.read() + connection.close() + assert response.status == 200, (name, response.status) + + def probe_ms(directory, count=16): """Median time of an in-place 4 KiB write + fdatasync on the store device.""" path = directory / 'device-probe' @@ -97,7 +110,8 @@ def load(port, connections, seconds, method, keys, mode, body=16): def main(): parser = argparse.ArgumentParser() - parser.add_argument('--module', type=Path, required=True, help='editor.pmod built by aup') + parser.add_argument('--module', type=Path, required=True, help='editor.pmod built by aup, or scores.pmod for --suite app') + parser.add_argument('--suite', choices=['documents', 'app'], default='documents') parser.add_argument('--directory', type=Path, default=Path('/var/tmp')) parser.add_argument('--seconds', type=float, default=5) parser.add_argument('--only', default='') @@ -130,6 +144,17 @@ def main(): ('put 16B, 64 conn, keep-alive', dict(connections=64, method='put', keys=64, mode='keep-alive')), ('put 16B, 8 conn, keep-alive', dict(connections=8, method='put', keys=8, mode='keep-alive')), ] + app = args.suite == 'app' + if app: + # Durable posts write (the method is put for the device probe); reads are a sum and a page. + workloads = [ + ('score, 64 conn, keep-alive', dict(connections=64, method='score', keys=64, mode='keep-alive', writes=True)), + ('score, 8 conn, keep-alive', dict(connections=8, method='score', keys=8, mode='keep-alive', writes=True)), + ('total, 64 conn, keep-alive', dict(connections=64, method='total', keys=64, mode='keep-alive')), + ('page 20 rows, 64 conn, keep-alive', dict(connections=64, method='page', keys=64, mode='keep-alive')), + ] + for name in ('pure-program', 'pure-serve-3', 'axum-atomic', 'drogon-atomic'): + del servers[name] # one owner of the board's states; atomic is durable for a log if args.workloads: workloads = [w for w in workloads if re.search(args.workloads, w[0])] chosen = [name for name in servers if not args.only or name in args.only.split(',')] @@ -144,12 +169,13 @@ def main(): process, port = start(servers[name](directory)) idle = rss_kib(process.pid, 'VmRSS:') for key in range(64): - put(port, f'k{key}', b'x' * 56, 0) + if app: post_score(port, f'k{key}') + else: put(port, f'k{key}', b'x' * 56, 0) for label, workload in workloads: if workload.get('big'): # the 1 MiB document is k0 for this workload put(port, 'k0', b'B' * (1 << 20), 1) - probe = probe_ms(base) if workload['method'] == 'put' else None + probe = probe_ms(base) if workload['method'] == 'put' or workload.get('writes') else None used = cpu_seconds(process.pid) measured = load(port, workload['connections'], args.seconds, workload['method'], workload['keys'], workload['mode']) diff --git a/benchmarks/http/scores.aup b/benchmarks/http/scores.aup new file mode 100644 index 0000000..8683bf2 --- /dev/null +++ b/benchmarks/http/scores.aup @@ -0,0 +1,22 @@ +# The scoreboard of the app benchmarks (run.py --suite app), which the axum +# and drogon baselines write by hand: pure-serve serves its views and inputs, +# with no routes, handlers or storage code written. +record Entry { + name: text + points = 0 +} +state board: list of Entry by name +input score = message { + points: u32 + name: bytes +} +on score { + if !board.has(text(score.name)) { board.add(Entry { name: text(score.name) }) } + board[text(score.name)].points = board[text(score.name)].points + score.points + reply score.name +} +view total = sum(e.points for e in board) +view leaders = each e in board { + name = e.name + points = e.points +} -- cgit v1.2.3