diff --git a/src/netplay.rs b/src/netplay.rs index b2e9b75..f80dbcc 100644 --- a/src/netplay.rs +++ b/src/netplay.rs @@ -149,7 +149,12 @@ fn send_message(netplay: &mut Netplay, message: NetplayMessage) { } } -fn process_reliable_messages(netplay: &mut Netplay) { +/// Drain the reliable channel and route every complete message into +/// `netplay.messages`, returning the decoded batch so callers can count +/// messages that share a name: several peers may ack within a single +/// channel poll, and the by-name `messages` map would collapse them into +/// one entry. +fn process_reliable_messages(netplay: &mut Netplay) -> Vec<(String, Vec)> { let packets = netplay .reliable_channel .receive() @@ -157,15 +162,18 @@ fn process_reliable_messages(netplay: &mut Netplay) { .map(|(_, data)| data.to_vec()) .collect::>>(); + let mut decoded = Vec::new(); for (name, data) in decode_packets( &mut netplay.incoming_message, &mut netplay.received_data, packets, ) { - netplay.messages.insert(name, data); + netplay.messages.insert(name.clone(), data.clone()); check_input_delay(netplay); check_disconnect(netplay); + decoded.push((name, data)); } + decoded } /// Poll `messages` for `name` until it arrives or `timeout` elapses, calling @@ -469,11 +477,20 @@ pub fn save_netplay_session(device: &mut device::Device, path: &std::path::Path) /// loading the broadcast session state before giving up on that peer. const NETPLAY_STATE_ACK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); +/// Count how many messages in a decoded batch carry `name`. Used for +/// ready-acks, where one ack per peer is expected and several peers can +/// acknowledge within a single channel poll. +fn count_messages_named(decoded: &[(String, Vec)], name: &str) -> usize { + decoded.iter().filter(|(decoded_name, _)| decoded_name == name).count() +} + /// Wait for `expected` acknowledgements named `name`, polling the reliable /// channel until all of them arrive or `timeout` elapses. /// /// Acks are counted, not attributed to a specific peer: `send_ready` -/// broadcasts to every peer, so the host receives one copy per peer. +/// broadcasts to every peer, so the host receives one copy per peer. The +/// count comes from the decoded batch of every poll — not from the +/// by-name `messages` map — so acks arriving together are all counted. fn wait_for_ready_all( netplay: &mut Netplay, name: &str, @@ -483,10 +500,9 @@ fn wait_for_ready_all( let now = std::time::Instant::now(); let mut received = 0; while received < expected { - process_reliable_messages(netplay); - if netplay.messages.remove(name).is_some() { - received += 1; - continue; + received += count_messages_named(&process_reliable_messages(netplay), name); + if received >= expected { + return true; } if now.elapsed() > timeout { return false; @@ -1239,7 +1255,7 @@ mod tests { /// there, not at 0. #[test] fn resume_state_reaches_peer_with_same_frame_metadata() { - use crate::savestates::{create_savestate_in_memory, load_savestate_from_data}; + use crate::savestates::create_savestate_in_memory; // Host: a session with distinctive emulator state, captured in // memory (issue #2 API) and wrapped in the issue #3 on-disk @@ -1318,6 +1334,7 @@ mod tests { #[test] fn resume_host_loads_state_from_disk_and_resets_frame() { use crate::device::Device; + use crate::savestates::create_savestate_in_memory; let mut host_device = Device::new(false); host_device.cpu.gpr[5] = 0xDEAD_BEEF; @@ -1368,23 +1385,29 @@ mod tests { let _ = std::fs::remove_file(&path); } - /// The host's 30s ack wait succeeds once every peer acknowledged and - /// times out (returning false) when acks are missing. + /// Ack counting is per decoded batch: several peers acknowledging + /// within a single channel poll are all counted, and unrelated + /// messages are ignored. + #[test] + fn ack_batch_counts_every_ack() { + let batch = vec![ + (NETPLAY_STATE_READY_MESSAGE.to_string(), vec![]), + (NETPLAY_STATE_READY_MESSAGE.to_string(), vec![]), + ("unrelated".to_string(), vec![1]), + ]; + assert_eq!( + count_messages_named(&batch, NETPLAY_STATE_READY_MESSAGE), + 2, + "both acks in one poll are counted; unrelated messages are not" + ); + } + + /// The host's 30s ack wait times out (returning false) when acks are + /// missing instead of hanging forever. #[test] fn wait_for_ready_all_counts_acks_and_times_out() { - // All acks present: succeeds immediately. - let mut netplay = mock_netplay(0); - netplay - .messages - .insert(NETPLAY_STATE_READY_MESSAGE.to_string(), vec![]); - assert!(wait_for_ready_all( - &mut netplay, - NETPLAY_STATE_READY_MESSAGE, - 1, - std::time::Duration::from_millis(200), - )); - - // Missing acks: gives up after the timeout. + // Missing acks (the mock channel never delivers): gives up after + // the timeout. let mut netplay = mock_netplay(0); assert!(!wait_for_ready_all( &mut netplay,