diff --git a/rc_core/src/persist/combat.rs b/rc_core/src/persist/combat.rs index 1cc9ca9..e68e832 100644 --- a/rc_core/src/persist/combat.rs +++ b/rc_core/src/persist/combat.rs @@ -619,7 +619,8 @@ fn default_multiplayer() -> super::MultiplayerConfig { super::MultiplayerConfig { players_per_game: 2, enabled: true, - autostart_after_s: 180, + autostart_after_s: super::multiplayer::default_match_autostart_after_s(), + continue_loading_after_s: super::multiplayer::default_continue_loading_after_s(), network: super::multiplayer::default_net_conf(), fakes: super::multiplayer::default_fake_users(), filler: super::multiplayer::default_filler_users(), diff --git a/rc_core/src/persist/config/cubes_json.rs b/rc_core/src/persist/config/cubes_json.rs index e27187f..7faeb42 100644 --- a/rc_core/src/persist/config/cubes_json.rs +++ b/rc_core/src/persist/config/cubes_json.rs @@ -401,12 +401,20 @@ impl super::ConfigProvider for CubeConfig { self.battle.multiplayer.players_per_game } - fn is_multiplayer_enabled(&self) -> bool { + /*fn is_multiplayer_enabled(&self) -> bool { self.battle.multiplayer.enabled } fn multiplayer_autostart_after(&self) -> std::time::Duration { std::time::Duration::from_secs(self.battle.multiplayer.autostart_after_s) + }*/ + + fn multiplayer_settings(&self) -> super::MultiplayerSettings { + super::MultiplayerSettings { + is_enabled: self.battle.multiplayer.enabled, + lobby_autostart_after: self.battle.multiplayer.autostart_after_s.map(std::time::Duration::from_secs), + loading_autostart_after: self.battle.multiplayer.continue_loading_after_s.map(std::time::Duration::from_secs), + } } fn network_config(&self) -> crate::persist::NetworkConf { diff --git a/rc_core/src/persist/config/mod.rs b/rc_core/src/persist/config/mod.rs index affcb4b..634b892 100644 --- a/rc_core/src/persist/config/mod.rs +++ b/rc_core/src/persist/config/mod.rs @@ -2,7 +2,7 @@ mod cubes_json; pub use cubes_json::CubeConfig; mod traits; -pub use traits::{ConfigProvider, DevMessageProvider, ServerConfig, GarageUpgrades, GarageUpgradeIncrement, ChatSystemConfig, GameEventSequence, GameEvents, GameRotationStrategy, GameEvent, GameMap, GameVisibility, GameType, SingleplayerConfig, VehicleInfo, VehicleDescriptor, QueueChangeMode, Point, Sphere, MapConfig, LinksConfig, FakePlayer, ClientEmulator, EnergyConfig, BattleArenaResolver, PitSettings, PitWinCondition, TeamDeathMatchSettings, ShopEntriesResolver, ShopAction, ShopGain, PromoCode}; +pub use traits::{ConfigProvider, DevMessageProvider, ServerConfig, GarageUpgrades, GarageUpgradeIncrement, ChatSystemConfig, GameEventSequence, GameEvents, GameRotationStrategy, GameEvent, GameMap, GameVisibility, GameType, SingleplayerConfig, VehicleInfo, VehicleDescriptor, QueueChangeMode, Point, Sphere, MapConfig, LinksConfig, FakePlayer, ClientEmulator, EnergyConfig, BattleArenaResolver, PitSettings, PitWinCondition, TeamDeathMatchSettings, ShopEntriesResolver, ShopAction, ShopGain, PromoCode, MultiplayerSettings}; mod validation; pub use validation::{SelfValidator, ValidationInfo, ValidationMessage}; diff --git a/rc_core/src/persist/config/traits.rs b/rc_core/src/persist/config/traits.rs index 644520b..b522d20 100644 --- a/rc_core/src/persist/config/traits.rs +++ b/rc_core/src/persist/config/traits.rs @@ -27,8 +27,7 @@ pub trait ConfigProvider { fn gamemodes(&self) -> crate::data::game_mode::GameModeConfigs; // FIXME fn singleplayer_details(&self) -> SingleplayerConfig; fn players_per_game(&self) -> usize; - fn is_multiplayer_enabled(&self) -> bool; - fn multiplayer_autostart_after(&self) -> std::time::Duration; + fn multiplayer_settings(&self) -> MultiplayerSettings; // FIXME don't use serializable types in traits fn network_config(&self) -> crate::persist::NetworkConf; fn maps(&self) -> std::collections::HashMap; @@ -502,3 +501,10 @@ pub struct PromoCode { pub value: f32, pub transaction: ShopAction, } + +#[derive(Debug)] +pub struct MultiplayerSettings { + pub is_enabled: bool, + pub lobby_autostart_after: Option, + pub loading_autostart_after: Option, +} diff --git a/rc_core/src/persist/multiplayer.rs b/rc_core/src/persist/multiplayer.rs index e612ea1..ba5756e 100644 --- a/rc_core/src/persist/multiplayer.rs +++ b/rc_core/src/persist/multiplayer.rs @@ -4,7 +4,10 @@ use serde::{Serialize, Deserialize}; pub struct MultiplayerConfig { pub players_per_game: usize, pub enabled: bool, - pub autostart_after_s: u64, + #[serde(default = "default_match_autostart_after_s")] + pub autostart_after_s: Option, + #[serde(default = "default_continue_loading_after_s")] + pub continue_loading_after_s: Option, #[serde(default = "default_net_conf")] pub network: NetworkConf, #[serde(default = "default_fake_users")] @@ -73,6 +76,14 @@ pub struct NetworkConf { pub max_bulk_resend: u8, } +pub(super) fn default_match_autostart_after_s() -> Option { + Some(180) +} + +pub(super) fn default_continue_loading_after_s() -> Option { + Some(30) +} + pub(super) fn default_net_conf() -> NetworkConf { NetworkConf { network_channel_ty: "3113".to_owned(), diff --git a/rc_lobby_room/src/lobby.rs b/rc_lobby_room/src/lobby.rs index f042c4b..bdcb631 100644 --- a/rc_lobby_room/src/lobby.rs +++ b/rc_lobby_room/src/lobby.rs @@ -55,17 +55,18 @@ pub struct QueueHandler { cpu_counter: std::sync::Arc, weapon_guesser: std::sync::Arc, change_strategy: GamemodeChangeStrategy, - autostart_after: std::time::Duration, + autostart_after: Option, autostart_task_started: std::sync::atomic::AtomicBool, } impl QueueHandler { pub fn new(conf: &oj_rc_core::ConfigImpl, game_host: &str, factory: std::sync::Arc, cpu_counter: std::sync::Arc, weapon_guesser: std::sync::Arc,) -> Self { let (domain, port_str) = game_host.split_once(':').expect("Invalid redirect address (must be domain:port)"); + let mp_settings = oj_rc_core::ConfigProvider::<()>::multiplayer_settings(conf); Self { users_in_queue: std::sync::Arc::new(tokio::sync::Mutex::new(HashMap::new())), users_per_game: oj_rc_core::ConfigProvider::<()>::players_per_game(conf), - is_enabled: oj_rc_core::ConfigProvider::<()>::is_multiplayer_enabled(conf), + is_enabled: mp_settings.is_enabled, hostname: domain.to_owned(), hostport: port_str.parse().expect("Invalid redirect port"), network_conf: crate::data::network::NetworkConfigData::from_conf(oj_rc_core::ConfigProvider::<()>::network_config(conf)), @@ -73,13 +74,13 @@ impl QueueHandler { cpu_counter, weapon_guesser, change_strategy: GamemodeChangeStrategy::from_core(>::server_config(conf).queue_mode), - autostart_after: oj_rc_core::ConfigProvider::<()>::multiplayer_autostart_after(conf), + autostart_after: mp_settings.lobby_autostart_after, autostart_task_started: std::sync::atomic::AtomicBool::new(false), } } fn ensure_autostart_task_running(&self) { - if self.autostart_task_started.swap(true, std::sync::atomic::Ordering::AcqRel) { + if self.autostart_task_started.swap(true, std::sync::atomic::Ordering::AcqRel) || self.autostart_after.is_none() { return; } @@ -91,7 +92,7 @@ impl QueueHandler { let factory = self.factory.clone(); let cpu_counter = self.cpu_counter.clone(); let weapon_guesser = self.weapon_guesser.clone(); - let autostart_after = self.autostart_after; + let autostart_after = self.autostart_after.unwrap(); tokio::spawn(async move { loop { diff --git a/rc_multiplayer/src/handler.rs b/rc_multiplayer/src/handler.rs index a68c9fb..e396221 100644 --- a/rc_multiplayer/src/handler.rs +++ b/rc_multiplayer/src/handler.rs @@ -29,7 +29,7 @@ impl literustlib_server::EventHandler for LnlEventHandler { type UserData = super::UserData; async fn on_receive(&self, data: Self::PacketData, header: &literustlib::packet::Header, peer: &std::sync::Arc< literustlib_server::Connection>, user: &Self::UserData, sender: &std::sync::Arc>) { - log::debug!("Got message {:?} (len: {}) from connection id {}", data.message_ty, data.data.len(), peer.id()); + log::trace!("Got message {:?} (len: {}) from connection id {}", data.message_ty, data.data.len(), peer.id()); match data.message_ty { crate::data::MessageType::ClientMsg => { if let Some(handler) = self.event_handlers.get(&data.variant) { @@ -47,7 +47,7 @@ impl literustlib_server::EventHandler for LnlEventHandler { } }, crate::data::MessageType::ServerMsg => { - log::debug!("Got message from server but I'm the server??? (ignoring)"); + log::warn!("Got message from server but I'm the server??? (ignoring) peer:{}", peer.id()); }, crate::data::MessageType::RobotMotion => { self.motion_handler.handle(&data.data, user).await; @@ -116,7 +116,7 @@ impl EventData { impl literustlib::packet::PacketData for EventData { fn parse(bytes: bytes::Bytes, _header: &literustlib::packet::Header) -> std::io::Result { - log::debug!("Got packet data ({}) {:?}", bytes.len(), &bytes[..]); + log::trace!("Got packet data ({}) {:?}", bytes.len(), &bytes[..]); if bytes.len() >= 6 { let data = bytes.slice(6..); let net_message_num = i16::from_le_bytes([bytes[0], bytes[1]]); diff --git a/rc_multiplayer/src/matches/aggregate.rs b/rc_multiplayer/src/matches/aggregate.rs index 8f0a607..22fa4b4 100644 --- a/rc_multiplayer/src/matches/aggregate.rs +++ b/rc_multiplayer/src/matches/aggregate.rs @@ -10,6 +10,7 @@ pub struct GameMatches { pit_settings: std::sync::Arc, tdm_settings: std::sync::Arc, factory: std::sync::Arc, + mp_settings: std::sync::Arc, } impl GameMatches { @@ -29,6 +30,7 @@ impl GameMatches { pit_settings: std::sync::Arc::new(>::pit_settings(conf)), tdm_settings: std::sync::Arc::new(>::tdm_settings(conf)), factory, + mp_settings: std::sync::Arc::new(>::multiplayer_settings(conf)), } } @@ -98,6 +100,7 @@ impl GameMatches { players, inner, fakes_handler, + self.mp_settings.clone(), ); Ok(engine.spawn()) }, @@ -130,6 +133,7 @@ impl GameMatches { players, inner, fakes_handler, + self.mp_settings.clone(), ); Ok(engine.spawn()) }, @@ -147,6 +151,7 @@ impl GameMatches { players, inner, fakes_handler, + self.mp_settings.clone(), ); Ok(engine.spawn()) }, @@ -164,6 +169,7 @@ impl GameMatches { players, inner, fakes_handler, + self.mp_settings.clone(), ); Ok(engine.spawn()) }, @@ -226,7 +232,7 @@ impl GameMatches { log::info!("Match message router has started"); while !rx.is_closed() { if let Some(msg) = rx.recv().await { - log::debug!("Match message router got a message"); + log::trace!("Match message router got a message"); match msg { super::GameMessage::NewConnection { user, game_guid, connection, response, sender } => { if let Some(tx) = self.matches.get(&game_guid) { diff --git a/rc_multiplayer/src/matches/generic.rs b/rc_multiplayer/src/matches/generic.rs index 3f23021..3a40afa 100644 --- a/rc_multiplayer/src/matches/generic.rs +++ b/rc_multiplayer/src/matches/generic.rs @@ -227,6 +227,33 @@ impl ConnectionMode { } } +#[repr(u8)] +#[derive(Debug, Copy, Clone)] +enum LoadingState { + Starting = 0, + InSync = 1, + InGame = 2, + End = 3, +} + +impl LoadingState { + #[inline] + pub(super) fn from_u8(num: u8) -> Self { + match num { + 0 => Self::Starting, + 1 => Self::InSync, + 2 => Self::InGame, + 3 => Self::End, + x => panic!("Unrecognized ConnectionMode {}", x), + } + } + + #[inline] + pub(super) fn to_u8(self) -> u8 { + self as u8 + } +} + struct UnclaimedStats { kills: tokio::sync::Mutex>>, } @@ -273,6 +300,9 @@ pub(super) struct GenericGamemodeEngine { //pub fake_users: std::collections::HashMap, pub fakes_handler: super::fake::Handler, unclaimed: UnclaimedStats, + loading_state: std::sync::Arc, + self_sender: Option>, + mp_config: std::sync::Arc, } impl GenericGamemodeEngine { @@ -286,6 +316,7 @@ impl GenericGamemodeEngine { players: Vec, custom: L, fakes_handler: super::fake::Handler, + mp_config: std::sync::Arc, ) -> Self { /*let fake_users = players.iter() @@ -312,6 +343,9 @@ impl GenericGamemodeEngine { custom_logic_handler: custom, fakes_handler, unclaimed: UnclaimedStats::new(), + loading_state: std::sync::Arc::new(std::sync::atomic::AtomicU8::new(LoadingState::Starting.to_u8())), + self_sender: None, + mp_config, } } @@ -446,8 +480,9 @@ impl GenericGamemodeEngine { } } - pub(super) fn spawn(self) -> tokio::sync::mpsc::Sender { + pub(super) fn spawn(mut self) -> tokio::sync::mpsc::Sender { let (tx, rx) = tokio::sync::mpsc::channel(super::CHANNEL_BOUND); + self.self_sender = Some(tx.clone()); tokio::spawn(self.run(rx)); tx } @@ -527,6 +562,9 @@ impl GenericGamemodeEngine { self.on_motion(user_id, motion).await; }, super::GameMessage::NoOp => {}, + super::GameMessage::LoadingTimeout { timeout, response } => { + self.on_timeout(timeout, response).await; + } } } } @@ -587,6 +625,7 @@ impl GenericGamemodeEngine { log::error!("Failed to mark player {} (user {}) as connected to game {}: {}", id, user_id, self.game_guid(), e); } let mut users = self.users.write().await; + let was_empty = users.is_empty(); users.insert(id, new_user.clone()); for fake_id in new_user.aliases.iter() { if let Some(player_desc) = self.user_descriptor(*fake_id) { @@ -597,6 +636,9 @@ impl GenericGamemodeEngine { log::warn!("Non-existent fake player id {} was encountered while connecting, ignoring", *fake_id); } } + if was_empty { + self.start_loading_sync_timeouter().await; + } } response.send(None).unwrap_or_default(); } @@ -740,7 +782,7 @@ impl GenericGamemodeEngine { } async fn on_request_loading_progress(&self, user_id: i32) { - log::info!("Got request loading progress"); + //log::info!("Got request loading progress"); if let Some(user_key) = self.user_key_by_user_id(user_id) { if let Some(user_info) = self.users.read().await.get(&user_key) { let mut client_ai_map = self.fakes_handler.get_client_ais().await; @@ -794,6 +836,58 @@ impl GenericGamemodeEngine { } } + async fn start_loading_sync_timeouter(&self) { + let game_state = self.loading_state.clone(); + let tx = self.self_sender.clone().unwrap(); + if let Some(loading_autostart_after) = self.mp_config.loading_autostart_after { + super::modes::trackers::Timeout::new(loading_autostart_after) + .on_timeout(move || { + // start sync + let tx_clone = tx.clone(); + let (tx_oneshot, rx_oneshot) = tokio::sync::oneshot::channel(); + tokio::spawn(async move { + tx_clone.send(crate::matches::GameMessage::LoadingTimeout { + timeout: crate::matches::messages::TimeoutVariant::WaitingForLoadingSync, + response: tx_oneshot, + }).await.unwrap_or_default(); + }); + rx_oneshot.blocking_recv().unwrap_or(true) + }) + .with_cancel_check(move || { + // check if sync is already started + let state = LoadingState::from_u8(game_state.load(std::sync::atomic::Ordering::Relaxed)); + !matches!(state, LoadingState::Starting) + }) + .start().await; + } + } + + async fn start_game_start_timeouter(&self) { + let game_state = self.loading_state.clone(); + let tx = self.self_sender.clone().unwrap(); + if let Some(loading_autostart_after) = self.mp_config.loading_autostart_after { + super::modes::trackers::Timeout::new(loading_autostart_after) + .on_timeout(move || { + // start sync + let tx_clone = tx.clone(); + let (tx_oneshot, rx_oneshot) = tokio::sync::oneshot::channel(); + tokio::spawn(async move { + tx_clone.send(crate::matches::GameMessage::LoadingTimeout { + timeout: crate::matches::messages::TimeoutVariant::WaitingForGameStart, + response: tx_oneshot, + }).await.unwrap_or_default(); + }); + rx_oneshot.blocking_recv().unwrap_or(true) + }) + .with_cancel_check(move || { + // check if sync is already started + let state = LoadingState::from_u8(game_state.load(std::sync::atomic::Ordering::Relaxed)); + !matches!(state, LoadingState::InSync) + }) + .start().await; + } + } + async fn on_request_loading_sync(&self, user_id: i32) { // wait for all users to be ready before transitioning to loading sync let mut ready_count = 0; @@ -805,7 +899,7 @@ impl GenericGamemodeEngine { log::warn!("Got RequestLoadingSync after user {} was already in/past WaitingForSync stage", user_id); continue; } - log::info!("User {} (player {}) is awaiting sync", user_id, player_id); + log::info!("User {} (player {}) is awaiting sync in game {}", user_id, player_id, self.game_guid()); user_desc.state.mode.store(ConnectionMode::WaitingForSync.to_u8(), std::sync::atomic::Ordering::Relaxed); ready_count += 1; } else if matches!(ConnectionMode::from_u8(user_desc.state.mode.load(std::sync::atomic::Ordering::Relaxed)), ConnectionMode::WaitingForSync) { @@ -813,17 +907,36 @@ impl GenericGamemodeEngine { } } let player_count = self.real_player_count(); - log::info!("Real players {}, ready players {}", player_count, ready_count); + //log::info!("Real players {}, ready players {}", player_count, ready_count); if ready_count == player_count { - log::info!("All players ({}) awaiting sync for game {}", player_count, self.game_guid()); - for (user_key, conn) in self.users.read().await.iter() { - let user_info = self.user_descriptor(*user_key).unwrap(); - let extra_packets = self.custom_logic_handler.extra_sync_events(self, conn, user_info).await; - self.spawn_send_sync_events(conn, conn.user.user_id(), *user_key, self.players_info(), extra_packets, self.map_config.clone()); - let user_desc = self.user_descriptor(*user_key).unwrap(); - user_desc.state.mode.store(ConnectionMode::Sync.to_u8(), std::sync::atomic::Ordering::Relaxed); + self.start_sync().await; + } + } + + async fn start_sync(&self) { + self.loading_state.store(LoadingState::InSync.to_u8(), std::sync::atomic::Ordering::Relaxed); + let ready_players = count_users_in_mode(ConnectionMode::WaitingForSync, self.descriptors.values()); + let player_count = self.real_player_count(); + log::info!("Starting sync for {}/{} players in game {}", ready_players, player_count, self.game_guid()); + let mut to_disconnect = Vec::with_capacity(player_count - ready_players); + for (user_key, conn) in self.users.read().await.iter() { + let user_info = self.user_descriptor(*user_key).unwrap(); + let extra_packets = self.custom_logic_handler.extra_sync_events(self, conn, user_info).await; + let user_desc = self.user_descriptor(*user_key).unwrap(); + self.spawn_send_sync_events(conn, user_desc.descriptor.user_id, *user_key, self.players_info(), extra_packets, self.map_config.clone()); + let old_mode = user_desc.state.mode.swap(ConnectionMode::Sync.to_u8(), std::sync::atomic::Ordering::Relaxed); + let old_mode = ConnectionMode::from_u8(old_mode); + if !matches!(old_mode, ConnectionMode::WaitingForSync) { + if let Some(user_id) = user_desc.descriptor.user_id { + to_disconnect.push(user_id); + //conn.connection.connection.goodbye(&conn.connection.sender).await; + } } } + for user_id in to_disconnect { + self.on_end_connection(user_id, false).await; + } + self.start_game_start_timeouter().await; } async fn on_load_complete(&self, user_id: i32) { @@ -847,31 +960,42 @@ impl GenericGamemodeEngine { return; } // wait for all users to be ready for starting game start countdown - let mut all_users_loading_complete = true; - for player_info in self.descriptors.values() { - if player_info.descriptor.user_id.is_none() { continue; } // skip non-user players - let mode = ConnectionMode::from_u8(player_info.state.mode.load(std::sync::atomic::Ordering::Relaxed)); - all_users_loading_complete &= matches!(mode, ConnectionMode::WaitingToStart); - } + let all_users_loading_complete = is_all_users_in_mode(ConnectionMode::WaitingToStart, self.descriptors.values()); // trigger game start if all_users_loading_complete { - let player_count = self.real_player_count(); - log::info!("All players ({}) are ready for game {}", player_count, self.game_guid()); - tokio::time::sleep(Self::END_OF_SYNC_DELAY).await; - self.fakes_handler.on_ready( - self.users.read().await.iter() - .map(|(id, real_player)| (*id, real_player.connection.clone())) - .collect() - ); - let game_start = chrono::Utc::now() + Self::COUNTDOWN_DURATION; - if self.custom_logic_handler.on_countdown_start(self, game_start).await { - let mut senders = Vec::new(); - for (player_id, conn) in self.users.read().await.iter() { - let user_desc = self.user_descriptor(*player_id).unwrap(); - senders.push((conn.connection.clone(), user_desc.state.clone())); + self.start_game().await; + } + } + + async fn start_game(&self) { + self.loading_state.store(LoadingState::InGame.to_u8(), std::sync::atomic::Ordering::Relaxed); + let ready_players = count_users_in_mode(ConnectionMode::WaitingToStart, self.descriptors.values()); + let player_count = self.real_player_count(); + log::info!("Starting game for {}/{} players in game {}", ready_players, player_count, self.game_guid()); + tokio::time::sleep(Self::END_OF_SYNC_DELAY).await; + self.fakes_handler.on_ready( + self.users.read().await.iter() + .map(|(id, real_player)| (*id, real_player.connection.clone())) + .collect() + ); + let game_start = chrono::Utc::now() + Self::COUNTDOWN_DURATION; + if self.custom_logic_handler.on_countdown_start(self, game_start).await { + let mut senders = Vec::with_capacity(self.descriptors.len()); + let mut to_disconnect = Vec::with_capacity(player_count - ready_players); + for (player_id, conn) in self.users.read().await.iter() { + let user_desc = self.user_descriptor(*player_id).unwrap(); + if let Some(user_id) = user_desc.descriptor.user_id { + let mode = ConnectionMode::from_u8(user_desc.state.mode.load(std::sync::atomic::Ordering::Relaxed)); + if !matches!(mode, ConnectionMode::WaitingToStart) { + to_disconnect.push(user_id); + } } - self.game_start.store(game_start.timestamp(), std::sync::atomic::Ordering::Relaxed); - super::countdown::match_countdown(senders, game_start); + senders.push((conn.connection.clone(), user_desc.state.clone())); + } + self.game_start.store(game_start.timestamp(), std::sync::atomic::Ordering::Relaxed); + super::countdown::match_countdown(senders, game_start); + for user_id in to_disconnect { + self.on_end_connection(user_id, false).await; } } } @@ -1201,6 +1325,40 @@ impl GenericGamemodeEngine { } } + async fn on_timeout(&self, timeout: super::messages::TimeoutVariant, response: tokio::sync::oneshot::Sender) { + let loading_state = LoadingState::from_u8(self.loading_state.load(std::sync::atomic::Ordering::Relaxed)); + match timeout { + super::messages::TimeoutVariant::WaitingForLoadingSync => { + if matches!(loading_state, LoadingState::Starting) { + let is_any_users_waiting = is_any_users_in_mode(ConnectionMode::WaitingForSync, self.descriptors.values()); + if is_any_users_waiting { + response.send(true).unwrap_or_default(); + log::info!("Reached max time waiting for loading sync, starting sync for game {}", self.game_guid()); + self.start_sync().await; + } else { + response.send(false).unwrap_or_default(); + } + } else { + response.send(true).unwrap_or_default(); + } + }, + super::messages::TimeoutVariant::WaitingForGameStart => { + if matches!(loading_state, LoadingState::InSync) { + let is_any_users_waiting = is_any_users_in_mode(ConnectionMode::WaitingToStart, self.descriptors.values()); + if is_any_users_waiting { + response.send(true).unwrap_or_default(); + log::info!("Reached max time waiting for game start, starting countdown for game {}", self.game_guid()); + self.start_game().await; + } else { + response.send(false).unwrap_or_default(); + } + } else { + response.send(true).unwrap_or_default(); + } + } + } + } + fn spawn_send_loading_events(&self, user: &UserConnection, player_id: u8, players: Vec>, client_ais: Vec) { let connection = user.connection.clone(); let user_id = user.user.user_id(); @@ -1248,15 +1406,15 @@ impl GenericGamemodeEngine { Ok(()) } - fn spawn_send_sync_events(&self, user: &UserConnection, user_id: i32, player_id: u8, players: Vec>, extra_packets: Vec, map: std::sync::Arc) { + fn spawn_send_sync_events(&self, user: &UserConnection, user_id: Option, player_id: u8, players: Vec>, extra_packets: Vec, map: std::sync::Arc) { let connection = user.connection.clone(); tokio::spawn(Self::send_sync_events_wrapper(connection, user_id, player_id, players, extra_packets, map)); //user.state.mode.store(ConnectionMode::Sync.to_u8(), std::sync::atomic::Ordering::Relaxed); } - async fn send_sync_events_wrapper(connection: UserSender, user_id: i32, player_id: u8, players: Vec>, extra_packets: Vec, map: std::sync::Arc) { + async fn send_sync_events_wrapper(connection: UserSender, user_id: Option, player_id: u8, players: Vec>, extra_packets: Vec, map: std::sync::Arc) { if let Err(e) = Self::send_sync_events(connection, player_id, players, extra_packets, map).await { - log::error!("Failed to send Sync events for user {}: {}", user_id, e); + log::error!("Failed to send Sync events for user {}: {}", user_id.unwrap_or(-1), e); } } @@ -1418,6 +1576,7 @@ impl GenericGamemodeEngine { if old { log::warn!("Game {} was marked as done again", self.game_guid()); } else { + //panic!("Game should not be done!"); log::debug!("Game {} is marked done (handler will exit once all players have disconnected)", self.game_guid()); } } @@ -1437,3 +1596,25 @@ impl GenericGamemodeEngine { distance < sphere.radius } } + +fn count_users_in_mode<'a>(wants_mode: ConnectionMode, descriptors: impl std::iter::Iterator) -> usize { + descriptors.filter(|desc| + desc.descriptor.user_id.is_some() + && desc.state.mode.load(std::sync::atomic::Ordering::Relaxed) == wants_mode.to_u8() + ).count() +} + +fn is_any_users_in_mode<'a>(wants_mode: ConnectionMode, mut descriptors: impl std::iter::Iterator) -> bool { + descriptors.any(|desc| + desc.descriptor.user_id.is_some() + && desc.state.mode.load(std::sync::atomic::Ordering::Relaxed) == wants_mode.to_u8() + ) +} + +fn is_all_users_in_mode<'a>(wants_mode: ConnectionMode, mut descriptors: impl std::iter::Iterator) -> bool { + descriptors.all(|desc| { + let mode = ConnectionMode::from_u8(desc.state.mode.load(std::sync::atomic::Ordering::Relaxed)); + desc.descriptor.user_id.is_some() + && (mode.to_u8() == wants_mode.to_u8() || matches!(mode, ConnectionMode::Disconnected)) + }) +} diff --git a/rc_multiplayer/src/matches/messages.rs b/rc_multiplayer/src/matches/messages.rs index 066e2de..a28c46c 100644 --- a/rc_multiplayer/src/matches/messages.rs +++ b/rc_multiplayer/src/matches/messages.rs @@ -1,3 +1,8 @@ +pub enum TimeoutVariant { + WaitingForLoadingSync, + WaitingForGameStart, +} + pub enum GameMessage { NewConnection { user: std::sync::Arc>, @@ -105,6 +110,10 @@ pub enum GameMessage { motion: rlnl::machine_motion::MachineMotion, }, NoOp, + LoadingTimeout { + timeout: TimeoutVariant, + response: tokio::sync::oneshot::Sender, + }, } impl GameMessage { @@ -136,6 +145,7 @@ impl GameMessage { Self::PlayerInputChanged { user_id, .. } => *user_id, Self::Motion { user_id, .. } => *user_id, Self::NoOp => unreachable!("NoOp is irrelevant for user ID"), + Self::LoadingTimeout { .. } => unreachable!("Timeout is irrelevant for user ID"), } } } diff --git a/rc_multiplayer/src/matches/modes/battle_arena.rs b/rc_multiplayer/src/matches/modes/battle_arena.rs index c9d40e5..902c95c 100644 --- a/rc_multiplayer/src/matches/modes/battle_arena.rs +++ b/rc_multiplayer/src/matches/modes/battle_arena.rs @@ -1322,6 +1322,11 @@ impl CustomGameLogic for BattleArenaLogic { if generic.is_game_done() { return true; } + let game_start = generic.game_start.load(std::sync::atomic::Ordering::Relaxed); + if game_start == i64::MIN || game_start > chrono::Utc::now().timestamp() { + // game has not started yet, player probably timed out while loading (which we can ignore) + return true; + } let player_id = player.descriptor.player_id; self.do_destruct_tasks(generic, player_id).await; self.player_tracking.disconnect_player(player_id).await; @@ -1620,12 +1625,18 @@ impl CustomGameLogic for BattleArenaLogic { } if generic.is_game_done() { self.abort_timer_sync().await; + #[cfg(debug_assertions)] + log::debug!("Game {} is already complete", generic.game_guid()); return true; } if self.check_if_match_time_is_done(generic).await { + #[cfg(debug_assertions)] + log::debug!("Out of time for game {}", generic.game_guid()); return true; } if generic.map_config.capture_points.is_empty() { + #[cfg(debug_assertions)] + log::debug!("No capture points for game {}", generic.game_guid()); return true; // don't bother trying to track whether players are in capture points since there are none } if let Some(player_team) = self.player_tracking.team(motion.player_id).await { @@ -1646,6 +1657,8 @@ impl CustomGameLogic for BattleArenaLogic { self.capture_tracking.on_exit(generic, was_in_point, motion.player_id, player_team as i8, self.config.num_segments as f32).await; } } + } else { + log::warn!("Unknown team for player {} in game {}", motion.player_id, generic.game_guid()); } if let Some(tick_info) = self.capture_tracking.tick(generic, self.config.num_segments as f32).await { // handle shield (de)activation diff --git a/rc_multiplayer/src/matches/modes/elimination.rs b/rc_multiplayer/src/matches/modes/elimination.rs index 8f0d673..071dc7d 100644 --- a/rc_multiplayer/src/matches/modes/elimination.rs +++ b/rc_multiplayer/src/matches/modes/elimination.rs @@ -418,6 +418,11 @@ impl CustomGameLogic for EliminationLogic { if generic.is_game_done() { return true; } + let game_start = generic.game_start.load(std::sync::atomic::Ordering::Relaxed); + if game_start == i64::MIN || game_start > chrono::Utc::now().timestamp() { + // game has not started yet, player probably timed out while loading (which we can ignore) + return true; + } if chrono::Utc::now().timestamp() >= generic.game_end() { generic.game_done(); self.abort_timer_sync().await; diff --git a/rc_multiplayer/src/matches/modes/mod.rs b/rc_multiplayer/src/matches/modes/mod.rs index a0b5ccb..f0368a6 100644 --- a/rc_multiplayer/src/matches/modes/mod.rs +++ b/rc_multiplayer/src/matches/modes/mod.rs @@ -14,7 +14,7 @@ pub use pit::PitLogic; mod team_death_match; pub use team_death_match::TeamDeathMatchLogic; -mod trackers; +pub(super) mod trackers; async fn respawn_player_after(after: chrono::DateTime, players: Vec, spawn: oj_rc_core::persist::config::Point, player_id: u8, alive_flag: std::sync::Arc) { let sleep_dur = after.signed_duration_since(chrono::Utc::now()).to_std().expect("Respawn duration too long to sleep"); diff --git a/rc_multiplayer/src/matches/modes/pit.rs b/rc_multiplayer/src/matches/modes/pit.rs index a8318c0..1ad6b62 100644 --- a/rc_multiplayer/src/matches/modes/pit.rs +++ b/rc_multiplayer/src/matches/modes/pit.rs @@ -353,6 +353,11 @@ impl CustomGameLogic for PitLogic { if generic.is_game_done() { return true; } + let game_start = generic.game_start.load(std::sync::atomic::Ordering::Relaxed); + if game_start == i64::MIN || game_start > chrono::Utc::now().timestamp() { + // game has not started yet, player probably timed out while loading (which we can ignore) + return true; + } let read_lock = generic.users.read().await; if read_lock.len() == 1 { // nobody to play against, automatically end the game diff --git a/rc_multiplayer/src/matches/modes/team_death_match.rs b/rc_multiplayer/src/matches/modes/team_death_match.rs index 4d1caf9..e9c4ec9 100644 --- a/rc_multiplayer/src/matches/modes/team_death_match.rs +++ b/rc_multiplayer/src/matches/modes/team_death_match.rs @@ -325,6 +325,11 @@ impl CustomGameLogic for TeamDeathMatchLogic { if generic.is_game_done() { return true; } + let game_start = generic.game_start.load(std::sync::atomic::Ordering::Relaxed); + if game_start == i64::MIN || game_start > chrono::Utc::now().timestamp() { + // game has not started yet, player probably timed out while loading (which we can ignore) + return true; + } if let Some(winning_team) = PlayerTracker::single_remaining_team(generic).await { log::info!("All players except those on team {} have disconnected, ending game {} early", winning_team, generic.game_guid()); self.do_win(WinReason::OutOfPlayers, generic, winning_team).await; diff --git a/rc_multiplayer/src/matches/modes/trackers/mod.rs b/rc_multiplayer/src/matches/modes/trackers/mod.rs index e56b83f..80d3fce 100644 --- a/rc_multiplayer/src/matches/modes/trackers/mod.rs +++ b/rc_multiplayer/src/matches/modes/trackers/mod.rs @@ -3,3 +3,6 @@ pub use surrender::{SurrenderGameTracker, SurrenderVoteResult}; mod ticker; pub use ticker::TickTracker; + +mod timeout; +pub use timeout::Timeout; diff --git a/rc_multiplayer/src/matches/modes/trackers/timeout.rs b/rc_multiplayer/src/matches/modes/trackers/timeout.rs new file mode 100644 index 0000000..86ee2ab --- /dev/null +++ b/rc_multiplayer/src/matches/modes/trackers/timeout.rs @@ -0,0 +1,84 @@ +const TIMEOUT_GRANULARITY: std::time::Duration = std::time::Duration::from_millis(100); + +pub struct WithTimeoutAction bool) + Send + 'static> { + on_timeout: TA, +} + +pub struct WithTimeoutAndCancelAction bool) + Send + Sync + 'static, CA: (FnMut() -> bool) + Send + 'static> { + // returns true if successful (return false to continuously re-run it) + on_timeout: std::sync::Arc, + // returns true if cancelled + is_cancelled: CA, +} + + +pub struct Timeout { + inner: T, + duration: std::time::Duration, +} + +impl Timeout<()> { + #[must_use] + pub fn new(duration: std::time::Duration) -> Self { + Self { + inner: (), + duration, + } + } + + #[must_use] + pub fn on_timeout bool) + Send + 'static>(self, action: TA) -> Timeout> { + Timeout { + inner: WithTimeoutAction { on_timeout: action }, + duration: self.duration, + } + } +} + +impl bool) + Send + Sync + 'static> Timeout> { + #[must_use] + pub fn with_cancel_check bool) + Send + 'static>(self, action: CA) -> Timeout> { + Timeout { + inner: WithTimeoutAndCancelAction { + on_timeout: std::sync::Arc::new(self.inner.on_timeout), + is_cancelled: action, + }, + duration: self.duration, + } + } + + /*async fn main_task(mut self2: Self) { + tokio::time::sleep(self2.duration).await; + while !(self2.inner.on_timeout)() { + tokio::time::sleep(TIMEOUT_GRANULARITY).await; + } + + } + + pub async fn start(self) -> tokio::task::JoinHandle<()> { + tokio::task::spawn(Self::main_task(self)) + }*/ +} + +impl bool) + Send + Sync + 'static, CA: (FnMut() -> bool) + Send + 'static> Timeout> { + async fn main_task(mut self, deadline: chrono::DateTime) { + loop { + if (self.inner.is_cancelled)() { return; } + let now = chrono::Utc::now(); + if now >= deadline { + let on_timeout_clone = self.inner.on_timeout.clone(); + if tokio::task::spawn_blocking(move || on_timeout_clone() ).await.unwrap_or(true) { + return; + } + } + tokio::time::sleep(TIMEOUT_GRANULARITY).await; + } + } + + pub async fn start(self) -> tokio::task::JoinHandle<()> { + let deadline = chrono::Utc::now() + self.duration; + tokio::task::spawn(self.main_task(deadline)) + } +} + +