warrentyexpired/ModernUO
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Updates networking to make pipe writer awaitable (#290)

Browse source 
- [X] Updates result so it is preallocated
- [X] Updates PipeWriter so it is awaitable
- [X] Simplifies NetState code to match changes
- [X] Adds a byte for empty checking so incoming/outgoing can have full pipes.

Bumps release version
This commit is contained in:
Kamron Batman 2020-10-27 11:22:07 -07:00 • committed by GitHub
parent 99bfff40df
commit 5f7e4bf050
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 292 additions and 210 deletions
Download patch file Download diff file Expand all files Collapse all files

54
Projects/Server.Tests/Network/Packets/Old/Outgoing/AccountPacketTests.cs
View file

@ -19,13 +19,13 @@ namespace Server.Tests.Network
secondMobile.RawName = null; secondMobile.RawName = null;
var account = new MockAccount(new[] { firstMobile, null, secondMobile }); 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(); using var ns = PacketTestUtilities.CreateTestNetState();
Packets.SendChangeCharacter(ns, account); Packets.SendChangeCharacter(ns, account);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), oldPacket); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
[Fact] [Fact]
@ -37,8 +37,8 @@ namespace Server.Tests.Network
Packets.SendClientVersionRequest(ns); Packets.SendClientVersionRequest(ns);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), expected); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
[Fact] [Fact]
@ -49,8 +49,8 @@ namespace Server.Tests.Network
using var ns = PacketTestUtilities.CreateTestNetState(); using var ns = PacketTestUtilities.CreateTestNetState();
Packets.SendCharacterDeleteResult(ns, DeleteResultType.BadRequest); Packets.SendCharacterDeleteResult(ns, DeleteResultType.BadRequest);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), expected); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
[Fact] [Fact]
@ -61,8 +61,8 @@ namespace Server.Tests.Network
using var ns = PacketTestUtilities.CreateTestNetState(); using var ns = PacketTestUtilities.CreateTestNetState();
Packets.SendPopupMessage(ns, PMMessage.LoginSyncError); Packets.SendPopupMessage(ns, PMMessage.LoginSyncError);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), expected); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
[Theory] [Theory]
@ -83,8 +83,8 @@ namespace Server.Tests.Network
var expected = new SupportedFeatures(ns).Compile(); var expected = new SupportedFeatures(ns).Compile();
Packets.SendSupportedFeature(ns); Packets.SendSupportedFeature(ns);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), expected); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
[Fact] [Fact]
@ -104,8 +104,8 @@ namespace Server.Tests.Network
using var ns = PacketTestUtilities.CreateTestNetState(); using var ns = PacketTestUtilities.CreateTestNetState();
Packets.SendLoginConfirmation(ns, m); Packets.SendLoginConfirmation(ns, m);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), expected); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
[Fact] [Fact]
@ -116,8 +116,8 @@ namespace Server.Tests.Network
using var ns = PacketTestUtilities.CreateTestNetState(); using var ns = PacketTestUtilities.CreateTestNetState();
Packets.SendLoginComplete(ns); Packets.SendLoginComplete(ns);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), expected); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
[Fact] [Fact]
@ -134,8 +134,8 @@ namespace Server.Tests.Network
using var ns = PacketTestUtilities.CreateTestNetState(); using var ns = PacketTestUtilities.CreateTestNetState();
Packets.SendCharacterListUpdate(ns, acct); Packets.SendCharacterListUpdate(ns, acct);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), expected); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
[Fact] [Fact]
@ -160,8 +160,8 @@ namespace Server.Tests.Network
Packets.SendCharacterList(ns); Packets.SendCharacterList(ns);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), expected); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
[Fact] [Fact]
@ -185,8 +185,8 @@ namespace Server.Tests.Network
Packets.SendCharacterList(ns); Packets.SendCharacterList(ns);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), expected); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
[Fact] [Fact]
@ -198,8 +198,8 @@ namespace Server.Tests.Network
using var ns = PacketTestUtilities.CreateTestNetState(); using var ns = PacketTestUtilities.CreateTestNetState();
Packets.SendAccountLoginRejected(ns, reason); Packets.SendAccountLoginRejected(ns, reason);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), expected); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
[Fact] [Fact]
@ -217,8 +217,8 @@ namespace Server.Tests.Network
Packets.SendAccountLoginAck(ns); Packets.SendAccountLoginAck(ns);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), expected); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
[Fact] [Fact]
@ -233,8 +233,8 @@ namespace Server.Tests.Network
Packets.SendPlayServerAck(ns, si, authId); Packets.SendPlayServerAck(ns, si, authId);
ns.SendPipe.Reader.TryRead(out var buffer); var result = ns.SendPipe.Reader.TryRead();
AssertThat.Equal(buffer.GetSpan(0), expected); AssertThat.Equal(result.Buffer[0].AsSpan(0), expected);
} }
} }
} }

82
Projects/Server.Tests/Network/PipeTests.cs
View file

@ -26,20 +26,18 @@ namespace Server.Tests.Network
DelayedExecute(() => DelayedExecute(() =>
{ {
; // Write some data into the pipe ; // Write some data into the pipe
writer.GetAvailable(out var buffer); var result = writer.TryGetMemory();
Assert.True(buffer.Length == 99); Assert.True(result.Buffer[0].Count == 99);
buffer[0] = 0x1; result.Buffer[0][0] = 0x1;
buffer[1] = 0x2; result.Buffer[0][1] = 0x2;
buffer[2] = 0x3; result.Buffer[0][2] = 0x3;
writer.Advance(3); writer.Advance(3);
writer.Flush(); writer.Flush();
}); });
var segments = new ArraySegment<byte>[2]; var result = await reader.Read();
await reader.Read(segments); Assert.Equal(3, result.Buffer[0].Count);
Assert.Equal(3, segments[0].Count);
} }
private bool _signal; private bool _signal;
@ -53,11 +51,11 @@ namespace Server.Tests.Network
while (count < 0x8000000) 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); Assert.True(first[i] == expected_value);
count++; count++;
@ -68,8 +66,8 @@ namespace Server.Tests.Network
} }
} }
var second = buffer.GetSpan(1); var second = result.Buffer[1];
for (int i = 0; i < second.Length; i++) for (int i = 0; i < second.Count; i++)
{ {
Assert.True(second[i] == expected_value); Assert.True(second[i] == expected_value);
count++; count++;
@ -80,7 +78,7 @@ namespace Server.Tests.Network
} }
} }
reader.Advance((uint)buffer.Length); reader.Advance((uint)result.Length);
} }
_signal = true; _signal = true;
@ -100,14 +98,14 @@ namespace Server.Tests.Network
while (count < 0x8000000) while (count < 0x8000000)
{ {
writer.GetAvailable(out var buffer); var result = writer.TryGetMemory();
if (buffer.Length < 16) if (result.Length < 16)
{ {
continue; 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, 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 reader = pipe.Reader;
var writer = pipe.Writer; 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()); Assert.Equal(0u, reader.GetAvailable());
reader.TryRead(out buffer); result = reader.TryRead();
Assert.Equal(0, buffer.Length); Assert.Equal(0, result.Length);
writer.Advance(7); writer.Advance(7);
writer.GetAvailable(out buffer); result = writer.TryGetMemory();
Assert.Equal(2, buffer.Length); Assert.Equal(2, result.Length);
Assert.Equal(7u, reader.GetAvailable()); Assert.Equal(7u, reader.GetAvailable());
reader.TryRead(out buffer); result = reader.TryRead();
Assert.Equal(7, buffer.Length); Assert.Equal(7, result.Length);
reader.Advance(4); reader.Advance(4);
writer.GetAvailable(out buffer); result = writer.TryGetMemory();
Assert.Equal(6, buffer.Length); Assert.Equal(6, result.Length);
Assert.Equal(3u, reader.GetAvailable()); Assert.Equal(3u, reader.GetAvailable());
reader.TryRead(out buffer); result = reader.TryRead();
Assert.Equal(3, buffer.Length); Assert.Equal(3, result.Length);
writer.Advance(3); writer.Advance(3);
writer.GetAvailable(out buffer); result = writer.TryGetMemory();
Assert.Equal(3, buffer.Length); Assert.Equal(3, result.Length);
Assert.Equal(6u, reader.GetAvailable()); Assert.Equal(6u, reader.GetAvailable());
reader.TryRead(out buffer); result = reader.TryRead();
Assert.Equal(6, buffer.Length); Assert.Equal(6, result.Length);
} }
[Fact] [Fact]
@ -191,19 +189,19 @@ namespace Server.Tests.Network
var reader = pipe.Reader; var reader = pipe.Reader;
var writer = pipe.Writer; var writer = pipe.Writer;
writer.GetAvailable(out var buffer); var result = writer.TryGetMemory();
Assert.True(buffer.Length == 9); 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.Advance(9);
writer.Flush(); writer.Flush();
Assert.Equal(9u, reader.GetAvailable()); 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++) for (int i = 0; i < 9; i++)
{ {
@ -211,9 +209,9 @@ namespace Server.Tests.Network
} }
reader.Advance(4); reader.Advance(4);
reader.TryRead(out buffer); result = reader.TryRead();
first = buffer.GetSpan(0); Assert.Equal(5, result.Length);
Assert.Equal(5, buffer.Length); first = result.Buffer[0];
Assert.Equal(4, first[0]); Assert.Equal(4, first[0]);
Assert.Equal(5, first[1]); Assert.Equal(5, first[1]);
Assert.Equal(6, first[2]); Assert.Equal(6, first[2]);

2
Projects/Server/Buffers/CircularBufferReader.cs
View file

@ -35,7 +35,7 @@ namespace Server.Network
{ {
} }
public CircularBufferReader(CircularBuffer<byte> buffer) : this(buffer.GetSpan(0), buffer.GetSpan(1)) public CircularBufferReader(ArraySegment<byte>[] buffer) : this(buffer[0], buffer[1])
{ {
} }

4
Projects/Server/Buffers/CircularBufferWriter.cs
View file

@ -33,6 +33,10 @@ namespace System.Buffers
{ {
} }
public CircularBufferWriter(ArraySegment<byte>[] buffer) : this(buffer[0], buffer[1])
{
}
public CircularBufferWriter(Span<byte> first, Span<byte> second) public CircularBufferWriter(Span<byte> first, Span<byte> second)
{ {
_first = first; _first = first;

48
Projects/Server/Network/NetState/NetState.cs
View file

@ -36,8 +36,8 @@ namespace Server.Network
public partial class NetState : IComparable<NetState>, IDisposable public partial class NetState : IComparable<NetState>, IDisposable
{ {
private static int RecvPipeSize = 1024 * 64; private static int RecvPipeSize = 1024 * 64 + 1;
private static int SendPipeSize = 1024 * 256; private static int SendPipeSize = 1024 * 256 + 1;
private static int GumpCap = 512; private static int GumpCap = 512;
private static int HuePickerCap = 512; private static int HuePickerCap = 512;
private static int MenuCap = 512; private static int MenuCap = 512;
@ -369,9 +369,17 @@ namespace Server.Network
NetworkState.Resume(ref m_NetworkState); NetworkState.Resume(ref m_NetworkState);
} }
public bool GetSendBuffer(out CircularBuffer<byte> cBuffer)
{
var result = SendPipe.Writer.TryGetMemory();
cBuffer = new CircularBuffer<byte>(result.Buffer);
return !(result.IsClosed || result.Length <= 0);
}
public virtual void Send(ref CircularBuffer<byte> buffer, int length) public virtual void Send(ref CircularBuffer<byte> buffer, int length)
{ {
if (Connection == null || BlockAllPackets || buffer.Length == 0 || length <= 0) if (Connection == null || BlockAllPackets || length <= 0)
{ {
return; return;
} }
@ -407,15 +415,16 @@ namespace Server.Network
if (buffer.Length > 0 && length > 0) if (buffer.Length > 0 && length > 0)
{ {
if (!SendPipe.Writer.GetAvailable(out var pipeBuffer)) var result = writer.TryGetMemory();
if (result.IsClosed)
{ {
p.OnSend(); p.OnSend();
return; 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); writer.Advance((uint)length);
// Flush at the end of the game loop // Flush at the end of the game loop
@ -461,24 +470,23 @@ namespace Server.Network
private async void SendTask(object state) private async void SendTask(object state)
{ {
var reader = SendPipe.Reader; var reader = SendPipe.Reader;
var segments = new ArraySegment<byte>[2];
try try
{ {
while (m_Running) while (m_Running)
{ {
var result = await reader.Read(segments); var result = await reader.Read();
if (result.Closed) if (result.IsClosed)
{ {
break; break;
} }
if (segments[0].Count + segments[1].Count <= 0) if (result.Length <= 0)
{ {
continue; continue;
} }
var bytesWritten = await Connection.SendAsync(segments, SocketFlags.None); var bytesWritten = await Connection.SendAsync(result.Buffer, SocketFlags.None);
if (bytesWritten > 0) if (bytesWritten > 0)
{ {
@ -510,7 +518,6 @@ namespace Server.Network
{ {
var socket = Connection; var socket = Connection;
var writer = RecvPipe.Writer; var writer = RecvPipe.Writer;
var segments = new ArraySegment<byte>[2];
try try
{ {
@ -521,24 +528,25 @@ namespace Server.Network
continue; continue;
} }
// TODO: Make awaitable var result = await writer.GetMemory();
if (!writer.GetAvailable(segments))
if (result.IsClosed)
{ {
break; break;
} }
if (segments[0].Count + segments[1].Count <= 0) if (result.Length <= 0)
{ {
continue; continue;
} }
var bytesWritten = await socket.ReceiveAsync(segments, SocketFlags.None); var bytesWritten = await socket.ReceiveAsync(result.Buffer, SocketFlags.None);
if (bytesWritten <= 0) if (bytesWritten <= 0)
{ {
break; break;
} }
DecodePacket(segments, ref bytesWritten); DecodePacket(result.Buffer, ref bytesWritten);
writer.Advance((uint)bytesWritten); writer.Advance((uint)bytesWritten);
m_NextCheckActivity = Core.TickCount + 90000; m_NextCheckActivity = Core.TickCount + 90000;
@ -583,12 +591,14 @@ namespace Server.Network
// Process as many packets as we can synchronously // Process as many packets as we can synchronously
while (true) while (true)
{ {
if (!reader.TryRead(out var buffer) || buffer.Length <= 0) var result = reader.TryRead();
if (result.IsClosed || result.Length <= 0)
{ {
return; return;
} }
var bytesProcessed = PacketHandlers.ProcessPacket(this, ref buffer); var bytesProcessed = PacketHandlers.ProcessPacket(this, result.Buffer);
if (bytesProcessed <= 0) if (bytesProcessed <= 0)
{ {

4
Projects/Server/Network/PacketHandlers.cs
View file

@ -277,9 +277,9 @@ namespace Server.Network
} }
} }
public static int ProcessPacket(NetState ns, ref CircularBuffer<byte> buffer) public static int ProcessPacket(NetState ns, ArraySegment<byte>[] buffer)
{ {
var reader = new CircularBufferReader(ref buffer); var reader = new CircularBufferReader(buffer);
var packetId = reader.ReadByte(); var packetId = reader.ReadByte();

24
Projects/Server/Network/Packets/AccountPackets.cs
View file

@ -61,7 +61,7 @@ namespace Server.Network
*/ */
public static void SendChangeCharacter(NetState ns, IAccount a) 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; return;
} }
@ -109,7 +109,7 @@ namespace Server.Network
*/ */
public static void SendClientVersionRequest(NetState ns) 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[0] = 0xBD; // Packet ID
buffer[1] = 0x00; buffer[1] = 0x00;
@ -127,7 +127,7 @@ namespace Server.Network
*/ */
public static void SendCharacterDeleteResult(NetState ns, DeleteResultType res) 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[0] = 0x85; // Packet ID
buffer[1] = (byte)res; buffer[1] = (byte)res;
@ -144,7 +144,7 @@ namespace Server.Network
*/ */
public static void SendPopupMessage(NetState ns, PMMessage msg) 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[0] = 0x53; // Packet ID
buffer[1] = (byte)msg; buffer[1] = (byte)msg;
@ -161,7 +161,7 @@ namespace Server.Network
*/ */
public static void SendSupportedFeature(NetState ns) 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; return;
} }
@ -206,7 +206,7 @@ namespace Server.Network
*/ */
public static void SendLoginConfirmation(NetState ns, Mobile m) 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; return;
} }
@ -247,7 +247,7 @@ namespace Server.Network
*/ */
public static void SendLoginComplete(NetState ns) 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 buffer[0] = 0x55; // Packet ID
@ -263,7 +263,7 @@ namespace Server.Network
*/ */
public static void SendCharacterListUpdate(NetState ns, IAccount a) 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; return;
} }
@ -316,7 +316,7 @@ namespace Server.Network
{ {
var acct = ns?.Account; var acct = ns?.Account;
if (acct == null || !ns.SendPipe.Writer.GetAvailable(out var buffer)) if (acct == null || !ns.GetSendBuffer(out var buffer))
{ {
return; return;
} }
@ -415,7 +415,7 @@ namespace Server.Network
*/ */
public static void SendAccountLoginRejected(NetState ns, ALRReason reason) 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[0] = 0x82; // Packet ID
buffer[1] = (byte)reason; buffer[1] = (byte)reason;
@ -432,7 +432,7 @@ namespace Server.Network
*/ */
public static void SendAccountLoginAck(NetState ns) 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; return;
} }
@ -469,7 +469,7 @@ namespace Server.Network
*/ */
public static void SendPlayServerAck(NetState ns, ServerInfo si, int authId) 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; return;
} }

284
Projects/Server/Network/Pipe.cs
View file

@ -14,7 +14,6 @@
*************************************************************************/ *************************************************************************/
using System; using System;
using System.Buffers;
using System.Runtime.CompilerServices; using System.Runtime.CompilerServices;
using System.Threading; using System.Threading;
@ -27,78 +26,125 @@ namespace Server.Network
public bool IsCompleted { get; } public bool IsCompleted { get; }
public T GetResult(); public T GetResult();
public void OnCompleted(Action continuation);
} }
public class Pipe<T> public class Pipe<T>
{ {
public readonly struct Result public struct Result
{ {
public bool Closed { get; } public ArraySegment<T>[] 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<T> 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<T>[segments];
}
} }
public class PipeWriter<T> public class PipeWriter : IPipeTask<Result>
{ {
private readonly Pipe<T> _pipe; private readonly Pipe<T> _pipe;
public PipeWriter(Pipe<T> pipe) => _pipe = pipe; private Result _result = new Result(2);
internal PipeWriter(Pipe<T> pipe) => _pipe = pipe;
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool GetAvailable(ArraySegment<T>[] segments) public uint GetAvailable()
{ {
var read = _pipe._readIdx; var read = _pipe._readIdx;
var write = _pipe._writeIdx; var write = _pipe._writeIdx;
if (read <= write) if (read <= write)
{ {
var readZero = read == 0; if (read == 0)
var sz = _pipe.Size - write - (readZero ? 1 : 0); {
return _pipe.Size - write - 1;
}
segments[0] = sz == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, (int)write, (int)sz); return _pipe.Size - write + (read - 1);
segments[1] = readZero ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, 0, (int)read - 1);
}
else
{
var sz = read - write - 1;
segments[0] = sz == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, (int)write, (int)sz);
segments[1] = ArraySegment<T>.Empty;
} }
return !_pipe._closed; return read - write - 1;
} }
[MethodImpl(MethodImplOptions.AggressiveInlining)] public Result TryGetMemory()
public bool GetAvailable(out CircularBuffer<T> buffer)
{ {
var read = _pipe._readIdx; var read = _pipe._readIdx;
var write = _pipe._writeIdx; var write = _pipe._writeIdx;
Span<T> first; _result.IsClosed = _pipe._closed;
Span<T> second;
if (read <= write) if (read <= write)
{ {
var readZero = read == 0; var readZero = read == 0;
var sz = _pipe.Size - write - (readZero ? 1 : 0); var sz = _pipe.Size - write - (readZero ? 1 : 0);
first = sz == 0 ? Span<T>.Empty : _pipe._buffer.AsSpan((int)write, (int)sz); _result.Buffer[0] = sz == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, (int)write, (int)sz);
second = readZero ? Span<T>.Empty : _pipe._buffer.AsSpan(0, (int)read - 1); _result.Buffer[1] = readZero ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, 0, (int)read - 1);
} }
else else
{ {
var sz = read - write - 1; var sz = read - write - 1;
first = sz == 0 ? Span<T>.Empty : _pipe._buffer.AsSpan((int)write, (int)sz); _result.Buffer[0] = sz == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, (int)write, (int)sz);
second = Span<T>.Empty; _result.Buffer[1] = ArraySegment<T>.Empty;
} }
buffer = new CircularBuffer<T>(first, second); return _result;
}
return !_pipe._closed; public IPipeTask<Result> GetMemory()
{
if (_pipe._writeAwaitBeginning)
{
throw new Exception("Double await on reader");
}
return this;
} }
public void Advance(uint count) public void Advance(uint count)
@ -171,7 +217,12 @@ namespace Server.Network
public void Flush() public void Flush()
{ {
var waiting = _pipe._awaitBeginning; if (_pipe._readIdx == _pipe._writeIdx)
{
return;
}
var waiting = _pipe._readAwaitBeginning;
if (!waiting) if (!waiting)
{ {
@ -182,20 +233,48 @@ namespace Server.Network
do do
{ {
continuation = _pipe._readerContinuation; continuation = _pipe._readContinuation;
} while (continuation == null); } while (continuation == null);
_pipe._readerContinuation = null; _pipe._readContinuation = null;
_pipe._awaitBeginning = false; _pipe._readAwaitBeginning = false;
ThreadPool.UnsafeQueueUserWorkItem(state => continuation(), true); 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<Result> 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<T> : IPipeTask<Result> public class PipeReader : IPipeTask<Result>
{ {
private readonly Pipe<T> _pipe; private readonly Pipe<T> _pipe;
private Result _result = new Result(2);
internal PipeReader(Pipe<T> pipe) => _pipe = pipe; internal PipeReader(Pipe<T> pipe) => _pipe = pipe;
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
@ -212,61 +291,37 @@ namespace Server.Network
return write + _pipe.Size - read; return write + _pipe.Size - read;
} }
private ArraySegment<T>[] _segments; public Result TryRead()
public IPipeTask<Result> Read(ArraySegment<T>[] segments)
{ {
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<T>.Empty : new ArraySegment<T>(_pipe._buffer, (int)read, (int)(write - read));
_result.Buffer[1] = ArraySegment<T>.Empty;
}
else
{
_result.Buffer[0] = _pipe.Size - read == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, (int)read, (int)(_pipe.Size - read));
_result.Buffer[1] = write == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, 0, (int)write);
}
return _result;
}
public IPipeTask<Result> Read()
{
if (_pipe._readAwaitBeginning)
{ {
throw new Exception("Double await on reader"); throw new Exception("Double await on reader");
} }
_segments = segments;
return this; return this;
} }
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool TryRead(out CircularBuffer<T> buffer)
{
var read = _pipe._readIdx;
var write = _pipe._writeIdx;
Span<T> first;
Span<T> second;
if (read <= write)
{
first = write - read == 0 ? Span<T>.Empty : _pipe._buffer.AsSpan((int)read, (int)(write - read));
second = Span<T>.Empty;
}
else
{
first = _pipe.Size - read == 0 ? Span<T>.Empty : _pipe._buffer.AsSpan((int)read, (int)(_pipe.Size - read));
second = write == 0 ? Span<T>.Empty : _pipe._buffer.AsSpan(0, (int)write);
}
buffer = new CircularBuffer<T>(first, second);
return !_pipe._closed;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void TryRead(ArraySegment<T>[] segments)
{
var read = _pipe._readIdx;
var write = _pipe._writeIdx;
if (read <= write)
{
segments[0] = write - read == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, (int)read, (int)(write - read));
segments[1] = ArraySegment<T>.Empty;
}
else
{
segments[0] = _pipe.Size - read == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, (int)read, (int)(_pipe.Size - read));
segments[1] = write == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, 0, (int)write);
}
}
public void Advance(uint count) public void Advance(uint count)
{ {
var read = _pipe._readIdx; var read = _pipe._readIdx;
@ -306,6 +361,33 @@ namespace Server.Network
_pipe._readIdx = read; _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 #region Awaitable
// The following makes it possible to await the reader. Do not use any of this directly. // 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; return true;
} }
_pipe._awaitBeginning = true; _pipe._readAwaitBeginning = true;
return false; return false;
} }
} }
public Result GetResult() public Result GetResult() => TryRead();
{
if (_pipe._closed)
{
_segments = null;
return new Result(true);
}
if (_segments != null) public void OnCompleted(Action continuation) => _pipe._readContinuation = continuation;
{
TryRead(_segments);
_segments = null;
}
return new Result(false);
}
public void OnCompleted(Action continuation) => _pipe._readerContinuation = continuation;
#endregion #endregion
} }
@ -353,8 +420,8 @@ namespace Server.Network
private volatile uint _readIdx; private volatile uint _readIdx;
private bool _closed; private bool _closed;
public PipeWriter<T> Writer { get; } public PipeWriter Writer { get; }
public PipeReader<T> Reader { get; } public PipeReader Reader { get; }
public uint Size => (uint)_buffer.Length; public uint Size => (uint)_buffer.Length;
@ -365,13 +432,16 @@ namespace Server.Network
_readIdx = 0; _readIdx = 0;
_closed = false; _closed = false;
Writer = new PipeWriter<T>(this); Writer = new PipeWriter(this);
Reader = new PipeReader<T>(this); Reader = new PipeReader(this);
} }
#region Awaitable #region Awaitable
private volatile bool _awaitBeginning; private volatile bool _readAwaitBeginning;
private volatile Action _readerContinuation; private volatile Action _readContinuation;
private volatile bool _writeAwaitBeginning;
private volatile Action _writeContinuation;
#endregion #endregion
} }