From da66df21f370da818f6f21614c50d76951b81015 Mon Sep 17 00:00:00 2001 From: Bojan Serafimov Date: Thu, 14 Apr 2022 11:00:52 -0400 Subject: [PATCH] Cleanup --- Cargo.lock | 67 ++++++-- pageserver/Cargo.toml | 2 +- pageserver/src/bin/psbench.rs | 175 ++++++++++++--------- test_runner/fixtures/zenith_fixtures.py | 10 +- test_runner/performance/test_pageserver.py | 2 +- 5 files changed, 170 insertions(+), 86 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1a9e261281..05fdeff6ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -306,17 +306,32 @@ dependencies = [ [[package]] name = "clap" -version = "3.0.14" +version = "3.1.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b63edc3f163b3c71ec8aa23f9bd6070f77edbf3d1d198b164afa90ff00e4ec62" +checksum = "71c47df61d9e16dc010b55dba1952a57d8c215dbb533fd13cdd13369aac73b1c" dependencies = [ "atty", "bitflags", + "clap_derive", "indexmap", + "lazy_static", "os_str_bytes", "strsim 0.10.0", "termcolor", - "textwrap 0.14.2", + "textwrap 0.15.0", +] + +[[package]] +name = "clap_derive" +version = "3.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3aab4734e083b809aaf5794e14e756d1c798d2c69c7f7de7a09a2f5214993c1" +dependencies = [ + "heck 0.4.0", + "proc-macro-error", + "proc-macro2", + "quote", + "syn", ] [[package]] @@ -335,7 +350,7 @@ version = "0.1.0" dependencies = [ "anyhow", "chrono", - "clap 3.0.14", + "clap 3.1.8", "env_logger", "hyper", "libc", @@ -947,6 +962,12 @@ dependencies = [ "unicode-segmentation", ] +[[package]] +name = "heck" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2540771e65fc8cb83cd6e8a237f70c319bd5c29f78ed1084ba5d50eeac86f7f9" + [[package]] name = "hermit-abi" version = "0.1.19" @@ -1512,7 +1533,7 @@ dependencies = [ "byteorder", "bytes", "chrono", - "clap 3.0.14", + "clap 3.1.8", "const_format", "crc32c", "crossbeam-utils", @@ -1820,6 +1841,30 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eb9f9e6e233e5c4a35559a617bf40a4ec447db2e84c20b55a6f83167b7e57872" +[[package]] +name = "proc-macro-error" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da25490ff9892aab3fcf7c36f08cfb902dd3e71ca0f9f9517bea02a73a5ce38c" +dependencies = [ + "proc-macro-error-attr", + "proc-macro2", + "quote", + "syn", + "version_check", +] + +[[package]] +name = "proc-macro-error-attr" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1be40180e52ecc98ad80b184934baf3d0d29f979574e439af5a55274b35f869" +dependencies = [ + "proc-macro2", + "quote", + "version_check", +] + [[package]] name = "proc-macro-hack" version = "0.5.19" @@ -1866,7 +1911,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "62941722fb675d463659e49c4f3fe1fe792ff24fe5bbaa9c08cd3b98a1c354f5" dependencies = [ "bytes", - "heck", + "heck 0.3.3", "itertools", "lazy_static", "log", @@ -1908,7 +1953,7 @@ version = "0.1.0" dependencies = [ "anyhow", "bytes", - "clap 3.0.14", + "clap 3.1.8", "fail", "futures", "hashbrown", @@ -2627,9 +2672,9 @@ dependencies = [ [[package]] name = "textwrap" -version = "0.14.2" +version = "0.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0066c8d12af8b5acd21e00547c3797fde4e8677254a7ee429176ccebbe93dd80" +checksum = "b1141d4d61095b28419e22cb0bbf02755f5e54e0526f97f1e3d1d160e60885fb" [[package]] name = "thiserror" @@ -3141,7 +3186,7 @@ dependencies = [ "anyhow", "byteorder", "bytes", - "clap 3.0.14", + "clap 3.1.8", "const_format", "crc32c", "daemonize", @@ -3408,7 +3453,7 @@ name = "zenith" version = "0.1.0" dependencies = [ "anyhow", - "clap 3.0.14", + "clap 3.1.8", "control_plane", "pageserver", "postgres 0.19.1 (git+https://github.com/zenithdb/rust-postgres.git?rev=2949d98df52587d562986aad155dd4e889e408b7)", diff --git a/pageserver/Cargo.toml b/pageserver/Cargo.toml index dccdca291c..de7ba30320 100644 --- a/pageserver/Cargo.toml +++ b/pageserver/Cargo.toml @@ -15,7 +15,7 @@ hyper = "0.14" itertools = "0.10.3" lazy_static = "1.4.0" log = "0.4.14" -clap = "3.0" +clap = { version = "3.1.8", features = ["derive"] } daemonize = "0.4.1" tokio = { version = "1.17", features = ["process", "sync", "macros", "fs", "rt", "io-util", "time"] } tokio-util = { version = "0.7", features = ["io"] } diff --git a/pageserver/src/bin/psbench.rs b/pageserver/src/bin/psbench.rs index ce04bf3c04..bc327ae917 100644 --- a/pageserver/src/bin/psbench.rs +++ b/pageserver/src/bin/psbench.rs @@ -3,9 +3,11 @@ //! Usually it's easier to write python perf tests, but here the performance //! of the tester matters, and the pagestream API is easier to call from rust. use bytes::{BufMut, BytesMut}; -use clap::{App, Arg}; +use clap::{Parser, Subcommand}; use pageserver::wal_metadata::{Page, WalEntryMetadata}; +use tokio::net::TcpStream; use std::fs::File; +use std::path::PathBuf; use std::time::Instant; use std::{ collections::HashSet, @@ -16,19 +18,12 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; use zenith_utils::{ lsn::Lsn, pq_proto::{BeMessage, FeMessage}, - GIT_VERSION, }; use anyhow::Result; const BYTES_IN_PAGE: usize = 8 * 1024; -pub fn read_lines_buffered(file_name: &str) -> impl Iterator { - BufReader::new(File::open(file_name).unwrap()) - .lines() - .map(|result| result.unwrap()) -} - pub async fn get_page( pagestream: &mut tokio::net::TcpStream, lsn: &Lsn, @@ -85,87 +80,125 @@ pub async fn get_page( Ok(page) } +struct Metadata { + // Parsed from metadata file + wal_metadata: Vec, + + // Derived from wal_metadata + total_wal_size: usize, + affected_pages: HashSet, + latest_lsn: Lsn, +} + +impl Metadata { + fn build(wal_metadata_path: &PathBuf) -> Result { + let wal_metadata_file = File::open(wal_metadata_path) + .expect("error opening wal_metadata"); + let wal_metadata: Vec = BufReader::new(wal_metadata_file) + .lines() + .map(|result| result.expect("error reading from file")) + .map(|line| serde_json::from_str(&line).expect("corrupt metadata file")) + .collect(); + + + let total_wal_size: usize = wal_metadata.iter().map(|m| m.size).sum(); + let affected_pages: HashSet<_> = wal_metadata + .iter() + .map(|m| m.affected_pages.clone()) + .flatten() + .collect(); + let latest_lsn = wal_metadata.iter().map(|m| m.lsn).max().unwrap(); + + Ok(Metadata { + wal_metadata, + total_wal_size, + affected_pages, + latest_lsn, + }) + } + + fn report_results(&self, durations: &Vec) -> Result<()> { + // Format is optimized for easy parsing from benchmark_fixture.py + println!("test_param num_pages {}", self.affected_pages.len()); + println!("test_param num_wal_entries {}", self.wal_metadata.len()); + println!("test_param total_wal_size {} bytes", self.total_wal_size); + println!( + "lower_is_better fastest {:?} microseconds", + durations.first().unwrap().as_micros() + ); + println!( + "lower_is_better median {:?} microseconds", + durations[durations.len() / 2].as_micros() + ); + println!( + "lower_is_better p99 {:?} microseconds", + durations[durations.len() - 1 - durations.len() / 100].as_micros() + ); + println!( + "lower_is_better slowest {:?} microseconds", + durations.last().unwrap().as_micros() + ); + Ok(()) + } +} + +async fn test_latest_pages(pagestream: &mut TcpStream, metadata: &Metadata) -> Result>{ + let mut durations: Vec = vec![]; + for page in &metadata.affected_pages { + let start = Instant::now(); + let _page_bytes = get_page(pagestream, &metadata.latest_lsn, &page, true).await?; + let duration = start.elapsed(); + + durations.push(duration); + } + durations.sort(); + Ok(durations) +} + +#[derive(Parser, Debug)] +struct Args { + wal_metadata_path: PathBuf, + tenant_hex: String, + timeline: String, + + #[clap(subcommand)] + test: PsbenchTest, +} + +#[derive(Subcommand, Debug)] +enum PsbenchTest { + GetLatestPages, +} + #[tokio::main] async fn main() -> Result<()> { - let arg_matches = App::new("LALALA") - .about("lalala") - .version(GIT_VERSION) - .arg( - Arg::new("wal_metadata_file") - .help("Path to wal metadata file") - .required(true) - .index(1), - ) - .arg(Arg::new("tenant_hex").help("TODO").required(true).index(2)) - .arg(Arg::new("timeline").help("TODO").required(true).index(3)) - .get_matches(); + let args = Args::parse(); - let metadata_file = arg_matches.value_of("wal_metadata_file").unwrap(); - let tenant_hex = arg_matches.value_of("tenant_hex").unwrap(); - let timeline = arg_matches.value_of("timeline").unwrap(); - - // Parse log lines - let wal_metadata: Vec = read_lines_buffered(metadata_file) - .map(|line| serde_json::from_str(&line).expect("corrupt metadata file")) - .collect(); + // Parse wal metadata from file + let metadata = Metadata::build(&args.wal_metadata_path)?; // Get raw TCP connection to the pageserver postgres protocol port - let mut socket = tokio::net::TcpStream::connect("localhost:15000").await?; + let mut pagestream = TcpStream::connect("localhost:15000").await?; let (client, conn) = tokio_postgres::Config::new() .host("127.0.0.1") .port(15000) .dbname("postgres") .user("zenith_admin") - .connect_raw(&mut socket, tokio_postgres::NoTls) + .connect_raw(&mut pagestream, tokio_postgres::NoTls) .await?; // Enter pagestream protocol - let init_query = format!("pagestream {} {}", tenant_hex, timeline); + let init_query = format!("pagestream {} {}", args.tenant_hex, args.timeline); tokio::select! { _ = conn => panic!("AAAA"), _ = client.query(init_query.as_str(), &[]) => (), }; - // Derive some variables - let total_wal_size: usize = wal_metadata.iter().map(|m| m.size).sum(); - let affected_pages: HashSet<_> = wal_metadata - .iter() - .map(|m| m.affected_pages.clone()) - .flatten() - .collect(); - let latest_lsn = wal_metadata.iter().map(|m| m.lsn).max().unwrap(); - - // Get all latest pages - let mut durations: Vec = vec![]; - for page in &affected_pages { - let start = Instant::now(); - let _page_bytes = get_page(&mut socket, &latest_lsn, &page, true).await?; - let duration = start.elapsed(); - - durations.push(duration); - } - - durations.sort(); - // Format is optimized for easy parsing from benchmark_fixture.py - println!("test_param num_pages {}", affected_pages.len()); - println!("test_param num_wal_entries {}", wal_metadata.len()); - println!("test_param total_wal_size {} bytes", total_wal_size); - println!( - "lower_is_better fastest {:?} microseconds", - durations.first().unwrap().as_micros() - ); - println!( - "lower_is_better median {:?} microseconds", - durations[durations.len() / 2].as_micros() - ); - println!( - "lower_is_better p99 {:?} microseconds", - durations[durations.len() - 1 - durations.len() / 100].as_micros() - ); - println!( - "lower_is_better slowest {:?} microseconds", - durations.last().unwrap().as_micros() - ); + // Run test + let durations = match args.test { + PsbenchTest::GetLatestPages => test_latest_pages(&mut pagestream, &metadata) + }.await?; + metadata.report_results(&durations)?; Ok(()) } diff --git a/test_runner/fixtures/zenith_fixtures.py b/test_runner/fixtures/zenith_fixtures.py index 2078644c52..ac895ddee7 100644 --- a/test_runner/fixtures/zenith_fixtures.py +++ b/test_runner/fixtures/zenith_fixtures.py @@ -1294,9 +1294,15 @@ class PsbenchBin: """A helper class for running the pageserver benchmarker tool.""" wal_metadata_path: str - def run(self, tenant_hex: str, timeline: str) -> str: + def test_latest_pages(self, tenant_hex: str, timeline: str) -> str: psbench_binpath = os.path.join(str(zenith_binpath), 'psbench') - args = [psbench_binpath, self.wal_metadata_path, tenant_hex, timeline] + args = [ + psbench_binpath, + self.wal_metadata_path, + tenant_hex, + timeline, + "get-latest-pages", + ] return subprocess.run(args, capture_output=True).stdout.decode("UTF-8").strip() diff --git a/test_runner/performance/test_pageserver.py b/test_runner/performance/test_pageserver.py index dd289eaeb3..9ee9293c89 100644 --- a/test_runner/performance/test_pageserver.py +++ b/test_runner/performance/test_pageserver.py @@ -51,5 +51,5 @@ def test_get_page(zenith_env_builder: ZenithEnvBuilder, pscur.execute(f"checkpoint {env.initial_tenant.hex} {timeline} 0") - output = psbench_bin.run(env.initial_tenant.hex, timeline) + output = psbench_bin.test_latest_pages(env.initial_tenant.hex, timeline) zenbenchmark.record_psbench_result(output)