Add NetworkLagDriver for lag simulation over low-level network transport layer

Source/Engine/Networking/Drivers/ENetDriver.cpp

@@ -59,7 +59,7 @@ bool ENetDriver::Initialize(NetworkPeer* host, const NetworkConfig& config)
 {
     _networkHost = host;
     _config = config;
-    _peerMap = Dictionary<uint32, void*>();
+    _peerMap.Clear();
 
     if (enet_initialize() != 0)
     {
@@ -94,7 +94,7 @@ bool ENetDriver::Listen()
     address.host = ENET_HOST_ANY;
 
     // Set host address if needed
-    if (_config.Address != String("any"))
+    if (_config.Address != TEXT("any"))
         enet_address_set_host(&address, _config.Address.ToStringAnsi().GetText());
 
     // Create ENet host
@@ -151,8 +151,7 @@ void ENetDriver::Disconnect()
 void ENetDriver::Disconnect(const NetworkConnection& connection)
 {
     const int connectionId = connection.ConnectionId;
-
-    void* peer = nullptr;
+    ENetPeer* peer;
     if (_peerMap.TryGet(connectionId, peer))
     {
         enet_peer_disconnect_now((ENetPeer*)peer, 0);
@@ -168,86 +167,74 @@ bool ENetDriver::PopEvent(NetworkEvent* eventPtr)
 {
     ENetEvent event;
     const int result = enet_host_service((ENetHost*)_host, &event, 0);
-
     if (result < 0)
         LOG(Error, "Failed to check ENet events!");
-
     if (result > 0)
     {
         // Copy sender data
         const uint32 connectionId = enet_peer_get_id(event.peer);
-        eventPtr->Sender = NetworkConnection();
         eventPtr->Sender.ConnectionId = connectionId;
 
         switch (event.type)
         {
         case ENET_EVENT_TYPE_CONNECT:
             eventPtr->EventType = NetworkEventType::Connected;
-
             if (IsServer())
                 _peerMap.Add(connectionId, event.peer);
             break;
-
         case ENET_EVENT_TYPE_DISCONNECT:
             eventPtr->EventType = NetworkEventType::Disconnected;
-
             if (IsServer())
                 _peerMap.Remove(connectionId);
             break;
-
         case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
             eventPtr->EventType = NetworkEventType::Timeout;
-
             if (IsServer())
                 _peerMap.Remove(connectionId);
             break;
-
         case ENET_EVENT_TYPE_RECEIVE:
             eventPtr->EventType = NetworkEventType::Message;
-
-        // Acquire message and copy message data
             eventPtr->Message = _networkHost->CreateMessage();
             eventPtr->Message.Length = event.packet->dataLength;
-            Memory::CopyItems(eventPtr->Message.Buffer, event.packet->data, event.packet->dataLength);
+            Platform::MemoryCopy(eventPtr->Message.Buffer, event.packet->data, event.packet->dataLength);
             break;
-
         default:
             break;
         }
-        return true; // Event
+
+        // Got event
+        return true;
     }
 
-    return false; // No events
+    // No events
+    return false;
 }
 
 void ENetDriver::SendMessage(const NetworkChannelType channelType, const NetworkMessage& message)
 {
     ASSERT(IsServer() == false);
-
     SendPacketToPeer((ENetPeer*)_peer, channelType, message);
 }
 
 void ENetDriver::SendMessage(NetworkChannelType channelType, const NetworkMessage& message, NetworkConnection target)
 {
     ASSERT(IsServer());
-
-    ENetPeer* peer = *(ENetPeer**)_peerMap.TryGet(target.ConnectionId);
-    ASSERT(peer != nullptr);
-    ASSERT(peer->state == ENET_PEER_STATE_CONNECTED);
-
-    SendPacketToPeer(peer, channelType, message);
+    ENetPeer* peer;
+    if (_peerMap.TryGet(target.ConnectionId, peer) && peer && peer->state == ENET_PEER_STATE_CONNECTED)
+    {
+        SendPacketToPeer(peer, channelType, message);
+    }
 }
 
 void ENetDriver::SendMessage(const NetworkChannelType channelType, const NetworkMessage& message, const Array<NetworkConnection, HeapAllocation>& targets)
 {
     ASSERT(IsServer());
-
+    ENetPeer* peer;
     for (NetworkConnection target : targets)
     {
-        ENetPeer* peer = *(ENetPeer**)_peerMap.TryGet(target.ConnectionId);
-        ASSERT(peer != nullptr);
-        ASSERT(peer->state == ENET_PEER_STATE_CONNECTED);
-
-        SendPacketToPeer(peer, channelType, message);
+        if (_peerMap.TryGet(target.ConnectionId, peer) && peer && peer->state == ENET_PEER_STATE_CONNECTED)
+        {
+            SendPacketToPeer(peer, channelType, message);
+        }
     }
 }

Source/Engine/Networking/Drivers/ENetDriver.h

@@ -45,5 +45,5 @@ private:
     NetworkPeer* _networkHost;
     void* _host = nullptr;
     void* _peer = nullptr;
-    Dictionary<uint32, void*> _peerMap;
+    Dictionary<uint32, struct _ENetPeer*> _peerMap;
 };

Source/Engine/Networking/Drivers/NetworkLagDriver.cpp

@@ -0,0 +1,186 @@
+// Copyright (c) 2012-2022 Wojciech Figat. All rights reserved.
+
+#include "NetworkLagDriver.h"
+#include "ENetDriver.h"
+#include "Engine/Core/Log.h"
+#include "Engine/Engine/Engine.h"
+#include "Engine/Engine/Time.h"
+
+NetworkLagDriver::NetworkLagDriver(const SpawnParams& params)
+    : ScriptingObject(params)
+{
+}
+
+NetworkLagDriver::~NetworkLagDriver()
+{
+    SetDriver(nullptr);
+}
+
+void NetworkLagDriver::SetDriver(INetworkDriver* value)
+{
+    if (_driver == value)
+        return;
+
+    // Cleanup created proxy driver object
+    if (auto* driver = FromInterface(_driver, INetworkDriver::TypeInitializer))
+        Delete(driver);
+
+    _driver = value;
+}
+
+String NetworkLagDriver::DriverName()
+{
+    if (!_driver)
+        return String::Empty;
+    return _driver->DriverName();
+}
+
+bool NetworkLagDriver::Initialize(NetworkPeer* host, const NetworkConfig& config)
+{
+    if (!_driver)
+    {
+        // Use ENet as default
+        _driver = New<ENetDriver>();
+    }
+    if (!_driver)
+    {
+        LOG(Error, "Missing Driver for Network Lag simulation.");
+        return true;
+    }
+    if (_driver->Initialize(host, config))
+        return true;
+    Engine::Update.Bind<NetworkLagDriver, &NetworkLagDriver::OnUpdate>(this);
+    return false;
+}
+
+void NetworkLagDriver::Dispose()
+{
+    if (!_driver)
+        return;
+    Engine::Update.Unbind<NetworkLagDriver, &NetworkLagDriver::OnUpdate>(this);
+    _driver->Dispose();
+    _messages.Clear();
+    _events.Clear();
+}
+
+bool NetworkLagDriver::Listen()
+{
+    if (!_driver)
+        return false;
+    return _driver->Listen();
+}
+
+bool NetworkLagDriver::Connect()
+{
+    if (!_driver)
+        return false;
+    return _driver->Connect();
+}
+
+void NetworkLagDriver::Disconnect()
+{
+    if (!_driver)
+        return;
+    return _driver->Disconnect();
+}
+
+void NetworkLagDriver::Disconnect(const NetworkConnection& connection)
+{
+    if (!_driver)
+        return;
+    _driver->Disconnect(connection);
+}
+
+bool NetworkLagDriver::PopEvent(NetworkEvent* eventPtr)
+{
+    if (!_driver)
+        return false;
+
+    // Try to pop event from the queue
+    for (int32 i = 0; i < _events.Count(); i++)
+    {
+        auto& e = _events[i];
+        if (e.Lag > 0.0)
+            continue;
+
+        *eventPtr = e.Event;
+        _events.RemoveAtKeepOrder(i);
+        return true;
+    }
+
+    // Consume any incoming events
+    while (_driver->PopEvent(eventPtr))
+    {
+        auto& e = _events.AddOne();
+        e.Lag = (double)Lag;
+        e.Event = *eventPtr;
+    }
+    return false;
+}
+
+void NetworkLagDriver::SendMessage(const NetworkChannelType channelType, const NetworkMessage& message)
+{
+    auto& msg = _messages.AddOne();
+    msg.Lag = (double)Lag;
+    msg.Type = 0;
+    msg.Message = message;
+    msg.MessageData.Set(message.Buffer, message.Length);
+}
+
+void NetworkLagDriver::SendMessage(NetworkChannelType channelType, const NetworkMessage& message, NetworkConnection target)
+{
+    auto& msg = _messages.AddOne();
+    msg.Lag = (double)Lag;
+    msg.Type = 1;
+    msg.Message = message;
+    msg.Target = target;
+    msg.MessageData.Set(message.Buffer, message.Length);
+}
+
+void NetworkLagDriver::SendMessage(const NetworkChannelType channelType, const NetworkMessage& message, const Array<NetworkConnection, HeapAllocation>& targets)
+{
+    auto& msg = _messages.AddOne();
+    msg.Lag = (double)Lag;
+    msg.Type = 2;
+    msg.Message = message;
+    msg.Targets = targets;
+    msg.MessageData.Set(message.Buffer, message.Length);
+}
+
+void NetworkLagDriver::OnUpdate()
+{
+    if (!_driver)
+        return;
+
+    // Update all pending messages and events
+    const double deltaTime = Time::Update.UnscaledDeltaTime.GetTotalMilliseconds();
+    for (int32 i = 0; i < _messages.Count(); i++)
+    {
+        auto& msg = _messages[i];
+        msg.Lag -= deltaTime;
+        if (msg.Lag > 0.0)
+            continue;
+
+        // Fix message to point to the current buffer
+        msg.Message.Buffer = msg.MessageData.Get();
+
+        switch (msg.Type)
+        {
+        case 0:
+            _driver->SendMessage(msg.ChannelType, msg.Message);
+            break;
+        case 1:
+            _driver->SendMessage(msg.ChannelType, msg.Message, msg.Target);
+            break;
+        case 2:
+            _driver->SendMessage(msg.ChannelType, msg.Message, msg.Targets);
+            break;
+        }
+        _messages.RemoveAt(i);
+    }
+    for (int32 i = 0; i < _events.Count(); i++)
+    {
+        auto& e = _events[i];
+        e.Lag -= deltaTime;
+    }
+}

Source/Engine/Networking/Drivers/NetworkLagDriver.h

@@ -0,0 +1,77 @@
+// Copyright (c) 2012-2022 Wojciech Figat. All rights reserved.
+
+#pragma once
+
+#include "Engine/Networking/INetworkDriver.h"
+#include "Engine/Networking/NetworkMessage.h"
+#include "Engine/Networking/NetworkConnection.h"
+#include "Engine/Networking/NetworkEvent.h"
+#include "Engine/Scripting/ScriptingObject.h"
+#include "Engine/Core/Collections/Array.h"
+
+/// <summary>
+/// Low-level network transport interface implementation that is proxy of another nested INetworkDriver implementation but with lag simulation feature.
+/// </summary>
+API_CLASS(Namespace="FlaxEngine.Networking", Sealed) class FLAXENGINE_API NetworkLagDriver : public ScriptingObject, public INetworkDriver
+{
+    DECLARE_SCRIPTING_TYPE(NetworkLagDriver);
+
+private:
+    struct LagMessage
+    {
+        double Lag;
+        int32 Type;
+        NetworkChannelType ChannelType;
+        NetworkMessage Message;
+        NetworkConnection Target;
+        Array<NetworkConnection> Targets;
+        Array<byte> MessageData; // TODO: use message buffers cache (of size config.MessageSize) to reduce dynamic memory allocations
+    };
+    struct LagEvent
+    {
+        double Lag;
+        NetworkEvent Event;
+    };
+
+    INetworkDriver* _driver = nullptr;
+    Array<LagMessage> _messages;
+    Array<LagEvent> _events;
+
+public:
+    ~NetworkLagDriver();
+
+    /// <summary>
+    /// Network lag value in milliseconds. Adds a delay between sending and receiving messages (2 * Lag is TTL).
+    /// </summary>
+    API_FIELD() float Lag = 100.0f;
+
+    /// <summary>
+    /// Gets or sets the nested INetworkDriver to use as a proxy with lags.
+    /// </summary>
+    API_PROPERTY() INetworkDriver* GetDriver() const
+    {
+        return _driver;
+    }
+
+    /// <summary>
+    /// Gets or sets the nested INetworkDriver to use as a proxy with lags.
+    /// </summary>
+    API_PROPERTY() void SetDriver(INetworkDriver* value);
+
+public:
+    // [INetworkDriver]
+    String DriverName() override;
+    bool Initialize(NetworkPeer* host, const NetworkConfig& config) override;
+    void Dispose() override;
+    bool Listen() override;
+    bool Connect() override;
+    void Disconnect() override;
+    void Disconnect(const NetworkConnection& connection) override;
+    bool PopEvent(NetworkEvent* eventPtr) override;
+    void SendMessage(NetworkChannelType channelType, const NetworkMessage& message) override;
+    void SendMessage(NetworkChannelType channelType, const NetworkMessage& message, NetworkConnection target) override;
+    void SendMessage(NetworkChannelType channelType, const NetworkMessage& message, const Array<NetworkConnection, HeapAllocation>& targets) override;
+
+private:
+    void OnUpdate();
+};