diff --git a/src/device.rs b/src/device.rs index 345994f..abfd524 100644 --- a/src/device.rs +++ b/src/device.rs @@ -164,7 +164,7 @@ pub struct Device { pub netplay: Option, pub ui: ui::Ui, byte_swap: usize, - cpu: cpu::Cpu, + pub cpu: cpu::Cpu, pif: pif::Pif, cart: cart_rom::Cart, memory: memory::Memory, diff --git a/src/device/controller.rs b/src/device/controller.rs index d5af046..7bb2b22 100644 --- a/src/device/controller.rs +++ b/src/device/controller.rs @@ -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, diff --git a/src/device/controller/rumble.rs b/src/device/controller/rumble.rs index 6356507..36d5aa5 100644 --- a/src/device/controller/rumble.rs +++ b/src/device/controller/rumble.rs @@ -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); + } } } diff --git a/src/device/pif.rs b/src/device/pif.rs index d94765f..8bab677 100644 --- a/src/device/pif.rs +++ b/src/device/pif.rs @@ -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); } diff --git a/src/device/vi.rs b/src/device/vi.rs index ccb7964..2915317 100644 --- a/src/device/vi.rs +++ b/src/device/vi.rs @@ -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); } diff --git a/src/netplay.rs b/src/netplay.rs index 2255e9c..050de35 100644 --- a/src/netplay.rs +++ b/src/netplay.rs @@ -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, + pub gui_ctx: egui::Context, +} + +pub struct PlayerData { + lag: u8, + count: u32, + pub reg_id: u32, + input_events: std::collections::HashMap, +} + +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 = [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 = [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 = [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, + 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(®id.to_be_bytes()); netplay.tcp_stream.write_all(&request).unwrap(); + netplay.tcp_stream.flush().unwrap(); + netplay + .tcp_stream + .shutdown(std::net::Shutdown::Both) + .unwrap(); } diff --git a/src/ui/gui.rs b/src/ui/gui.rs index 8303584..61fbb3e 100644 --- a/src/ui/gui.rs +++ b/src/ui/gui.rs @@ -22,6 +22,7 @@ pub struct GopherEguiApp { dinput: bool, show_vru_dialog: bool, vru_window_receiver: Option>>, + netplay_error_receiver: Option>, vru_word_notifier: Option>, vru_word_list: Vec, 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, + tokio::sync::mpsc::Receiver, + ) = tokio::sync::mpsc::channel(8); + let (vru_window_notifier, vru_window_receiver): ( tokio::sync::mpsc::Sender>, tokio::sync::mpsc::Receiver>, @@ -271,6 +278,9 @@ pub fn open_rom(app: &mut GopherEguiApp, ctx: &egui::Context) { tokio::sync::mpsc::Receiver, ) = 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()); } diff --git a/src/ui/gui/gui_netplay.rs b/src/ui/gui/gui_netplay.rs index 54a57ec..b0ea54b 100644 --- a/src/ui/gui/gui_netplay.rs +++ b/src/ui/gui/gui_netplay.rs @@ -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(); }; });