From 4849f2659aaeda42ad9546117bec364573cd2d53 Mon Sep 17 00:00:00 2001 From: Nyx JPN Date: Wed, 7 Oct 2026 08:22:37 +0900 Subject: [PATCH 1/3] Implement GekkoNet Rollback Using Orca Snapshots --- CMakeLists.txt | 4 +- Externals/GekkoNet/CMakeLists.txt | 33 + Externals/GekkoNet/LICENSE | 24 + Externals/GekkoNet/include/backend.h | 207 ++ Externals/GekkoNet/include/compression.h | 61 + Externals/GekkoNet/include/event.h | 106 + Externals/GekkoNet/include/gekko_types.h | 19 + Externals/GekkoNet/include/gekkonet.h | 217 ++ Externals/GekkoNet/include/input.h | 89 + Externals/GekkoNet/include/net.h | 155 + Externals/GekkoNet/include/session.h | 218 ++ Externals/GekkoNet/include/storage.h | 34 + Externals/GekkoNet/include/sync.h | 57 + Externals/GekkoNet/src/backend.cpp | 945 +++++ Externals/GekkoNet/src/event.cpp | 291 ++ Externals/GekkoNet/src/game_session.cpp | 555 +++ Externals/GekkoNet/src/gekkonet.cpp | 225 ++ Externals/GekkoNet/src/input.cpp | 303 ++ Externals/GekkoNet/src/net.cpp | 118 + Externals/GekkoNet/src/player.cpp | 38 + Externals/GekkoNet/src/spectator_session.cpp | 242 ++ Externals/GekkoNet/src/storage.cpp | 34 + Externals/GekkoNet/src/stress_session.cpp | 167 + Externals/GekkoNet/src/sync.cpp | 165 + Externals/GekkoNet/thirdparty/zpp/LICENSE | 21 + Externals/GekkoNet/thirdparty/zpp/README.md | 366 ++ .../GekkoNet/thirdparty/zpp/serializer.h | 3282 +++++++++++++++++ Externals/licenses.md | 2 + Source/Core/AudioCommon/Mixer.cpp | 77 +- Source/Core/AudioCommon/Mixer.h | 17 + Source/Core/Common/ENet.cpp | 163 + Source/Core/Common/ENet.h | 26 + Source/Core/Common/TraversalClient.cpp | 5 +- Source/Core/Core/CMakeLists.txt | 11 + Source/Core/Core/Config/NetplaySettings.cpp | 6 + Source/Core/Core/Config/NetplaySettings.h | 4 + Source/Core/Core/Core.cpp | 3 + Source/Core/Core/CoreTiming.cpp | 49 +- Source/Core/Core/CoreTiming.h | 15 + Source/Core/Core/HLE/HLE.cpp | 8 +- Source/Core/Core/HLE/HLE_Misc.cpp | 6 + Source/Core/Core/HLE/HLE_Misc.h | 1 + Source/Core/Core/HW/Memmap.cpp | 99 +- Source/Core/Core/HW/Memmap.h | 23 +- Source/Core/Core/HW/SI/SI.cpp | 15 + Source/Core/Core/HW/SI/SI.h | 2 + Source/Core/Core/HW/SI/SI_DeviceGCAdapter.cpp | 6 +- .../Core/Core/HW/SI/SI_DeviceGCController.cpp | 6 +- Source/Core/Core/IOS/FS/HostBackend/FS.cpp | 90 +- Source/Core/Core/IOS/FS/HostBackend/FS.h | 67 + Source/Core/Core/IOS/FS/HostBackend/File.cpp | 5 + .../Core/Core/IOS/FS/HostBackend/Journal.cpp | 248 ++ Source/Core/Core/IOS/Network/Socket.cpp | 3 + Source/Core/Core/IOS/SDIO/SDIOSlot0.cpp | 6 +- Source/Core/Core/MemTools.cpp | 15 + Source/Core/Core/MemTools.h | 1 + Source/Core/Core/NetPlayClient.cpp | 103 +- Source/Core/Core/NetPlayClient.h | 1 + Source/Core/Core/NetPlayProto.h | 1 - Source/Core/Core/NetPlayServer.cpp | 54 +- Source/Core/Core/PowerPC/JitInterface.cpp | 4 +- Source/Core/Core/PowerPC/PowerPC.cpp | 19 +- Source/Core/Core/Rollback/Cow.cpp | 599 +++ Source/Core/Core/Rollback/Cow.h | 106 + Source/Core/Core/Rollback/GekkoRollback.cpp | 1109 ++++++ Source/Core/Core/Rollback/GekkoRollback.h | 46 + Source/Core/Core/Rollback/Rollback.cpp | 423 +++ Source/Core/Core/Rollback/Rollback.h | 122 + Source/Core/Core/Rollback/UndoLog.cpp | 182 + Source/Core/Core/Rollback/UndoLog.h | 126 + Source/Core/Core/State.cpp | 10 + Source/Core/Core/State.h | 7 + Source/Core/DolphinQt/MainWindow.cpp | 7 +- .../Core/DolphinQt/NetPlay/NetPlayDialog.cpp | 92 +- Source/Core/DolphinQt/NetPlay/NetPlayDialog.h | 2 + Source/Core/VideoCommon/BPStructs.cpp | 5 +- Source/Core/VideoCommon/OpcodeDecoding.cpp | 16 +- Source/Core/VideoCommon/Present.cpp | 33 +- Source/Core/VideoCommon/TextureCacheBase.cpp | 15 + Source/Core/VideoCommon/VideoBackendBase.cpp | 2 +- Source/Core/VideoCommon/VideoState.cpp | 67 +- Source/Core/VideoCommon/VideoState.h | 20 + 82 files changed, 11987 insertions(+), 139 deletions(-) create mode 100644 Externals/GekkoNet/CMakeLists.txt create mode 100644 Externals/GekkoNet/LICENSE create mode 100644 Externals/GekkoNet/include/backend.h create mode 100644 Externals/GekkoNet/include/compression.h create mode 100644 Externals/GekkoNet/include/event.h create mode 100644 Externals/GekkoNet/include/gekko_types.h create mode 100644 Externals/GekkoNet/include/gekkonet.h create mode 100644 Externals/GekkoNet/include/input.h create mode 100644 Externals/GekkoNet/include/net.h create mode 100644 Externals/GekkoNet/include/session.h create mode 100644 Externals/GekkoNet/include/storage.h create mode 100644 Externals/GekkoNet/include/sync.h create mode 100644 Externals/GekkoNet/src/backend.cpp create mode 100644 Externals/GekkoNet/src/event.cpp create mode 100644 Externals/GekkoNet/src/game_session.cpp create mode 100644 Externals/GekkoNet/src/gekkonet.cpp create mode 100644 Externals/GekkoNet/src/input.cpp create mode 100644 Externals/GekkoNet/src/net.cpp create mode 100644 Externals/GekkoNet/src/player.cpp create mode 100644 Externals/GekkoNet/src/spectator_session.cpp create mode 100644 Externals/GekkoNet/src/storage.cpp create mode 100644 Externals/GekkoNet/src/stress_session.cpp create mode 100644 Externals/GekkoNet/src/sync.cpp create mode 100644 Externals/GekkoNet/thirdparty/zpp/LICENSE create mode 100644 Externals/GekkoNet/thirdparty/zpp/README.md create mode 100644 Externals/GekkoNet/thirdparty/zpp/serializer.h create mode 100644 Source/Core/Core/IOS/FS/HostBackend/Journal.cpp create mode 100644 Source/Core/Core/Rollback/Cow.cpp create mode 100644 Source/Core/Core/Rollback/Cow.h create mode 100644 Source/Core/Core/Rollback/GekkoRollback.cpp create mode 100644 Source/Core/Core/Rollback/GekkoRollback.h create mode 100644 Source/Core/Core/Rollback/Rollback.cpp create mode 100644 Source/Core/Core/Rollback/Rollback.h create mode 100644 Source/Core/Core/Rollback/UndoLog.cpp create mode 100644 Source/Core/Core/Rollback/UndoLog.h diff --git a/CMakeLists.txt b/CMakeLists.txt index 08f774c168ac..fe339e988b27 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -762,7 +762,9 @@ if(WIN32 OR LINUX OR CMAKE_SYSTEM_NAME STREQUAL "FreeBSD" OR CMAKE_SYSTEM_NAME S add_definitions(-DHAVE_CPPIPC) endif() -######################################## +add_subdirectory(Externals/GekkoNet) + +######################################## # Pre-build events: Define configuration variables and write SCM info header # diff --git a/Externals/GekkoNet/CMakeLists.txt b/Externals/GekkoNet/CMakeLists.txt new file mode 100644 index 000000000000..134556d4a910 --- /dev/null +++ b/Externals/GekkoNet/CMakeLists.txt @@ -0,0 +1,33 @@ +add_library(GekkoNet STATIC + src/backend.cpp + src/event.cpp + src/game_session.cpp + src/gekkonet.cpp + src/input.cpp + src/net.cpp + src/player.cpp + src/spectator_session.cpp + src/storage.cpp + src/stress_session.cpp + src/sync.cpp +) + +target_include_directories(GekkoNet +PUBLIC + ${CMAKE_CURRENT_SOURCE_DIR}/include +PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR}/thirdparty +) + +# Dolphin supplies the UDP transport through its traversed ENet socket. Exclude +# GekkoNet's standalone Asio adapter so Dolphin does not vendor a second stack. +target_compile_definitions(GekkoNet PUBLIC GEKKONET_STATIC GEKKONET_NO_ASIO) +target_compile_features(GekkoNet PUBLIC cxx_std_17) + +if(COMMAND dolphin_disable_warnings) + dolphin_disable_warnings(GekkoNet) +endif() + +if(MSVC) + target_compile_options(GekkoNet PRIVATE /GR /EHsc /W0) +endif() diff --git a/Externals/GekkoNet/LICENSE b/Externals/GekkoNet/LICENSE new file mode 100644 index 000000000000..e6b74de324b0 --- /dev/null +++ b/Externals/GekkoNet/LICENSE @@ -0,0 +1,24 @@ +BSD 2-Clause License + +Copyright (c) 2024-2026, Jamie Meyer + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, this + list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, + this list of conditions and the following disclaimer in the documentation + and/or other materials provided with the distribution. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/Externals/GekkoNet/include/backend.h b/Externals/GekkoNet/include/backend.h new file mode 100644 index 000000000000..3f8e55d09de6 --- /dev/null +++ b/Externals/GekkoNet/include/backend.h @@ -0,0 +1,207 @@ +#pragma once + +#include "gekkonet.h" + +#include "gekko_types.h" + +#include "compression.h" +#include "net.h" +#include "event.h" + +#include +#include +#include +#include +#include +#include + +namespace Gekko { + + enum PlayerStatus { + Initiating, + Connected, + Disconnected, + }; + + struct InputCache { + Frame last_acked_frame = -1; + Frame last_input_frame = -1; + std::vector packets; + + bool IsValid(Frame current_ack, Frame current_last_input) const; + }; + + struct AdvantageHistory { + public: + void Init(); + + void Update(Frame frame); + + f32 GetAverageAdvantage(); + + void SetLocalAdvantage(i8 adv); + + void SetRemoteAdvantage(i8 adv); + + private: + static const i32 HISTORY_SIZE = 26; + + i8 _local_frame_adv; + + i8 _remote_frame_adv; + + i8 _local[HISTORY_SIZE]; + + i8 _remote[HISTORY_SIZE]; + }; + + class Player + { + public: + Player(Handle phandle, GekkoPlayerType type, NetAddress* addr, u32 magic = 0); + + GekkoPlayerType GetType(); + + PlayerStatus GetStatus(); + + void SetStatus(PlayerStatus type); + + void SetChecksum(Frame frame, u32 checksum); + + public: + Handle handle; + + u8 sync_num; + + u16 session_magic; + + NetStats stats; + + NetAddress address; + + std::map session_health; + + InputCache input_cache; + + u64 last_input_send_time = 0; + + AdvantageHistory adv_history; + + private: + GekkoPlayerType _type; + + PlayerStatus _status; + }; + + class MessageSystem { + public: + MessageSystem(); + + void Init(u8 num_players, u32 input_size); + + void AddInput(Frame input_frame, Handle player, u8 input[], bool remote = false); + + void AddSpectatorInput(Frame input_frame, u8 input[]); + + void SendPendingOutput(GekkoNetAdapter* host); + + void HandleData(GekkoNetAdapter* host, GekkoNetResult** data, u32 length); + + void SendInputAck(Handle player, Frame frame, i8 local_advantage); + + Frame GetLastAddedInput(bool spectator = false); + + bool CheckStatusActors(); + + void SendSessionHealth(Frame frame, u32 checksum); + + void SendNetworkHealth(); + + Frame GetLastAddedInputFrom(Handle player); + + std::deque>& GetNetPlayerQueue(Handle player); + + public: + std::vector> locals; + + std::vector> remotes; + + std::vector> spectators; + + SessionEventSystem session_events; + + std::map local_health; + + struct NetInputQueue { + Frame last_added_input = -1; + std::deque> inputs; + + NetInputQueue(const NetInputQueue&) = delete; + NetInputQueue& operator=(const NetInputQueue&) = delete; + + NetInputQueue(NetInputQueue&&) = default; + NetInputQueue& operator=(NetInputQueue&&) = default; + + NetInputQueue() = default; + + void TrimToAck(Frame min_ack, u32 max_size); + }; + + private: + void SendSyncRequest(NetAddress* addr); + + void SendSyncResponse(NetAddress* addr, u16 magic); + + void SendInputsToPeer(Player* peer, GekkoNetAdapter* host, bool spectator); + + std::vector GetRemoteHandlesForAddress(NetAddress* addr); + + Player* GetPlayerByHandle(Handle handle); + + Frame GetMinLastAckedFrame(bool spectator = false); + + void HandleTooFarBehindActors(bool spectator = false); + + u64 TimeSinceEpoch(); + + void SendDataToAll(NetData* pkt, GekkoNetAdapter* host, bool spectators_only = false); + + void SendDataTo(NetData* pkt, GekkoNetAdapter* host); + + void ParsePacket(NetAddress& addr, NetPacket& pkt, u32 packet_size); + + void OnSyncRequest(NetAddress& addr, NetPacket& pkt); + + void OnSyncResponse(NetAddress& addr, NetPacket& pkt); + + void OnInputs(NetAddress& addr, NetPacket& pkt); + + void OnInputAck(NetAddress& addr, NetPacket& pkt); + + void OnSessionHealth(NetAddress& addr, NetPacket& pkt); + + void OnNetworkHealth(NetAddress& addr, NetPacket& pkt); + + private: + const u32 MAX_INPUT_QUEUE_SIZE = 128; + const u32 NUM_TO_SYNC = 4; + + u32 _input_size; + + u16 _session_magic; + + u8 _num_players; + + // input queue for each player for either sending or receiving + std::vector _net_player_queue; + + // input queue for spectator inputs + NetInputQueue _net_spectator_queue; + + std::queue> _pending_output; + + std::vector _bin_buffer; + + u64 _last_sent_network_check; + }; +} diff --git a/Externals/GekkoNet/include/compression.h b/Externals/GekkoNet/include/compression.h new file mode 100644 index 000000000000..3baf3d6898bf --- /dev/null +++ b/Externals/GekkoNet/include/compression.h @@ -0,0 +1,61 @@ +#pragma once + +#include "gekko_types.h" + +#include +#include + +namespace Gekko { + struct Compression { + + static void PrintArray(const uint8_t* data, u32 length) { + for (u32 i = 0; i < length; i++) { + std::cout << +data[i] << ' '; + } + std::cout << std::endl; + std::cout << "_______________\n"; + } + + static std::vector RLEEncode(const uint8_t* data, u32 length) { + std::vector result; + + u32 idx = 0; + u8 count = 0; + + while (idx < length) { + count = 1; + while (count != UINT8_MAX && idx + 1 < length && data[idx] == data[idx + 1]) { + idx++; + count++; + } + result.push_back(count); + result.push_back(data[idx]); + idx++; + } + return result; + } + + static std::vector RLEDecode(const uint8_t* data, u32 length) { + std::vector result; + + u32 idx = 0; + u8 count = 0; + u8 value = 0; + + while (idx < length) { + count = data[idx]; + value = data[idx + 1]; + + for (i32 x = 0; x < count; x++) { + result.push_back(value); + } + + idx += 2; + } + + return result; + } + + Compression() = delete; + }; +} diff --git a/Externals/GekkoNet/include/event.h b/Externals/GekkoNet/include/event.h new file mode 100644 index 000000000000..c757c77c1210 --- /dev/null +++ b/Externals/GekkoNet/include/event.h @@ -0,0 +1,106 @@ +#pragma once + +#include "gekkonet.h" +#include "gekko_types.h" +#include "sync.h" +#include "storage.h" + +#include +#include + +namespace Gekko { + struct GameEventBuffer { + public: + void Init(u32 input_size); + + GekkoGameEvent* GetEvent(bool advance); + + void Reset(); + + private: + u32 _input_size = 0; + + u16 _index_others = 0; + + u16 _index_advance = 0; + + std::vector> _buffer_advance; + + std::vector> _buffer_others; + + std::vector> _input_memory_buffer; + }; + + struct GameEventSystem { + public: + void Init(u32 input_size); + + bool AddAdvanceEvent(SyncSystem& sync, bool rolling_back, bool running_ahead = false); + + void AddSaveEvent(SyncSystem& sync, StateStorage& storage, Frame* last_saved_frame = nullptr); + + void AddLoadEvent(SyncSystem& sync, StateStorage& storage); + + void AddRunaheadSaveEvent(SyncSystem& sync, StateStorage& storage); + + void AddRunaheadLoadEvent(StateStorage& storage); + + std::vector& GetEvents(); + + void Reset(); + + void Clear(); + + i32 Count(); + + GekkoGameEvent** Data(); + + private: + std::vector _current_events; + + GameEventBuffer _event_buffer; + }; + + struct SessionEventBuffer { + public: + SessionEventBuffer(); + + GekkoSessionEvent* GetEvent(); + + void Reset(); + + private: + u16 _index; + + std::vector> _buffer; + }; + + struct SessionEventSystem { + public: + void Reset(); + + std::vector& GetRecentEvents(); + + void AddPlayerSyncingEvent(Handle handle, u8 sync, u8 max); + + void AddPlayerConnectedEvent(Handle handle); + + void AddPlayerDisconnectedEvent(Handle handle); + + void AddSessionStartedEvent(); + + void AddSpectatorPausedEvent(); + + void AddSpectatorUnpausedEvent(); + + void AddDesyncDetectedEvent(Frame frame, Handle remote, u32 check_local, u32 check_remote); + + private: + void AddEvent(GekkoSessionEvent* ev); + + private: + std::vector _events; + + SessionEventBuffer _event_buffer; + }; +} diff --git a/Externals/GekkoNet/include/gekko_types.h b/Externals/GekkoNet/include/gekko_types.h new file mode 100644 index 000000000000..497dc5ebd7f1 --- /dev/null +++ b/Externals/GekkoNet/include/gekko_types.h @@ -0,0 +1,19 @@ +#pragma once + +#include + +typedef uint8_t u8; +typedef uint16_t u16; +typedef uint32_t u32; +typedef uint64_t u64; + +typedef int8_t i8; +typedef int16_t i16; +typedef int32_t i32; +typedef int64_t i64; + +typedef float f32; +typedef double f64; + +typedef int32_t Frame; +typedef int32_t Handle; diff --git a/Externals/GekkoNet/include/gekkonet.h b/Externals/GekkoNet/include/gekkonet.h new file mode 100644 index 000000000000..3c4327db775c --- /dev/null +++ b/Externals/GekkoNet/include/gekkonet.h @@ -0,0 +1,217 @@ +/* +BSD 2-Clause License + +Copyright (c) 2024-2026, Jamie Meyer + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, this + list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, + this list of conditions and the following disclaimer in the documentation + and/or other materials provided with the distribution. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ +#pragma once + +#ifdef __cplusplus +extern "C" { +#else +#include +#endif + +#ifdef _WIN32 +#ifdef GEKKONET_STATIC + // Static library - no import/export needed +#define GEKKONET_API +#else +#ifdef GEKKONET_DLL_EXPORT + // Building the DLL +#define GEKKONET_API __declspec(dllexport) +#else + // Using the DLL +#define GEKKONET_API __declspec(dllimport) +#endif +#endif +#else + // Non-Windows platforms don't need special export macros +#define GEKKONET_API +#endif + +// GekkoNet is mostly self contained memory wise using internal memory buffers. +// Technically the user shouldnt need to free or create any memory (look at the examples). +// The only case the user needs to create memory is when slotting in their own GekkoNetAdapter +typedef struct GekkoSession GekkoSession; + +typedef enum GekkoSessionType { + GekkoGameSession, // session for an active player. + GekkoStressSession, // session to test your local simulation for state desyncs. + GekkoSpectateSession, // session for spectators watching an active player. +} GekkoSessionType; + +typedef struct GekkoConfig { + unsigned char num_players; + unsigned char max_spectators; + unsigned char input_prediction_window; + unsigned int spectator_delay; + unsigned int input_size; + unsigned int state_size; + bool limited_saving; + bool desync_detection; + unsigned int check_distance; +} GekkoConfig; + +typedef enum GekkoPlayerType { + GekkoLocalPlayer, + GekkoRemotePlayer, + GekkoSpectator +} GekkoPlayerType; + +typedef struct GekkoNetAddress { + void* data; + unsigned int size; +} GekkoNetAddress; + +typedef struct GekkoNetResult { + GekkoNetAddress addr; + unsigned int data_len; + void* data; +} GekkoNetResult; + +typedef struct GekkoNetAdapter { + // send data to another peer + void (*send_data)(GekkoNetAddress* addr, const char* data, int length); + // receive all packets accumulated between the last frame and now + // the results within the array will be freed after use using the free_data function + // the array itself won't be touched + GekkoNetResult** (*receive_data)(int* length); + // free data function so gekkonet can cleanup data that the user created. + void (*free_data)(void* data_ptr); +} GekkoNetAdapter; + +typedef enum GekkoGameEventType { + GekkoEmptyGameEvent = -1, + GekkoAdvanceEvent, + GekkoSaveEvent, + GekkoLoadEvent +} GekkoGameEventType; + +typedef struct GekkoGameEvent { + GekkoGameEventType type; + + union GekkoEventData { + // events + struct GekkoAdvance { + int frame; + unsigned int input_len; + unsigned char* inputs; + bool rolling_back; + bool running_ahead; + } adv; + struct GekkoSave { + int frame; + unsigned int* checksum; + unsigned int* state_len; + unsigned char* state; + } save; + struct GekkoLoad { + int frame; + unsigned int state_len; + unsigned char* state; + } load; + } data; +} GekkoGameEvent; + +typedef enum GekkoSessionEventType { + GekkoEmptySessionEvent = -1, + GekkoPlayerSyncing, + GekkoPlayerConnected, + GekkoPlayerDisconnected, + GekkoSessionStarted, + GekkoSpectatorPaused, + GekkoSpectatorUnpaused, + GekkoDesyncDetected +} GekkoSessionEventType; + +typedef struct GekkoSessionEvent { + GekkoSessionEventType type; + + union GekkoSessionData { + struct GekkoSyncing { + int handle; + unsigned char current; + unsigned char max; + } syncing; + struct GekkoConnected { + int handle; + } connected; + struct GekkoDisconnected { + int handle; + } disconnected; + struct GekkoDesynced { + int frame; + unsigned int local_checksum; + unsigned int remote_checksum; + int remote_handle; + } desynced; + } data; +} GekkoSessionEvent; + +typedef struct GekkoNetworkStats { + float kb_sent; + float kb_received; + unsigned short last_ping; + float avg_ping; + float jitter; +} GekkoNetworkStats; + +// Public Facing API +GEKKONET_API bool gekko_create(GekkoSession** session, GekkoSessionType session_type); + +GEKKONET_API bool gekko_destroy(GekkoSession** session); + +GEKKONET_API void gekko_start(GekkoSession* session, GekkoConfig* config); + +GEKKONET_API void gekko_net_adapter_set(GekkoSession* session, GekkoNetAdapter* adapter); + +GEKKONET_API int gekko_add_actor(GekkoSession* session, GekkoPlayerType player_type, GekkoNetAddress* addr); + +GEKKONET_API void gekko_set_local_delay(GekkoSession* session, int player, unsigned char delay); + +GEKKONET_API void gekko_set_runahead(GekkoSession* session, unsigned char runahead); + +GEKKONET_API void gekko_add_local_input(GekkoSession* session, int player, void* input); + +GEKKONET_API GekkoGameEvent** gekko_update_session(GekkoSession* session, int* count); + +GEKKONET_API GekkoSessionEvent** gekko_session_events(GekkoSession* session, int* count); + +GEKKONET_API float gekko_frames_ahead(GekkoSession* session); + +GEKKONET_API void gekko_network_stats(GekkoSession* session, int player, GekkoNetworkStats* stats); + +GEKKONET_API void gekko_network_poll(GekkoSession* session); + +#ifndef GEKKONET_NO_ASIO + +GEKKONET_API GekkoNetAdapter* gekko_default_adapter(unsigned short port); + +GEKKONET_API bool gekko_default_adapter_destroy(); + +#endif // GEKKONET_NO_ASIO + +#ifdef __cplusplus +} +#endif diff --git a/Externals/GekkoNet/include/input.h b/Externals/GekkoNet/include/input.h new file mode 100644 index 000000000000..4f8dd4a6826f --- /dev/null +++ b/Externals/GekkoNet/include/input.h @@ -0,0 +1,89 @@ +#pragma once + +#include "gekko_types.h" +#include +#include + +namespace Gekko { + struct GameInput { + public: + GameInput(); + + void Init(GameInput* other); + + void Init(Frame frame_num, u8* inp, u32 inp_len); + + bool IsEqualTo(u8* other); + + void Clear(); + + public: + static const i32 NULL_FRAME = -1; + + Frame frame; + + std::unique_ptr input; + + u32 input_len; + }; + + struct InputBuffer { + static const u32 DEFAULT_BUFF_SIZE = 128; + + InputBuffer(); + + void Init(u8 delay, u8 input_window, u32 input_size, u32 buffer_size = DEFAULT_BUFF_SIZE); + + void AddLocalInput(Frame frame, u8* input); + + void AddInput(Frame frame, u8* input); + + void SetDelay(u8 delay); + + u8 GetDelay(); + + void SetInputPredictionWindow(u8 input_window); + + Frame GetIncorrectPredictionFrame(); + + std::unique_ptr GetInput(Frame frame, bool prediction = false); + + void SetRunaheadMode(bool running_ahead); + + Frame GetLastReceivedFrame(); + + void ClearIncorrectFrames(Frame clear_limit); + + private: + void ResetPrediction(); + + u32 PreviousFrame(Frame frame); + + bool HandleInputPrediction(Frame frame); + + bool CanPredictInput(); + + private: + bool _running_ahead; + + u8 _input_delay; + + u8 _input_prediction_window; + + u32 _input_size; + + u32 _buff_size; + + std::unique_ptr _empty_input; + + Frame _last_received_input; + + Frame _last_predicted_input; + + Frame _first_predicted_input; + + std::deque _incorrent_predicted_inputs; + + std::deque> _inputs; + }; +} diff --git a/Externals/GekkoNet/include/net.h b/Externals/GekkoNet/include/net.h new file mode 100644 index 000000000000..2aa1104e07a5 --- /dev/null +++ b/Externals/GekkoNet/include/net.h @@ -0,0 +1,155 @@ +#pragma once + +#include "gekkonet.h" +#include "gekko_types.h" + +#include +#include +#include + +#include "zpp/serializer.h" + +namespace Gekko { + struct NetAddress { + NetAddress(); + NetAddress(void* data, u32 size); + + u8* GetAddress(); + u32 GetSize(); + + void Copy(NetAddress* other); + bool Equals(NetAddress& other); + + private: + std::unique_ptr _data; + u32 _size; + }; + + enum PacketType : u8 { + Inputs = 1, + SpectatorInputs, + InputAck, + SyncRequest, + SyncResponse, + SessionHealth, + NetworkHealth + }; + + struct MsgHeader { + PacketType type; + u16 magic; + + template + static void serialize(Archive& a, Self& s) { + a(s.type, s.magic); + } + }; + + struct MsgBody : public zpp::serializer::polymorphic { + }; + + struct InputMsg : MsgBody { + Frame start_frame; + u16 input_count; + u16 total_size; + bool compressed; + + std::vector inputs; + + void Copy(const InputMsg* other) { + start_frame = other->start_frame; + input_count = other->input_count; + total_size = other->total_size; + compressed = other->compressed; + + inputs.clear(); + inputs.insert(inputs.begin(), other->inputs.begin(), other->inputs.end()); + } + + template + static void serialize(Archive& a, Self& s) { + a(s.start_frame, s.input_count, s.total_size, s.compressed, s.inputs); + } + }; + + struct InputAckMsg : MsgBody { + Frame ack_frame; + i8 frame_advantage; + + template + static void serialize(Archive& a, Self& s) { + a(s.ack_frame, s.frame_advantage); + } + }; + + struct SyncMsg : MsgBody { + u16 rng_data; + + template + static void serialize(Archive& a, Self& s) { + a(s.rng_data); + } + }; + + struct SessionHealthMsg : MsgBody { + Frame frame; + u32 checksum; + + template + static void serialize(Archive& a, Self& s) { + a(s.frame, s.checksum); + } + }; + + struct NetworkHealthMsg : MsgBody { + u64 send_time; + bool received; + + template + static void serialize(Archive& a, Self& s) { + a(s.send_time, s.received); + } + }; + + struct NetPacket { + MsgHeader header; + std::unique_ptr body; + }; + + struct NetData { + NetAddress addr; + NetPacket pkt; + }; + + struct NetStats { + static const u64 DISCONNECT_TIMEOUT = 5000; + static const u64 SYNC_MSG_DELAY = 200; + static const u64 NET_CHECK_DELAY = 500; + static const u64 INPUT_RETRY_INTERVAL = 200; + static const u32 RTT_HISTORY_SIZE = 10; + + Frame last_acked_frame = -1; + u64 last_sent_sync_message = 0; + u64 last_received_message = 0; + u64 last_received_frame = 0; + + float kb_sent_per_sec = 0; + float kb_received_per_sec = 0; + u32 bytes_sent_accum = 0; + u32 bytes_received_accum = 0; + u64 last_bandwidth_update = 0; + + std::vector rtt; + + void AddRTT(u16 rtt_ms); + void UpdateBandwidth(); + float CalculateJitter(); + float CalculateAvgRTT(); + u32 LastRTT(); + }; + + struct NetInputData { + std::vector handles; + InputMsg input; + }; +} diff --git a/Externals/GekkoNet/include/session.h b/Externals/GekkoNet/include/session.h new file mode 100644 index 000000000000..4295024bd60e --- /dev/null +++ b/Externals/GekkoNet/include/session.h @@ -0,0 +1,218 @@ +#pragma once + +#include +#include +#include + +#include "gekkonet.h" +#include "gekko_types.h" +#include "backend.h" +#include "event.h" +#include "sync.h" +#include "storage.h" + +// define GekkoSession internally +struct GekkoSession { + virtual void Init(GekkoConfig* config) = 0; + virtual void SetLocalDelay(i32 player, u8 delay) = 0; + virtual void SetRunahead(u8 runahead) = 0; + virtual void SetNetAdapter(GekkoNetAdapter* adapter) = 0; + virtual i32 AddActor(GekkoPlayerType type, GekkoNetAddress* addr) = 0; + virtual void AddLocalInput(i32 player, void* input) = 0; + virtual GekkoGameEvent** UpdateSession(i32* count) = 0; + virtual GekkoSessionEvent** Events(i32* count) = 0; + virtual f32 FramesAhead() = 0; + virtual void NetworkStats(i32 player, GekkoNetworkStats* stats) = 0; + virtual void NetworkPoll() = 0; + virtual ~GekkoSession() = default; +}; + +namespace Gekko { + + class GameSession : public GekkoSession { + public: + GameSession(); + + void Init(GekkoConfig* config) override; + + void SetLocalDelay(i32 player, u8 delay) override; + + void SetRunahead(u8 runahead) override; + + void SetNetAdapter(GekkoNetAdapter* adapter) override; + + i32 AddActor(GekkoPlayerType type, GekkoNetAddress* addr) override; + + void AddLocalInput(i32 player, void* input) override; + + GekkoGameEvent** UpdateSession(i32* count) override; + + GekkoSessionEvent** Events(i32* count) override; + + f32 FramesAhead() override; + + void NetworkStats(i32 player, GekkoNetworkStats* stats) override; + + void NetworkPoll() override; + + private: + void Poll(); + + bool AllActorsValid(); + + void HandleReceivedInputs(); + + void SendLocalInputs(); + + u8 GetMinLocalDelay(); + + bool IsPlayingLocally(); + + bool IsLockstepActive() const; + + void AddDisconnectedPlayerInputs(); + + void SendSpectatorInputs(); + + void HandleRollback(); + + void HandleSavingConfirmedFrame(); + + void HandleRunahead(); + + void RewindRunahead(); + + void SendSessionHealthCheck(); + + void SendNetworkHealthCheck(); + + void SessionIntegrityCheck(); + + private: + bool _started; + + Frame _last_saved_frame; + + Frame _last_sent_healthcheck; + + Frame _runahead_start_frame; + + u8 _runahead_frames; + + std::unique_ptr _disconnected_input; + + GekkoConfig _config; + + SyncSystem _sync; + + GekkoNetAdapter* _host; + + MessageSystem _msg; + + StateStorage _storage; + + GameEventSystem _game_events; + }; + + class SpectatorSession : public GekkoSession { + public: + SpectatorSession(); + + void Init(GekkoConfig* config) override; + + void SetLocalDelay(i32 player, u8 delay) override; + + void SetRunahead(u8 runahead) override {} + + void SetNetAdapter(GekkoNetAdapter* adapter) override; + + i32 AddActor(GekkoPlayerType type, GekkoNetAddress* addr) override; + + void AddLocalInput(i32 player, void* input) override; + + GekkoGameEvent** UpdateSession(i32* count) override; + + GekkoSessionEvent** Events(i32* count) override; + + f32 FramesAhead() override; + + void NetworkStats(i32 player, GekkoNetworkStats* stats) override; + + void NetworkPoll() override; + + private: + void Poll(); + + bool AllActorsValid(); + + void HandleReceivedInputs(); + + bool ShouldDelaySpectator(); + + private: + bool _started; + + bool _delay_spectator; + + Frame _last_saved_frame; + + GekkoConfig _config; + + SyncSystem _sync; + + GekkoNetAdapter* _host; + + MessageSystem _msg; + + GameEventSystem _game_events; + }; + + class StressSession : public GekkoSession { + public: + StressSession(); + + void Init(GekkoConfig* config) override; + + void SetLocalDelay(i32 player, u8 delay) override; + + void SetRunahead(u8 runahead) override {} + + void SetNetAdapter(GekkoNetAdapter* adapter) override; + + i32 AddActor(GekkoPlayerType type, GekkoNetAddress* addr) override; + + void AddLocalInput(i32 player, void* input) override; + + GekkoGameEvent** UpdateSession(i32* count) override; + + GekkoSessionEvent** Events(i32* count) override; + + f32 FramesAhead() override; + + void NetworkStats(i32 player, GekkoNetworkStats* stats) override; + + void NetworkPoll() override; + + private: + void HandleRollback(); + + void CheckForDesyncs(Frame check_frame); + + private: + GekkoConfig _config; + + SyncSystem _sync; + + StateStorage _storage; + + SessionEventSystem _session_events; + + GameEventSystem _game_events; + + std::vector _locals; + + u32 _check_distance; + + std::map _checksum_history; + }; +} diff --git a/Externals/GekkoNet/include/storage.h b/Externals/GekkoNet/include/storage.h new file mode 100644 index 000000000000..232c6932bda2 --- /dev/null +++ b/Externals/GekkoNet/include/storage.h @@ -0,0 +1,34 @@ +#pragma once + +#include "gekko_types.h" +#include "input.h" + +#include +#include + +namespace Gekko { + struct StateEntry { + Frame frame = GameInput::NULL_FRAME; + std::unique_ptr state; + u32 state_len = 0; + u32 checksum = 0; + }; + + class StateStorage { + public: + StateStorage(); + + void Init(u32 num_states, u32 state_size, bool limited); + + StateEntry* GetState(Frame frame); + + StateEntry* GetRunaheadState(); + + private: + u32 _max_num_states; + + std::vector> _states; + + StateEntry _runahead_state; + }; +} diff --git a/Externals/GekkoNet/include/sync.h b/Externals/GekkoNet/include/sync.h new file mode 100644 index 000000000000..28c64ab8a2a4 --- /dev/null +++ b/Externals/GekkoNet/include/sync.h @@ -0,0 +1,57 @@ +#pragma once + +#include + +#include "gekko_types.h" +#include "input.h" + +namespace Gekko { + + class SyncSystem { + public: + SyncSystem(); + + void Init(u8 num_players, u32 input_size, u32 buffer_size = InputBuffer::DEFAULT_BUFF_SIZE); + + void AddLocalInput(Handle player, u8* input); + + void AddRemoteInput(Handle player, u8* input, Frame frame); + + void IncrementFrame(); + + bool GetCurrentInputs(std::unique_ptr& inputs, Frame& frame); + + void SetRunaheadMode(bool running_ahead); + + bool GetSpectatorInputs(std::unique_ptr& inputs, Frame frame); + + bool GetLocalInput(Handle player, std::unique_ptr& input, Frame frame); + + void SetLocalDelay(Handle player, u8 delay); + + u8 GetLocalDelay(Handle player); + + void SetInputPredictionWindow(Handle player, u8 input_window); + + Frame GetCurrentFrame() const; + + void SetCurrentFrame(Frame frame); + + Frame GetMinIncorrectFrame(); + + Frame GetMinReceivedFrame(); + + Frame GetLastReceivedFrom(Handle player); + + void ClearIncorrectFramesUpTo(Frame clear_limit); + + private: + u8 _num_players; + + u32 _input_size; + + Frame _current_frame; + + std::unique_ptr _input_buffers; + }; +} diff --git a/Externals/GekkoNet/src/backend.cpp b/Externals/GekkoNet/src/backend.cpp new file mode 100644 index 000000000000..9cf4aa8d6d0d --- /dev/null +++ b/Externals/GekkoNet/src/backend.cpp @@ -0,0 +1,945 @@ +#include "backend.h" + +#include +#include +#include +#include +#include + +namespace +{ + u16 GenerateSessionMagic() + { + static std::atomic counter{1}; + + const auto now = std::chrono::steady_clock::now().time_since_epoch().count(); + const unsigned int sequence = counter.fetch_add(1, std::memory_order_relaxed); + std::uintptr_t stack_addr = reinterpret_cast(&sequence); + + std::uint64_t mixed = static_cast(now); + mixed ^= static_cast(sequence) * 0x9e3779b97f4a7c15ull; + mixed ^= static_cast(stack_addr); + mixed ^= mixed >> 33; + mixed *= 0xff51afd7ed558ccdull; + mixed ^= mixed >> 33; + + u16 magic = static_cast((mixed ^ (mixed >> 16) ^ (mixed >> 32) ^ (mixed >> 48)) & 0xffffu); + return magic == 0 ? 1 : magic; + } +} + +// register poly types. +namespace +{ + zpp::serializer::register_types< + zpp::serializer::make_type, + zpp::serializer::make_type, + zpp::serializer::make_type, + zpp::serializer::make_type, + zpp::serializer::make_type + > _; +} + +Gekko::MessageSystem::MessageSystem() +{ + _num_players = 0; + _input_size = 0; + _last_sent_network_check = 0; + + _session_magic = GenerateSessionMagic(); + + session_events = SessionEventSystem(); +} + +void Gekko::MessageSystem::Init(u8 num_players, u32 input_size) +{ + _num_players = num_players; + _input_size = input_size; + + _net_player_queue.resize(num_players); + +} + + +bool Gekko::InputCache::IsValid(Frame current_ack, Frame current_last_input) const +{ + return !packets.empty() + && last_acked_frame == current_ack + && last_input_frame == current_last_input; +} + +void Gekko::MessageSystem::NetInputQueue::TrimToAck(Frame min_ack, u32 max_size) +{ + if (inputs.empty()) return; + + if (min_ack != INT_MAX) { + const Frame oldest = last_added_input - (Frame)inputs.size() + 1; + const Frame acked = std::max((Frame)0, min_ack - oldest + 1); + const u32 target = (u32)inputs.size() - std::min((u32)acked, (u32)inputs.size()); + + while (inputs.size() > target) { + inputs.pop_front(); + } + return; + } + + while (inputs.size() > max_size) { + inputs.pop_front(); + } +} + +void Gekko::MessageSystem::AddInput(Frame input_frame, Handle player, u8 input[], bool remote) +{ + auto& input_q = _net_player_queue[player]; + if (input_q.last_added_input + 1 == input_frame) { + input_q.last_added_input++; + input_q.inputs.push_back(std::make_unique(_input_size)); + std::memcpy(input_q.inputs.back().get(), input, _input_size); + } + + // discard acked inputs (local) or just cap the queue (remote) + Frame min_ack = remote ? (Frame)INT_MAX : GetMinLastAckedFrame(false); + input_q.TrimToAck(min_ack, MAX_INPUT_QUEUE_SIZE); +} + +void Gekko::MessageSystem::AddSpectatorInput(Frame input_frame, u8 input[]) +{ + auto& input_q = _net_spectator_queue; + if (input_q.last_added_input + 1 == input_frame) { + input_q.last_added_input++; + const size_t num_players = locals.size() + remotes.size(); + input_q.inputs.push_back(std::make_unique(_input_size * num_players)); + std::memcpy(input_q.inputs.back().get(), input, _input_size * num_players); + } + + // discard acked inputs and cap the queue + input_q.TrimToAck(GetMinLastAckedFrame(true), MAX_INPUT_QUEUE_SIZE); +} + +void Gekko::MessageSystem::SendPendingOutput(GekkoNetAdapter* host) +{ + // send per-peer input packets to remotes + if (!remotes.empty() && !locals.empty()) { + for (auto& peer : remotes) { + SendInputsToPeer(peer.get(), host, false); + } + } + // check for disconnects (runs even in spectator sessions with no locals) + if (!remotes.empty()) { + HandleTooFarBehindActors(false); + } + + // send per-peer input packets to spectators + if (!spectators.empty() && !locals.empty()) { + for (auto& peer : spectators) { + SendInputsToPeer(peer.get(), host, true); + } + } + // check for disconnects + if (!spectators.empty()) { + HandleTooFarBehindActors(true); + } + + // drain remaining messages (acks, sync, health, etc.) + while (!_pending_output.empty()) { + auto& pkt = _pending_output.front(); + if ((pkt->pkt.header.type == SessionHealth || pkt->pkt.header.type == NetworkHealth) && pkt->addr.GetSize() == 0) { + // send to remotes + SendDataToAll(pkt.get(), host); + // send to spectators + SendDataToAll(pkt.get(), host, true); + } + else { + SendDataTo(pkt.get(), host); + } + _pending_output.pop(); + } +} + +void Gekko::MessageSystem::HandleData(GekkoNetAdapter* host, GekkoNetResult** data, u32 length) +{ + for (u32 i = 0; i < length; i++) { + auto res = data[i]; + auto addr = NetAddress(res->addr.data, res->addr.size); + + _bin_buffer.clear(); + + try { + _bin_buffer.insert(_bin_buffer.begin(), (u8*)res->data, (u8*)res->data + res->data_len); + + NetPacket pkt; + zpp::serializer::memory_input_archive in(_bin_buffer); + in(pkt.header, pkt.body); + + ParsePacket(addr, pkt, res->data_len); + } + catch (const std::exception&) { + printf("failed to deserialize packet\n"); + } + + // cleanup :) + host->free_data(res->addr.data); + host->free_data(res->data); + host->free_data(res); + } +} + +void Gekko::MessageSystem::SendSyncRequest(NetAddress* addr) +{ + if (!addr) { + return; + } + + _pending_output.push(std::make_unique()); + auto& message = _pending_output.back(); + + message->addr.Copy(addr); + + message->pkt.header.type = SyncRequest; + message->pkt.header.magic = 0; + + auto body = std::make_unique(); + body->rng_data = _session_magic; + + message->pkt.body = std::move(body); +} + +void Gekko::MessageSystem::SendSyncResponse(NetAddress* addr, u16 magic) +{ + if (!addr || magic == 0) { + return; + } + + _pending_output.push(std::make_unique()); + auto& message = _pending_output.back(); + + message->addr.Copy(addr); + message->pkt.header.type = SyncResponse; + message->pkt.header.magic = magic; + + auto body = std::make_unique(); + body->rng_data = _session_magic; + + message->pkt.body = std::move(body); +} + +void Gekko::MessageSystem::SendInputAck(Handle player, Frame frame, i8 local_advantage) +{ + auto plyr = GetPlayerByHandle(player); + + if (!plyr) { + return; + } + + _pending_output.push(std::make_unique()); + auto& message = _pending_output.back(); + + message->addr.Copy(&plyr->address); + message->pkt.header.magic = plyr->session_magic; + message->pkt.header.type = InputAck; + + auto body = std::make_unique(); + body->ack_frame = frame; + body->frame_advantage = local_advantage; + + message->pkt.body = std::move(body); +} + +std::vector Gekko::MessageSystem::GetRemoteHandlesForAddress(NetAddress* addr) +{ + auto result = std::vector(); + for (auto& player: remotes) { + if (player->address.Equals(*addr)) { + result.push_back(player->handle); + } + } + return result; +} + +Gekko::Player* Gekko::MessageSystem::GetPlayerByHandle(Handle handle) +{ + std::vector>* current = &locals; + for (u32 i = 0; i < 2; i++) + { + if (i == 1) { + current = &remotes; + } + + for (auto& player : *current) { + if (player->handle == handle) { + return player.get(); + } + } + } + + return nullptr; +} + +Frame Gekko::MessageSystem::GetMinLastAckedFrame(bool spectator) +{ + Frame min = INT_MAX; + for (auto& player : spectator ? spectators : remotes) { + if (player->GetStatus() == Connected) { + min = std::min(player->stats.last_acked_frame, min); + } + } + return min; +} + +Frame Gekko::MessageSystem::GetLastAddedInput(bool spectator) +{ + if (spectator) return _net_spectator_queue.last_added_input; + + Frame result = INT_MAX; + for (auto& local : locals) { + result = std::min(_net_player_queue[local->handle].last_added_input, result); + } + + return result; +} + +bool Gekko::MessageSystem::CheckStatusActors() +{ + i32 result = 0; + u64 now = TimeSinceEpoch(); + + std::vector>* current = &remotes; + + for (u32 i = 0; i < 2; i++) + { + if (i == 1) { + current = &spectators; + } + for (auto& player : *current) { + if (player->GetStatus() == Initiating) { + if (player->stats.last_sent_sync_message + NetStats::SYNC_MSG_DELAY < now) { + if (player->sync_num == 0) { + SendSyncRequest(&player->address); + player->stats.last_sent_sync_message = now; + } + else if (player->sync_num < NUM_TO_SYNC) { + SendSyncResponse(&player->address, player->session_magic); + player->stats.last_sent_sync_message = now; + } + else { + player->SetStatus(Connected); + session_events.AddPlayerConnectedEvent(player->handle); + result++; + } + } + result--; + } + } + } + + return result == 0; +} + +void Gekko::MessageSystem::SendSessionHealth(Frame frame, u32 checksum) +{ + _pending_output.push(std::make_unique()); + auto& message = _pending_output.back(); + + // the address and magic is set later so dont worry about it now + message->pkt.header.type = SessionHealth; + + auto body = std::make_unique(); + body->frame = frame; + body->checksum = checksum; + + message->pkt.body = std::move(body); +} + +void Gekko::MessageSystem::SendNetworkHealth() +{ + u64 now = TimeSinceEpoch(); + + // dont want to spam the network with network health packets + if (_last_sent_network_check + NetStats::NET_CHECK_DELAY > now) { + return; + } + + _pending_output.push(std::make_unique()); + auto& message = _pending_output.back(); + + // the address and magic is set later so dont worry about it now + message->pkt.header.type = NetworkHealth; + + auto body = std::make_unique(); + body->send_time = now; + body->received = false; + + message->pkt.body = std::move(body); + + _last_sent_network_check = now; +} + +Frame Gekko::MessageSystem::GetLastAddedInputFrom(Handle player) +{ + return _net_player_queue[player].last_added_input; +} + +std::deque>& Gekko::MessageSystem::GetNetPlayerQueue(Handle player) +{ + return _net_player_queue[player].inputs; +} + +void Gekko::MessageSystem::HandleTooFarBehindActors(bool spectator) +{ + const u64 now = TimeSinceEpoch(); + for (auto& actor : spectator ? spectators : remotes) { + if (actor->GetStatus() == Connected) { + // give the actor a chance to save itself. + if (actor->stats.last_received_message == 0) { + actor->stats.last_received_message = now; + continue; + } + // check whether messages are being sent if not disconnect. + const u64 msg_diff = now - actor->stats.last_received_message; + if (msg_diff > NetStats::DISCONNECT_TIMEOUT) { + session_events.AddPlayerDisconnectedEvent(actor->handle); + actor->SetStatus(Disconnected); + actor->sync_num = 0; + } + } + } +} + +u64 Gekko::MessageSystem::TimeSinceEpoch() +{ + using namespace std::chrono; + return duration_cast(steady_clock::now().time_since_epoch()).count(); +} + +void Gekko::MessageSystem::SendDataToAll(NetData* pkt, GekkoNetAdapter* host, bool spectators_only) +{ + auto& actors = spectators_only ? spectators : remotes; + + std::vector body_buffer; + + try { + zpp::serializer::memory_output_archive out(body_buffer); + out(pkt->pkt.body); + } + catch (const std::exception&) + { + printf("failed to serialize packet body\n"); + return; + } + + + for (auto& actor : actors) { + _bin_buffer.clear(); + if (actor->address.GetSize() != 0 && actor->GetStatus() != Disconnected) { + + pkt->addr.Copy(&actor->address); + pkt->pkt.header.magic = actor->session_magic; + + try { + zpp::serializer::memory_output_archive out(_bin_buffer); + out(pkt->pkt.header); + } + catch (const std::exception&) + { + printf("failed to serialize packet header\n"); + continue; + } + + _bin_buffer.insert( + _bin_buffer.end(), + body_buffer.begin(), + body_buffer.end() + ); + + auto addr = GekkoNetAddress(); + addr.data = actor->address.GetAddress(); + addr.size = actor->address.GetSize(); + + host->send_data(&addr, (char*)_bin_buffer.data(), (int)_bin_buffer.size()); + actor->stats.bytes_sent_accum += (u32)_bin_buffer.size(); + } + } +} + +void Gekko::MessageSystem::SendDataTo(NetData* pkt, GekkoNetAdapter* host) +{ + _bin_buffer.clear(); + + try { + zpp::serializer::memory_output_archive out(_bin_buffer); + out(pkt->pkt.header, pkt->pkt.body); + } + catch (const std::exception&) + { + printf("failed to serialize packet\n"); + return; + } + + auto addr = GekkoNetAddress(); + addr.data = pkt->addr.GetAddress(); + addr.size = pkt->addr.GetSize(); + + host->send_data(&addr, (char*)_bin_buffer.data(), (int)_bin_buffer.size()); + + u32 sent_size = (u32)_bin_buffer.size(); + std::vector>* current = &remotes; + for (u32 i = 0; i < 2; i++) + { + if (i == 1) { + current = &spectators; + } + + for (auto& actor : *current) { + if (actor->address.Equals(pkt->addr)) { + actor->stats.bytes_sent_accum += sent_size; + } + } + } +} + +void Gekko::MessageSystem::ParsePacket(NetAddress& addr, NetPacket& pkt, u32 packet_size) +{ + u64 now = TimeSinceEpoch(); + // update receive timers. + std::vector>* current = &remotes; + for (u32 i = 0; i < 2; i++) + { + if (i == 1) { + current = &spectators; + } + + for (auto& player : *current) { + if (player->address.Equals(addr)) { + player->stats.last_received_message = now; + player->stats.bytes_received_accum += packet_size; + } + } + } + + // handle packet. + if (pkt.header.magic != _session_magic) { + if (pkt.header.type == SyncRequest) { + OnSyncRequest(addr, pkt); + } + else { + printf("dropped packet!\n"); + } + } + else { + switch (pkt.header.type) + { + case SyncResponse: + OnSyncResponse(addr, pkt); + return; + case Inputs: + case SpectatorInputs: + OnInputs(addr, pkt); + return; + case InputAck: + OnInputAck(addr, pkt); + return; + case SessionHealth: + OnSessionHealth(addr, pkt); + return; + case NetworkHealth: + OnNetworkHealth(addr, pkt); + return; + default: + assert(false && "cannot process an unknown event!"); + return; + } + } +} + +void Gekko::MessageSystem::OnSyncRequest(NetAddress& addr, NetPacket& pkt) +{ + i32 should_send = 0; + u64 now = TimeSinceEpoch(); + auto body = (SyncMsg*)pkt.body.get(); + + // handle requests and set the peer its session magic for both remotes and spectators + std::vector>* current = &remotes; + for (u32 i = 0; i < 2; i++) + { + if (i == 1) { + current = &spectators; + } + + for (auto& player : *current) { + if (player->address.Equals(addr)) { + if (player->GetStatus() == Connected) { + should_send++; + continue; + } + + player->session_magic = body->rng_data; + if (player->sync_num == 0) { + player->stats.last_sent_sync_message = now; + should_send++; + } + } + } + } + + if (should_send > 0) { + // send a packet containing the local session magic + SendSyncResponse(&addr, body->rng_data); + } +} + +void Gekko::MessageSystem::OnSyncResponse(NetAddress& addr, NetPacket& pkt) +{ + i32 should_send = 0; + u64 now = TimeSinceEpoch(); + auto body = (SyncMsg*)pkt.body.get(); + + // handle sync responses for both remotes and spectators + std::vector>* current = &remotes; + for (u32 i = 0; i < 2; i++) + { + if (i == 1) { + current = &spectators; + } + + for (auto& player : *current) { + if (player->GetStatus() == Connected) { + // connected but the remote is still asking ? maybe high packet loss? send a response again + should_send++; + continue; + } + + if (player->address.Equals(addr)) { + player->session_magic = body->rng_data; + if (player->sync_num < NUM_TO_SYNC) { + player->sync_num++; + should_send++; + player->stats.last_sent_sync_message = now; + session_events.AddPlayerSyncingEvent( + player->handle, + player->sync_num, + NUM_TO_SYNC + ); + continue; + } + + if (player->sync_num >= NUM_TO_SYNC) { + player->SetStatus(Connected); + session_events.AddPlayerConnectedEvent(player->handle); + continue; + } + } + } + } + + if (should_send > 0) { + // send a packet containing the local session magic + SendSyncResponse(&addr, body->rng_data); + } +} + +void Gekko::MessageSystem::OnInputs(NetAddress& addr, NetPacket& pkt) +{ + auto body = (InputMsg*)pkt.body.get(); + + // RLE decompress if the sender compressed this packet + if (body->compressed) { + auto decompressed = Compression::RLEDecode(body->inputs.data(), (u32)body->inputs.size()); + body->inputs = std::move(decompressed); + } + + const Frame start_frame = body->start_frame; + const u32 input_count = body->input_count; + + const bool is_spectator = (pkt.header.type == SpectatorInputs); + + if (is_spectator) { + for (u32 frame_idx = 0; frame_idx < input_count; frame_idx++) { + const Frame recv_frame = start_frame + frame_idx; + const u32 frame_offset = frame_idx * _num_players * _input_size; + + for (u32 player = 0; player < _num_players; player++) { + u8* input = &body->inputs[frame_offset + player * _input_size]; + AddInput(recv_frame, player, input, true); + } + } + } else { + auto handles = GetRemoteHandlesForAddress(&addr); + const u32 player_count = (u32)handles.size(); + + for (u32 i = 0; i < player_count; i++) { + const u32 player_offset = i * input_count * _input_size; + + for (u32 frame_idx = 0; frame_idx < input_count; frame_idx++) { + const Frame recv_frame = start_frame + frame_idx; + u8* input = &body->inputs[player_offset + frame_idx * _input_size]; + AddInput(recv_frame, handles[i], input, true); + } + + auto player = GetPlayerByHandle(handles[i]); + if (player) { + player->stats.last_received_frame = TimeSinceEpoch(); + } + } + } +} + +void Gekko::MessageSystem::OnInputAck(NetAddress& addr, NetPacket& pkt) +{ + auto body = (InputAckMsg*)pkt.body.get(); + const Frame ack_frame = body->ack_frame; + const i8 remote_advantage = (i8)body->frame_advantage; + + for (auto& player : remotes) { + if (player->address.Equals(addr) && player->stats.last_acked_frame < ack_frame) { + player->stats.last_acked_frame = ack_frame; + player->adv_history.SetRemoteAdvantage(remote_advantage); + } + } + + for (auto& player : spectators) { + if (player->address.Equals(addr) && player->stats.last_acked_frame < ack_frame) { + player->stats.last_acked_frame = ack_frame; + } + } +} + +void Gekko::MessageSystem::OnSessionHealth(NetAddress& addr, NetPacket& pkt) +{ + auto body = (SessionHealthMsg*)pkt.body.get(); + + const Frame frame = body->frame; + const u32 checksum = body->checksum; + + for (auto& player : remotes) { + if (player->address.Equals(addr)) { + player->SetChecksum(frame, checksum); + + for (auto iter = player->session_health.begin(); + iter != player->session_health.end(); ) { + if (iter->first < (_net_player_queue[player->handle].last_added_input - 128)) { + iter = player->session_health.erase(iter); + } else { + ++iter; + } + } + break; + } + } +} + +void Gekko::MessageSystem::OnNetworkHealth(NetAddress& addr, NetPacket& pkt) +{ + auto body = (NetworkHealthMsg*)pkt.body.get(); + + // ok if its not a returned packet then update it and send it back to its specifc peer. + if (!body->received) { + // find the sender in remotes or spectators + Player* player = nullptr; + auto handles = GetRemoteHandlesForAddress(&addr); + if (!handles.empty()) { + player = GetPlayerByHandle(handles.at(0)); + } + + // check spectators if not found in remotes + if (!player) { + for (auto& spec : spectators) { + if (spec->address.Equals(addr)) { + player = spec.get(); + break; + } + } + } + + if (!player) { + return; + } + + _pending_output.push(std::make_unique()); + auto& message = _pending_output.back(); + + message->pkt.header.magic = player->session_magic; + message->pkt.header.type = NetworkHealth; + + auto new_body = std::make_unique(); + new_body->send_time = body->send_time; + new_body->received = true; + + message->pkt.body = std::move(new_body); + message->addr.Copy(&addr); + return; + } + + // else update network stats + u16 rtt_ms = (u16)(TimeSinceEpoch() - body->send_time); + std::vector>* current = &remotes; + + for (u32 i = 0; i < 2; i++) + { + if (i == 1) { + current = &spectators; + } + + for (auto& actor : *current) { + if (addr.Equals(actor->address)) { + actor->stats.AddRTT(rtt_ms); + } + } + } +} + +void Gekko::MessageSystem::SendInputsToPeer(Player* peer, GekkoNetAdapter* host, bool spectator) +{ + if (peer->address.GetSize() == 0 || peer->GetStatus() == Disconnected) { + return; + } + + const u32 MAX_INPUT_SIZE = 512; + const auto packet_type = spectator ? SpectatorInputs : Inputs; + auto& queue = spectator ? _net_spectator_queue : _net_player_queue[locals[0]->handle]; + const u32 num_players = spectator ? _num_players : (u32)locals.size(); + const u32 inputs_per_packet = MAX_INPUT_SIZE / (_input_size * num_players); + const u32 q_size = (u32)queue.inputs.size(); + + if (q_size == 0) return; + + const Frame queue_oldest_frame = queue.last_added_input - (Frame)q_size + 1; + const Frame last_input = spectator ? _net_spectator_queue.last_added_input : GetLastAddedInput(false); + + // peer is caught up, nothing to send + if (peer->stats.last_acked_frame >= last_input) return; + + // check per-peer cache + if (peer->input_cache.IsValid(peer->stats.last_acked_frame, last_input)) { + // cache hit: nothing new to send, this is a pure re-send — rate limit it + const u64 now = TimeSinceEpoch(); + if (peer->last_input_send_time + NetStats::INPUT_RETRY_INTERVAL > now) { + return; + } + for (const auto& cached_msg : peer->input_cache.packets) { + NetData data; + data.addr.Copy(&peer->address); + data.pkt.header.type = packet_type; + data.pkt.header.magic = peer->session_magic; + + auto message = std::make_unique(); + message->Copy(&cached_msg); + data.pkt.body = std::move(message); + + SendDataTo(&data, host); + } + peer->last_input_send_time = now; + return; + } + + // cache miss: rebuild packets for this peer + const Frame peer_start_frame = std::max(peer->stats.last_acked_frame + 1, queue_oldest_frame); + const u32 peer_start_idx = (u32)(peer_start_frame - queue_oldest_frame); + if (peer_start_idx >= q_size) return; + + const u32 peer_input_count = q_size - peer_start_idx; + + if (peer_input_count == 0) return; + + peer->input_cache.packets.clear(); + + const u32 packet_count = (peer_input_count + inputs_per_packet - 1) / inputs_per_packet; + + for (u32 pc = 0; pc < packet_count; pc++) { + const u32 input_start_idx = peer_start_idx + pc * inputs_per_packet; + const u32 input_end_idx = std::min(peer_start_idx + peer_input_count, input_start_idx + inputs_per_packet); + const u32 input_count = input_end_idx - input_start_idx; + + InputMsg msg; + msg.start_frame = queue_oldest_frame + input_start_idx; + + if (spectator) { + for (u32 i = input_start_idx; i < input_end_idx; i++) { + const auto& p_input = queue.inputs.at(i); + msg.inputs.insert(msg.inputs.end(), + p_input.get(), + p_input.get() + _input_size * num_players); + } + } + else { + for (u32 player = 0; player < num_players; player++) { + const auto& player_queue = _net_player_queue[locals[player]->handle]; + for (u32 i = input_start_idx; i < input_end_idx; i++) { + const auto& p_input = player_queue.inputs.at(i); + msg.inputs.insert(msg.inputs.end(), + p_input.get(), + p_input.get() + _input_size); + } + } + } + + // RLE compress only when it actually reduces size + msg.compressed = false; + auto compressed = Compression::RLEEncode(msg.inputs.data(), (u32)msg.inputs.size()); + if (compressed.size() < msg.inputs.size()) { + msg.inputs = std::move(compressed); + msg.compressed = true; + } + + msg.total_size = (u16)msg.inputs.size(); + msg.input_count = input_count; + + // cache and send + InputMsg cached_msg; + cached_msg.Copy(&msg); + peer->input_cache.packets.push_back(std::move(cached_msg)); + + NetData data; + data.addr.Copy(&peer->address); + data.pkt.header.type = packet_type; + data.pkt.header.magic = peer->session_magic; + data.pkt.body = std::make_unique(std::move(msg)); + + SendDataTo(&data, host); + } + + peer->last_input_send_time = TimeSinceEpoch(); + + // update cache keys + peer->input_cache.last_acked_frame = peer->stats.last_acked_frame; + peer->input_cache.last_input_frame = last_input; +} + +void Gekko::AdvantageHistory::Init() +{ + _local_frame_adv = 0; + _remote_frame_adv = 0; + std::memset(_local, 0, HISTORY_SIZE * sizeof(i8)); + std::memset(_remote, 0, HISTORY_SIZE * sizeof(i8)); +} + +void Gekko::AdvantageHistory::Update(Frame frame) +{ + const u32 update_frame = std::max(frame, 0); + _local[update_frame % HISTORY_SIZE] = _local_frame_adv; + _remote[update_frame % HISTORY_SIZE] = _remote_frame_adv; +} + +f32 Gekko::AdvantageHistory::GetAverageAdvantage() +{ + f32 sum_local = 0.f; + f32 sum_remote = 0.f; + + for (i32 i = 0; i < HISTORY_SIZE; i++) { + sum_local += _local[i]; + sum_remote += _remote[i]; + } + + f32 avg_local = sum_local / HISTORY_SIZE; + f32 avg_remote = sum_remote / HISTORY_SIZE; + + // return the frames ahead (halved: each peer corrects its share of the gap) + return (avg_local - avg_remote) / 2.f; +} + +void Gekko::AdvantageHistory::SetLocalAdvantage(i8 adv) { + _local_frame_adv = adv; +} + +void Gekko::AdvantageHistory::SetRemoteAdvantage(i8 adv) { + _remote_frame_adv = adv; +} diff --git a/Externals/GekkoNet/src/event.cpp b/Externals/GekkoNet/src/event.cpp new file mode 100644 index 000000000000..0fc61bfd3833 --- /dev/null +++ b/Externals/GekkoNet/src/event.cpp @@ -0,0 +1,291 @@ +#include "event.h" + +#include +#include +#include + +void Gekko::GameEventBuffer::Init(u32 input_size) +{ + _input_size = input_size; + _index_advance = 0; + _index_others = 0; +} + +void Gekko::GameEventBuffer::Reset() +{ + _index_advance = 0; + _index_others = 0; + + // clear the inputs + for (auto& ev : _buffer_advance){ + if (ev->type == GekkoEmptyGameEvent) { + break; + } + // reuse the input buffer rather than destroying it. + // set event to empty to be used again. + ev->type = GekkoEmptyGameEvent; + } + + for (auto& ev : _buffer_others) { + if (ev->type == GekkoEmptyGameEvent) { + break; + } + // set event to empty to be used again. + ev->type = GekkoEmptyGameEvent; + } +} + +GekkoGameEvent* Gekko::GameEventBuffer::GetEvent(bool advance) +{ + auto& buff = advance ? _buffer_advance : _buffer_others; + u16& idx = advance ? _index_advance : _index_others; + + idx++; + + if (buff.size() < idx) { + buff.push_back(std::make_unique()); + + if (advance) { + buff.back()->data.adv.input_len = _input_size; + // add more input space when needed + if (_input_memory_buffer.size() < idx) { + _input_memory_buffer.push_back(std::make_unique(_input_size)); + } + buff.back()->data.adv.inputs = _input_memory_buffer.back().get(); + } + } + + assert(idx != 0); + + return buff[idx - 1].get(); +} + +Gekko::SessionEventBuffer::SessionEventBuffer() +{ + _index = 0; +} + +GekkoSessionEvent* Gekko::SessionEventBuffer::GetEvent() +{ + _index++; + + if(_buffer.size() < _index) { + _buffer.push_back(std::make_unique()); + } + + assert(_index != 0); + + return _buffer[_index - 1].get(); +} + +void Gekko::SessionEventBuffer::Reset() +{ + _index = 0; + + for (auto& ev : _buffer) { + if (ev->type == GekkoEmptySessionEvent) { + break; + } + + ev->type = GekkoEmptySessionEvent; + } +} + +void Gekko::SessionEventSystem::Reset() +{ + _event_buffer.Reset(); + _events.clear(); +} + +void Gekko::SessionEventSystem::AddEvent(GekkoSessionEvent* ev) +{ + _events.push_back(ev); +} + +std::vector& Gekko::SessionEventSystem::GetRecentEvents() +{ + return _events; +} + +void Gekko::SessionEventSystem::AddPlayerSyncingEvent(Handle handle, u8 sync, u8 max) +{ + auto ev = _event_buffer.GetEvent(); + ev->type = GekkoPlayerSyncing; + ev->data.syncing.handle = handle; + ev->data.syncing.current = sync; + ev->data.syncing.max = max; + AddEvent(ev); +} + +void Gekko::SessionEventSystem::AddPlayerConnectedEvent(Handle handle) +{ + auto ev = _event_buffer.GetEvent(); + ev->type = GekkoPlayerConnected; + ev->data.connected.handle = handle; + AddEvent(ev); +} + +void Gekko::SessionEventSystem::AddPlayerDisconnectedEvent(Handle handle) +{ + auto ev = _event_buffer.GetEvent(); + ev->type = GekkoPlayerDisconnected; + ev->data.disconnected.handle = handle; + AddEvent(ev); +} + +void Gekko::SessionEventSystem::AddSessionStartedEvent() +{ + auto ev = _event_buffer.GetEvent(); + ev->type = GekkoSessionStarted; + AddEvent(ev); +} + +void Gekko::SessionEventSystem::AddSpectatorPausedEvent() +{ + auto ev = _event_buffer.GetEvent(); + ev->type = GekkoSpectatorPaused; + AddEvent(ev); +} + +void Gekko::SessionEventSystem::AddSpectatorUnpausedEvent() +{ + auto ev = _event_buffer.GetEvent(); + ev->type = GekkoSpectatorUnpaused; + AddEvent(ev); +} + +void Gekko::SessionEventSystem::AddDesyncDetectedEvent(Frame frame, Handle remote, u32 check_local, u32 check_remote) +{ + auto ev = _event_buffer.GetEvent(); + ev->type = GekkoDesyncDetected; + ev->data.desynced.frame = frame; + ev->data.desynced.remote_handle = remote; + ev->data.desynced.local_checksum = check_local; + ev->data.desynced.remote_checksum = check_remote; + AddEvent(ev); +} + +void Gekko::GameEventSystem::Init(u32 input_size) { + _event_buffer.Init(input_size); + _event_buffer.Reset(); + _current_events.clear(); +} + +bool Gekko::GameEventSystem::AddAdvanceEvent(SyncSystem& sync, bool rolling_back, bool running_ahead) +{ + Frame frame = GameInput::NULL_FRAME; + std::unique_ptr inputs; + if (!sync.GetCurrentInputs(inputs, frame)) { + return false; + } + + _current_events.push_back(_event_buffer.GetEvent(true)); + + auto event = _current_events.back(); + + event->type = GekkoAdvanceEvent; + event->data.adv.frame = frame; + event->data.adv.rolling_back = rolling_back; + event->data.adv.running_ahead = running_ahead; + + if (event->data.adv.inputs) { + std::memcpy(event->data.adv.inputs, inputs.get(), event->data.adv.input_len); + } + + return true; +} + +void Gekko::GameEventSystem::AddSaveEvent(SyncSystem& sync, StateStorage& storage, Frame* last_saved_frame) +{ + const Frame frame_to_save = sync.GetCurrentFrame(); + + auto state = storage.GetState(frame_to_save); + state->frame = frame_to_save; + + _current_events.push_back(_event_buffer.GetEvent(false)); + + auto event = _current_events.back(); + event->type = GekkoSaveEvent; + + event->data.save.frame = frame_to_save; + event->data.save.state = state->state.get(); + event->data.save.checksum = &state->checksum; + event->data.save.state_len = &state->state_len; + + if (last_saved_frame) { + *last_saved_frame = frame_to_save; + } +} + +void Gekko::GameEventSystem::AddLoadEvent(SyncSystem& sync, StateStorage& storage) +{ + const Frame frame_to_load = sync.GetCurrentFrame(); + + auto state = storage.GetState(frame_to_load); + + _current_events.push_back(_event_buffer.GetEvent(false)); + + auto event = _current_events.back(); + event->type = GekkoLoadEvent; + + event->data.load.frame = frame_to_load; + event->data.load.state = state->state.get(); + event->data.load.state_len = state->state_len; +} + +void Gekko::GameEventSystem::AddRunaheadSaveEvent(SyncSystem& sync, StateStorage& storage) +{ + const Frame frame_to_save = sync.GetCurrentFrame(); + + auto state = storage.GetRunaheadState(); + state->frame = frame_to_save; + + _current_events.push_back(_event_buffer.GetEvent(false)); + + auto event = _current_events.back(); + event->type = GekkoSaveEvent; + + event->data.save.frame = frame_to_save; + event->data.save.state = state->state.get(); + event->data.save.checksum = &state->checksum; + event->data.save.state_len = &state->state_len; +} + +void Gekko::GameEventSystem::AddRunaheadLoadEvent(StateStorage& storage) +{ + auto state = storage.GetRunaheadState(); + + _current_events.push_back(_event_buffer.GetEvent(false)); + + auto event = _current_events.back(); + event->type = GekkoLoadEvent; + + event->data.load.frame = state->frame; + event->data.load.state = state->state.get(); + event->data.load.state_len = state->state_len; +} + +std::vector& Gekko::GameEventSystem::GetEvents() +{ + return _current_events; +} + +void Gekko::GameEventSystem::Reset() +{ + _event_buffer.Reset(); +} + +void Gekko::GameEventSystem::Clear() +{ + _current_events.clear(); +} + +i32 Gekko::GameEventSystem::Count() +{ + return (i32)_current_events.size(); +} + +GekkoGameEvent** Gekko::GameEventSystem::Data() +{ + return _current_events.data(); +} diff --git a/Externals/GekkoNet/src/game_session.cpp b/Externals/GekkoNet/src/game_session.cpp new file mode 100644 index 000000000000..2501acf9a048 --- /dev/null +++ b/Externals/GekkoNet/src/game_session.cpp @@ -0,0 +1,555 @@ +#include "session.h" + +#include + +Gekko::GameSession::GameSession() +{ + _host = nullptr; + _started = false; + _last_saved_frame = GameInput::NULL_FRAME - 1; + _disconnected_input = nullptr; + _last_sent_healthcheck = GameInput::NULL_FRAME; + _runahead_start_frame = GameInput::NULL_FRAME; + _runahead_frames = 0; + _config = GekkoConfig(); +} + +void Gekko::GameSession::Init(GekkoConfig* config) +{ + _host = nullptr; + + _started = false; + + // get given configs + std::memcpy(&_config, config, sizeof(GekkoConfig)); + + // setup input buffer for the players + _sync.Init(_config.num_players, _config.input_size); + + // setup message system. + _msg.Init(_config.num_players, _config.input_size); + + // setup game event system + _game_events.Init(_config.input_size * _config.num_players); + + // setup state storage + _storage.Init(_config.input_prediction_window, _config.state_size, _config.limited_saving); + + // setup disconnected input for disconnected player within the session + _disconnected_input = std::make_unique(_config.input_size); + std::memset(_disconnected_input.get(), 0, _config.input_size); + + // we only detect desyncs whenever we are not limited saving for now. + _config.desync_detection = _config.limited_saving ? false : _config.desync_detection; +} + +void Gekko::GameSession::SetRunahead(u8 runahead) +{ + _runahead_frames = runahead; +} + +void Gekko::GameSession::SetLocalDelay(i32 player, u8 delay) +{ + for (u32 i = 0; i < _msg.locals.size(); i++) { + if (_msg.locals[i]->handle == player) { + _sync.SetLocalDelay(player, delay); + } + } +} + +void Gekko::GameSession::SetNetAdapter(GekkoNetAdapter* adapter) +{ + _host = adapter; +} + +i32 Gekko::GameSession::AddActor(GekkoPlayerType type, GekkoNetAddress* addr) +{ + const i32 ERR = -1; + std::unique_ptr address; + + if (addr) { + address = std::make_unique(addr->data, addr->size); + } + + if (type == GekkoSpectator) { + if (_msg.spectators.size() >= _config.max_spectators) { + return ERR; + } + + u32 new_handle = _config.num_players + (u32)_msg.spectators.size(); + _msg.spectators.push_back(std::make_unique(new_handle, type, address.get())); + + return new_handle; + } + else { + if (_started || _msg.locals.size() + _msg.remotes.size() >= _config.num_players) { + return ERR; + } + + u32 new_handle = (u32)(_msg.locals.size() + _msg.remotes.size()); + + if (type == GekkoLocalPlayer) { + _msg.locals.push_back(std::make_unique(new_handle, type, address.get())); + } + else { + // require an address when specifing a remote player + if (addr == nullptr) { + return ERR; + } + + _msg.remotes.push_back(std::make_unique(new_handle, type, address.get())); + _sync.SetInputPredictionWindow(new_handle, _config.input_prediction_window); + } + + return new_handle; + } +} + +void Gekko::GameSession::AddLocalInput(i32 player, void* input) +{ + u8* inp = (u8*)input; + + for (u32 i = 0; i < _msg.locals.size(); i++) { + if (_msg.locals[i]->handle == player) { + _sync.AddLocalInput(player, inp); + break; + } + } +} + +GekkoGameEvent** Gekko::GameSession::UpdateSession(i32* count) +{ + // reset session events + _msg.session_events.Reset(); + + // connection Handling + Poll(); + + // clear GameEvents + _game_events.Clear(); + + // gameplay + if (AllActorsValid()) { + // reset the game event buffer before doing anything else + _game_events.Reset(); + + // add inputs so we can continue the session. + AddDisconnectedPlayerInputs(); + + // rewind any runahead frames from the previous tick + RewindRunahead(); + + // check if we need to rollback + HandleRollback(); + + // check if we need to save the confirmed frame + HandleSavingConfirmedFrame(); + + // send a healthcheck if applicable + SendSessionHealthCheck(); + + // check if the session is still doing alright. + SessionIntegrityCheck(); + + // then advance the session + if (_game_events.AddAdvanceEvent(_sync, false)) { + if (!_config.limited_saving || + IsPlayingLocally() && + _sync.GetCurrentFrame() % _config.input_prediction_window == 0) { + _game_events.AddSaveEvent(_sync, _storage, &_last_saved_frame); + } + _sync.IncrementFrame(); + } + + // run ahead if configured + HandleRunahead(); + } + + *count = _game_events.Count(); + return _game_events.Data(); +} + +GekkoSessionEvent** Gekko::GameSession::Events(i32* count) +{ + *count = (i32)_msg.session_events.GetRecentEvents().size(); + return _msg.session_events.GetRecentEvents().data(); +} + +f32 Gekko::GameSession::FramesAhead() +{ + if (!_started) { + return 0.f; + } + + f32 sum = 0.f; + i32 count = 0; + for (auto& remote : _msg.remotes) { + if (remote->GetStatus() == Connected) { + sum += remote->adv_history.GetAverageAdvantage(); + count++; + } + } + return count > 0 ? sum / (f32)count : 0.f; +} + +void Gekko::GameSession::NetworkStats(i32 player, GekkoNetworkStats* stats) +{ + std::vector>* current = &_msg.remotes; + + for (u32 i = 0; i < 2; i++) + { + if (i == 1) { + current = &_msg.spectators; + } + + for (auto& actor : *current) { + if (actor->handle == player) { + actor->stats.UpdateBandwidth(); + stats->kb_sent = actor->stats.kb_sent_per_sec; + stats->kb_received = actor->stats.kb_received_per_sec; + stats->last_ping = actor->stats.LastRTT(); + stats->jitter = actor->stats.CalculateJitter(); + stats->avg_ping = actor->stats.CalculateAvgRTT(); + return; + } + } + } +} + +void Gekko::GameSession::NetworkPoll() +{ + Poll(); +} + +void Gekko::GameSession::HandleSavingConfirmedFrame() +{ + if (IsLockstepActive() || !_config.limited_saving || + IsPlayingLocally()) { + return; + } + + const Frame confirmed_frame = _sync.GetMinReceivedFrame(); + const Frame current = _sync.GetCurrentFrame(); + const Frame diff = current - (_last_saved_frame + 1); + + if (diff <= _config.input_prediction_window) { + return; + } + + assert(_last_saved_frame < confirmed_frame); + + const Frame sync_frame = _last_saved_frame; + const Frame frame_to_save = std::min(current - 1, confirmed_frame); + + _sync.SetCurrentFrame(sync_frame); + _game_events.AddLoadEvent(_sync, _storage); + _sync.IncrementFrame(); + + for (Frame frame = sync_frame + 1; frame < current; frame++) { + _game_events.AddAdvanceEvent(_sync, true); + if (frame == frame_to_save) { + _game_events.AddSaveEvent(_sync, _storage, &_last_saved_frame); + } + _sync.IncrementFrame(); + } + + // make sure that we are back where we started. + assert(_sync.GetCurrentFrame() == current); +} + +void Gekko::GameSession::SendSessionHealthCheck() +{ + if (!_config.desync_detection) { + return; + } + + const Frame current = _sync.GetCurrentFrame(); + const Frame confirmed = (current - _config.input_prediction_window) - 1; + + if (confirmed <= GameInput::NULL_FRAME) { + return; + } + + if (confirmed <= _last_sent_healthcheck) { + return; + } + + auto sav = _storage.GetState(confirmed); + + assert(sav->frame == confirmed); + + _last_sent_healthcheck = confirmed; + + _msg.local_health[confirmed] = sav->checksum; + + _msg.SendSessionHealth(confirmed, sav->checksum); + + for (auto iter = _msg.local_health.begin(); + iter != _msg.local_health.end(); ) { + if (iter->first < (confirmed - 100)) { + iter = _msg.local_health.erase(iter); + } else { + ++iter; + } + } +} + +void Gekko::GameSession::SendNetworkHealthCheck() +{ + _msg.SendNetworkHealth(); +} + +void Gekko::GameSession::SessionIntegrityCheck() +{ + if (!_config.desync_detection) { + return; + } + + for (auto iter = _msg.local_health.begin(); + iter != _msg.local_health.end(); ) { + + for (auto& player : _msg.remotes) { + if (player->session_health.count(iter->first)) { + if (player->session_health[iter->first] != iter->second) { + _msg.session_events.AddDesyncDetectedEvent( + iter->first, + player->handle, + iter->second, + player->session_health[iter->first] + ); + } + player->session_health.erase(iter->first); + } + } + + ++iter; + } +} + +void Gekko::GameSession::AddDisconnectedPlayerInputs() +{ + for (auto& player : _msg.remotes) { + if (player->GetStatus() == Disconnected) { + const Frame last_recv = _sync.GetLastReceivedFrom(player->handle) + 1; + const Frame current = _sync.GetCurrentFrame(); + for (Frame i = last_recv; i < current; i++) { + _sync.AddRemoteInput(player->handle, _disconnected_input.get(), i); + } + } + } +} + +void Gekko::GameSession::SendSpectatorInputs() +{ + const Frame current = _msg.GetLastAddedInput(true) + 1; + const Frame confirmed = _sync.GetMinReceivedFrame(); + + std::unique_ptr inputs; + for (Frame frame = current; frame <= confirmed; frame++) { + if (!_sync.GetSpectatorInputs(inputs, frame)) { + break; + } + _msg.AddSpectatorInput(frame, inputs.get()); + } +} + +void Gekko::GameSession::HandleRollback() +{ + Frame current = _sync.GetCurrentFrame(); + if (_last_saved_frame == GameInput::NULL_FRAME - 1) { + _sync.SetCurrentFrame(current - 1); + _game_events.AddSaveEvent(_sync, _storage, &_last_saved_frame); + _sync.IncrementFrame(); + } + + if (IsLockstepActive() || IsPlayingLocally()) { + return; + } + + current = _sync.GetCurrentFrame(); + const Frame min = _sync.GetMinIncorrectFrame(); + + // Don't roll back before the pre-frame baseline. + if (min == GameInput::NULL_FRAME) { + return; + } + + const Frame sync_frame = _config.limited_saving ? _last_saved_frame : min - 1; + const Frame frame_to_save = std::min(current - 1, min); + + // load the sync frame + _sync.SetCurrentFrame(sync_frame); + _game_events.AddLoadEvent(_sync, _storage); + _sync.IncrementFrame(); + + for (Frame frame = sync_frame + 1; frame < current; frame++) { + _game_events.AddAdvanceEvent(_sync, true); + if (!_config.limited_saving || frame == frame_to_save) { + _game_events.AddSaveEvent(_sync, _storage, &_last_saved_frame); + } + _sync.IncrementFrame(); + } + + // clear the marked mispredictions up to this point in the input buffer + _sync.ClearIncorrectFramesUpTo(current); + + // make sure that we are back where we started. + assert(_sync.GetCurrentFrame() == current); +} + +void Gekko::GameSession::Poll() +{ + // return if no host is defined. + if (!_host) { + return; + } + + // fetch data from network + int length = 0; + auto data = _host->receive_data(&length); + + // process the data we received + _msg.HandleData(_host, data, length); + + // handle received inputs + HandleReceivedInputs(); + + // add local input for the network + SendLocalInputs(); + + // send inputs to spectators + SendSpectatorInputs(); + + // send network health update + SendNetworkHealthCheck(); + + // now send data + _msg.SendPendingOutput(_host); +} + +bool Gekko::GameSession::AllActorsValid() +{ + if (!_started) { + if (!_msg.CheckStatusActors()) { + return false; + } + + // if none returned that the session is ready! + _msg.session_events.AddSessionStartedEvent(); + + _started = true; + + return true; + } + + return true; +} + +void Gekko::GameSession::HandleReceivedInputs() +{ + for (auto& remote : _msg.remotes) { + if (remote->GetStatus() != Connected) continue; + + { + auto handle = remote->handle; + const Frame last_recv = _sync.GetLastReceivedFrom(handle) + 1; + const Frame last_added = _msg.GetLastAddedInputFrom(handle); + + assert(last_added - last_recv <= 128); // more then 128 frames behind sounds incorrect. + + auto& input_q = _msg.GetNetPlayerQueue(handle); + const Frame min_frame = last_added - (i32)input_q.size() + 1; + const Frame current_frame = _sync.GetCurrentFrame(); + const Frame local_delay = (Frame)GetMinLocalDelay(); + for (int i = last_recv; i <= last_added; i++) { + if (i >= min_frame) { + int current_idx = i - min_frame; + u8* input = input_q[current_idx].get(); + _sync.AddRemoteInput(handle, input, i); + const i8 local_adv = (i8)(current_frame - i - local_delay); + _msg.SendInputAck(handle, i, local_adv); + } + } + } + } +} + +void Gekko::GameSession::SendLocalInputs() +{ + if (!_msg.locals.empty() && _started) { + const Frame current = _msg.GetLastAddedInput(false) + 1; + const Frame delay = GetMinLocalDelay(); + + auto input = std::make_unique(_config.input_size); + for (Frame frame = current; frame <= current + delay; frame++) { + for (auto& player : _msg.locals) { + if (!_sync.GetLocalInput(player->handle, input, frame)) { + return; + } + _msg.AddInput(frame, player->handle, input.get()); + } + // Record per-peer advantage snapshot once per actual game frame + if (frame == current) { + const Frame current_frame = _sync.GetCurrentFrame(); + for (auto& remote : _msg.remotes) { + if (remote->GetStatus() == Connected) { + const i8 local_adv = (i8)(current_frame - _sync.GetLastReceivedFrom(remote->handle) - (Frame)delay); + remote->adv_history.SetLocalAdvantage(local_adv); + remote->adv_history.Update(frame); + } + } + } + } + } +} + +u8 Gekko::GameSession::GetMinLocalDelay() +{ + u8 min = UINT8_MAX; + for (auto& player : _msg.locals) { + min = std::min(_sync.GetLocalDelay(player->handle), min); + } + return min; +} + +bool Gekko::GameSession::IsPlayingLocally() +{ + return _msg.remotes.empty() && !_msg.locals.empty(); +} + +bool Gekko::GameSession::IsLockstepActive() const +{ + return _config.input_prediction_window == 0; +} + +void Gekko::GameSession::RewindRunahead() +{ + if (_runahead_start_frame == GameInput::NULL_FRAME) { + return; + } + + _game_events.AddRunaheadLoadEvent(_storage); + _runahead_start_frame = GameInput::NULL_FRAME; +} + +void Gekko::GameSession::HandleRunahead() +{ + if (IsLockstepActive() || _runahead_frames == 0) { + return; + } + + _runahead_start_frame = _sync.GetCurrentFrame(); + _game_events.AddRunaheadSaveEvent(_sync, _storage); + + _sync.SetRunaheadMode(true); + for (u8 i = 0; i < _runahead_frames; i++) { + if (!_game_events.AddAdvanceEvent(_sync, false, true)) { + break; + } + _sync.IncrementFrame(); + } + _sync.SetRunaheadMode(false); + + // Reset back to the real frame so AddLocalInput and network logic see the correct frame + _sync.SetCurrentFrame(_runahead_start_frame); +} diff --git a/Externals/GekkoNet/src/gekkonet.cpp b/Externals/GekkoNet/src/gekkonet.cpp new file mode 100644 index 000000000000..261ff53c86f8 --- /dev/null +++ b/Externals/GekkoNet/src/gekkonet.cpp @@ -0,0 +1,225 @@ +#include "gekkonet.h" + +#include "session.h" + +GEKKONET_API bool gekko_create(GekkoSession** session, GekkoSessionType session_type) +{ + if (*session) { + return false; + } + + switch (session_type) { + case GekkoSessionType::GekkoGameSession: + *session = new Gekko::GameSession(); + break; + + case GekkoSessionType::GekkoStressSession: + *session = new Gekko::StressSession(); + break; + + case GekkoSessionType::GekkoSpectateSession: + *session = new Gekko::SpectatorSession(); + break; + + default: + return false; + } + + return true; +} + +bool gekko_destroy(GekkoSession** session) +{ + if (session && *session) { + delete *session; + *session = nullptr; + return true; + } + + return false; +} + +void gekko_start(GekkoSession* session, GekkoConfig* config) +{ + session->Init(config); +} + +void gekko_net_adapter_set(GekkoSession* session, GekkoNetAdapter* adapter) +{ + session->SetNetAdapter(adapter); +} + +int gekko_add_actor(GekkoSession* session, GekkoPlayerType player_type, GekkoNetAddress* addr) +{ + return session->AddActor(player_type, !addr ? nullptr : addr); +} + +void gekko_set_local_delay(GekkoSession* session, int player, unsigned char delay) +{ + session->SetLocalDelay(player, delay); +} + +void gekko_set_runahead(GekkoSession* session, unsigned char runahead) +{ + session->SetRunahead(runahead); +} + +void gekko_add_local_input(GekkoSession* session, int player, void* input) +{ + session->AddLocalInput(player, input); +} + +GekkoGameEvent** gekko_update_session(GekkoSession* session, int* count) +{ + return session->UpdateSession(count); +} + +GekkoSessionEvent** gekko_session_events(GekkoSession* session, int* count) +{ + return session->Events(count); +} + +float gekko_frames_ahead(GekkoSession* session) +{ + return session->FramesAhead(); +} + +void gekko_network_stats(GekkoSession* session, int player, GekkoNetworkStats* stats) +{ + session->NetworkStats(player, stats); +} + +void gekko_network_poll(GekkoSession* session) +{ + session->NetworkPoll(); +} + +#ifndef GEKKONET_NO_ASIO + +#ifdef _WIN32 +#ifndef _WIN32_WINNT +#define _WIN32_WINNT 0x0A00 +#endif +#endif // _WIN32 + +#define ASIO_STANDALONE + +#include "asio/asio.hpp" +#include + +static char _buffer[1024]; +static asio::error_code _ec; +static asio::io_context _io_ctx; +static asio::ip::udp::endpoint _remote; +static std::vector _results; +static asio::ip::udp::socket* _socket = nullptr; + +static asio::ip::udp::endpoint STOE(const std::string& str) { + std::string::size_type colon_pos = str.find(':'); + if (colon_pos == std::string::npos) { + throw std::invalid_argument("Invalid endpoint string"); + } + + std::string address = str.substr(0, colon_pos); + u16 port = (u16)(std::stoi(str.substr(colon_pos + 1))); + + return asio::ip::udp::endpoint(asio::ip::address::from_string(address), port); +} + + +static void asio_send(GekkoNetAddress* addr, const char* data, int length) { + std::string address((char*)addr->data, addr->size); + + auto endpoint = STOE(address); + + _socket->send_to(asio::buffer(data, length), endpoint, 0, _ec); + + if (_ec) { + std::cerr << "send failed: " << _ec.message() << std::endl; + } +} + +static GekkoNetResult** asio_receive(int* length) { + _results.clear(); + + while (true) { + const u32 len = (u32)_socket->receive_from(asio::buffer(_buffer), _remote, 0, _ec); + if (_ec && _ec != asio::error::would_block) { + // std::cout << "receive failed: " << _ec.message() << std::endl; + continue; + } + else if (!_ec) { + std::string endpoint = _remote.address().to_string() + ":" + std::to_string(_remote.port()); + + GekkoNetResult* res = reinterpret_cast(std::malloc(sizeof(*res))); + _results.push_back(res); + + res->addr.data = std::malloc(endpoint.size()); + res->addr.size = (u32)endpoint.size(); + std::memcpy(res->addr.data, endpoint.c_str(), res->addr.size); + + res->data_len = len; + res->data = std::malloc(len); + + std::memcpy(res->data, _buffer, len); + } + else { + break; + } + } + + *length = (int)_results.size(); + + return _results.data(); +} + +static void asio_free(void* data_ptr) { + std::free(data_ptr); +} + +static GekkoNetAdapter default_sock{ + asio_send, + asio_receive, + asio_free +}; + +GekkoNetAdapter* gekko_default_adapter(unsigned short port) { + // in case this has been called before. + if (_socket) { + asio::error_code ec; + _socket->shutdown(asio::socket_base::shutdown_type::shutdown_both, ec); + _socket->close(ec); + delete _socket; + _socket = nullptr; + } + + // setup socket with SO_REUSEADDR and error checking + asio::error_code ec; + auto* sock = new asio::ip::udp::socket(_io_ctx); + sock->open(asio::ip::udp::v4(), ec); + if (!ec) { + sock->set_option(asio::socket_base::reuse_address(true), ec); + sock->bind(asio::ip::udp::endpoint(asio::ip::udp::v4(), port), ec); + } + if (ec) { + std::cerr << "GekkoNet: Failed to bind socket on port " << port << ": " << ec.message() << '\n'; + delete sock; + return nullptr; + } + sock->non_blocking(true, ec); + _socket = sock; + + return &default_sock; +} + +bool gekko_default_adapter_destroy() +{ + if (_socket) { + delete _socket; + _socket = nullptr; + return true; + } + return false; +} + +#endif // GEKKONET_NO_ASIO diff --git a/Externals/GekkoNet/src/input.cpp b/Externals/GekkoNet/src/input.cpp new file mode 100644 index 000000000000..c0f726df70fc --- /dev/null +++ b/Externals/GekkoNet/src/input.cpp @@ -0,0 +1,303 @@ +#include "input.h" + +#include +#include + +Gekko::InputBuffer::InputBuffer() { + _empty_input = nullptr; + + _input_size = 0; + _buff_size = DEFAULT_BUFF_SIZE; + _input_delay = 0; + _input_prediction_window = 0; + _running_ahead = false; + + _last_received_input = GameInput::NULL_FRAME; + _last_predicted_input = GameInput::NULL_FRAME; + _first_predicted_input = GameInput::NULL_FRAME; + + _incorrent_predicted_inputs.clear(); +} + +void Gekko::InputBuffer::Init(u8 delay, u8 input_window, u32 input_size, u32 buffer_size) +{ + _input_delay = delay; + _input_size = input_size; + _buff_size = buffer_size; + _input_prediction_window = input_window; + + _last_received_input = GameInput::NULL_FRAME; + + _last_predicted_input = GameInput::NULL_FRAME; + _first_predicted_input = GameInput::NULL_FRAME; + + _incorrent_predicted_inputs.clear(); + + // init GameInput array + _empty_input = std::make_unique(_input_size); + std::memset(_empty_input.get(), 0, _input_size); + + for (u32 i = 0; i < _buff_size; i++) { + _inputs.push_back(std::make_unique()); + _inputs[i]->Init(GameInput::NULL_FRAME, _empty_input.get(), _input_size); + } +} + +void Gekko::InputBuffer::AddLocalInput(Frame frame, u8* input) +{ + if (_inputs[frame % _buff_size]->frame == GameInput::NULL_FRAME && _input_delay > 0) { + for (i32 i = 0; i < _input_delay; i++) { + AddInput(i, _empty_input.get()); + } + } + AddInput(frame + _input_delay, input); +} + +void Gekko::InputBuffer::AddInput(Frame frame, u8* input) +{ + // only allow sequential input insertion + if (frame != _last_received_input + 1) { + return; + } + + const Frame idx = frame % _buff_size; + if (_input_prediction_window > 0 && _first_predicted_input == frame) { + if (!_inputs[idx]->IsEqualTo(input)) { + // incorrect prediction + _incorrent_predicted_inputs.push_back(_first_predicted_input); + + // last prediction frame ? add correct input and reset prediction + if (_first_predicted_input == _last_predicted_input) { + _inputs[idx]->Init(frame, input, _input_size); + ResetPrediction(); + } else { + + // repeat the correct inputs instead of the wrong one. + const Frame diff = _last_predicted_input - _first_predicted_input; + for (Frame i = 0; i <= diff; i++) { + const Frame pred_frame = _first_predicted_input + i; + _inputs[pred_frame % _buff_size]->Init(pred_frame, input, _input_size); + } + + // move along this frame since the previous is verified now. + _first_predicted_input++; + } + } else { + + // correct prediction + if (_first_predicted_input == _last_predicted_input) { + ResetPrediction(); + } else { + _first_predicted_input++; + } + } + } else { + // not a predicted input + _inputs[idx]->Init(frame, input, _input_size); + } + + // always advance the buffer + _last_received_input++; +} + +void Gekko::InputBuffer::SetDelay(u8 delay) +{ + // early return AddLocalInput will handle it + if (_last_received_input == GameInput::NULL_FRAME || _input_delay == delay) { + _input_delay = delay; + return; + } + + // when our current delay is smaller then the new delay + // all we have to do is expand the delay with the last input we received + if (_input_delay < delay) { + _input_delay = delay; + + Frame last_input = _last_received_input; + u8* prev = _inputs[last_input % _buff_size]->input.get(); + + for (i32 i = 1; i <= _input_delay; i++) { + AddInput(last_input + i, prev); + } + + return; + } + + // if our current delay is bigger then our new delay the plan is different. + // since we probably already sent the inputs up to _last_received_input + // which includes the previous delay we should not modify any inputs that happend before that + // this will probably mean that it will drop local inputs until the local machine catches up + if (_input_delay > delay) { + _input_delay = delay; + return; + } +} + +u8 Gekko::InputBuffer::GetDelay() +{ + return _input_delay; +} + +void Gekko::InputBuffer::SetInputPredictionWindow(u8 input_window) +{ + _input_prediction_window = input_window; +} + +Frame Gekko::InputBuffer::GetIncorrectPredictionFrame() +{ + return _incorrent_predicted_inputs.empty() ? GameInput::NULL_FRAME : _incorrent_predicted_inputs.front(); +} + +void Gekko::InputBuffer::ResetPrediction() +{ + _first_predicted_input = GameInput::NULL_FRAME; + _last_predicted_input = GameInput::NULL_FRAME; +} + +Frame Gekko::InputBuffer::GetLastReceivedFrame() +{ + return _last_received_input; +} + +void Gekko::InputBuffer::ClearIncorrectFrames(Frame clear_limit) +{ + while (!_incorrent_predicted_inputs.empty() && _incorrent_predicted_inputs.front() <= clear_limit) { + _incorrent_predicted_inputs.pop_front(); + } +} + + +bool Gekko::InputBuffer::HandleInputPrediction(Frame frame) +{ + const u32 prev_input = PreviousFrame(frame); + if (_first_predicted_input == GameInput::NULL_FRAME) { + // if the prev frame happends to be an empty frame add a dummy input + if (_inputs[prev_input % _buff_size]->frame == GameInput::NULL_FRAME) { + _inputs[frame % _buff_size]->Init(frame, _empty_input.get(), _input_size); + } + else { + // predict by copying the previous input + _inputs[frame % _buff_size]->Init(_inputs[prev_input % _buff_size].get()); + _inputs[frame % _buff_size]->frame = frame; + } + // set prediction values + _first_predicted_input = frame; + _last_predicted_input = frame; + return true; + } + else { + // continue predicting if the diff is within the window + const u32 diff = _last_predicted_input - _first_predicted_input + 1; + if (_input_prediction_window > diff) { + _inputs[frame % _buff_size]->Init(_inputs[prev_input % _buff_size].get()); + _inputs[frame % _buff_size]->frame = frame; + // move the last predicted input along with the requested frame + _last_predicted_input = frame; + return true; + } + return false; + } +} + +bool Gekko::InputBuffer::CanPredictInput() { + const Frame diff = std::abs(_last_predicted_input) - std::abs(_first_predicted_input) + 1; + return _input_prediction_window > 0 && diff < _input_prediction_window; +} + +u32 Gekko::InputBuffer::PreviousFrame(Frame frame) +{ + return frame - 1 < 0 ? _buff_size - frame - 1 : frame - 1; +} + +void Gekko::InputBuffer::SetRunaheadMode(bool running_ahead) +{ + _running_ahead = running_ahead; +} + +std::unique_ptr Gekko::InputBuffer::GetInput(Frame frame, bool prediction) +{ + auto inp = std::make_unique(); + + if (_last_received_input < frame) { + // no input? check if we should predict the input + if (prediction) { + if (_last_predicted_input != GameInput::NULL_FRAME && + frame <= _last_predicted_input) { + // return existing prediction + inp->Init(_inputs[frame % _buff_size].get()); + + } else if (!_running_ahead && CanPredictInput() && HandleInputPrediction(frame)) { + // generate new prediction + inp->Init(_inputs[frame % _buff_size].get()); + + } else if (_running_ahead) { + // return last known input without advancing prediction state + const Frame ref = _last_predicted_input != GameInput::NULL_FRAME ? _last_predicted_input : _last_received_input; + if (ref != GameInput::NULL_FRAME) { + inp->Init(_inputs[ref % _buff_size].get()); + } + } + } + return inp; + } + + if (_inputs[frame % _buff_size]->frame != frame || + _inputs[frame % _buff_size]->frame == GameInput::NULL_FRAME) { + return inp; + } + + inp->Init(_inputs[frame % _buff_size].get()); + return inp; +} + +void Gekko::GameInput::Init(GameInput* other) +{ + frame = other->frame; + input_len = other->input_len; + + if (input) { + std::memcpy(input.get(), other->input.get(), input_len); + return; + } + + input = std::make_unique(input_len); + + if (input) { + std::memcpy(input.get(), other->input.get(), input_len); + } +} + +void Gekko::GameInput::Init(Frame frame_num, u8* inp, u32 inp_len) +{ + frame = frame_num; + input_len = inp_len; + + if (input) { + std::memcpy(input.get(), inp, input_len); + return; + } + + input = std::make_unique(input_len); + + if (input) { + std::memcpy(input.get(), inp, input_len); + } +} + +bool Gekko::GameInput::IsEqualTo(u8* other) +{ + return input_len != 0 && std::memcmp(input.get(), other, input_len) == 0; +} + +void Gekko::GameInput::Clear() +{ + std::memset(input.get(), 0, input_len); + frame = NULL_FRAME; +} + +Gekko::GameInput::GameInput() +{ + frame = NULL_FRAME; + input_len = 0; + input = std::unique_ptr(); +} diff --git a/Externals/GekkoNet/src/net.cpp b/Externals/GekkoNet/src/net.cpp new file mode 100644 index 000000000000..3b98c53ce1ed --- /dev/null +++ b/Externals/GekkoNet/src/net.cpp @@ -0,0 +1,118 @@ +#include "gekko_types.h" +#include "net.h" + +#include +#include + +u32 Gekko::NetAddress::GetSize() +{ + return _size; +} + +Gekko::NetAddress::NetAddress(void* data, u32 size) +{ + _size = size; + _data = std::make_unique(_size); + // copy address data + std::memcpy(_data.get(), data, _size); +} + +Gekko::NetAddress::NetAddress() +{ + _size = 0; + _data = nullptr; +} + +void Gekko::NetAddress::Copy(NetAddress* other) +{ + if (!other) { + return; + } + + _size = other->_size; + + if (_data) { + _data.reset(); + } + + _data = std::make_unique(_size); + // copy address data + std::memcpy(_data.get(), other->GetAddress(), _size); +} + +bool Gekko::NetAddress::Equals(NetAddress& other) +{ + return _size == other._size && std::memcmp(_data.get(), other._data.get(), _size) == 0; +} + +u8* Gekko::NetAddress::GetAddress() +{ + return _data.get(); +} + +float Gekko::NetStats::CalculateJitter() +{ + if (rtt.size() < 2) { + return 0.f; + } + + float sum = 0.f; + for (i32 i = 1; i < rtt.size(); ++i) { + sum += std::abs((float)rtt[i] - (float)rtt[i - 1]); + } + + float jitter = sum / (rtt.size() - 1); + return jitter; +} + +float Gekko::NetStats::CalculateAvgRTT() +{ + if (rtt.empty()) { + return 0.f; + } + + float sum = 0.f; + for (i32 i = 0; i < rtt.size(); i++) { + sum += rtt[i]; + } + + float avg_rtt = sum / rtt.size(); + return avg_rtt; +} + +void Gekko::NetStats::AddRTT(u16 rtt_ms) +{ + rtt.push_back(rtt_ms); + while (rtt.size() > RTT_HISTORY_SIZE) { + rtt.erase(rtt.begin()); + } +} + +void Gekko::NetStats::UpdateBandwidth() +{ + using namespace std::chrono; + u64 now = duration_cast(steady_clock::now().time_since_epoch()).count(); + + if (last_bandwidth_update == 0) { + last_bandwidth_update = now; + return; + } + + u64 elapsed = now - last_bandwidth_update; + if (elapsed >= 1000) { + kb_sent_per_sec = (bytes_sent_accum * 1000.f / elapsed) / 1024.f; + kb_received_per_sec = (bytes_received_accum * 1000.f / elapsed) / 1024.f; + bytes_sent_accum = 0; + bytes_received_accum = 0; + last_bandwidth_update = now; + } +} + +u32 Gekko::NetStats::LastRTT() +{ + if (rtt.empty()) { + return 0; + } + + return rtt.back(); +} diff --git a/Externals/GekkoNet/src/player.cpp b/Externals/GekkoNet/src/player.cpp new file mode 100644 index 000000000000..4550ee15f39c --- /dev/null +++ b/Externals/GekkoNet/src/player.cpp @@ -0,0 +1,38 @@ +#include "backend.h" + +Gekko::Player::Player(Handle phandle, GekkoPlayerType type, NetAddress* addr, u32 magic) +{ + handle = phandle; + sync_num = 0; + session_magic = magic; + + address.Copy(addr); + stats = NetStats(); + + stats.last_acked_frame = -1; + stats.last_sent_sync_message = 0; + + _type = type; + _status = _type == GekkoLocalPlayer ? Connected : Initiating; + adv_history.Init(); +} + +GekkoPlayerType Gekko::Player::GetType() +{ + return _type; +} + +Gekko::PlayerStatus Gekko::Player::GetStatus() +{ + return _status; +} + +void Gekko::Player::SetStatus(PlayerStatus type) +{ + _status = type; +} + +void Gekko::Player::SetChecksum(Frame frame, u32 checksum) +{ + session_health[frame] = checksum; +} diff --git a/Externals/GekkoNet/src/spectator_session.cpp b/Externals/GekkoNet/src/spectator_session.cpp new file mode 100644 index 000000000000..33b987e548ff --- /dev/null +++ b/Externals/GekkoNet/src/spectator_session.cpp @@ -0,0 +1,242 @@ +#include "session.h" + + +Gekko::SpectatorSession::SpectatorSession() +{ + _host = nullptr; + _started = false; + _delay_spectator = false; + _last_saved_frame = GameInput::NULL_FRAME - 1; + _config = GekkoConfig(); +} + +void Gekko::SpectatorSession::Init(GekkoConfig* config) +{ + _host = nullptr; + + _started = false; + + // get given configs + std::memcpy(&_config, config, sizeof(GekkoConfig)); + + // setup input buffer for the players (add size for spectator delay) + u32 buffer_size = InputBuffer::DEFAULT_BUFF_SIZE + _config.spectator_delay; + _sync.Init(_config.num_players, _config.input_size, buffer_size); + + // setup message system. + _msg.Init(_config.num_players, _config.input_size); + + // setup game event system + _game_events.Init(_config.input_size * _config.num_players); + + // start paused so the buffer fills before playback begins + _delay_spectator = (_config.spectator_delay > 0); +} + +void Gekko::SpectatorSession::SetLocalDelay(i32 player, u8 delay) +{ + // no-op: spectators don't have local players +} + +void Gekko::SpectatorSession::SetNetAdapter(GekkoNetAdapter* adapter) +{ + _host = adapter; +} + +i32 Gekko::SpectatorSession::AddActor(GekkoPlayerType type, GekkoNetAddress* addr) +{ + const i32 ERR = -1; + std::unique_ptr address; + + // only accept a single remote player (the host) + if (type != GekkoRemotePlayer) { + return ERR; + } + + if (addr == nullptr) { + return ERR; + } + + if (!_msg.remotes.empty()) { + return ERR; + } + + address = std::make_unique(addr->data, addr->size); + + u32 new_handle = (u32)_msg.remotes.size(); + _msg.remotes.push_back(std::make_unique(new_handle, type, address.get())); + + return new_handle; +} + +void Gekko::SpectatorSession::AddLocalInput(i32 player, void* input) +{ + // no-op: spectators don't add local input +} + +GekkoGameEvent** Gekko::SpectatorSession::UpdateSession(i32* count) +{ + // reset session events + _msg.session_events.Reset(); + + // connection handling + Poll(); + + // clear GameEvents + _game_events.Clear(); + + // gameplay + if (AllActorsValid()) { + // reset the game event buffer before doing anything else + _game_events.Reset(); + + // spectator session buffer + if (ShouldDelaySpectator()) { + *count = _game_events.Count(); + return _game_events.Data(); + } + + // then advance the session + if (_game_events.AddAdvanceEvent(_sync, false)) { + _sync.IncrementFrame(); + } + } + + *count = _game_events.Count(); + return _game_events.Data(); +} + +GekkoSessionEvent** Gekko::SpectatorSession::Events(i32* count) +{ + *count = (i32)_msg.session_events.GetRecentEvents().size(); + return _msg.session_events.GetRecentEvents().data(); +} + +f32 Gekko::SpectatorSession::FramesAhead() +{ + return 0.f; +} + +void Gekko::SpectatorSession::NetworkStats(i32 player, GekkoNetworkStats* stats) +{ + for (auto& actor : _msg.remotes) { + if (actor->handle == player) { + actor->stats.UpdateBandwidth(); + stats->kb_sent = actor->stats.kb_sent_per_sec; + stats->kb_received = actor->stats.kb_received_per_sec; + stats->last_ping = actor->stats.LastRTT(); + stats->jitter = actor->stats.CalculateJitter(); + stats->avg_ping = actor->stats.CalculateAvgRTT(); + return; + } + } +} + +void Gekko::SpectatorSession::NetworkPoll() +{ + Poll(); +} + +void Gekko::SpectatorSession::Poll() +{ + // return if no host is defined. + if (!_host) { + return; + } + + // fetch data from network + int length = 0; + auto data = _host->receive_data(&length); + + // process the data we received + _msg.HandleData(_host, data, length); + + // handle received inputs + HandleReceivedInputs(); + + // send network health update + _msg.SendNetworkHealth(); + + // now send data + _msg.SendPendingOutput(_host); +} + +bool Gekko::SpectatorSession::AllActorsValid() +{ + if (!_started) { + if (!_msg.CheckStatusActors()) { + return false; + } + + // if none returned that the session is ready! + _msg.session_events.AddSessionStartedEvent(); + if (_config.spectator_delay > 0) { + _msg.session_events.AddSpectatorPausedEvent(); + } + _started = true; + + return true; + } + + return true; +} + +void Gekko::SpectatorSession::HandleReceivedInputs() +{ + for (auto& remote : _msg.remotes) { + if (remote->GetStatus() != Connected) continue; + + // spectators receive combined inputs for ALL players from the host + std::vector handles; + for (u32 i = 0; i < _config.num_players; i++) { + handles.push_back(i); + } + + for (u32 i = 0; i < handles.size(); i++) { + auto handle = handles[i]; + const Frame last_recv = _sync.GetLastReceivedFrom(handle) + 1; + const Frame last_added = _msg.GetLastAddedInputFrom(handle); + + auto& input_q = _msg.GetNetPlayerQueue(handle); + const Frame min_frame = last_added - (i32)input_q.size() + 1; + for (int i = last_recv; i <= last_added; i++) { + if (i >= min_frame) { + int current_idx = i - min_frame; + u8* input = input_q[current_idx].get(); + _sync.AddRemoteInput(handle, input, i); + _msg.SendInputAck(handle, i, 0); + } + } + } + } +} + +bool Gekko::SpectatorSession::ShouldDelaySpectator() +{ + if (_config.spectator_delay == 0) { + return false; + } + + const u32 delay = _config.spectator_delay; + const Frame current = _sync.GetCurrentFrame(); + const Frame min = _sync.GetMinReceivedFrame(); + const u32 diff = std::abs(min - current); + + if (_delay_spectator) { + if (diff >= delay) { + _delay_spectator = false; + _msg.session_events.AddSpectatorUnpausedEvent(); + return false; + } + return true; + } + + // Re-pause only when the buffer is completely exhausted + if (diff == 0) { + _delay_spectator = true; + _msg.session_events.AddSpectatorPausedEvent(); + return true; + } + + return false; +} diff --git a/Externals/GekkoNet/src/storage.cpp b/Externals/GekkoNet/src/storage.cpp new file mode 100644 index 000000000000..82c87030d4ff --- /dev/null +++ b/Externals/GekkoNet/src/storage.cpp @@ -0,0 +1,34 @@ +#include "storage.h" + +Gekko::StateStorage::StateStorage() +{ + _max_num_states = 0; +} + + +void Gekko::StateStorage::Init(u32 num_states, u32 state_size, bool limited) +{ + const u32 num = limited ? 2 : num_states + 2; + _max_num_states = num; + + for (u32 i = 0; i < _max_num_states; i++) { + _states.push_back(std::make_unique()); + _states.back().get()->state = std::make_unique(state_size); + _states.back().get()->state_len = state_size; + } + + _runahead_state.state = std::make_unique(state_size); + _runahead_state.state_len = state_size; + _runahead_state.frame = GameInput::NULL_FRAME; +} + +Gekko::StateEntry* Gekko::StateStorage::GetRunaheadState() +{ + return &_runahead_state; +} + +Gekko::StateEntry* Gekko::StateStorage::GetState(Frame frame) +{ + frame = frame < 0 ? frame + _max_num_states : frame; + return _states[frame % _max_num_states].get(); +} diff --git a/Externals/GekkoNet/src/stress_session.cpp b/Externals/GekkoNet/src/stress_session.cpp new file mode 100644 index 000000000000..6c24972c7345 --- /dev/null +++ b/Externals/GekkoNet/src/stress_session.cpp @@ -0,0 +1,167 @@ +#include "session.h" +#include + + +Gekko::StressSession::StressSession() +{ + _config = GekkoConfig(); + _sync = SyncSystem(); + _check_distance = 0; +} + +void Gekko::StressSession::Init(GekkoConfig* config) +{ + std::memcpy(&_config, config, sizeof(GekkoConfig)); + + // check distance + _check_distance = _config.check_distance; + + // setup input buffer for the players + _sync.Init(_config.num_players, _config.input_size); + + // setup game event system + _game_events.Init(_config.input_size * _config.num_players); + + // setup state storage + _storage.Init(_check_distance, _config.state_size, false); + + // setup checksum history for comparisons + _checksum_history.clear(); +} + +void Gekko::StressSession::SetLocalDelay(i32 player, u8 delay) +{ + for (u32 i = 0; i < _locals.size(); i++) { + if (_locals[i].handle == player) { + _sync.SetLocalDelay(player, delay); + } + } +} + +void Gekko::StressSession::SetNetAdapter(GekkoNetAdapter* adapter) +{ + // no adapters used in stress sessions. +} + +i32 Gekko::StressSession::AddActor(GekkoPlayerType type, GekkoNetAddress* addr) +{ + if (type != GekkoLocalPlayer) return -1; + + u32 new_handle = (u32)_locals.size(); + std::unique_ptr address; + + if (addr) { + address = std::make_unique(addr->data, addr->size); + } + + _locals.push_back(Player(new_handle, type, address.get())); + + return new_handle; +} + +void Gekko::StressSession::AddLocalInput(i32 player, void* input) +{ + u8* inp = (u8*)input; + + for (u32 i = 0; i < _locals.size(); i++) { + if (_locals[i].handle == player) { + _sync.AddLocalInput(player, inp); + break; + } + } +} + +GekkoGameEvent** Gekko::StressSession::UpdateSession(i32* count) +{ + _game_events.Clear(); + _session_events.Reset(); + _game_events.Reset(); + + Frame current = _sync.GetCurrentFrame(); + if (_check_distance > 0 && current > _check_distance) { + // once whe have gone far enough forward start rollback back and comparing checksums + for (Frame check_frame = current - _check_distance; check_frame < current; check_frame++) { + CheckForDesyncs(check_frame); + } + // rollback according to check distance + HandleRollback(); + } + + if (current - 1 == GameInput::NULL_FRAME - 1) { + // initial gamestate save + _game_events.AddSaveEvent(_sync, _storage); + } + + // advance the session forward + if (_game_events.AddAdvanceEvent(_sync, false)) { + _game_events.AddSaveEvent(_sync, _storage); + _sync.IncrementFrame(); + } + + *count = _game_events.Count(); + return _game_events.Data(); +} + +GekkoSessionEvent** Gekko::StressSession::Events(i32* count) +{ + *count = (i32)_session_events.GetRecentEvents().size(); + return _session_events.GetRecentEvents().data(); +} + +f32 Gekko::StressSession::FramesAhead() +{ + return 0.f; +} + +void Gekko::StressSession::NetworkStats(i32 player, GekkoNetworkStats* stats) +{ + // no stats for now. +} + +void Gekko::StressSession::NetworkPoll() +{ + // stress sessions are local only +} + +void Gekko::StressSession::HandleRollback() +{ + const Frame current = _sync.GetCurrentFrame(); + const Frame past = current - _check_distance; + + // load the sync frame + _sync.SetCurrentFrame(past); + _game_events.AddLoadEvent(_sync, _storage); + _sync.IncrementFrame(); + + for (Frame frame = past + 1; frame < current; frame++) { + _game_events.AddAdvanceEvent(_sync, true); + _game_events.AddSaveEvent(_sync, _storage); + _sync.IncrementFrame(); + } + + // make sure that we are back where we started. + assert(_sync.GetCurrentFrame() == current); +} + +void Gekko::StressSession::CheckForDesyncs(Frame check_frame) +{ + const Frame oldest = check_frame - _check_distance; + + // clear old checksum frames + _checksum_history.erase(_checksum_history.begin(), _checksum_history.lower_bound(oldest)); + + const StateEntry* state = _storage.GetState(check_frame); + // check if there is a valid entry in the state storage + if (state->frame == check_frame) { + if (_checksum_history.count(check_frame)) { + // found in the history! compare checksums and send event if incorrect. + u32 recorded_checksum = _checksum_history[check_frame]; + if (state->checksum != recorded_checksum) { + _session_events.AddDesyncDetectedEvent(check_frame, 0, state->checksum, recorded_checksum); + } + } else { + // not in the history? then add it. + _checksum_history.emplace(check_frame, state->checksum); + } + } +} diff --git a/Externals/GekkoNet/src/sync.cpp b/Externals/GekkoNet/src/sync.cpp new file mode 100644 index 000000000000..b1b516010297 --- /dev/null +++ b/Externals/GekkoNet/src/sync.cpp @@ -0,0 +1,165 @@ +#include "sync.h" + +#include +#include +#include + +Gekko::SyncSystem::SyncSystem() +{ + _current_frame = GameInput::NULL_FRAME; + _input_size = 0; + _num_players = 0; + _input_buffers = nullptr; +} + +void Gekko::SyncSystem::Init(u8 num_players, u32 input_size, u32 buffer_size) +{ + _input_size = input_size; + _num_players = num_players; + _current_frame = GameInput::NULL_FRAME + 1; + + _input_buffers = std::make_unique(num_players); + // on creation setup input buffers + for (int i = 0; i < _num_players; i++) { + _input_buffers[i].Init(0, 0, input_size, buffer_size); + } +} + +void Gekko::SyncSystem::AddLocalInput(Handle player, u8* input) +{ + // drop inputs from incorrect handles + if (player >= _num_players || player < 0) { + return; + } + + _input_buffers[player].AddLocalInput(_current_frame, input); +} + +void Gekko::SyncSystem::AddRemoteInput(Handle player, u8* input, Frame frame) +{ + // drop inputs from incorrect handles + if (player >= _num_players || player < 0) { + return; + } + + _input_buffers[player].AddInput(frame, input); +} + +void Gekko::SyncSystem::IncrementFrame() +{ + _current_frame++; +} + +bool Gekko::SyncSystem::GetSpectatorInputs(std::unique_ptr& inputs, Frame frame) +{ + auto all_input = std::make_unique(_input_size * _num_players); + for (u8 i = 0; i < _num_players; i++) { + auto inp = _input_buffers[i].GetInput(frame); + + if (inp->frame == GameInput::NULL_FRAME) { + return false; + } + + std::memcpy(all_input.get() + (i * _input_size), (void*)inp->input.get(), _input_size); + } + inputs = std::move(all_input); + return true; +} + +void Gekko::SyncSystem::SetRunaheadMode(bool running_ahead) +{ + for (u8 i = 0; i < _num_players; i++) { + _input_buffers[i].SetRunaheadMode(running_ahead); + } +} + +bool Gekko::SyncSystem::GetCurrentInputs(std::unique_ptr& inputs, Frame& frame) +{ + auto all_input = std::make_unique(_input_size * _num_players); + for (u8 i = 0; i < _num_players; i++) { + auto inp = _input_buffers[i].GetInput(_current_frame, true); + + if (inp->frame == GameInput::NULL_FRAME) { + return false; + } + + std::memcpy(all_input.get() + (i * _input_size), (void*)inp->input.get(), _input_size); + } + frame = _current_frame; + inputs = std::move(all_input); + return true; +} + +bool Gekko::SyncSystem::GetLocalInput(Handle player, std::unique_ptr& input, Frame frame) +{ + auto inp = _input_buffers[player].GetInput(frame); + if (inp->frame == GameInput::NULL_FRAME) { + return false; + } + std::memcpy(input.get(), inp->input.get(), _input_size); + return true; +} + +void Gekko::SyncSystem::SetLocalDelay(Handle player, u8 delay) +{ + _input_buffers[player].SetDelay(delay); +} + +u8 Gekko::SyncSystem::GetLocalDelay(Handle player) +{ + return _input_buffers[player].GetDelay(); +} + +void Gekko::SyncSystem::SetInputPredictionWindow(Handle player, u8 input_window) +{ + _input_buffers[player].SetInputPredictionWindow(input_window); +} + +Frame Gekko::SyncSystem::GetCurrentFrame() const +{ + return _current_frame; +} + +void Gekko::SyncSystem::SetCurrentFrame(Frame frame) +{ + _current_frame = frame; +} + +Frame Gekko::SyncSystem::GetMinIncorrectFrame() +{ + Frame min = INT_MAX; + for (i32 i = 0; i < _num_players; i++) { + Frame frame = _input_buffers[i].GetIncorrectPredictionFrame(); + if (frame == GameInput::NULL_FRAME) { + continue; + } + min = std::min(frame, min); + } + return min == INT_MAX ? GameInput::NULL_FRAME : min; +} + +Frame Gekko::SyncSystem::GetMinReceivedFrame() +{ + Frame min = INT_MAX; + for (i32 i = 0; i < _num_players; i++) { + Frame frame = _input_buffers[i].GetLastReceivedFrame(); + min = std::min(frame, min); + } + return min == INT_MAX ? GameInput::NULL_FRAME : min; +} + +Frame Gekko::SyncSystem::GetLastReceivedFrom(Handle player) +{ + if (player >= 0 && player < _num_players) { + return _input_buffers[player].GetLastReceivedFrame(); + } + + return GameInput::NULL_FRAME; +} + +void Gekko::SyncSystem::ClearIncorrectFramesUpTo(Frame clear_limit) +{ + for (i32 i = 0; i < _num_players; i++) { + _input_buffers[i].ClearIncorrectFrames(clear_limit); + } +} diff --git a/Externals/GekkoNet/thirdparty/zpp/LICENSE b/Externals/GekkoNet/thirdparty/zpp/LICENSE new file mode 100644 index 000000000000..ffb4dcdcb536 --- /dev/null +++ b/Externals/GekkoNet/thirdparty/zpp/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2019 Eyal Z + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/Externals/GekkoNet/thirdparty/zpp/README.md b/Externals/GekkoNet/thirdparty/zpp/README.md new file mode 100644 index 000000000000..118e7d006192 --- /dev/null +++ b/Externals/GekkoNet/thirdparty/zpp/README.md @@ -0,0 +1,366 @@ +zpp serializer +============== +A single header only standard C++ serialization framework. + +Abstract +-------- +In C++ there is no standard way of taking an object as-is and transforming it into a language +independent representation, that is, serialize it. + +Serialization frameworks are really common in C++, and they all come with difference promises and have their advantages and disadvantages. +I've had the pleasure to seek a serialization framework that would turn my classes into data +in the most effortless manner, without caring much about the format, and without doing unnecessary logic. + +Some frameworks support many common formats such as json, xml, and such; Some frameworks provide +you with the highest level of performance using zero copy techniques, thus supporting only binary format; +Some frameworks require you to run a script that generates the C++ code that serializes your classes; + +While there are many excellent serialization frameworks, the diversity of the features and complexity often +make them hard to adopt, some of them require you to change existing code to integrate them, or even set up +your build environment differently. + +I finally reached the conclusion that I do not need all those features. +I definitly do not want to either pay unnecessary price in performance to serialize my classes into a textual format, change the code of +my already existing classes, modify my build systems, write my classes in another format and compile it into C++. + +What I needed was to have my classes serialized in a zero overhead manner into binary, with the ability to serialize +objects by their dynamic type, allowing easy dispatch logic between a server and client side, with little to no +change to my already existing classes. + +Motivation +---------- +Provide a single, simple header file, that would enable one to: +* Enable save & load any STL container / string / utility into and from a binary form, in a zero overhead approach. +* Enable save & load any object, by adding only a few lines to any class, without breaking existing code. +* Enable save & load the dynamic type of any object, by a simple one-liner. + +Contents +-------- +* To enable save & load of any object, add the following lines to your class, `object_1, object_2, ...` being the non-static +data members of the class. +```cpp + friend zpp::serializer::access; + template + static void serialize(Archive & archive, Self & self) + { + archive(self.object_1, self.object_2, ...); + } +``` +If your class does not have a default constructor, define one as private. +* To enable save & load of the dynamic type of any object, you have to register it, and have the base type derive from `zpp::serializer::polymorphic`. +Given the classes `v1::protocol::client_hello`, `v1::protocol::server_hello`, and `v1::protocol::sleep`, all derive from `protocol::command`. +```cpp +// protocol.h +class protocol::command : public zpp::serializer::polymorphic +{ +public: + virtual void operator()(protocol::context &) = 0; + virtual ~command() = default; +}; + +// protocol.cpp +namespace +{ +zpp::serializer::register_types< + zpp::serializer::make_type, + zpp::serializer::make_type, + zpp::serializer::make_type, + // ... +> _; +} +``` +* Save and load objects into a vector of data, in this example we show polymorphic serialization which +has an overhead of 8 bytes serialization id, per polymorphic object being serialized. +```cpp +// The data of the objects we serialize, the vector will grow and shrink as we serialize +// data to/from the vector. +std::vector data; + +// Turns an object into data. +zpp::serializer::memory_output_archive out(data); + +// Turns data into objects. +zpp::serializer::memory_input_archive in(data); + +// Create a sleep command. +std::unique_ptr command = std::make_unique(60s); + +// Serialize a unique pointer of an object whose zpp::serializer::polymorphic is a base class, +// prepends 8 bytes of the serialization id, then the derived class is serialized. +out(command); + +// ... +// Deserializes a unique pointer of an object whose zpp::serializer::polymorphic is a base class, +// loads 8 bytes of the serialization id, constructs a `v1::protocol::sleep` then deseializes into it. +in(command); + +// Run the command, any command has its own logic. +(*command)(protocol_context); +``` + +* You can serialize multiple objects in the same line: +```cpp +out(object_1, object_2, ...); +in(object_1, object_2, ...); +``` + +* You can request serializtion without the polymorphic overhead, thus the static type +is serialized and only this type can be loaded in the other end. +``` +out(*command); +in(*command); +``` + +* Serializing STL containers and strings, first stores a 4 byte size, then the elements: +``` +std::vector v = { 1, 2, 3, 4 }; +out(v); +in(v); +``` +The reason why the default size type is of 4 bytes (i.e `std::uint32_t`) is that most programs +almost never reach a case of a container being more than ~4 billion items, and it may be unjust to +pay the price of 8 bytes size by default. + +* For specific size types that are not 4 bytes, use `zpp::serializer::size_is()`: +``` +std::vector v = { 1, 2, 3, 4 }; +out(zpp::serializer::size_is(v)); +in(zpp::serializer::size_is(v)); +``` +Make sure that the size type is large enough for the serialized object, otherwise less items +will be serialized, according to conversion rules of unsigned types. Uncareful use may lead to +erroneuos code. + +* You may use `memory_view_input_archive`/`memory_view_output_archive` that receives a view type or pointer and size rather than +a vector which requires ownership and memory allocation. In contrary to the owning archives, the view types are not altered, and +you should use the `offset()` function to determine the position of the processed input and output. + +* Serialization using argument dependent lookup is also possible: +```cpp +namespace my_namespace +{ +struct adl +{ + int x; + int y; +}; + +template +void serialize(Archive & archive, adl & adl) +{ + archive(adl.x, adl.y); +} + +template +void serialize(Archive & archive, const adl & adl) +{ + archive(adl.x, adl.y); +} +} +``` + +* Objects that do not derive from `zpp::serializer::polymorphic` do not need registration +and have no overhead at all. + +Example +------- +```cpp +#include "serializer.h" +#include +#include + +class point +{ +public: + point() = default; + point(int x, int y) noexcept : + m_x(x), + m_y(y) + { + } + + friend zpp::serializer::access; + template + static void serialize(Archive & archive, Self & self) + { + archive(self.m_x, self.m_y); + } + + int get_x() const noexcept + { + return m_x; + } + + int get_y() const noexcept + { + return m_y; + } + +private: + int m_x = 0; + int m_y = 0; +}; + +class person : public zpp::serializer::polymorphic +{ +public: + person() = default; + explicit person(std::string name) noexcept : + m_name(std::move(name)) + { + } + + friend zpp::serializer::access; + template + static void serialize(Archive & archive, Self & self) + { + archive(self.m_name); + } + + const std::string & get_name() const noexcept + { + return m_name; + } + + virtual void print() const + { + std::cout << "person: " << m_name; + } + +private: + std::string m_name; +}; + +class student : public person +{ +public: + student() = default; + student(std::string name, std::string university) noexcept : + person(std::move(name)), + m_university(std::move(university)) + { + } + + friend zpp::serializer::access; + template + static void serialize(Archive & archive, Self & self) + { + person::serialize(archive, self); + archive(self.m_university); + } + + virtual void print() const + { + std::cout << "student: " << person::get_name() << ' ' << m_university << '\n'; + } + +private: + std::string m_university; +}; + +namespace +{ +zpp::serializer::register_types< + zpp::serializer::make_type, + zpp::serializer::make_type +> _; +} // + +static void foo() +{ + std::vector data; + zpp::serializer::memory_input_archive in(data); + zpp::serializer::memory_output_archive out(data); + + out(point(1337, 1338)); + + point my_point; + in(my_point); + + std::cout << my_point.get_x() << ' ' << my_point.get_y() << '\n'; +} + +static void bar() +{ + std::vector data; + zpp::serializer::memory_input_archive in(data); + zpp::serializer::memory_output_archive out(data); + + std::unique_ptr my_person = std::make_unique("1337", "1337University"); + out(my_person); + + my_person = nullptr; + in(my_person); + + my_person->print(); +} + +static void foobar() +{ + std::vector data; + zpp::serializer::memory_input_archive in(data); + zpp::serializer::memory_output_archive out(data); + + out(zpp::serializer::as_polymorphic(student("1337", "1337University"))); + + std::unique_ptr my_person; + in(my_person); + + my_person->print(); +} +``` + +Freestanding Implementation +-------------------------- +The library also supports experimental freestanding mode, to allow running in an environment +without exceptions and rtti. + +To enable freestanding mode, define `ZPP_SERIALIZER_FREESTANDING` preprocessing macro. + +In this mode polymorphic serialization is not supported, and error checking +is done via return values. + +The returned error type is `zpp::serializer::freestanding::error`. The numeric value of the error is of +the values in the enum class `zpp::serializer::error` and is accessible by `code()` member function. +The error message is accessible by calling the `message()` member function, as a `std::string_view`. + +In this mode serialization functions should be declared with `auto` as the return type, and return the result +from the `archive`, like so: +```cpp + template + static auto serialize(Archive & archive, Self & self) + { + return archive(self.m_x, self.m_y); + } +``` + +Error checking is done like so: +```cpp + std::vector data; + zpp::serializer::memory_input_archive in(data); + zpp::serializer::memory_output_archive out(data); + + if (auto result = out(point(1337, 1338)); !result) { + std::cout << "Error: " << result.code() << " message: " << result.message() << '\n'; + // return failure / throw + } + + point my_point; + if (auto result = in(my_point); !result) { + std::cout << "Error: " << result.code() << " message: " << result.message() << '\n'; + // return failure / throw + } + + std::cout << my_point.get_x() << ' ' << my_point.get_y() << '\n'; +``` + +A Python Version +---------------- +A compact python version of the library can be found here: https://github.com/eyalz800/zpp_serializer_py. +You can use this library to intercommunicate with this one. The python version does not support variant/optional. + +Requirements +------------ +This framework requires a fully compliant C++14 compiler, including RTTI and exceptions enabled. +One can easily overcome the RTTI requirement by using the following project: https://github.com/eyalz800/type_info. +Disclaimer: registering polymorphic types can be slower in C++14 compared to C++17 due to the use of `shared_timed_mutex` instead of `shared_mutex`. diff --git a/Externals/GekkoNet/thirdparty/zpp/serializer.h b/Externals/GekkoNet/thirdparty/zpp/serializer.h new file mode 100644 index 000000000000..8c7adb86afa6 --- /dev/null +++ b/Externals/GekkoNet/thirdparty/zpp/serializer.h @@ -0,0 +1,3282 @@ +#ifndef ZPP_SERIALIZER_H +#define ZPP_SERIALIZER_H + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#if __cplusplus >= 201703L +#include +#include +#endif +#ifdef ZPP_SERIALIZER_FREESTANDING +#include +#else +#include +#include +#include +#endif + +namespace zpp +{ +/** + * Supports serialization of objects and polymorphic objects. + * Example of non polymorphic serialization: + * ~~~ + * class point + * { + * public: + * point() = default; + * point(int x, int y) noexcept : + * m_x(x), + * m_y(y) + * { + * } + * + * friend zpp::serializer::access; + * template + * static void serialize(Archive & archive, Self & self) + * { + * archive(self.m_x, self.m_y); + * } + * + * int get_x() const noexcept + * { + * return m_x; + * } + * + * int get_y() const noexcept + * { + * return m_y; + * } + * + * private: + * int m_x = 0; + * int m_y = 0; + * }; + * + * static void foo() + * { + * std::vector data; + * zpp::serializer::memory_input_archive in(data); + * zpp::serializer::memory_output_archive out(data); + * + * out(point(1337, 1338)); + * + * point my_point; + * in(my_point); + * + * std::cout << my_point.get_x() << ' ' << my_point.get_y() << '\n'; + * } + * ~~~ + * + * Example of polymorphic serialization: + * ~~~ + * class person : public zpp::serializer::polymorphic + * { + * public: + * person() = default; + * explicit person(std::string name) noexcept : + * m_name(std::move(name)) + * { + * } + * + * friend zpp::serializer::access; + * template + * static void serialize(Archive & archive, Self & self) + * { + * archive(self.m_name); + * } + * + * const std::string & get_name() const noexcept + * { + * return m_name; + * } + * + * virtual void print() const + * { + * std::cout << "person: " << m_name; + * } + * + * private: + * std::string m_name; + * }; + * + * class student : public person + * { + * public: + * student() = default; + * student(std::string name, std::string university) noexcept : + * person(std::move(name)), + * m_university(std::move(university)) + * { + * } + * + * friend zpp::serializer::access; + * template + * static void serialize(Archive & archive, Self & self) + * { + * person::serialize(archive, self); + * archive(self.m_university); + * } + * + * virtual void print() const + * { + * std::cout << "student: " << person::get_name() << ' ' << + * m_university << '\n'; + * } + * + * private: + * std::string m_university; + * }; + * + * namespace + * { + * zpp::serializer::register_types< + * zpp::serializer::make_type, + * zpp::serializer::make_type > _; } // + * + * static void foo() + * { + * std::vector data; + * zpp::serializer::memory_input_archive in(data); + * zpp::serializer::memory_output_archive out(data); + * + * std::unique_ptr my_person = + * std::make_unique("1337", "1337University"); out(my_person); + * + * my_person = nullptr; + * in(my_person); + * + * my_person->print(); + * } + * + * static void bar() + * { + * std::vector data; + * zpp::serializer::memory_input_archive in(data); + * zpp::serializer::memory_output_archive out(data); + * + * out(zpp::serializer::as_polymorphic(student("1337", + * "1337University"))); + * + * std::unique_ptr my_person; + * in(my_person); + * + * my_person->print(); + * } + * ~~~ + */ +namespace serializer +{ +#ifdef ZPP_SERIALIZER_FREESTANDING +namespace freestanding +{ +/** + * Returns the error category for a given error code enumeration type, + * using an argument dependent lookup of a user implemented category + * function. + */ +template +decltype(auto) category() +{ + return category(ErrorCode{}); +} + +/** + * The error category which responsible for translating error codes to + * error messages. + */ +class error_category +{ +public: + /** + * Returns the error category name. + */ + virtual std::string_view name() const noexcept = 0; + + /** + * Return the error message for a given error code. + * For success codes, it is unspecified what value is returned. + * For convienience, you may return zpp::error::no_error for success. + * All other codes must return non empty string views. + */ + virtual std::string_view message(int code) const noexcept = 0; + + /** + * Returns true if success code, else false. + */ + bool success(int code) const + { + return code == m_success_code; + } + +protected: + /** + * Creates an error category whose success code is 'success_code'. + */ + constexpr error_category(int success_code) : + m_success_code(success_code) + { + } + + /** + * Destroys the error category. + */ + ~error_category() = default; + +private: + /** + * The success code. + */ + int m_success_code{}; +}; + +/** + * Creates an error category, whose name and success + * code are specified, as well as a message translation + * logic that returns the error message for every error code. + * Note: message translation must not throw. + */ +template +constexpr auto make_error_category(std::string_view name, + ErrorCode success_code, + Messages && messages) +{ + // Create a category with the name and messages. + class category final : public error_category, + private std::remove_reference_t + { + public: + constexpr category(std::string_view name, + ErrorCode success_code, + Messages && messages) : + error_category( + std::underlying_type_t(success_code)), + std::remove_reference_t( + std::forward(messages)), + m_name(name) + { + } + + std::string_view name() const noexcept override + { + return m_name; + } + + std::string_view message(int code) const noexcept override + { + return this->operator()(ErrorCode{code}); + } + + private: + std::string_view m_name; + } category(name, success_code, std::forward(messages)); + + // Return the category. + return category; +} + +/** + * Represents an error to be initialized from an error code + * enumeration. + * The error code enumeration must have 'int' as underlying type. + * Defining an error code enum and a category for it goes as follows. + * Example: + * ~~~ + * namespace my_namespace + * { + * enum class my_error : int + * { + * success = 0, + * something_bad = 1, + * something_really_bad = 2, + * }; + * + * inline const zpp::error_category & category(my_error) + * { + * constexpr static auto error_category = + * zpp::make_error_category("my_category", my_error::success, + * [](auto code) -> std::string_view { + * switch (code) { + * case my_error::success: + * return zpp::error::no_error; + * case my_code:something_bad: + * return "Something bad happened."; + * case my_error::something_really_bad: + * return "Something really bad happened."; + * default: + * return "Unknown error occurred."; + * } + * } + * ); + * return error_category; + * } + * } // my_namespace + * ~~~ + */ +class error +{ +public: + /** + * Disables default construction. + */ + error() = delete; + + /** + * Constructs an error from an error code enumeration, the + * category is looked by using argument dependent lookup on a + * function named 'category' that receives the error code + * enumeration value. + */ + template + error(ErrorCode error_code) : + m_category(std::addressof( + zpp::serializer::freestanding::category())), + m_code(std::underlying_type_t(error_code)) + { + } + + /** + * Constructs an error from an error code enumeration, the + * category is given explicitly in this overload. + */ + template + error(ErrorCode error_code, const error_category & category) : + m_category(std::addressof(category)), + m_code(std::underlying_type_t(error_code)) + { + } + + /** + * Returns the error category. + */ + const error_category & category() const + { + return *m_category; + } + + /** + * Returns the error code. + */ + int code() const + { + return m_code; + } + + /** + * Returns the error message. Calling this on + * a success error is implementation defined according + * to the error category. + */ + std::string_view message() const + { + return m_category->message(m_code); + } + + /** + * Returns true if the error indicates success, else false. + */ + explicit operator bool() const + { + return m_category->success(m_code); + } + + /** + * No error message value. + */ + static constexpr std::string_view no_error{}; + +private: + /** + * The error category. + */ + const error_category * m_category{}; + + /** + * The error code. + */ + int m_code{}; +}; +} // namespace freestanding + +enum class error : int +{ + success = 0, + out_of_range = 1, + variant_is_valueless = 2, + null_pointer_serialization = 3, +}; + +inline const freestanding::error_category & category(error) +{ + constexpr static auto error_category = + freestanding::make_error_category( + "zpp::serializer", + error::success, + [](auto code) -> std::string_view { + switch (code) { + case error::success: + return freestanding::error::no_error; + case error::out_of_range: + return "[zpp::serializer] Out of range error"; + case error::variant_is_valueless: + return "[zpp::serializer] Cannot serialize a " + "valueless variant."; + case error::null_pointer_serialization: + return "[zpp::serializer] Cannot serialize a null " + "pointer."; + default: + return "[zpp::serializer] Unknown error occurred."; + } + }); + return error_category; +} +#endif // ZPP_SERIALIZER_FREESTANDING + +namespace detail +{ +/** + * Map any sequence of types to void. + */ +template +using void_t = void; + +/** + * Tests if all conditions are true, empty means true. + * Example: + * ~~~ + * all_of::value == false + * all_of::value == true + * all_of::value == false + * all_of<>::value == true + * ~~~ + */ +template +struct all_of : std::true_type +{ +}; + +template +struct all_of : std::false_type +{ +}; + +template +struct all_of : all_of +{ +}; + +template <> +struct all_of : std::true_type +{ +}; + +/** + * Remove const of container value_type + */ +template +struct container_nonconst_value_type +{ + using type = std::remove_const_t; +}; + +/** + * Same as above, except in case of std::map and std::unordered_map, and + * similar, we also need to remove the const of the key type. + */ +template