From b65e8c90948490afb4eb42dd8a2b5a9478894f47 Mon Sep 17 00:00:00 2001 From: Lawrence Chen <54008264+lawrencecchen@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:42:20 -0700 Subject: [PATCH] test: inject memory-instrumented transport workload --- .../tests/interactive_latency_e2e.rs | 186 ++++++++++++------ 1 file changed, 126 insertions(+), 60 deletions(-) diff --git a/cmux-tui/crates/cmux-remote/tests/interactive_latency_e2e.rs b/cmux-tui/crates/cmux-remote/tests/interactive_latency_e2e.rs index e65195fb426..ffcabdb9433 100644 --- a/cmux-tui/crates/cmux-remote/tests/interactive_latency_e2e.rs +++ b/cmux-tui/crates/cmux-remote/tests/interactive_latency_e2e.rs @@ -16,15 +16,10 @@ use url::Url; use zeroize::Zeroizing; const MAXIMUM_FRAME_BYTES: usize = 65_535; -const BULK_BYTES_PER_DIRECTION: usize = 64 * 1024 * 1024; const BULK_CHUNK_BYTES: usize = 4 * 1024; -const BULK_FRAME_COUNT: usize = BULK_BYTES_PER_DIRECTION / BULK_CHUNK_BYTES; // Keep this regression strictly sequential so every sample models one // keystroke round trip. The black-box transport proof harness separately // requires 1,000 markers overlapping a 64 MiB transfer. -const ECHO_COUNT: usize = 256; -const BULK_FRAMES_PER_ECHO: usize = BULK_FRAME_COUNT / ECHO_COUNT; -const RECEIVER_FENCE_FRAMES_PER_ECHO: usize = BULK_FRAMES_PER_ECHO / 2; const CLIENT_BULK_STREAM: u64 = 41; const SERVER_BULK_STREAM: u64 = 42; const INTERACTIVE_STREAM: u64 = 7; @@ -32,11 +27,56 @@ const ECHO_TIMEOUT: Duration = Duration::from_secs(5); const TEST_TIMEOUT: Duration = Duration::from_secs(90); const P95_BOUND: Duration = Duration::from_millis(250); const P99_BOUND: Duration = Duration::from_secs(1); -const _: () = { - assert!(BULK_FRAME_COUNT == ECHO_COUNT * BULK_FRAMES_PER_ECHO); - assert!(RECEIVER_FENCE_FRAMES_PER_ECHO < BULK_FRAMES_PER_ECHO); + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct Workload { + bulk_bytes_per_direction: usize, + echo_count: usize, + enforce_latency_bounds: bool, +} + +const DEFAULT_WORKLOAD: Workload = Workload { + bulk_bytes_per_direction: 64 * 1024 * 1024, + echo_count: 256, + enforce_latency_bounds: true, +}; +// Memory instrumentation observes the same transport lifecycle at much higher +// per-event cost. Reduce only event volume, while retaining both directions, +// receiver fences, payload digests, and the unchanged final watchdog. +const MEMORY_INSTRUMENTED_WORKLOAD: Workload = Workload { + bulk_bytes_per_direction: 4 * 1024 * 1024, + echo_count: 32, + enforce_latency_bounds: false, }; +impl Workload { + fn from_environment() -> Self { + match std::env::var("CMUX_TEST_MEMORY_INSTRUMENTATION") { + Err(std::env::VarError::NotPresent) => DEFAULT_WORKLOAD, + Ok(value) if value == "1" => MEMORY_INSTRUMENTED_WORKLOAD, + Ok(value) => panic!("CMUX_TEST_MEMORY_INSTRUMENTATION must be 1, got {value:?}"), + Err(error) => panic!("CMUX_TEST_MEMORY_INSTRUMENTATION is not Unicode: {error}"), + } + } + + fn bulk_frame_count(self) -> usize { + assert_eq!(self.bulk_bytes_per_direction % BULK_CHUNK_BYTES, 0); + self.bulk_bytes_per_direction / BULK_CHUNK_BYTES + } + + fn bulk_frames_per_echo(self) -> usize { + let bulk_frame_count = self.bulk_frame_count(); + assert_eq!(bulk_frame_count % self.echo_count, 0); + bulk_frame_count / self.echo_count + } + + fn receiver_fence_frames_per_echo(self) -> usize { + let bulk_frames_per_echo = self.bulk_frames_per_echo(); + assert!(bulk_frames_per_echo > 1); + bulk_frames_per_echo / 2 + } +} + #[derive(Debug)] struct BulkReport { started: Instant, @@ -70,6 +110,7 @@ struct EchoFenceReport { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn interactive_echo_stays_responsive_during_bidirectional_bulk_transfer() { tokio::time::timeout(TEST_TIMEOUT, async { + let workload = Workload::from_environment(); let state = tempdir().unwrap(); let auth = AuthDatabase::load_or_create(state.path(), "latency-e2e", false).unwrap(); let (daemon, mut accepted) = RemoteDaemon::new(auth.clone(), SessionLimits::default()); @@ -129,9 +170,9 @@ async fn interactive_echo_stays_responsive_during_bidirectional_bulk_transfer() // Compute reference digests outside the measured overlap window. The // receivers hash bytes with optimized dependency code instead of doing - // 128 MiB of debug-build byte assertions on async worker threads. - let expected_client_bulk = expected_bulk_digest(0x39); - let expected_server_bulk = expected_bulk_digest(0xa7); + // per-byte assertions on async worker threads. + let expected_client_bulk = expected_bulk_digest(workload, 0x39); + let expected_server_bulk = expected_bulk_digest(workload, 0xa7); // A query releases exactly one fixed Bulk epoch in each direction. The // responder waits for both remote receivers to observe half that epoch // before replying. Every measured RTT therefore causally contains new @@ -144,23 +185,29 @@ async fn interactive_echo_stays_responsive_during_bidirectional_bulk_transfer() let (echo_tx, mut echo_rx) = mpsc::channel(1); let server_receive = tokio::spawn(run_server_receiver( + workload, daemon_client.clone(), query_tx, daemon_bulk_progress_tx, )); - let client_receive = - tokio::spawn(run_client_receiver(client.clone(), echo_tx, client_bulk_progress_tx)); + let client_receive = tokio::spawn(run_client_receiver( + workload, + client.clone(), + echo_tx, + client_bulk_progress_tx, + )); let responder = tokio::spawn(run_echo_responder( + workload, daemon_client.clone(), query_rx, daemon_bulk_progress_rx, client_bulk_progress_rx, )); - let client_bulk = tokio::spawn(send_client_bulk(client.clone(), client_epoch_rx)); - let server_bulk = tokio::spawn(send_server_bulk(daemon_client, server_epoch_rx)); + let client_bulk = tokio::spawn(send_client_bulk(workload, client.clone(), client_epoch_rx)); + let server_bulk = tokio::spawn(send_server_bulk(workload, daemon_client, server_epoch_rx)); - let mut latencies = Vec::with_capacity(ECHO_COUNT); - for index in 0..ECHO_COUNT { + let mut latencies = Vec::with_capacity(workload.echo_count); + for index in 0..workload.echo_count { let payload = echo_payload(index); let issued = Instant::now(); client @@ -187,13 +234,19 @@ async fn interactive_echo_stays_responsive_during_bidirectional_bulk_transfer() let server_receive = server_receive.await.unwrap(); let client_receive = client_receive.await.unwrap(); - assert_eq!(client_bulk.bytes, BULK_BYTES_PER_DIRECTION); - assert_eq!(server_bulk.bytes, BULK_BYTES_PER_DIRECTION); + assert_eq!(client_bulk.bytes, workload.bulk_bytes_per_direction); + assert_eq!(server_bulk.bytes, workload.bulk_bytes_per_direction); assert_eq!(server_receive.bytes, client_bulk.bytes); assert_eq!(client_receive.bytes, server_bulk.bytes); - assert_eq!(responder.responses, ECHO_COUNT); - assert!(responder.daemon_received_client_bulk >= receiver_fence_target(ECHO_COUNT - 1)); - assert!(responder.client_received_daemon_bulk >= receiver_fence_target(ECHO_COUNT - 1)); + assert_eq!(responder.responses, workload.echo_count); + assert!( + responder.daemon_received_client_bulk + >= receiver_fence_target(workload, workload.echo_count - 1) + ); + assert!( + responder.client_received_daemon_bulk + >= receiver_fence_target(workload, workload.echo_count - 1) + ); assert_eq!( server_receive.digest, expected_client_bulk, "client-to-daemon Bulk digest changed", @@ -208,18 +261,20 @@ async fn interactive_echo_stays_responsive_during_bidirectional_bulk_transfer() assert!(client_receive.bulk_finished <= client_receive.finished); let metrics = latency_metrics(latencies); - assert!( - metrics.p95 < P95_BOUND, - "Interactive p95 {:?} exceeded conservative {:?} bound", - metrics.p95, - P95_BOUND, - ); - assert!( - metrics.p99 < P99_BOUND, - "Interactive p99 {:?} exceeded conservative {:?} bound", - metrics.p99, - P99_BOUND, - ); + if workload.enforce_latency_bounds { + assert!( + metrics.p95 < P95_BOUND, + "Interactive p95 {:?} exceeded conservative {:?} bound", + metrics.p95, + P95_BOUND, + ); + assert!( + metrics.p99 < P99_BOUND, + "Interactive p99 {:?} exceeded conservative {:?} bound", + metrics.p99, + P99_BOUND, + ); + } assert!( metrics.max < ECHO_TIMEOUT, "Interactive max {:?} exceeded finite {:?} bound", @@ -233,10 +288,11 @@ async fn interactive_echo_stays_responsive_during_bidirectional_bulk_transfer() let aggregate_mib = (client_bulk.bytes + server_bulk.bytes) as f64 / (1024.0 * 1024.0); let aggregate_mib_per_second = aggregate_mib / transfer_duration.as_secs_f64(); eprintln!( - "interactive-under-bulk: echoes={ECHO_COUNT} bulk_each_mib={} \ + "interactive-under-bulk: echoes={} bulk_each_mib={} \ p50={:?} p95={:?} p99={:?} max={:?} transfer={:?} aggregate_mib_s={:.1} \ receiver_fenced_mib={:.2}/{:.2}", - BULK_BYTES_PER_DIRECTION / (1024 * 1024), + workload.echo_count, + workload.bulk_bytes_per_direction / (1024 * 1024), metrics.p50, metrics.p95, metrics.p99, @@ -255,16 +311,18 @@ async fn interactive_echo_stays_responsive_during_bidirectional_bulk_transfer() } async fn send_client_bulk( + workload: Workload, client: Arc, mut epochs: mpsc::Receiver, ) -> BulkReport { let mut started = None; - for expected_epoch in 0..ECHO_COUNT { + let bulk_frames_per_echo = workload.bulk_frames_per_echo(); + for expected_epoch in 0..workload.echo_count { let epoch = epochs.recv().await.expect("client Bulk epoch source stopped early"); assert_eq!(epoch, expected_epoch, "client Bulk epoch changed order"); started.get_or_insert_with(Instant::now); - let first = epoch * BULK_FRAMES_PER_ECHO; - for index in first..first + BULK_FRAMES_PER_ECHO { + let first = epoch * bulk_frames_per_echo; + for index in first..first + bulk_frames_per_echo { client .send( Lane::Bulk, @@ -276,20 +334,22 @@ async fn send_client_bulk( .unwrap(); } } - BulkReport { started: started.unwrap(), bytes: BULK_BYTES_PER_DIRECTION } + BulkReport { started: started.unwrap(), bytes: workload.bulk_bytes_per_direction } } async fn send_server_bulk( + workload: Workload, daemon: Arc, mut epochs: mpsc::Receiver, ) -> BulkReport { let mut started = None; - for expected_epoch in 0..ECHO_COUNT { + let bulk_frames_per_echo = workload.bulk_frames_per_echo(); + for expected_epoch in 0..workload.echo_count { let epoch = epochs.recv().await.expect("server Bulk epoch source stopped early"); assert_eq!(epoch, expected_epoch, "server Bulk epoch changed order"); started.get_or_insert_with(Instant::now); - let first = epoch * BULK_FRAMES_PER_ECHO; - for index in first..first + BULK_FRAMES_PER_ECHO { + let first = epoch * bulk_frames_per_echo; + for index in first..first + bulk_frames_per_echo { daemon .send( Lane::Bulk, @@ -301,10 +361,11 @@ async fn send_server_bulk( .unwrap(); } } - BulkReport { started: started.unwrap(), bytes: BULK_BYTES_PER_DIRECTION } + BulkReport { started: started.unwrap(), bytes: workload.bulk_bytes_per_direction } } async fn run_server_receiver( + workload: Workload, daemon: Arc, query_tx: mpsc::Sender<(usize, Bytes)>, progress: watch::Sender, @@ -314,7 +375,8 @@ async fn run_server_receiver( let mut bulk_finished = None; let mut echo_count = 0; let mut bulk_digest = Sha256::new(); - while bulk_index < BULK_FRAME_COUNT || echo_count < ECHO_COUNT { + let bulk_frame_count = workload.bulk_frame_count(); + while bulk_index < bulk_frame_count || echo_count < workload.echo_count { let frame = daemon.receive().await.unwrap().expect("client connection closed early"); match (frame.lane, frame.stream) { (Lane::Bulk, CLIENT_BULK_STREAM) => { @@ -326,7 +388,7 @@ async fn run_server_receiver( bulk_digest.update(&frame.payload); bulk_index += 1; progress.send_replace(bulk_index * BULK_CHUNK_BYTES); - if bulk_index == BULK_FRAME_COUNT { + if bulk_index == bulk_frame_count { bulk_finished = Some(received_at); } } @@ -341,8 +403,8 @@ async fn run_server_receiver( (lane, stream) => panic!("unexpected client frame on {lane}/{stream}"), } } - assert_eq!(bulk_index, BULK_FRAME_COUNT); - assert_eq!(echo_count, ECHO_COUNT); + assert_eq!(bulk_index, bulk_frame_count); + assert_eq!(echo_count, workload.echo_count); ReceiveReport { started: started.unwrap(), bulk_finished: bulk_finished.unwrap(), @@ -353,6 +415,7 @@ async fn run_server_receiver( } async fn run_client_receiver( + workload: Workload, client: Arc, echo_tx: mpsc::Sender, progress: watch::Sender, @@ -362,7 +425,8 @@ async fn run_client_receiver( let mut bulk_finished = None; let mut echo_count = 0; let mut bulk_digest = Sha256::new(); - while bulk_index < BULK_FRAME_COUNT || echo_count < ECHO_COUNT { + let bulk_frame_count = workload.bulk_frame_count(); + while bulk_index < bulk_frame_count || echo_count < workload.echo_count { let frame = client.receive().await.unwrap().expect("daemon connection closed early"); match (frame.lane, frame.stream) { (Lane::Bulk, SERVER_BULK_STREAM) => { @@ -374,7 +438,7 @@ async fn run_client_receiver( bulk_digest.update(&frame.payload); bulk_index += 1; progress.send_replace(bulk_index * BULK_CHUNK_BYTES); - if bulk_index == BULK_FRAME_COUNT { + if bulk_index == bulk_frame_count { bulk_finished = Some(received_at); } } @@ -385,8 +449,8 @@ async fn run_client_receiver( (lane, stream) => panic!("unexpected daemon frame on {lane}/{stream}"), } } - assert_eq!(bulk_index, BULK_FRAME_COUNT); - assert_eq!(echo_count, ECHO_COUNT); + assert_eq!(bulk_index, bulk_frame_count); + assert_eq!(echo_count, workload.echo_count); ReceiveReport { started: started.unwrap(), bulk_finished: bulk_finished.unwrap(), @@ -397,6 +461,7 @@ async fn run_client_receiver( } async fn run_echo_responder( + workload: Workload, daemon: Arc, mut queries: mpsc::Receiver<(usize, Bytes)>, mut daemon_received_client_bulk: watch::Receiver, @@ -404,11 +469,11 @@ async fn run_echo_responder( ) -> EchoFenceReport { let mut daemon_progress = 0; let mut client_progress = 0; - for expected_index in 0..ECHO_COUNT { + for expected_index in 0..workload.echo_count { let (index, payload) = queries.recv().await.expect("Interactive query source stopped early"); assert_eq!(index, expected_index, "Interactive query changed order"); - let target = receiver_fence_target(index); + let target = receiver_fence_target(workload, index); daemon_progress = wait_for_receiver_progress( &mut daemon_received_client_bulk, target, @@ -427,7 +492,7 @@ async fn run_echo_responder( .unwrap(); } EchoFenceReport { - responses: ECHO_COUNT, + responses: workload.echo_count, daemon_received_client_bulk: daemon_progress, client_received_daemon_bulk: client_progress, } @@ -447,8 +512,9 @@ async fn wait_for_receiver_progress( received } -fn receiver_fence_target(echo_index: usize) -> usize { - (echo_index * BULK_FRAMES_PER_ECHO + RECEIVER_FENCE_FRAMES_PER_ECHO) * BULK_CHUNK_BYTES +fn receiver_fence_target(workload: Workload, echo_index: usize) -> usize { + (echo_index * workload.bulk_frames_per_echo() + workload.receiver_fence_frames_per_echo()) + * BULK_CHUNK_BYTES } fn bulk_payload(seed: u8, index: usize) -> Bytes { @@ -469,9 +535,9 @@ fn assert_bulk_frame_header(payload: &Bytes, expected_index: usize) { assert_eq!(u64::from_be_bytes(encoded_index), expected_index as u64); } -fn expected_bulk_digest(seed: u8) -> [u8; 32] { +fn expected_bulk_digest(workload: Workload, seed: u8) -> [u8; 32] { let mut digest = Sha256::new(); - for index in 0..BULK_FRAME_COUNT { + for index in 0..workload.bulk_frame_count() { digest.update(bulk_payload(seed, index)); } digest.finalize().into() @@ -485,7 +551,7 @@ fn echo_payload(index: usize) -> Bytes { } fn latency_metrics(mut samples: Vec) -> LatencyMetrics { - assert_eq!(samples.len(), ECHO_COUNT); + assert!(!samples.is_empty()); samples.sort_unstable(); LatencyMetrics { p50: percentile(&samples, 50),