Make Message and Client pools thread-safe using ConcurrentDictionary

[CrossV4]

Summary of Changes

I have developed a highly efficient, multi-threaded, and thread-safe networking architectural pattern for RiptideNetworking, specifically designed for high-concurrency projects like MMORPGs.

The Problem: Thread Contention & Race Conditions

When offloading network I/O operations to a dedicated background thread (e.g., executing Server.Update() asynchronously to keep the main game loop responsive), serious race conditions occur. The primary culprit is Riptide’s internal message pool (Message.Create(), Message.Release()). Accessing the message pool concurrently from both the Main Thread (Game Logic) and the Background Thread (Network Loop) corrupts internal byte arrays, leading to network desynchronization, corrupted data, and server crashes.

The Solution: The Command Pattern & Asynchronous Action Queue

To bridge the gap between the Main Thread and the Network Thread safely without expensive resource locking, I implemented a custom Command Pattern execution pipeline combined with a ConcurrentQueue.

1. Incoming Request Queuing: Network message handlers running on the background thread no longer execute game logic or manipulate the message pool directly. Instead, they deserialize raw data into lightweight, immutable INetworkCommand structures and enqueue them into a thread-safe ConcurrentQueue.

public static readonly ConcurrentQueue<INetworkCommand> ACTION_QUEUE = new();
public interface INetworkCommand
{
    ushort ClientId { get; }
    void Execute();
}

private class PPromo : INetworkCommand
{
    public ushort ClientId { get; set; }
    public string promocode { get; set; }

    public void Execute()
    {
        if (CheckIsInList(ClientId, out Player player))
        {
            bool success = TryClaimPromoLocal(player, promocode);
            Message message = Message.Create(MessageSendMode.Reliable, ServerToClientId.PromoResult);
            message.AddBool(success);
            NetworkManager.Singleton.Server.Send(message, ClientId);
        }
    }
}
[MessageHandler((ushort)ClientToServerId.PromoRequest)]
private static void PROCMO_ACTIVATE(ushort fromClientId, Message message)
{
    NetworkManager.ACTION_QUEUE.Enqueue(new PPromo() { ClientId = fromClientId, promocode = message.GetString() });
}

2. Main Thread Thread-Safe Execution & Frame Budgeting: The Main Thread dequeues and processes these commands sequentially during its standard Update() loop. To ensure server stability and prevent CPU spikes caused by a massive surge of packets, I integrated a Frame Budget Monitor using a Stopwatch. If processing takes longer than the allocated frame time budget (MaxFrameMs), the execution breaks, deferred actions are carried over to the next frame, and a warning is logged.

private readonly System.Diagnostics.Stopwatch SW = new();
private void Update()
{
        SW.Restart();
        while (ACTION_QUEUE.TryDequeue(out INetworkCommand CR))
        {
            CR.Execute();

            ACTION_REQUEST_COUNTER.AddOrUpdate(CR.ClientId, 0, (o, k) => (byte)(k - 1));

            // Frame Budgeting Control (e.g., ~33.3ms for a 30-Tick Server)
            if (SW.ElapsedMilliseconds > MaxFrameMs)
            {
                Debug.LogWarning($"{{{TICK_FORDEBUG}}} - [Performans] Bu frame'de {SW.ElapsedMilliseconds}ms harcandı! (Dikkat: {CR.GetType().Name}). Kalan istekleri bir sonrakine aktarılıyor.");
                break;
            }
        }
}

3. Isolated Background Network Loop: With the main loop freed from networking I/O overhead, Server.Update() runs continuously on a high-priority background worker thread, throttled naturally to prevent CPU core exhaustion.

            SERVER_THREAD = new Thread(() => ServerThreadLoop())
            {
                Name = "MokSha-Rin Riptide Server",
                IsBackground = true
            };
            SERVER_THREAD.Start();
    private void ServerThreadLoop()
    {
        try
        {
            while (!GLOBAL_SOURCE.Token.IsCancellationRequested && !QUIT_FLAG)
            {
                // read Riptide packages and fill the actionqueue
                if (Server != null && Server.IsRunning)
                {
                    Server.Update();
                }

                //preventing cpu lock making the thread sleep for a tiny duration and also check for requesting server is going to shutdown
                Thread.Sleep(16);
                GLOBAL_SOURCE.Token.ThrowIfCancellationRequested();
            }
        }
        catch (Exception ex)
        {
            Debug.LogError($"{{{TICK_FORDEBUG}}} - [SERVER - CORE] Server error: {ex.Message}");
        }
        finally
        {
            Debug.Log($"{{{TICK_FORDEBUG}}} - [SERVER - CORE] Server has shutdown (Loop end).");
        }
    }

4. Why we need this ?

public void Execute()
    {
        if (CheckIsInList(ClientId, out Player player))
        {
            bool success = TryClaimPromoLocal(player, promocode);
            Message message = Message.Create(MessageSendMode.Reliable, ServerToClientId.PromoResult);
            message.AddBool(success);
            NetworkManager.Singleton.Server.Send(message, ClientId);
        }
    }

Message message = Message.Create(MessageSendMode.Reliable, ServerToClientId.PromoResult);
Why This is Vital for the Community
By enforcing this design, Riptide’s pooling mechanism (Message.Create) is only ever touched from a single thread sequence during actual processing, completely eliminating Riptide-pool-related Multi-Thread crashes. I wanted to share this architectural solution with the Riptide community to help anyone struggling to transition their dedicated servers to an asynchronous, high-performance, and crash-proof multi-threaded environment. Hope it helps!

[tom-weiland]

Thanks for the PR, Erol!

Regarding the // Modified from Erol Bircan lines you added, I'd prefer to keep file headers limited to mentioning the license and the primary copyright to avoid "header bloat." The commit history (and git blame) already tracks individual contributions and modifications, and it does so much more precisely than manual comments like this do.

Additionally, a number of your code comments read like they were AI generated and they don't fit in with the rest of the code-base, and some of the XML doc comments (the /// ones) you added have typos/grammatical mistakes or don't make a whole lot of sense.

Could you please remove those lines from the headers and do an "AI tone"/grammar/typo pass on your comments? (If English is not your first language I can also try to find some time to help with this.) I’m happy to take another look at the logic changes once that's done 🙂

[CrossV4]

uhh yess, i am a turkish guy i couldnt properly write a professional english so i wote the lines in turkish then i translate them with ai 😅😅.

Okay, i will remove the command lines and i will try to write it withouit ai (but i dont think i will writte them professionally)

[CrossV4]

Summary & Purpose of the Additional Changes

In the original implementation, when a message sequence/order mismatch or an over-read (capacity overflow) occurs, Riptide primarily logs a warning. While this is sufficient for standard debugging and local testing, it creates a vulnerability in production environments—especially in competitive or secure multiplayer games. Malicious clients could intentionally send malformed or out-of-order payloads to manipulate the server state or trigger unexpected behavior without being disconnected.

To address this, these additional changes introduce an explicit Exception Prevention Toggle (preventException). By turning this feature off, developers can force Riptide to throw a strict ReadingOrderOrCapacityException whenever a serialization mismatch or unexpected message sequence is detected. This allows the application layer to catch the exception, identify potential network attacks/exploits, and immediately disconnect malicious clients.

---

API Usage & Configuration

The feature is fully backwards-compatible as preventException defaults to true (maintaining Riptide's default warning-only behavior). Developers can explicitly opt-in to strict exception throwing via initialization parameters or standalone method calls.

For Servers:

// Via initialization
Server.Start(Port, MaximumClient, 0, true, false); // Last parameter sets preventException to false

// Or via standalone method at runtime
Server.SetPreventException(false);

For Clients:

// Via initialization
Client.Connect("1.1.1.1:11", 5, 0, null, true, false); // Last parameter sets preventException to false

// Or via standalone method at runtime
Client.ChangeExceptionPrevention(false);

---

Practical Example: Anti-Cheat & Exploit Mitigation

With preventException set to false, you can safely wrap your message handlers in a try-catch block to detect structural manipulation or malicious packet tampering:

[MessageHandler((ushort)ClientToServerId.Equip)]
private static void CLIENT_EQUIP(ushort fromClientId, Message message)
{
    try
    {
        NetworkManager.ACTION_QUEUE.Enqueue(new PEquip() 
        { 
            ClientId = fromClientId, 
            A = (ArmoryType)(message.GetByte() % (byte)ArmoryType.ArmoryTypeEnd) 
        });
    }
    catch (Riptide.ReadingOrderOrCapacityException ex)
    {
        // Caught possible network attack, structural exploit, or synchronization bug!
        Console.WriteLine($"Security alert while reading EQUIP message from Client {fromClientId}: {ex.Message}");
        
        // Immediately disconnect the client and flag them
        NetworkManager.Singleton.Server.DisconnectClient(
            fromClientId, 
            Message.Create().AddByte((byte)AcceptantFlags.PossibleTriedCheat)
        );
    }
}

Why this is necessary for production environments:

1. Security & Exploit Prevention: Prevents hackers from flooding or fuzzing messages to desynchronize the internal bit stream or trigger logic bypasses.
2. Immediate Threat Containment: Gives developers direct control to instantly kick or ban a player the exact moment a packet structure anomaly occurs.
3. Opt-in Flexibility: Zero breaking changes for existing projects since the default behavior remains unchanged (true).