diff --git a/docs/runtime-diagnostics.md b/docs/runtime-diagnostics.md new file mode 100644 index 0000000..3ed89ea --- /dev/null +++ b/docs/runtime-diagnostics.md @@ -0,0 +1,13 @@ +# Runtime Diagnostics + +Set `DIAGNOSTICS=1` to print aggregate pipeline measurements at process exit: + +- `diagnostics.elapsed_seconds`: wall-clock duration. +- `diagnostics.generated_seeds`, `diagnostics.csv_bytes`, and `diagnostics.csv_mib_per_second`: generated payload volume and wall-clock payload rate. +- `diagnostics.generation_aggregate_seconds` and `diagnostics.generation_average_ms`: CSV-generation cost. +- `diagnostics.channel_send_wait_aggregate_seconds`: worker time blocked sending to the bounded channel. +- `diagnostics.writer_connection_aggregate_seconds`: PostgreSQL connection setup time. +- `diagnostics.batch_receive_aggregate_seconds`, `diagnostics.transaction_start_aggregate_seconds`, `diagnostics.star_copy_aggregate_seconds`, `diagnostics.planet_copy_aggregate_seconds`, and `diagnostics.commit_aggregate_seconds`: writer-stage time. +- `diagnostics.transactions` and `diagnostics.batch_size_min`, `diagnostics.batch_size_average`, and `diagnostics.batch_size_max`: actual, rather than configured, batching behavior. + +All stage durations are aggregate thread time. They can exceed `diagnostics.elapsed_seconds` when worker or writer threads overlap. Diagnostics are disabled by default. diff --git a/inserter/src/diagnostics.rs b/inserter/src/diagnostics.rs new file mode 100644 index 0000000..424231c --- /dev/null +++ b/inserter/src/diagnostics.rs @@ -0,0 +1,176 @@ +use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; +use std::sync::LazyLock; +use std::time::{Duration, Instant}; + +static ENABLED: LazyLock = LazyLock::new(|| { + matches!( + std::env::var("DIAGNOSTICS") + .unwrap_or_default() + .to_ascii_lowercase() + .as_str(), + "1" | "true" | "yes" | "on" + ) +}); + +static GENERATED_SEEDS: AtomicU64 = AtomicU64::new(0); +static CSV_BYTES: AtomicU64 = AtomicU64::new(0); +static GENERATION_NS: AtomicU64 = AtomicU64::new(0); +static SEND_WAIT_NS: AtomicU64 = AtomicU64::new(0); +static CONNECTIONS: AtomicU64 = AtomicU64::new(0); +static CONNECTION_NS: AtomicU64 = AtomicU64::new(0); +static BATCH_RECEIVE_NS: AtomicU64 = AtomicU64::new(0); +static TRANSACTION_START_NS: AtomicU64 = AtomicU64::new(0); +static STAR_COPY_NS: AtomicU64 = AtomicU64::new(0); +static PLANET_COPY_NS: AtomicU64 = AtomicU64::new(0); +static COMMIT_NS: AtomicU64 = AtomicU64::new(0); +static BATCHES: AtomicU64 = AtomicU64::new(0); +static BATCHED_SEEDS: AtomicU64 = AtomicU64::new(0); +static MIN_BATCH_SIZE: AtomicU64 = AtomicU64::new(u64::MAX); +static MAX_BATCH_SIZE: AtomicU64 = AtomicU64::new(0); + +pub fn enabled() -> bool { + *ENABLED +} + +pub struct ReportOnDrop(Option); + +impl Drop for ReportOnDrop { + fn drop(&mut self) { + if let Some(start) = self.0 { + report(start.elapsed()); + } + } +} + +pub fn report_on_drop() -> ReportOnDrop { + ReportOnDrop(enabled().then(Instant::now)) +} + +fn add_duration(counter: &AtomicU64, duration: Duration) { + let nanos = duration.as_nanos().min(u64::MAX as u128) as u64; + counter.fetch_add(nanos, Relaxed); +} + +pub fn record_generation(duration: Duration, csv_bytes: usize) { + GENERATED_SEEDS.fetch_add(1, Relaxed); + CSV_BYTES.fetch_add(csv_bytes as u64, Relaxed); + add_duration(&GENERATION_NS, duration); +} + +pub fn record_send_wait(duration: Duration) { + add_duration(&SEND_WAIT_NS, duration); +} + +pub fn record_connection(duration: Duration) { + CONNECTIONS.fetch_add(1, Relaxed); + add_duration(&CONNECTION_NS, duration); +} + +pub fn record_batch_receive(duration: Duration) { + add_duration(&BATCH_RECEIVE_NS, duration); +} + +pub fn record_transaction_start(duration: Duration) { + add_duration(&TRANSACTION_START_NS, duration); +} + +pub fn record_star_copy(duration: Duration) { + add_duration(&STAR_COPY_NS, duration); +} + +pub fn record_planet_copy(duration: Duration) { + add_duration(&PLANET_COPY_NS, duration); +} + +pub fn record_commit(duration: Duration, batch_size: usize) { + let batch_size = batch_size as u64; + add_duration(&COMMIT_NS, duration); + BATCHES.fetch_add(1, Relaxed); + BATCHED_SEEDS.fetch_add(batch_size, Relaxed); + MIN_BATCH_SIZE.fetch_min(batch_size, Relaxed); + MAX_BATCH_SIZE.fetch_max(batch_size, Relaxed); +} + +fn seconds(counter: &AtomicU64) -> f64 { + counter.load(Relaxed) as f64 / 1_000_000_000.0 +} + +fn report(elapsed: Duration) { + let elapsed_seconds = elapsed.as_secs_f64(); + let generated_seeds = GENERATED_SEEDS.load(Relaxed); + let csv_bytes = CSV_BYTES.load(Relaxed); + let batches = BATCHES.load(Relaxed); + let batched_seeds = BATCHED_SEEDS.load(Relaxed); + let min_batch_size = if batches == 0 { + 0 + } else { + MIN_BATCH_SIZE.load(Relaxed) + }; + let average_batch_size = if batches == 0 { + 0.0 + } else { + batched_seeds as f64 / batches as f64 + }; + let average_generation_ms = if generated_seeds == 0 { + 0.0 + } else { + seconds(&GENERATION_NS) * 1000.0 / generated_seeds as f64 + }; + let csv_mib_per_second = if elapsed_seconds == 0.0 { + 0.0 + } else { + csv_bytes as f64 / 1_048_576.0 / elapsed_seconds + }; + + println!("diagnostics.elapsed_seconds={:.6}", elapsed_seconds); + println!("diagnostics.generated_seeds={}", generated_seeds); + println!("diagnostics.csv_bytes={}", csv_bytes); + println!("diagnostics.csv_mib_per_second={:.6}", csv_mib_per_second); + println!( + "diagnostics.generation_aggregate_seconds={:.6}", + seconds(&GENERATION_NS) + ); + println!( + "diagnostics.generation_average_ms={:.6}", + average_generation_ms + ); + println!( + "diagnostics.channel_send_wait_aggregate_seconds={:.6}", + seconds(&SEND_WAIT_NS) + ); + println!( + "diagnostics.writer_connections={}", + CONNECTIONS.load(Relaxed) + ); + println!( + "diagnostics.writer_connection_aggregate_seconds={:.6}", + seconds(&CONNECTION_NS) + ); + println!( + "diagnostics.batch_receive_aggregate_seconds={:.6}", + seconds(&BATCH_RECEIVE_NS) + ); + println!( + "diagnostics.transaction_start_aggregate_seconds={:.6}", + seconds(&TRANSACTION_START_NS) + ); + println!( + "diagnostics.star_copy_aggregate_seconds={:.6}", + seconds(&STAR_COPY_NS) + ); + println!( + "diagnostics.planet_copy_aggregate_seconds={:.6}", + seconds(&PLANET_COPY_NS) + ); + println!( + "diagnostics.commit_aggregate_seconds={:.6}", + seconds(&COMMIT_NS) + ); + println!("diagnostics.transactions={}", batches); + println!("diagnostics.batch_size_min={}", min_batch_size); + println!("diagnostics.batch_size_average={:.3}", average_batch_size); + println!( + "diagnostics.batch_size_max={}", + MAX_BATCH_SIZE.load(Relaxed) + ); +} diff --git a/inserter/src/main.rs b/inserter/src/main.rs index 83ade0e..f259fb9 100644 --- a/inserter/src/main.rs +++ b/inserter/src/main.rs @@ -1,5 +1,6 @@ mod algorithm; mod checkpoint; +mod diagnostics; mod generate_csv; mod logging; mod metrics; @@ -92,6 +93,7 @@ fn run() -> Result<()> { // capture start time for performance evaluation let start = Instant::now(); + let _diagnostics_report = diagnostics::report_on_drop(); if *BENCHMARK { log_info!("Benchmark mode enabled; database writes are disabled"); diff --git a/inserter/src/threads.rs b/inserter/src/threads.rs index 2bf3baf..f13281e 100644 --- a/inserter/src/threads.rs +++ b/inserter/src/threads.rs @@ -1,6 +1,7 @@ use std::ops::Range; use std::sync::atomic::Ordering::Relaxed; use std::time::Duration; +use std::time::Instant; use std::io::Write; use crossbeam_channel::{Receiver, RecvTimeoutError, Sender}; use postgres::{Client, NoTls}; @@ -10,9 +11,18 @@ use crate::{log_info, COMMITTED_SEEDS, COMMIT_COUNT, DB_STR, PROGRESS_WORKERS, R use crate::misc::{COPY_PLANET, COPY_STAR}; pub fn worker_thread(seeds: Range, send: Sender<(String, String)>, id: usize) -> Result<()> { + let diagnostics_enabled = crate::diagnostics::enabled(); for seed in seeds { + let generation_start = diagnostics_enabled.then(Instant::now); let entry = gen_formatted(seed, STAR_COUNT, REC_MULTIPLIER).expect("gen_formatted failed"); + if let Some(start) = generation_start { + crate::diagnostics::record_generation(start.elapsed(), entry.0.len() + entry.1.len()); + } + let send_start = diagnostics_enabled.then(Instant::now); send.send(entry)?; + if let Some(start) = send_start { + crate::diagnostics::record_send_wait(start.elapsed()); + } PROGRESS_WORKERS[id].fetch_add(1, Relaxed); } @@ -20,9 +30,15 @@ pub fn worker_thread(seeds: Range, send: Sender<(String, String)>, id: usiz Ok(()) } pub fn commit_thread(rec: Receiver<(String, String)>) -> Result<()> { + let diagnostics_enabled = crate::diagnostics::enabled(); + let connection_start = diagnostics_enabled.then(Instant::now); let mut client = Client::connect(&*DB_STR.as_str(), NoTls)?; + if let Some(start) = connection_start { + crate::diagnostics::record_connection(start.elapsed()); + } 'outer: loop { + let receive_start = diagnostics_enabled.then(Instant::now); let mut batch: Vec<(String, String)> = Vec::with_capacity(*COMMIT_COUNT); 'inner: for i in 0..*COMMIT_COUNT { match rec.recv_timeout(Duration::new(1, 0)) { @@ -31,9 +47,17 @@ pub fn commit_thread(rec: Receiver<(String, String)>) -> Result<()> { Err(RecvTimeoutError::Disconnected) => if i == 0 {break 'outer} else {break 'inner}, } } + if let Some(start) = receive_start { + crate::diagnostics::record_batch_receive(start.elapsed()); + } + let transaction_start = diagnostics_enabled.then(Instant::now); let mut txn = client.transaction()?; + if let Some(start) = transaction_start { + crate::diagnostics::record_transaction_start(start.elapsed()); + } + let star_copy_start = diagnostics_enabled.then(Instant::now); { let mut scpy = txn.copy_in(COPY_STAR)?; for (star, _) in &batch { @@ -41,7 +65,11 @@ pub fn commit_thread(rec: Receiver<(String, String)>) -> Result<()> { } scpy.finish()?; } + if let Some(start) = star_copy_start { + crate::diagnostics::record_star_copy(start.elapsed()); + } + let planet_copy_start = diagnostics_enabled.then(Instant::now); { let mut pcpy = txn.copy_in(COPY_PLANET)?; for (_, planet) in &batch { @@ -49,8 +77,15 @@ pub fn commit_thread(rec: Receiver<(String, String)>) -> Result<()> { } pcpy.finish()?; } + if let Some(start) = planet_copy_start { + crate::diagnostics::record_planet_copy(start.elapsed()); + } + let commit_start = diagnostics_enabled.then(Instant::now); txn.commit()?; + if let Some(start) = commit_start { + crate::diagnostics::record_commit(start.elapsed(), batch.len()); + } COMMITTED_SEEDS.fetch_add(batch.len() as i32, std::sync::atomic::Ordering::SeqCst); } log_info!("Writer thread terminated"); @@ -65,4 +100,4 @@ pub fn writer_sink(rec: Receiver<(String, String)>) -> Result<()> { } } Ok(()) -} \ No newline at end of file +}