redhat-performance
diff --git a/‎src/benchmark.rs‎
Lines changed: 2 additions & 0 deletions b/‎src/benchmark.rs‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎src/benchmark_blocking.rs‎
Lines changed: 2 additions & 0 deletions b/‎src/benchmark_blocking.rs‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎src/ipc/mod.rs‎
Lines changed: 32 additions & 0 deletions b/‎src/ipc/mod.rs‎
Lines changed: 32 additions & 0 deletions
diff --git a/‎src/ipc/posix_message_queue.rs‎
Lines changed: 34 additions & 9 deletions b/‎src/ipc/posix_message_queue.rs‎
Lines changed: 34 additions & 9 deletions
diff --git a/‎src/ipc/posix_message_queue_blocking.rs‎
Lines changed: 9 additions & 6 deletions b/‎src/ipc/posix_message_queue_blocking.rs‎
Lines changed: 9 additions & 6 deletions
@@ -494,6 +494,8 @@ impl BenchmarkRunner {
             self.config.msg_count,
             self.config.duration,
             self.config.warmup_iterations,
+            self.config.one_way,
+            self.config.round_trip,
         );
 
         // Run warmup if configured
 
@@ -720,6 +720,8 @@ impl BlockingBenchmarkRunner {
             self.config.msg_count,
             self.config.duration,
             self.config.warmup_iterations,
+            self.config.one_way,
+            self.config.round_trip,
         );
 
         // Run warmup if configured
 
@@ -1574,3 +1574,35 @@ mod tests {
         );
     }
 }
+
+#[test]
+fn test_timestamp_offset_update_in_serialized_buffer() {
+    use super::*;
+
+    // Create a message with timestamp 0
+    let mut msg = Message::new(42, vec![1, 2, 3, 4], MessageType::OneWay);
+    msg.timestamp = 0;
+
+    // Serialize it
+    let mut serialized = bincode::serialize(&msg).unwrap();
+
+    // Verify timestamp is 0 in bytes 8-15
+    let extracted_ts = u64::from_le_bytes(serialized[8..16].try_into().unwrap());
+    assert_eq!(extracted_ts, 0, "Timestamp should be 0 initially");
+
+    // Now update the timestamp bytes
+    let new_ts: u64 = 123456789;
+    let ts_bytes = new_ts.to_le_bytes();
+    serialized[8..16].copy_from_slice(&ts_bytes);
+
+    // Verify it was updated
+    let updated_ts = u64::from_le_bytes(serialized[8..16].try_into().unwrap());
+    assert_eq!(updated_ts, new_ts, "Timestamp should be updated");
+
+    // Deserialize and verify
+    let deserialized: Message = bincode::deserialize(&serialized).unwrap();
+    assert_eq!(
+        deserialized.timestamp, new_ts,
+        "Deserialized timestamp should match"
+    );
+}
@@ -524,7 +524,14 @@ impl IpcTransport for PosixMessageQueueTransport {
     /// error, indicating that the message queue is full. The transport will
     /// retry with a backoff, and a warning will be logged on the first
     /// occurrence.
+    ///
+    /// ## Timestamp Accuracy
+    ///
+    /// To ensure accurate one-way latency measurement, the timestamp is updated
+    /// immediately before the `mq_send` syscall inside the blocking task. This
+    /// excludes async scheduling overhead from the measured latency.
     async fn send(&mut self, message: &Message) -> Result<bool> {
+        // Pre-serialize with current timestamp (will be updated before send)
         let data = message.to_bytes()?;
         let fd_ref = self
             .mq_fd
@@ -538,15 +545,26 @@ impl IpcTransport for PosixMessageQueueTransport {
         // Get the priority from the config, which was set during transport creation.
         let priority = self.config.as_ref().map_or(0, |c| c.pmq_priority);
 
+        // Pre-compute timestamp offset for efficient in-place updates
+        let ts_offset = Message::timestamp_offset();
+
         // Use non-blocking send with exponential backoff for queue-full conditions
         let mut retry_delay_ms = 1;
         let max_retries = 100; // More retries since each one is much faster
 
         for attempt in 0..max_retries {
             let start_time = std::time::Instant::now();
             let result = tokio::task::spawn_blocking({
-                let data = data.clone();
+                let mut data = data.clone();
+                let ts_range = ts_offset.clone();
                 move || {
+                    // CRITICAL: Update timestamp immediately before mq_send syscall
+                    // This ensures accurate latency measurement by excluding async
+                    // scheduling overhead from the measured one-way latency.
+                    let ts_now = crate::ipc::get_monotonic_time_ns();
+                    let ts_bytes = ts_now.to_le_bytes();
+                    data[ts_range].copy_from_slice(&ts_bytes);
+
                     // Reconstruct MqdT from raw fd for the blocking operation
                     let fd = unsafe { MqdT::from_raw_fd(raw_fd) };
                     // std::mem::forget(fd); // Don't close the fd when this MqdT drops
@@ -649,6 +667,11 @@ impl IpcTransport for PosixMessageQueueTransport {
     /// - **Temporary**: Queue empty (EAGAIN) - retried automatically
     /// - **Permanent**: Invalid queue state or deserialization failure
     /// - **Resource**: No queue available or transport not initialized
+    ///
+    /// ## Polling Strategy
+    ///
+    /// Uses moderate polling (100–500µs backoff, 1000 retries) to
+    /// balance latency impact against CPU usage.
     async fn receive(&mut self) -> Result<Message> {
         let fd_ref = self
             .mq_fd
@@ -657,9 +680,10 @@ impl IpcTransport for PosixMessageQueueTransport {
         let raw_fd = fd_ref.as_raw_fd();
         let max_msg_size = self.max_msg_size;
 
-        // Use non-blocking receive with exponential backoff for empty queue conditions
-        let mut retry_delay_ms = 1;
-        let max_retries = 100;
+        // Use non-blocking receive with moderate polling
+        // Balance between latency impact and CPU usage
+        let mut retry_delay_us = 100; // Start with 100 microseconds
+        let max_retries = 1000; // Reasonable retry count
 
         for attempt in 0..max_retries {
             let result = tokio::task::spawn_blocking({
@@ -682,7 +706,9 @@ impl IpcTransport for PosixMessageQueueTransport {
                         "Received message {} bytes via POSIX message queue",
                         buffer.len()
                     );
-                    return Message::from_bytes(&buffer);
+                    let message = Message::from_bytes(&buffer)?;
+
+                    return Ok(message);
                 }
                 Err(Errno::EAGAIN) => {
                     // Queue is empty, wait and retry
@@ -692,10 +718,9 @@ impl IpcTransport for PosixMessageQueueTransport {
                             max_retries
                         ));
                     }
-                    // Justification: Short, exponentially increasing delay to wait for a message to arrive
-                    // in an empty queue without busy-waiting.
-                    tokio::time::sleep(Duration::from_millis(retry_delay_ms)).await;
-                    retry_delay_ms = (retry_delay_ms * 2).min(10); // Cap at 10ms
+                    // Use moderate delays (100-500 microseconds) to balance latency and CPU
+                    tokio::time::sleep(Duration::from_micros(retry_delay_us)).await;
+                    retry_delay_us = (retry_delay_us * 2).min(500); // Cap at 500µs
                 }
                 Err(e) => {
                     return Err(anyhow!("Failed to receive message: {}", e));
 
@@ -391,18 +391,21 @@ impl BlockingTransport for BlockingPosixMessageQueue {
             ));
         }
 
-        // Capture timestamp immediately before send and update bytes in buffer
-        message_with_timestamp.set_timestamp_now();
-        let timestamp_bytes = message_with_timestamp.timestamp.to_le_bytes();
+        // Pre-compute the timestamp offset for efficient in-place updates
         let ts_offset = Message::timestamp_offset();
-        serialized[ts_offset].copy_from_slice(&timestamp_bytes);
 
-        // Send immediately - no intervening work
+        // Send with retry loop - update timestamp on each attempt
         // Use a retry loop with timeouts since mq_send can fail with EAGAIN
         let start = std::time::Instant::now();
         let timeout = Duration::from_secs(5);
 
         loop {
+            // CRITICAL: Capture timestamp immediately before each send attempt
+            // This ensures accurate latency measurement even under backpressure
+            let ts_now = crate::ipc::get_monotonic_time_ns();
+            let timestamp_bytes = ts_now.to_le_bytes();
+            serialized[ts_offset.clone()].copy_from_slice(&timestamp_bytes);
+
             match mq_send(fd, &serialized, self.priority) {
                 Ok(()) => {
                     trace!("Message ID {} sent successfully", message.id);
@@ -415,7 +418,7 @@ impl BlockingTransport for BlockingPosixMessageQueue {
                             "Timeout: message queue full, possible backpressure"
                         ));
                     }
-                    // Yield and retry
+                    // Yield and retry - timestamp will be updated on next iteration
                     std::thread::yield_now();
                     std::thread::sleep(Duration::from_millis(10));
                 }