From: AeonLucid Date: Sat, 31 Oct 2020 23:28:33 +0000 (+0100) Subject: Use a normal byte array for MessageReader X-Git-Tag: v1.2.2~47^2~7 X-Git-Url: https://git.deb.at/?a=commitdiff_plain;h=f33a184d10e9d1cde887a360c733ede122b02da7;p=rhonda%2Fimpostor.git Use a normal byte array for MessageReader --- diff --git a/src/Impostor.Api/Net/Messages/IMessageReader.cs b/src/Impostor.Api/Net/Messages/IMessageReader.cs index 3a99946..d3320f3 100644 --- a/src/Impostor.Api/Net/Messages/IMessageReader.cs +++ b/src/Impostor.Api/Net/Messages/IMessageReader.cs @@ -12,7 +12,12 @@ namespace Impostor.Api.Net.Messages /// /// Gets the buffer of the message. /// - ReadOnlyMemory Buffer { get; } + byte[] Buffer { get; } + + /// + /// Gets the offset of our current in the entire . + /// + int Offset { get; } /// /// Gets the current position of the reader. @@ -54,8 +59,8 @@ namespace Impostor.Api.Net.Messages void CopyTo(IMessageWriter writer); - IMessageReader Slice(int start); + void Seek(int position); - IMessageReader Slice(int start, int length); + IMessageReader Copy(int offset = 0); } } diff --git a/src/Impostor.Hazel/MessageData.cs b/src/Impostor.Hazel/MessageData.cs deleted file mode 100644 index 60ab1c1..0000000 --- a/src/Impostor.Hazel/MessageData.cs +++ /dev/null @@ -1,24 +0,0 @@ -using System; -using System.Buffers; - -namespace Impostor.Hazel -{ - public readonly struct MessageData - { - private readonly IMemoryOwner _data; - private readonly int _length; - - public MessageData(IMemoryOwner data, int length) - { - _data = data; - _length = length; - } - - public ReadOnlyMemory Buffer => _data.Memory.Slice(0, _length); - - public void Return() - { - _data.Dispose(); - } - } -} \ No newline at end of file diff --git a/src/Impostor.Hazel/MessageReader.cs b/src/Impostor.Hazel/MessageReader.cs index ab6aea8..82c998e 100644 --- a/src/Impostor.Hazel/MessageReader.cs +++ b/src/Impostor.Hazel/MessageReader.cs @@ -12,28 +12,34 @@ namespace Impostor.Hazel private readonly ObjectPool _pool; private bool _inUse; - public byte Tag { get; private set; } - public ReadOnlyMemory Buffer { get; private set; } - public int Position { get; set; } - public int Length => Buffer.Length; + private byte _tag; internal MessageReader(ObjectPool pool) { _pool = pool; } - public void Update(ReadOnlyMemory 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 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() @@ -42,19 +48,21 @@ namespace Impostor.Hazel 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; } @@ -76,35 +84,35 @@ namespace Impostor.Hazel 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; } @@ -112,7 +120,7 @@ namespace Impostor.Hazel 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; } @@ -125,7 +133,7 @@ namespace Impostor.Hazel public ReadOnlyMemory ReadBytes(int length) { - var output = Buffer.Slice(Position, length); + var output = Buffer.AsMemory(ReadPosition, length); Position += length; return output; } @@ -165,27 +173,25 @@ namespace Impostor.Hazel { 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() diff --git a/src/Impostor.Hazel/Udp/UdpClientConnection.cs b/src/Impostor.Hazel/Udp/UdpClientConnection.cs index d2f5e59..4d80ddd 100644 --- a/src/Impostor.Hazel/Udp/UdpClientConnection.cs +++ b/src/Impostor.Hazel/Udp/UdpClientConnection.cs @@ -26,8 +26,8 @@ namespace Impostor.Hazel.Udp private readonly Timer _reliablePacketTimer; private readonly SemaphoreSlim _connectWaitLock; - private readonly MemoryPool _pool; - private readonly Channel _channel; + private readonly ArrayPool _pool; + private readonly Channel _channel; private Task _listenTask; private Task _handleTask; @@ -48,8 +48,8 @@ namespace Impostor.Hazel.Udp _reliablePacketTimer = new Timer(ManageReliablePacketsInternal, null, 100, Timeout.Infinite); _connectWaitLock = new SemaphoreSlim(1, 1); - _pool = MemoryPool.Shared; - _channel = Channel.CreateUnbounded(new UnboundedChannelOptions + _pool = ArrayPool.Shared; + _channel = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true, SingleWriter = true @@ -171,27 +171,8 @@ namespace Impostor.Hazel.Udp return; } - await HandleAsync(data.Buffer); - } - } - - private async ValueTask HandleAsync(ReadOnlyMemory 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); } } diff --git a/src/Impostor.Hazel/Udp/UdpConnection.Reliable.cs b/src/Impostor.Hazel/Udp/UdpConnection.Reliable.cs index 25c3633..a7a4309 100644 --- a/src/Impostor.Hazel/Udp/UdpConnection.Reliable.cs +++ b/src/Impostor.Hazel/Udp/UdpConnection.Reliable.cs @@ -285,7 +285,11 @@ namespace Impostor.Hazel.Udp { 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); diff --git a/src/Impostor.Hazel/Udp/UdpConnection.cs b/src/Impostor.Hazel/Udp/UdpConnection.cs index 46ad376..5288d3c 100644 --- a/src/Impostor.Hazel/Udp/UdpConnection.cs +++ b/src/Impostor.Hazel/Udp/UdpConnection.cs @@ -31,14 +31,14 @@ namespace Impostor.Hazel.Udp _readerPool = readerPool; _stoppingCts = new CancellationTokenSource(); - Pipeline = Channel.CreateUnbounded(new UnboundedChannelOptions + Pipeline = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true, SingleWriter = true }); } - internal Channel Pipeline { get; } + internal Channel Pipeline { get; } public Task StartAsync() { @@ -85,18 +85,17 @@ namespace Impostor.Hazel.Udp 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) { @@ -104,16 +103,6 @@ namespace Impostor.Hazel.Udp Dispose(true); break; } - finally - { - result.Return(); - } - } - - // Exhaust pipeline. - while (Pipeline.Reader.TryRead(out var mem)) - { - mem.Return(); } } @@ -203,11 +192,14 @@ namespace Impostor.Hazel.Udp // 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: @@ -216,7 +208,7 @@ namespace Impostor.Hazel.Udp //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! @@ -230,12 +222,18 @@ namespace Impostor.Hazel.Udp 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; } diff --git a/src/Impostor.Hazel/Udp/UdpConnectionListener.cs b/src/Impostor.Hazel/Udp/UdpConnectionListener.cs index b0f7cef..ee4c8fc 100644 --- a/src/Impostor.Hazel/Udp/UdpConnectionListener.cs +++ b/src/Impostor.Hazel/Udp/UdpConnectionListener.cs @@ -180,7 +180,8 @@ namespace Impostor.Hazel.Udp await client.StartAsync(); } - await WriteToClientAsync(client, data.Buffer); + // Write to client. + await client.Pipeline.Writer.WriteAsync(data.Buffer); } } catch (Exception e) @@ -189,26 +190,6 @@ namespace Impostor.Hazel.Udp } } - private async ValueTask WriteToClientAsync(UdpConnection client, ReadOnlyMemory 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; diff --git a/src/Impostor.Server/Net/Client.cs b/src/Impostor.Server/Net/Client.cs index bfe29a6..0036cdd 100644 --- a/src/Impostor.Server/Net/Client.cs +++ b/src/Impostor.Server/Net/Client.cs @@ -148,7 +148,7 @@ namespace Impostor.Server.Net 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 diff --git a/src/Impostor.Server/Recorder/ClientRecorder.cs b/src/Impostor.Server/Recorder/ClientRecorder.cs index 03ad965..5763c70 100644 --- a/src/Impostor.Server/Recorder/ClientRecorder.cs +++ b/src/Impostor.Server/Recorder/ClientRecorder.cs @@ -27,7 +27,7 @@ namespace Impostor.Server.Recorder public override async ValueTask HandleMessageAsync(IMessageReader reader, MessageType messageType) { - var messageCopy = reader.Slice(0); + using var messageCopy = reader.Copy(); // Trigger connect event. if (_isFirst) diff --git a/src/Impostor.Server/Recorder/PacketRecorder.cs b/src/Impostor.Server/Recorder/PacketRecorder.cs index f3ff48b..c145f72 100644 --- a/src/Impostor.Server/Recorder/PacketRecorder.cs +++ b/src/Impostor.Server/Recorder/PacketRecorder.cs @@ -146,7 +146,7 @@ namespace Impostor.Server.Recorder 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) diff --git a/src/Impostor.Tests/Hazel/MessageReaderTests.cs b/src/Impostor.Tests/Hazel/MessageReaderTests.cs index 9ac0132..da13696 100644 --- a/src/Impostor.Tests/Hazel/MessageReaderTests.cs +++ b/src/Impostor.Tests/Hazel/MessageReaderTests.cs @@ -113,6 +113,65 @@ namespace Impostor.Tests.Hazel 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() { @@ -135,7 +194,7 @@ namespace Impostor.Tests.Hazel 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); @@ -146,6 +205,46 @@ namespace Impostor.Tests.Hazel 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() { diff --git a/src/Impostor.Tools.Proxy/Program.cs b/src/Impostor.Tools.Proxy/Program.cs index d891bdf..b97dd94 100644 --- a/src/Impostor.Tools.Proxy/Program.cs +++ b/src/Impostor.Tools.Proxy/Program.cs @@ -85,10 +85,10 @@ namespace Impostor.Tools.Proxy 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 || @@ -99,7 +99,7 @@ namespace Impostor.Tools.Proxy } else { - reader.Position += 1; + reader.Seek(reader.Position + 1); } var isSent = ipSrc.StartsWith("192."); diff --git a/src/Impostor.Tools.ServerReplay/Program.cs b/src/Impostor.Tools.ServerReplay/Program.cs index f656801..5aaa954 100644 --- a/src/Impostor.Tools.ServerReplay/Program.cs +++ b/src/Impostor.Tools.ServerReplay/Program.cs @@ -165,7 +165,7 @@ namespace Impostor.Tools.ServerReplay var buffer = reader.ReadBytes(length); using var message = _readerPool.Get(); - message.Update(tag, buffer); + message.Update(buffer, tag: tag); if (tag == MessageFlags.HostGame) {