Netplay input sync (#169)

* Netplay sync inputs

* use wrapping add

* non blocking

* more

* more

* more

* more

* more

* more

* more

* more

* more

* more

* more

* more

* message

* more

* close

* fixes

* redirect rumble

* more

* fix

* more

* messages
This commit is contained in:
Logan McNaughton 2025-01-22 21:31:58 +01:00 committed by GitHub
parent 9257e8c66d
commit 195412a48e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 288 additions and 16 deletions

View File

@ -164,7 +164,7 @@ pub struct Device {
pub netplay: Option<netplay::Netplay>,
pub ui: ui::Ui,
byte_swap: usize,
cpu: cpu::Cpu,
pub cpu: cpu::Cpu,
pif: pif::Pif,
cart: cart_rom::Cart,
memory: memory::Memory,

View File

@ -1,4 +1,5 @@
use crate::device;
use crate::netplay;
use crate::ui;
pub mod mempak;
@ -52,7 +53,17 @@ pub fn process(device: &mut device::Device, channel: usize) {
}
JCMD_CONTROLLER_READ => {
let offset = device.pif.channels[channel].rx_buf.unwrap();
let input = ui::input::get(&mut device.ui, channel);
let input = if device.netplay.is_none() {
ui::input::get(&mut device.ui, channel)
} else {
if device.netplay.as_ref().unwrap().player_number as usize == channel {
let local_input = ui::input::get(&mut device.ui, 0);
netplay::send_input(device.netplay.as_ref().unwrap(), local_input);
}
netplay::get_input(device, channel)
};
device.pif.ram[offset..offset + 4].copy_from_slice(&input.0.to_ne_bytes());
if input.1 {
// pak change button pressed
@ -151,7 +162,12 @@ fn pak_switch_event(device: &mut device::Device) {
for (i, channel) in device.pif.channels.iter_mut().enumerate() {
if channel.change_pak != PakType::None {
if channel.change_pak == PakType::RumblePak {
device::ui::input::set_rumble(&mut device.ui, i, 0);
if device.netplay.is_none() {
device::ui::input::set_rumble(&mut device.ui, i, 0);
} else if device.netplay.as_ref().unwrap().player_number as usize == i {
device::ui::input::set_rumble(&mut device.ui, 0, 0);
}
let handler = device::controller::PakHandler {
read: device::controller::mempak::read,
write: device::controller::mempak::write,

View File

@ -15,6 +15,10 @@ pub fn read(device: &mut device::Device, _channel: usize, address: u16, data: us
pub fn write(device: &mut device::Device, channel: usize, address: u16, data: usize, size: usize) {
if address == 0xc000 {
let rumble = device.pif.ram[data + size - 1];
device::ui::input::set_rumble(&mut device.ui, channel, rumble);
if device.netplay.is_none() {
device::ui::input::set_rumble(&mut device.ui, channel, rumble);
} else if device.netplay.as_ref().unwrap().player_number as usize == channel {
device::ui::input::set_rumble(&mut device.ui, 0, rumble);
}
}
}

View File

@ -100,6 +100,7 @@ pub fn update_pif_ram(device: &mut device::Device) -> u64 {
for k in 0..PIF_CHANNELS_COUNT {
active_channels += process_channel(device, k)
}
(24000 + (active_channels * 30000)) as u64
}
@ -232,7 +233,12 @@ pub fn init(device: &mut device::Device) {
};
for i in 0..4 {
if device.ui.config.input.controller_enabled[i] {
if device.netplay.is_none() {
if device.ui.config.input.controller_enabled[i] {
device.pif.channels[i].pak_handler = Some(mempak_handler);
device.pif.channels[i].process = Some(device::controller::process);
}
} else if device.netplay.as_ref().unwrap().player_data[i].reg_id != 0 {
device.pif.channels[i].pak_handler = Some(mempak_handler);
device.pif.channels[i].process = Some(device::controller::process);
}

View File

@ -1,4 +1,5 @@
use crate::device;
use crate::netplay;
use crate::ui;
use governor::clock::Clock;
@ -124,8 +125,14 @@ fn vertical_interrupt_event(device: &mut device::Device) {
}
*/
let mut enable_speed_limiter = true;
if device.netplay.is_some() {
netplay::send_sync_check(device);
enable_speed_limiter = !device.netplay.as_ref().unwrap().fast_forward;
}
device.vi.vi_counter += 1;
if device.vi.vi_counter % LIMIT_BUFFER == 0 {
if device.vi.vi_counter % LIMIT_BUFFER == 0 && enable_speed_limiter {
speed_limiter(device);
}

View File

@ -1,12 +1,13 @@
use crate::device;
use crate::ui;
use eframe::egui;
use std::io::{Read, Write};
//UDP packet formats
//const UDP_SEND_KEY_INFO: u8 = 0;
//const UDP_RECEIVE_KEY_INFO: u8 = 1;
//const UDP_REQUEST_KEY_INFO: u8 = 2;
//const UDP_RECEIVE_KEY_INFO_GRATUITOUS: u8 = 3;
//const UDP_SYNC_DATA: u8 = 4;
const UDP_SEND_KEY_INFO: u8 = 0;
const UDP_RECEIVE_KEY_INFO: u8 = 1;
const UDP_REQUEST_KEY_INFO: u8 = 2;
const UDP_RECEIVE_KEY_INFO_GRATUITOUS: u8 = 3;
const UDP_SYNC_DATA: u8 = 4;
//TCP packet formats
const TCP_SEND_SAVE: u8 = 1;
@ -18,9 +19,34 @@ const TCP_GET_REGISTRATION: u8 = 6;
const TCP_DISCONNECT_NOTICE: u8 = 7;
pub struct Netplay {
// udp_socket: std::net::UdpSocket,
pub tcp_stream: std::net::TcpStream,
udp_socket: std::net::UdpSocket,
tcp_stream: std::net::TcpStream,
pub player_number: u8,
pub player_data: [PlayerData; 4],
vi_counter: u32,
status: u8,
buffer_target: u8,
pub fast_forward: bool,
pub error_notifier: tokio::sync::mpsc::Sender<String>,
pub gui_ctx: egui::Context,
}
pub struct PlayerData {
lag: u8,
count: u32,
pub reg_id: u32,
input_events: std::collections::HashMap<u32, InputEvent>,
}
struct InputEvent {
input: u32,
plugin: u8,
}
fn send_error(netplay: &mut Netplay, error: String) {
netplay.error_notifier.try_send(error).unwrap();
netplay.gui_ctx.request_repaint(); // this is so the window pops up right away
}
pub fn send_save(netplay: &mut Netplay, save_type: &str, save_data: &[u8], size: usize) {
@ -46,10 +72,151 @@ pub fn receive_save(netplay: &mut Netplay, save_type: &str, save_data: &mut Vec<
*save_data = response;
}
pub fn send_sync_check(device: &mut device::Device) {
let netplay = device.netplay.as_mut().unwrap();
if netplay.vi_counter % 600 == 0 {
let mut request: Vec<u8> = [UDP_SYNC_DATA].to_vec();
request.extend_from_slice(&(netplay.vi_counter).to_be_bytes());
for i in 0..device::cop0::COP0_REGS_COUNT as usize {
request.extend_from_slice(&(device.cpu.cop0.regs[i] as u32).to_be_bytes());
}
netplay.udp_socket.send(&request).unwrap();
}
netplay.vi_counter = netplay.vi_counter.wrapping_add(1);
}
pub fn send_input(netplay: &Netplay, input: (u32, bool)) {
let mut request: Vec<u8> = [UDP_SEND_KEY_INFO].to_vec();
request.push(netplay.player_number);
request.extend_from_slice(
&(netplay.player_data[netplay.player_number as usize].count).to_be_bytes(),
);
request.extend_from_slice(&(input.0).to_be_bytes());
request.push(input.1 as u8);
netplay.udp_socket.send(&request).unwrap();
}
pub fn get_input(device: &mut device::Device, channel: usize) -> (u32, bool) {
let netplay = device.netplay.as_mut().unwrap();
let mut input = None;
let timeout = std::time::Instant::now() + std::time::Duration::from_secs(10);
let mut request_timer = std::time::Instant::now() + std::time::Duration::from_millis(5);
request_input(netplay, channel);
while input.is_none() {
if std::time::Instant::now() > request_timer {
// sends a request packet every 5ms
request_input(netplay, channel);
request_timer = std::time::Instant::now() + std::time::Duration::from_millis(5);
}
process_incoming(netplay);
input = netplay.player_data[channel]
.input_events
.remove(&netplay.player_data[channel].count);
if std::time::Instant::now() > timeout {
send_error(
netplay,
"Timed out waiting for input. Lost connection to server".to_string(),
);
input = Some(InputEvent {
input: 0,
plugin: 0,
});
device.cpu.running = false;
}
}
netplay.fast_forward = netplay.player_data[channel].lag > 0
&& netplay.player_data[channel].input_events.len() as u8 > netplay.buffer_target;
netplay.player_data[channel].count = netplay.player_data[channel].count.wrapping_add(1);
(
input.as_ref().unwrap().input,
input.as_ref().unwrap().plugin != 0,
)
}
fn request_input(netplay: &Netplay, channel: usize) {
let mut request: Vec<u8> = [UDP_REQUEST_KEY_INFO].to_vec();
request.push(channel as u8); //The player we need input for
request.extend_from_slice(
&(netplay.player_data[netplay.player_number as usize].reg_id).to_be_bytes(),
);
request.extend_from_slice(&(netplay.player_data[channel].count).to_be_bytes());
request.push(0); //spectator mode
request.push(netplay.player_data[channel].input_events.len() as u8);
netplay.udp_socket.send(&request).unwrap();
}
fn process_incoming(netplay: &mut Netplay) {
let mut buf: [u8; 1024] = [0; 1024];
while let Ok(_incoming) = netplay.udp_socket.recv(&mut buf) {
match buf[0] {
UDP_RECEIVE_KEY_INFO | UDP_RECEIVE_KEY_INFO_GRATUITOUS => {
let player = buf[1] as usize;
//current_status is a status update from the server
//it will let us know if another player has disconnected, or the games have desynced
let current_status = buf[2];
if buf[0] == UDP_RECEIVE_KEY_INFO {
netplay.player_data[player].lag = buf[3];
}
if current_status != netplay.status {
if ((current_status & 0x1) ^ (netplay.status & 0x1)) != 0 {
send_error(
netplay,
format!("Players have desynced at VI {}", netplay.vi_counter),
);
}
for dis in 1..5 {
if ((current_status & (0x1 << dis)) ^ (netplay.status & (0x1 << dis))) != 0
{
send_error(netplay, format!("Player {} has disconnected", dis));
}
}
netplay.status = current_status;
}
let mut buffer_offset = 5;
for _i in 0..buf[4] {
let count = u32::from_be_bytes(
buf[buffer_offset..buffer_offset + 4].try_into().unwrap(),
);
buffer_offset += 4;
if (count.wrapping_sub(netplay.player_data[player].count)) > (u32::MAX / 2) {
//event doesn't need to be recorded
buffer_offset += 5;
continue;
}
let input = u32::from_be_bytes(
buf[buffer_offset..buffer_offset + 4].try_into().unwrap(),
);
buffer_offset += 4;
let plugin = buf[buffer_offset];
buffer_offset += 1;
let input_event = InputEvent { input, plugin };
netplay.player_data[player]
.input_events
.insert(count, input_event);
}
}
_ => {
panic! {"unknown UDP packet"}
}
}
}
}
pub fn init(
mut peer_addr: std::net::SocketAddr,
session: ui::gui::gui_netplay::NetplayRoom,
player_number: u8,
error_notifier: tokio::sync::mpsc::Sender<String>,
gui_ctx: egui::Context,
) -> Netplay {
peer_addr.set_port(session.port.unwrap() as u16);
let udp_socket = if peer_addr.is_ipv4() {
@ -60,6 +227,7 @@ pub fn init(
.expect("couldn't bind to address")
};
udp_socket.connect(peer_addr).unwrap();
udp_socket.set_nonblocking(true).unwrap();
let mut stream = std::net::TcpStream::connect(peer_addr).unwrap();
@ -83,7 +251,7 @@ pub fn init(
if response[0] != 1 {
panic!("Failed to register player");
}
let _buffer_target = response[1];
let buffer_target = response[1];
let request: [u8; 1] = [TCP_GET_REGISTRATION];
stream.write_all(&request).unwrap();
@ -96,9 +264,41 @@ pub fn init(
reg_id[i] = u32::from_be_bytes(response[(i * 6)..(i * 6) + 4].try_into().unwrap());
}
Netplay {
// udp_socket,
udp_socket,
tcp_stream: stream,
player_number,
vi_counter: 0,
status: 0,
buffer_target,
fast_forward: false,
error_notifier,
gui_ctx,
player_data: [
PlayerData {
lag: 0,
count: 0,
reg_id: reg_id[0],
input_events: std::collections::HashMap::new(),
},
PlayerData {
lag: 0,
count: 0,
reg_id: reg_id[1],
input_events: std::collections::HashMap::new(),
},
PlayerData {
lag: 0,
count: 0,
reg_id: reg_id[2],
input_events: std::collections::HashMap::new(),
},
PlayerData {
lag: 0,
count: 0,
reg_id: reg_id[3],
input_events: std::collections::HashMap::new(),
},
],
}
}
@ -109,4 +309,9 @@ pub fn close(device: &mut device::Device) {
request[1..5].copy_from_slice(&regid.to_be_bytes());
netplay.tcp_stream.write_all(&request).unwrap();
netplay.tcp_stream.flush().unwrap();
netplay
.tcp_stream
.shutdown(std::net::Shutdown::Both)
.unwrap();
}

View File

@ -22,6 +22,7 @@ pub struct GopherEguiApp {
dinput: bool,
show_vru_dialog: bool,
vru_window_receiver: Option<tokio::sync::mpsc::Receiver<Vec<String>>>,
netplay_error_receiver: Option<tokio::sync::mpsc::Receiver<String>>,
vru_word_notifier: Option<tokio::sync::mpsc::Sender<String>>,
vru_word_list: Vec<String>,
pub netplay: gui_netplay::GuiNetplay,
@ -105,6 +106,7 @@ impl GopherEguiApp {
emulate_vru: game_ui.config.input.emulate_vru,
show_vru_dialog: false,
dinput: false,
netplay_error_receiver: None,
vru_window_receiver: None,
vru_word_notifier: None,
vru_word_list: Vec::new(),
@ -261,6 +263,11 @@ pub fn open_rom(app: &mut GopherEguiApp, ctx: &egui::Context) {
player_number = Some(app.netplay.player_number);
}
let (netplay_error_notifier, netplay_error_receiver): (
tokio::sync::mpsc::Sender<String>,
tokio::sync::mpsc::Receiver<String>,
) = tokio::sync::mpsc::channel(8);
let (vru_window_notifier, vru_window_receiver): (
tokio::sync::mpsc::Sender<Vec<String>>,
tokio::sync::mpsc::Receiver<Vec<String>>,
@ -271,6 +278,9 @@ pub fn open_rom(app: &mut GopherEguiApp, ctx: &egui::Context) {
tokio::sync::mpsc::Receiver<String>,
) = tokio::sync::mpsc::channel(1);
if netplay {
app.netplay_error_receiver = Some(netplay_error_receiver);
}
if emulate_vru && !netplay {
app.vru_window_receiver = Some(vru_window_receiver);
app.vru_word_notifier = Some(vru_word_notifier);
@ -312,6 +322,8 @@ pub fn open_rom(app: &mut GopherEguiApp, ctx: &egui::Context) {
peer_addr.unwrap(),
session.unwrap(),
player_number.unwrap(),
netplay_error_notifier,
gui_ctx,
));
device::run_game(rom_contents, data_dir, &mut device, fullscreen);
netplay::close(&mut device);
@ -351,6 +363,13 @@ impl eframe::App for GopherEguiApp {
gui_netplay::netplay_wait(self, ctx);
}
if self.netplay_error_receiver.is_some() {
let result = self.netplay_error_receiver.as_mut().unwrap().try_recv();
if result.is_ok() {
self.netplay.error = result.unwrap();
}
}
if !self.netplay.error.is_empty() {
gui_netplay::netplay_error(self, ctx, self.netplay.error.clone());
}

View File

@ -347,6 +347,9 @@ pub fn netplay_create(app: &mut GopherEguiApp, ctx: &egui::Context) {
ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| {
if ui.button("Close").clicked() {
let socket = app.netplay.socket.as_mut().unwrap();
socket.close(None).unwrap();
while socket.read().is_ok() {}
app.netplay = Default::default();
};
})
@ -542,6 +545,9 @@ pub fn netplay_join(app: &mut GopherEguiApp, ctx: &egui::Context) {
);
ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| {
if ui.button("Close").clicked() {
let socket = app.netplay.socket.as_mut().unwrap();
socket.close(None).unwrap();
while socket.read().is_ok() {}
app.netplay = Default::default();
};
if app.netplay.join_rom_label.is_empty() {
@ -647,8 +653,9 @@ pub fn netplay_wait(app: &mut GopherEguiApp, ctx: &egui::Context) {
port: app.netplay.waiting_session.as_ref().unwrap().port,
}),
};
let socket = app.netplay.socket.as_mut().unwrap();
if !app.netplay.socket_waiting {
let socket = app.netplay.socket.as_mut().unwrap();
socket
.send(tungstenite::Message::Binary(tungstenite::Bytes::from(
serde_json::to_vec(&motd_message).unwrap(),
@ -684,6 +691,7 @@ pub fn netplay_wait(app: &mut GopherEguiApp, ctx: &egui::Context) {
port: app.netplay.waiting_session.as_ref().unwrap().port,
}),
};
let socket = app.netplay.socket.as_mut().unwrap();
socket
.send(tungstenite::Message::Binary(tungstenite::Bytes::from(
serde_json::to_vec(&begin_game).unwrap(),
@ -715,6 +723,7 @@ pub fn netplay_wait(app: &mut GopherEguiApp, ctx: &egui::Context) {
}),
};
app.netplay.chat_message.clear();
let socket = app.netplay.socket.as_mut().unwrap();
socket
.send(tungstenite::Message::Binary(tungstenite::Bytes::from(
serde_json::to_vec(&send_chat).unwrap(),
@ -724,6 +733,7 @@ pub fn netplay_wait(app: &mut GopherEguiApp, ctx: &egui::Context) {
}
if app.netplay.socket_waiting {
let socket = app.netplay.socket.as_mut().unwrap();
let data = socket.read();
if data.is_ok() {
let message: NetplayMessage =
@ -748,6 +758,8 @@ pub fn netplay_wait(app: &mut GopherEguiApp, ctx: &egui::Context) {
}
}
app.netplay.player_number = player as u8;
socket.close(None).unwrap();
while socket.read().is_ok() {}
gui::open_rom(app, ctx);
app.netplay = Default::default();
return;
@ -835,6 +847,9 @@ pub fn netplay_wait(app: &mut GopherEguiApp, ctx: &egui::Context) {
}
ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| {
if ui.button("Close").clicked() {
let socket = app.netplay.socket.as_mut().unwrap();
socket.close(None).unwrap();
while socket.read().is_ok() {}
app.netplay = Default::default();
};
});