bones_framework/
networking.rs

1#![doc = include_str!("./networking.md")]
2
3use self::{input::NetworkControls, socket::Socket};
4use crate::input::{DenseInput, DenseInputConfig};
5use crate::networking::online::OnlineMatchmakerResponse;
6pub use crate::networking::random::RngGenerator;
7use crate::prelude::*;
8use bones_matchmaker_proto::{MATCH_ALPN, PLAY_ALPN};
9use ggrs::P2PSession;
10use instant::Duration;
11use once_cell::sync::Lazy;
12use std::{fmt::Debug, marker::PhantomData, sync::Arc};
13use tracing::{debug, error, info, trace, warn};
14
15#[cfg(feature = "net-debug")]
16use {
17    self::debug::{NetworkDebugMessage, PlayerSyncState, NETWORK_DEBUG_CHANNEL},
18    ggrs::{NetworkStats, PlayerHandle},
19};
20
21pub use iroh;
22
23pub mod input;
24pub mod lan;
25pub mod online;
26pub mod online_lobby;
27pub mod online_matchmaking;
28pub mod random;
29pub mod socket;
30
31#[cfg(feature = "net-debug")]
32pub mod debug;
33
34/// Runtime, needed to execute network related calls.
35pub static RUNTIME: Lazy<tokio::runtime::Runtime> =
36    Lazy::new(|| tokio::runtime::Runtime::new().expect("unable to crate tokio runtime"));
37
38/// Indicates if input from networking is confirmed, predicted, or if player is disconnected.
39#[derive(Debug, Copy, Clone, PartialEq, Eq)]
40pub enum NetworkInputStatus {
41    /// The input of this player for this frame is an actual received input.
42    Confirmed,
43    /// The input of this player for this frame is predicted.
44    Predicted,
45    /// The player has disconnected at or prior to this frame, so this input is a dummy.
46    Disconnected,
47}
48
49impl From<ggrs::InputStatus> for NetworkInputStatus {
50    fn from(value: ggrs::InputStatus) -> Self {
51        match value {
52            ggrs::InputStatus::Confirmed => NetworkInputStatus::Confirmed,
53            ggrs::InputStatus::Predicted => NetworkInputStatus::Predicted,
54            ggrs::InputStatus::Disconnected => NetworkInputStatus::Disconnected,
55        }
56    }
57}
58
59/// Module prelude.
60pub mod prelude {
61    pub use super::{
62        input, lan, online, random, DisconnectedPlayers, RngGenerator, SyncingInfo, RUNTIME,
63    };
64
65    #[cfg(feature = "net-debug")]
66    pub use super::debug::prelude::*;
67}
68
69/// Muliplier for framerate that will be used when playing an online match.
70///
71/// Lowering the frame rate a little for online matches reduces bandwidth and may help overall
72/// gameplay. This may not be necessary once we improve network performance.
73///
74/// Note that FPS is provided as an integer to ggrs, so network modified fps is rounded to nearest int,
75/// which is then used to compute timestep so ggrs and networking match.
76pub const NETWORK_FRAME_RATE_FACTOR: f32 = 0.9;
77
78/// Default frame rate to run at if user provides none
79pub const NETWORK_DEFAULT_SIMULATION_FRAME_RATE: f32 = 60.0;
80
81/// Number of frames client may predict beyond confirmed frame before freezing and waiting
82/// for inputs from other players. Default value if not specified in [`GgrsSessionRunnerInfo`].
83pub const NETWORK_MAX_PREDICTION_WINDOW_DEFAULT: usize = 7;
84
85// todo test as zero?
86
87/// Amount of frames GGRS will delay local input.
88pub const NETWORK_LOCAL_INPUT_DELAY_DEFAULT: usize = 2;
89
90/// Possible errors returned by network loop.
91pub enum NetworkError {
92    /// The session was disconnected.
93    Disconnected,
94}
95
96/// The [`ggrs::Config`] implementation used by Jumpy.
97#[derive(Debug)]
98pub struct GgrsConfig<T: DenseInput + Debug> {
99    phantom: PhantomData<T>,
100}
101
102impl<T: DenseInput + Debug> ggrs::Config for GgrsConfig<T> {
103    type Input = T;
104    type State = World;
105    /// Addresses are the same as the player handle for our custom socket.
106    type Address = usize;
107}
108
109/// The network endpoint used for all network communications.
110static NETWORK_ENDPOINT: tokio::sync::OnceCell<iroh::Endpoint> = tokio::sync::OnceCell::const_new();
111
112/// Get the network endpoint used for all communications.
113pub async fn get_network_endpoint() -> &'static iroh::Endpoint {
114    NETWORK_ENDPOINT
115        .get_or_init(|| async move {
116            let secret_key = iroh::key::SecretKey::generate();
117            iroh::Endpoint::builder()
118                .alpns(vec![MATCH_ALPN.to_vec(), PLAY_ALPN.to_vec()])
119                .discovery(Box::new(
120                    iroh::discovery::ConcurrentDiscovery::from_services(vec![
121                        Box::new(
122                            iroh::discovery::local_swarm_discovery::LocalSwarmDiscovery::new(
123                                secret_key.public(),
124                            )
125                            .unwrap(),
126                        ),
127                        Box::new(iroh::discovery::dns::DnsDiscovery::n0_dns()),
128                        Box::new(iroh::discovery::pkarr::PkarrPublisher::n0_dns(
129                            secret_key.clone(),
130                        )),
131                    ]),
132                ))
133                .secret_key(secret_key)
134                .bind()
135                .await
136                .unwrap()
137        })
138        .await
139}
140
141/// Resource containing the [`NetworkSocket`] implementation while there is a connection to a
142/// network game.
143///
144/// This is inserted into the world after a match has been established by a network matchmaker.
145#[derive(Clone, HasSchema, Deref, DerefMut)]
146#[schema(no_default)]
147pub struct NetworkMatchSocket(Arc<dyn NetworkSocket>);
148
149/// Wraps [`ggrs::Message`] with included `match_id`, used to determine if message received
150/// from current match.
151#[derive(Debug, Clone, Serialize, Deserialize)]
152pub struct GameMessage {
153    /// Socket match id
154    pub match_id: u8,
155    /// Wrapped message
156    pub message: ggrs::Message,
157}
158
159/// Automatically implemented for [`NetworkSocket`] + [`ggrs::NonBlockingSocket<usize>`].
160pub trait GgrsSocket: NetworkSocket + ggrs::NonBlockingSocket<usize> {}
161impl<T> GgrsSocket for T where T: NetworkSocket + ggrs::NonBlockingSocket<usize> {}
162
163/// Trait that must be implemented by socket connections establish by matchmakers.
164///
165/// The [`NetworkMatchSocket`] resource will contain an instance of this trait and will be used by
166/// the game to send network messages after a match has been established.
167pub trait NetworkSocket: Sync + Send {
168    /// Get a GGRS socket from this network socket.
169    fn ggrs_socket(&self) -> Socket;
170    /// Send a reliable message to the given [`SocketTarget`].
171    fn send_reliable(&self, target: SocketTarget, message: &[u8]);
172    /// Receive reliable messages from other players. The `usize` is the index of the player that
173    /// sent the message.
174    fn recv_reliable(&self) -> Vec<(u32, Vec<u8>)>;
175    /// Close the connection.
176    fn close(&self);
177    /// Get the player index of the local player.
178    fn player_idx(&self) -> u32;
179    /// Get the player count for this network match.
180    fn player_count(&self) -> u32;
181
182    /// Increment match id so messages from previous match that are still in flight
183    /// will be filtered out. Used when starting new session with existing socket.
184    fn increment_match_id(&mut self);
185}
186
187/// The destination for a reliable network message.
188pub enum SocketTarget {
189    /// Send to a specific player.
190    Player(u32),
191    /// Broadcast to all players.
192    All,
193}
194
195/// Resource updated each frame exposing syncing/networking information in the current session.
196#[derive(HasSchema, Clone)]
197#[schema(no_default)]
198pub struct SyncingInfo {
199    /// Current frame of simulation step
200    current_frame: i32,
201    /// The random seed for this session
202    random_seed: u64,
203    /// The additional online info.
204    online_info: Option<OnlineSyncingInfo>,
205}
206
207/// Holds data for an online session
208#[derive(HasSchema, Clone)]
209#[schema(no_default)]
210pub struct OnlineSyncingInfo {
211    /// Last confirmed frame by all clients.
212    /// Anything that occurred on this frame is agreed upon by all clients.
213    last_confirmed_frame: i32,
214    /// Socket
215    socket: Socket,
216    /// Networking stats for each connected player, stored at the \[player_idx\] index for each respective player.
217    players_network_stats: SVec<PlayerNetworkStats>,
218    /// The local player's index
219    local_player_idx: usize,
220    /// The local input delay set for this session
221    local_frame_delay: usize,
222    /// List of disconnected players (their idx)
223    disconnected_players: SVec<usize>,
224}
225
226impl SyncingInfo {
227    /// Checks if the session is online.
228    pub fn is_online(&self) -> bool {
229        self.online_info.is_some()
230    }
231
232    /// Checks if the session is offline.
233    pub fn is_offline(&self) -> bool {
234        self.online_info.is_none()
235    }
236
237    /// Getter for the current frame (number).
238    pub fn current_frame(&self) -> i32 {
239        self.current_frame
240    }
241
242    /// Getter for the last confirmed frame (number).
243    pub fn last_confirmed_frame(&self) -> i32 {
244        self.online_info
245            .as_ref()
246            .map(|info| info.last_confirmed_frame)
247            .unwrap_or(self.current_frame)
248    }
249    /// Getter for socket.
250    pub fn socket(&self) -> Option<&Socket> {
251        self.online_info.as_ref().map(|info| &info.socket)
252    }
253
254    /// Mutable getter for socket.
255    pub fn socket_mut(&mut self) -> Option<&mut Socket> {
256        self.online_info.as_mut().map(|info| &mut info.socket)
257    }
258
259    /// Getter for a single player's network stats using their player_idx
260    pub fn player_network_stats(&self, player_idx: usize) -> Option<PlayerNetworkStats> {
261        self.online_info
262            .as_ref()
263            .map(|info| info.players_network_stats.get(player_idx).cloned())?
264    }
265
266    /// Getter for all players' network stats, including local player (set to default). This maintains index == player_idx.
267    pub fn players_network_stats(&self) -> SVec<PlayerNetworkStats> {
268        self.online_info
269            .as_ref()
270            .map(|info| info.players_network_stats.clone())
271            .unwrap_or_default()
272    }
273
274    /// Getter for remote player network stats (filtering out local player). This does not maintain index == player_idx.
275    pub fn remote_players_network_stats(&self) -> SVec<PlayerNetworkStats> {
276        self.online_info
277            .as_ref()
278            .map(|info| {
279                info.players_network_stats
280                    .iter()
281                    .filter(|&stats| stats.ping != 0 || stats.kbps_sent != 0)
282                    .cloned()
283                    .collect()
284            })
285            .unwrap_or_default()
286    }
287
288    /// Calculates the total kilobits per second sent across all remote players.
289    pub fn total_kbps_sent(&self) -> usize {
290        self.remote_players_network_stats()
291            .iter()
292            .map(|stats| stats.kbps_sent)
293            .sum()
294    }
295
296    /// Calculates the average kilobits per second sent across all remote players.
297    pub fn averaged_kbps_sent(&self) -> f32 {
298        let remote_stats = self.remote_players_network_stats();
299        if remote_stats.is_empty() {
300            0.0
301        } else {
302            let total_kbps: usize = remote_stats.iter().map(|stats| stats.kbps_sent).sum();
303            total_kbps as f32 / remote_stats.len() as f32
304        }
305    }
306
307    /// Returns the highest number of local frames behind across all remote players.
308    pub fn highest_local_frames_behind(&self) -> i32 {
309        self.remote_players_network_stats()
310            .iter()
311            .map(|stats| stats.local_frames_behind)
312            .max()
313            .unwrap_or(0)
314    }
315
316    /// Returns the highest number of remote frames behind across all remote players.
317    pub fn highest_remote_frames_behind(&self) -> i32 {
318        self.remote_players_network_stats()
319            .iter()
320            .map(|stats| stats.remote_frames_behind)
321            .max()
322            .unwrap_or(0)
323    }
324
325    /// Calculates the average ping across all remote players.
326    pub fn averaged_ping(&self) -> u128 {
327        let remote_stats = self.remote_players_network_stats();
328        if remote_stats.is_empty() {
329            0
330        } else {
331            let total_ping: u128 = remote_stats.iter().map(|stats| stats.ping).sum();
332            total_ping / remote_stats.len() as u128
333        }
334    }
335
336    /// Returns the lowest ping across all remote players.
337    pub fn lowest_ping(&self) -> u128 {
338        self.remote_players_network_stats()
339            .iter()
340            .map(|stats| stats.ping)
341            .min()
342            .unwrap_or(0)
343    }
344
345    /// Returns the highest ping across all remote players.
346    pub fn highest_ping(&self) -> u128 {
347        self.remote_players_network_stats()
348            .iter()
349            .map(|stats| stats.ping)
350            .max()
351            .unwrap_or(0)
352    }
353
354    /// Getter for the local player index, if offline defaults to None.
355    pub fn local_player_idx_checked(&self) -> Option<usize> {
356        self.online_info.as_ref().map(|info| info.local_player_idx)
357    }
358
359    /// Getter for the local player index, if offline defaults to 0.
360    pub fn local_player_idx(&self) -> usize {
361        self.online_info
362            .as_ref()
363            .map(|info| info.local_player_idx)
364            .unwrap_or(0)
365    }
366
367    /// Getter for the local frame delay.
368    pub fn local_frame_delay(&self) -> usize {
369        self.online_info
370            .as_ref()
371            .map(|info| info.local_frame_delay)
372            .unwrap_or(0)
373    }
374
375    /// Getter for the number of players, if offline defaults to 0.
376    pub fn players_count(&self) -> usize {
377        self.online_info
378            .as_ref()
379            .map(|info| info.players_network_stats.len())
380            .unwrap_or(0)
381    }
382
383    /// Getter for the number of players, if offline defaults to None.
384    pub fn players_count_checked(&self) -> Option<usize> {
385        self.online_info
386            .as_ref()
387            .map(|info| info.players_network_stats.len())
388    }
389
390    /// Getter for the list of active players (idx) which are connected. Offline returns empty list.
391    pub fn active_players(&self) -> SVec<usize> {
392        self.online_info
393            .as_ref()
394            .map(|info| {
395                let total_players = info.players_network_stats.len();
396                (0..total_players)
397                    .filter(|&id| !info.disconnected_players.contains(&id))
398                    .collect()
399            })
400            .unwrap_or_default()
401    }
402
403    /// Getter for the list of active players (idx) which are connected. Offline returns None.
404    pub fn active_players_checked(&self) -> Option<SVec<usize>> {
405        self.online_info.as_ref().map(|info| {
406            let total_players = info.players_network_stats.len();
407            (0..total_players)
408                .filter(|&id| !info.disconnected_players.contains(&id))
409                .collect()
410        })
411    }
412
413    /// Getter for the list of players which have been disconnected (their idx). Offline returns empty list.
414    pub fn disconnected_players(&self) -> SVec<usize> {
415        self.online_info
416            .as_ref()
417            .map(|info| info.disconnected_players.clone())
418            .unwrap_or_default()
419    }
420
421    /// Getter for the list of players which have been disconnected (their idx). Offline returns None.
422    pub fn disconnected_players_checked(&self) -> Option<SVec<usize>> {
423        self.online_info
424            .as_ref()
425            .map(|info| info.disconnected_players.clone())
426    }
427
428    /// Getter for the random seed.
429    pub fn random_seed(&self) -> u64 {
430        self.random_seed
431    }
432}
433
434/// Resource tracking which players have been disconnected.
435/// May not be in world if no disconnects.
436///
437/// If rollback to frame before disconnect, player handle is still included here.
438#[derive(HasSchema, Clone, Default)]
439pub struct DisconnectedPlayers {
440    /// Handles of players that have been disconnected.
441    pub disconnected_players: Vec<usize>,
442}
443
444/// [`SessionRunner`] implementation that uses [`ggrs`] for network play.
445///
446/// This is where the whole `ggrs` integration is implemented.
447pub struct GgrsSessionRunner<'a, InputTypes: DenseInputConfig<'a>> {
448    /// The last player input we detected.
449    pub last_player_input: InputTypes::Dense,
450
451    /// The GGRS peer-to-peer session.
452    pub session: P2PSession<GgrsConfig<InputTypes::Dense>>,
453
454    /// Local player idx.
455    pub player_idx: u32,
456
457    /// Index of local player, computed from player_is_local
458    pub local_player_idx: u32,
459
460    /// The frame time accumulator, used to produce a fixed refresh rate.
461    pub accumulator: f64,
462
463    /// Timestamp of last time session was run to compute delta time.
464    pub last_run: Option<Instant>,
465
466    /// FPS from game adjusted with constant network factor (may be slightly slower)
467    pub network_fps: f64,
468
469    /// FPS from game not adjusted with network factor.
470    pub original_fps: f64,
471
472    /// Session runner's input collector.
473    pub input_collector: InputTypes::InputCollector,
474
475    /// Is local input disabled? (No input will be used if set)
476    pub local_input_disabled: bool,
477
478    /// Players who have been reported disconnected by ggrs
479    disconnected_players: Vec<usize>,
480
481    /// Store copy of socket to be able to restart session runner with existing socket.
482    socket: Socket,
483
484    /// Local input delay ggrs session was initialized with
485    local_input_delay: usize,
486
487    /// The random seed used for this session
488    pub random_seed: u64,
489}
490
491/// The info required to create a [`GgrsSessionRunner`].
492#[derive(Clone)]
493pub struct GgrsSessionRunnerInfo {
494    /// The socket that will be converted into GGRS socket implementation.
495    pub socket: Socket,
496    /// The local player idx
497    pub player_idx: u32,
498    /// the player count.
499    pub player_count: u32,
500
501    /// Max prediction window (max number of frames client may predict ahead of last confirmed frame)
502    /// `None` will use Bone's default.
503    pub max_prediction_window: Option<usize>,
504
505    /// Local input delay (local inputs + remote inputs will be buffered and sampled this many frames later)
506    /// Increasing helps with mitigating pops when remote user changes input quickly, and reduces amount of frames
507    /// client will end up predicted ahead from others, helps with high latency.
508    ///
509    /// `None` will use Bone's default.
510    pub local_input_delay: Option<usize>,
511    /// The random seed used for this session
512    pub random_seed: u64,
513}
514
515impl GgrsSessionRunnerInfo {
516    /// See [`GgrsSessionRunnerInfo`] fields for info on arguments.
517    pub fn new(
518        socket: Socket,
519        max_prediction_window: Option<usize>,
520        local_input_delay: Option<usize>,
521        random_seed: u64,
522    ) -> Self {
523        let player_idx = socket.player_idx();
524        let player_count = socket.player_count();
525        Self {
526            socket,
527            player_idx,
528            player_count,
529            max_prediction_window,
530            local_input_delay,
531            random_seed,
532        }
533    }
534}
535
536impl<'a, InputTypes> GgrsSessionRunner<'a, InputTypes>
537where
538    InputTypes: DenseInputConfig<'a>,
539{
540    /// Creates a new session runner from a `OnlineMatchmakerResponse::GameStarting`
541    /// Any input values set as `None` will be set to default.
542    /// If response is not `GameStarting` returns None.
543    pub fn new_networked_game_starting(
544        target_fps: Option<f32>,
545        max_prediction_window: Option<usize>,
546        local_input_delay: Option<usize>,
547        matchmaker_resp_game_starting: OnlineMatchmakerResponse,
548    ) -> Option<Self> {
549        if let OnlineMatchmakerResponse::GameStarting {
550            socket,
551            player_idx: _,
552            player_count: _,
553            random_seed,
554        } = matchmaker_resp_game_starting
555        {
556            Some(Self::new(
557                target_fps,
558                GgrsSessionRunnerInfo::new(
559                    socket.ggrs_socket(),
560                    max_prediction_window,
561                    local_input_delay,
562                    random_seed,
563                ),
564            ))
565        } else {
566            None
567        }
568    }
569
570    /// Creates a new session runner from scratch.
571    pub fn new(target_fps: Option<f32>, info: GgrsSessionRunnerInfo) -> Self
572    where
573        Self: Sized,
574    {
575        let simulation_fps = target_fps.unwrap_or(NETWORK_DEFAULT_SIMULATION_FRAME_RATE);
576
577        // Modified FPS may not be an integer, but ggrs requires integer fps, so we clamp and round
578        // to integer so our computed timestep will match  that of ggrs.
579        let network_fps = (simulation_fps * NETWORK_FRAME_RATE_FACTOR) as f64;
580        let network_fps = network_fps
581            .max(usize::MIN as f64)
582            .min(usize::MAX as f64)
583            .round() as usize;
584
585        // There may be value in dynamically negotitaing these values based on client's pings
586        // before starting the match.
587        let max_prediction = info
588            .max_prediction_window
589            .unwrap_or(NETWORK_MAX_PREDICTION_WINDOW_DEFAULT);
590        let local_input_delay = info
591            .local_input_delay
592            .unwrap_or(NETWORK_LOCAL_INPUT_DELAY_DEFAULT);
593
594        // Notify debugger of setting
595        #[cfg(feature = "net-debug")]
596        NETWORK_DEBUG_CHANNEL
597            .sender
598            .try_send(NetworkDebugMessage::SetMaxPrediction(max_prediction))
599            .unwrap();
600
601        let mut builder = ggrs::SessionBuilder::new()
602            .with_num_players(info.player_count as usize)
603            .with_input_delay(local_input_delay)
604            .with_fps(network_fps)
605            .unwrap()
606            .with_max_prediction_window(max_prediction)
607            .unwrap();
608
609        let local_player_idx = info.player_idx;
610        for i in 0..info.player_count {
611            if i == info.player_idx {
612                builder = builder
613                    .add_player(ggrs::PlayerType::Local, i as usize)
614                    .unwrap();
615            } else {
616                builder = builder
617                    .add_player(ggrs::PlayerType::Remote(i as usize), i as usize)
618                    .unwrap();
619            }
620        }
621
622        let session = builder.start_p2p_session(info.socket.clone()).unwrap();
623
624        Self {
625            last_player_input: InputTypes::Dense::default(),
626            session,
627            player_idx: info.player_idx,
628            local_player_idx,
629            accumulator: default(),
630            last_run: None,
631            network_fps: network_fps as f64,
632            original_fps: simulation_fps as f64,
633            disconnected_players: default(),
634            input_collector: InputTypes::InputCollector::default(),
635            socket: info.socket.clone(),
636            local_input_delay,
637            local_input_disabled: false,
638            random_seed: info.random_seed,
639        }
640    }
641}
642
643impl<InputTypes> SessionRunner for GgrsSessionRunner<'static, InputTypes>
644where
645    InputTypes: DenseInputConfig<'static> + 'static,
646{
647    fn step(&mut self, frame_start: Instant, world: &mut World, stages: &mut SystemStages) {
648        let step: f64 = 1.0 / self.network_fps;
649
650        let last_run = self.last_run.unwrap_or(frame_start);
651        let delta = (frame_start - last_run).as_secs_f64();
652        self.accumulator += delta;
653
654        let mut skip_frames: u32 = 0;
655
656        // Collect inputs and update controls
657        self.input_collector.apply_inputs(world);
658
659        // save local players dense input for use with ggrs
660        self.last_player_input = self.input_collector.get_state();
661
662        #[cfg(feature = "net-debug")]
663        // Current frame before we start network update loop
664        let current_frame_original = self.session.current_frame();
665
666        for event in self.session.events() {
667            match event {
668                ggrs::GgrsEvent::Synchronizing { addr, total, count } => {
669                    info!(player=%addr, %total, progress=%count, "Syncing network player");
670
671                    #[cfg(feature = "net-debug")]
672                    NETWORK_DEBUG_CHANNEL
673                        .sender
674                        .try_send(NetworkDebugMessage::PlayerSync((
675                            PlayerSyncState::SyncInProgress,
676                            addr,
677                        )))
678                        .unwrap();
679                }
680                ggrs::GgrsEvent::Synchronized { addr } => {
681                    info!(player=%addr, "Syncrhonized network client");
682
683                    #[cfg(feature = "net-debug")]
684                    NETWORK_DEBUG_CHANNEL
685                        .sender
686                        .try_send(NetworkDebugMessage::PlayerSync((
687                            PlayerSyncState::Sychronized,
688                            addr,
689                        )))
690                        .unwrap();
691                }
692                ggrs::GgrsEvent::Disconnected { addr } => {
693                    warn!(player=%addr, "Player Disconnected");
694                    self.disconnected_players.push(addr);
695
696                    #[cfg(feature = "net-debug")]
697                    NETWORK_DEBUG_CHANNEL
698                        .sender
699                        .try_send(NetworkDebugMessage::DisconnectedPlayers(
700                            self.disconnected_players.clone(),
701                        ))
702                        .unwrap();
703                } //return Err(SessionError::Disconnected)},
704                ggrs::GgrsEvent::NetworkInterrupted { addr, .. } => {
705                    info!(player=%addr, "Network player interrupted");
706                }
707                ggrs::GgrsEvent::NetworkResumed { addr } => {
708                    info!(player=%addr, "Network player re-connected");
709                }
710                ggrs::GgrsEvent::WaitRecommendation {
711                    skip_frames: skip_count,
712                } => {
713                    info!(
714                        "Skipping {skip_count} frames to give network players a chance to catch up"
715                    );
716                    skip_frames = skip_count;
717
718                    #[cfg(feature = "net-debug")]
719                    NETWORK_DEBUG_CHANNEL
720                        .sender
721                        .try_send(NetworkDebugMessage::SkipFrame {
722                            frame: current_frame_original,
723                            count: skip_count,
724                        })
725                        .unwrap();
726                }
727                ggrs::GgrsEvent::DesyncDetected {
728                    frame,
729                    local_checksum,
730                    remote_checksum,
731                    addr,
732                } => {
733                    error!(%frame, %local_checksum, %remote_checksum, player=%addr, "Network de-sync detected");
734                }
735            }
736        }
737
738        loop {
739            if self.accumulator >= step {
740                self.accumulator -= step;
741
742                if !self.local_input_disabled {
743                    self.session
744                        .add_local_input(self.local_player_idx as usize, self.last_player_input)
745                        .unwrap();
746                } else {
747                    // If local input is disabled, we still submit a default value representing no-inputs.
748                    // This way if input is disabled current inputs will not be held down indefinitely.
749                    self.session
750                        .add_local_input(
751                            self.local_player_idx as usize,
752                            InputTypes::Dense::default(),
753                        )
754                        .unwrap();
755                }
756
757                #[cfg(feature = "net-debug")]
758                {
759                    let current_frame = self.session.current_frame();
760                    let confirmed_frame = self.session.confirmed_frame();
761
762                    NETWORK_DEBUG_CHANNEL
763                        .sender
764                        .try_send(NetworkDebugMessage::FrameUpdate {
765                            current: current_frame,
766                            last_confirmed: confirmed_frame,
767                        })
768                        .unwrap();
769                }
770
771                if skip_frames > 0 {
772                    skip_frames = skip_frames.saturating_sub(1);
773                    continue;
774                }
775
776                match self.session.advance_frame() {
777                    Ok(requests) => {
778                        for request in requests {
779                            match request {
780                                ggrs::GgrsRequest::SaveGameState { cell, frame } => {
781                                    cell.save(frame, Some(world.clone()), None)
782                                }
783                                ggrs::GgrsRequest::LoadGameState { cell, .. } => {
784                                    world.load_snapshot(cell.load().unwrap_or_default());
785                                }
786                                ggrs::GgrsRequest::AdvanceFrame {
787                                    inputs: network_inputs,
788                                } => {
789                                    // Input has been consumed, signal that we are in new input frame
790
791                                    // TODO: only call `InputCollector::advance_frame`
792                                    // when we are in a "prediction frame", not a "rollback catchup frame".
793                                    // If there are requests to load a snapshot and re-run frames the
794                                    // input collector's input has already been defined for those frames.
795                                    self.input_collector.advance_frame();
796
797                                    // Fetch the PlayerNetworkStats for each remote player, guaranteeing each one is inserted into the index matching its handle
798                                    let mut players_network_stats: Vec<PlayerNetworkStats> = vec![
799                                        PlayerNetworkStats::default();
800                                        self.session.remote_player_handles().len() + 1 // + 1 for the local player to maintain correct length
801                                    ];
802                                    for handle in self.session.remote_player_handles().iter() {
803                                        if let Ok(stats) = self.session.network_stats(*handle) {
804                                            players_network_stats[*handle] =
805                                                PlayerNetworkStats::from_ggrs_network_stats(
806                                                    *handle, stats,
807                                                );
808                                        }
809                                    }
810
811                                    // Create and insert the RngGenerator resource if it doesn't exist
812                                    if world.resources.get::<RngGenerator>().is_none() {
813                                        let rng_generator = RngGenerator::new(self.random_seed);
814                                        world.insert_resource(rng_generator);
815                                    }
816
817                                    // TODO: Make sure SyncingInfo is initialized immediately when session is created,
818                                    // even before a frame has advanced.
819                                    // The existance of this resource may be used to determine if in an online match, and there could
820                                    // be race if expected it to exist but testing before first frame advance.
821                                    world.insert_resource(SyncingInfo {
822                                        current_frame: self.session.current_frame(),
823                                        random_seed: self.random_seed,
824                                        online_info: Some(OnlineSyncingInfo {
825                                            last_confirmed_frame: self.session.confirmed_frame(),
826                                            socket: self.socket.clone(),
827                                            players_network_stats: players_network_stats.into(),
828                                            local_player_idx: self.local_player_idx as usize,
829                                            local_frame_delay: self.local_input_delay,
830                                            disconnected_players: self
831                                                .disconnected_players
832                                                .clone()
833                                                .into(),
834                                        }),
835                                    });
836
837                                    // Disconnected players persisted on session runner, and updated each frame.
838                                    // This avoids a rollback from changing resource state.
839                                    world.insert_resource(DisconnectedPlayers {
840                                        disconnected_players: self.disconnected_players.clone(),
841                                    });
842
843                                    {
844                                        world
845                                            .init_resource::<Time>()
846                                            .advance_exact(Duration::from_secs_f64(step));
847
848                                        // update game controls from ggrs inputs
849                                        let mut player_inputs =
850                                            world.init_resource::<InputTypes::Controls>();
851                                        for (player_idx, (input, status)) in
852                                            network_inputs.into_iter().enumerate()
853                                        {
854                                            trace!(
855                                                "Net player({player_idx}) local: {}, status: {status:?}, input: {:?}",
856                                                self.local_player_idx as usize == player_idx,
857                                                input
858                                            );
859                                            player_inputs.network_update(
860                                                player_idx,
861                                                &input,
862                                                status.into(),
863                                            );
864                                        }
865                                    }
866
867                                    // Run game session stages, advancing simulation
868                                    stages.run(world);
869                                }
870                            }
871                        }
872                    }
873                    Err(e) => match e {
874                        ggrs::GgrsError::NotSynchronized => {
875                            debug!("Waiting for network clients to sync")
876                        }
877                        ggrs::GgrsError::PredictionThreshold => {
878                            warn!("Freezing game while waiting for network to catch-up.");
879
880                            #[cfg(feature = "net-debug")]
881                            NETWORK_DEBUG_CHANNEL
882                                .sender
883                                .try_send(NetworkDebugMessage::FrameFroze {
884                                    frame: self.session.current_frame(),
885                                })
886                                .unwrap();
887                        }
888                        e => error!("Network protocol error: {e}"),
889                    },
890                }
891            } else {
892                break;
893            }
894        }
895
896        self.last_run = Some(frame_start);
897
898        // Fetch GGRS network stats of remote players and send to net debug tool
899        #[cfg(feature = "net-debug")]
900        {
901            let mut network_stats: Vec<(PlayerHandle, NetworkStats)> = vec![];
902            for handle in self.session.remote_player_handles().iter() {
903                if let Ok(stats) = self.session.network_stats(*handle) {
904                    network_stats.push((*handle, stats));
905                }
906            }
907            if !network_stats.is_empty() {
908                NETWORK_DEBUG_CHANNEL
909                    .sender
910                    .try_send(NetworkDebugMessage::NetworkStats { network_stats })
911                    .unwrap();
912            }
913        }
914    }
915
916    fn restart_session(&mut self) {
917        // Rebuild session info from runner + create new runner
918
919        // Increment match id so messages from previous match that are still in flight
920        // will be filtered out.
921        self.socket.increment_match_id();
922
923        let runner_info = GgrsSessionRunnerInfo {
924            socket: self.socket.clone(),
925            player_idx: self.player_idx,
926            player_count: self.session.num_players().try_into().unwrap(),
927            max_prediction_window: Some(self.session.max_prediction()),
928            local_input_delay: Some(self.local_input_delay),
929            random_seed: self.random_seed,
930        };
931        *self = GgrsSessionRunner::new(Some(self.original_fps as f32), runner_info);
932    }
933
934    fn disable_local_input(&mut self, input_disabled: bool) {
935        self.local_input_disabled = input_disabled;
936    }
937}
938
939/// A schema-compatible wrapper for ggrs `NetworkStats` struct contains networking stats.
940#[derive(Debug, Default, Clone, Copy, HasSchema)]
941pub struct PlayerNetworkStats {
942    /// The idx of the player these stats are for. Included here for self-attesting/ease-of-access.
943    pub player_idx: usize,
944    /// The length of the queue containing UDP packets which have not yet been acknowledged by the end client.
945    /// The length of the send queue is a rough indication of the quality of the connection. The longer the send queue, the higher the round-trip time between the
946    /// clients. The send queue will also be longer than usual during high packet loss situations.
947    pub send_queue_len: usize,
948    /// The roundtrip packet transmission time as calculated by GGRS.
949    pub ping: u128,
950    /// The estimated bandwidth used between the two clients, in kilobits per second.
951    pub kbps_sent: usize,
952
953    /// The number of frames GGRS calculates that the local client is behind the remote client at this instant in time.
954    /// For example, if at this instant the current game client is running frame 1002 and the remote game client is running frame 1009,
955    /// this value will mostly likely roughly equal 7.
956    pub local_frames_behind: i32,
957    /// The same as [`local_frames_behind`], but calculated from the perspective of the remote player.
958    ///
959    /// [`local_frames_behind`]: #structfield.local_frames_behind
960    pub remote_frames_behind: i32,
961}
962
963impl PlayerNetworkStats {
964    /// Creates a new PlayerNetworkStats from a player index and a ggrs NetworkStats.
965    pub fn from_ggrs_network_stats(player_idx: usize, stats: ggrs::NetworkStats) -> Self {
966        Self {
967            player_idx,
968            send_queue_len: stats.send_queue_len,
969            ping: stats.ping,
970            kbps_sent: stats.kbps_sent,
971            local_frames_behind: stats.local_frames_behind,
972            remote_frames_behind: stats.remote_frames_behind,
973        }
974    }
975}
bones_framework/networking.rs

bones_framework/
networking.rs
