redhat-performance
diff --git a/‎Cargo.lock‎
Lines changed: 11 additions & 0 deletions b/‎Cargo.lock‎
Lines changed: 11 additions & 0 deletions
diff --git a/‎Cargo.toml‎
Lines changed: 2 additions & 0 deletions b/‎Cargo.toml‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎examples/blocking_basic.rs‎
Lines changed: 4 additions & 0 deletions b/‎examples/blocking_basic.rs‎
Lines changed: 4 additions & 0 deletions
diff --git a/‎examples/blocking_comparison.rs‎
Lines changed: 2 additions & 0 deletions b/‎examples/blocking_comparison.rs‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎src/benchmark.rs‎
Lines changed: 3 additions & 1 deletion b/‎src/benchmark.rs‎
Lines changed: 3 additions & 1 deletion
diff --git a/‎src/cli.rs‎
Lines changed: 128 additions & 13 deletions b/‎src/cli.rs‎
Lines changed: 128 additions & 13 deletions
diff --git a/‎src/ipc/mod.rs‎
Lines changed: 15 additions & 0 deletions b/‎src/ipc/mod.rs‎
Lines changed: 15 additions & 0 deletions
diff --git a/‎src/ipc/posix_message_queue_blocking.rs‎
Lines changed: 38 additions & 0 deletions b/‎src/ipc/posix_message_queue_blocking.rs‎
Lines changed: 38 additions & 0 deletions
diff --git a/‎src/ipc/shared_memory_blocking.rs‎
Lines changed: 36 additions & 0 deletions b/‎src/ipc/shared_memory_blocking.rs‎
Lines changed: 36 additions & 0 deletions
diff --git a/‎src/ipc/shared_memory_direct.rs‎
Lines changed: 9 additions & 0 deletions b/‎src/ipc/shared_memory_direct.rs‎
Lines changed: 9 additions & 0 deletions
@@ -53,6 +53,8 @@ thiserror = "2.0.16"
 tokio-macros = "=2.5.0"  # tokio-macros 2.6.0+ requires Rust 1.71+
 mio = "=1.0.4"  # mio 1.1.0+ requires Rust 1.71+
 core_affinity = "0.8.3"
+# MSRV: ctrlc 3.5+ uses zeroed() in statics (requires Rust 1.75+)
+ctrlc = "=3.4.5"
 os_pipe = "1.1.5"
 # MSRV: zmij 1.0.20+ requires Rust 1.71+
 zmij = "=1.0.19"
 
@@ -102,6 +102,10 @@ fn main() -> Result<()> {
         buffer_size: None,
         pmq_priority: 0,
 
+        // Standalone mode flags
+        server: false,
+        client: false,
+
         // Internal flag (not for external use)
         internal_run_as_server: false,
 
 
@@ -287,6 +287,8 @@ fn create_args(
         client_affinity: None,
         buffer_size: None,
         pmq_priority: 0,
+        server: false,
+        client: false,
         internal_run_as_server: false,
         verbose: 0,
         quiet: false,
 
@@ -377,6 +377,8 @@ impl BenchmarkConfig {
 /// #     pmq_priority: 0,
 /// #     include_first_message: false,
 /// #     blocking: false,
+/// #     server: false,
+/// #     client: false,
 /// #     internal_run_as_server: false,
 /// #     socket_path: None,
 /// #     shared_memory_name: None,
@@ -766,7 +768,7 @@ impl BenchmarkRunner {
         cmd.arg("-s").arg(self.config.message_size.to_string());
 
         if let Some(duration) = self.config.duration {
-            cmd.arg("-d").arg(format!("{}s", duration.as_secs()));
+            cmd.arg("-d").arg(format!("{}s", duration.as_secs_f64()));
         } else if let Some(count) = self.config.msg_count {
             cmd.arg("-i").arg(count.to_string());
         }
 
@@ -82,6 +82,7 @@ const TIMING: &str = "Timing";
 const CONCURRENCY: &str = "Concurrency";
 const OUTPUT_AND_LOGGING: &str = "Output and Logging";
 const ADVANCED: &str = "Advanced";
+const STANDALONE: &str = "Standalone Mode";
 
 /// Returns the default IPC mechanism based on the target OS.
 ///
@@ -366,27 +367,87 @@ pub struct Args {
     #[arg(long, default_value_t = false, help_heading = ADVANCED)]
     pub shm_direct: bool,
 
-    /// (Internal) Run the process in server-only mode.
+    /// Run in standalone server mode.
     ///
-    /// This is a hidden flag used by the benchmark runner to spawn a child
-    /// process that acts only as the server. It is not intended for direct
-    /// use by users.
-    #[arg(long, hide = true)]
-    pub internal_run_as_server: bool,
+    /// Starts the process as a server that listens for incoming client
+    /// connections. The server waits indefinitely for a client to connect,
+    /// receives messages, and responds to round-trip requests.
+    ///
+    /// For TCP and UDS, the server accepts multiple concurrent connections
+    /// (one handler thread per client). After the first client connects,
+    /// the server waits 2 seconds before checking if all clients have
+    /// disconnected. Additional clients should connect within this window.
+    ///
+    /// This enables running the server and client as independent processes,
+    /// potentially on different hosts or in different environments (e.g.,
+    /// host and container).
+    ///
+    /// Mutually exclusive with --client.
+    ///
+    /// # Examples
+    ///
+    /// ```bash
+    /// # Terminal 1: start server
+    /// ipc-benchmark --server -m uds --blocking
+    ///
+    /// # Terminal 2: start client
+    /// ipc-benchmark --client -m uds --blocking -i 10000
+    /// ```
+    #[arg(long, conflicts_with = "client", help_heading = STANDALONE)]
+    pub server: bool,
 
-    // --- Transport-specific arguments for internal use ---
-    /// (Internal) Specifies the exact socket path for UDS.
-    #[arg(long, hide = true)]
+    /// Run in standalone client mode.
+    ///
+    /// Connects to an already-running server and executes the benchmark
+    /// workload. If the server is not yet available, the client retries
+    /// the connection with backoff for up to 30 seconds before failing.
+    ///
+    /// Mutually exclusive with --server.
+    ///
+    /// # Examples
+    ///
+    /// ```bash
+    /// # Terminal 1: start server
+    /// ipc-benchmark --server -m tcp --blocking
+    ///
+    /// # Terminal 2: start client (retries until server is ready)
+    /// ipc-benchmark --client -m tcp --blocking -i 5000
+    /// ```
+    #[arg(long, conflicts_with = "server", help_heading = STANDALONE)]
+    pub client: bool,
+
+    /// Socket path for Unix Domain Sockets in standalone mode.
+    ///
+    /// Specifies the filesystem path for the UDS socket. Both server
+    /// and client must use the same path. If not specified, a default
+    /// path in the system temp directory is used.
+    #[arg(long, help_heading = STANDALONE)]
     pub socket_path: Option<String>,
 
-    /// (Internal) Specifies the exact name for Shared Memory.
-    #[arg(long, hide = true)]
+    /// Shared memory segment name for standalone mode.
+    ///
+    /// Specifies the name of the POSIX shared memory segment. Both
+    /// server and client must use the same name. If not specified,
+    /// defaults to "ipc_benchmark_shm".
+    #[arg(long, help_heading = STANDALONE)]
     pub shared_memory_name: Option<String>,
 
-    /// (Internal) Specifies the exact name for the POSIX Message Queue.
-    #[arg(long, hide = true)]
+    /// POSIX Message Queue name for standalone mode.
+    ///
+    /// Specifies the name of the message queue. Both server and client
+    /// must use the same name. If not specified, defaults to
+    /// "ipc_benchmark_pmq".
+    #[arg(long, help_heading = STANDALONE)]
     pub message_queue_name: Option<String>,
 
+    /// (Internal) Run the process in server-only mode.
+    ///
+    /// This is a hidden flag used by the benchmark runner to spawn a child
+    /// process that acts only as the server. It is not intended for direct
+    /// use by users.
+    #[arg(long, hide = true)]
+    pub internal_run_as_server: bool,
+
     /// (Internal) File path for server to write measured latencies.
     ///
     /// Used internally when spawning the server process to communicate
@@ -993,4 +1054,58 @@ mod tests {
         assert_eq!(args.message_size, 1024);
         assert_eq!(args.msg_count, 1000);
     }
+
+    #[test]
+    fn test_server_flag() {
+        let args = Args::parse_from(["ipc-benchmark", "--server", "-m", "tcp"]);
+        assert!(args.server);
+        assert!(!args.client);
+    }
+
+    #[test]
+    fn test_client_flag() {
+        let args = Args::parse_from(["ipc-benchmark", "--client", "-m", "tcp"]);
+        assert!(args.client);
+        assert!(!args.server);
+    }
+
+    #[test]
+    fn test_server_client_mutually_exclusive() {
+        let result = Args::try_parse_from(["ipc-benchmark", "--server", "--client", "-m", "tcp"]);
+        assert!(result.is_err(), "--server and --client should conflict");
+    }
+
+    #[test]
+    fn test_standalone_defaults() {
+        let args = Args::parse_from(["ipc-benchmark", "-m", "tcp"]);
+        assert!(!args.server);
+        assert!(!args.client);
+    }
+
+    #[test]
+    fn test_standalone_endpoint_flags() {
+        let args = Args::parse_from([
+            "ipc-benchmark",
+            "--server",
+            "-m",
+            "tcp",
+            "--socket-path",
+            "/tmp/test.sock",
+            "--shared-memory-name",
+            "my_shm",
+            "--message-queue-name",
+            "my_pmq",
+        ]);
+        assert!(args.server);
+        assert_eq!(args.socket_path, Some("/tmp/test.sock".to_string()));
+        assert_eq!(args.shared_memory_name, Some("my_shm".to_string()));
+        assert_eq!(args.message_queue_name, Some("my_pmq".to_string()));
+    }
+
+    #[test]
+    fn test_server_with_blocking() {
+        let args = Args::parse_from(["ipc-benchmark", "--server", "-m", "tcp", "--blocking"]);
+        assert!(args.server);
+        assert!(args.blocking);
+    }
 }
@@ -1185,6 +1185,21 @@ pub trait BlockingTransport: Send {
     /// timeout mechanisms if needed (SO_RCVTIMEO on sockets, etc).
     fn receive_blocking(&mut self) -> Result<Message>;
 
+    /// Receive a message and capture a monotonic timestamp immediately
+    /// after the raw bytes are read but before deserialization.
+    ///
+    /// This provides more accurate one-way latency measurement by
+    /// excluding deserialization overhead from the receive timestamp.
+    ///
+    /// The default implementation captures the timestamp after
+    /// `receive_blocking()` returns (including deserialization).
+    /// Transport implementations should override this to place the
+    /// timestamp between raw I/O and deserialization.
+    fn receive_blocking_timed(&mut self) -> Result<(Message, u64)> {
+        let msg = self.receive_blocking()?;
+        Ok((msg, get_monotonic_time_ns()))
+    }
+
     /// Close the transport and release resources.
     ///
     /// This method cleanly shuts down the transport, closing connections
 
@@ -471,6 +471,44 @@ impl BlockingTransport for BlockingPosixMessageQueue {
         Ok(message)
     }
 
+    fn receive_blocking_timed(&mut self) -> Result<(Message, u64)> {
+        trace!("Waiting to receive timed message via blocking POSIX message queue");
+
+        let fd = self.recv_fd.as_ref().ok_or_else(|| {
+            anyhow!(
+                "Cannot receive: message queue not initialized. \
+                 Call start_server_blocking() or start_client_blocking() first."
+            )
+        })?;
+
+        let mut buffer = vec![0u8; self.max_msg_size];
+
+        let data = loop {
+            let mut priority = 0u32;
+            match mq_receive(fd, &mut buffer, &mut priority) {
+                Ok(size) => {
+                    buffer.truncate(size);
+                    break buffer;
+                }
+                Err(Errno::EAGAIN) => {
+                    std::thread::yield_now();
+                    std::thread::sleep(Duration::from_millis(10));
+                }
+                Err(e) => {
+                    return Err(anyhow!("Failed to receive message: {}", e));
+                }
+            }
+        };
+
+        let receive_time_ns = crate::ipc::get_monotonic_time_ns();
+
+        let message: Message =
+            bincode::deserialize(&data).context("Failed to deserialize message")?;
+
+        trace!("Received message ID {}", message.id);
+        Ok((message, receive_time_ns))
+    }
+
     fn close_blocking(&mut self) -> Result<()> {
         debug!("Closing blocking POSIX message queue transport");
         self.cleanup_queues();
 
@@ -756,6 +756,42 @@ impl BlockingTransport for BlockingSharedMemory {
         Ok(message)
     }
 
+    fn receive_blocking_timed(&mut self) -> Result<(Message, u64)> {
+        self.ensure_peer_ready()?;
+
+        let ring_buffer = self.ring_buffer.ok_or_else(|| {
+            anyhow!(
+                "Cannot receive: shared memory not initialized. \
+                 Call start_server_blocking() or start_client_blocking() first."
+            )
+        })?;
+
+        #[cfg(unix)]
+        let data = unsafe { (*ring_buffer).read_data_blocking()? };
+
+        #[cfg(not(unix))]
+        let data = loop {
+            match unsafe { (*ring_buffer).read_data() } {
+                Ok(d) => break d,
+                Err(_) => {
+                    if unsafe { (*ring_buffer).shutdown.load(Ordering::Acquire) } {
+                        return Err(anyhow!("Connection closed"));
+                    }
+                    thread::yield_now();
+                    thread::sleep(Duration::from_micros(100));
+                }
+            }
+        };
+
+        // Capture timestamp after raw read, before deserialization
+        let receive_time_ns = crate::ipc::get_monotonic_time_ns();
+
+        let message: Message =
+            bincode::deserialize(&data).context("Failed to deserialize message")?;
+
+        Ok((message, receive_time_ns))
+    }
+
     fn close_blocking(&mut self) -> Result<()> {
         debug!("Closing blocking shared memory transport");
 
 
@@ -649,6 +649,15 @@ impl BlockingTransport for BlockingSharedMemoryDirect {
         Ok(message)
     }
 
+    fn receive_blocking_timed(&mut self) -> Result<(Message, u64)> {
+        // SHM-direct has no deserialization (direct memcpy), so the
+        // timestamp is captured immediately after the data read and
+        // before mutex unlock/signal. This uses the default implementation
+        // since there's no meaningful deserialization to exclude.
+        let msg = self.receive_blocking()?;
+        Ok((msg, crate::ipc::get_monotonic_time_ns()))
+    }
+
     fn close_blocking(&mut self) -> Result<()> {
         debug!("Closing direct memory SHM transport");