erik/acdream
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

perf(net): own one pooled async receive

Browse source 
Replace timeout-polled in-world UDP receives with one cancellable caller-buffered socket operation. Transfer only right-sized pooled datagrams through the FIFO, return every ownership edge deterministically, and send caller spans without a transport copy while preserving handshake pacing and ACK order.
This commit is contained in:
Erik 2026-07-25 05:50:18 +02:00
parent a2a1e5916d
commit 7211bb1bf7
10 changed files with 547 additions and 87 deletions
Download patch file Download diff file Expand all files Collapse all files

91
src/AcDream.Core.Net/NetClient.cs
View file

@ -4,22 +4,20 @@ using System.Net.Sockets;
namespace AcDream.Core.Net;
/// <summary>
/// Minimum-viable UDP transport for acdream. Wraps a <see cref="UdpClient"/>
/// with synchronous send + timeout-based receive — good enough for the
/// Phase 4.6 handshake smoke test and early state-machine bring-up.
/// UDP transport for acdream. The live world owns one cancellable,
/// caller-buffered asynchronous receive after the synchronous handshake.
///
/// <para>
/// <b>Not yet provided</b> (deferred to a later phase once the handshake
/// actually works): background receive thread, outbound queue, ack/retransmit
/// window, heartbeat timer, concurrent send/receive. The acdream game loop
/// will need a real async pump eventually but building that now would be
/// debugging two things at once when we hit the first protocol mismatch.
/// Sends consume caller spans directly through <see cref="Socket"/> without
/// materializing a transport-only array copy.
/// </para>
/// </summary>
public sealed class NetClient : IDisposable
{
private readonly UdpClient _udp;
private readonly IPEndPoint _remote;
private readonly IPEndPoint _anyRemote =
new(IPAddress.Any, 0);
/// <summary>Last <see cref="Socket.ReceiveTimeout"/> value applied to the
/// underlying socket, so repeated <see cref="Receive"/> calls with the
@ -57,7 +55,10 @@ public sealed class NetClient : IDisposable
/// </summary>
public void Send(ReadOnlySpan<byte> datagram)
{
_udp.Send(datagram.ToArray(), datagram.Length, _remote);
_udp.Client.SendTo(
datagram,
SocketFlags.None,
_remote);
}
/// <summary>
@ -68,7 +69,63 @@ public sealed class NetClient : IDisposable
/// </summary>
public void Send(IPEndPoint remote, ReadOnlySpan<byte> datagram)
{
_udp.Send(datagram.ToArray(), datagram.Length, remote);
_udp.Client.SendTo(
datagram,
SocketFlags.None,
remote);
}
/// <summary>
/// Receive into caller-owned storage. Returns the received byte count,
/// or <c>-1</c> when <paramref name="timeout"/> expires.
/// </summary>
public int Receive(
Span<byte> destination,
TimeSpan timeout,
out IPEndPoint? from)
{
int timeoutMs = checked((int)timeout.TotalMilliseconds);
if (timeoutMs != _lastAppliedReceiveTimeoutMs)
{
_udp.Client.ReceiveTimeout = timeoutMs;
_lastAppliedReceiveTimeoutMs = timeoutMs;
}
try
{
EndPoint remote = _anyRemote;
int length = _udp.Client.ReceiveFrom(
destination,
SocketFlags.None,
ref remote);
from = (IPEndPoint)remote;
return length;
}
catch (SocketException ex)
when (ex.SocketErrorCode == SocketError.TimedOut)
{
from = null;
return -1;
}
}
/// <summary>
/// One cancellable asynchronous receive into caller-owned storage.
/// The returned memory remains owned by the caller.
/// </summary>
public async ValueTask<NetReceiveResult> ReceiveAsync(
Memory<byte> destination,
CancellationToken cancellationToken)
{
SocketReceiveFromResult result =
await _udp.Client.ReceiveFromAsync(
destination,
SocketFlags.None,
_anyRemote,
cancellationToken).ConfigureAwait(false);
return new NetReceiveResult(
result.ReceivedBytes,
(IPEndPoint)result.RemoteEndPoint);
}
/// <summary>
@ -79,7 +136,7 @@ public sealed class NetClient : IDisposable
/// </summary>
public byte[]? Receive(TimeSpan timeout, out IPEndPoint? from)
{
int timeoutMs = (int)timeout.TotalMilliseconds;
int timeoutMs = checked((int)timeout.TotalMilliseconds);
if (timeoutMs != _lastAppliedReceiveTimeoutMs)
{
_udp.Client.ReceiveTimeout = timeoutMs;
@ -88,11 +145,12 @@ public sealed class NetClient : IDisposable
try
{
IPEndPoint any = new(IPAddress.Any, 0);
var bytes = _udp.Receive(ref any);
byte[] bytes = _udp.Receive(ref any);
from = any;
return bytes;
}
catch (SocketException ex) when (ex.SocketErrorCode == SocketError.TimedOut)
catch (SocketException ex)
when (ex.SocketErrorCode == SocketError.TimedOut)
{
from = null;
return null;
@ -129,3 +187,10 @@ public sealed class NetClient : IDisposable
public void Dispose() => _udp.Dispose();
}
/// <summary>
/// Result of receiving one UDP datagram into caller-owned memory.
/// </summary>
public readonly record struct NetReceiveResult(
int Length,
IPEndPoint RemoteEndPoint);

240
src/AcDream.Core.Net/WorldSession.cs
View file

@ -1,3 +1,4 @@
using System.Buffers;
using System.Buffers.Binary;
using System.Diagnostics;
using System.Net;
@ -14,7 +15,13 @@ internal interface IWorldSessionTransport : IDisposable
{
void Send(ReadOnlySpan<byte> datagram);
void Send(IPEndPoint remote, ReadOnlySpan<byte> datagram);
byte[]? Receive(TimeSpan timeout, out IPEndPoint? from);
int Receive(
Span<byte> destination,
TimeSpan timeout,
out IPEndPoint? from);
ValueTask<NetReceiveResult> ReceiveAsync(
Memory<byte> destination,
CancellationToken cancellationToken);
}
internal sealed class NetClientWorldSessionTransport(IPEndPoint remote)
@ -27,8 +34,16 @@ internal sealed class NetClientWorldSessionTransport(IPEndPoint remote)
public void Send(IPEndPoint endpoint, ReadOnlySpan<byte> datagram) =>
_client.Send(endpoint, datagram);
public byte[]? Receive(TimeSpan timeout, out IPEndPoint? from) =>
_client.Receive(timeout, out from);
public int Receive(
Span<byte> destination,
TimeSpan timeout,
out IPEndPoint? from) =>
_client.Receive(destination, timeout, out from);
public ValueTask<NetReceiveResult> ReceiveAsync(
Memory<byte> destination,
CancellationToken cancellationToken) =>
_client.ReceiveAsync(destination, cancellationToken);
public void Dispose() => _client.Dispose();
}
@ -677,15 +692,29 @@ public sealed class WorldSession : IDisposable
private int _characterLogOffConfirmed;
private int _disposeStarted;
// Phase A.3: background receive thread buffers raw UDP datagrams into
// a channel so the render thread never blocks on socket I/O.
private readonly Channel<byte[]> _inboundQueue =
Channel.CreateUnbounded<byte[]>(
// Retail CNetLayerPacket stores ProtoHeader separately beside
// m_Data[65484] (named-retail acclient.h, type 3705). Keep one full UDP
// receive buffer at the socket edge; only the actual byte count crosses
// the queue boundary in a right-sized ArrayPool bucket.
internal const int MaxInboundDatagramBytes = ushort.MaxValue;
// Phase A.3 / Slice H-c1: one asynchronous socket owner buffers raw UDP
// datagrams into a FIFO channel so the render thread never blocks on I/O.
private readonly Channel<PooledInboundDatagram> _inboundQueue =
Channel.CreateUnbounded<PooledInboundDatagram>(
new UnboundedChannelOptions
{ SingleReader = true, SingleWriter = true });
private Thread? _netThread;
private Task? _netReceiveTask;
private readonly CancellationTokenSource _netCancel = new();
internal readonly record struct PooledInboundDatagram(
byte[] Buffer,
int Length)
{
public ReadOnlyMemory<byte> Memory =>
Buffer.AsMemory(0, Length);
}
/// <summary>
/// Phase 4.10 latch — true after we've sent the LoginComplete game
/// action in response to PlayerCreate. Prevents re-sending if the
@ -775,11 +804,28 @@ public sealed class WorldSession : IDisposable
Packet? cr = null;
while (DateTime.UtcNow < deadline && cr is null)
{
var bytes = _net.Receive(deadline - DateTime.UtcNow, out _);
if (bytes is null) break;
var dec = PacketCodec.TryDecode(bytes, null);
if (dec.IsOk && dec.Packet!.Header.HasFlag(PacketHeaderFlags.ConnectRequest))
cr = dec.Packet;
PooledInboundDatagram? received =
ReceiveBlocking(deadline - DateTime.UtcNow);
if (received is null)
break;
PooledInboundDatagram datagram = received.Value;
try
{
var dec = PacketCodec.TryDecode(
datagram.Memory.Span,
inboundIsaac: null);
if (dec.IsOk
&& dec.Packet!.Header.HasFlag(
PacketHeaderFlags.ConnectRequest))
{
cr = dec.Packet;
}
}
finally
{
ReturnInboundDatagram(datagram);
}
}
if (cr is null) { Transition(State.Failed); throw new TimeoutException("ConnectRequest not received"); }
@ -876,12 +922,7 @@ public sealed class WorldSession : IDisposable
// During Connect() and EnterWorld(), PumpOnce() read directly
// from the socket (blocking). From here on, Tick() drains the
// channel instead.
_netThread = new Thread(NetReceiveLoop)
{
IsBackground = true,
Name = "acdream.net-recv",
};
_netThread.Start();
_netReceiveTask = NetReceiveLoopAsync();
}
internal readonly record struct EnterWorldSelection(
@ -933,12 +974,20 @@ public sealed class WorldSession : IDisposable
{
int processed = 0;
long start = Stopwatch.GetTimestamp();
while (_inboundQueue.Reader.TryRead(out var bytes))
while (_inboundQueue.Reader.TryRead(
out PooledInboundDatagram datagram))
{
ProcessDatagram(bytes);
try
{
ProcessDatagram(datagram.Memory.Span);
}
finally
{
ReturnInboundDatagram(datagram);
}
processed++;
// Bound ONLY in-world: the handshake uses the blocking PumpOnce path, never Tick
// (the net thread that feeds _inboundQueue starts at Transition(State.InWorld)).
// (the async receive owner starts at Transition(State.InWorld)).
// Acks are queued per packet inside ProcessDatagram BEFORE the heavy handler, so
// deferring the tail only delays the tail's acks a few frames — within ACE's
// tolerance (holtburger defers acks on a flush cadence). The tail stays queued
@ -958,20 +1007,19 @@ public sealed class WorldSession : IDisposable
=> state == State.InWorld && nowTicks - startTicks >= budgetTicks;
/// <summary>
/// Phase A.3: background receive loop. Runs on a dedicated daemon
/// thread started at the end of <see cref="EnterWorld"/>. Continuously
/// pulls raw UDP datagrams from the kernel buffer via
/// <see cref="NetClient.Receive"/> and writes them into
/// Phase A.3 / Slice H-c1: asynchronous receive loop. It owns exactly
/// one outstanding socket receive from the end of
/// <see cref="EnterWorld"/> until cancellation, then writes pooled raw
/// datagrams into
/// <see cref="_inboundQueue"/> for the render thread to drain in
/// <see cref="Tick"/>. Does NOT decode, reassemble, or dispatch —
/// all of that stays on the render thread to avoid ISAAC/assembler
/// thread-safety issues.
///
/// <para>
/// The 250ms receive timeout is the heartbeat: if no packet arrives
/// within 250ms, the loop re-checks the cancellation token and
/// tries again. On shutdown, <see cref="Dispose"/> cancels the token
/// and joins the thread.
/// Cancellation is wired directly into the socket receive. Idle
/// sessions therefore create neither timeout exceptions nor polling
/// wakeups. On shutdown, <see cref="Dispose"/> cancels and joins the task.
/// </para>
///
/// <para>
@ -989,16 +1037,33 @@ public sealed class WorldSession : IDisposable
/// per-iteration, logged, and the loop keeps polling.
/// </para>
/// </summary>
private void NetReceiveLoop()
private async Task NetReceiveLoopAsync()
{
byte[] receiveBuffer = ArrayPool<byte>.Shared.Rent(
MaxInboundDatagramBytes);
try
{
while (!_netCancel.Token.IsCancellationRequested)
{
byte[]? bytes;
byte[]? queuedBuffer = null;
bool ownershipTransferred = false;
try
{
bytes = _net.Receive(TimeSpan.FromMilliseconds(250), out _);
NetReceiveResult result =
await _net.ReceiveAsync(
receiveBuffer.AsMemory(
0,
MaxInboundDatagramBytes),
_netCancel.Token).ConfigureAwait(false);
queuedBuffer = ArrayPool<byte>.Shared.Rent(
result.Length);
receiveBuffer.AsSpan(0, result.Length).CopyTo(
queuedBuffer);
ownershipTransferred =
_inboundQueue.Writer.TryWrite(
new PooledInboundDatagram(
queuedBuffer,
result.Length));
}
catch (System.Net.Sockets.SocketException ex)
{
@ -1007,21 +1072,55 @@ public sealed class WorldSession : IDisposable
// keep receiving. Does NOT tear down _inboundQueue.
Console.Error.WriteLine(
$"[net] receive error (continuing): {ex.SocketErrorCode} {ex.Message}");
continue;
}
if (bytes is not null)
_inboundQueue.Writer.TryWrite(bytes);
finally
{
if (!ownershipTransferred
&& queuedBuffer is not null)
{
ArrayPool<byte>.Shared.Return(queuedBuffer);
}
}
}
}
catch (OperationCanceledException) { /* graceful shutdown */ }
catch (ObjectDisposedException) { /* NetClient disposed before thread noticed */ }
finally
{
ArrayPool<byte>.Shared.Return(receiveBuffer);
_inboundQueue.Writer.TryComplete();
}
}
private PooledInboundDatagram? ReceiveBlocking(TimeSpan timeout)
{
byte[] buffer = ArrayPool<byte>.Shared.Rent(
MaxInboundDatagramBytes);
try
{
int length = _net.Receive(
buffer.AsSpan(0, MaxInboundDatagramBytes),
timeout,
out _);
if (length < 0)
{
ArrayPool<byte>.Shared.Return(buffer);
return null;
}
return new PooledInboundDatagram(buffer, length);
}
catch
{
ArrayPool<byte>.Shared.Return(buffer);
throw;
}
}
private static void ReturnInboundDatagram(
PooledInboundDatagram datagram) =>
ArrayPool<byte>.Shared.Return(datagram.Buffer);
/// <summary>
/// Blocking single-datagram pump used during Connect/EnterWorld.
/// Returns true if a datagram was processed.
@ -1034,14 +1133,27 @@ public sealed class WorldSession : IDisposable
private bool PumpOnce(out List<uint> opcodesThisCall)
{
opcodesThisCall = new List<uint>();
var bytes = _net.Receive(TimeSpan.FromMilliseconds(250), out _);
if (bytes is null) return false;
ProcessDatagram(bytes, opcodesThisCall);
return true;
PooledInboundDatagram? received =
ReceiveBlocking(TimeSpan.FromMilliseconds(250));
if (received is null)
return false;
PooledInboundDatagram datagram = received.Value;
try
{
ProcessDatagram(
datagram.Memory.Span,
opcodesThisCall);
return true;
}
finally
{
ReturnInboundDatagram(datagram);
}
}
private void ProcessDatagram(
byte[] bytes,
ReadOnlySpan<byte> bytes,
List<uint>? opcodesOut = null,
bool dispatchWorldEvents = true)
{
@ -2120,11 +2232,16 @@ public sealed class WorldSession : IDisposable
Console.Error.WriteLine(
$"[session] transport disconnect failed: {result.TransportDisconnectError.Message}");
// Phase A.3: shut down the background receive thread. Cancel the
// token → the 250ms receive timeout fires → loop exits → join.
// Slice H-c1: cancel the outstanding socket receive directly, join
// its task, then return every queued buffer before transport dispose.
_netCancel.Cancel();
_netReceiveTask?.GetAwaiter().GetResult();
_inboundQueue.Writer.TryComplete();
_netThread?.Join(TimeSpan.FromSeconds(2));
while (_inboundQueue.Reader.TryRead(
out PooledInboundDatagram queued))
{
ReturnInboundDatagram(queued);
}
_netCancel.Dispose();
_net.Dispose();
@ -2224,16 +2341,20 @@ public sealed class WorldSession : IDisposable
WaitForCharacterLogOffConfirmation(
_inboundQueue.Reader,
timeout,
bytes =>
datagram =>
{
ProcessDatagram(bytes, dispatchWorldEvents: false);
ProcessDatagram(
datagram.Memory.Span,
dispatchWorldEvents: false);
return Volatile.Read(ref _characterLogOffConfirmed) != 0;
});
},
ReturnInboundDatagram);
internal static bool WaitForCharacterLogOffConfirmation(
ChannelReader<byte[]> reader,
internal static bool WaitForCharacterLogOffConfirmation<T>(
ChannelReader<T> reader,
TimeSpan timeout,
Func<byte[], bool> processAndCheckConfirmation)
Func<T, bool> processAndCheckConfirmation,
Action<T>? release = null)
{
ArgumentNullException.ThrowIfNull(reader);
ArgumentNullException.ThrowIfNull(processAndCheckConfirmation);
@ -2243,11 +2364,24 @@ public sealed class WorldSession : IDisposable
{
while (!timeoutSource.IsCancellationRequested)
{
while (reader.TryRead(out byte[]? bytes))
while (reader.TryRead(out T? item))
{
if (timeoutSource.IsCancellationRequested)
{
release?.Invoke(item);
return false;
if (processAndCheckConfirmation(bytes))
}
bool confirmed;
try
{
confirmed = processAndCheckConfirmation(item);
}
finally
{
release?.Invoke(item);
}
if (confirmed)
return true;
}