Completely rework the network stack for faster lower latency messages
This commit is contained in:
1 parent
a689f2656f
commit
ca39e184ed
7 files changed
+133
-129
No files matched your search
@@ -23,18 +23,36 @@ namespace Mistox.Client {
|
|||||||
async Task ReceiveThread() {
|
async Task ReceiveThread() {
|
||||||
using( NetworkStream ns = Server.GetStream() ) {
|
using( NetworkStream ns = Server.GetStream() ) {
|
||||||
try {
|
try {
|
||||||
|
byte[] headerBuffer = new byte[DataPacket.HeaderLength];
|
||||||
|
byte[] payloadBuffer = new byte[65535]; // Upper limit for safety
|
||||||
while( ClientInterface.ServerClientInstance.Alive ) {
|
while( ClientInterface.ServerClientInstance.Alive ) {
|
||||||
byte[] StreamData = new byte[1024];
|
|
||||||
int bytesRead = await ns.ReadAsync(StreamData, 0, StreamData.Length);
|
// Read the header off
|
||||||
if( bytesRead > 0 ) {
|
await ns.ReadExactlyAsync(headerBuffer, 0, 3);
|
||||||
dynamic data = Serializer.tReceive( StreamData.Sub( 0, bytesRead ) );
|
|
||||||
if( data != null ) {
|
// Parse the header
|
||||||
ClientInterface.ServerClientInstance.InvokeOnReceived( data );
|
byte packetbyte = headerBuffer[0];
|
||||||
}
|
ushort dataLength = (ushort)((headerBuffer[1] << 8) | headerBuffer[2]);
|
||||||
|
|
||||||
|
// Make sure the data looks valid
|
||||||
|
if (dataLength > payloadBuffer.Length) {
|
||||||
|
throw new Exception("Incoming TCP payload is too large for the pre-allocated buffer.");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Read off the data
|
||||||
|
await ns.ReadExactlyAsync(payloadBuffer, 0, dataLength);
|
||||||
|
|
||||||
|
// Create the dataobject
|
||||||
|
dynamic dataobject = DataPacket.Deserialize( packetbyte, dataLength, payloadBuffer.AsMemory(DataPacket.HeaderLength, dataLength) );
|
||||||
|
|
||||||
|
// Return the object
|
||||||
|
ClientInterface.ServerClientInstance.InvokeOnReceived(dataobject);
|
||||||
|
|
||||||
}
|
}
|
||||||
} catch( Exception e ) {
|
} catch (Exception e) {
|
||||||
Console.WriteLine( "TCP Receive exception : " + e.Message );
|
ClientInterface.ServerClientInstance.Alive = false;
|
||||||
|
Dispose();
|
||||||
|
Console.WriteLine("TCP Receive exception: " + e.Message);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -10,7 +10,6 @@ namespace Mistox.Client {
|
|||||||
public class mUDPClient : IDisposable {
|
public class mUDPClient : IDisposable {
|
||||||
|
|
||||||
Socket udpClient;
|
Socket udpClient;
|
||||||
|
|
||||||
IPEndPoint remoteHost;
|
IPEndPoint remoteHost;
|
||||||
|
|
||||||
public mUDPClient( long RemoteIP, int Port ) {
|
public mUDPClient( long RemoteIP, int Port ) {
|
||||||
@@ -25,14 +24,29 @@ namespace Mistox.Client {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async Task ReceiveThread() {
|
async Task ReceiveThread() {
|
||||||
|
byte[] buffer = new byte[65535];
|
||||||
while( ClientInterface.ServerClientInstance.Alive ) {
|
while( ClientInterface.ServerClientInstance.Alive ) {
|
||||||
try {
|
try {
|
||||||
byte[] buffer = new byte[1024];
|
|
||||||
|
// Read the data out of the UDP packet with a buffer that has the maximum possible UDP packet size
|
||||||
|
// As UDP reads the whole next packet then drops. Im guaranteed to receive 1 packet at a time
|
||||||
int bytesRead = await udpClient.ReceiveAsync( buffer );
|
int bytesRead = await udpClient.ReceiveAsync( buffer );
|
||||||
dynamic data = Serializer.uReceive( buffer.Sub(0, bytesRead) );
|
|
||||||
if( data != null ) {
|
// Create the data readers
|
||||||
ClientInterface.ServerClientInstance.InvokeOnReceived( data );
|
ReadOnlyMemory<byte> dataMemory = buffer.AsMemory(0, bytesRead);
|
||||||
}
|
ReadOnlySpan<byte> dataPacket = dataMemory.Span;
|
||||||
|
|
||||||
|
// Split apart the packet
|
||||||
|
byte dataType = dataPacket[0];
|
||||||
|
ushort DataLength = (ushort)((dataPacket[1] << 8) | dataPacket[2]);
|
||||||
|
ReadOnlyMemory<byte> data = dataMemory.Slice(DataPacket.HeaderLength, DataLength);
|
||||||
|
|
||||||
|
// Extract the object
|
||||||
|
dynamic buildObj = DataPacket.Deserialize( dataType, DataLength, data );
|
||||||
|
|
||||||
|
// Return the object
|
||||||
|
ClientInterface.ServerClientInstance.InvokeOnReceived( buildObj );
|
||||||
|
|
||||||
} catch( Exception e ) {
|
} catch( Exception e ) {
|
||||||
Console.WriteLine( "UDP Receive exception : " + e.Message );
|
Console.WriteLine( "UDP Receive exception : " + e.Message );
|
||||||
}
|
}
|
||||||
@@ -50,4 +64,4 @@ namespace Mistox.Client {
|
|||||||
udpClient = null;
|
udpClient = null;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -5,7 +5,7 @@
|
|||||||
</PropertyGroup>
|
</PropertyGroup>
|
||||||
|
|
||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
<PackageReference Include="MsgPack.Cli" Version="1.0.1" />
|
<PackageReference Include="MessagePack" Version="3.1.7" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
</Project>
|
</Project>
|
||||||
@@ -1,66 +0,0 @@
|
|||||||
using System;
|
|
||||||
|
|
||||||
namespace Mistox {
|
|
||||||
|
|
||||||
static class Extensions {
|
|
||||||
public static T[] Join<T>( this T[] first, T[] second ) {
|
|
||||||
T[] bytes = new T[first.Length + second.Length];
|
|
||||||
Buffer.BlockCopy( first, 0, bytes, 0, first.Length );
|
|
||||||
Buffer.BlockCopy( second, 0, bytes, first.Length, second.Length );
|
|
||||||
return bytes;
|
|
||||||
}
|
|
||||||
|
|
||||||
public static T[] Sub<T>( this T[] data, int index, int length ) {
|
|
||||||
T[] result = new T[length];
|
|
||||||
Array.Copy( data, index, result, 0, length );
|
|
||||||
return result;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
public class Serializer {
|
|
||||||
|
|
||||||
static byte[] TBufferedData = [];
|
|
||||||
public static dynamic tReceive( byte [] BytesRead ) {
|
|
||||||
// Append the new data onto the old data
|
|
||||||
TBufferedData = TBufferedData.Join( BytesRead );
|
|
||||||
|
|
||||||
// If the total of the data is bigger than the header
|
|
||||||
if( TBufferedData.Length > DataPacket.HeaderLength ) {
|
|
||||||
DataPacket packet = new DataPacket() {
|
|
||||||
PacketType = BytesRead[0],
|
|
||||||
DataLength = BitConverter.ToUInt16( BytesRead.Sub(1,2) )
|
|
||||||
};
|
|
||||||
|
|
||||||
// If the total of the data is bigger then the whole packet
|
|
||||||
if (TBufferedData.Length > DataPacket.HeaderLength + packet.DataLength) {
|
|
||||||
packet.PacketData = TBufferedData.Sub( DataPacket.HeaderLength, packet.DataLength );
|
|
||||||
TBufferedData = TBufferedData.Sub( DataPacket.HeaderLength, TBufferedData.Length - ( DataPacket.HeaderLength + packet.DataLength ) );
|
|
||||||
return packet.Deserialize();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
|
|
||||||
static byte[] UBufferedData = [];
|
|
||||||
public static dynamic uReceive( byte [] BytesRead ) {
|
|
||||||
// Append the new data onto the old data
|
|
||||||
UBufferedData = UBufferedData.Join( BytesRead );
|
|
||||||
|
|
||||||
// If the total of the data is bigger than the header
|
|
||||||
if( UBufferedData.Length > DataPacket.HeaderLength ) {
|
|
||||||
DataPacket packet = new DataPacket() {
|
|
||||||
PacketType = BytesRead[0],
|
|
||||||
DataLength = BitConverter.ToUInt16( BytesRead.Sub(1,2) )
|
|
||||||
};
|
|
||||||
|
|
||||||
// If the total of the data is bigger then the whole packet
|
|
||||||
if (UBufferedData.Length > DataPacket.HeaderLength + packet.DataLength) {
|
|
||||||
packet.PacketData = UBufferedData.Sub( DataPacket.HeaderLength, packet.DataLength );
|
|
||||||
UBufferedData = UBufferedData.Sub( DataPacket.HeaderLength, UBufferedData.Length - ( DataPacket.HeaderLength + packet.DataLength ) );
|
|
||||||
return packet.Deserialize();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -19,21 +19,39 @@ namespace Mistox.Server {
|
|||||||
receiveThread.Start();
|
receiveThread.Start();
|
||||||
}
|
}
|
||||||
|
|
||||||
public async Task ReceiveThread() {
|
async Task ReceiveThread() {
|
||||||
while( _client.Connected && _connection.Connected && ServerInterface.ServerInstance.Alive ) {
|
using( NetworkStream ns = _client.GetStream() ) {
|
||||||
try {
|
try {
|
||||||
_connection.LastHeartbeat = DateTime.UtcNow;
|
byte[] headerBuffer = new byte[DataPacket.HeaderLength];
|
||||||
byte[] StreamData = new byte[1024];
|
byte[] payloadBuffer = new byte[65535]; // Upper limit for safety
|
||||||
int bytesRead = await _client.GetStream().ReadAsync(StreamData, 0, StreamData.Length);
|
while( _connection.Connected ) {
|
||||||
if( bytesRead > 0 ) {
|
|
||||||
dynamic data = Serializer.tReceive( StreamData.Sub( 0, bytesRead ) );
|
// Read the header off
|
||||||
if( data != null ) {
|
await ns.ReadExactlyAsync(headerBuffer, 0, 3);
|
||||||
ServerInterface.ServerInstance.InvokeDataReceived( data );
|
|
||||||
|
// Parse the header
|
||||||
|
byte packetbyte = headerBuffer[0];
|
||||||
|
ushort dataLength = (ushort)((headerBuffer[1] << 8) | headerBuffer[2]);
|
||||||
|
|
||||||
|
// Make sure the data looks valid
|
||||||
|
if (dataLength > payloadBuffer.Length) {
|
||||||
|
throw new Exception("Incoming TCP payload is too large for the pre-allocated buffer.");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Read off the data
|
||||||
|
await ns.ReadExactlyAsync(payloadBuffer, 0, dataLength);
|
||||||
|
|
||||||
|
// Create the dataobject
|
||||||
|
dynamic dataobject = DataPacket.Deserialize( packetbyte, dataLength, payloadBuffer.AsMemory(DataPacket.HeaderLength, dataLength) );
|
||||||
|
|
||||||
|
// Return the object
|
||||||
|
ServerInterface.ServerInstance.InvokeDataReceived(dataobject);
|
||||||
|
|
||||||
}
|
}
|
||||||
} catch( Exception ) {
|
} catch (Exception e) {
|
||||||
Console.WriteLine( "User disconnected" );
|
ClientInterface.ServerClientInstance.Alive = false;
|
||||||
Dispose();
|
Dispose();
|
||||||
|
Console.WriteLine("TCP Receive exception: " + e.Message);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,9 +1,8 @@
|
|||||||
using System;
|
using System;
|
||||||
using System.Collections.Generic;
|
using System.Collections.Generic;
|
||||||
using System.IO;
|
|
||||||
using System.Reflection;
|
using System.Reflection;
|
||||||
|
using MessagePack;
|
||||||
using Mistox.Attributes;
|
using Mistox.Attributes;
|
||||||
using MsgPack.Serialization;
|
|
||||||
|
|
||||||
namespace Mistox {
|
namespace Mistox {
|
||||||
public class DataPacket {
|
public class DataPacket {
|
||||||
@@ -11,52 +10,72 @@ namespace Mistox {
|
|||||||
// The length of the packet header in bytes
|
// The length of the packet header in bytes
|
||||||
public static readonly int HeaderLength = 3;
|
public static readonly int HeaderLength = 3;
|
||||||
|
|
||||||
// Data Packet Properties
|
// 8b. 16b. value of datalength
|
||||||
public byte PacketType { get; set; } = 0x00;
|
// Data Packet [ Type | DataLength | Data ]
|
||||||
public ushort DataLength { get; set; } = 0;
|
|
||||||
public byte[] PacketData { get; set; } = [];
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
// 8b. 16b. ?
|
|
||||||
// Data Packet [ Type | DataLength | Data ]
|
|
||||||
|
|
||||||
// Helper classes
|
// Helper classes
|
||||||
public static byte[] Serialize<T>( T Object ) {
|
public static byte[] Serialize<T>( T Object ) {
|
||||||
|
|
||||||
|
// Get your custom packet attribute to find the ID
|
||||||
mPacket attribute = typeof(T).GetCustomAttribute<mPacket>(false);
|
mPacket attribute = typeof(T).GetCustomAttribute<mPacket>(false);
|
||||||
DataPacket packet = new DataPacket();
|
if (attribute == null) {
|
||||||
if (attribute != null) {
|
throw new Exception($"The datastructure {typeof(T).Name} is not an mPacket");
|
||||||
using ( MemoryStream stream = new MemoryStream() ) {
|
|
||||||
MessagePackSerializer serializer = MessagePackSerializer.Get<T>();
|
|
||||||
serializer.Pack( stream, Object );
|
|
||||||
packet.PacketData = stream.ToArray();
|
|
||||||
packet.DataLength = (ushort)packet.PacketData.Length;
|
|
||||||
return new byte[]{ packet.PacketType }.Join( BitConverter.GetBytes( packet.DataLength )).Join(packet.PacketData);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
throw new Exception( "The datastructure " + typeof(T).Name + " is not an mPacket" );
|
|
||||||
|
// Serialize the payload directly to a temporary byte array.
|
||||||
|
byte[] payloadData = MessagePackSerializer.Serialize(Object);
|
||||||
|
ushort dataLength = (ushort)payloadData.Length;
|
||||||
|
|
||||||
|
// Allocate an array for the final output (1 byte Type + 2 bytes Length + Payload)
|
||||||
|
byte[] finalPacket = new byte[HeaderLength + dataLength];
|
||||||
|
|
||||||
|
// Write the 3-byte header directly into the final array
|
||||||
|
finalPacket[0] = attribute.ID;
|
||||||
|
finalPacket[1] = (byte)((dataLength >> 8) & 0xFF); // High byte of ushort
|
||||||
|
finalPacket[2] = (byte)(dataLength & 0xFF); // Low byte of ushort
|
||||||
|
|
||||||
|
// Copy the payload directly into the final array right after the header
|
||||||
|
// This replaces all your .Join() allocations with a single ultra-fast CPU copy
|
||||||
|
Buffer.BlockCopy(payloadData, 0, finalPacket, HeaderLength, dataLength);
|
||||||
|
|
||||||
|
return finalPacket;
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public object Deserialize() {
|
public static object Deserialize( byte PacketType, ushort DataLength, ReadOnlyMemory<byte> Data ) {
|
||||||
Type type = Type.GetType( NetworkPackets[PacketType] );
|
|
||||||
MessagePackSerializer Serilizer = MessagePackSerializer.Get(type);
|
// Sanity check: Ensure the packet we received actually has the payload bytes it claims to have
|
||||||
using( MemoryStream ms = new MemoryStream( PacketData ) ) {
|
if ( DataLength <= Data.Length ) {
|
||||||
return Serilizer.Unpack( ms );
|
throw new Exception("Malformed packet: Payload size mismatch.");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Get the type from the tag
|
||||||
|
if (!NetworkPackets.TryGetValue(PacketType, out Type resolvedType)) {
|
||||||
|
throw new Exception($"Unknown packet type ID: {PacketType}");
|
||||||
|
}
|
||||||
|
|
||||||
|
// 4. Create the low-level reader struct directly over the memory slice
|
||||||
|
var reader = new MessagePackReader(Data);
|
||||||
|
|
||||||
|
// Extract the object
|
||||||
|
dynamic obj = MessagePackSerializer.Deserialize( resolvedType, ref reader, MessagePackSerializerOptions.Standard );
|
||||||
|
|
||||||
|
return obj;
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
// Fills out the NetworkPacket Lookup Dictionary
|
// Fills out the NetworkPacket Lookup Dictionary
|
||||||
public static readonly Dictionary<byte, string> NetworkPackets = new();
|
public static readonly Dictionary<byte, Type> NetworkPackets = new();
|
||||||
public static void BuildNetworkPacketList() {
|
public static void BuildNetworkPacketList() {
|
||||||
Type[] allClasses = Assembly.GetExecutingAssembly().GetTypes();
|
Type[] allClasses = Assembly.GetExecutingAssembly().GetTypes();
|
||||||
foreach( Type cur in allClasses) {
|
foreach( Type cur in allClasses) {
|
||||||
mPacket attribute = cur.GetCustomAttribute<mPacket>(false);
|
mPacket attribute = cur.GetCustomAttribute<mPacket>(false);
|
||||||
if (attribute != null) {
|
if (attribute != null) {
|
||||||
NetworkPackets[attribute.ID] = cur.FullName + "," + cur.Assembly.FullName;
|
NetworkPackets[attribute.ID] = cur;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,9 +1,10 @@
|
|||||||
1. Wire up the event handlers for the Clients
|
0. Messages sent from clients to clients or clients to the server dont have any information about who it came from on the receiving side?
|
||||||
2. Allow the server to setup a server name
|
Probably need to route the event handlers throught the client objects to be able to pull that information off of the packets as well.
|
||||||
3. Allow the client to setup a user name
|
1. Allow the server to setup a server name
|
||||||
4. Have the client connect to the server with a welcome packet
|
2. Allow the client to setup a user name
|
||||||
5. Have the server use the welcome packet to fillout the user information on the server
|
3. Have the client connect to the server with a welcome packet
|
||||||
6. Respond to the client from the server with a server list of UDPEndpoints
|
4. Have the server use the welcome packet to fillout the user information on the server
|
||||||
|
5. Respond to the client from the server with a server list of UDPEndpoints
|
||||||
|
|
||||||
Test that the datapacket is formatted correctly. The packet size might be off from the actual packet size
|
Test that the datapacket is formatted correctly. The packet size might be off from the actual packet size
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user