preserve-emulation/gopher64
1
0
Fork 0
mirror of https://github.com/gopher64/gopher64.git synced 2026-07-11 01:25:20 +02:00
Code Issues Packages Projects Releases Wiki Activity

Use mpsc instead of mutex (#1065)

Browse Source 
* use mpsc

* more

* more

* more
This commit is contained in:
Logan McNaughton
2026-06-12 16:16:06 +02:00
committed by GitHub
parent 4cc85d595f
commit be74e3358a
1 changed files with 25 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/netplay.rs
+25 -23
View File
@@ -11,24 +11,21 @@ impl ggrs::Config for GgrsConfig {
type Address = matchbox_socket::PeerId;
}
static DISCONNECTED_PEERS: std::sync::LazyLock<
std::sync::Mutex<rustc_hash::FxHashSet<matchbox_socket::PeerId>>,
> = std::sync::LazyLock::new(|| std::sync::Mutex::new(rustc_hash::FxHashSet::default()));
pub struct MatchboxChannel(matchbox_socket::WebRtcChannel);
pub struct MatchboxChannel {
channel: matchbox_socket::WebRtcChannel,
disconnected_peers: tokio::sync::mpsc::Sender<matchbox_socket::PeerId>,
}
impl ggrs::NonBlockingSocket<matchbox_socket::PeerId> for MatchboxChannel {
fn send_to(&mut self, msg: &ggrs::Message, addr: &matchbox_socket::PeerId) {
let encoded = postcard::to_stdvec(msg).expect("serialization failed");
if let Err(_) = self.0.try_send(encoded.into(), *addr)
&& let Ok(mut disconnected_peers) = DISCONNECTED_PEERS.lock()
{
disconnected_peers.insert(*addr);
if self.channel.try_send(encoded.into(), *addr).is_err() {
self.disconnected_peers.try_send(*addr).unwrap();
}
}
fn receive_all_messages(&mut self) -> Vec<(matchbox_socket::PeerId, ggrs::Message)> {
self.0
self.channel
.receive()
.iter()
.filter_map(|(peer, packet)| {
@@ -51,6 +48,7 @@ pub struct Netplay {
pub inputs: Vec<(ui::input::InputData, ggrs::InputStatus)>,
pub requests: std::collections::VecDeque<ggrs::GgrsRequest<GgrsConfig>>,
pub incoming_message: Vec<u8>,
pub disconnected_peers: tokio::sync::mpsc::Receiver<matchbox_socket::PeerId>,
}
#[derive(serde::Serialize, serde::Deserialize)]
@@ -238,17 +236,15 @@ 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,
);
}
while let Ok(addr) = netplay.disconnected_peers.try_recv() {
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,
);
}
}
}
@@ -360,7 +356,11 @@ pub fn init(
}
});
let matchbox_channel = MatchboxChannel(socket.take_channel(0).unwrap());
let (disconnected_peers_tx, disconnected_peers_rx) = tokio::sync::mpsc::channel(100);
let matchbox_channel = MatchboxChannel {
channel: socket.take_channel(0).unwrap(),
disconnected_peers: disconnected_peers_tx,
};
let mut reliable_channel = socket.take_channel(1).unwrap();
let now = std::time::Instant::now();
let timeout = std::time::Duration::from_secs(10);
@@ -413,7 +413,8 @@ pub fn init(
}
}
if matchbox_channel.0.config().max_retransmits != Some(0) || matchbox_channel.0.config().ordered
if matchbox_channel.channel.config().max_retransmits != Some(0)
|| matchbox_channel.channel.config().ordered
{
eprintln!("Sending GGRS traffic over reliable channel");
}
@@ -431,6 +432,7 @@ pub fn init(
}
Some(Netplay {
disconnected_peers: disconnected_peers_rx,
incoming_message: vec![],
input_delay,
session,