From 60b19d0770fcb86afe73b7b5176413370b9f1dae Mon Sep 17 00:00:00 2001 From: toti330 Date: Mon, 13 Jul 2026 13:48:06 +0000 Subject: [PATCH 1/4] Add timed progress logging --- docs/progress-logging.md | 16 ++ inserter/src/main.rs | 6 + inserter/src/progress.rs | 392 +++++++++++++++++++++++++++++++++++++++ inserter/src/threads.rs | 37 +++- 4 files changed, 449 insertions(+), 2 deletions(-) create mode 100644 docs/progress-logging.md create mode 100644 inserter/src/progress.rs diff --git a/docs/progress-logging.md b/docs/progress-logging.md new file mode 100644 index 0000000..e9068f9 --- /dev/null +++ b/docs/progress-logging.md @@ -0,0 +1,16 @@ +# Progress Logging + +Progress logging is enabled by default and runs in a dedicated sleeping thread. Set `PROGRESS_LOG=0` to disable the logger and all additional stage timers. Set `PROGRESS_LOG_INTERVAL_SECS` to a positive integer number of seconds; the default is `30`. + +The logger prints `generation started` before workers launch, then emits time-based reports containing: + +- Generated, committed, and overall progress as exact counts and percentages. +- Interval and whole-run generation and commit rates in seeds/second. +- Elapsed time and channel occupancy. +- Aggregate worker-thread, database connection, active database, and writer-wait durations. While workers remain active, worker time is estimated as elapsed time per active worker; the final value is measured. +- Active and completed worker counts, writers currently receiving or executing database work, database batch count, dominant aggregate thread-time stage, and a queue-based likely bottleneck. +- Linux process RSS, peak RSS, bytes read, and bytes written from `/proc`; unsupported platforms report `n/a`. + +Aggregate stage durations are thread time, so overlapping workers or writers can produce totals larger than wall-clock elapsed time. `likely_bottleneck` is a queue-based diagnostic: at least `80%` occupancy indicates database pressure, at most `20%` indicates generation pressure, and an empty generation phase with outstanding commits indicates database drain. + +Instrumentation does not add a timer or branch to each generation pass. Worker timing occurs once per worker lifetime, database timing occurs once per batch, and process/resource collection runs only in the monitor thread at the configured interval. diff --git a/inserter/src/main.rs b/inserter/src/main.rs index 83ade0e..519ec8b 100644 --- a/inserter/src/main.rs +++ b/inserter/src/main.rs @@ -4,6 +4,7 @@ mod generate_csv; mod logging; mod metrics; mod misc; +mod progress; mod threads; use crate::checkpoint::{load_workloads, write_checkpoints}; @@ -109,6 +110,8 @@ fn run() -> Result<()> { log_info!("Loading workloads..."); let workloads = load_workloads().context("failed to load workloads")?; let (entry_sender, entry_reciever): (Sender<(String, String)>, Receiver<(String, String)>) = bounded(*CHANNEL_SIZE); + let planned_seeds = workloads.iter().map(|workload| workload.len() as u64).sum(); + let progress_logger = progress::ProgressLogger::start(planned_seeds, entry_reciever.clone())?; let mut work_handles = vec![]; let mut commit_handles = vec![]; @@ -194,6 +197,9 @@ fn run() -> Result<()> { .map_err(|_| anyhow!("writer thread panicked"))? .context("writer thread failed")?; } + if let Some(logger) = progress_logger { + logger.finish()?; + } let elapsed = start.elapsed(); let per_second = (*END_SEED - *START_SEED) as f32 / elapsed.as_secs() as f32; diff --git a/inserter/src/progress.rs b/inserter/src/progress.rs new file mode 100644 index 0000000..ffc6ac1 --- /dev/null +++ b/inserter/src/progress.rs @@ -0,0 +1,392 @@ +use crate::{ + BENCHMARK, CHANNEL_SIZE, COMMITTED_SEEDS, COMMIT_COUNT, WORKER_THREADS, WRITER_THREADS, +}; +use anyhow::{Context, Result}; +use crossbeam_channel::Receiver; +use std::fs; +use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; +use std::sync::{Arc, Condvar, LazyLock, Mutex}; +use std::thread::{self, JoinHandle}; +use std::time::{Duration, Instant}; + +static ENABLED: LazyLock = LazyLock::new(|| { + !matches!( + std::env::var("PROGRESS_LOG") + .unwrap_or_else(|_| "1".to_string()) + .to_ascii_lowercase() + .as_str(), + "0" | "false" | "no" | "off" + ) +}); + +static WORKER_TIME_NS: AtomicU64 = AtomicU64::new(0); +static WORKERS_FINISHED: AtomicU64 = AtomicU64::new(0); +static ACTIVE_WORKERS: AtomicU64 = AtomicU64::new(0); +static DB_CONNECTION_NS: AtomicU64 = AtomicU64::new(0); +static DB_TIME_NS: AtomicU64 = AtomicU64::new(0); +static DB_WAIT_NS: AtomicU64 = AtomicU64::new(0); +static DB_BATCHES: AtomicU64 = AtomicU64::new(0); +static WRITERS_WAITING: AtomicU64 = AtomicU64::new(0); +static WRITERS_DB_ACTIVE: AtomicU64 = AtomicU64::new(0); + +pub fn enabled() -> bool { + *ENABLED +} + +fn add_duration(counter: &AtomicU64, duration: Duration) { + counter.fetch_add(duration.as_nanos().min(u64::MAX as u128) as u64, Relaxed); +} + +pub fn record_worker(duration: Duration) { + add_duration(&WORKER_TIME_NS, duration); + WORKERS_FINISHED.fetch_add(1, Relaxed); + ACTIVE_WORKERS.fetch_sub(1, Relaxed); +} + +pub fn worker_started() { + ACTIVE_WORKERS.fetch_add(1, Relaxed); +} + +pub fn record_db_connection(duration: Duration) { + add_duration(&DB_CONNECTION_NS, duration); +} + +pub fn record_db_wait(duration: Duration) { + add_duration(&DB_WAIT_NS, duration); + WRITERS_WAITING.fetch_sub(1, Relaxed); +} + +pub fn db_wait_started() { + WRITERS_WAITING.fetch_add(1, Relaxed); +} + +pub fn record_db_batch(duration: Duration) { + add_duration(&DB_TIME_NS, duration); + DB_BATCHES.fetch_add(1, Relaxed); + WRITERS_DB_ACTIVE.fetch_sub(1, Relaxed); +} + +pub fn db_batch_started() { + WRITERS_DB_ACTIVE.fetch_add(1, Relaxed); +} + +struct Snapshot { + at: Instant, + generated: u64, + committed: u64, +} + +struct ProcessStats { + rss_kib: Option, + peak_rss_kib: Option, + read_bytes: Option, + write_bytes: Option, +} + +pub struct ProgressLogger { + stop: Arc<(Mutex, Condvar)>, + handle: Option>, +} + +impl ProgressLogger { + pub fn start(planned_seeds: u64, receiver: Receiver<(String, String)>) -> Result> { + if !enabled() { + return Ok(None); + } + + let interval_seconds = std::env::var("PROGRESS_LOG_INTERVAL_SECS") + .unwrap_or_else(|_| "30".to_string()) + .parse::() + .context("PROGRESS_LOG_INTERVAL_SECS must be a positive integer")?; + if interval_seconds == 0 { + anyhow::bail!("PROGRESS_LOG_INTERVAL_SECS must be greater than 0"); + } + let interval = Duration::from_secs(interval_seconds); + eprintln!( + "[progress] generation started: seeds={} workers={} writers={} commit_count={} interval={}s benchmark={}", + planned_seeds, + *WORKER_THREADS, + *WRITER_THREADS, + *COMMIT_COUNT, + interval_seconds, + *BENCHMARK + ); + + let stop = Arc::new((Mutex::new(false), Condvar::new())); + let thread_stop = Arc::clone(&stop); + let handle = thread::Builder::new() + .name("progress_logger".to_string()) + .spawn(move || monitor(planned_seeds, receiver, interval, thread_stop)) + .context("failed to spawn progress logger")?; + + Ok(Some(Self { + stop, + handle: Some(handle), + })) + } + + pub fn finish(mut self) -> Result<()> { + self.signal_stop(); + if let Some(handle) = self.handle.take() { + handle + .join() + .map_err(|_| anyhow::anyhow!("progress logger panicked"))?; + } + Ok(()) + } + + fn signal_stop(&self) { + let (lock, condition) = &*self.stop; + if let Ok(mut stopped) = lock.lock() { + *stopped = true; + condition.notify_one(); + } + } +} + +impl Drop for ProgressLogger { + fn drop(&mut self) { + self.signal_stop(); + } +} + +fn monitor( + planned_seeds: u64, + receiver: Receiver<(String, String)>, + interval: Duration, + stop: Arc<(Mutex, Condvar)>, +) { + let started = Instant::now(); + let mut previous = Snapshot { + at: started, + generated: generated_seeds(), + committed: committed_seeds(), + }; + + loop { + let (lock, condition) = &*stop; + let stopped = match lock.lock() { + Ok(stopped) => stopped, + Err(_) => return, + }; + let (stopped, timeout) = match condition.wait_timeout(stopped, interval) { + Ok(result) => result, + Err(_) => return, + }; + if *stopped { + break; + } + if timeout.timed_out() { + previous = report(started, previous, planned_seeds, receiver.len(), false); + } + } + + report(started, previous, planned_seeds, receiver.len(), true); +} + +fn report( + started: Instant, + previous: Snapshot, + planned_seeds: u64, + queue: usize, + complete: bool, +) -> Snapshot { + let now = Instant::now(); + let generated = generated_seeds(); + let committed = committed_seeds(); + let elapsed = now.duration_since(started); + let period = now + .duration_since(previous.at) + .as_secs_f64() + .max(f64::EPSILON); + let generated_rate = generated.saturating_sub(previous.generated) as f64 / period; + let committed_rate = committed.saturating_sub(previous.committed) as f64 / period; + let average_generated_rate = generated as f64 / elapsed.as_secs_f64().max(f64::EPSILON); + let average_committed_rate = committed as f64 / elapsed.as_secs_f64().max(f64::EPSILON); + let generated_percent = percent(generated, planned_seeds); + let committed_percent = percent(committed, planned_seeds); + let overall = if *BENCHMARK { generated } else { committed }; + let status = if complete { "complete" } else { "running" }; + let generation_time = estimated_generation_time(elapsed); + + eprintln!( + "[progress] status={} elapsed={} progress={}/{} ({:.2}%) generated={}/{} ({:.2}%) committed={}/{} ({:.2}%) queue={}/{}", + status, + format_duration(elapsed), + overall, + planned_seeds, + percent(overall, planned_seeds), + generated, + planned_seeds, + generated_percent, + committed, + planned_seeds, + committed_percent, + queue, + *CHANNEL_SIZE + ); + eprintln!( + "[progress] rates: generated_interval={:.3}/s generated_average={:.3}/s committed_interval={:.3}/s committed_average={:.3}/s", + generated_rate, + average_generated_rate, + committed_rate, + average_committed_rate + ); + eprintln!( + "[progress] timing: generation_worker={} db_connection={} db_active={} writer_wait={} workers_active={} workers_finished={}/{} writers_waiting={} writers_db_active={} db_batches={} dominant_thread_time={} likely_bottleneck={}", + format_nanos(generation_time), + format_nanos(DB_CONNECTION_NS.load(Relaxed)), + format_nanos(DB_TIME_NS.load(Relaxed)), + format_nanos(DB_WAIT_NS.load(Relaxed)), + ACTIVE_WORKERS.load(Relaxed), + WORKERS_FINISHED.load(Relaxed), + *WORKER_THREADS, + WRITERS_WAITING.load(Relaxed), + WRITERS_DB_ACTIVE.load(Relaxed), + DB_BATCHES.load(Relaxed), + dominant_stage(generation_time), + likely_bottleneck(generated, committed, planned_seeds, queue) + ); + print_process_stats(read_process_stats()); + + Snapshot { + at: now, + generated, + committed, + } +} + +fn generated_seeds() -> u64 { + crate::PROGRESS_WORKERS + .iter() + .take(*WORKER_THREADS as usize) + .map(|progress| progress.load(Relaxed).max(0) as u64) + .sum() +} + +fn committed_seeds() -> u64 { + COMMITTED_SEEDS.load(Relaxed).max(0) as u64 +} + +fn percent(value: u64, total: u64) -> f64 { + if total == 0 { + 100.0 + } else { + value.min(total) as f64 / total as f64 * 100.0 + } +} + +fn estimated_generation_time(elapsed: Duration) -> u64 { + let active_time = elapsed + .as_nanos() + .saturating_mul(ACTIVE_WORKERS.load(Relaxed) as u128) + .min(u64::MAX as u128) as u64; + WORKER_TIME_NS.load(Relaxed).saturating_add(active_time) +} + +fn dominant_stage(generation_time: u64) -> &'static str { + let stages = [ + ("generation", generation_time), + ("database", DB_TIME_NS.load(Relaxed)), + ("writer_wait", DB_WAIT_NS.load(Relaxed)), + ]; + stages + .into_iter() + .max_by_key(|(_, duration)| *duration) + .filter(|(_, duration)| *duration > 0) + .map(|(stage, _)| stage) + .unwrap_or("warming_up") +} + +fn likely_bottleneck( + generated: u64, + committed: u64, + planned_seeds: u64, + queue: usize, +) -> &'static str { + if *BENCHMARK { + return "generation"; + } + if generated >= planned_seeds && committed < planned_seeds { + return "database_drain"; + } + if *CHANNEL_SIZE > 0 && queue.saturating_mul(100) >= *CHANNEL_SIZE * 80 { + return "database"; + } + if *CHANNEL_SIZE == 0 || queue.saturating_mul(100) <= *CHANNEL_SIZE * 20 { + return "generation"; + } + "balanced_or_inconclusive" +} + +fn format_nanos(nanos: u64) -> String { + format_duration(Duration::from_nanos(nanos)) +} + +fn format_duration(duration: Duration) -> String { + let seconds = duration.as_secs_f64(); + if seconds >= 60.0 { + format!("{}m{:.3}s", duration.as_secs() / 60, seconds % 60.0) + } else { + format!("{:.3}s", seconds) + } +} + +fn read_process_stats() -> ProcessStats { + let status = fs::read_to_string("/proc/self/status").unwrap_or_default(); + let io = fs::read_to_string("/proc/self/io").unwrap_or_default(); + ProcessStats { + rss_kib: parse_value(&status, "VmRSS:"), + peak_rss_kib: parse_value(&status, "VmHWM:"), + read_bytes: parse_value(&io, "read_bytes:"), + write_bytes: parse_value(&io, "write_bytes:"), + } +} + +fn parse_value(contents: &str, key: &str) -> Option { + contents + .lines() + .find(|line| line.starts_with(key)) + .and_then(|line| line.split_whitespace().nth(1)) + .and_then(|value| value.parse().ok()) +} + +fn print_process_stats(stats: ProcessStats) { + eprintln!( + "[progress] resources: rss_mib={} peak_rss_mib={} process_read_mib={} process_write_mib={}", + format_mib(stats.rss_kib.map(|value| value * 1024)), + format_mib(stats.peak_rss_kib.map(|value| value * 1024)), + format_mib(stats.read_bytes), + format_mib(stats.write_bytes) + ); +} + +fn format_mib(bytes: Option) -> String { + bytes + .map(|value| format!("{:.3}", value as f64 / 1_048_576.0)) + .unwrap_or_else(|| "n/a".to_string()) +} + +#[cfg(test)] +mod tests { + use super::{format_duration, parse_value, percent}; + use std::time::Duration; + + #[test] + fn formats_short_and_long_durations() { + assert_eq!(format_duration(Duration::from_millis(1250)), "1.250s"); + assert_eq!(format_duration(Duration::from_millis(61_250)), "1m1.250s"); + } + + #[test] + fn percentage_is_bounded() { + assert_eq!(percent(50, 100), 50.0); + assert_eq!(percent(200, 100), 100.0); + } + + #[test] + fn parses_proc_values() { + assert_eq!(parse_value("VmRSS:\t2048 kB\n", "VmRSS:"), Some(2048)); + assert_eq!(parse_value("read_bytes: 4096\n", "read_bytes:"), Some(4096)); + } +} diff --git a/inserter/src/threads.rs b/inserter/src/threads.rs index 2bf3baf..37ece5f 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,28 +11,57 @@ 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 progress_enabled = crate::progress::enabled(); + let worker_started = progress_enabled.then(Instant::now); + if progress_enabled { + crate::progress::worker_started(); + } for seed in seeds { let entry = gen_formatted(seed, STAR_COUNT, REC_MULTIPLIER).expect("gen_formatted failed"); send.send(entry)?; PROGRESS_WORKERS[id].fetch_add(1, Relaxed); } + if let Some(started) = worker_started { + crate::progress::record_worker(started.elapsed()); + } log_info!("Worker thread {} finished sucessfully", id); Ok(()) } pub fn commit_thread(rec: Receiver<(String, String)>) -> Result<()> { + let progress_enabled = crate::progress::enabled(); + let connection_started = progress_enabled.then(Instant::now); let mut client = Client::connect(&*DB_STR.as_str(), NoTls)?; + if let Some(started) = connection_started { + crate::progress::record_db_connection(started.elapsed()); + } 'outer: loop { + let wait_started = progress_enabled.then(Instant::now); + if progress_enabled { + crate::progress::db_wait_started(); + } 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)) { Ok(msg) => batch.push(msg), Err(RecvTimeoutError::Timeout) => panic!("commit_thread: recv_timeout reached - channel stall detected (>1s lull)"), - Err(RecvTimeoutError::Disconnected) => if i == 0 {break 'outer} else {break 'inner}, + Err(RecvTimeoutError::Disconnected) => if i == 0 { + if let Some(started) = wait_started { + crate::progress::record_db_wait(started.elapsed()); + } + break 'outer + } else {break 'inner}, } } + if let Some(started) = wait_started { + crate::progress::record_db_wait(started.elapsed()); + } + let db_started = progress_enabled.then(Instant::now); + if progress_enabled { + crate::progress::db_batch_started(); + } let mut txn = client.transaction()?; { @@ -51,6 +81,9 @@ pub fn commit_thread(rec: Receiver<(String, String)>) -> Result<()> { } txn.commit()?; + if let Some(started) = db_started { + crate::progress::record_db_batch(started.elapsed()); + } COMMITTED_SEEDS.fetch_add(batch.len() as i32, std::sync::atomic::Ordering::SeqCst); } log_info!("Writer thread terminated"); @@ -65,4 +98,4 @@ pub fn writer_sink(rec: Receiver<(String, String)>) -> Result<()> { } } Ok(()) -} \ No newline at end of file +} From 6f92afc940d59a2f27acf8f5a39e2754ffdf0adb Mon Sep 17 00:00:00 2001 From: toti330 Date: Mon, 13 Jul 2026 13:51:16 +0000 Subject: [PATCH 2/4] Isolate progress timing hooks --- docs/progress-logging.md | 6 +-- inserter/src/main.rs | 5 ++- inserter/src/progress.rs | 87 ++++++++++++++++++---------------------- inserter/src/threads.rs | 35 +--------------- 4 files changed, 45 insertions(+), 88 deletions(-) diff --git a/docs/progress-logging.md b/docs/progress-logging.md index e9068f9..d477870 100644 --- a/docs/progress-logging.md +++ b/docs/progress-logging.md @@ -7,10 +7,10 @@ The logger prints `generation started` before workers launch, then emits time-ba - Generated, committed, and overall progress as exact counts and percentages. - Interval and whole-run generation and commit rates in seeds/second. - Elapsed time and channel occupancy. -- Aggregate worker-thread, database connection, active database, and writer-wait durations. While workers remain active, worker time is estimated as elapsed time per active worker; the final value is measured. -- Active and completed worker counts, writers currently receiving or executing database work, database batch count, dominant aggregate thread-time stage, and a queue-based likely bottleneck. +- Aggregate worker and writer-pipeline durations. Writer-pipeline time includes connection, channel receive, COPY, and commit time; use the independent runtime-diagnostics branch when those database stages must be separated. While threads remain active, time is estimated as elapsed time per active thread; final values are measured. +- Active and completed worker/writer counts, dominant aggregate thread-time stage, and a queue-based likely bottleneck. - Linux process RSS, peak RSS, bytes read, and bytes written from `/proc`; unsupported platforms report `n/a`. Aggregate stage durations are thread time, so overlapping workers or writers can produce totals larger than wall-clock elapsed time. `likely_bottleneck` is a queue-based diagnostic: at least `80%` occupancy indicates database pressure, at most `20%` indicates generation pressure, and an empty generation phase with outstanding commits indicates database drain. -Instrumentation does not add a timer or branch to each generation pass. Worker timing occurs once per worker lifetime, database timing occurs once per batch, and process/resource collection runs only in the monitor thread at the configured interval. +Instrumentation does not add a timer or branch to each generation pass. Timing occurs once per worker or writer lifetime, and process/resource collection runs only in the monitor thread at the configured interval. diff --git a/inserter/src/main.rs b/inserter/src/main.rs index 519ec8b..321c932 100644 --- a/inserter/src/main.rs +++ b/inserter/src/main.rs @@ -9,6 +9,7 @@ mod threads; use crate::checkpoint::{load_workloads, write_checkpoints}; use crate::misc::check_db_connection; +use crate::progress::{commit_thread, worker_thread}; use crate::{metrics::write_metrics, threads::*}; use anyhow::{anyhow, Context, Result}; use crossbeam_channel::{bounded, Receiver, Sender}; @@ -110,11 +111,11 @@ fn run() -> Result<()> { log_info!("Loading workloads..."); let workloads = load_workloads().context("failed to load workloads")?; let (entry_sender, entry_reciever): (Sender<(String, String)>, Receiver<(String, String)>) = bounded(*CHANNEL_SIZE); - let planned_seeds = workloads.iter().map(|workload| workload.len() as u64).sum(); - let progress_logger = progress::ProgressLogger::start(planned_seeds, entry_reciever.clone())?; let mut work_handles = vec![]; let mut commit_handles = vec![]; + let planned_seeds = workloads.iter().map(|workload| workload.len() as u64).sum(); + let progress_logger = progress::ProgressLogger::start(planned_seeds, entry_reciever.clone())?; log_info!("Starting worker threads..."); // Launch worker threads for (id, work) in workloads.iter().enumerate() { diff --git a/inserter/src/progress.rs b/inserter/src/progress.rs index ffc6ac1..c334d91 100644 --- a/inserter/src/progress.rs +++ b/inserter/src/progress.rs @@ -2,8 +2,9 @@ use crate::{ BENCHMARK, CHANNEL_SIZE, COMMITTED_SEEDS, COMMIT_COUNT, WORKER_THREADS, WRITER_THREADS, }; use anyhow::{Context, Result}; -use crossbeam_channel::Receiver; +use crossbeam_channel::{Receiver, Sender}; use std::fs; +use std::ops::Range; use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; use std::sync::{Arc, Condvar, LazyLock, Mutex}; use std::thread::{self, JoinHandle}; @@ -22,12 +23,9 @@ static ENABLED: LazyLock = LazyLock::new(|| { static WORKER_TIME_NS: AtomicU64 = AtomicU64::new(0); static WORKERS_FINISHED: AtomicU64 = AtomicU64::new(0); static ACTIVE_WORKERS: AtomicU64 = AtomicU64::new(0); -static DB_CONNECTION_NS: AtomicU64 = AtomicU64::new(0); -static DB_TIME_NS: AtomicU64 = AtomicU64::new(0); -static DB_WAIT_NS: AtomicU64 = AtomicU64::new(0); -static DB_BATCHES: AtomicU64 = AtomicU64::new(0); -static WRITERS_WAITING: AtomicU64 = AtomicU64::new(0); -static WRITERS_DB_ACTIVE: AtomicU64 = AtomicU64::new(0); +static WRITER_TIME_NS: AtomicU64 = AtomicU64::new(0); +static WRITERS_FINISHED: AtomicU64 = AtomicU64::new(0); +static ACTIVE_WRITERS: AtomicU64 = AtomicU64::new(0); pub fn enabled() -> bool { *ENABLED @@ -37,37 +35,30 @@ fn add_duration(counter: &AtomicU64, duration: Duration) { counter.fetch_add(duration.as_nanos().min(u64::MAX as u128) as u64, Relaxed); } -pub fn record_worker(duration: Duration) { - add_duration(&WORKER_TIME_NS, duration); +pub fn worker_thread(seeds: Range, sender: Sender<(String, String)>, id: usize) -> Result<()> { + if !enabled() { + return crate::threads::worker_thread(seeds, sender, id); + } + ACTIVE_WORKERS.fetch_add(1, Relaxed); + let started = Instant::now(); + let result = crate::threads::worker_thread(seeds, sender, id); + add_duration(&WORKER_TIME_NS, started.elapsed()); WORKERS_FINISHED.fetch_add(1, Relaxed); ACTIVE_WORKERS.fetch_sub(1, Relaxed); + result } -pub fn worker_started() { - ACTIVE_WORKERS.fetch_add(1, Relaxed); -} - -pub fn record_db_connection(duration: Duration) { - add_duration(&DB_CONNECTION_NS, duration); -} - -pub fn record_db_wait(duration: Duration) { - add_duration(&DB_WAIT_NS, duration); - WRITERS_WAITING.fetch_sub(1, Relaxed); -} - -pub fn db_wait_started() { - WRITERS_WAITING.fetch_add(1, Relaxed); -} - -pub fn record_db_batch(duration: Duration) { - add_duration(&DB_TIME_NS, duration); - DB_BATCHES.fetch_add(1, Relaxed); - WRITERS_DB_ACTIVE.fetch_sub(1, Relaxed); -} - -pub fn db_batch_started() { - WRITERS_DB_ACTIVE.fetch_add(1, Relaxed); +pub fn commit_thread(receiver: Receiver<(String, String)>) -> Result<()> { + if !enabled() { + return crate::threads::commit_thread(receiver); + } + ACTIVE_WRITERS.fetch_add(1, Relaxed); + let started = Instant::now(); + let result = crate::threads::commit_thread(receiver); + add_duration(&WRITER_TIME_NS, started.elapsed()); + WRITERS_FINISHED.fetch_add(1, Relaxed); + ACTIVE_WRITERS.fetch_sub(1, Relaxed); + result } struct Snapshot { @@ -207,7 +198,8 @@ fn report( let committed_percent = percent(committed, planned_seeds); let overall = if *BENCHMARK { generated } else { committed }; let status = if complete { "complete" } else { "running" }; - let generation_time = estimated_generation_time(elapsed); + let generation_time = estimated_thread_time(&WORKER_TIME_NS, &ACTIVE_WORKERS, elapsed); + let writer_time = estimated_thread_time(&WRITER_TIME_NS, &ACTIVE_WRITERS, elapsed); eprintln!( "[progress] status={} elapsed={} progress={}/{} ({:.2}%) generated={}/{} ({:.2}%) committed={}/{} ({:.2}%) queue={}/{}", @@ -233,18 +225,16 @@ fn report( average_committed_rate ); eprintln!( - "[progress] timing: generation_worker={} db_connection={} db_active={} writer_wait={} workers_active={} workers_finished={}/{} writers_waiting={} writers_db_active={} db_batches={} dominant_thread_time={} likely_bottleneck={}", + "[progress] timing: generation_worker={} writer_pipeline={} workers_active={} workers_finished={}/{} writers_active={} writers_finished={}/{} dominant_thread_time={} likely_bottleneck={}", format_nanos(generation_time), - format_nanos(DB_CONNECTION_NS.load(Relaxed)), - format_nanos(DB_TIME_NS.load(Relaxed)), - format_nanos(DB_WAIT_NS.load(Relaxed)), + format_nanos(writer_time), ACTIVE_WORKERS.load(Relaxed), WORKERS_FINISHED.load(Relaxed), *WORKER_THREADS, - WRITERS_WAITING.load(Relaxed), - WRITERS_DB_ACTIVE.load(Relaxed), - DB_BATCHES.load(Relaxed), - dominant_stage(generation_time), + ACTIVE_WRITERS.load(Relaxed), + WRITERS_FINISHED.load(Relaxed), + if *BENCHMARK { 0 } else { *WRITER_THREADS as u64 }, + dominant_stage(generation_time, writer_time), likely_bottleneck(generated, committed, planned_seeds, queue) ); print_process_stats(read_process_stats()); @@ -276,19 +266,18 @@ fn percent(value: u64, total: u64) -> f64 { } } -fn estimated_generation_time(elapsed: Duration) -> u64 { +fn estimated_thread_time(completed: &AtomicU64, active: &AtomicU64, elapsed: Duration) -> u64 { let active_time = elapsed .as_nanos() - .saturating_mul(ACTIVE_WORKERS.load(Relaxed) as u128) + .saturating_mul(active.load(Relaxed) as u128) .min(u64::MAX as u128) as u64; - WORKER_TIME_NS.load(Relaxed).saturating_add(active_time) + completed.load(Relaxed).saturating_add(active_time) } -fn dominant_stage(generation_time: u64) -> &'static str { +fn dominant_stage(generation_time: u64, writer_time: u64) -> &'static str { let stages = [ ("generation", generation_time), - ("database", DB_TIME_NS.load(Relaxed)), - ("writer_wait", DB_WAIT_NS.load(Relaxed)), + ("writer_pipeline", writer_time), ]; stages .into_iter() diff --git a/inserter/src/threads.rs b/inserter/src/threads.rs index 37ece5f..5938315 100644 --- a/inserter/src/threads.rs +++ b/inserter/src/threads.rs @@ -1,7 +1,6 @@ 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}; @@ -11,57 +10,28 @@ 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 progress_enabled = crate::progress::enabled(); - let worker_started = progress_enabled.then(Instant::now); - if progress_enabled { - crate::progress::worker_started(); - } for seed in seeds { let entry = gen_formatted(seed, STAR_COUNT, REC_MULTIPLIER).expect("gen_formatted failed"); send.send(entry)?; PROGRESS_WORKERS[id].fetch_add(1, Relaxed); } - if let Some(started) = worker_started { - crate::progress::record_worker(started.elapsed()); - } log_info!("Worker thread {} finished sucessfully", id); Ok(()) } pub fn commit_thread(rec: Receiver<(String, String)>) -> Result<()> { - let progress_enabled = crate::progress::enabled(); - let connection_started = progress_enabled.then(Instant::now); let mut client = Client::connect(&*DB_STR.as_str(), NoTls)?; - if let Some(started) = connection_started { - crate::progress::record_db_connection(started.elapsed()); - } 'outer: loop { - let wait_started = progress_enabled.then(Instant::now); - if progress_enabled { - crate::progress::db_wait_started(); - } 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)) { Ok(msg) => batch.push(msg), Err(RecvTimeoutError::Timeout) => panic!("commit_thread: recv_timeout reached - channel stall detected (>1s lull)"), - Err(RecvTimeoutError::Disconnected) => if i == 0 { - if let Some(started) = wait_started { - crate::progress::record_db_wait(started.elapsed()); - } - break 'outer - } else {break 'inner}, + Err(RecvTimeoutError::Disconnected) => if i == 0 {break 'outer} else {break 'inner}, } } - if let Some(started) = wait_started { - crate::progress::record_db_wait(started.elapsed()); - } - let db_started = progress_enabled.then(Instant::now); - if progress_enabled { - crate::progress::db_batch_started(); - } let mut txn = client.transaction()?; { @@ -81,9 +51,6 @@ pub fn commit_thread(rec: Receiver<(String, String)>) -> Result<()> { } txn.commit()?; - if let Some(started) = db_started { - crate::progress::record_db_batch(started.elapsed()); - } COMMITTED_SEEDS.fetch_add(batch.len() as i32, std::sync::atomic::Ordering::SeqCst); } log_info!("Writer thread terminated"); From c70e7c6361a4e317247631fec0a970ff8d521901 Mon Sep 17 00:00:00 2001 From: toti330 Date: Mon, 13 Jul 2026 13:52:15 +0000 Subject: [PATCH 3/4] Avoid tuning setup overlap --- inserter/src/main.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/inserter/src/main.rs b/inserter/src/main.rs index 321c932..41a4bc9 100644 --- a/inserter/src/main.rs +++ b/inserter/src/main.rs @@ -114,8 +114,6 @@ fn run() -> Result<()> { let mut work_handles = vec![]; let mut commit_handles = vec![]; - let planned_seeds = workloads.iter().map(|workload| workload.len() as u64).sum(); - let progress_logger = progress::ProgressLogger::start(planned_seeds, entry_reciever.clone())?; log_info!("Starting worker threads..."); // Launch worker threads for (id, work) in workloads.iter().enumerate() { @@ -148,6 +146,8 @@ fn run() -> Result<()> { } } + let planned_seeds = workloads.iter().map(|workload| workload.len() as u64).sum(); + let progress_logger = progress::ProgressLogger::start(planned_seeds, entry_reciever.clone())?; log_info!("Starting main thread loop"); // Main thread takes checkpoints and displays metrics to the terminal if *TUI { From 01cc0fbd256eb8f2e84e9fb6a3c6907b20a3664e Mon Sep 17 00:00:00 2001 From: toti330 Date: Mon, 13 Jul 2026 13:53:18 +0000 Subject: [PATCH 4/4] Avoid tuning module overlap --- inserter/src/main.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/inserter/src/main.rs b/inserter/src/main.rs index 41a4bc9..ed7e31f 100644 --- a/inserter/src/main.rs +++ b/inserter/src/main.rs @@ -4,8 +4,8 @@ mod generate_csv; mod logging; mod metrics; mod misc; -mod progress; mod threads; +mod progress; use crate::checkpoint::{load_workloads, write_checkpoints}; use crate::misc::check_db_connection;