/// <summary>
/// Gets the buffer of the message.
/// </summary>
- ReadOnlyMemory<byte> Buffer { get; }
+ byte[] Buffer { get; }
+
+ /// <summary>
+ /// Gets the offset of our current <see cref="IMessageReader"/> in the entire <see cref="Buffer"/>.
+ /// </summary>
+ int Offset { get; }
/// <summary>
/// Gets the current position of the reader.
void CopyTo(IMessageWriter writer);
- IMessageReader Slice(int start);
+ void Seek(int position);
- IMessageReader Slice(int start, int length);
+ IMessageReader Copy(int offset = 0);
}
}
+++ /dev/null
-using System;
-using System.Buffers;
-
-namespace Impostor.Hazel
-{
- public readonly struct MessageData
- {
- private readonly IMemoryOwner<byte> _data;
- private readonly int _length;
-
- public MessageData(IMemoryOwner<byte> data, int length)
- {
- _data = data;
- _length = length;
- }
-
- public ReadOnlyMemory<byte> Buffer => _data.Memory.Slice(0, _length);
-
- public void Return()
- {
- _data.Dispose();
- }
- }
-}
\ No newline at end of file
private readonly ObjectPool<MessageReader> _pool;
private bool _inUse;
- public byte Tag { get; private set; }
- public ReadOnlyMemory<byte> Buffer { get; private set; }
- public int Position { get; set; }
- public int Length => Buffer.Length;
+ private byte _tag;
internal MessageReader(ObjectPool<MessageReader> pool)
{
_pool = pool;
}
- public void Update(ReadOnlyMemory<byte> buffer)
- {
- Update(byte.MaxValue, buffer);
- }
+ public byte[] Buffer { get; private set; }
+
+ public int Offset { get; internal set; }
+
+ public int Position { get; internal set; }
+
+ public int Length { get; internal set; }
+
+ public byte Tag { get; private set; }
+
+ private int ReadPosition => Offset + Position;
- public void Update(byte tag, ReadOnlyMemory<byte> buffer)
+ public void Update(byte[] buffer, int offset = 0, int position = 0, int? length = null, byte tag = byte.MaxValue)
{
_inUse = true;
- Tag = tag;
Buffer = buffer;
- Position = 0;
+ Offset = offset;
+ Position = position;
+ Length = length ?? buffer.Length;
+ Tag = tag;
}
internal void Reset()
Tag = byte.MaxValue;
Buffer = null;
+ Offset = 0;
Position = 0;
+ Length = 0;
}
public IMessageReader ReadMessage()
{
var length = ReadUInt16();
var tag = FastByte();
- var pos = Position;
+ var pos = ReadPosition;
Position += length;
var reader = _pool.Get();
- reader.Update(tag, Buffer.Slice(pos, length));
+ reader.Update(Buffer, pos, 0, length, tag);
return reader;
}
public ushort ReadUInt16()
{
- var output = BinaryPrimitives.ReadUInt16LittleEndian(Buffer.Span.Slice(Position));
+ var output = BinaryPrimitives.ReadUInt16LittleEndian(Buffer.AsSpan(ReadPosition));
Position += sizeof(ushort);
return output;
}
public short ReadInt16()
{
- var output = BinaryPrimitives.ReadInt16LittleEndian(Buffer.Span.Slice(Position));
+ var output = BinaryPrimitives.ReadInt16LittleEndian(Buffer.AsSpan(ReadPosition));
Position += sizeof(short);
return output;
}
public uint ReadUInt32()
{
- var output = BinaryPrimitives.ReadUInt32LittleEndian(Buffer.Span.Slice(Position));
+ var output = BinaryPrimitives.ReadUInt32LittleEndian(Buffer.AsSpan(ReadPosition));
Position += sizeof(uint);
return output;
}
public int ReadInt32()
{
- var output = BinaryPrimitives.ReadInt32LittleEndian(Buffer.Span.Slice(Position));
+ var output = BinaryPrimitives.ReadInt32LittleEndian(Buffer.AsSpan(ReadPosition));
Position += sizeof(int);
return output;
}
public unsafe float ReadSingle()
{
- var output = BinaryPrimitives.ReadSingleLittleEndian(Buffer.Span.Slice(Position));
+ var output = BinaryPrimitives.ReadSingleLittleEndian(Buffer.AsSpan(ReadPosition));
Position += sizeof(float);
return output;
}
public string ReadString()
{
var len = ReadPackedInt32();
- var output = Encoding.UTF8.GetString(Buffer.Span.Slice(Position, len));
+ var output = Encoding.UTF8.GetString(Buffer.AsSpan(ReadPosition, len));
Position += len;
return output;
}
public ReadOnlyMemory<byte> ReadBytes(int length)
{
- var output = Buffer.Slice(Position, length);
+ var output = Buffer.AsMemory(ReadPosition, length);
Position += length;
return output;
}
{
writer.Write((ushort) Length);
writer.Write((byte) Tag);
- writer.Write(Buffer);
+ writer.Write(Buffer.AsMemory(Offset, Length));
}
- public IMessageReader Slice(int start)
+ public void Seek(int position)
{
- var reader = _pool.Get();
- reader.Update(Tag, Buffer.Slice(start));
- return reader;
+ Position = position;
}
- public IMessageReader Slice(int start, int length)
+ public IMessageReader Copy(int offset = 0)
{
var reader = _pool.Get();
- reader.Update(Tag, Buffer.Slice(start, length));
+ reader.Update(Buffer, Offset + offset, Position, Length - offset, Tag);
return reader;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private byte FastByte()
{
- return Buffer.Span[Position++];
+ return Buffer[Offset + Position++];
}
public void Dispose()
private readonly Timer _reliablePacketTimer;
private readonly SemaphoreSlim _connectWaitLock;
- private readonly MemoryPool<byte> _pool;
- private readonly Channel<MessageData> _channel;
+ private readonly ArrayPool<byte> _pool;
+ private readonly Channel<byte[]> _channel;
private Task _listenTask;
private Task _handleTask;
_reliablePacketTimer = new Timer(ManageReliablePacketsInternal, null, 100, Timeout.Infinite);
_connectWaitLock = new SemaphoreSlim(1, 1);
- _pool = MemoryPool<byte>.Shared;
- _channel = Channel.CreateUnbounded<MessageData>(new UnboundedChannelOptions
+ _pool = ArrayPool<byte>.Shared;
+ _channel = Channel.CreateUnbounded<byte[]>(new UnboundedChannelOptions
{
SingleReader = true,
SingleWriter = true
return;
}
- await HandleAsync(data.Buffer);
- }
- }
-
- private async ValueTask HandleAsync(ReadOnlyMemory<byte> memory)
- {
- // Rent memory.
- var dest = _pool.Rent(memory.Length);
-
- // Copy data.
- memory.CopyTo(dest.Memory);
-
- try
- {
// Write to client.
- await Pipeline.Writer.WriteAsync(new MessageData(dest, memory.Length));
- }
- catch (ChannelClosedException)
- {
- // Clean up.
- dest.Dispose();
+ await Pipeline.Writer.WriteAsync(data.Buffer);
}
}
{
if (await ProcessReliableReceive(message.Buffer, 1))
{
- await InvokeDataReceived(message.Slice(3), MessageType.Reliable);
+ message.Offset += 3;
+ message.Length -= 3;
+ message.Position = 0;
+
+ await InvokeDataReceived(message, MessageType.Reliable);
}
Statistics.LogReliableReceive(message.Length - 3, message.Length);
_readerPool = readerPool;
_stoppingCts = new CancellationTokenSource();
- Pipeline = Channel.CreateUnbounded<MessageData>(new UnboundedChannelOptions
+ Pipeline = Channel.CreateUnbounded<byte[]>(new UnboundedChannelOptions
{
SingleReader = true,
SingleWriter = true
});
}
- internal Channel<MessageData> Pipeline { get; }
+ internal Channel<byte[]> Pipeline { get; }
public Task StartAsync()
{
private async Task ReadAsync()
{
- // Read loop.
+ var reader = new MessageReader(_readerPool);
+
while (!_stoppingCts.IsCancellationRequested)
{
var result = await Pipeline.Reader.ReadAsync(_stoppingCts.Token);
try
{
- using (var reader = _readerPool.Get())
- {
- reader.Update(byte.MaxValue, result.Buffer);
- await HandleReceive(reader);
- }
+ reader.Update(result);
+
+ await HandleReceive(reader);
}
catch (Exception e)
{
Dispose(true);
break;
}
- finally
- {
- result.Return();
- }
- }
-
- // Exhaust pipeline.
- while (Pipeline.Reader.TryRead(out var mem))
- {
- mem.Return();
}
}
// Slice 4 bytes to get handshake data.
if (_listener != null)
{
- await _listener.InvokeNewConnection(message.Slice(4), this);
+ using (var handshake = message.Copy(4))
+ {
+ await _listener.InvokeNewConnection(handshake, this);
+ }
}
}
- switch (message.Buffer.Span[0])
+ switch (message.Buffer[0])
{
//Handle reliable receives
case (byte)MessageType.Reliable:
//Handle acknowledgments
case (byte)UdpSendOption.Acknowledgement:
- AcknowledgementMessageReceive(message.Buffer.Span);
+ AcknowledgementMessageReceive(message.Buffer);
break;
//We need to acknowledge hello and ping messages but dont want to invoke any events!
break;
case (byte)UdpSendOption.Disconnect:
- await DisconnectRemote("The remote sent a disconnect request", message.Slice(1));
+ using (var reader = message.Copy(1))
+ {
+ await DisconnectRemote("The remote sent a disconnect request", reader);
+ }
break;
//Treat everything else as unreliable
default:
- await InvokeDataReceived(message.Slice(1), MessageType.Unreliable);
+ using (var reader = message.Copy(1))
+ {
+ await InvokeDataReceived(reader, MessageType.Unreliable);
+ }
Statistics.LogUnreliableReceive(message.Length - 1, message.Length);
break;
}
await client.StartAsync();
}
- await WriteToClientAsync(client, data.Buffer);
+ // Write to client.
+ await client.Pipeline.Writer.WriteAsync(data.Buffer);
}
}
catch (Exception e)
}
}
- private async ValueTask WriteToClientAsync(UdpConnection client, ReadOnlyMemory<byte> memory)
- {
- // Rent memory.
- var dest = _pool.Rent(memory.Length);
-
- // Copy data.
- memory.CopyTo(dest.Memory);
-
- try
- {
- // Write to client.
- await client.Pipeline.Writer.WriteAsync(new MessageData(dest, memory.Length));
- }
- catch (ChannelClosedException)
- {
- // Clean up.
- dest.Dispose();
- }
- }
-
#if DEBUG
public int TestDropRate = -1;
private int dropCounter = 0;
var toPlayer = flag == MessageFlags.GameDataTo;
// Handle packet.
- var readerCopy = reader.Slice(reader.Position);
+ using var readerCopy = reader.Copy();
// TODO: Return value, either a bool (to cancel) or a writer (to cancel (null) or modify/overwrite).
try
public override async ValueTask HandleMessageAsync(IMessageReader reader, MessageType messageType)
{
- var messageCopy = reader.Slice(0);
+ using var messageCopy = reader.Copy();
// Trigger connect event.
if (_isFirst)
context.Writer.Write((byte) messageType);
context.Writer.Write((byte) reader.Tag);
context.Writer.Write((int) reader.Buffer.Length);
- context.Writer.Write(reader.Buffer.Span);
+ context.Writer.Write(reader.Buffer);
}
private static void WriteGameCode(PacketSerializationContext context, in GameCode gameCode)
Assert.Equal(Test1, message.ReadSingle());
}
+ [Fact]
+ public void CopyMessage()
+ {
+ var readerPool = CreateReaderPool();
+
+ // Create message.
+ const int msgLength = 18;
+ const byte Test1 = 12;
+ const byte Test2 = 146;
+
+ var msg = new MessageWriter(2048);
+
+ msg.StartMessage(1);
+ msg.StartMessage(2);
+ msg.Write(Test1);
+ msg.Write(Test2);
+ msg.StartMessage(2);
+ msg.Write(Test1);
+ msg.Write(Test2);
+ msg.StartMessage(2);
+ msg.Write(Test1);
+ msg.Write(Test2);
+ msg.EndMessage();
+ msg.EndMessage();
+ msg.EndMessage();
+ msg.EndMessage();
+
+ // Read message.
+ using var reader = readerPool.Get();
+
+ reader.Update(msg.Buffer);
+
+ // Read first message.
+ using var messageOne = reader.ReadMessage();
+
+ Assert.Equal(1, messageOne.Tag);
+ Assert.Equal(0, messageOne.Position);
+ Assert.Equal(3, messageOne.Offset);
+ Assert.Equal(msgLength - 3, messageOne.Length);
+
+ using var messageTwo = messageOne.ReadMessage();
+
+ Assert.Equal(2, messageTwo.Tag);
+ Assert.Equal(0, messageTwo.Position);
+ Assert.Equal(6, messageTwo.Offset);
+ Assert.Equal(msgLength - 6, messageTwo.Length);
+ Assert.Equal(Test1, messageTwo.ReadByte());
+ Assert.Equal(Test2, messageTwo.ReadByte());
+
+ using var messageThree = messageTwo.ReadMessage();
+
+ Assert.Equal(2, messageThree.Tag);
+ Assert.Equal(0, messageThree.Position);
+ Assert.Equal(11, messageThree.Offset);
+ Assert.Equal(msgLength - 11, messageThree.Length);
+ Assert.Equal(Test1, messageThree.ReadByte());
+ Assert.Equal(Test2, messageThree.ReadByte());
+ }
+
[Fact]
public void CopySubMessage()
{
var handleMessage = handleReader.ReadMessage();
Assert.Equal(1, handleMessage.Tag);
- var parentReader = handleMessage.Slice(handleMessage.Position);
+ using var parentReader = handleMessage.Copy();
Assert.Equal(1, parentReader.Tag);
Assert.Equal(Test2, reader.ReadByte());
}
+ [Fact]
+ public void CopyToMessage()
+ {
+ var expected = new byte[]
+ {
+ 0x2A, 0x00, 0x01, 0x27, 0x00, 0x02, 0x26, 0x54,
+ 0x68, 0x69, 0x73, 0x20, 0x69, 0x73, 0x20, 0x61,
+ 0x20, 0x6C, 0x6F, 0x6E, 0x67, 0x20, 0x70, 0x61,
+ 0x63, 0x6B, 0x65, 0x74, 0x20, 0x74, 0x6F, 0x20,
+ 0x74, 0x65, 0x73, 0x74, 0x20, 0x63, 0x6F, 0x70,
+ 0x79, 0x69, 0x6E, 0x67, 0x2E
+ };
+
+ var readerPool = CreateReaderPool();
+
+ // Create packet.
+ var msg = new MessageWriter(2048);
+ msg.StartMessage(1);
+ msg.StartMessage(2);
+ msg.Write("This is a long packet to test copying.");
+ msg.EndMessage();
+ msg.EndMessage();
+
+ // Create a reader.
+ var reader = readerPool.Get();
+
+ reader.Update(msg.Buffer);
+
+ // Read the initial message.
+ var message = reader.ReadMessage();
+
+ // Copy the message to a new writer.
+ var writer = new MessageWriter(2048);
+
+ message.CopyTo(writer);
+
+ // Compare.
+ Assert.Equal(expected, writer.ToByteArray(true));
+ }
+
[Fact]
public void ReadMessageLength()
{
reader.Update(stream.ToArray());
- var option = reader.Buffer.Span[0];
+ var option = reader.Buffer[0];
if (option == (byte) MessageType.Reliable)
{
- reader.Position += 3;
+ reader.Seek(reader.Position + 3);
}
else if (option == (byte) UdpSendOption.Acknowledgement ||
option == (byte) UdpSendOption.Ping ||
}
else
{
- reader.Position += 1;
+ reader.Seek(reader.Position + 1);
}
var isSent = ipSrc.StartsWith("192.");
var buffer = reader.ReadBytes(length);
using var message = _readerPool.Get();
- message.Update(tag, buffer);
+ message.Update(buffer, tag: tag);
if (tag == MessageFlags.HostGame)
{
projects / rhonda / impostor.git / commitdiff