diff --git a/crates/irtt-cli/tests/client_multi_target.rs b/crates/irtt-cli/tests/client_multi_target.rs index 987592a..d6f39af 100644 --- a/crates/irtt-cli/tests/client_multi_target.rs +++ b/crates/irtt-cli/tests/client_multi_target.rs @@ -4,7 +4,7 @@ use std::{ io::Read, net::{SocketAddr, UdpSocket}, process::{Command, Output, Stdio}, - sync::{mpsc, Arc, Barrier}, + sync::mpsc, thread, thread::JoinHandle, time::{Duration, Instant}, @@ -425,9 +425,26 @@ fn finite_multi_target_peer_close_is_accepted_as_completion() { #[test] fn continuous_all_peer_closed_targets_exit_nonzero() { let params = test_params(None, Duration::from_millis(10)); - let first_probe_barrier = Arc::new(Barrier::new(2)); - let a = start_synchronized_peer_close_server(params.clone(), Arc::clone(&first_probe_barrier)); - let b = start_synchronized_peer_close_server(params, first_probe_barrier); + let (a_seen_tx, a_seen_rx) = mpsc::channel(); + let (a_release_tx, a_release_rx) = mpsc::channel(); + let (b_seen_tx, b_seen_rx) = mpsc::channel(); + let (b_release_tx, b_release_rx) = mpsc::channel(); + let a = start_synchronized_peer_close_server(params.clone(), a_seen_tx, a_release_rx); + let b = start_synchronized_peer_close_server(params, b_seen_tx, b_release_rx); + let coordinator = thread::spawn(move || { + a_seen_rx + .recv_timeout(Duration::from_secs(2)) + .expect("timed out waiting for target A's first probe"); + b_seen_rx + .recv_timeout(Duration::from_secs(2)) + .expect("timed out waiting for target B's first probe"); + a_release_tx + .send(()) + .expect("target A stopped waiting for peer-close release"); + b_release_tx + .send(()) + .expect("target B stopped waiting for peer-close release"); + }); let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli")); command.args([ "--duration", @@ -447,6 +464,7 @@ fn continuous_all_peer_closed_targets_exit_nonzero() { ]); let output = run_with_timeout(command, Duration::from_secs(3)); + coordinator.join().unwrap(); a.join(); b.join(); @@ -863,7 +881,8 @@ fn start_peer_close_server(params: Params) -> FakeServer { fn start_synchronized_peer_close_server( params: Params, - first_probe_barrier: Arc, + first_probe_seen_tx: mpsc::Sender<()>, + release_peer_close_rx: mpsc::Receiver<()>, ) -> FakeServer { let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); socket @@ -887,7 +906,12 @@ fn start_synchronized_peer_close_server( "expected a complete echo probe before synchronized peer-close reply" ); let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap()); - first_probe_barrier.wait(); + first_probe_seen_tx + .send(()) + .expect("peer-close coordinator stopped waiting for first probe"); + release_peer_close_rx + .recv_timeout(Duration::from_secs(2)) + .expect("timed out waiting to release synchronized peer-close reply"); socket .send_to( &echo_reply_packet(