diff --git a/docs/plans/2026-07-25-modern-runtime-slice-h.md b/docs/plans/2026-07-25-modern-runtime-slice-h.md index 8ee3e3f4..7d2ab057 100644 --- a/docs/plans/2026-07-25-modern-runtime-slice-h.md +++ b/docs/plans/2026-07-25-modern-runtime-slice-h.md @@ -135,6 +135,19 @@ behavior is outside this performance-only unit. ## 4. H-c — ordered, allocation-conscious network I/O +H-c is executed as three independently reversible units: + +1. **H-c1 — pooled receive owner — COMPLETE.** One cancellable async socket + receive, a retained full-size edge buffer, right-sized pooled FIFO + datagrams, unchanged blocking handshake cadence, and direct span sends. + Evidence: + [`../research/2026-07-25-slice-h-c1-pooled-receive-owner.md`](../research/2026-07-25-slice-h-c1-pooled-receive-owner.md). +2. **H-c2 — borrowed decode — ACTIVE.** Parse and verify headers, optional + fields, and fragments as borrowed views. Copy only multi-fragment state + that must survive the current datagram. +3. **H-c3 — direct outbound framing — PENDING.** Write packet and fragment + framing into caller storage and remove intermediate payload arrays. + Retail/transport invariants: - one outstanding receive; diff --git a/docs/research/2026-07-25-slice-h-c1-pooled-receive-owner.md b/docs/research/2026-07-25-slice-h-c1-pooled-receive-owner.md new file mode 100644 index 00000000..3ff5b314 --- /dev/null +++ b/docs/research/2026-07-25-slice-h-c1-pooled-receive-owner.md @@ -0,0 +1,70 @@ +# Slice H-c1 — cancellable pooled receive owner + +## Result + +The in-world transport now has one asynchronous socket receive owner: + +```text +one retained full-size socket buffer + -> ReceiveFromAsync(cancellation token) + -> copy actual datagram length into a right-sized ArrayPool bucket + -> FIFO Channel + -> render-thread decode/dispatch + -> return bucket exactly once +``` + +The synchronous `PumpOnce` handshake path keeps its existing 250 ms pacing, +but receives into a temporarily rented caller buffer rather than accepting a +fresh `UdpClient.Receive` array. The normal in-world path no longer wakes four +times per idle second and no longer uses `SocketException(TimedOut)` as its +heartbeat. + +`NetClient.Send` also consumes its caller's `ReadOnlySpan` directly +through `Socket.SendTo`; the transport-only `datagram.ToArray()` copy is gone. +Higher-level outbound framing allocations are intentionally left to H-c3. + +## Retail and protocol boundary + +Behavioral invariants remain unchanged: + +- one outstanding receive; +- kernel arrival order is FIFO channel order; +- decode, ISAAC, fragment assembly, ACK decisions, and event dispatch remain + on the existing render/update thread; +- ACK generation remains inside `ProcessDatagram`, immediately after decode + acceptance; +- `LinkStatusHolder::OnHeartbeat` (`0x004113D0`) last-heard placement remains + immediately after successful decode; +- blocking `PumpOnce` handshake cadence is unchanged. + +The initial design review caught that 484 bytes is a fragment limit, not a +safe UDP receive limit. Named-retail `CNetLayerPacket` (type 3705 in +`acclient.h`) contains a separate `ProtoHeader` plus `m_Data[65484]`. +Accordingly the socket edge retains a full UDP buffer. Only the received byte +count crosses the queue boundary, preventing small queued packets from each +pinning a 64 KiB bucket. + +Holtburger's receive/finalize path remains the cross-reference for +ACK-per-accepted-packet ordering. The transport ownership mechanism is a +modern .NET adaptation; it does not change AC wire behavior. + +## Deterministic evidence + +- real loopback `ReceiveAsync` writes a 1,500-byte datagram into caller-owned + memory; +- cancelling an idle receive terminates the one outstanding operation; +- a scripted recoverable `SocketException` does not kill the receive owner; +- three accepted packets retain arrival order and produce ACKs in the exact + sequence `41, 42, 43`; +- blocking handshake timeout behavior and cached socket-timeout tests remain + unchanged; +- graceful confirmation drain releases every consumed pooled datagram, and + shutdown cancels/joins the receive task before returning unread buffers. + +## Remaining H-c work + +H-c2 removes `Packet`, `BodyBytes`, `PacketHeaderOptional`, `List`, and +single-fragment payload copies from the production decode path using borrowed +memory with an explicit lifetime boundary. H-c3 writes fragment and packet +framing directly into caller storage. + diff --git a/src/AcDream.Core.Net/NetClient.cs b/src/AcDream.Core.Net/NetClient.cs index d3ec4a93..99780053 100644 --- a/src/AcDream.Core.Net/NetClient.cs +++ b/src/AcDream.Core.Net/NetClient.cs @@ -4,22 +4,20 @@ using System.Net.Sockets; namespace AcDream.Core.Net; /// -/// Minimum-viable UDP transport for acdream. Wraps a -/// 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. /// /// -/// Not yet provided (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 without +/// materializing a transport-only array copy. /// /// public sealed class NetClient : IDisposable { private readonly UdpClient _udp; private readonly IPEndPoint _remote; + private readonly IPEndPoint _anyRemote = + new(IPAddress.Any, 0); /// Last value applied to the /// underlying socket, so repeated calls with the @@ -57,7 +55,10 @@ public sealed class NetClient : IDisposable /// public void Send(ReadOnlySpan datagram) { - _udp.Send(datagram.ToArray(), datagram.Length, _remote); + _udp.Client.SendTo( + datagram, + SocketFlags.None, + _remote); } /// @@ -68,7 +69,63 @@ public sealed class NetClient : IDisposable /// public void Send(IPEndPoint remote, ReadOnlySpan datagram) { - _udp.Send(datagram.ToArray(), datagram.Length, remote); + _udp.Client.SendTo( + datagram, + SocketFlags.None, + remote); + } + + /// + /// Receive into caller-owned storage. Returns the received byte count, + /// or -1 when expires. + /// + public int Receive( + Span 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; + } + } + + /// + /// One cancellable asynchronous receive into caller-owned storage. + /// The returned memory remains owned by the caller. + /// + public async ValueTask ReceiveAsync( + Memory destination, + CancellationToken cancellationToken) + { + SocketReceiveFromResult result = + await _udp.Client.ReceiveFromAsync( + destination, + SocketFlags.None, + _anyRemote, + cancellationToken).ConfigureAwait(false); + return new NetReceiveResult( + result.ReceivedBytes, + (IPEndPoint)result.RemoteEndPoint); } /// @@ -79,7 +136,7 @@ public sealed class NetClient : IDisposable /// 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(); } + +/// +/// Result of receiving one UDP datagram into caller-owned memory. +/// +public readonly record struct NetReceiveResult( + int Length, + IPEndPoint RemoteEndPoint); diff --git a/src/AcDream.Core.Net/WorldSession.cs b/src/AcDream.Core.Net/WorldSession.cs index c4599165..d1c551dd 100644 --- a/src/AcDream.Core.Net/WorldSession.cs +++ b/src/AcDream.Core.Net/WorldSession.cs @@ -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 datagram); void Send(IPEndPoint remote, ReadOnlySpan datagram); - byte[]? Receive(TimeSpan timeout, out IPEndPoint? from); + int Receive( + Span destination, + TimeSpan timeout, + out IPEndPoint? from); + ValueTask ReceiveAsync( + Memory 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 datagram) => _client.Send(endpoint, datagram); - public byte[]? Receive(TimeSpan timeout, out IPEndPoint? from) => - _client.Receive(timeout, out from); + public int Receive( + Span destination, + TimeSpan timeout, + out IPEndPoint? from) => + _client.Receive(destination, timeout, out from); + + public ValueTask ReceiveAsync( + Memory 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 _inboundQueue = - Channel.CreateUnbounded( + // 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 _inboundQueue = + Channel.CreateUnbounded( 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 Memory => + Buffer.AsMemory(0, Length); + } + /// /// 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; /// - /// Phase A.3: background receive loop. Runs on a dedicated daemon - /// thread started at the end of . Continuously - /// pulls raw UDP datagrams from the kernel buffer via - /// and writes them into + /// Phase A.3 / Slice H-c1: asynchronous receive loop. It owns exactly + /// one outstanding socket receive from the end of + /// until cancellation, then writes pooled raw + /// datagrams into /// for the render thread to drain in /// . Does NOT decode, reassemble, or dispatch — /// all of that stays on the render thread to avoid ISAAC/assembler /// thread-safety issues. /// /// - /// 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, 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, cancels and joins the task. /// /// /// @@ -989,16 +1037,33 @@ public sealed class WorldSession : IDisposable /// per-iteration, logged, and the loop keeps polling. /// /// - private void NetReceiveLoop() + private async Task NetReceiveLoopAsync() { + byte[] receiveBuffer = ArrayPool.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.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.Shared.Return(queuedBuffer); + } + } } } catch (OperationCanceledException) { /* graceful shutdown */ } catch (ObjectDisposedException) { /* NetClient disposed before thread noticed */ } finally { + ArrayPool.Shared.Return(receiveBuffer); _inboundQueue.Writer.TryComplete(); } } + private PooledInboundDatagram? ReceiveBlocking(TimeSpan timeout) + { + byte[] buffer = ArrayPool.Shared.Rent( + MaxInboundDatagramBytes); + try + { + int length = _net.Receive( + buffer.AsSpan(0, MaxInboundDatagramBytes), + timeout, + out _); + if (length < 0) + { + ArrayPool.Shared.Return(buffer); + return null; + } + + return new PooledInboundDatagram(buffer, length); + } + catch + { + ArrayPool.Shared.Return(buffer); + throw; + } + } + + private static void ReturnInboundDatagram( + PooledInboundDatagram datagram) => + ArrayPool.Shared.Return(datagram.Buffer); + /// /// 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 opcodesThisCall) { opcodesThisCall = new List(); - 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 bytes, List? 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 reader, + internal static bool WaitForCharacterLogOffConfirmation( + ChannelReader reader, TimeSpan timeout, - Func processAndCheckConfirmation) + Func processAndCheckConfirmation, + Action? 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; } diff --git a/tests/AcDream.App.Tests/Net/LiveSessionControllerTests.cs b/tests/AcDream.App.Tests/Net/LiveSessionControllerTests.cs index 91883825..17c822e4 100644 --- a/tests/AcDream.App.Tests/Net/LiveSessionControllerTests.cs +++ b/tests/AcDream.App.Tests/Net/LiveSessionControllerTests.cs @@ -15,12 +15,20 @@ public sealed class LiveSessionControllerTests public void Send(IPEndPoint remote, ReadOnlySpan datagram) { } - public byte[]? Receive(TimeSpan timeout, out IPEndPoint? from) + public int Receive( + Span destination, + TimeSpan timeout, + out IPEndPoint? from) { from = null; - return null; + return -1; } + public ValueTask ReceiveAsync( + Memory destination, + CancellationToken cancellationToken) => + throw new OperationCanceledException(cancellationToken); + public void Dispose() { } } diff --git a/tests/AcDream.App.Tests/Net/LiveSessionHostTests.cs b/tests/AcDream.App.Tests/Net/LiveSessionHostTests.cs index 7f8729c0..747a5ec6 100644 --- a/tests/AcDream.App.Tests/Net/LiveSessionHostTests.cs +++ b/tests/AcDream.App.Tests/Net/LiveSessionHostTests.cs @@ -381,12 +381,20 @@ public sealed class LiveSessionHostTests public void Send(ReadOnlySpan datagram) { } public void Send(IPEndPoint remote, ReadOnlySpan datagram) { } - public byte[]? Receive(TimeSpan timeout, out IPEndPoint? from) + public int Receive( + Span destination, + TimeSpan timeout, + out IPEndPoint? from) { from = null; - return null; + return -1; } + public ValueTask ReceiveAsync( + Memory destination, + CancellationToken cancellationToken) => + throw new OperationCanceledException(cancellationToken); + public void Dispose() { } } } diff --git a/tests/AcDream.App.Tests/Net/LiveSessionLifecycleHostTests.cs b/tests/AcDream.App.Tests/Net/LiveSessionLifecycleHostTests.cs index a0ed5b26..2d20f49a 100644 --- a/tests/AcDream.App.Tests/Net/LiveSessionLifecycleHostTests.cs +++ b/tests/AcDream.App.Tests/Net/LiveSessionLifecycleHostTests.cs @@ -99,12 +99,20 @@ public sealed class LiveSessionLifecycleHostTests public void Send(IPEndPoint remote, ReadOnlySpan datagram) { } - public byte[]? Receive(TimeSpan timeout, out IPEndPoint? from) + public int Receive( + Span destination, + TimeSpan timeout, + out IPEndPoint? from) { from = null; - return null; + return -1; } + public ValueTask ReceiveAsync( + Memory destination, + CancellationToken cancellationToken) => + throw new OperationCanceledException(cancellationToken); + public void Dispose() { } } } diff --git a/tests/AcDream.Core.Net.Tests/NetClientTests.cs b/tests/AcDream.Core.Net.Tests/NetClientTests.cs index 23110ac4..7e6841d7 100644 --- a/tests/AcDream.Core.Net.Tests/NetClientTests.cs +++ b/tests/AcDream.Core.Net.Tests/NetClientTests.cs @@ -96,4 +96,50 @@ public class NetClientTests Assert.Null(result); Assert.Null(from); } + + [Fact] + public async Task ReceiveAsync_LoopbackWritesIntoCallerOwnedMemory() + { + using var serverSide = + new NetClient(new IPEndPoint(IPAddress.Loopback, 0)); + using var clientSide = + new NetClient(new IPEndPoint( + IPAddress.Loopback, + serverSide.LocalEndPoint.Port)); + byte[] outbound = Enumerable.Range(0, 1_500) + .Select(static value => (byte)value) + .ToArray(); + byte[] destination = new byte[2_048]; + using var timeout = + new CancellationTokenSource(TimeSpan.FromSeconds(2)); + + clientSide.Send(outbound); + NetReceiveResult result = await serverSide.ReceiveAsync( + destination, + timeout.Token); + + Assert.Equal(outbound.Length, result.Length); + Assert.Equal( + outbound, + destination.AsSpan(0, result.Length).ToArray()); + Assert.Equal( + clientSide.LocalEndPoint.Port, + result.RemoteEndPoint.Port); + } + + [Fact] + public async Task ReceiveAsync_IdleCancellationStopsOutstandingReceive() + { + using var client = + new NetClient(new IPEndPoint(IPAddress.Loopback, 1)); + byte[] destination = new byte[32]; + using var cancellation = new CancellationTokenSource(); + + ValueTask receive = + client.ReceiveAsync(destination, cancellation.Token); + cancellation.Cancel(); + + await Assert.ThrowsAnyAsync( + async () => await receive); + } } diff --git a/tests/AcDream.Core.Net.Tests/WorldSessionConstructionTests.cs b/tests/AcDream.Core.Net.Tests/WorldSessionConstructionTests.cs index 0718a742..6a6bdf9d 100644 --- a/tests/AcDream.Core.Net.Tests/WorldSessionConstructionTests.cs +++ b/tests/AcDream.Core.Net.Tests/WorldSessionConstructionTests.cs @@ -30,12 +30,20 @@ public sealed class WorldSessionConstructionTests public void Send(IPEndPoint remote, ReadOnlySpan datagram) { } - public byte[]? Receive(TimeSpan timeout, out IPEndPoint? from) + public int Receive( + Span destination, + TimeSpan timeout, + out IPEndPoint? from) { from = null; - return null; + return -1; } + public ValueTask ReceiveAsync( + Memory destination, + CancellationToken cancellationToken) => + throw new OperationCanceledException(cancellationToken); + public void Dispose() { } } } diff --git a/tests/AcDream.Core.Net.Tests/WorldSessionNetReceiveLoopResilienceTests.cs b/tests/AcDream.Core.Net.Tests/WorldSessionNetReceiveLoopResilienceTests.cs index ff0ca527..2529f77c 100644 --- a/tests/AcDream.Core.Net.Tests/WorldSessionNetReceiveLoopResilienceTests.cs +++ b/tests/AcDream.Core.Net.Tests/WorldSessionNetReceiveLoopResilienceTests.cs @@ -1,6 +1,7 @@ using System.Net; using System.Net.Sockets; using System.Reflection; +using AcDream.Core.Net.Packets; namespace AcDream.Core.Net.Tests; @@ -19,7 +20,7 @@ namespace AcDream.Core.Net.Tests; public sealed class WorldSessionNetReceiveLoopResilienceTests { [Fact] - public void NetReceiveLoop_NonTimeoutSocketException_LogsAndKeepsReceiving() + public async Task NetReceiveLoop_NonTimeoutSocketException_LogsAndKeepsReceiving() { var transport = new ScriptedTransport(); var session = new WorldSession( @@ -27,7 +28,8 @@ public sealed class WorldSessionNetReceiveLoopResilienceTests transport); MethodInfo loopMethod = typeof(WorldSession).GetMethod( - "NetReceiveLoop", BindingFlags.NonPublic | BindingFlags.Instance)!; + "NetReceiveLoopAsync", + BindingFlags.NonPublic | BindingFlags.Instance)!; // Deliberately does NOT redirect Console.Error: xUnit runs test // classes in this assembly in parallel by default, and Console.Error @@ -36,12 +38,7 @@ public sealed class WorldSessionNetReceiveLoopResilienceTests // swallow its output or restore the wrong writer. The behavioral // assertions below (call count + queue drain) fully cover the fix // without needing to observe the log line's exact text. - var thread = new Thread(() => loopMethod.Invoke(session, null)) - { - IsBackground = true, - Name = "test.net-receive-loop", - }; - thread.Start(); + var task = (Task)loopMethod.Invoke(session, null)!; // The scripted transport throws SocketException on call 1, returns // one datagram on call 2, then throws ObjectDisposedException on @@ -50,7 +47,8 @@ public sealed class WorldSessionNetReceiveLoopResilienceTests // would exit after call 1 and this thread would never terminate on // its own via the scripted ObjectDisposedException (call 3 would // never be reached). - bool exited = thread.Join(TimeSpan.FromSeconds(5)); + await task.WaitAsync(TimeSpan.FromSeconds(5)); + bool exited = task.IsCompletedSuccessfully; Assert.True(exited, "NetReceiveLoop did not exit after the scripted ObjectDisposedException — " + "the non-timeout SocketException likely killed the loop before it reached call 3."); @@ -66,6 +64,48 @@ public sealed class WorldSessionNetReceiveLoopResilienceTests Assert.Equal(1, processed); } + [Fact] + public async Task NetReceiveLoopAsync_PreservesArrivalAndAckOrder() + { + var transport = new OrderedDatagramTransport( + BuildPacket(sequence: 41), + BuildPacket(sequence: 42), + BuildPacket(sequence: 43)); + var session = new WorldSession( + new IPEndPoint(IPAddress.Loopback, 9000), + transport); + MethodInfo loopMethod = typeof(WorldSession).GetMethod( + "NetReceiveLoopAsync", + BindingFlags.NonPublic | BindingFlags.Instance)!; + + var task = (Task)loopMethod.Invoke(session, null)!; + await task.WaitAsync(TimeSpan.FromSeconds(5)); + + Assert.Equal(3, session.Tick()); + uint[] acked = transport.Sent + .Select(static bytes => + PacketCodec.TryDecode( + bytes, + inboundIsaac: null)) + .Select(static decoded => + { + Assert.True(decoded.IsOk, decoded.Error.ToString()); + return decoded.Packet!.Optional.AckSequence; + }) + .ToArray(); + Assert.Equal([41u, 42u, 43u], acked); + } + + private static byte[] BuildPacket(uint sequence) => + PacketCodec.Encode( + new PacketHeader + { + Sequence = sequence, + Flags = PacketHeaderFlags.None, + }, + ReadOnlySpan.Empty, + outboundIsaac: null); + private sealed class ScriptedTransport : IWorldSessionTransport { private int _calls; @@ -76,10 +116,20 @@ public sealed class WorldSessionNetReceiveLoopResilienceTests public void Send(IPEndPoint remote, ReadOnlySpan datagram) { } - public byte[]? Receive(TimeSpan timeout, out IPEndPoint? from) + public int Receive( + Span destination, + TimeSpan timeout, + out IPEndPoint? from) + { + from = null; + return -1; + } + + public ValueTask ReceiveAsync( + Memory destination, + CancellationToken cancellationToken) { int n = Interlocked.Increment(ref _calls); - from = null; switch (n) { case 1: @@ -87,8 +137,14 @@ public sealed class WorldSessionNetReceiveLoopResilienceTests // WSAECONNRESET off a stale peer's ICMP port-unreachable. throw new SocketException((int)SocketError.ConnectionReset); case 2: - from = new IPEndPoint(IPAddress.Loopback, 9000); - return new byte[] { 0xAA, 0xBB, 0xCC, 0xDD }; + new byte[] { 0xAA, 0xBB, 0xCC, 0xDD } + .CopyTo(destination); + return ValueTask.FromResult( + new NetReceiveResult( + 4, + new IPEndPoint( + IPAddress.Loopback, + 9000))); default: // Simulates NetClient having been disposed out from // under the loop during shutdown — the pre-existing, @@ -99,4 +155,48 @@ public sealed class WorldSessionNetReceiveLoopResilienceTests public void Dispose() { } } + + private sealed class OrderedDatagramTransport( + params byte[][] datagrams) : IWorldSessionTransport + { + private int _next; + + public List Sent { get; } = []; + + public void Send(ReadOnlySpan datagram) => + Sent.Add(datagram.ToArray()); + + public void Send( + IPEndPoint remote, + ReadOnlySpan datagram) => + Sent.Add(datagram.ToArray()); + + public int Receive( + Span destination, + TimeSpan timeout, + out IPEndPoint? from) + { + from = null; + return -1; + } + + public ValueTask ReceiveAsync( + Memory destination, + CancellationToken cancellationToken) + { + int index = Interlocked.Increment(ref _next) - 1; + if (index >= datagrams.Length) + throw new ObjectDisposedException( + nameof(OrderedDatagramTransport)); + + byte[] source = datagrams[index]; + source.CopyTo(destination); + return ValueTask.FromResult( + new NetReceiveResult( + source.Length, + new IPEndPoint(IPAddress.Loopback, 9000))); + } + + public void Dispose() { } + } }