[Showcase] ENet + Unity ECS (5000 real time player simulation)

https://www.gamasutra.com/view/feature/129854/propagation_of_visual_entity_.php?page=6
https://web.archive.org/web/20181107181434/https://gafferongames.com/post/snapshot_compression/

 

you are awesome !!!!!!
thank you

 

Compression, shorts for xyz on position, and then smallest three compression on rotation most likely (or compressing each xyzw on the quat to 1 byte each)

 

Compression algorithms designed for VLQ are available here. Depending on values range, you can achieve even smaller final size of encoded data.

 

Hello! Very nice demonstration!

I'm attempting to implement something very similar, but I'm having a hard time with the ringbuffer. I made a simple test with a ringbuffer of IntPtr (or actually a struct with intpr and length but maybe I should just assume a static length) and when I read it I don't always get the expected data. I was suspecting I wouldn't because the packet gets disposed immediately upon recieving (as per the examples) meaning it will free the data (I assume) and then that data can be overwritten by something completely random potentially.

Is the idea to not dispose the packet until after deserialize? It doesn't feel super obvious how to know when it's safe to dispose, since my quick attempt at disposing it in the consumer thread made a good ol' crash. Should I make a buffer of packets requiring destruction by the producer thread after consumer has done stuff? I'm getting the feeling I'm missing something obvious here.

Or am I simply required to copy the data off the packet?

 

I was thinking about putting the whole packet into ringbuffer from network thread.
Then read in logic thread and enqueue whole packet into second ringbuffer, which will be consumed by network thread and disposed, that way you won't copy/allocate anything else

But performance should be tested, I personally just allocate memory for whole payload and copy data to that, then dispose packet

 

If you are pre-allocating packets and then putting into the queue, you should never dispose them since ENet does that for you automatically (read common mistakes). If a packet was disposed, it means that unmanaged memory was freed, so ENet will be unable to access it. You should dispose a packet only at receiving and if the packet is not sent further.

If you are using the unmanaged memory allocator for your payload before creating a packet, then you should free it only after enqueuing the packet for sending to ENet and only if you are not using PacketFlags.NoAllocate flag. That way ENet will copy payload internally to make it independent from your application's logic. If you want to avoid memory copying, you can use PacketFlags.NoAllocate flag with Packet.SetFreeCallback(), so ENet will notify when you can free/return back to the pool a memory block.

 

Sorry I forgot to specify I was talking about received packets, those need to be disposed manually.

 

At receiving you can enqueue inter-thread message with a packet for another thread, then deserialize a payload there and only then dispose the packet if your deserialized data now stored somewhere else/processed in place. It's safe to dispose a packet in any thread, the memory there becomes independent as soon as you obtained it from ENet.

 

I will drop you a private message with a sample of implementation.

 

Ah alright! Wonder why it crashed for me then, maybe I did something else wrong that’s unrelated. Unity crashed when I stopped play, and as far as i remember that was the only thing I did, to move deserialize of packet to other thread.

I’ll do some more tests as soon as I have time. I appreciate the explanations a lot

 

Do you stop network thread on exit?

 

Hi Everyone!
im new to ENet and im trying to implement it with ECS.

i did a simple implementation just to make it work than focus on Performance but the Editor is crashing every time i stop the Play Mode.

can anyone help me with that ?

Code (CSharp):

1. #if CLIENT_BUILD
2. using CWBR.Client.Components;
3. using ENet;
4. using System.Collections;
5. using System.Collections.Generic;
6. using System.Runtime.InteropServices;
7. using Unity.Collections;
8. using Unity.Entities;
9. using Unity.Jobs;
10. using Unity.Networking.Transport;
11. using Unity.Networking.Transport.LowLevel.Unsafe;
12. using UnityEngine;
13. [AlwaysUpdateSystem]
14. [UpdateInGroup(typeof(InitializationSystemGroup))]
15. public class ClientEnetManagerSystem : ComponentSystem
16. {
17.     bool InGame = false;
18.     private const int timeLimitToWaitForEvents = 0;
19.     ENet.Event netEvent;
20.     Host client;
21.     public Peer serverPeer;
22.  
23.     protected override void OnCreate()
24.     {
25. #if CLIENT_BUILD && !SERVER_BUILD
26.          ENet.Library.Initialize();
27. #endif
28.     }
29.    
30.     public void JoinServer(uint SecretKeyValue, string Ip, ushort port)
31.     {
32.         Debug.Log("[Client] JoinServer Called");
33.        
34.  
35.         client = new Host();
36.         Address address = new Address();
37.  
38.         address.Port = port;
39.         address.SetIP(Ip);
40.         client.Create();
41.         client.EnableCompression();
42.  
43.         // Connect Server
44.         serverPeer = client.Connect(address, 2, SecretKeyValue);
45.         InGame = true;
46.     }
47.  
48.  
49.     protected override void OnUpdate()
50.     {
51.  
52.         if(InGame == false)
53.             return;
54.        
55.         while (client.Service(timeLimitToWaitForEvents, out netEvent) > 0)
56.         {
57.             switch (netEvent.Type)
58.             {
59.                 case ENet.EventType.None:
60.                     break;
61.  
62.                 case ENet.EventType.Connect:
63.                     Debug.Log("[Client] Client connected to server");
64.  
65.                     break;
66.  
67.                 case ENet.EventType.Disconnect:
68.                     Debug.Log("[Client] Client disconnected from server");
69.                     break;
70.  
71.                 case ENet.EventType.Timeout:
72.                     Debug.Log("[Client] Client connection timeout");
73.                     break;
74.  
75.                 case ENet.EventType.Receive:
76.  
77.                     DataStreamReader reader;
78.                     unsafe {
79.                         reader =  DataStreamUnsafeUtility.CreateReaderFromExistingData((byte*)netEvent.Packet.NativeData, netEvent.Packet.Length);
80.                     }
81.  
82.                     Debug.Log($"[Client] Got Message From Server with length : {reader.Length}");
83.                     //dispose the packet
84.                
85.                     netEvent.Packet.Dispose();
86.                     break;
87.             }
88.         }
89.  
90.         return;
91.     }
92.  
93.  
94.  
95.  
96.  
97.     public void Disconnect()
98.     {
99.         InGame = false;
100.  
101.         Debug.Log($"[Client] Bytes Sent : {client.BytesSent}");
102.         Debug.Log($"[Client] Bytes Received : {client.BytesReceived}");
103.        
104.         client.Flush();
105.       //  client.Dispose();
106.     }
107.  
108.     protected override void OnDestroy()
109.     {
110.  
111. #if CLIENT_BUILD && !SERVER_BUILD
112.         ENet.Library.Deinitialize();
113. #endif
114.     }
115. }
116. #endif

Code (CSharp):

1. #if SERVER_BUILD
2. using CWBR.ClientAndServer.Structs;
3. using CWBR.Server.Components;
4. using ENet;
5. using System;
6. using System.Collections;
7. using System.Collections.Generic;
8. using System.Runtime.InteropServices;
9. using Unity.Collections;
10. using Unity.Entities;
11. using Unity.Jobs;
12. using Unity.Networking.Transport;
13. using Unity.Networking.Transport.LowLevel.Unsafe;
14. using UnityEngine;
15.  
16.  
17. [AlwaysUpdateSystem]
18. [UpdateInGroup(typeof(InitializationSystemGroup))]
19. public class ServerEnetManagerSystem : ComponentSystem
20. {
21.     // Max Players that can join this Server
22.     const int peerLimit = 32;
23.  
24.     private const int timeLimitToWaitForEvents = 0;
25.     const string localIp = "127.0.0.1";
26.     NativeHashMap<uint, byte> PlayerIDFromENetID;
27.     bool startedListening = false;
28.  
29.     ENet.Event netEvent;
30.     Host server;
31.  
32.  
33.     protected override void OnCreate()
34.     {
35.      
36.         Debug.Log("ENet Initialized");
37.         ENet.Library.Initialize();
38.     }
39.  
40.  
41.     public void StartListening(ushort port)
42.     {  
43.         server = new Host();
44.         Address address = new Address();
45.         address.SetIP(localIp);
46.         address.Port = port;
47.         server.Create(address, peerLimit, 2);
48.         server.EnableCompression();
49.         startedListening = true;
50.         Debug.Log("[Server] is Listening");
51.     }
52.      
53.  
54.     protected override void OnUpdate()
55.     {
56.         if (!startedListening)
57.             return;
58.        
59.         while (server.Service(timeLimitToWaitForEvents, out netEvent) > 0)
60.         {
61.             switch (netEvent.Type)
62.             {
63.                 case ENet.EventType.None:
64.                     break;
65.  
66.                 case ENet.EventType.Connect:
67.  
68.                     Debug.Log("[Server] Client connected - ID: " + netEvent.Peer.ID + ", IP: " + netEvent.Peer.IP);
69.  
70.                     // Identify User
71.                     // Player Identified
72.                     int playerIndex = NativeArrayExtensions.IndexOf(ServerGameManager.instance.players_SecretKey, netEvent.Data);
73.                     if (playerIndex >= 0)
74.                     {
75.                         Debug.Log("[SERVER] Player Identified");
76.  
77.                     }
78.                     // Player Failed his identification
79.                     else
80.                     {
81.                         Debug.Log("[SERVER] Player Failed the Identification");
82.  
83.  
84.                         // Close this Connection  
85.                         // Notify Player that he Failed the identification
86.                         netEvent.Peer.DisconnectNow(ENetSuppliedData.IdentificationFailed);
87.  
88.                     }
89.  
90.                     break;
91.  
92.  
93.                 case ENet.EventType.Disconnect | ENet.EventType.Timeout:
94.                     Debug.Log("[Server] Client disconnected - ID: " + netEvent.Peer.ID + ", IP: " + netEvent.Peer.IP);
95.  
96.                     //TODO: Handle Case where all Players are Disconnected
97.                     break;
98.                    
99.                 case ENet.EventType.Receive:
100.                     Debug.Log("[Server] Got received from Client-ID: " + netEvent.Peer.ID + ", IP: " + netEvent.Peer.IP + ", Channel ID: " + netEvent.ChannelID + ", Data length: " + netEvent.Packet.Length);
101.  
102.                     DataStreamReader reader;
103.                     unsafe
104.                     {
105.                         reader = DataStreamUnsafeUtility.CreateReaderFromExistingData((byte*)netEvent.Packet.NativeData, netEvent.Packet.Length);
106.                     }
107.                     //dispose the packet
108.                     netEvent.Packet.Dispose();
109.                     break;
110.             }
111.         }
112.  
113.         return;
114.     }
115.  
116.  
117.  
118.     public void DisconnectAllClients()
119.     {
120.         startedListening = false;
121.  
122.         Debug.Log($"[Server] Bytes Sent : {server.BytesSent}");
123.         Debug.Log($"[Server] Bytes Received : {server.BytesReceived}");
124.  
125.         PlayerIDFromENetID.Dispose();
126.  
127.         server.Flush();
128.       //  server.Dispose();
129.     }
130.  
131.     protected override void OnDestroy()
132.     {
133.         Debug.Log("ENet Deinitialized");
134.         ENet.Library.Deinitialize();
135.     }
136. }
137. #endif

@wobes your Showcase is Awesome!!
is it possible to get an example of code how you are passing The Data between ENet Worker Thread and the MainThread ?


Thanks!

 

@sabriboughanmi01 any chance of getting a stack trace of crash? But my first assumption that you do not break the loop so you cause a while loop death. Sometimes what may happen is that you access already destroyed Peer, so it causes a crash when it tries to get Peer.ID, be aware of it.

Try to use something like

Code (CSharp):

1.   private volatile bool isUpdatingNetworkThread;
2.  
3. while (isUpdatingNetworkThread)
4.              // while service code here
5.  
6.   protected override void OnDestroy()
7.         {
8.             isUpdatingNetworkThread = false;
9.         }
10.  

Best scenario would be to wait for all the ENet events get dispatched and only then break the loop and deinitialize the library. Here you may find a useful example with a most likely proper break: https://github.com/nxrighthere/ENet-CSharp/issues/62

As for IO between ENet worker thread and main thread what I do is I have two RingBuffers of <IntPtr>, IntPtr here is simply a pointer to allocated struct (command, transportEvent) with data Enum, IntPtr where Enum is identifier of command type, such as (kick, start server, connect, disconnect) and IntPtr is the actual payload of the command.

Then I simply dequeue the command and dispatch it based on the command type. Think of it as serialization/deserialization network messages just locally.

Cheers.

 

hi @wobes as you can see in my example im already changing the boolean value that authorizes the Server/Client loops iteration in my "DisconnectAllClients" & "Disconnect" functions they are always called but at the end the Editor still Crash


my first impression is maybe the Server and Client Loops cant be both in the Same Thread.

can you please tell me how can i get the stack trace of the crash ? (Sorry i never did that before )

 

@wobes after reading the Unity Log file, i found out that the ENet.Library.Deinitialize() was called before my Host.Flush() .
so how do you Call the "Deinitialize" after all your systems using the ENet are disposed ?

 

Because you are deinitializing the library immediately instead of the graceful shutdown. You need to use a sort of checkpoint and initiate sequential deinitialization process in `OnDestroy()` method.

 

@nxrighthere do you have a good approach to Send Dynamic Buffers with ENet ?

is it ok to use Host.Flush() multiple times per Frame ?

Code (CSharp):

1. [UpdateInGroup(typeof(PresentationSystemGroup))]
2.     public class ClientEnetSendSystem : ComponentSystem
3.     {
4.         EntityQuery entityQuery;
5.         ClientEnetMassageReceiverSystem clientEnetMassageReceiverSystem;
6.  
7.         protected override void OnCreate()
8.         {
9.             entityQuery = GetEntityQuery(ComponentType.ReadOnly<LocalPlayerTag>(), ComponentType.ReadOnly<LocalPlayerConnectedStatus>(), ComponentType.ReadOnly<OutgoingUnreliableMessagesAuthoring>(), typeof(Peer));
10.             clientEnetMassageReceiverSystem = World.Active.GetOrCreateSystem<ClientEnetMassageReceiverSystem>();
11.         }
12.  
13.         protected override void OnUpdate()
14.         {
15.             Entities.With(entityQuery).ForEach((Entity entity, ref Peer peer) =>
16.             {
17.                 var unreliableMessages =  EntityManager.GetBuffer<OutgoingUnreliableMessages>(entity);
18.  
19.  
20.                 if (unreliableMessages.Length == 0 )
21.                     return;
22.  
23.                 // in Case LocalPlayer was Disconnected in the Middle of the Frame
24.                 if (peer.State != PeerState.Connected)
25.                 {
26.                     unreliableMessages.Clear();
27.                     return;
28.                 }
29.  
30.  
31.                 Packet packet = default;
32.                 unsafe
33.                 {
34.                     packet.Create(new IntPtr(unreliableMessages.GetUnsafePtr()), unreliableMessages.Length, PacketFlags.None);
35.                 }
36.                 peer.Send(0,ref packet);
37.                 clientEnetMassageReceiverSystem.client.Flush();
38.  
39.                 unreliableMessages.Clear();
40.             });
41.         }
42.        
43.     }

 

Nope, I'm not using any ECS abstractions made by Unity. Probably @wobes and @e199 can give some advises since they are both using it extensively.

For a moderate amount of messages - yes. Basically, you trade multiplexing efficiently in favor of latency by sending packets instantly after enqueuing instead of waiting for aggregation.

This check is redundant, such case is not possible. Disconnecting might occur only during events dispatching using the service function since the control flow is single-threaded.

 

You may use this:

Code (CSharp):

1.     public static class Bootstrap
2.     {
3.         [RuntimeInitializeOnLoadMethod(RuntimeInitializeLoadType.SubsystemRegistration)]
4.         private static void InitializeOnSubsystemRegistration()
5.         {
6.             ENet.Library.Initialize();
7.         }
8.  
9.         [RuntimeInitializeOnLoadMethod(RuntimeInitializeLoadType.BeforeSceneLoad)]
10.         private static void InitializeBeforeSceneLoad()
11.         {
12.             PlayerLoopManager.RegisterDomainUnload(Deinitialize, int.MaxValue);      
13.         }
14.  
15.         private static void Deinitialize()
16.         {
17.             ENet.Library.Deinitialize();
18.         }
19.     }

That will ensure that Deinitialize is called after ECS worlds are disposed.

 

@Opeth001 DynamicBuffers of what? Serialized messages? Just messages our just components? There are multiple approaches.

 

All my systems are writing to 3 DynamicBuffers of byte. OutGoingReliableMessages, OutGoingUnreliableMessages, IncommingMessages, Received Messages are writen to the Incomming DynamicBuffer and then processed inside a Job. Messages that need to be sent are collected and writen to the outgoing DynamicBuffers inside Jobs then passed to the ENet Thread, after the sending process the OutgoingDynamicBuffers need be cleared in order to fill them again with new messages.

for the moment I’m trying to make ENet work so im using it into the main thread.

for a more detailed question,
What would be the best approach to pass multiple Peers Data (two Dynamic Buffers each Peer ) to the ENet thread and await the sending process in the main Thread correctly in order to fill them again?

 

If by await you mean blocking the main thread until sending is done then there's not much point in using concurrency because mutual exclusion will lead to thread contention, starvation, cache-coherence traffic, and so on. The primary goal behind the message-passing approach between threads is to avoid any stagnation of progress with a stopless conveyor of data through inter-thread messages.

As an option, you can try to use per entity atomic counters or a sort of lock-free/wait-free concurrent container which will be mapped one to one with entities and serve for an indication of the transmission and buffers state.

 

Thank you!
I will do as you suggested.

Also is there a way to set Custom data like an int or struct to a Peer ?
im looking for a way to link Peers to their entities in Server side, this way i dont have to look for the player entity each time an EventType.Receive is triggered.

 

You can use `Peer.Data` for that and assign a value or a pointer to a memory block for each peer.

 

Awesome Thank you!

 

hello @nxrighthere ,
i have few questions about Enet
1) Does Enet has a Discord channel or a thread here ? (that would be really awesome)
2) does Enet have a simulation pipeline ? ( simulating high latency, packet loss ... )
3) does Enet have examples on how to use Memory callbacks ? (i want to use them with Native containers)

 

Nope, you can use third-party software to simulate bad network conditions or implement such logic in socket functions for sending/receiving messages.

These callbacks are intended only for general-purpose memory management, the example is here.

 

Hi, verry interresting thread guys.

@wobes how did you manage do handle 5k connexions on 1 server ? The limit of the library is 4095, did you modify the sources of the library ?

 

You can slightly modify the protocol to increase that limitation, apply this and then this.

 

OK I see thanks

 

I found him here... https://www.gafferongames.com/#posts