Add opt-in pipeline runtime diagnostics

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.

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<bool> = 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<Instant>);
+
+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)
+    );
+}

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");

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,19 +11,34 @@ 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<i32>, 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);
 
     }
     log_info!("Worker thread {} finished sucessfully", id);
     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,26 +47,45 @@ 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 {
                 scpy.write_all(star.as_bytes()).expect("writing to scpy failed");
             }
             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 {
                 pcpy.write_all(planet.as_bytes()).expect("writing to pcpy failed");
             }
             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(())
-}
+}