rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-10-25 05:03:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cleanup

Browse Source
This commit is contained in:
Bojan Serafimov
2022-04-14 11:00:52 -04:00
parent 0b53968db4
commit da66df21f3
5 changed files with 170 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+56 -11
View File
@@ -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)",
pageserver/Cargo.toml
+1 -1
View File
@@ -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"] }
pageserver/src/bin/psbench.rs
+104 -71
View File
@@ -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<Item = String> {
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<WalEntryMetadata>,
// Derived from wal_metadata
total_wal_size: usize,
affected_pages: HashSet<Page>,
latest_lsn: Lsn,
}
impl Metadata {
fn build(wal_metadata_path: &PathBuf) -> Result<Metadata> {
let wal_metadata_file = File::open(wal_metadata_path)
.expect("error opening wal_metadata");
let wal_metadata: Vec<WalEntryMetadata> = 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<Duration>) -> 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<Vec<Duration>>{
let mut durations: Vec<Duration> = 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<WalEntryMetadata> = 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<Duration> = 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(())
}
test_runner/fixtures/zenith_fixtures.py
+8 -2
View File
@@ -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()
test_runner/performance/test_pageserver.py
+1 -1
View File
@@ -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)