diff options
Diffstat (limited to 'benchmarks/http')
| -rw-r--r-- | benchmarks/http/README.md | 12 | ||||
| -rw-r--r-- | benchmarks/http/axum-baseline/Cargo.lock | 518 | ||||
| -rw-r--r-- | benchmarks/http/axum-baseline/Cargo.toml | 17 | ||||
| -rw-r--r-- | benchmarks/http/axum-baseline/src/main.rs | 159 | ||||
| -rw-r--r-- | benchmarks/http/load.c | 209 | ||||
| -rw-r--r-- | benchmarks/http/run.py | 133 |
6 files changed, 1048 insertions, 0 deletions
diff --git a/benchmarks/http/README.md b/benchmarks/http/README.md new file mode 100644 index 0000000..3fc5740 --- /dev/null +++ b/benchmarks/http/README.md @@ -0,0 +1,12 @@ +# HTTP 벤치마크 + +Pure의 문서 서버들과 같은 계약의 axum 서비스를 같은 기기·코어 배정·부하 생성기로 잰다([075](../../docs/research/075-http-host-against-axum.md)). + +```sh +cc -O2 -std=c17 -o load load.c +(cd axum-baseline && cargo build --release) +make -C ../../prototypes/connections build/pure-connections-module +make -C ../../prototypes/connections BUILD=build/tree STORE=store-tree.c build/tree/pure-connections +aup build ../../../aup/examples/editor.aup -o editor.pmod +python3 run.py --module editor.pmod --directory /var/tmp # 저장소를 둘 장치에서 +``` diff --git a/benchmarks/http/axum-baseline/Cargo.lock b/benchmarks/http/axum-baseline/Cargo.lock new file mode 100644 index 0000000..a3a7869 --- /dev/null +++ b/benchmarks/http/axum-baseline/Cargo.lock @@ -0,0 +1,518 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "form_urlencoded", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-baseline" +version = "0.1.0" +dependencies = [ + "axum", + "bytes", + "tokio", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", +] + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "slab", +] + +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + +[[package]] +name = "hyper" +version = "1.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27b501faa50e7a26c3d3560ca625132f4078a17771f4810baf70475ae48cbe43" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", +] + +[[package]] +name = "hyper-util" +version = "0.1.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddc03d96684f9226b8a787cdb71488417b53ab5ea8fdb1dac946cb9431cc8bff" +dependencies = [ + "bytes", + "http", + "http-body", + "hyper", + "pin-project-lite", + "tokio", + "tower-service", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + +[[package]] +name = "log" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" + +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + +[[package]] +name = "mio" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "serde_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.16.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9395f0f0eee849a9b707b2f06bb92a6a422090e2123bb2ef8e87a0e61892a8e" + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "syn" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "libc", + "mio", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "futures-core", + "futures-util", + "pin-project-lite", + "sync_wrapper", + "tokio", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "log", + "pin-project-lite", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + +[[package]] +name = "unicode-ident" +version = "1.0.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/benchmarks/http/axum-baseline/Cargo.toml b/benchmarks/http/axum-baseline/Cargo.toml new file mode 100644 index 0000000..1a26451 --- /dev/null +++ b/benchmarks/http/axum-baseline/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "axum-baseline" +version = "0.1.0" +edition = "2021" +publish = false + +# A reference service for Pure's HTTP benchmarks: the same document contract +# as Pure's document server, written the way an axum service usually is. +[dependencies] +axum = "0.8" +tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "fs", "sync", "signal"] } +bytes = "1" + +[profile.release] +lto = "fat" +codegen-units = 1 +panic = "abort" diff --git a/benchmarks/http/axum-baseline/src/main.rs b/benchmarks/http/axum-baseline/src/main.rs new file mode 100644 index 0000000..6684bfc --- /dev/null +++ b/benchmarks/http/axum-baseline/src/main.rs @@ -0,0 +1,159 @@ +//! Reference service for Pure's HTTP benchmarks: the same contract as Pure's +//! document server (GET/PUT /objects/{key}, 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). +//! +//! 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 +//! handler would (not crash-safe: a crash after truncation loses it). +//! --atomic writes DIR/key.tmp, fsyncs it, renames it over DIR/key and fsyncs +//! the directory: the usual crash-safe replacement. +use axum::{ + body::Bytes, + extract::{Path, State}, + http::{HeaderMap, HeaderValue, StatusCode}, + response::{IntoResponse, Response}, + routing::get, + Router, +}; +use std::{collections::HashMap, path::PathBuf, sync::Arc}; +use tokio::sync::Mutex; + +#[derive(Default)] +struct Document { + bytes: Vec<u8>, + revision: u64, +} + +struct Service { + documents: Mutex<HashMap<String, Arc<Mutex<Document>>>>, + durable: Option<PathBuf>, + atomic: bool, +} + +fn valid_key(key: &str) -> bool { + !key.is_empty() + && key.len() <= 64 + && key.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') +} + +fn tagged(status: StatusCode, revision: u64, error: &'static str, body: Bytes) -> Response { + let mut response = (status, body).into_response(); + let headers = response.headers_mut(); + headers.insert("X-Pure-Error", HeaderValue::from_static(error)); + if revision > 0 { + headers.insert("ETag", HeaderValue::from_str(&format!("\"{revision}\"")).unwrap()); + } + response +} + +async fn document(service: &Service, key: &str) -> Arc<Mutex<Document>> { + let mut documents = service.documents.lock().await; + documents.entry(key.to_string()).or_default().clone() +} + +async fn show(State(service): State<Arc<Service>>, Path(key): Path<String>) -> Response { + if !valid_key(&key) { + return tagged(StatusCode::BAD_REQUEST, 0, "request", Bytes::new()); + } + let document = document(&service, &key).await; + let document = document.lock().await; + if document.revision == 0 { + return tagged(StatusCode::NOT_FOUND, 0, "not_found", Bytes::new()); + } + tagged(StatusCode::OK, document.revision, "none", Bytes::copy_from_slice(&document.bytes)) +} + +async fn edit( + State(service): State<Arc<Service>>, + Path(key): Path<String>, + headers: HeaderMap, + body: Bytes, +) -> Response { + let expected = match (headers.get("if-match"), headers.get("if-none-match")) { + (Some(tag), None) => tag + .to_str() + .ok() + .and_then(|t| t.strip_prefix('"')?.strip_suffix('"')?.parse::<u64>().ok()), + (None, Some(star)) if star == "*" => Some(0), + _ => None, + }; + let Some(expected) = expected.filter(|_| valid_key(&key) && body.len() >= 8) else { + return tagged(StatusCode::BAD_REQUEST, 0, "request", Bytes::new()); + }; + let start = u32::from_le_bytes(body[0..4].try_into().unwrap()) as usize; + let remove = u32::from_le_bytes(body[4..8].try_into().unwrap()) as usize; + let document = document(&service, &key).await; + let mut document = document.lock().await; + if document.revision != expected { + return tagged(StatusCode::PRECONDITION_FAILED, document.revision, "conflict", Bytes::new()); + } + if start > document.bytes.len() || remove > document.bytes.len() - start { + return tagged(StatusCode::PAYLOAD_TOO_LARGE, 0, "limit", Bytes::new()); + } + let insert = body.slice(8..); + document.bytes.splice(start..start + remove, insert.iter().copied()); + if let Some(directory) = &service.durable { + let (directory, path, atomic) = (directory.clone(), directory.join(&key), service.atomic); + let bytes = document.bytes.clone(); + let written = tokio::task::spawn_blocking(move || -> std::io::Result<()> { + use std::io::Write; + if !atomic { + let mut file = std::fs::File::create(path)?; + file.write_all(&bytes)?; + return file.sync_data(); + } + let temporary = path.with_extension("tmp"); + let mut file = std::fs::File::create(&temporary)?; + file.write_all(&bytes)?; + file.sync_all()?; + std::fs::rename(&temporary, &path)?; + std::fs::File::open(&directory)?.sync_all() + }) + .await; + if !matches!(written, Ok(Ok(()))) { + return tagged(StatusCode::INTERNAL_SERVER_ERROR, 0, "storage", Bytes::new()); + } + } + document.revision += 1; + tagged(StatusCode::OK, document.revision, "none", insert) +} + +#[tokio::main] +async fn main() { + let arguments: Vec<String> = std::env::args().collect(); + let mut port = 8768u16; + let mut durable = None; + let mut atomic = false; + let mut at = 1; + while at + 1 < arguments.len() { + match arguments[at].as_str() { + "--port" => port = arguments[at + 1].parse().expect("port"), + "--durable" => durable = Some(PathBuf::from(&arguments[at + 1])), + "--atomic" => { + durable = Some(PathBuf::from(&arguments[at + 1])); + atomic = true; + } + other => panic!("unknown argument {other}"), + } + at += 2; + } + 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 app = Router::new() + .route("/objects/{key}", get(show).put(edit)) + .with_state(service); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", port)).await.expect("bind"); + println!("{{\"port\":{}}}", listener.local_addr().unwrap().port()); + use std::io::Write; + std::io::stdout().flush().unwrap(); + axum::serve(listener, app) + .with_graceful_shutdown(async { + tokio::signal::ctrl_c().await.ok(); + }) + .await + .unwrap(); +} diff --git a/benchmarks/http/load.c b/benchmarks/http/load.c new file mode 100644 index 0000000..a54ec1c --- /dev/null +++ b/benchmarks/http/load.c @@ -0,0 +1,209 @@ +#define _GNU_SOURCE +/* 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] + * 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. + * Prints one JSON line: requests/s, statuses, latency percentiles (µs). */ +#include <arpa/inet.h> +#include <errno.h> +#include <fcntl.h> +#include <netinet/in.h> +#include <netinet/tcp.h> +#include <stdbool.h> +#include <stdint.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/epoll.h> +#include <sys/socket.h> +#include <time.h> +#include <unistd.h> + +enum { BUCKETS = 64 * 16, RESPONSE = 1 << 21 }; + +typedef struct { + int fd, key; + bool connected; + char request[512]; + size_t request_size, sent, body_size, body_sent; + char *response; + size_t received; + uint64_t started, revision; +} Connection; + +static uint64_t now_us(void) { + struct timespec t; + clock_gettime(CLOCK_MONOTONIC, &t); + return (uint64_t)t.tv_sec * 1000000u + (uint64_t)t.tv_nsec / 1000u; +} + +/* Log-linear histogram: 16 buckets per power of two. */ +static uint64_t histogram[BUCKETS]; +static unsigned bucket(uint64_t us) { + if (us < 16) return (unsigned)us; + unsigned power = 63 - (unsigned)__builtin_clzll(us); + unsigned index = (power - 3) * 16 + (unsigned)((us >> (power - 4)) & 15); + return index < BUCKETS ? index : BUCKETS - 1; +} +static uint64_t bucket_value(unsigned index) { + if (index < 16) return index; + unsigned power = index / 16 + 3; + return ((uint64_t)16 + index % 16) << (power - 4); +} +static uint64_t percentile(uint64_t total, double p) { + uint64_t want = (uint64_t)(total * p), seen = 0; + for (unsigned i = 0; i < BUCKETS; ++i) { + seen += histogram[i]; + if (seen > want) return bucket_value(i); + } + return 0; +} + +static int port, connections_count, keys; +static bool put, keep_alive; +static char *body; +static size_t body_bytes; +static int epoll_fd; + +static void prepare(Connection *c) { + char condition[64] = ""; + if (put) { + if (c->revision) snprintf(condition, sizeof(condition), "If-Match: \"%llu\"\r\n", (unsigned long long)c->revision); + else snprintf(condition, sizeof(condition), "If-None-Match: *\r\n"); + } + c->request_size = (size_t)snprintf(c->request, sizeof(c->request), + "%s /objects/k%d HTTP/1.1\r\nHost: localhost\r\nContent-Length: %zu\r\n%s%s\r\n", + put ? "PUT" : "GET", c->key, put ? body_bytes + 8 : 0, condition, keep_alive ? "" : "Connection: close\r\n"); + c->sent = c->received = c->body_sent = 0; + c->body_size = put ? body_bytes + 8 : 0; + c->started = now_us(); +} + +static void open_connection(Connection *c) { + c->fd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0); + int one = 1; + setsockopt(c->fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)); + struct sockaddr_in address = {.sin_family = AF_INET, .sin_port = htons((uint16_t)port), + .sin_addr.s_addr = htonl(INADDR_LOOPBACK)}; + connect(c->fd, (struct sockaddr *)&address, sizeof(address)); + c->connected = false; + struct epoll_event event = {.events = EPOLLOUT | EPOLLIN, .data.ptr = c}; + epoll_ctl(epoll_fd, EPOLL_CTL_ADD, c->fd, &event); +} + +static void reopen(Connection *c) { + epoll_ctl(epoll_fd, EPOLL_CTL_DEL, c->fd, NULL); + close(c->fd); + open_connection(c); +} + +static uint64_t completed, statuses[6], errors; + +/* A complete response in c->response, or false. */ +static bool complete(Connection *c, int *status, bool *closing, size_t *length) { + char *end = memmem(c->response, c->received, "\r\n\r\n", 4); + if (!end) return false; + size_t header = (size_t)(end - c->response) + 4, content = 0; + char *field = memmem(c->response, header, "Content-Length:", 15); + if (!field) field = memmem(c->response, header, "content-length:", 15); + if (field) content = strtoul(field + 15, NULL, 10); + if (c->received < header + content) return false; + *status = atoi(c->response + 9); + *closing = memmem(c->response, header, "onnection: close", 16) != NULL; + char *tag = memmem(c->response, header, "ETag: \"", 7); + if (!tag) tag = memmem(c->response, header, "etag: \"", 7); + if (tag) c->revision = strtoull(tag + 7, NULL, 10); + *length = header + content; + return true; +} + +int main(int argc, char **argv) { + if (argc < 6) { + fputs("load PORT CONNECTIONS SECONDS get|put 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"); + keys = atoi(argv[5]); + keep_alive = argc > 6 && !strcmp(argv[6], "keep-alive"); + body_bytes = argc > 7 ? strtoul(argv[7], NULL, 10) : 16; + body = calloc(body_bytes + 8, 1); + memset(body + 8, 'x', body_bytes); + epoll_fd = epoll_create1(0); + Connection *all = calloc((size_t)connections_count, sizeof(*all)); + for (int i = 0; i < connections_count; ++i) { + all[i].key = i % keys; + all[i].response = malloc(RESPONSE); + open_connection(&all[i]); + prepare(&all[i]); + } + uint64_t start = now_us(), stop = start + (uint64_t)(seconds * 1e6); + struct epoll_event events[256]; + while (now_us() < stop) { + int ready = epoll_wait(epoll_fd, events, 256, 100); + for (int e = 0; e < ready; ++e) { + Connection *c = events[e].data.ptr; + if (events[e].events & (EPOLLERR | EPOLLHUP) && !(events[e].events & EPOLLIN)) { + ++errors; + reopen(c); + prepare(c); + continue; + } + if (events[e].events & EPOLLOUT) { + c->connected = true; + while (c->sent < c->request_size) { + ssize_t n = send(c->fd, c->request + c->sent, c->request_size - c->sent, MSG_NOSIGNAL); + if (n <= 0) break; + c->sent += (size_t)n; + } + while (c->sent == c->request_size && c->body_sent < c->body_size) { + ssize_t n = send(c->fd, body + c->body_sent, c->body_size - c->body_sent, MSG_NOSIGNAL); + if (n <= 0) break; + c->body_sent += (size_t)n; + } + if (c->sent == c->request_size && c->body_sent == c->body_size) { + struct epoll_event event = {.events = EPOLLIN, .data.ptr = c}; + epoll_ctl(epoll_fd, EPOLL_CTL_MOD, c->fd, &event); + } + } + if (events[e].events & EPOLLIN) { + ssize_t n = recv(c->fd, c->response + c->received, RESPONSE - c->received, 0); + if (n <= 0) { + if (n < 0 && errno == EAGAIN) continue; + ++errors; + reopen(c); + prepare(c); + continue; + } + c->received += (size_t)n; + int status; + bool closing; + size_t length; + if (!complete(c, &status, &closing, &length)) continue; + ++completed; + ++histogram[bucket(now_us() - c->started)]; + ++statuses[status / 100 < 6 ? status / 100 : 0]; + if (!keep_alive || closing) reopen(c); + else { + struct epoll_event event = {.events = EPOLLOUT | EPOLLIN, .data.ptr = c}; + epoll_ctl(epoll_fd, EPOLL_CTL_MOD, c->fd, &event); + } + prepare(c); + } + } + } + double elapsed = (double)(now_us() - start) / 1e6; + printf("{\"requests_per_second\":%.0f,\"completed\":%llu,\"2xx\":%llu,\"4xx\":%llu,\"5xx\":%llu,\"errors\":%llu," + "\"p50_us\":%llu,\"p99_us\":%llu,\"max_us\":%llu}\n", + (double)completed / elapsed, (unsigned long long)completed, (unsigned long long)statuses[2], + (unsigned long long)statuses[4], (unsigned long long)statuses[5], (unsigned long long)errors, + (unsigned long long)percentile(completed, 0.5), (unsigned long long)percentile(completed, 0.99), + (unsigned long long)percentile(completed, 0.99999)); + return 0; +} diff --git a/benchmarks/http/run.py b/benchmarks/http/run.py new file mode 100644 index 0000000..8202440 --- /dev/null +++ b/benchmarks/http/run.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 +"""HTTP benchmarks: Pure's document servers against an axum service with the +same contract, on the same machine, with the same load generator. + +Servers are pinned to CPUs 0-2 and the load to CPU 3; stores live under +--directory (use the device you care about: fsync costs are the device's). +Prints one JSON object per (server, workload) and a summary table. +""" +import argparse +import json +import os +import select +import shutil +import signal +import struct +import subprocess +import sys +import tempfile +import time +import http.client +from pathlib import Path + +HERE = Path(__file__).resolve().parent +PURE = HERE.parent.parent / 'prototypes' / 'connections' + + +def start(args): + process = subprocess.Popen(['taskset', '-c', '0-2'] + [str(a) for a in args], + stdout=subprocess.PIPE, stderr=subprocess.PIPE) + ready, _, _ = select.select([process.stdout], [], [], 20) + assert ready, 'server did not start' + port = json.loads(process.stdout.readline())['port'] + return process, port + + +def rss_kib(pid, field): + for line in Path(f'/proc/{pid}/status').read_text().splitlines(): + if line.startswith(field): + return int(line.split()[1]) + return 0 + + +def put(port, key, body, expected): + connection = http.client.HTTPConnection('127.0.0.1', port, timeout=30) + headers = {'Content-Length': str(len(body) + 8)} + headers['If-None-Match' if expected == 0 else 'If-Match'] = '*' if expected == 0 else f'"{expected}"' + connection.request('PUT', f'/objects/{key}', struct.pack('<II', 0, 0) + body, headers) + response = connection.getresponse() + response.read() + connection.close() + assert response.status == 200, (key, response.status) + + +def load(port, connections, seconds, method, keys, mode, body=16): + output = subprocess.run(['taskset', '-c', '3', str(HERE / 'load'), str(port), str(connections), str(seconds), + method, str(keys), mode, str(body)], capture_output=True, text=True, check=True) + return json.loads(output.stdout) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--module', type=Path, required=True, help='editor.pmod built by aup') + parser.add_argument('--directory', type=Path, default=Path('/var/tmp')) + parser.add_argument('--seconds', type=float, default=5) + parser.add_argument('--only', default='') + parser.add_argument('--budget', type=int, default=64 << 20, help="Pure's payload budget in bytes") + args = parser.parse_args() + axum = HERE / 'axum-baseline/target/release/axum-baseline' + base = Path(tempfile.mkdtemp(dir=args.directory, prefix='pure-http-bench-')) + servers = { + 'pure-program': lambda d: [PURE / 'build/tree/pure-connections', '--program', PURE / 'examples/editor.pure', + '--store', d, '--port', '0', '--budget', str(args.budget)], + 'pure-module': lambda d: [PURE / 'build/pure-connections-module', '--module', args.module, '--store', d, + '--port', '0', '--budget', str(args.budget)], + 'axum-memory': lambda d: [axum, '--port', '0'], + 'axum-durable': lambda d: [axum, '--port', '0', '--durable', d], + 'axum-atomic': lambda d: [axum, '--port', '0', '--atomic', d], + } + workloads = [ + ('get 64B, 64 conn, close', dict(connections=64, method='get', keys=64, mode='close')), + ('get 64B, 64 conn, keep-alive', dict(connections=64, method='get', keys=64, mode='keep-alive')), + ('get 1MiB, 8 conn, keep-alive', dict(connections=8, method='get', keys=1, mode='keep-alive', big=True)), + ('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')), + ] + results = [] + try: + for name, command in servers.items(): + if args.only and name not in args.only.split(','): + continue + directory = base / name + process, port = start(command(directory)) + idle = rss_kib(process.pid, 'VmRSS:') + for key in range(64): + put(port, f'k{key}', b'x' * 56, 0) + big = b'B' * (1 << 20) + put(port, 'big', big, 0) + for label, workload in workloads: + keys = workload['keys'] + if workload.get('big'): + # the 1 MiB document is k0 for this workload + put(port, 'k0', big, 1) + if workload['method'] == 'put': + # fresh keys per run so every connection starts at If-None-Match: * + pass + measured = load(port, workload['connections'], args.seconds, workload['method'], keys, + workload['mode']) + measured.update(server=name, workload=label, rss_kib=rss_kib(process.pid, 'VmRSS:'), + ok_per_second=round(measured['2xx'] / args.seconds)) + results.append(measured) + print(json.dumps(measured), flush=True) + peak = rss_kib(process.pid, 'VmHWM:') + process.send_signal(signal.SIGINT if name.startswith('axum') else signal.SIGTERM) + process.communicate(timeout=30) + for r in results: + if r['server'] == name: + r['idle_rss_kib'], r['peak_rss_kib'] = idle, peak + finally: + shutil.rmtree(base, ignore_errors=True) + print() + print(f"{'workload':32s} {'server':14s} {'2xx/s':>9s} {'p50 us':>8s} {'p99 us':>8s} {'errors':>7s} {'non-2xx':>8s}") + for r in results: + print(f"{r['workload']:32s} {r['server']:14s} {r['ok_per_second']:9d} {r['p50_us']:8d} {r['p99_us']:8d} " + f"{r['errors']:7d} {r['4xx'] + r['5xx']:8d}") + print() + for name in servers: + rows = [r for r in results if r['server'] == name] + if rows: + print(f"{name:14s} idle RSS {rows[0]['idle_rss_kib']} KiB, peak RSS {rows[0]['peak_rss_kib']} KiB") + + +if __name__ == '__main__': + sys.exit(main()) |