From a7870d708bfebb647471264500ab056a4de86594 Mon Sep 17 00:00:00 2001 From: Bojan Serafimov Date: Wed, 27 Apr 2022 12:18:13 -0400 Subject: [PATCH] Cleanup --- pageserver/src/bin/psbench.rs | 80 ++++++++++++++-------- pageserver/src/wal_metadata.rs | 2 +- postgres_ffi/src/waldecoder-async.rs | 1 + test_runner/performance/test_pageserver.py | 3 + 4 files changed, 57 insertions(+), 29 deletions(-) create mode 100644 postgres_ffi/src/waldecoder-async.rs diff --git a/pageserver/src/bin/psbench.rs b/pageserver/src/bin/psbench.rs index 4e39f1f32d..c11116250d 100644 --- a/pageserver/src/bin/psbench.rs +++ b/pageserver/src/bin/psbench.rs @@ -1,11 +1,22 @@ //! Pageserver benchmark tool //! -//! 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. +//! This tool connects directly to a pageserver, issues queries and measures performance. +//! +//! Ideally the tool would be ablle to stream WAL into the pageserver, and (possibly +//! simultaneously) make read requests. Currently wal streaming is not implemented, +//! so this tool assumes the pageserver is prepopulated with some data, and only +//! issues read queries. It also currently assumes that the pageserver writes out some +//! metadata describing the write access pattern on the workload that was performed on it. +//! +//! This tool runs a variety of workloads. See the Args enum below, or run the tool +//! with --help to see the available workloads. +//! use bytes::{BufMut, BytesMut}; use clap::{Parser, Subcommand}; use futures::future; use pageserver::wal_metadata::{Page, WalEntryMetadata}; +use postgres_ffi::pg_constants::BLCKSZ; +use rand::seq::SliceRandom; use rand::thread_rng; use std::fs::File; use std::path::{Path, PathBuf}; @@ -17,16 +28,14 @@ use std::{ }; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; +use zenith_utils::zid::{ZTenantId, ZTimelineId}; use zenith_utils::{ lsn::Lsn, pq_proto::{BeMessage, FeMessage}, }; -use rand::seq::SliceRandom; use anyhow::Result; -const BYTES_IN_PAGE: usize = 8 * 1024; - /// Client for the pageserver's pagestream API struct PagestreamApi { stream: TcpStream, @@ -34,7 +43,7 @@ struct PagestreamApi { /// Good enough implementation for these tests impl PagestreamApi { - async fn connect(tenant_hex: &str, timeline: &str) -> Result { + async fn connect(tenant: &ZTenantId, timeline: &ZTimelineId) -> Result { let mut stream = TcpStream::connect("localhost:15000").await?; // Connect to pageserver @@ -48,9 +57,9 @@ impl PagestreamApi { .await?; // Enter pagestream protocol - let init_query = format!("pagestream {} {}", tenant_hex, timeline); + let init_query = format!("pagestream {} {}", tenant, timeline); tokio::select! { - _ = conn => panic!("AAAA"), + _ = conn => panic!("connection closed during pagestream initialization"), _ = client.query(init_query.as_str(), &[]) => (), }; @@ -90,7 +99,7 @@ impl PagestreamApi { 102 => { let mut page = Vec::::new(); cursor.read_to_end(&mut page).await?; - if page.len() != BYTES_IN_PAGE { + if page.len() != (BLCKSZ as usize) { panic!("Expected 8kb page, got: {:?}", page.len()); } page @@ -165,9 +174,9 @@ impl Metadata { latencies[latencies.len() / 2].as_micros() ); println!( - "lower_is_better average {:.3} microseconds", - (latencies.iter().map(|l| l.as_micros()).sum::() as f64) / ( - latencies.len() as f64) + "lower_is_better average {:.2} microseconds", + (latencies.iter().map(|l| l.as_micros()).sum::() as f64) + / (latencies.len() as f64) ); println!( "lower_is_better p99 {:?} microseconds", @@ -182,7 +191,12 @@ impl Metadata { } /// Sequentially get the latest version of each page and report latencies -async fn test_latest_pages(api: &mut PagestreamApi, metadata: &Metadata) -> Result> { +async fn test_latest_pages( + tenant: &ZTenantId, + timeline: &ZTimelineId, + metadata: &Metadata, +) -> Result> { + let mut api = PagestreamApi::connect(tenant, timeline).await.unwrap(); let mut latencies: Vec = vec![]; let mut page_order: Vec<&Page> = metadata.affected_pages.iter().collect(); page_order.shuffle(&mut thread_rng()); @@ -198,12 +212,12 @@ async fn test_latest_pages(api: &mut PagestreamApi, metadata: &Metadata) -> Resu #[derive(Parser, Debug, Clone)] struct Args { + // TODO maybe one metadata file per timeline? wal_metadata_path: PathBuf, // TODO get these from wal metadata - // TODO test multi-timeline parallel reads - tenant_hex: String, - timeline: String, + tenant: ZTenantId, + timeline: ZTimelineId, // TODO change to `clients_per_timeline` #[clap(long, default_value = "1")] @@ -215,15 +229,15 @@ struct Args { #[derive(Subcommand, Debug, Clone)] enum PsbenchTest { + /// Query the latest version of each page, in a random sequential order. + /// If multiple clients are used, all clients will independently query + /// every page in a different random order. GetLatestPages, -} - -async fn run_client(args: Args, metadata: Metadata) -> Vec { - let mut pagestream = PagestreamApi::connect(&args.tenant_hex, &args.timeline).await.unwrap(); - match args.test { - PsbenchTest::GetLatestPages => test_latest_pages(&mut pagestream, &metadata), - } - .await.unwrap() + // TODO add more tests: + // - Query with realistic read pattern + // - Query every page after every change of that page + // - Query all pages at given point in time + // - etc. } #[tokio::main] @@ -231,10 +245,20 @@ async fn main() -> Result<()> { let args = Args::parse(); let metadata = Metadata::build(&args.wal_metadata_path).unwrap(); - // TODO explicitly spawn a thread for each? - let latencies: Vec = future::join_all((0..args.num_clients).map(|_| { - run_client(args.clone(), metadata.clone()) - })).await.into_iter().flatten().collect(); + let latencies: Vec = match args.test { + PsbenchTest::GetLatestPages => { + // TODO explicitly spawn a thread for each? + future::join_all( + (0..args.num_clients) + .map(|_| test_latest_pages(&args.tenant, &args.timeline, &metadata)), + ) + .await + .into_iter() + .map(|v| v.unwrap()) + .flatten() + .collect() + } + }; println!("test_param num_clients {}", args.num_clients); metadata.report_latency(&latencies).unwrap(); diff --git a/pageserver/src/wal_metadata.rs b/pageserver/src/wal_metadata.rs index a298ba9612..c527ebb184 100644 --- a/pageserver/src/wal_metadata.rs +++ b/pageserver/src/wal_metadata.rs @@ -21,6 +21,7 @@ use zenith_utils::lsn::Lsn; use crate::{config::PageServerConf, walrecord::DecodedBkpBlock}; +// TODO make a directory instead, and write one file per (tenant, timeline). pub static WAL_METADATA_FILE: OnceCell = OnceCell::new(); #[derive(Copy, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Debug, Serialize, Deserialize)] @@ -73,7 +74,6 @@ impl From<&DecodedBkpBlock> for Page { } } -// TODO include tenant and timeline #[derive(Debug, Serialize, Deserialize, Clone)] pub struct WalEntryMetadata { pub lsn: Lsn, diff --git a/postgres_ffi/src/waldecoder-async.rs b/postgres_ffi/src/waldecoder-async.rs new file mode 100644 index 0000000000..8b13789179 --- /dev/null +++ b/postgres_ffi/src/waldecoder-async.rs @@ -0,0 +1 @@ + diff --git a/test_runner/performance/test_pageserver.py b/test_runner/performance/test_pageserver.py index 41df6ca656..0ac0e22c78 100644 --- a/test_runner/performance/test_pageserver.py +++ b/test_runner/performance/test_pageserver.py @@ -55,3 +55,6 @@ def test_get_page(zenith_env_builder: ZenithEnvBuilder, output = psbench_bin.test_latest_pages(env.initial_tenant.hex, timeline, num_clients=8) zenbenchmark.record_psbench_result("8_clients", output) + + +# TODO test concurrent get_page requests from 8 different timelines