diff --git a/Projects/Server.Tests/Network/Packets/Old/Outgoing/AccountPacketTests.cs b/Projects/Server.Tests/Network/Packets/Old/Outgoing/AccountPacketTests.cs index c82fd5c12..eb18742ad 100644 --- a/Projects/Server.Tests/Network/Packets/Old/Outgoing/AccountPacketTests.cs +++ b/Projects/Server.Tests/Network/Packets/Old/Outgoing/AccountPacketTests.cs @@ -19,13 +19,13 @@ namespace Server.Tests.Network secondMobile.RawName = null; var account = new MockAccount(new[] { firstMobile, null, secondMobile }); - var oldPacket = new ChangeCharacter(account).Compile(); + var expected = new ChangeCharacter(account).Compile(); using var ns = PacketTestUtilities.CreateTestNetState(); Packets.SendChangeCharacter(ns, account); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), oldPacket); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } [Fact] @@ -37,8 +37,8 @@ namespace Server.Tests.Network Packets.SendClientVersionRequest(ns); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), expected); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } [Fact] @@ -49,8 +49,8 @@ namespace Server.Tests.Network using var ns = PacketTestUtilities.CreateTestNetState(); Packets.SendCharacterDeleteResult(ns, DeleteResultType.BadRequest); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), expected); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } [Fact] @@ -61,8 +61,8 @@ namespace Server.Tests.Network using var ns = PacketTestUtilities.CreateTestNetState(); Packets.SendPopupMessage(ns, PMMessage.LoginSyncError); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), expected); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } [Theory] @@ -83,8 +83,8 @@ namespace Server.Tests.Network var expected = new SupportedFeatures(ns).Compile(); Packets.SendSupportedFeature(ns); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), expected); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } [Fact] @@ -104,8 +104,8 @@ namespace Server.Tests.Network using var ns = PacketTestUtilities.CreateTestNetState(); Packets.SendLoginConfirmation(ns, m); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), expected); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } [Fact] @@ -116,8 +116,8 @@ namespace Server.Tests.Network using var ns = PacketTestUtilities.CreateTestNetState(); Packets.SendLoginComplete(ns); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), expected); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } [Fact] @@ -134,8 +134,8 @@ namespace Server.Tests.Network using var ns = PacketTestUtilities.CreateTestNetState(); Packets.SendCharacterListUpdate(ns, acct); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), expected); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } [Fact] @@ -160,8 +160,8 @@ namespace Server.Tests.Network Packets.SendCharacterList(ns); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), expected); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } [Fact] @@ -185,8 +185,8 @@ namespace Server.Tests.Network Packets.SendCharacterList(ns); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), expected); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } [Fact] @@ -198,8 +198,8 @@ namespace Server.Tests.Network using var ns = PacketTestUtilities.CreateTestNetState(); Packets.SendAccountLoginRejected(ns, reason); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), expected); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } [Fact] @@ -217,8 +217,8 @@ namespace Server.Tests.Network Packets.SendAccountLoginAck(ns); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), expected); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } [Fact] @@ -233,8 +233,8 @@ namespace Server.Tests.Network Packets.SendPlayServerAck(ns, si, authId); - ns.SendPipe.Reader.TryRead(out var buffer); - AssertThat.Equal(buffer.GetSpan(0), expected); + var result = ns.SendPipe.Reader.TryRead(); + AssertThat.Equal(result.Buffer[0].AsSpan(0), expected); } } } diff --git a/Projects/Server.Tests/Network/PipeTests.cs b/Projects/Server.Tests/Network/PipeTests.cs index 9ba78bc69..b89c1b1d6 100644 --- a/Projects/Server.Tests/Network/PipeTests.cs +++ b/Projects/Server.Tests/Network/PipeTests.cs @@ -26,20 +26,18 @@ namespace Server.Tests.Network DelayedExecute(() => { ; // Write some data into the pipe - writer.GetAvailable(out var buffer); - Assert.True(buffer.Length == 99); - buffer[0] = 0x1; - buffer[1] = 0x2; - buffer[2] = 0x3; + var result = writer.TryGetMemory(); + Assert.True(result.Buffer[0].Count == 99); + result.Buffer[0][0] = 0x1; + result.Buffer[0][1] = 0x2; + result.Buffer[0][2] = 0x3; writer.Advance(3); writer.Flush(); }); - var segments = new ArraySegment[2]; - await reader.Read(segments); - - Assert.Equal(3, segments[0].Count); + var result = await reader.Read(); + Assert.Equal(3, result.Buffer[0].Count); } private bool _signal; @@ -53,11 +51,11 @@ namespace Server.Tests.Network while (count < 0x8000000) { - reader.TryRead(out var buffer); + var result = reader.TryRead(); - var first = buffer.GetSpan(0); + var first = result.Buffer[0]; - for (int i = 0; i < first.Length; i++) + for (int i = 0; i < first.Count; i++) { Assert.True(first[i] == expected_value); count++; @@ -68,8 +66,8 @@ namespace Server.Tests.Network } } - var second = buffer.GetSpan(1); - for (int i = 0; i < second.Length; i++) + var second = result.Buffer[1]; + for (int i = 0; i < second.Count; i++) { Assert.True(second[i] == expected_value); count++; @@ -80,7 +78,7 @@ namespace Server.Tests.Network } } - reader.Advance((uint)buffer.Length); + reader.Advance((uint)result.Length); } _signal = true; @@ -100,14 +98,14 @@ namespace Server.Tests.Network while (count < 0x8000000) { - writer.GetAvailable(out var buffer); + var result = writer.TryGetMemory(); - if (buffer.Length < 16) + if (result.Length < 16) { continue; } - buffer.CopyFrom(new[] { + result.CopyFrom(new[] { expected_value, expected_value, expected_value, expected_value, expected_value, expected_value, expected_value, expected_value, expected_value, expected_value, expected_value, expected_value, @@ -136,33 +134,33 @@ namespace Server.Tests.Network var reader = pipe.Reader; var writer = pipe.Writer; - writer.GetAvailable(out var buffer); + var result = writer.TryGetMemory(); - Assert.Equal(9, buffer.Length); + Assert.Equal(9, result.Length); Assert.Equal(0u, reader.GetAvailable()); - reader.TryRead(out buffer); - Assert.Equal(0, buffer.Length); + result = reader.TryRead(); + Assert.Equal(0, result.Length); writer.Advance(7); - writer.GetAvailable(out buffer); - Assert.Equal(2, buffer.Length); + result = writer.TryGetMemory(); + Assert.Equal(2, result.Length); Assert.Equal(7u, reader.GetAvailable()); - reader.TryRead(out buffer); - Assert.Equal(7, buffer.Length); + result = reader.TryRead(); + Assert.Equal(7, result.Length); reader.Advance(4); - writer.GetAvailable(out buffer); - Assert.Equal(6, buffer.Length); + result = writer.TryGetMemory(); + Assert.Equal(6, result.Length); Assert.Equal(3u, reader.GetAvailable()); - reader.TryRead(out buffer); - Assert.Equal(3, buffer.Length); + result = reader.TryRead(); + Assert.Equal(3, result.Length); writer.Advance(3); - writer.GetAvailable(out buffer); - Assert.Equal(3, buffer.Length); + result = writer.TryGetMemory(); + Assert.Equal(3, result.Length); Assert.Equal(6u, reader.GetAvailable()); - reader.TryRead(out buffer); - Assert.Equal(6, buffer.Length); + result = reader.TryRead(); + Assert.Equal(6, result.Length); } [Fact] @@ -191,19 +189,19 @@ namespace Server.Tests.Network var reader = pipe.Reader; var writer = pipe.Writer; - writer.GetAvailable(out var buffer); - Assert.True(buffer.Length == 9); + var result = writer.TryGetMemory(); + Assert.Equal(9, result.Length); - buffer.CopyFrom(new byte[] { 0, 1, 2, 3, 4, 5, 6, 7, 8 }); + result.CopyFrom(new byte[] { 0, 1, 2, 3, 4, 5, 6, 7, 8 }); writer.Advance(9); writer.Flush(); Assert.Equal(9u, reader.GetAvailable()); - reader.TryRead(out buffer); + result = reader.TryRead(); - var first = buffer.GetSpan(0); + var first = result.Buffer[0]; for (int i = 0; i < 9; i++) { @@ -211,9 +209,9 @@ namespace Server.Tests.Network } reader.Advance(4); - reader.TryRead(out buffer); - first = buffer.GetSpan(0); - Assert.Equal(5, buffer.Length); + result = reader.TryRead(); + Assert.Equal(5, result.Length); + first = result.Buffer[0]; Assert.Equal(4, first[0]); Assert.Equal(5, first[1]); Assert.Equal(6, first[2]); diff --git a/Projects/Server/Buffers/CircularBufferReader.cs b/Projects/Server/Buffers/CircularBufferReader.cs index 3b7198cd2..3238e8b67 100644 --- a/Projects/Server/Buffers/CircularBufferReader.cs +++ b/Projects/Server/Buffers/CircularBufferReader.cs @@ -35,7 +35,7 @@ namespace Server.Network { } - public CircularBufferReader(CircularBuffer buffer) : this(buffer.GetSpan(0), buffer.GetSpan(1)) + public CircularBufferReader(ArraySegment[] buffer) : this(buffer[0], buffer[1]) { } diff --git a/Projects/Server/Buffers/CircularBufferWriter.cs b/Projects/Server/Buffers/CircularBufferWriter.cs index f67fb88c6..7519f10cc 100644 --- a/Projects/Server/Buffers/CircularBufferWriter.cs +++ b/Projects/Server/Buffers/CircularBufferWriter.cs @@ -33,6 +33,10 @@ namespace System.Buffers { } + public CircularBufferWriter(ArraySegment[] buffer) : this(buffer[0], buffer[1]) + { + } + public CircularBufferWriter(Span first, Span second) { _first = first; diff --git a/Projects/Server/Network/NetState/NetState.cs b/Projects/Server/Network/NetState/NetState.cs index 587ca0553..fccc7b95c 100644 --- a/Projects/Server/Network/NetState/NetState.cs +++ b/Projects/Server/Network/NetState/NetState.cs @@ -36,8 +36,8 @@ namespace Server.Network public partial class NetState : IComparable, IDisposable { - private static int RecvPipeSize = 1024 * 64; - private static int SendPipeSize = 1024 * 256; + private static int RecvPipeSize = 1024 * 64 + 1; + private static int SendPipeSize = 1024 * 256 + 1; private static int GumpCap = 512; private static int HuePickerCap = 512; private static int MenuCap = 512; @@ -369,9 +369,17 @@ namespace Server.Network NetworkState.Resume(ref m_NetworkState); } + public bool GetSendBuffer(out CircularBuffer cBuffer) + { + var result = SendPipe.Writer.TryGetMemory(); + cBuffer = new CircularBuffer(result.Buffer); + + return !(result.IsClosed || result.Length <= 0); + } + public virtual void Send(ref CircularBuffer buffer, int length) { - if (Connection == null || BlockAllPackets || buffer.Length == 0 || length <= 0) + if (Connection == null || BlockAllPackets || length <= 0) { return; } @@ -407,15 +415,16 @@ namespace Server.Network if (buffer.Length > 0 && length > 0) { - if (!SendPipe.Writer.GetAvailable(out var pipeBuffer)) + var result = writer.TryGetMemory(); + if (result.IsClosed) { p.OnSend(); return; } - if (pipeBuffer.Length >= length) + if (result.Length >= length) { - pipeBuffer.CopyFrom(buffer.AsSpan(0, length)); + result.CopyFrom(buffer.AsSpan(0, length)); writer.Advance((uint)length); // Flush at the end of the game loop @@ -461,24 +470,23 @@ namespace Server.Network private async void SendTask(object state) { var reader = SendPipe.Reader; - var segments = new ArraySegment[2]; try { while (m_Running) { - var result = await reader.Read(segments); - if (result.Closed) + var result = await reader.Read(); + if (result.IsClosed) { break; } - if (segments[0].Count + segments[1].Count <= 0) + if (result.Length <= 0) { continue; } - var bytesWritten = await Connection.SendAsync(segments, SocketFlags.None); + var bytesWritten = await Connection.SendAsync(result.Buffer, SocketFlags.None); if (bytesWritten > 0) { @@ -510,7 +518,6 @@ namespace Server.Network { var socket = Connection; var writer = RecvPipe.Writer; - var segments = new ArraySegment[2]; try { @@ -521,24 +528,25 @@ namespace Server.Network continue; } - // TODO: Make awaitable - if (!writer.GetAvailable(segments)) + var result = await writer.GetMemory(); + + if (result.IsClosed) { break; } - if (segments[0].Count + segments[1].Count <= 0) + if (result.Length <= 0) { continue; } - var bytesWritten = await socket.ReceiveAsync(segments, SocketFlags.None); + var bytesWritten = await socket.ReceiveAsync(result.Buffer, SocketFlags.None); if (bytesWritten <= 0) { break; } - DecodePacket(segments, ref bytesWritten); + DecodePacket(result.Buffer, ref bytesWritten); writer.Advance((uint)bytesWritten); m_NextCheckActivity = Core.TickCount + 90000; @@ -583,12 +591,14 @@ namespace Server.Network // Process as many packets as we can synchronously while (true) { - if (!reader.TryRead(out var buffer) || buffer.Length <= 0) + var result = reader.TryRead(); + + if (result.IsClosed || result.Length <= 0) { return; } - var bytesProcessed = PacketHandlers.ProcessPacket(this, ref buffer); + var bytesProcessed = PacketHandlers.ProcessPacket(this, result.Buffer); if (bytesProcessed <= 0) { diff --git a/Projects/Server/Network/PacketHandlers.cs b/Projects/Server/Network/PacketHandlers.cs index bca8cb673..2ed9b13d7 100644 --- a/Projects/Server/Network/PacketHandlers.cs +++ b/Projects/Server/Network/PacketHandlers.cs @@ -277,9 +277,9 @@ namespace Server.Network } } - public static int ProcessPacket(NetState ns, ref CircularBuffer buffer) + public static int ProcessPacket(NetState ns, ArraySegment[] buffer) { - var reader = new CircularBufferReader(ref buffer); + var reader = new CircularBufferReader(buffer); var packetId = reader.ReadByte(); diff --git a/Projects/Server/Network/Packets/AccountPackets.cs b/Projects/Server/Network/Packets/AccountPackets.cs index e79e3fa84..2766a8f27 100644 --- a/Projects/Server/Network/Packets/AccountPackets.cs +++ b/Projects/Server/Network/Packets/AccountPackets.cs @@ -61,7 +61,7 @@ namespace Server.Network */ public static void SendChangeCharacter(NetState ns, IAccount a) { - if (ns == null || a == null || !ns.SendPipe.Writer.GetAvailable(out var buffer)) + if (ns == null || a == null || !ns.GetSendBuffer(out var buffer)) { return; } @@ -109,7 +109,7 @@ namespace Server.Network */ public static void SendClientVersionRequest(NetState ns) { - if (ns != null && ns.SendPipe.Writer.GetAvailable(out var buffer)) + if (ns != null && ns.GetSendBuffer(out var buffer)) { buffer[0] = 0xBD; // Packet ID buffer[1] = 0x00; @@ -127,7 +127,7 @@ namespace Server.Network */ public static void SendCharacterDeleteResult(NetState ns, DeleteResultType res) { - if (ns != null && ns.SendPipe.Writer.GetAvailable(out var buffer)) + if (ns != null && ns.GetSendBuffer(out var buffer)) { buffer[0] = 0x85; // Packet ID buffer[1] = (byte)res; @@ -144,7 +144,7 @@ namespace Server.Network */ public static void SendPopupMessage(NetState ns, PMMessage msg) { - if (ns != null && ns.SendPipe.Writer.GetAvailable(out var buffer)) + if (ns != null && ns.GetSendBuffer(out var buffer)) { buffer[0] = 0x53; // Packet ID buffer[1] = (byte)msg; @@ -161,7 +161,7 @@ namespace Server.Network */ public static void SendSupportedFeature(NetState ns) { - if (ns == null || !ns.SendPipe.Writer.GetAvailable(out var buffer)) + if (ns == null || !ns.GetSendBuffer(out var buffer)) { return; } @@ -206,7 +206,7 @@ namespace Server.Network */ public static void SendLoginConfirmation(NetState ns, Mobile m) { - if (ns == null || !ns.SendPipe.Writer.GetAvailable(out var buffer)) + if (ns == null || !ns.GetSendBuffer(out var buffer)) { return; } @@ -247,7 +247,7 @@ namespace Server.Network */ public static void SendLoginComplete(NetState ns) { - if (ns != null && ns.SendPipe.Writer.GetAvailable(out var buffer)) + if (ns != null && ns.GetSendBuffer(out var buffer)) { buffer[0] = 0x55; // Packet ID @@ -263,7 +263,7 @@ namespace Server.Network */ public static void SendCharacterListUpdate(NetState ns, IAccount a) { - if (ns == null || a == null || !ns.SendPipe.Writer.GetAvailable(out var buffer)) + if (ns == null || a == null || !ns.GetSendBuffer(out var buffer)) { return; } @@ -316,7 +316,7 @@ namespace Server.Network { var acct = ns?.Account; - if (acct == null || !ns.SendPipe.Writer.GetAvailable(out var buffer)) + if (acct == null || !ns.GetSendBuffer(out var buffer)) { return; } @@ -415,7 +415,7 @@ namespace Server.Network */ public static void SendAccountLoginRejected(NetState ns, ALRReason reason) { - if (ns != null && ns.SendPipe.Writer.GetAvailable(out var buffer)) + if (ns != null && ns.GetSendBuffer(out var buffer)) { buffer[0] = 0x82; // Packet ID buffer[1] = (byte)reason; @@ -432,7 +432,7 @@ namespace Server.Network */ public static void SendAccountLoginAck(NetState ns) { - if (ns == null || !ns.SendPipe.Writer.GetAvailable(out var buffer)) + if (ns == null || !ns.GetSendBuffer(out var buffer)) { return; } @@ -469,7 +469,7 @@ namespace Server.Network */ public static void SendPlayServerAck(NetState ns, ServerInfo si, int authId) { - if (ns == null || !ns.SendPipe.Writer.GetAvailable(out var buffer)) + if (ns == null || !ns.GetSendBuffer(out var buffer)) { return; } diff --git a/Projects/Server/Network/Pipe.cs b/Projects/Server/Network/Pipe.cs index b7c94d999..067cc6836 100644 --- a/Projects/Server/Network/Pipe.cs +++ b/Projects/Server/Network/Pipe.cs @@ -14,7 +14,6 @@ *************************************************************************/ using System; -using System.Buffers; using System.Runtime.CompilerServices; using System.Threading; @@ -27,78 +26,125 @@ namespace Server.Network public bool IsCompleted { get; } public T GetResult(); - - public void OnCompleted(Action continuation); } public class Pipe { - public readonly struct Result + public struct Result { - public bool Closed { get; } + public ArraySegment[] Buffer { get; } + public bool IsClosed { get; set; } - public Result(bool closed) => Closed = closed; + public int Length + { + get + { + var length = 0; + for (int i = 0; i < Buffer.Length; i++) + { + length += Buffer[i].Count; + } + + return length; + } + } + + public void CopyFrom(ReadOnlySpan bytes) + { + var remaining = bytes.Length; + var offset = 0; + + if (remaining == 0) + { + return; + } + + for (int i = 0; i < Buffer.Length; i++) + { + var buffer = Buffer[i]; + var sz = Math.Min(remaining, buffer.Count); + bytes.Slice(offset, sz).CopyTo(buffer); + + remaining -= sz; + offset += sz; + + if (remaining == 0) + { + return; + } + } + + throw new OutOfMemoryException(); + } + + public Result(int segments) + { + IsClosed = false; + Buffer = new ArraySegment[segments]; + } } - public class PipeWriter + public class PipeWriter : IPipeTask { private readonly Pipe _pipe; - public PipeWriter(Pipe pipe) => _pipe = pipe; + private Result _result = new Result(2); + + internal PipeWriter(Pipe pipe) => _pipe = pipe; [MethodImpl(MethodImplOptions.AggressiveInlining)] - public bool GetAvailable(ArraySegment[] segments) + public uint GetAvailable() { var read = _pipe._readIdx; var write = _pipe._writeIdx; if (read <= write) { - var readZero = read == 0; - var sz = _pipe.Size - write - (readZero ? 1 : 0); + if (read == 0) + { + return _pipe.Size - write - 1; + } - segments[0] = sz == 0 ? ArraySegment.Empty : new ArraySegment(_pipe._buffer, (int)write, (int)sz); - segments[1] = readZero ? ArraySegment.Empty : new ArraySegment(_pipe._buffer, 0, (int)read - 1); - } - else - { - var sz = read - write - 1; - - segments[0] = sz == 0 ? ArraySegment.Empty : new ArraySegment(_pipe._buffer, (int)write, (int)sz); - segments[1] = ArraySegment.Empty; + return _pipe.Size - write + (read - 1); } - return !_pipe._closed; + return read - write - 1; } - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public bool GetAvailable(out CircularBuffer buffer) + public Result TryGetMemory() { var read = _pipe._readIdx; var write = _pipe._writeIdx; - Span first; - Span second; + _result.IsClosed = _pipe._closed; if (read <= write) { var readZero = read == 0; var sz = _pipe.Size - write - (readZero ? 1 : 0); - first = sz == 0 ? Span.Empty : _pipe._buffer.AsSpan((int)write, (int)sz); - second = readZero ? Span.Empty : _pipe._buffer.AsSpan(0, (int)read - 1); + _result.Buffer[0] = sz == 0 ? ArraySegment.Empty : new ArraySegment(_pipe._buffer, (int)write, (int)sz); + _result.Buffer[1] = readZero ? ArraySegment.Empty : new ArraySegment(_pipe._buffer, 0, (int)read - 1); } else { var sz = read - write - 1; - first = sz == 0 ? Span.Empty : _pipe._buffer.AsSpan((int)write, (int)sz); - second = Span.Empty; + _result.Buffer[0] = sz == 0 ? ArraySegment.Empty : new ArraySegment(_pipe._buffer, (int)write, (int)sz); + _result.Buffer[1] = ArraySegment.Empty; } - buffer = new CircularBuffer(first, second); + return _result; + } - return !_pipe._closed; + public IPipeTask GetMemory() + { + if (_pipe._writeAwaitBeginning) + { + throw new Exception("Double await on reader"); + } + + return this; } public void Advance(uint count) @@ -171,7 +217,12 @@ namespace Server.Network public void Flush() { - var waiting = _pipe._awaitBeginning; + if (_pipe._readIdx == _pipe._writeIdx) + { + return; + } + + var waiting = _pipe._readAwaitBeginning; if (!waiting) { @@ -182,20 +233,48 @@ namespace Server.Network do { - continuation = _pipe._readerContinuation; + continuation = _pipe._readContinuation; } while (continuation == null); - _pipe._readerContinuation = null; - _pipe._awaitBeginning = false; + _pipe._readContinuation = null; + _pipe._readAwaitBeginning = false; ThreadPool.UnsafeQueueUserWorkItem(state => continuation(), true); } + + #region Awaitable + + // The following makes it possible to await the writer. Do not use any of this directly. + + public IPipeTask GetAwaiter() => this; + + public bool IsCompleted + { + get + { + if (GetAvailable() > 0) + { + return true; + } + + _pipe._writeAwaitBeginning = true; + return false; + } + } + + public Result GetResult() => TryGetMemory(); + + public void OnCompleted(Action continuation) => _pipe._writeContinuation = continuation; + + #endregion } - public class PipeReader : IPipeTask + public class PipeReader : IPipeTask { private readonly Pipe _pipe; + private Result _result = new Result(2); + internal PipeReader(Pipe pipe) => _pipe = pipe; [MethodImpl(MethodImplOptions.AggressiveInlining)] @@ -212,61 +291,37 @@ namespace Server.Network return write + _pipe.Size - read; } - private ArraySegment[] _segments; - - public IPipeTask Read(ArraySegment[] segments) + public Result TryRead() { - if (_pipe._awaitBeginning) + var read = _pipe._readIdx; + var write = _pipe._writeIdx; + + _result.IsClosed = _pipe._closed; + + if (read <= write) + { + _result.Buffer[0] = write - read == 0 ? ArraySegment.Empty : new ArraySegment(_pipe._buffer, (int)read, (int)(write - read)); + _result.Buffer[1] = ArraySegment.Empty; + } + else + { + _result.Buffer[0] = _pipe.Size - read == 0 ? ArraySegment.Empty : new ArraySegment(_pipe._buffer, (int)read, (int)(_pipe.Size - read)); + _result.Buffer[1] = write == 0 ? ArraySegment.Empty : new ArraySegment(_pipe._buffer, 0, (int)write); + } + + return _result; + } + + public IPipeTask Read() + { + if (_pipe._readAwaitBeginning) { throw new Exception("Double await on reader"); } - _segments = segments; return this; } - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public bool TryRead(out CircularBuffer buffer) - { - var read = _pipe._readIdx; - var write = _pipe._writeIdx; - - Span first; - Span second; - - if (read <= write) - { - first = write - read == 0 ? Span.Empty : _pipe._buffer.AsSpan((int)read, (int)(write - read)); - second = Span.Empty; - } - else - { - first = _pipe.Size - read == 0 ? Span.Empty : _pipe._buffer.AsSpan((int)read, (int)(_pipe.Size - read)); - second = write == 0 ? Span.Empty : _pipe._buffer.AsSpan(0, (int)write); - } - - buffer = new CircularBuffer(first, second); - return !_pipe._closed; - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void TryRead(ArraySegment[] segments) - { - var read = _pipe._readIdx; - var write = _pipe._writeIdx; - - if (read <= write) - { - segments[0] = write - read == 0 ? ArraySegment.Empty : new ArraySegment(_pipe._buffer, (int)read, (int)(write - read)); - segments[1] = ArraySegment.Empty; - } - else - { - segments[0] = _pipe.Size - read == 0 ? ArraySegment.Empty : new ArraySegment(_pipe._buffer, (int)read, (int)(_pipe.Size - read)); - segments[1] = write == 0 ? ArraySegment.Empty : new ArraySegment(_pipe._buffer, 0, (int)write); - } - } - public void Advance(uint count) { var read = _pipe._readIdx; @@ -306,6 +361,33 @@ namespace Server.Network _pipe._readIdx = read; } + public void Commit() + { + if (_pipe._readIdx == (_pipe._writeIdx + 1) % _pipe.Size) + { + return; + } + + var waiting = _pipe._writeAwaitBeginning; + + if (!waiting) + { + return; + } + + Action continuation; + + do + { + continuation = _pipe._writeContinuation; + } while (continuation == null); + + _pipe._writeContinuation = null; + _pipe._readAwaitBeginning = false; + + ThreadPool.UnsafeQueueUserWorkItem(state => continuation(), true); + } + #region Awaitable // The following makes it possible to await the reader. Do not use any of this directly. @@ -321,29 +403,14 @@ namespace Server.Network return true; } - _pipe._awaitBeginning = true; + _pipe._readAwaitBeginning = true; return false; } } - public Result GetResult() - { - if (_pipe._closed) - { - _segments = null; - return new Result(true); - } + public Result GetResult() => TryRead(); - if (_segments != null) - { - TryRead(_segments); - _segments = null; - } - - return new Result(false); - } - - public void OnCompleted(Action continuation) => _pipe._readerContinuation = continuation; + public void OnCompleted(Action continuation) => _pipe._readContinuation = continuation; #endregion } @@ -353,8 +420,8 @@ namespace Server.Network private volatile uint _readIdx; private bool _closed; - public PipeWriter Writer { get; } - public PipeReader Reader { get; } + public PipeWriter Writer { get; } + public PipeReader Reader { get; } public uint Size => (uint)_buffer.Length; @@ -365,13 +432,16 @@ namespace Server.Network _readIdx = 0; _closed = false; - Writer = new PipeWriter(this); - Reader = new PipeReader(this); + Writer = new PipeWriter(this); + Reader = new PipeReader(this); } #region Awaitable - private volatile bool _awaitBeginning; - private volatile Action _readerContinuation; + private volatile bool _readAwaitBeginning; + private volatile Action _readContinuation; + + private volatile bool _writeAwaitBeginning; + private volatile Action _writeContinuation; #endregion }