From 067e59fadfa1c89be21b4e3dcc9beb44071fc6df Mon Sep 17 00:00:00 2001 From: Matt Currier Date: Fri, 6 Mar 2026 16:28:08 -0500 Subject: [PATCH 1/4] fix: Fix streaming output timestamps showing same second for all messages Pass send timestamp to MessageLatencyRecord instead of capturing it when the record is created. This makes the timestamp represent when the message was sent, so gaps between timestamps now match the actual latency. Changes: - MessageLatencyRecord::new() and new_combined() now require send_timestamp_ns parameter instead of capturing SystemTime::now() internally - Added MessageLatencyRecord::current_timestamp_ns() helper for call sites that need to capture the current wall-clock time - Updated all streaming record creation sites in benchmark.rs and benchmark_blocking.rs to capture and pass send timestamps Cherry-picked from container-to-container-ipc branch (6cb8f9b). Host-container-specific changes excluded (host_container.rs not present). AI-assisted-by: Claude claude-4.6-opus-high-thinking (Anthropic) Made-with: Cursor --- src/benchmark.rs | 12 +++++++++ src/benchmark_blocking.rs | 12 +++++++++ src/results.rs | 54 ++++++++++++++++++++++++++++----------- 3 files changed, 63 insertions(+), 15 deletions(-) diff --git a/src/benchmark.rs b/src/benchmark.rs index bb084be4f..d25aef31f 100644 --- a/src/benchmark.rs +++ b/src/benchmark.rs @@ -1128,12 +1128,16 @@ port={}", // Stream latency if enabled if let Some(ref mut manager) = results_manager { + // Use current timestamp since this is server-measured latency read from file + let send_timestamp_ns = + crate::results::MessageLatencyRecord::current_timestamp_ns(); let record = crate::results::MessageLatencyRecord::new( line_num, self.mechanism, self.config.message_size, crate::metrics::LatencyType::OneWay, latency, + send_timestamp_ns, ); manager.stream_latency_record(&record).await?; } @@ -1266,12 +1270,16 @@ port={}", for (i, latency) in latencies.iter().enumerate() { metrics_collector.record_message(self.config.message_size, Some(*latency))?; if let Some(ref mut manager) = results_manager { + // Use current timestamp since latencies are recorded after collection + let send_timestamp_ns = + crate::results::MessageLatencyRecord::current_timestamp_ns(); let record = crate::results::MessageLatencyRecord::new( i as u64, self.mechanism, self.config.message_size, crate::metrics::LatencyType::RoundTrip, *latency, + send_timestamp_ns, ); manager.stream_latency_record(&record).await?; } @@ -1558,12 +1566,16 @@ port={}", for (i, &one_way_latency) in one_way_latencies.iter().enumerate() { one_way_metrics.record_message(self.config.message_size, Some(one_way_latency))?; if let Some(ref mut manager) = results_manager { + // Use current timestamp since latencies are recorded after collection + let send_timestamp_ns = + crate::results::MessageLatencyRecord::current_timestamp_ns(); let record = crate::results::MessageLatencyRecord::new_combined( i as u64, self.mechanism, self.config.message_size, one_way_latency, round_trip_latencies[i], + send_timestamp_ns, ); manager.write_streaming_record_direct(&record).await?; } diff --git a/src/benchmark_blocking.rs b/src/benchmark_blocking.rs index 8a025b3d8..8207ca018 100644 --- a/src/benchmark_blocking.rs +++ b/src/benchmark_blocking.rs @@ -1132,12 +1132,16 @@ impl BlockingBenchmarkRunner { // Stream latency if enabled if let Some(ref mut manager) = results_manager { + // Use current timestamp since this is server-measured latency read from file + let send_timestamp_ns = + crate::results::MessageLatencyRecord::current_timestamp_ns(); let record = crate::results::MessageLatencyRecord::new( i as u64, self.mechanism, self.config.message_size, crate::metrics::LatencyType::OneWay, latency, + send_timestamp_ns, ); let _ = manager.stream_latency_record(&record); } @@ -1226,6 +1230,9 @@ impl BlockingBenchmarkRunner { } while start_time.elapsed() < duration { + // Capture send timestamp for streaming record (wall clock) + let send_timestamp_ns = + crate::results::MessageLatencyRecord::current_timestamp_ns(); let send_time = Instant::now(); let message = Message::new(i, payload.clone(), MessageType::Request); @@ -1245,6 +1252,7 @@ impl BlockingBenchmarkRunner { self.config.message_size, crate::metrics::LatencyType::RoundTrip, latency, + send_timestamp_ns, ); let _ = manager.stream_latency_record(&record); } @@ -1271,6 +1279,9 @@ impl BlockingBenchmarkRunner { } for i in 0..msg_count { + // Capture send timestamp for streaming record (wall clock) + let send_timestamp_ns = + crate::results::MessageLatencyRecord::current_timestamp_ns(); let send_time = Instant::now(); let message = Message::new(i as u64, payload.clone(), MessageType::Request); client_transport.send_blocking(&message)?; @@ -1293,6 +1304,7 @@ impl BlockingBenchmarkRunner { self.config.message_size, crate::metrics::LatencyType::RoundTrip, latency, + send_timestamp_ns, ); let _ = manager.stream_latency_record(&record); } diff --git a/src/results.rs b/src/results.rs index 043c33fc0..a7f3067af 100644 --- a/src/results.rs +++ b/src/results.rs @@ -153,21 +153,18 @@ impl MessageLatencyRecord { /// - `message_size`: Size of the message payload /// - `latency_type`: Type of latency measurement /// - `latency`: Measured latency duration + /// - `send_timestamp_ns`: Unix timestamp (nanoseconds) when the message was sent /// /// ## Returns - /// New record with current system timestamp + /// New record with the provided send timestamp pub fn new( message_id: u64, mechanism: IpcMechanism, message_size: usize, latency_type: LatencyType, latency: Duration, + send_timestamp_ns: u64, ) -> Self { - let timestamp_ns = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_nanos() as u64; - let latency_ns = latency.as_nanos() as u64; let (one_way_latency_ns, round_trip_latency_ns) = match latency_type { LatencyType::OneWay => (Some(latency_ns), None), @@ -175,7 +172,7 @@ impl MessageLatencyRecord { }; Self { - timestamp_ns, + timestamp_ns: send_timestamp_ns, message_id, mechanism, message_size, @@ -192,23 +189,20 @@ impl MessageLatencyRecord { /// - `message_size`: Size of the message payload /// - `one_way_latency`: One-way latency measurement /// - `round_trip_latency`: Round-trip latency measurement + /// - `send_timestamp_ns`: Unix timestamp (nanoseconds) when the message was sent /// /// ## Returns - /// New record with both latency measurements and current timestamp + /// New record with both latency measurements and the provided send timestamp pub fn new_combined( message_id: u64, mechanism: IpcMechanism, message_size: usize, one_way_latency: Duration, round_trip_latency: Duration, + send_timestamp_ns: u64, ) -> Self { - let timestamp_ns = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_nanos() as u64; - Self { - timestamp_ns, + timestamp_ns: send_timestamp_ns, message_id, mechanism, message_size, @@ -240,6 +234,21 @@ impl MessageLatencyRecord { pub fn is_combined(&self) -> bool { self.one_way_latency_ns.is_some() && self.round_trip_latency_ns.is_some() } + + /// Get current Unix timestamp in nanoseconds + /// + /// Helper function to capture the current time as a Unix timestamp + /// for use with `new()` and `new_combined()` methods. + /// + /// ## Returns + /// Unix timestamp in nanoseconds since epoch + #[inline] + pub fn current_timestamp_ns() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() as u64 + } } /// Complete benchmark results for a specific IPC mechanism @@ -1942,6 +1951,7 @@ mod tests { let mut mgr = ResultsManager::new(None, None).unwrap(); mgr.enable_per_message_streaming(&path).unwrap(); + let timestamp = MessageLatencyRecord::current_timestamp_ns(); #[cfg(unix)] let r1 = MessageLatencyRecord::new( 1, @@ -1949,6 +1959,7 @@ mod tests { 128, LatencyType::OneWay, Duration::from_micros(10), + timestamp, ); #[cfg(not(unix))] let r1 = MessageLatencyRecord::new( @@ -1957,6 +1968,7 @@ mod tests { 128, LatencyType::OneWay, Duration::from_micros(10), + timestamp, ); #[cfg(unix)] let r2 = MessageLatencyRecord::new( @@ -1965,6 +1977,7 @@ mod tests { 128, LatencyType::RoundTrip, Duration::from_micros(20), + timestamp, ); #[cfg(not(unix))] let r2 = MessageLatencyRecord::new( @@ -1973,6 +1986,7 @@ mod tests { 128, LatencyType::RoundTrip, Duration::from_micros(20), + timestamp, ); let rt = Runtime::new().unwrap(); @@ -2002,6 +2016,7 @@ mod tests { mgr.both_tests_enabled = true; // Insert pending records out-of-order. + let timestamp = MessageLatencyRecord::current_timestamp_ns(); #[cfg(unix)] let r_high = MessageLatencyRecord::new( 5, @@ -2009,6 +2024,7 @@ mod tests { 128, LatencyType::OneWay, Duration::from_micros(50), + timestamp, ); #[cfg(not(unix))] let r_high = MessageLatencyRecord::new( @@ -2017,6 +2033,7 @@ mod tests { 128, LatencyType::OneWay, Duration::from_micros(50), + timestamp, ); #[cfg(unix)] let r_low = MessageLatencyRecord::new( @@ -2025,6 +2042,7 @@ mod tests { 128, LatencyType::OneWay, Duration::from_micros(20), + timestamp, ); #[cfg(not(unix))] let r_low = MessageLatencyRecord::new( @@ -2033,6 +2051,7 @@ mod tests { 128, LatencyType::OneWay, Duration::from_micros(20), + timestamp, ); mgr.pending_records.insert(r_high.message_id, r_high); @@ -2071,6 +2090,7 @@ mod tests { let mut mgr = ResultsManager::new(None, None).unwrap(); mgr.enable_csv_streaming(&path).unwrap(); + let timestamp = MessageLatencyRecord::current_timestamp_ns(); #[cfg(unix)] let r1 = MessageLatencyRecord::new( 1, @@ -2078,6 +2098,7 @@ mod tests { 64, LatencyType::OneWay, Duration::from_micros(5), + timestamp, ); #[cfg(not(unix))] let r1 = MessageLatencyRecord::new( @@ -2086,6 +2107,7 @@ mod tests { 64, LatencyType::OneWay, Duration::from_micros(5), + timestamp, ); let rt = Runtime::new().unwrap(); @@ -2272,12 +2294,14 @@ mod tests { #[test] fn test_message_latency_record_new_combined() { + let timestamp = MessageLatencyRecord::current_timestamp_ns(); let record = MessageLatencyRecord::new_combined( 42, IpcMechanism::TcpSocket, 256, Duration::from_micros(100), Duration::from_micros(200), + timestamp, ); assert_eq!(record.message_id, 42); @@ -2286,7 +2310,7 @@ mod tests { assert_eq!(record.one_way_latency_ns, Some(100_000)); // 100 micros = 100,000 ns assert_eq!(record.round_trip_latency_ns, Some(200_000)); assert!(record.is_combined()); - assert!(record.timestamp_ns > 0); + assert_eq!(record.timestamp_ns, timestamp); } #[test] From 88bfcc458cf419cb80c566d78935ea585ebca4d4 Mon Sep 17 00:00:00 2001 From: Matt Currier Date: Tue, 10 Mar 2026 15:13:00 -0400 Subject: [PATCH 2/4] test/docs: Add streaming timestamp tests and sequential test execution documentation - Add 3 new unit tests for MessageLatencyRecord timestamp handling: test_new_uses_provided_send_timestamp, test_new_combined_uses_provided_send_timestamp, test_current_timestamp_ns_returns_recent_value - Add 2 end-to-end streaming tests in benchmark.rs: test_one_way_streaming_captures_send_timestamp, test_round_trip_streaming_captures_send_timestamp - Document sequential one-way/round-trip test execution in README.md (new "Test Execution Order" section), CONFIG.md, dashboard README, and run() doc comments in benchmark.rs and benchmark_blocking.rs - All 332 tests pass (331 passed, 1 ignored), zero clippy warnings AI-assisted-by: Claude claude-4.6-opus-high-thinking Made-with: Cursor --- CONFIG.md | 6 +++ README.md | 21 +++++++- src/benchmark.rs | 100 ++++++++++++++++++++++++++++++++++++++ src/benchmark_blocking.rs | 4 ++ src/results.rs | 59 ++++++++++++++++++++++ utils/dashboard/README.md | 2 +- 6 files changed, 190 insertions(+), 2 deletions(-) diff --git a/CONFIG.md b/CONFIG.md index 8473ef9f7..a677ad6aa 100644 --- a/CONFIG.md +++ b/CONFIG.md @@ -27,6 +27,12 @@ This document provides detailed information about configuring the IPC Benchmark ### Test Configuration +One-way and round-trip tests run **sequentially**, not +simultaneously. Each test spawns its own server process, runs to +completion, and tears down before the next test starts. By default +both are enabled. For duration-based tests (`-d`), each test gets +its own full time window. + | Option | Type | Default | Description | |--------|------|---------|-------------| | `--one-way` | Boolean | `true` | Enable one-way latency tests | diff --git a/README.md b/README.md index 4d551552c..b9c499928 100644 --- a/README.md +++ b/README.md @@ -166,6 +166,24 @@ This benchmark suite uses **high-precision monotonic clocks** to measure true IP - **Windows**: Falls back to system time (less precise) - **Characteristics**: Monotonic clocks measure time from system boot and are unaffected by NTP adjustments, daylight saving time, or manual clock changes +#### Test Execution Order + +One-way and round-trip tests run **sequentially**, never +simultaneously. Each test spawns its own server process, runs to +completion, and tears down before the next test starts: + +1. **Warmup** (if configured) +2. **One-way test** — client sends messages; server measures + receive latency +3. **Round-trip test** — client sends a message, waits for the + server's response, then sends the next + +By default both tests are enabled. Use `--one-way` or +`--round-trip` to run only one. For duration-based tests (`-d`), +each test gets its own full time window, so one-way typically +produces far more records than round-trip (fire-and-forget vs +send-wait-receive per message). + #### Timestamp Capture Points **For One-Way Latency Tests:** @@ -390,7 +408,8 @@ ipc-benchmark --log-file stderr # Continue running tests even if one mechanism fails ipc-benchmark -m all --continue-on-error -# Run only round-trip tests +# Run only round-trip tests (one-way and round-trip run +# sequentially by default; use these flags to select one) ipc-benchmark --round-trip --no-one-way # Custom percentiles for latency analysis diff --git a/src/benchmark.rs b/src/benchmark.rs index d25aef31f..348749a11 100644 --- a/src/benchmark.rs +++ b/src/benchmark.rs @@ -2632,4 +2632,104 @@ mod tests { ); } } + + /// Exercises the one-way streaming path with per-message + /// streaming enabled (not combined mode), covering the + /// send_timestamp_ns capture in run_one_way_test. + #[tokio::test] + async fn test_one_way_streaming_captures_send_timestamp() { + let tmp = tempfile::NamedTempFile::new().unwrap(); + let path = tmp.path().to_path_buf(); + + let args = Args { + mechanisms: vec![{ + #[cfg(unix)] + { + IpcMechanism::UnixDomainSocket + } + #[cfg(not(unix))] + { + IpcMechanism::TcpSocket + } + }], + message_size: 64, + msg_count: 5, + duration: None, + concurrency: 1, + warmup_iterations: 0, + one_way: true, + round_trip: false, + include_first_message: true, + ..Default::default() + }; + let config = BenchmarkConfig::from_args(&args).unwrap(); + let mechanism = args.mechanisms[0]; + let runner = BenchmarkRunner::new(config, mechanism, args.clone()); + + let mut rm = crate::results::ResultsManager::new(None, None).unwrap(); + rm.enable_per_message_streaming(&path).unwrap(); + + let res = runner.run(Some(&mut rm)).await; + assert!(res.is_ok(), "one-way streaming run failed: {:?}", res.err()); + rm.finalize().await.unwrap(); + + let s = std::fs::read_to_string(&path).expect("read streaming file"); + assert!( + s.contains("\"headings\""), + "Streaming output should have headings" + ); + assert!(s.contains("\"data\""), "Streaming output should have data"); + } + + /// Exercises the round-trip streaming path with per-message + /// streaming enabled (not combined mode), covering the + /// send_timestamp_ns capture in run_round_trip_test. + #[tokio::test] + async fn test_round_trip_streaming_captures_send_timestamp() { + let tmp = tempfile::NamedTempFile::new().unwrap(); + let path = tmp.path().to_path_buf(); + + let args = Args { + mechanisms: vec![{ + #[cfg(unix)] + { + IpcMechanism::UnixDomainSocket + } + #[cfg(not(unix))] + { + IpcMechanism::TcpSocket + } + }], + message_size: 64, + msg_count: 5, + duration: None, + concurrency: 1, + warmup_iterations: 0, + one_way: false, + round_trip: true, + include_first_message: true, + ..Default::default() + }; + let config = BenchmarkConfig::from_args(&args).unwrap(); + let mechanism = args.mechanisms[0]; + let runner = BenchmarkRunner::new(config, mechanism, args.clone()); + + let mut rm = crate::results::ResultsManager::new(None, None).unwrap(); + rm.enable_per_message_streaming(&path).unwrap(); + + let res = runner.run(Some(&mut rm)).await; + assert!( + res.is_ok(), + "round-trip streaming run failed: {:?}", + res.err() + ); + rm.finalize().await.unwrap(); + + let s = std::fs::read_to_string(&path).expect("read streaming file"); + assert!( + s.contains("\"headings\""), + "Streaming output should have headings" + ); + assert!(s.contains("\"data\""), "Streaming output should have data"); + } } diff --git a/src/benchmark_blocking.rs b/src/benchmark_blocking.rs index 8207ca018..c60982bd3 100644 --- a/src/benchmark_blocking.rs +++ b/src/benchmark_blocking.rs @@ -694,6 +694,10 @@ impl BlockingBenchmarkRunner { /// /// ## Test Execution Flow /// + /// Tests run **sequentially**, not simultaneously. Each test + /// spawns its own server process, runs to completion, and tears + /// down before the next test starts. + /// /// 1. **Validation**: Verify core availability and configuration /// 2. **Warmup Phase**: Run warmup iterations to stabilize performance /// 3. **One-way Testing**: Measure one-way latency if enabled diff --git a/src/results.rs b/src/results.rs index a7f3067af..064b554ba 100644 --- a/src/results.rs +++ b/src/results.rs @@ -2454,4 +2454,63 @@ mod tests { assert_eq!(results.test_config.concurrency, 2); assert_eq!(results.test_config.message_size, 512); } + + /// Verify that `MessageLatencyRecord::new()` uses the + /// caller-provided `send_timestamp_ns` instead of capturing + /// `SystemTime::now()` internally. + #[test] + fn test_new_uses_provided_send_timestamp() { + let fixed_ts: u64 = 1_700_000_000_000_000_000; + let record = MessageLatencyRecord::new( + 42, + IpcMechanism::TcpSocket, + 1024, + LatencyType::OneWay, + Duration::from_micros(100), + fixed_ts, + ); + assert_eq!( + record.timestamp_ns, fixed_ts, + "timestamp_ns should match the provided value" + ); + } + + /// Verify that `MessageLatencyRecord::new_combined()` uses the + /// caller-provided `send_timestamp_ns`. + #[test] + fn test_new_combined_uses_provided_send_timestamp() { + let fixed_ts: u64 = 1_600_000_000_000_000_000; + let record = MessageLatencyRecord::new_combined( + 7, + IpcMechanism::SharedMemory, + 512, + Duration::from_micros(50), + Duration::from_micros(120), + fixed_ts, + ); + assert_eq!( + record.timestamp_ns, fixed_ts, + "timestamp_ns should match the provided value" + ); + } + + /// Verify that `current_timestamp_ns()` returns a plausible, + /// monotonically non-decreasing wall-clock timestamp. + #[test] + fn test_current_timestamp_ns_returns_recent_value() { + let before = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() as u64; + let ts = MessageLatencyRecord::current_timestamp_ns(); + let after = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() as u64; + assert!( + ts >= before && ts <= after, + "current_timestamp_ns should be between before ({}) and after ({}), got {}", + before, after, ts + ); + } } diff --git a/utils/dashboard/README.md b/utils/dashboard/README.md index 9eb66401f..26ea3aac7 100644 --- a/utils/dashboard/README.md +++ b/utils/dashboard/README.md @@ -71,7 +71,7 @@ For complete ipc-benchmark parameter documentation, see the [main README](../../ - **Performance Overview**: AI-generated insights and key metrics cards - **Head-to-Head Comparisons**: Interactive comparison matrices for P50/Max latency - **Statistical Analysis**: Comprehensive latency and throughput breakdowns -- **Data Tables**: Sortable pivot tables for one-way vs round-trip performance +- **Data Tables**: Sortable pivot tables for one-way vs round-trip performance (these tests run sequentially, not simultaneously) ### **Time Series Analysis** - **Interactive Visualizations**: Real-time scatter plots with advanced controls From c3c16a31361e0c375ffb08d7d6c9be48ad9349fe Mon Sep 17 00:00:00 2001 From: Matt Currier Date: Fri, 13 Mar 2026 12:51:11 -0400 Subject: [PATCH 3/4] fix: Capture streaming timestamps at message-send time for all code paths The previous commit (68afe07) added a send_timestamp_ns parameter to MessageLatencyRecord but only the blocking round-trip path was actually capturing timestamps at send time. The async round-trip, async combined, and all one-way paths were still using current_timestamp_ns() at record-creation time (post-test), causing all timestamps within a run to cluster into the same second. Changes: - Async round-trip: capture wall-clock timestamp inside the spawned client future before each send(), return Vec<(Duration, u64)> - Async combined: same pattern for one-way latency vector - One-way (async + blocking): server now writes "wall_send_ns,latency_ns" per line (wall_send_ns = wall_clock_now - latency); client readers parse and use the server-computed send timestamp - Add parse_latency_file_line() with 7 unit tests covering valid input, missing commas, empty lines, non-numeric values, and extra commas - Enhance existing end-to-end streaming tests to validate timestamps fall within the test execution window and are not all identical - Document streaming output column definitions in README.md including timestamp_ns semantics and accuracy note for one-way clock mixing All 265+ unit tests pass, clippy clean, no scope creep. AI-assisted-by: Claude claude-4.6-opus-high-thinking (Anthropic) Made-with: Cursor --- README.md | 34 +++++- src/benchmark.rs | 210 ++++++++++++++++++++++++++++++++------ src/benchmark_blocking.rs | 13 +-- src/main.rs | 87 +++++++++++----- 4 files changed, 275 insertions(+), 69 deletions(-) diff --git a/README.md b/README.md index b9c499928..827c6dd7a 100644 --- a/README.md +++ b/README.md @@ -31,8 +31,16 @@ This benchmark suite provides a systematic way to evaluate the performance of va ### Output Formats - **JSON**: Optional, machine-readable structured output for final, aggregated results. Generated only when the `--output-file` flag is used. -- **Streaming JSON**: Real-time, per-message latency data written to a file in a columnar JSON format. This allows for efficient, live monitoring of long-running tests. The format consists of a `headings` array and a `data` array containing value arrays for each message. -- **Streaming CSV**: Real-time, per-message latency data written to a standard CSV file. This format is ideal for easy import into spreadsheets and data analysis tools. +- **Streaming JSON**: Real-time, per-message latency data written + to a file in a columnar JSON format. This allows for efficient, + live monitoring of long-running tests. The format consists of a + `headings` array and a `data` array containing value arrays for + each message. See [Streaming Output Columns](#streaming-output-columns) + for the column definitions. +- **Streaming CSV**: Real-time, per-message latency data written + to a standard CSV file. The columns match the streaming JSON + headings. This format is ideal for easy import into spreadsheets + and data analysis tools. - **Console Output**: User-friendly, color-coded summaries on `stdout`. Includes a configuration summary at startup and a detailed results summary upon completion. - **Detailed Logs**: Structured, timestamped logs written to a file or `stderr` for diagnostics. @@ -197,6 +205,28 @@ send-wait-receive per message). 3. **Client Side**: Timestamp captured after receiving response 4. **Latency Calculation**: Total elapsed time from send to receive +#### Streaming Output Columns + +The per-message streaming output (JSON and CSV) contains the +following columns: + +| Column | Type | Description | +|--------|------|-------------| +| `timestamp_ns` | `u64` | Approximate wall-clock time (nanoseconds since Unix epoch) when the message was sent. For round-trip and combined tests this is captured on the client immediately before `send()`. For one-way tests the server approximates it as `wall_clock_now - latency`. | +| `message_id` | `u64` | Zero-based sequential message identifier. | +| `mechanism` | `string` | IPC mechanism name (e.g. `"TcpSocket"`, `"UnixDomainSocket"`). | +| `message_size` | `u64` | Payload size in bytes. | +| `one_way_latency_ns` | `u64` or `null` | One-way latency in nanoseconds, or `null` if this record is round-trip only. | +| `round_trip_latency_ns` | `u64` or `null` | Round-trip latency in nanoseconds, or `null` if this record is one-way only. | + +> **Note on `timestamp_ns` accuracy:** For one-way tests the +> server computes the send timestamp by subtracting the measured +> monotonic latency from its current wall-clock time. This mixes +> two clock domains (wall-clock and monotonic) and is subject to +> minor drift from NTP adjustments, but it is the best +> approximation available without clock synchronization between +> the client and server processes. + ### What's Measured The latency measurements include: diff --git a/src/benchmark.rs b/src/benchmark.rs index 348749a11..cd58697e5 100644 --- a/src/benchmark.rs +++ b/src/benchmark.rs @@ -234,6 +234,35 @@ pub struct BenchmarkConfig { pub client_affinity: Option, } +/// Parse a line from the server-written latency file. +/// +/// The expected format is `"wall_send_ns,latency_ns"` where +/// `wall_send_ns` is the approximate wall-clock time the message +/// was sent (computed server-side as `wall_now - latency`) and +/// `latency_ns` is the measured one-way IPC latency in nanoseconds. +/// +/// # Errors +/// +/// Returns an error if the line does not contain exactly two +/// comma-separated `u64` values. +pub fn parse_latency_file_line(line: &str) -> anyhow::Result<(u64, u64)> { + let parts: Vec<&str> = line.splitn(2, ',').collect(); + if parts.len() != 2 { + anyhow::bail!( + "Invalid latency file line (expected \ + wall_send_ns,latency_ns): {}", + line + ); + } + let wall_send_ns: u64 = parts[0] + .parse() + .with_context(|| format!("Failed to parse wall_send_ns from: {}", parts[0]))?; + let latency_ns: u64 = parts[1] + .parse() + .with_context(|| format!("Failed to parse latency_ns from: {}", parts[1]))?; + Ok((wall_send_ns, latency_ns)) +} + impl BenchmarkConfig { /// Create benchmark configuration from CLI arguments /// @@ -1117,27 +1146,22 @@ port={}", .await .context("Failed to read line from latency file")? { - let latency_ns: u64 = line - .parse() - .with_context(|| format!("Failed to parse latency from line: {}", line))?; + // Parse "wall_send_ns,latency_ns" format written + // by the server process. + let (wall_send_ns, latency_ns) = parse_latency_file_line(&line)?; let latency = Duration::from_nanos(latency_ns); - // Record in metrics collector metrics_collector.record_message(self.config.message_size, Some(latency))?; - // Stream latency if enabled if let Some(ref mut manager) = results_manager { - // Use current timestamp since this is server-measured latency read from file - let send_timestamp_ns = - crate::results::MessageLatencyRecord::current_timestamp_ns(); let record = crate::results::MessageLatencyRecord::new( line_num, self.mechanism, self.config.message_size, crate::metrics::LatencyType::OneWay, latency, - send_timestamp_ns, + wall_send_ns, ); manager.stream_latency_record(&record).await?; } @@ -1192,7 +1216,7 @@ port={}", let transport_config_clone = transport_config.clone(); let client_future = async move { - let mut latencies = Vec::new(); + let mut latencies: Vec<(Duration, u64)> = Vec::new(); client_transport .start_client(&transport_config_clone) .await?; @@ -1210,6 +1234,7 @@ port={}", } while start_time.elapsed() < duration { + let wall_ts = crate::results::MessageLatencyRecord::current_timestamp_ns(); let send_time = Instant::now(); let message = Message::new(i, payload.clone(), MessageType::Request); @@ -1231,7 +1256,7 @@ port={}", .await .is_ok() { - latencies.push(send_time.elapsed()); + latencies.push((send_time.elapsed(), wall_ts)); } } _ => { @@ -1247,6 +1272,7 @@ port={}", msg_count + 1 }; for i in 0..iterations { + let wall_ts = crate::results::MessageLatencyRecord::current_timestamp_ns(); let send_time = Instant::now(); let message = Message::new(i as u64, payload.clone(), MessageType::Request); client_transport.send(&message).await?; @@ -1255,31 +1281,28 @@ port={}", } client_transport.receive().await?; if i > 0 || client_config.include_first_message { - latencies.push(send_time.elapsed()); + latencies.push((send_time.elapsed(), wall_ts)); } } } client_transport.close().await?; - Ok::, anyhow::Error>(latencies) + Ok::, anyhow::Error>(latencies) }; // Execute client work with proper affinity using spawn_with_affinity let latencies = crate::utils::spawn_with_affinity(client_future, self.config.client_affinity).await?; - for (i, latency) in latencies.iter().enumerate() { + for (i, (latency, wall_ts)) in latencies.iter().enumerate() { metrics_collector.record_message(self.config.message_size, Some(*latency))?; if let Some(ref mut manager) = results_manager { - // Use current timestamp since latencies are recorded after collection - let send_timestamp_ns = - crate::results::MessageLatencyRecord::current_timestamp_ns(); let record = crate::results::MessageLatencyRecord::new( i as u64, self.mechanism, self.config.message_size, crate::metrics::LatencyType::RoundTrip, *latency, - send_timestamp_ns, + *wall_ts, ); manager.stream_latency_record(&record).await?; } @@ -1510,8 +1533,8 @@ port={}", let transport_config_clone = transport_config.clone(); let client_future = async move { - let mut one_way_latencies = Vec::new(); - let mut round_trip_latencies = Vec::new(); + let mut one_way_latencies: Vec<(Duration, u64)> = Vec::new(); + let mut round_trip_latencies: Vec = Vec::new(); client_transport .start_client(&transport_config_clone) .await?; @@ -1522,6 +1545,7 @@ port={}", if let Some(duration) = client_config.duration { let mut i = 0u64; while start_time.elapsed() < duration { + let wall_ts = crate::results::MessageLatencyRecord::current_timestamp_ns(); let send_start = Instant::now(); let message = Message::new(i, payload.clone(), MessageType::Request); @@ -1529,7 +1553,7 @@ port={}", let one_way_latency = send_start.elapsed(); if client_transport.receive().await.is_ok() { let round_trip_latency = send_start.elapsed(); - one_way_latencies.push(one_way_latency); + one_way_latencies.push((one_way_latency, wall_ts)); round_trip_latencies.push(round_trip_latency); i += 1; } else { @@ -1542,48 +1566,46 @@ port={}", } else { let msg_count = client_config.msg_count.unwrap_or_default(); for i in 0..msg_count { + let wall_ts = crate::results::MessageLatencyRecord::current_timestamp_ns(); let send_start = Instant::now(); let message = Message::new(i as u64, payload.clone(), MessageType::Request); client_transport.send(&message).await?; let one_way_latency = send_start.elapsed(); client_transport.receive().await?; let round_trip_latency = send_start.elapsed(); - one_way_latencies.push(one_way_latency); + one_way_latencies.push((one_way_latency, wall_ts)); round_trip_latencies.push(round_trip_latency); } } client_transport.close().await?; - Ok::<(Vec, Vec), anyhow::Error>(( + Ok::<(Vec<(Duration, u64)>, Vec), anyhow::Error>(( one_way_latencies, round_trip_latencies, )) }; - // Execute client work with proper affinity using spawn_with_affinity + // Execute client work with proper affinity let (one_way_latencies, round_trip_latencies) = crate::utils::spawn_with_affinity(client_future, self.config.client_affinity).await?; - for (i, &one_way_latency) in one_way_latencies.iter().enumerate() { - one_way_metrics.record_message(self.config.message_size, Some(one_way_latency))?; + for (i, (one_way_latency, wall_ts)) in one_way_latencies.iter().enumerate() { + one_way_metrics.record_message(self.config.message_size, Some(*one_way_latency))?; if let Some(ref mut manager) = results_manager { - // Use current timestamp since latencies are recorded after collection - let send_timestamp_ns = - crate::results::MessageLatencyRecord::current_timestamp_ns(); let record = crate::results::MessageLatencyRecord::new_combined( i as u64, self.mechanism, self.config.message_size, - one_way_latency, + *one_way_latency, round_trip_latencies[i], - send_timestamp_ns, + *wall_ts, ); manager.write_streaming_record_direct(&record).await?; } } - for round_trip_latency in round_trip_latencies { + for round_trip_latency in &round_trip_latencies { round_trip_metrics - .record_message(self.config.message_size, Some(round_trip_latency))?; + .record_message(self.config.message_size, Some(*round_trip_latency))?; } // --- Cleanup --- @@ -2633,6 +2655,69 @@ mod tests { } } + #[test] + fn test_parse_latency_file_line_valid() { + let (wall, lat) = super::parse_latency_file_line("1700000000000000000,42000").unwrap(); + assert_eq!(wall, 1_700_000_000_000_000_000); + assert_eq!(lat, 42_000); + } + + #[test] + fn test_parse_latency_file_line_zeros() { + let (wall, lat) = super::parse_latency_file_line("0,0").unwrap(); + assert_eq!(wall, 0); + assert_eq!(lat, 0); + } + + #[test] + fn test_parse_latency_file_line_missing_comma() { + let result = super::parse_latency_file_line("123456789"); + assert!(result.is_err()); + let msg = result.unwrap_err().to_string(); + assert!( + msg.contains("wall_send_ns,latency_ns"), + "Error should mention expected format, got: {}", + msg + ); + } + + #[test] + fn test_parse_latency_file_line_empty() { + assert!(super::parse_latency_file_line("").is_err()); + } + + #[test] + fn test_parse_latency_file_line_non_numeric_first() { + let result = super::parse_latency_file_line("abc,789"); + assert!(result.is_err()); + let msg = result.unwrap_err().to_string(); + assert!( + msg.contains("wall_send_ns"), + "Error should mention wall_send_ns, got: {}", + msg + ); + } + + #[test] + fn test_parse_latency_file_line_non_numeric_second() { + let result = super::parse_latency_file_line("123,xyz"); + assert!(result.is_err()); + let msg = result.unwrap_err().to_string(); + assert!( + msg.contains("latency_ns"), + "Error should mention latency_ns, got: {}", + msg + ); + } + + #[test] + fn test_parse_latency_file_line_extra_commas() { + // splitn(2, ',') treats "1,2,3" as ["1", "2,3"]. + // "2,3" is not a valid u64, so this should fail. + let result = super::parse_latency_file_line("1,2,3"); + assert!(result.is_err()); + } + /// Exercises the one-way streaming path with per-message /// streaming enabled (not combined mode), covering the /// send_timestamp_ns capture in run_one_way_test. @@ -2669,8 +2754,10 @@ mod tests { let mut rm = crate::results::ResultsManager::new(None, None).unwrap(); rm.enable_per_message_streaming(&path).unwrap(); + let before_ns = crate::results::MessageLatencyRecord::current_timestamp_ns(); let res = runner.run(Some(&mut rm)).await; assert!(res.is_ok(), "one-way streaming run failed: {:?}", res.err()); + let after_ns = crate::results::MessageLatencyRecord::current_timestamp_ns(); rm.finalize().await.unwrap(); let s = std::fs::read_to_string(&path).expect("read streaming file"); @@ -2679,6 +2766,34 @@ mod tests { "Streaming output should have headings" ); assert!(s.contains("\"data\""), "Streaming output should have data"); + + // Validate that recorded timestamps fall within the + // test execution window and are not all identical + // (which was the original bug). + let parsed: serde_json::Value = serde_json::from_str(&s).expect("valid JSON"); + let data = parsed["data"].as_array().expect("data should be an array"); + assert!(!data.is_empty(), "streaming output should contain records"); + let timestamps: Vec = data + .iter() + .map(|row| row[0].as_u64().expect("timestamp_ns")) + .collect(); + for &ts in ×tamps { + assert!( + ts >= before_ns && ts <= after_ns, + "timestamp {} outside test window [{}, {}]", + ts, + before_ns, + after_ns + ); + } + if timestamps.len() > 1 { + let all_same = timestamps.windows(2).all(|w| w[0] == w[1]); + assert!( + !all_same, + "all timestamps are identical — \ + likely still capturing at record-creation time" + ); + } } /// Exercises the round-trip streaming path with per-message @@ -2717,12 +2832,14 @@ mod tests { let mut rm = crate::results::ResultsManager::new(None, None).unwrap(); rm.enable_per_message_streaming(&path).unwrap(); + let before_ns = crate::results::MessageLatencyRecord::current_timestamp_ns(); let res = runner.run(Some(&mut rm)).await; assert!( res.is_ok(), "round-trip streaming run failed: {:?}", res.err() ); + let after_ns = crate::results::MessageLatencyRecord::current_timestamp_ns(); rm.finalize().await.unwrap(); let s = std::fs::read_to_string(&path).expect("read streaming file"); @@ -2731,5 +2848,32 @@ mod tests { "Streaming output should have headings" ); assert!(s.contains("\"data\""), "Streaming output should have data"); + + // Validate that recorded timestamps fall within the + // test execution window and are not all identical. + let parsed: serde_json::Value = serde_json::from_str(&s).expect("valid JSON"); + let data = parsed["data"].as_array().expect("data should be an array"); + assert!(!data.is_empty(), "streaming output should contain records"); + let timestamps: Vec = data + .iter() + .map(|row| row[0].as_u64().expect("timestamp_ns")) + .collect(); + for &ts in ×tamps { + assert!( + ts >= before_ns && ts <= after_ns, + "timestamp {} outside test window [{}, {}]", + ts, + before_ns, + after_ns + ); + } + if timestamps.len() > 1 { + let all_same = timestamps.windows(2).all(|w| w[0] == w[1]); + assert!( + !all_same, + "all timestamps are identical — \ + likely still capturing at record-creation time" + ); + } } } diff --git a/src/benchmark_blocking.rs b/src/benchmark_blocking.rs index c60982bd3..a3af14da8 100644 --- a/src/benchmark_blocking.rs +++ b/src/benchmark_blocking.rs @@ -1125,27 +1125,22 @@ impl BlockingBenchmarkRunner { for (i, line) in reader.lines().enumerate() { let line = line.context("Failed to read line from latency file")?; - let latency_ns: u64 = line - .parse() - .with_context(|| format!("Failed to parse latency from line: {}", line))?; + // Parse "wall_send_ns,latency_ns" format written + // by the server process. + let (wall_send_ns, latency_ns) = crate::benchmark::parse_latency_file_line(&line)?; let latency = std::time::Duration::from_nanos(latency_ns); - // Record in metrics collector metrics_collector.record_message(self.config.message_size, Some(latency))?; - // Stream latency if enabled if let Some(ref mut manager) = results_manager { - // Use current timestamp since this is server-measured latency read from file - let send_timestamp_ns = - crate::results::MessageLatencyRecord::current_timestamp_ns(); let record = crate::results::MessageLatencyRecord::new( i as u64, self.mechanism, self.config.message_size, crate::metrics::LatencyType::OneWay, latency, - send_timestamp_ns, + wall_send_ns, ); let _ = manager.stream_latency_record(&record); } diff --git a/src/main.rs b/src/main.rs index dfb25e5fe..894ea741c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -42,6 +42,7 @@ use ipc_benchmark::{ results_blocking::BlockingResultsManager, }; use std::io::{self, Write}; +use std::time::{SystemTime, UNIX_EPOCH}; use tracing::{debug, error, info, warn}; use tracing_subscriber::{filter::LevelFilter, prelude::*, Layer}; @@ -679,7 +680,7 @@ fn run_server_mode_blocking(args: cli::Args) -> Result<()> { // Buffer latencies in memory instead of per-message file I/O // This avoids the massive overhead of writing to disk for each message let latency_file_path = args.internal_latency_file.clone(); - let mut latency_buffer: Vec = if latency_file_path.is_some() { + let mut latency_buffer: Vec<(u64, u64)> = if latency_file_path.is_some() { Vec::with_capacity(100_000) // Pre-allocate for performance } else { Vec::new() @@ -695,7 +696,12 @@ fn run_server_mode_blocking(args: cli::Args) -> Result<()> { let latency_ns = receive_time_ns.saturating_sub(message.timestamp); if should_buffer_latency(latency_file_path.is_some(), message.id) { - latency_buffer.push(latency_ns); + let wall_now_ns = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() as u64; + let wall_send_ns = wall_now_ns.saturating_sub(latency_ns); + latency_buffer.push((wall_send_ns, latency_ns)); } // Check for shutdown message (used by PMQ and other queue-based transports) @@ -876,7 +882,7 @@ async fn run_server_mode(args: cli::Args) -> Result<()> { // Buffer latencies in memory instead of per-message file I/O // This avoids the massive overhead of writing to disk for each message let latency_file_path = args.internal_latency_file.clone(); - let mut latency_buffer: Vec = if latency_file_path.is_some() { + let mut latency_buffer: Vec<(u64, u64)> = if latency_file_path.is_some() { Vec::with_capacity(100_000) // Pre-allocate for performance } else { Vec::new() @@ -895,7 +901,12 @@ async fn run_server_mode(args: cli::Args) -> Result<()> { let latency_ns = receive_time_ns.saturating_sub(msg.timestamp); if should_buffer_latency(latency_file_path.is_some(), msg.id) { - latency_buffer.push(latency_ns); + let wall_now_ns = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() as u64; + let wall_send_ns = wall_now_ns.saturating_sub(latency_ns); + latency_buffer.push((wall_send_ns, latency_ns)); } // Message received @@ -995,14 +1006,20 @@ fn should_buffer_latency(latency_file_enabled: bool, message_id: u64) -> bool { /// Write a buffer of latency values to a file. /// -/// Each latency is written as a single line containing the -/// nanosecond value in decimal. This format matches what the -/// client-side benchmark reader expects. +/// Each entry is written as a single line containing a +/// `"wall_send_ns,latency_ns"` pair. `wall_send_ns` is the +/// approximate wall-clock send time (computed as `wall_now - latency` +/// on the server) and `latency_ns` is the measured one-way IPC +/// latency. This format matches what `parse_latency_file_line()` +/// in the client-side benchmark reader expects. /// /// # Errors /// /// Returns an error if the file cannot be created or written. -fn write_latency_buffer(path: &str, buffer: &[u64]) -> Result<()> { +fn write_latency_buffer( + path: &str, + buffer: &[(u64, u64)], +) -> Result<()> { debug!( "Writing {} buffered latencies to file: {}", buffer.len(), @@ -1010,8 +1027,8 @@ fn write_latency_buffer(path: &str, buffer: &[u64]) -> Result<()> { ); let mut file = std::fs::File::create(path) .with_context(|| format!("Failed to create latency file: {}", path))?; - for latency_ns in buffer { - writeln!(file, "{}", latency_ns).ok(); + for &(wall_send_ns, latency_ns) in buffer { + writeln!(file, "{},{}", wall_send_ns, latency_ns).ok(); } debug!("Finished writing latencies to file"); Ok(()) @@ -1051,9 +1068,9 @@ mod tests { assert!(!should_buffer_latency(false, u64::MAX)); } - /// Verify that write_latency_buffer produces one decimal - /// u64 value per line, matching the format that the - /// client-side reader expects. + /// Verify that write_latency_buffer produces one + /// "wall_send_ns,latency_ns" pair per line, matching + /// the format that parse_latency_file_line() expects. #[test] fn test_write_latency_buffer_format() { let dir = std::env::temp_dir(); @@ -1062,18 +1079,25 @@ mod tests { .to_string_lossy() .to_string(); - let latencies: Vec = vec![100, 200, 999, 0, 42]; - write_latency_buffer(&path, &latencies).unwrap(); + let entries: Vec<(u64, u64)> = vec![ + (1000, 100), + (2000, 200), + (3000, 999), + (0, 0), + (5000, 42), + ]; + write_latency_buffer(&path, &entries).unwrap(); let file = std::fs::File::open(&path).unwrap(); - let lines: Vec = BufReader::new(file).lines().map(|l| l.unwrap()).collect(); + let lines: Vec = + BufReader::new(file).lines().map(|l| l.unwrap()).collect(); assert_eq!(lines.len(), 5); - assert_eq!(lines[0], "100"); - assert_eq!(lines[1], "200"); - assert_eq!(lines[2], "999"); - assert_eq!(lines[3], "0"); - assert_eq!(lines[4], "42"); + assert_eq!(lines[0], "1000,100"); + assert_eq!(lines[1], "2000,200"); + assert_eq!(lines[2], "3000,999"); + assert_eq!(lines[3], "0,0"); + assert_eq!(lines[4], "5000,42"); let _ = std::fs::remove_file(&path); } @@ -1102,19 +1126,29 @@ mod tests { /// matching the same parse logic the benchmark reader uses. #[test] fn test_write_latency_buffer_round_trip_parse() { + use ipc_benchmark::benchmark::parse_latency_file_line; + let dir = std::env::temp_dir(); let path = dir .join("test_latency_buffer_roundtrip.txt") .to_string_lossy() .to_string(); - let original: Vec = vec![1, u64::MAX - 1, 0, 123_456_789]; + let original: Vec<(u64, u64)> = vec![ + (1, 1), + (u64::MAX - 1, u64::MAX - 1), + (0, 0), + (999_999_999, 123_456_789), + ]; write_latency_buffer(&path, &original).unwrap(); let file = std::fs::File::open(&path).unwrap(); - let parsed: Vec = BufReader::new(file) + let parsed: Vec<(u64, u64)> = BufReader::new(file) .lines() - .filter_map(|l| l.ok().and_then(|s| s.trim().parse::().ok())) + .filter_map(|l| { + l.ok() + .and_then(|s| parse_latency_file_line(&s).ok()) + }) .collect(); assert_eq!(parsed, original); @@ -1126,7 +1160,10 @@ mod tests { /// an invalid path (e.g. a non-existent directory). #[test] fn test_write_latency_buffer_invalid_path() { - let result = write_latency_buffer("/no/such/directory/latencies.txt", &[1, 2, 3]); + let result = write_latency_buffer( + "/no/such/directory/latencies.txt", + &[(1000, 1), (2000, 2), (3000, 3)], + ); assert!(result.is_err(), "writing to invalid path should fail"); } } From bc1c30a1f427724d50b3c50683054541e80b7207 Mon Sep 17 00:00:00 2001 From: Matt Currier Date: Fri, 10 Apr 2026 10:59:04 -0400 Subject: [PATCH 4/4] fix: Capture wall-clock timestamp immediately and trim latency lines - Move SystemTime::now() capture to immediately after get_monotonic_time_ns() in both blocking and async server paths, eliminating minor drift from the should_buffer_latency branch - Add .trim() to parse_latency_file_line() input for cross-platform robustness against Windows \r\n line endings - All tests passing AI-assisted-by: Claude claude-4.6-opus-high-thinking Made-with: Cursor --- src/benchmark.rs | 1 + src/main.rs | 38 +++++++++++++------------------------- 2 files changed, 14 insertions(+), 25 deletions(-) diff --git a/src/benchmark.rs b/src/benchmark.rs index cd58697e5..2c418ce01 100644 --- a/src/benchmark.rs +++ b/src/benchmark.rs @@ -246,6 +246,7 @@ pub struct BenchmarkConfig { /// Returns an error if the line does not contain exactly two /// comma-separated `u64` values. pub fn parse_latency_file_line(line: &str) -> anyhow::Result<(u64, u64)> { + let line = line.trim(); let parts: Vec<&str> = line.splitn(2, ',').collect(); if parts.len() != 2 { anyhow::bail!( diff --git a/src/main.rs b/src/main.rs index 894ea741c..04eb00b36 100644 --- a/src/main.rs +++ b/src/main.rs @@ -693,13 +693,13 @@ fn run_server_mode_blocking(args: cli::Args) -> Result<()> { // Calculate actual IPC latency: receive_time - send_time // Use monotonic clock to avoid NTP adjustments affecting measurements let receive_time_ns = get_monotonic_time_ns(); + let wall_now_ns = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() as u64; let latency_ns = receive_time_ns.saturating_sub(message.timestamp); if should_buffer_latency(latency_file_path.is_some(), message.id) { - let wall_now_ns = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_nanos() as u64; let wall_send_ns = wall_now_ns.saturating_sub(latency_ns); latency_buffer.push((wall_send_ns, latency_ns)); } @@ -898,13 +898,13 @@ async fn run_server_mode(args: cli::Args) -> Result<()> { // Calculate actual IPC latency: receive_time - send_time // Use monotonic clock to avoid NTP adjustments affecting measurements let receive_time_ns = get_monotonic_time_ns(); + let wall_now_ns = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() as u64; let latency_ns = receive_time_ns.saturating_sub(msg.timestamp); if should_buffer_latency(latency_file_path.is_some(), msg.id) { - let wall_now_ns = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_nanos() as u64; let wall_send_ns = wall_now_ns.saturating_sub(latency_ns); latency_buffer.push((wall_send_ns, latency_ns)); } @@ -1016,10 +1016,7 @@ fn should_buffer_latency(latency_file_enabled: bool, message_id: u64) -> bool { /// # Errors /// /// Returns an error if the file cannot be created or written. -fn write_latency_buffer( - path: &str, - buffer: &[(u64, u64)], -) -> Result<()> { +fn write_latency_buffer(path: &str, buffer: &[(u64, u64)]) -> Result<()> { debug!( "Writing {} buffered latencies to file: {}", buffer.len(), @@ -1079,18 +1076,12 @@ mod tests { .to_string_lossy() .to_string(); - let entries: Vec<(u64, u64)> = vec![ - (1000, 100), - (2000, 200), - (3000, 999), - (0, 0), - (5000, 42), - ]; + let entries: Vec<(u64, u64)> = + vec![(1000, 100), (2000, 200), (3000, 999), (0, 0), (5000, 42)]; write_latency_buffer(&path, &entries).unwrap(); let file = std::fs::File::open(&path).unwrap(); - let lines: Vec = - BufReader::new(file).lines().map(|l| l.unwrap()).collect(); + let lines: Vec = BufReader::new(file).lines().map(|l| l.unwrap()).collect(); assert_eq!(lines.len(), 5); assert_eq!(lines[0], "1000,100"); @@ -1145,10 +1136,7 @@ mod tests { let file = std::fs::File::open(&path).unwrap(); let parsed: Vec<(u64, u64)> = BufReader::new(file) .lines() - .filter_map(|l| { - l.ok() - .and_then(|s| parse_latency_file_line(&s).ok()) - }) + .filter_map(|l| l.ok().and_then(|s| parse_latency_file_line(&s).ok())) .collect(); assert_eq!(parsed, original);