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.

5 files changed, 310 insertions, 11 deletions

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<HashMap<String, Arc<Mutex<Document>>>>,
     durable: Option<PathBuf>,
     atomic: bool,
+    board: Mutex<Board>,
+}
+
+/// The scoreboard: points by name, in name (byte) order, and the log.
+struct Board {
+    points: BTreeMap<String, u64>,
+    log: Option<Arc<std::fs::File>>,
 }
 
 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<T: std::str::FromStr>(query: &Option<String>, name: &str) -> Option<T> {
+    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<Arc<Service>>, RawQuery(query): RawQuery, body: Bytes) -> Response {
+    let (Some(points), Ok(name)) = (query_value::<u32>(&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<Arc<Service>>) -> Response {
+    let board = service.board.lock().await;
+    text(StatusCode::OK, "none", board.points.values().sum::<u64>().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<Arc<Service>>, RawQuery(query): RawQuery) -> Response {
+    let from = query_value::<usize>(&query, "from").unwrap_or(0);
+    let count = query_value::<usize>(&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<PathBuf>) -> 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<String> = 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 <unistd.h>
 #include <cstdio>
 #include <cstring>
+#include <map>
 #include <memory>
 #include <mutex>
 #include <string>
@@ -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<std::string, uint64_t> 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<std::mutex> guard(board_lock);
+    if (board_log >= 0) {
+        uint32_t length = (uint32_t)name.size(), added = (uint32_t)points;
+        std::string record(reinterpret_cast<const char *>(&length), 4);
+        record += name;
+        record.append(reinterpret_cast<const char *>(&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<std::mutex> 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<std::mutex> 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<void(const HttpResponsePtr &)> &&callback) {
+            callback(score(request));
+        },
+        {Post});
+    app().registerHandler(
+        "/views/total",
+        [](const HttpRequestPtr &, std::function<void(const HttpResponsePtr &)> &&callback) { callback(total()); },
+        {Get});
+    app().registerHandler(
+        "/views/leaders",
+        [](const HttpRequestPtr &request, std::function<void(const HttpResponsePtr &)> &&callback) {
+            callback(leaders(request));
+        },
+        {Get});
     app().registerHandler(
         "/objects/{key}",
         [](const HttpRequestPtr &request, std::function<void(const HttpResponsePtr &)> &&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 <arpa/inet.h>
 #include <errno.h>
@@ -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
+}