Skip to content

Test remote latency under memory instrumentation #10025

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lawrencecchen merged 1 commit into from
Aug 12, 2026
Merged

Test remote latency under memory instrumentation #10025

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
186 changes: 126 additions & 60 deletions cmux-tui/crates/cmux-remote/tests/interactive_latency_e2e.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,27 +16,67 @@ 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;
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,
Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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",
Expand All @@ -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",
Expand All @@ -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,
Expand All @@ -255,16 +311,18 @@ async fn interactive_echo_stays_responsive_during_bidirectional_bulk_transfer()
}

async fn send_client_bulk(
workload: Workload,
client: Arc<ClientConnection>,
mut epochs: mpsc::Receiver<usize>,
) -> 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,
Expand All @@ -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<ServerConnection>,
mut epochs: mpsc::Receiver<usize>,
) -> 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,
Expand All @@ -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<ServerConnection>,
query_tx: mpsc::Sender<(usize, Bytes)>,
progress: watch::Sender<usize>,
Expand All @@ -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) => {
Expand All @@ -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);
}
}
Expand All @@ -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(),
Expand All @@ -353,6 +415,7 @@ async fn run_server_receiver(
}

async fn run_client_receiver(
workload: Workload,
client: Arc<ClientConnection>,
echo_tx: mpsc::Sender<Bytes>,
progress: watch::Sender<usize>,
Expand All @@ -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) => {
Expand All @@ -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);
}
}
Expand All @@ -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(),
Expand All @@ -397,18 +461,19 @@ async fn run_client_receiver(
}

async fn run_echo_responder(
workload: Workload,
daemon: Arc<ServerConnection>,
mut queries: mpsc::Receiver<(usize, Bytes)>,
mut daemon_received_client_bulk: watch::Receiver<usize>,
mut client_received_daemon_bulk: watch::Receiver<usize>,
) -> 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,
Expand All @@ -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,
}
Expand All @@ -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 {
Expand All @@ -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()
Expand All @@ -485,7 +551,7 @@ fn echo_payload(index: usize) -> Bytes {
}

fn latency_metrics(mut samples: Vec<Duration>) -> LatencyMetrics {
assert_eq!(samples.len(), ECHO_COUNT);
assert!(!samples.is_empty());
samples.sort_unstable();
LatencyMetrics {
p50: percentile(&samples, 50),
Expand Down