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

feat: Networking v3 using epoll/kqueue (#813)

Browse source 
* Replaces networking internals with epoll/kqueue
This commit is contained in:
Kamron Batman 2021-10-03 18:04:53 -07:00 • committed by GitHub
parent 8eaa859332
commit a845f8a1c0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 686 additions and 198 deletions
Download patch file Download diff file Expand all files Collapse all files

303
Projects/Server/Network/NetState/NetState.cs Normal file → Executable file
View file

@ -21,7 +21,7 @@ using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Runtime.InteropServices;
using Server.Accounting;
using Server.Diagnostics;
using Server.Gumps;
@ -43,12 +43,15 @@ namespace Server.Network
private const int RecvPipeSize = 1024 * 64;
private const int SendPipeSize = 1024 * 256;
private static int GumpCap = 512;
private static int HuePickerCap = 512;
private static int MenuCap = 512;
private static int PacketPerSecondThreshold = 3000;
private const int GumpCap = 512;
private const int HuePickerCap = 512;
private const int MenuCap = 512;
private const int PacketPerSecondThreshold = 3000;
private static NetState[] _polledStates = new NetState[2048];
private static readonly IPollGroup _pollGroup = PollGroup.Create();
private static readonly Queue<NetState> FlushPending = new(2048);
private static readonly Queue<NetState> FlushedPartials = new(2048);
private static readonly ConcurrentQueue<NetState> Disposed = new();
public static NetStateCreatedCallback CreatedCallback { get; set; }
@ -56,7 +59,7 @@ namespace Server.Network
private readonly string _toString;
private ClientVersion _version;
private long _nextActivityCheck;
private int _running;
private bool _running = true;
private volatile DecodePacket _packetDecoder;
private volatile EncodePacket _packetEncoder;
private bool _flushQueued;
@ -66,14 +69,12 @@ namespace Server.Network
internal int _authId;
internal int _seed;
internal ParserState _parserState = ParserState.Uninitialized;
internal ProtocolState _protocolState = ProtocolState.Uninitialized;
internal RecvState _recvState = RecvState.Uninitialized;
internal SendState _sendState = SendState.Uninitialized;
internal ParserState _parserState = ParserState.AwaitingNextPacket;
internal ProtocolState _protocolState = ProtocolState.AwaitingSeed;
internal GCHandle _handle;
internal enum ParserState
{
Uninitialized,
AwaitingNextPacket,
AwaitingPartialPacket,
ProcessingPacket,
@ -83,7 +84,6 @@ namespace Server.Network
internal enum ProtocolState
{
Uninitialized,
AwaitingSeed, // Based on the way the seed arrives, we know if this is a login server or a game server connection
LoginServer_AwaitingLogin,
@ -96,32 +96,6 @@ namespace Server.Network
Error
}
internal enum RecvState
{
Uninitialized,
AwaitingMemory,
AwaitingRecv,
DataReceived,
Exited,
}
internal enum SendState
{
Uninitialized,
AwaitingData,
Sending,
SendCompleted,
Exited,
}
public static void Configure()
{
GumpCap = ServerConfiguration.GetOrUpdateSetting("netstate.gumpCap", GumpCap);
HuePickerCap = ServerConfiguration.GetOrUpdateSetting("netstate.huePickerCap", HuePickerCap);
MenuCap = ServerConfiguration.GetOrUpdateSetting("netstate.menuCap", MenuCap);
PacketPerSecondThreshold = ServerConfiguration.GetOrUpdateSetting("netstate.packetsPerSecondThreshold", PacketPerSecondThreshold);
}
public static void Initialize()
{
Timer.DelayCall(TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(1.5), CheckAllAlive);
@ -138,6 +112,7 @@ namespace Server.Network
RecvPipe = new Pipe<byte>(GC.AllocateUninitializedArray<byte>(RecvPipeSize));
SendPipe = new Pipe<byte>(GC.AllocateUninitializedArray<byte>(SendPipeSize));
_nextActivityCheck = Core.TickCount + 30000;
ConnectedOn = Core.Now;
try
{
@ -151,7 +126,17 @@ namespace Server.Network
_toString = "(error)";
}
ConnectedOn = Core.Now;
_handle = GCHandle.Alloc(this);
try
{
_pollGroup.Add(this);
}
catch (Exception ex)
{
TraceException(ex);
Disconnect("Unable to add socket to poll group");
}
CreatedCallback?.Invoke(this);
}
@ -180,12 +165,6 @@ namespace Server.Network
public List<SecureTrade> Trades { get; }
public bool Running
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => _running == 1;
}
public bool Seeded { get; set; }
public Pipe<byte> RecvPipe { get; }
@ -524,33 +503,21 @@ namespace Server.Network
}
}
internal void Start()
{
if (Interlocked.CompareExchange(ref _running, 1, 0) == 1 || Connection == null)
{
return;
}
_parserState = ParserState.AwaitingNextPacket;
_protocolState = ProtocolState.AwaitingSeed;
ThreadPool.UnsafeQueueUserWorkItem(RecvTask, null);
ThreadPool.UnsafeQueueUserWorkItem(SendTask, null);
}
public void HandleReceive()
{
if (!Running)
if (!_running)
{
return;
}
ReceiveData();
var reader = RecvPipe.Reader;
try
{
// Process as many packets as we can synchronously
while (Running && _parserState != ParserState.Error && _protocolState != ProtocolState.Error)
while (_running && _parserState != ParserState.Error && _protocolState != ProtocolState.Error)
{
var result = reader.TryRead();
var length = result.Length;
@ -574,12 +541,6 @@ namespace Server.Network
{
switch (_protocolState)
{
case ProtocolState.Uninitialized:
{
HandleError(packetId, packetLength);
return;
}
case ProtocolState.AwaitingSeed:
{
if (packetId == 0xEF)
@ -803,83 +764,56 @@ namespace Server.Network
return ParserState.AwaitingNextPacket;
}
private async void SendTask(object state)
private bool Flush()
{
_flushQueued = false;
if (Connection == null)
{
return true;
}
SendPipe.Writer.Flush();
var reader = SendPipe.Reader;
var result = reader.TryRead();
if (result.IsClosed || result.Length == 0)
{
return true;
}
var bytesWritten = 0;
try
{
while (Running)
{
_sendState = SendState.AwaitingData;
var result = await reader.Read();
if (result.IsClosed || !Running)
{
break;
}
if (result.Length <= 0)
{
continue;
}
_sendState = SendState.Sending;
var bytesWritten = await Connection.SendAsync(result.Buffer, SocketFlags.None);
_sendState = SendState.SendCompleted;
if (bytesWritten > 0)
{
_nextActivityCheck = Core.TickCount + 90000;
reader.Advance((uint)bytesWritten);
}
}
// Grab any remaining data and flush it
var data = reader.TryRead();
if (data.Length > 0)
{
_sendState = SendState.Sending;
Connection.Send(data.Buffer, SocketFlags.None);
reader.Advance((uint)data.Length);
}
bytesWritten = Connection.Send(result.Buffer, SocketFlags.None);
}
catch (SocketException ex)
{
// If the user closes the connection (or the recv side does)
// between the check for m_Running above and the call to SendAsync,
// we can still get a socket exception here. That's ok.
// Socket exceptions are generally ok, just spammy
#if DEBUG
Console.WriteLine(ex);
#endif
Disconnect(string.Empty);
}
catch (Exception ex)
{
#if DEBUG
Console.WriteLine(ex);
#endif
Disconnect($"Disconnected with error: {ex}");
TraceException(ex);
}
finally
{
try
{
Connection.Shutdown(SocketShutdown.Both);
Connection.Close();
}
catch (Exception ex)
{
TraceException(ex);
}
Disconnect("Exiting SendTask.");
_sendState = SendState.Exited;
if (bytesWritten > 0)
{
_nextActivityCheck = Core.TickCount + 90000;
reader.Advance((uint)bytesWritten);
}
return bytesWritten == result.Length;
}
private void DecodePacket(ArraySegment<byte>[] buffer, ref int length)
@ -888,47 +822,21 @@ namespace Server.Network
_packetDecoder?.Invoke(cBuffer, ref length);
}
private async void RecvTask(object state)
private void ReceiveData()
{
var socket = Connection;
var writer = RecvPipe.Writer;
var result = writer.TryGetMemory();
if (result.IsClosed || result.Length == 0)
{
return;
}
var bytesWritten = 0;
try
{
while (Running)
{
_recvState = RecvState.AwaitingMemory;
var result = await writer.GetMemory();
if (result.IsClosed || !Running)
{
break;
}
if (result.Length <= 0)
{
continue;
}
_recvState = RecvState.AwaitingRecv;
var bytesWritten = await socket.ReceiveAsync(result.Buffer, SocketFlags.None);
if (bytesWritten <= 0)
{
break;
}
_recvState = RecvState.DataReceived;
DecodePacket(result.Buffer, ref bytesWritten);
writer.Advance((uint)bytesWritten);
_nextActivityCheck = Core.TickCount + 90000;
// No need to flush
}
Disconnect(string.Empty);
bytesWritten = Connection.Receive(result.Buffer, SocketFlags.None);
}
catch (SocketException ex)
{
@ -945,31 +853,20 @@ namespace Server.Network
#if DEBUG
Console.WriteLine(ex);
#endif
Disconnect("RecvTask exited unexpectedly.");
Disconnect($"Disconnected with error: {ex}");
TraceException(ex);
}
finally
{
_recvState = RecvState.Exited;
}
}
public static void HandleAllReceives()
{
foreach (var ns in TcpServer.Instances)
if (bytesWritten <= 0)
{
ns.HandleReceive();
}
}
public void Flush()
{
if (Connection != null)
{
SendPipe.Writer.Flush();
Disconnect(string.Empty);
return;
}
_flushQueued = false;
DecodePacket(result.Buffer, ref bytesWritten);
writer.Advance((uint)bytesWritten);
_nextActivityCheck = Core.TickCount + 90000;
}
public static void FlushAll()
@ -982,16 +879,36 @@ namespace Server.Network
public static void Slice()
{
while (FlushPending.Count != 0)
int count = _pollGroup.Poll(ref _polledStates);
if (count > 0)
{
FlushPending.Dequeue()?.Flush();
for (int i = 0; i < count; i++)
{
_polledStates[i].HandleReceive();
_polledStates[i] = null;
}
}
while (FlushPending.TryDequeue(out var ns))
{
if (!ns.Flush())
{
// Incomplete data, so we need to requeue
FlushedPartials.Enqueue(ns);
}
}
var hasDisposes = !Disposed.IsEmpty;
while (Disposed.TryDequeue(out var ns))
{
TcpServer.Instances.Remove(ns);
ns.Dispose();
}
if (hasDisposes)
{
_pollGroup.Poll(ref _polledStates);
}
}
public void CheckAlive(long curTicks)
@ -1045,11 +962,13 @@ namespace Server.Network
public void Disconnect(string reason)
{
if (Interlocked.CompareExchange(ref _running, 0, 1) == 0)
if (!_running)
{
return;
}
_running = false;
try
{
if (_disconnectReason != string.Empty)
@ -1063,7 +982,6 @@ namespace Server.Network
}
_disconnectReason = reason;
Disposed.Enqueue(this);
}
@ -1095,6 +1013,11 @@ namespace Server.Network
{
TraceDisconnect(_disconnectReason, _toString);
if (_running)
{
throw new Exception("Disconnected a NetState that is still running.");
}
#if THREADGUARD
if (Thread.CurrentThread != Core.Thread)
{
@ -1106,8 +1029,18 @@ namespace Server.Network
}
#endif
RecvPipe.Writer.Close();
SendPipe.Writer.Close();
TcpServer.Instances.Remove(this);
try
{
_pollGroup.Remove(this);
}
catch (Exception ex)
{
TraceException(ex);
}
Connection.Close();
_handle.Free();
var m = Mobile;
Mobile = null;