From 4cc85d595f06049cccd971f9c7196cf67195f584 Mon Sep 17 00:00:00 2001 From: Logan McNaughton <848146+loganmc10@users.noreply.github.com> Date: Fri, 12 Jun 2026 15:34:21 +0200 Subject: [PATCH] pass channel to ggrs (#1064) * pass channel to ggrs * more * more * more * more * more * more * more * more * more * more * more * more * more * more * more --- src/lib.rs | 3 +- src/netplay.rs | 72 +++++++++++++++++++++++++++++----------- src/retroachievements.rs | 3 +- 3 files changed, 54 insertions(+), 24 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 144b1b73..584e0134 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -228,10 +228,9 @@ pub fn run(args: Args, arg_count: usize) -> std::io::Result<()> { tx.send(false).unwrap(); } - let join_handle = std::thread::spawn(move || { + tokio::task::block_in_place(move || { rx.blocking_recv().unwrap(); }); - join_handle.join().unwrap(); ra_config } else { diff --git a/src/netplay.rs b/src/netplay.rs index 4bd29185..ecb5cdc4 100644 --- a/src/netplay.rs +++ b/src/netplay.rs @@ -11,21 +11,24 @@ impl ggrs::Config for GgrsConfig { type Address = matchbox_socket::PeerId; } -pub struct MatchboxSocket(matchbox_socket::WebRtcSocket); +static DISCONNECTED_PEERS: std::sync::LazyLock< + std::sync::Mutex>, +> = std::sync::LazyLock::new(|| std::sync::Mutex::new(rustc_hash::FxHashSet::default())); -impl ggrs::NonBlockingSocket for MatchboxSocket { +pub struct MatchboxChannel(matchbox_socket::WebRtcChannel); + +impl ggrs::NonBlockingSocket for MatchboxChannel { fn send_to(&mut self, msg: &ggrs::Message, addr: &matchbox_socket::PeerId) { let encoded = postcard::to_stdvec(msg).expect("serialization failed"); - let channel = self.0.get_channel_mut(0).unwrap(); - if channel.config().max_retransmits != Some(0) || channel.config().ordered { - eprintln!("Sending GGRS traffic over reliable channel"); + if let Err(_) = self.0.try_send(encoded.into(), *addr) + && let Ok(mut disconnected_peers) = DISCONNECTED_PEERS.lock() + { + disconnected_peers.insert(*addr); } - channel.send(encoded.into(), *addr); } fn receive_all_messages(&mut self) -> Vec<(matchbox_socket::PeerId, ggrs::Message)> { - let channel = self.0.get_channel_mut(0).unwrap(); - channel + self.0 .receive() .iter() .filter_map(|(peer, packet)| { @@ -61,7 +64,12 @@ fn send_message(netplay: &mut Netplay, message: NetplayMessage) { let chunks = data.chunks(16384).collect::>(); for peer in netplay.peers.iter() { for chunk in chunks.iter() { - netplay.reliable_channel.send(chunk.to_vec().into(), *peer); + if let Err(e) = netplay + .reliable_channel + .try_send(chunk.to_vec().into(), *peer) + { + eprintln!("Failed to send message: {}", e); + } } } } @@ -86,6 +94,7 @@ fn process_reliable_messages(netplay: &mut Netplay) { .messages .insert(decoded_message.name, decoded_message.data); netplay.incoming_message.clear(); + check_input_delay(netplay); } } } @@ -119,7 +128,9 @@ fn send_player_number( }; let data = postcard::to_stdvec(&message).unwrap(); for peer in peers { - channel.send(data.clone().into(), peer); + if let Err(e) = channel.try_send(data.clone().into(), peer) { + eprintln!("Failed to send message: {}", e); + } } } @@ -226,6 +237,23 @@ pub fn in_rollback(netplay: Option<&Netplay>) -> bool { } } +fn process_disconnected_peers(netplay: &mut Netplay) { + if let Ok(mut disconnected_peers) = DISCONNECTED_PEERS.lock() { + for addr in disconnected_peers.drain() { + for handle in netplay.session.handles_by_address(addr) { + if let Err(e) = netplay.session.disconnect_player(handle) { + eprintln!("Error disconnecting player: {}", e); + } else { + ui::video::onscreen_message( + &format!("Player {} disconnected", handle + 1), + ui::video::MESSAGE_LENGTH_MESSAGE_SHORT, + ); + } + } + } + } +} + pub fn process_requests( device: &mut device::Device, ) -> Vec<(ui::input::InputData, ggrs::InputStatus)> { @@ -260,6 +288,7 @@ pub fn process_requests( fn process_netplay(device: &mut device::Device) { let netplay = device.netplay.as_mut().unwrap(); + process_disconnected_peers(netplay); netplay.session.poll_remote_clients(); for event in netplay.session.events() { @@ -267,9 +296,8 @@ fn process_netplay(device: &mut device::Device) { ggrs::GgrsEvent::Synchronizing { .. } => {} ggrs::GgrsEvent::Synchronized { .. } => {} ggrs::GgrsEvent::Disconnected { .. } => { - eprintln!("peer disconnected"); ui::video::onscreen_message( - "Peer disconnected", + "Lost connection to peer", ui::video::MESSAGE_LENGTH_MESSAGE_LONG, ); } @@ -293,7 +321,6 @@ fn process_netplay(device: &mut device::Device) { } process_reliable_messages(netplay); - check_input_delay(netplay); advance_frame(device); } @@ -323,7 +350,7 @@ pub fn init( input_delay: usize, pal: bool, ) -> Option { - let (socket, loop_fut) = matchbox_socket::WebRtcSocketBuilder::new(server_addr) + let (mut socket, loop_fut) = matchbox_socket::WebRtcSocketBuilder::new(server_addr) .add_unreliable_channel() .add_reliable_channel() .build(); @@ -332,16 +359,16 @@ pub fn init( eprintln!("WebRTC loop failed: {}", e); } }); - let mut matchbox_socket = MatchboxSocket(socket); - let mut reliable_channel = matchbox_socket.0.take_channel(1).unwrap(); + + let matchbox_channel = MatchboxChannel(socket.take_channel(0).unwrap()); + let mut reliable_channel = socket.take_channel(1).unwrap(); let now = std::time::Instant::now(); let timeout = std::time::Duration::from_secs(10); let mut player_numbers = std::collections::BTreeMap::new(); player_numbers.insert(player_number, None); loop { - matchbox_socket.0.update_peers(); - let peers = matchbox_socket - .0 + socket.update_peers(); + let peers = socket .connected_peers() .collect::>(); @@ -386,7 +413,12 @@ pub fn init( } } - let mut session = session_builder.start_p2p_session(matchbox_socket).unwrap(); + if matchbox_channel.0.config().max_retransmits != Some(0) || matchbox_channel.0.config().ordered + { + eprintln!("Sending GGRS traffic over reliable channel"); + } + + let mut session = session_builder.start_p2p_session(matchbox_channel).unwrap(); let now = std::time::Instant::now(); while session.current_state() != ggrs::SessionState::Running { diff --git a/src/retroachievements.rs b/src/retroachievements.rs index 0f225410..ee81cbf6 100644 --- a/src/retroachievements.rs +++ b/src/retroachievements.rs @@ -150,10 +150,9 @@ pub fn load_game( let tx_ptr = Box::into_raw(Box::new(tx)) as *mut std::ffi::c_void; ra_load_game(rom.as_ptr(), rom_size, tx_ptr); }; - let join_handle = std::thread::spawn(move || { + tokio::task::block_in_place(move || { rx.blocking_recv().unwrap(); }); - join_handle.join().unwrap(); let mut c_title = std::ptr::null(); let mut c_image_url = std::ptr::null();