From 1c0c13a247260585cc37a0b5784848e8f549109e Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 13 Jul 2026 13:05:54 +0000 Subject: [PATCH 1/3] Add opt-in pipeline diagnostics --- README.md | 1 + inserter/src/diagnostics.rs | 166 ++++++++++++++++++++++++++++++++++++ inserter/src/main.rs | 2 + inserter/src/threads.rs | 37 +++++++- 4 files changed, 205 insertions(+), 1 deletion(-) create mode 100644 inserter/src/diagnostics.rs diff --git a/README.md b/README.md index e7ecd37..79d9d43 100644 --- a/README.md +++ b/README.md @@ -59,6 +59,7 @@ The Rust generator is configured via environment variables: - `WRITER_THREADS`: Number of threads pushing data to Postgres. - `CHECKPOINT_FILE`: Path to the progress tracking file (default: `checkpoints.txt`). - `BENCHMARK=1`: Disables DB writes to test pure generation throughput. +- `DIAGNOSTICS=1`: Prints worker, channel, COPY, commit, payload-byte, and actual batch-size measurements after the run. Stage durations are aggregate thread time and can exceed wall-clock time when threads overlap. ## Sharp Edges diff --git a/inserter/src/diagnostics.rs b/inserter/src/diagnostics.rs new file mode 100644 index 0000000..cdd2274 --- /dev/null +++ b/inserter/src/diagnostics.rs @@ -0,0 +1,166 @@ +use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; +use std::sync::LazyLock; +use std::time::Duration; + +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 +} + +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 +} + +pub fn report(elapsed: Duration) { + if !enabled() { + return; + } + + 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..1b49dcb 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; @@ -196,6 +197,7 @@ fn run() -> Result<()> { } let elapsed = start.elapsed(); + diagnostics::report(elapsed); let per_second = (*END_SEED - *START_SEED) as f32 / elapsed.as_secs() as f32; println!("seeds/sec: {:?}", per_second); Ok(()) 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 +} From 22aa6596e3b0740760052654c1d2b20ec2507f2c Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 13 Jul 2026 13:23:09 +0000 Subject: [PATCH 2/3] Isolate diagnostics documentation --- README.md | 1 - docs/runtime-diagnostics.md | 13 +++++++++++++ 2 files changed, 13 insertions(+), 1 deletion(-) create mode 100644 docs/runtime-diagnostics.md diff --git a/README.md b/README.md index 79d9d43..e7ecd37 100644 --- a/README.md +++ b/README.md @@ -59,7 +59,6 @@ The Rust generator is configured via environment variables: - `WRITER_THREADS`: Number of threads pushing data to Postgres. - `CHECKPOINT_FILE`: Path to the progress tracking file (default: `checkpoints.txt`). - `BENCHMARK=1`: Disables DB writes to test pure generation throughput. -- `DIAGNOSTICS=1`: Prints worker, channel, COPY, commit, payload-byte, and actual batch-size measurements after the run. Stage durations are aggregate thread time and can exceed wall-clock time when threads overlap. ## Sharp Edges 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. From 274cd37ce947a1676ae3cefe0450efd63980f157 Mon Sep 17 00:00:00 2001 From: toti330 Date: Mon, 13 Jul 2026 13:30:55 +0000 Subject: [PATCH 3/3] Isolate diagnostics timing hook --- inserter/src/diagnostics.rs | 22 ++++++++++++++++------ inserter/src/main.rs | 2 +- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/inserter/src/diagnostics.rs b/inserter/src/diagnostics.rs index cdd2274..424231c 100644 --- a/inserter/src/diagnostics.rs +++ b/inserter/src/diagnostics.rs @@ -1,6 +1,6 @@ use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; use std::sync::LazyLock; -use std::time::Duration; +use std::time::{Duration, Instant}; static ENABLED: LazyLock = LazyLock::new(|| { matches!( @@ -32,6 +32,20 @@ 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); @@ -81,11 +95,7 @@ fn seconds(counter: &AtomicU64) -> f64 { counter.load(Relaxed) as f64 / 1_000_000_000.0 } -pub fn report(elapsed: Duration) { - if !enabled() { - return; - } - +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); diff --git a/inserter/src/main.rs b/inserter/src/main.rs index 1b49dcb..f259fb9 100644 --- a/inserter/src/main.rs +++ b/inserter/src/main.rs @@ -93,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"); @@ -197,7 +198,6 @@ fn run() -> Result<()> { } let elapsed = start.elapsed(); - diagnostics::report(elapsed); let per_second = (*END_SEED - *START_SEED) as f32 / elapsed.as_secs() as f32; println!("seeds/sec: {:?}", per_second); Ok(())