From 55456fe55a4c428e738c95d4480b0f316b7805ae Mon Sep 17 00:00:00 2001 From: sunsi Date: Sat, 15 Aug 2026 13:25:45 +0800 Subject: [PATCH 1/6] test: cover TimedBatch deadline-wait retention, extension, chunk re-arm, and stop (issue-204) --- .../Runtime/SendPumpTests.cs | 212 ++++++++++++++++++ 1 file changed, 212 insertions(+) diff --git a/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs b/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs index 7017b10a..e1e1d2a2 100644 --- a/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs +++ b/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs @@ -1,3 +1,4 @@ +using System.Collections.Generic; using System.Diagnostics; using System.IO.Pipelines; using System.Threading; @@ -89,6 +90,186 @@ public async Task TimedBatchShouldFlushAtExactProviderLatencyAndReturnItsOwner() } } + [Test] + public async Task TimedBatchShouldDeliverFrameSentAfterDeadlineFlushThroughRetainedRead() + { + var clock = new ManualTimeProvider(); + var maxLatency = TimeSpan.FromMilliseconds(100); + var input = new Pipe(); + var output = new Pipe(); + using var context = new SharpLinkRuntimeContextBuilder() + .UseTimeProvider(clock) + .Build(includeGeneratedAssemblyCatalog: false); + var session = RpcSessionTestFixture.CreateSessionOverTestTransport( + "timed-batch-retained-read", + input.Reader, + output.Writer, + RpcSessionTestFixture.ClientOptions( + context, + new RpcSessionFlushOptions(1024 * 1024, maxLatency))); + try + { + var first = CreateFrame(session, 32, requestId: 1); + session.SendPacket(first); + await WaitUntilAsync(() => clock.ActiveTimerCount > 0); + clock.Advance(maxLatency); + + await ConsumeAvailableAsync(output.Reader); + await WaitUntilAsync(() => session.QueuedSendBytes == 0); + EnsureReturned(first, "the deadline flush must return the first frame owner"); + Ensure(clock.ActiveTimerCount == 0, + "the deadline timer must be disposed after the timed-out flush"); + + // The pump must have kept the unconsumed pending read and re-observed it: the next + // frame wakes the pump through that retained registration. Dropping the read would + // leave the fresh registration waiting behind the stale one and time this out. + var second = CreateFrame(session, 32, requestId: 2); + await session.SendPacketAndFlushAsync(second).AsTask().WaitAsync(TimeSpan.FromSeconds(2)); + EnsureReturned(second, + "a frame after a deadline flush must be delivered through the retained pending read"); + await ConsumeAvailableAsync(output.Reader); + } + finally + { + await session.DisposeAsync(); + await output.Reader.CompleteAsync(); + await input.Writer.CompleteAsync(); + } + } + + [Test] + public async Task TimedBatchShouldExtendBatchForFrameArrivingBeforeDeadline() + { + var clock = new ManualTimeProvider(); + var maxLatency = TimeSpan.FromMilliseconds(100); + var provider = new TimerArmRecordingTimeProvider(clock); + var input = new Pipe(); + var output = new Pipe(); + using var context = new SharpLinkRuntimeContextBuilder() + .UseTimeProvider(provider) + .Build(includeGeneratedAssemblyCatalog: false); + var session = RpcSessionTestFixture.CreateSessionOverTestTransport( + "timed-batch-extension", + input.Reader, + output.Writer, + RpcSessionTestFixture.ClientOptions( + context, + new RpcSessionFlushOptions(1024 * 1024, maxLatency))); + var first = CreateFrame(session, 32, requestId: 1); + var second = CreateFrame(session, 32, requestId: 2); + try + { + session.SendPacket(first); + await WaitUntilAsync(() => provider.WasArmed(maxLatency)); + clock.Advance(TimeSpan.FromMilliseconds(50)); + + session.SendPacket(second); + // The arriving frame wins the deadline race. The pump then re-arms one timer for + // the remaining latency, which is the durable observation point (the transient + // dispose-then-rearm handoff is too short to poll for). + await WaitUntilAsync(() => provider.WasArmed(TimeSpan.FromMilliseconds(50))); + + clock.Advance(TimeSpan.FromMilliseconds(50)); + var read = await output.Reader.ReadAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(2)); + var expectedBytes = 2 * (ProtocolV2Constants.HeaderBytes + 32); + Ensure(read.Buffer.Length >= expectedBytes, + "both frames must share one flush at the first frame's deadline"); + output.Reader.AdvanceTo(read.Buffer.End); + + await WaitUntilAsync(() => session.QueuedSendBytes == 0); + EnsureReturned(first, "the extended batch must return the first frame owner"); + EnsureReturned(second, "the extended batch must return the second frame owner"); + } + finally + { + await session.DisposeAsync(); + await output.Reader.CompleteAsync(); + await input.Writer.CompleteAsync(); + } + } + + [Test] + public async Task TimedBatchShouldRearmAcrossMaximumTimerDelayChunks() + { + var clock = new ManualTimeProvider(); + var chunk = TimeSpan.FromMilliseconds(int.MaxValue); + var maxLatency = chunk + TimeSpan.FromMilliseconds(1); + var input = new Pipe(); + var output = new Pipe(); + using var context = new SharpLinkRuntimeContextBuilder() + .UseTimeProvider(clock) + .Build(includeGeneratedAssemblyCatalog: false); + var session = RpcSessionTestFixture.CreateSessionOverTestTransport( + "timed-batch-chunk-rearm", + input.Reader, + output.Writer, + RpcSessionTestFixture.ClientOptions( + context, + new RpcSessionFlushOptions(1024 * 1024, maxLatency))); + var frame = CreateFrame(session, 32, requestId: 1); + try + { + session.SendPacket(frame); + await WaitUntilAsync(() => clock.ActiveTimerCount > 0); + Ensure(clock.EarliestTimerTimestamp == clock.GetTimestamp() + chunk.Ticks, + "a deadline beyond the maximum timer delay must be armed as one full chunk"); + + clock.Advance(chunk); + Ensure(session.QueuedSendBytes > 0, + "an expiring timer-delay chunk must not flush a batch whose deadline is still ahead"); + await WaitUntilAsync(() => + clock.EarliestTimerTimestamp == clock.GetTimestamp() + TimeSpan.FromMilliseconds(1).Ticks); + + clock.Advance(TimeSpan.FromMilliseconds(1)); + await ConsumeAvailableAsync(output.Reader); + await WaitUntilAsync(() => session.QueuedSendBytes == 0); + EnsureReturned(frame, "the re-armed deadline flush must return the frame owner"); + Ensure(clock.ActiveTimerCount == 0, "the final deadline timer must be disposed"); + } + finally + { + await session.DisposeAsync(); + await output.Reader.CompleteAsync(); + await input.Writer.CompleteAsync(); + } + } + + [Test] + public async Task TimedBatchDeadlineWaitShouldExitWhenSessionIsDisposed() + { + var clock = new ManualTimeProvider(); + var maxLatency = TimeSpan.FromMilliseconds(100); + var input = new Pipe(); + var output = new Pipe(); + using var context = new SharpLinkRuntimeContextBuilder() + .UseTimeProvider(clock) + .Build(includeGeneratedAssemblyCatalog: false); + var session = RpcSessionTestFixture.CreateSessionOverTestTransport( + "timed-batch-dispose-during-wait", + input.Reader, + output.Writer, + RpcSessionTestFixture.ClientOptions( + context, + new RpcSessionFlushOptions(1024 * 1024, maxLatency))); + var frame = CreateFrame(session, 32, requestId: 1); + try + { + session.SendPacket(frame); + await WaitUntilAsync(() => clock.ActiveTimerCount > 0); + + await session.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(2)); + EnsureReturned(frame, "dispose during the deadline wait must return the frame owner"); + Ensure(session.QueuedSendBytes == 0, "dispose during the deadline wait must release queued bytes"); + Ensure(clock.ActiveTimerCount == 0, + "the deadline timer must be disposed when the pump stops mid-wait"); + } + finally + { + await output.Reader.CompleteAsync(); + await input.Writer.CompleteAsync(); + } + } + [Test] public async Task FullByteQueueShouldFailFastWithoutClosingHealthySession() { @@ -333,6 +514,37 @@ public override ITimer CreateTimer( } } + private sealed class TimerArmRecordingTimeProvider(ManualTimeProvider inner) : TimeProvider + { + private readonly Lock _gate = new(); + private readonly List _armedDueTimes = []; + + public override long TimestampFrequency => inner.TimestampFrequency; + + public override TimeZoneInfo LocalTimeZone => inner.LocalTimeZone; + + public override DateTimeOffset GetUtcNow() => inner.GetUtcNow(); + + public override long GetTimestamp() => inner.GetTimestamp(); + + public override ITimer CreateTimer( + TimerCallback callback, + object? state, + TimeSpan dueTime, + TimeSpan period) + { + lock (_gate) + _armedDueTimes.Add(dueTime); + return inner.CreateTimer(callback, state, dueTime, period); + } + + internal bool WasArmed(TimeSpan dueTime) + { + lock (_gate) + return _armedDueTimes.Contains(dueTime); + } + } + private static void Ensure(bool condition, string message) { if (!condition) From 2b1f7f582d799247c52c5668ca1766f5538fa4f3 Mon Sep 17 00:00:00 2001 From: sunsi Date: Sat, 15 Aug 2026 13:25:49 +0800 Subject: [PATCH 2/6] perf: race TimedBatch deadline wait with a pooled read and per-wait timer, dropping AsTask/WhenAny allocations (issue-204) --- src/SharpLink.Runtime/DeadlineReadRace.cs | 150 +++++++++++++++++++ src/SharpLink.Runtime/RpcSession.SendPump.cs | 34 +++-- 2 files changed, 169 insertions(+), 15 deletions(-) create mode 100644 src/SharpLink.Runtime/DeadlineReadRace.cs diff --git a/src/SharpLink.Runtime/DeadlineReadRace.cs b/src/SharpLink.Runtime/DeadlineReadRace.cs new file mode 100644 index 00000000..a4894da6 --- /dev/null +++ b/src/SharpLink.Runtime/DeadlineReadRace.cs @@ -0,0 +1,150 @@ +using System.Runtime.CompilerServices; +using System.Threading.Tasks.Sources; + +namespace SharpLink.Runtime; + +/// +/// Races a pending channel read against a deadline timer without +/// , +/// , or a per-pump +/// . A single +/// instance is reused for every deadline wait of one send pump, so only the arm itself +/// allocates: one continuation closure +/// per wait plus one from the owner's . +/// +/// +/// +/// When the timer wins, the read is deliberately left unconsumed: its +/// stays registered on the channel and the owner is expected to retain and re-observe it later +/// (pending-read retention). When the read wins, the timer is disposed and the result is +/// surfaced through the returned . +/// +/// +/// The instance is single-flight: an arm must be fully awaited before the next arm. The owner +/// (a single-threaded send pump) satisfies this by construction. A read that completes while an +/// arm is being set up is still handled correctly: the continuation registered by +/// runs inline for completed +/// tasks, which is why the timer is created before the continuation is registered, and why each +/// arm captures its own read so a late continuation from a previous arm can never act on the +/// current arm's state (the identity check makes stale completions no-ops). +/// +/// +internal sealed class DeadlineReadRace : IValueTaskSource, IDisposable +{ + internal enum RaceOutcome + { + Pending, + DataAvailable, + ReadClosed, + TimedOut, + } + + private static readonly TimerCallback s_timerCallback = + static state => ((DeadlineReadRace)state!).OnTimerFired(); + + private readonly TimeProvider _timeProvider; + private ManualResetValueTaskSourceCore _core; + private Task? _read; + private ITimer? _timer; + private RaceOutcome _outcome; + private int _readAbandoned; + + internal DeadlineReadRace(TimeProvider timeProvider) + { + _timeProvider = timeProvider ?? throw new ArgumentNullException(nameof(timeProvider)); + _core = new ManualResetValueTaskSourceCore + { + RunContinuationsAsynchronously = true, + }; + } + + /// + /// Gets how the most recent wait resolved. Only meaningful after the value task returned by + /// has been awaited to completion. + /// + internal RaceOutcome Outcome => + (RaceOutcome)Volatile.Read(ref Unsafe.As(ref _outcome)); + + /// + /// Waits until completes or expires. + /// The returned value task completes with the read's result when the read wins, and with + /// false when the timer wins; a faulted or canceled read is propagated. The read is + /// consumed exactly once by the winner and stays available to the owner otherwise. + /// + internal ValueTask WaitForReadOrTimeout(Task read, TimeSpan timeout) + { + ArgumentNullException.ThrowIfNull(read); + if (read.IsCompleted) + { + // Data arrived (or the channel closed) between the caller's completedness check and + // this arm: surface the already-available outcome without starting a race. + Volatile.Write(ref Unsafe.As(ref _outcome), + (int)(read.IsCompletedSuccessfully && read.Result + ? RaceOutcome.DataAvailable + : RaceOutcome.ReadClosed)); + return new ValueTask(read); + } + + _read = read; + Volatile.Write(ref Unsafe.As(ref _outcome), (int)RaceOutcome.Pending); + Volatile.Write(ref _readAbandoned, 0); + _core.Reset(); + + // The timer must be armed before the read continuation is registered: a read that + // completes in this window invokes the continuation inline, and the read-win path + // disposes the timer it expects to exist. + _timer = _timeProvider.CreateTimer(s_timerCallback, this, timeout, Timeout.InfiniteTimeSpan); + + // Each arm registers a fresh closure capturing its own read. A closure from an earlier + // arm that fires late must not be able to act on this arm's state: the identity check + // against the current read makes stale completions no-ops. + read.GetAwaiter().UnsafeOnCompleted(() => OnReadCompleted(read)); + return new ValueTask(this, _core.Version); + } + + private void OnReadCompleted(Task read) + { + if (!ReferenceEquals(read, _read)) + return; // Stale completion from a previous arm: never touch the current cycle's state. + + if (Interlocked.Exchange(ref _readAbandoned, 1) != 0) + return; // The timer won first: the read stays unconsumed for later reuse. + + _timer!.Dispose(); + if (read.IsCompletedSuccessfully) + { + Volatile.Write(ref Unsafe.As(ref _outcome), + (int)(read.Result ? RaceOutcome.DataAvailable : RaceOutcome.ReadClosed)); + _core.SetResult(read.Result); + } + else + { + Volatile.Write(ref Unsafe.As(ref _outcome), (int)RaceOutcome.ReadClosed); + _core.SetException( + (Exception?)read.Exception ?? new InvalidOperationException("pending read failed.")); + } + } + + private void OnTimerFired() + { + if (Interlocked.Exchange(ref _readAbandoned, 1) != 0) + return; // The read completed first and disposed the timer. + + _timer!.Dispose(); + Volatile.Write(ref Unsafe.As(ref _outcome), (int)RaceOutcome.TimedOut); + _core.SetResult(false); + } + + bool IValueTaskSource.GetResult(short token) => _core.GetResult(token); + + ValueTaskSourceStatus IValueTaskSource.GetStatus(short token) => _core.GetStatus(token); + + void IValueTaskSource.OnCompleted( + Action continuation, + object? state, + short token, + ValueTaskSourceOnCompletedFlags flags) => + _core.OnCompleted(continuation, state, token, flags); + + public void Dispose() => _timer?.Dispose(); +} diff --git a/src/SharpLink.Runtime/RpcSession.SendPump.cs b/src/SharpLink.Runtime/RpcSession.SendPump.cs index bf7416e2..8319801f 100644 --- a/src/SharpLink.Runtime/RpcSession.SendPump.cs +++ b/src/SharpLink.Runtime/RpcSession.SendPump.cs @@ -23,10 +23,10 @@ private enum FlushMode private readonly Action _onTransportFaulted; private readonly Channel _queue; private readonly Lock _admissionGate = new(); + private readonly DeadlineReadRace _deadlineRace; private readonly Task _pumpTask; private TaskCompletionSource? _capacityChanged; private Task? _pendingReadWait; - private CancellationTokenSource? _delayCancellation; private long _queuedBytes; private int _stopped; private int _faulted; @@ -86,6 +86,7 @@ public SendPump( SingleWriter = false, AllowSynchronousContinuations = false }); + _deadlineRace = new DeadlineReadRace(_timeProvider); _pumpTask = RunAsync(); } @@ -202,7 +203,7 @@ await WaitForMoreUntilDeadlineAsync(batchDeadline).ConfigureAwait(false)) } finally { - _delayCancellation?.Dispose(); + _deadlineRace.Dispose(); ReleaseBatch(pending, terminalException); DrainQueuedFrames(terminalException); PulseCapacityWaiters(); @@ -230,7 +231,7 @@ private async ValueTask FlushAndReleaseAsync(List pending) private async ValueTask WaitForMoreUntilDeadlineAsync(long batchDeadline) { - var waitToRead = _queue.Reader.WaitToReadAsync(_sessionCancellation); + var waitToRead = _queue.Reader.WaitToReadAsync(CancellationToken.None); if (waitToRead.IsCompletedSuccessfully) return waitToRead.Result; @@ -246,22 +247,25 @@ private async ValueTask WaitForMoreUntilDeadlineAsync(long batchDeadline) return false; var delay = remaining > MaximumTimerDelay ? MaximumTimerDelay : remaining; - var delayCancellation = _delayCancellation; - if (delayCancellation is null || delayCancellation.IsCancellationRequested) - { - delayCancellation = new CancellationTokenSource(); - _delayCancellation = delayCancellation; - } - var delayTask = Task.Delay(delay, _timeProvider, delayCancellation.Token); - if (await Task.WhenAny(pendingRead, delayTask).ConfigureAwait(false) == pendingRead) + if (await _deadlineRace.WaitForReadOrTimeout(pendingRead, delay).ConfigureAwait(false)) { _pendingReadWait = null; - delayCancellation.Cancel(); - return await pendingRead.ConfigureAwait(false); + return true; } - if (remaining <= MaximumTimerDelay) - return false; + switch (_deadlineRace.Outcome) + { + case DeadlineReadRace.RaceOutcome.ReadClosed: + _pendingReadWait = null; + return false; + case DeadlineReadRace.RaceOutcome.TimedOut when remaining > MaximumTimerDelay: + // A chunk of a very long deadline expired: re-arm the same retained read. + continue; + default: + // The deadline expired and the pending read was not consumed: it stays + // retained in _pendingReadWait for WaitToReadAsync to re-observe. + return false; + } } } From 443f4f731a831d3fd66e861e335748e4fe9fb342 Mon Sep 17 00:00:00 2001 From: sunsi Date: Sat, 15 Aug 2026 13:39:17 +0800 Subject: [PATCH 3/6] fix: guard deadline race timer callbacks by arm generation; record test timer arms after install (issue-204 codex review) --- src/SharpLink.Runtime/DeadlineReadRace.cs | 19 ++- .../Runtime/SendPumpTests.cs | 115 +++++++++++++++++- 2 files changed, 127 insertions(+), 7 deletions(-) diff --git a/src/SharpLink.Runtime/DeadlineReadRace.cs b/src/SharpLink.Runtime/DeadlineReadRace.cs index a4894da6..496582de 100644 --- a/src/SharpLink.Runtime/DeadlineReadRace.cs +++ b/src/SharpLink.Runtime/DeadlineReadRace.cs @@ -26,7 +26,11 @@ namespace SharpLink.Runtime; /// runs inline for completed /// tasks, which is why the timer is created before the continuation is registered, and why each /// arm captures its own read so a late continuation from a previous arm can never act on the -/// current arm's state (the identity check makes stale completions no-ops). +/// current arm's state (the identity check makes stale completions no-ops). For the same reason +/// each arm stamps its timer callback with a monotonically increasing generation: disposing a +/// fired timer does not guarantee that an already queued callback has finished running, so a +/// stale callback must recognize that it no longer belongs to the current arm before it may +/// touch the timer or the completion source. /// /// internal sealed class DeadlineReadRace : IValueTaskSource, IDisposable @@ -39,15 +43,13 @@ internal enum RaceOutcome TimedOut, } - private static readonly TimerCallback s_timerCallback = - static state => ((DeadlineReadRace)state!).OnTimerFired(); - private readonly TimeProvider _timeProvider; private ManualResetValueTaskSourceCore _core; private Task? _read; private ITimer? _timer; private RaceOutcome _outcome; private int _readAbandoned; + private long _armGeneration; internal DeadlineReadRace(TimeProvider timeProvider) { @@ -93,7 +95,9 @@ internal ValueTask WaitForReadOrTimeout(Task read, TimeSpan timeout) // The timer must be armed before the read continuation is registered: a read that // completes in this window invokes the continuation inline, and the read-win path // disposes the timer it expects to exist. - _timer = _timeProvider.CreateTimer(s_timerCallback, this, timeout, Timeout.InfiniteTimeSpan); + var generation = Interlocked.Increment(ref _armGeneration); + _timer = _timeProvider.CreateTimer( + state => OnTimerFired(generation), this, timeout, Timeout.InfiniteTimeSpan); // Each arm registers a fresh closure capturing its own read. A closure from an earlier // arm that fires late must not be able to act on this arm's state: the identity check @@ -125,8 +129,11 @@ private void OnReadCompleted(Task read) } } - private void OnTimerFired() + private void OnTimerFired(long generation) { + if (generation != Volatile.Read(ref _armGeneration)) + return; // A queued timer callback from an earlier arm: never touch the current arm. + if (Interlocked.Exchange(ref _readAbandoned, 1) != 0) return; // The read completed first and disposed the timer. diff --git a/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs b/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs index e1e1d2a2..66cab14c 100644 --- a/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs +++ b/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs @@ -188,6 +188,64 @@ public async Task TimedBatchShouldExtendBatchForFrameArrivingBeforeDeadline() } } + [Test] + public async Task TimedBatchShouldIgnoreStaleTimerCallbackFromPreviousArm() + { + var clock = new ManualTimeProvider(); + var maxLatency = TimeSpan.FromMilliseconds(100); + var provider = new StaleCallbackTimeProvider(clock); + var input = new Pipe(); + var output = new Pipe(); + using var context = new SharpLinkRuntimeContextBuilder() + .UseTimeProvider(provider) + .Build(includeGeneratedAssemblyCatalog: false); + var session = RpcSessionTestFixture.CreateSessionOverTestTransport( + "timed-batch-stale-timer-callback", + input.Reader, + output.Writer, + RpcSessionTestFixture.ClientOptions( + context, + new RpcSessionFlushOptions(1024 * 1024, maxLatency))); + var first = CreateFrame(session, 32, requestId: 1); + var second = CreateFrame(session, 32, requestId: 2); + try + { + session.SendPacket(first); + await WaitUntilAsync(() => clock.ActiveTimerCount > 0); + clock.Advance(TimeSpan.FromMilliseconds(50)); + + session.SendPacket(second); + // The arriving frame wins the first race; the pump re-arms for the remaining + // latency, which is the durable observation that the first arm is superseded. + await WaitUntilAsync(() => provider.WasArmed(TimeSpan.FromMilliseconds(50))); + + // The first arm's timer callback fires out of band, as if it had been dequeued by + // the timer queue but not yet executed. It belongs to a superseded arm and must not + // disarm the current deadline timer or complete the current wait. + provider.InvokeArmedCallback(0); + await Task.Delay(50); + Ensure(clock.ActiveTimerCount == 1, + "a stale timer callback must not disarm the current deadline timer"); + + clock.Advance(TimeSpan.FromMilliseconds(50)); + var read = await output.Reader.ReadAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(2)); + var expectedBytes = 2 * (ProtocolV2Constants.HeaderBytes + 32); + Ensure(read.Buffer.Length >= expectedBytes, + "both frames must share one flush at the first frame's deadline"); + output.Reader.AdvanceTo(read.Buffer.End); + + await WaitUntilAsync(() => session.QueuedSendBytes == 0); + EnsureReturned(first, "the deadline flush must return the first frame owner"); + EnsureReturned(second, "the deadline flush must return the second frame owner"); + } + finally + { + await session.DisposeAsync(); + await output.Reader.CompleteAsync(); + await input.Writer.CompleteAsync(); + } + } + [Test] public async Task TimedBatchShouldRearmAcrossMaximumTimerDelayChunks() { @@ -533,9 +591,55 @@ public override ITimer CreateTimer( TimeSpan dueTime, TimeSpan period) { + var timer = inner.CreateTimer(callback, state, dueTime, period); + // Publish only after the timer is actually installed: the test advances the manual + // clock once the arm is observed, and the due time must be relative to the clock + // position at arm time. + lock (_gate) + _armedDueTimes.Add(dueTime); + return timer; + } + + internal bool WasArmed(TimeSpan dueTime) + { + lock (_gate) + return _armedDueTimes.Contains(dueTime); + } + } + + /// + /// Wraps and keeps every armed timer callback invocable + /// out of band, simulating a fired timer whose callback is still queued after a later arm + /// replaced the race state (timer-queue disposal cannot cancel an already dequeued work + /// item). + /// + private sealed class StaleCallbackTimeProvider(ManualTimeProvider inner) : TimeProvider + { + private readonly Lock _gate = new(); + private readonly List<(TimerCallback Callback, object? State)> _armedCallbacks = []; + private readonly List _armedDueTimes = []; + + public override long TimestampFrequency => inner.TimestampFrequency; + + public override TimeZoneInfo LocalTimeZone => inner.LocalTimeZone; + + public override DateTimeOffset GetUtcNow() => inner.GetUtcNow(); + + public override long GetTimestamp() => inner.GetTimestamp(); + + public override ITimer CreateTimer( + TimerCallback callback, + object? state, + TimeSpan dueTime, + TimeSpan period) + { + var timer = inner.CreateTimer(callback, state, dueTime, period); lock (_gate) + { + _armedCallbacks.Add((callback, state)); _armedDueTimes.Add(dueTime); - return inner.CreateTimer(callback, state, dueTime, period); + } + return timer; } internal bool WasArmed(TimeSpan dueTime) @@ -543,6 +647,15 @@ internal bool WasArmed(TimeSpan dueTime) lock (_gate) return _armedDueTimes.Contains(dueTime); } + + internal void InvokeArmedCallback(int index) + { + TimerCallback callback; + object? state; + lock (_gate) + (callback, state) = _armedCallbacks[index]; + callback(state); + } } private static void Ensure(bool condition, string message) From 0a5c8c2f4509c6eac47e4711995f0d33ff4899a7 Mon Sep 17 00:00:00 2001 From: sunsi Date: Sat, 15 Aug 2026 19:31:28 +0800 Subject: [PATCH 4/6] fix: claim deadline-race arms atomically by generation and publish timers before arming (issue-204 codex review round 2) --- src/SharpLink.Runtime/DeadlineReadRace.cs | 68 +++++++++---------- .../Runtime/SendPumpTests.cs | 65 ++++++++++++------ 2 files changed, 75 insertions(+), 58 deletions(-) diff --git a/src/SharpLink.Runtime/DeadlineReadRace.cs b/src/SharpLink.Runtime/DeadlineReadRace.cs index 496582de..aa0a356b 100644 --- a/src/SharpLink.Runtime/DeadlineReadRace.cs +++ b/src/SharpLink.Runtime/DeadlineReadRace.cs @@ -9,8 +9,8 @@ namespace SharpLink.Runtime; /// , or a per-pump /// . A single /// instance is reused for every deadline wait of one send pump, so only the arm itself -/// allocates: one continuation closure -/// per wait plus one from the owner's . +/// allocates: two continuation closures per wait plus one from the +/// owner's . /// /// /// @@ -21,16 +21,18 @@ namespace SharpLink.Runtime; /// /// /// The instance is single-flight: an arm must be fully awaited before the next arm. The owner -/// (a single-threaded send pump) satisfies this by construction. A read that completes while an -/// arm is being set up is still handled correctly: the continuation registered by +/// (a single-threaded send pump) satisfies this by construction. Callbacks that outlive their +/// arm are neutralized by an atomic claim: each arm publishes a unique token, and the read +/// callback and the timer callback race to claim that token with a single +/// . A stale callback's token no +/// longer matches the published one, so it can never dispose a later arm's timer or complete a +/// later arm's source, no matter how late it runs. The timer is additionally created in a +/// disabled state and armed via only after the +/// field that owns it has been published, so a deadline already in the past can never invoke a +/// callback that observes an unpublished timer. A read that completes while an arm is being set +/// up is still handled correctly: the continuation registered by /// runs inline for completed -/// tasks, which is why the timer is created before the continuation is registered, and why each -/// arm captures its own read so a late continuation from a previous arm can never act on the -/// current arm's state (the identity check makes stale completions no-ops). For the same reason -/// each arm stamps its timer callback with a monotonically increasing generation: disposing a -/// fired timer does not guarantee that an already queued callback has finished running, so a -/// stale callback must recognize that it no longer belongs to the current arm before it may -/// touch the timer or the completion source. +/// tasks, and the timer field is already published at that point. /// /// internal sealed class DeadlineReadRace : IValueTaskSource, IDisposable @@ -43,13 +45,15 @@ internal enum RaceOutcome TimedOut, } + private const long ReadClaimBit = 1; + private const long TimerClaimBit = 2; + private readonly TimeProvider _timeProvider; private ManualResetValueTaskSourceCore _core; - private Task? _read; private ITimer? _timer; private RaceOutcome _outcome; - private int _readAbandoned; private long _armGeneration; + private long _armClaim; internal DeadlineReadRace(TimeProvider timeProvider) { @@ -87,32 +91,25 @@ internal ValueTask WaitForReadOrTimeout(Task read, TimeSpan timeout) return new ValueTask(read); } - _read = read; + var token = (++_armGeneration) << 2; Volatile.Write(ref Unsafe.As(ref _outcome), (int)RaceOutcome.Pending); - Volatile.Write(ref _readAbandoned, 0); _core.Reset(); - // The timer must be armed before the read continuation is registered: a read that - // completes in this window invokes the continuation inline, and the read-win path - // disposes the timer it expects to exist. - var generation = Interlocked.Increment(ref _armGeneration); + // Publish the arm token before either callback can run, then publish the timer before + // it can fire: create it disabled, arm it via Change, and only then register the read + // continuation (which runs inline for a read that completes during the setup). + Volatile.Write(ref _armClaim, token); _timer = _timeProvider.CreateTimer( - state => OnTimerFired(generation), this, timeout, Timeout.InfiniteTimeSpan); - - // Each arm registers a fresh closure capturing its own read. A closure from an earlier - // arm that fires late must not be able to act on this arm's state: the identity check - // against the current read makes stale completions no-ops. - read.GetAwaiter().UnsafeOnCompleted(() => OnReadCompleted(read)); + _ => OnTimerFired(token), this, Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan); + _timer.Change(timeout, Timeout.InfiniteTimeSpan); + read.GetAwaiter().UnsafeOnCompleted(() => OnReadCompleted(read, token)); return new ValueTask(this, _core.Version); } - private void OnReadCompleted(Task read) + private void OnReadCompleted(Task read, long token) { - if (!ReferenceEquals(read, _read)) - return; // Stale completion from a previous arm: never touch the current cycle's state. - - if (Interlocked.Exchange(ref _readAbandoned, 1) != 0) - return; // The timer won first: the read stays unconsumed for later reuse. + if (Interlocked.CompareExchange(ref _armClaim, token | ReadClaimBit, token) != token) + return; // Superseded arm or already claimed by the timer: the read stays unconsumed. _timer!.Dispose(); if (read.IsCompletedSuccessfully) @@ -129,13 +126,10 @@ private void OnReadCompleted(Task read) } } - private void OnTimerFired(long generation) + private void OnTimerFired(long token) { - if (generation != Volatile.Read(ref _armGeneration)) - return; // A queued timer callback from an earlier arm: never touch the current arm. - - if (Interlocked.Exchange(ref _readAbandoned, 1) != 0) - return; // The read completed first and disposed the timer. + if (Interlocked.CompareExchange(ref _armClaim, token | TimerClaimBit, token) != token) + return; // Superseded arm or already claimed by the read. _timer!.Dispose(); Volatile.Write(ref Unsafe.As(ref _outcome), (int)RaceOutcome.TimedOut); diff --git a/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs b/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs index 66cab14c..d0638449 100644 --- a/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs +++ b/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs @@ -95,10 +95,11 @@ public async Task TimedBatchShouldDeliverFrameSentAfterDeadlineFlushThroughRetai { var clock = new ManualTimeProvider(); var maxLatency = TimeSpan.FromMilliseconds(100); + var provider = new TimerArmRecordingTimeProvider(clock); var input = new Pipe(); var output = new Pipe(); using var context = new SharpLinkRuntimeContextBuilder() - .UseTimeProvider(clock) + .UseTimeProvider(provider) .Build(includeGeneratedAssemblyCatalog: false); var session = RpcSessionTestFixture.CreateSessionOverTestTransport( "timed-batch-retained-read", @@ -111,7 +112,7 @@ public async Task TimedBatchShouldDeliverFrameSentAfterDeadlineFlushThroughRetai { var first = CreateFrame(session, 32, requestId: 1); session.SendPacket(first); - await WaitUntilAsync(() => clock.ActiveTimerCount > 0); + await WaitUntilAsync(() => provider.WasArmed(maxLatency)); clock.Advance(maxLatency); await ConsumeAvailableAsync(output.Reader); @@ -211,7 +212,7 @@ public async Task TimedBatchShouldIgnoreStaleTimerCallbackFromPreviousArm() try { session.SendPacket(first); - await WaitUntilAsync(() => clock.ActiveTimerCount > 0); + await WaitUntilAsync(() => provider.WasArmed(maxLatency)); clock.Advance(TimeSpan.FromMilliseconds(50)); session.SendPacket(second); @@ -252,10 +253,11 @@ public async Task TimedBatchShouldRearmAcrossMaximumTimerDelayChunks() var clock = new ManualTimeProvider(); var chunk = TimeSpan.FromMilliseconds(int.MaxValue); var maxLatency = chunk + TimeSpan.FromMilliseconds(1); + var provider = new TimerArmRecordingTimeProvider(clock); var input = new Pipe(); var output = new Pipe(); using var context = new SharpLinkRuntimeContextBuilder() - .UseTimeProvider(clock) + .UseTimeProvider(provider) .Build(includeGeneratedAssemblyCatalog: false); var session = RpcSessionTestFixture.CreateSessionOverTestTransport( "timed-batch-chunk-rearm", @@ -268,7 +270,7 @@ public async Task TimedBatchShouldRearmAcrossMaximumTimerDelayChunks() try { session.SendPacket(frame); - await WaitUntilAsync(() => clock.ActiveTimerCount > 0); + await WaitUntilAsync(() => provider.WasArmed(chunk)); Ensure(clock.EarliestTimerTimestamp == clock.GetTimestamp() + chunk.Ticks, "a deadline beyond the maximum timer delay must be armed as one full chunk"); @@ -297,10 +299,11 @@ public async Task TimedBatchDeadlineWaitShouldExitWhenSessionIsDisposed() { var clock = new ManualTimeProvider(); var maxLatency = TimeSpan.FromMilliseconds(100); + var provider = new TimerArmRecordingTimeProvider(clock); var input = new Pipe(); var output = new Pipe(); using var context = new SharpLinkRuntimeContextBuilder() - .UseTimeProvider(clock) + .UseTimeProvider(provider) .Build(includeGeneratedAssemblyCatalog: false); var session = RpcSessionTestFixture.CreateSessionOverTestTransport( "timed-batch-dispose-during-wait", @@ -313,7 +316,7 @@ public async Task TimedBatchDeadlineWaitShouldExitWhenSessionIsDisposed() try { session.SendPacket(frame); - await WaitUntilAsync(() => clock.ActiveTimerCount > 0); + await WaitUntilAsync(() => provider.WasArmed(maxLatency)); await session.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(2)); EnsureReturned(frame, "dispose during the deadline wait must return the frame owner"); @@ -566,9 +569,11 @@ public override ITimer CreateTimer( TimeSpan period) { var timer = inner.CreateTimer(callback, state, dueTime, period); - if (dueTime == expectedDueTime) - _expectedTimerArmed.TrySetResult(); - return timer; + return new HookedTimer(timer, changedDueTime => + { + if (changedDueTime == expectedDueTime) + _expectedTimerArmed.TrySetResult(); + }); } } @@ -592,12 +597,14 @@ public override ITimer CreateTimer( TimeSpan period) { var timer = inner.CreateTimer(callback, state, dueTime, period); - // Publish only after the timer is actually installed: the test advances the manual - // clock once the arm is observed, and the due time must be relative to the clock - // position at arm time. - lock (_gate) - _armedDueTimes.Add(dueTime); - return timer; + return new HookedTimer(timer, changedDueTime => + { + // The deadline race creates its timers disabled and arms them via Change, so + // the arm is only observable on the Change hook, after the timer is installed + // relative to the current clock position. + lock (_gate) + _armedDueTimes.Add(changedDueTime); + }); } internal bool WasArmed(TimeSpan dueTime) @@ -633,13 +640,14 @@ public override ITimer CreateTimer( TimeSpan dueTime, TimeSpan period) { - var timer = inner.CreateTimer(callback, state, dueTime, period); lock (_gate) - { _armedCallbacks.Add((callback, state)); - _armedDueTimes.Add(dueTime); - } - return timer; + var timer = inner.CreateTimer(callback, state, dueTime, period); + return new HookedTimer(timer, changedDueTime => + { + lock (_gate) + _armedDueTimes.Add(changedDueTime); + }); } internal bool WasArmed(TimeSpan dueTime) @@ -658,6 +666,21 @@ internal void InvokeArmedCallback(int index) } } + private sealed class HookedTimer( + ITimer inner, + Action onChangedDueTime) : ITimer + { + public bool Change(TimeSpan dueTime, TimeSpan period) + { + onChangedDueTime(dueTime); + return inner.Change(dueTime, period); + } + + public void Dispose() => inner.Dispose(); + + public ValueTask DisposeAsync() => inner.DisposeAsync(); + } + private static void Ensure(bool condition, string message) { if (!condition) From 95e33bbe8311e8a7ddb600b65672e0d116eb2b81 Mon Sep 17 00:00:00 2001 From: sunsi Date: Sat, 15 Aug 2026 19:31:29 +0800 Subject: [PATCH 5/6] test: deliver transport output faults through the session so a paused flush observes cancellation (issue-204) --- .../Runtime/SendPumpIdleShutdownTests.cs | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/test/SharpLink.UnitTests/Runtime/SendPumpIdleShutdownTests.cs b/test/SharpLink.UnitTests/Runtime/SendPumpIdleShutdownTests.cs index a91273b4..4ffeaee8 100644 --- a/test/SharpLink.UnitTests/Runtime/SendPumpIdleShutdownTests.cs +++ b/test/SharpLink.UnitTests/Runtime/SendPumpIdleShutdownTests.cs @@ -76,10 +76,17 @@ public async Task PumpBlockedInFlushExitsWhenTransportOutputFaults() var flush = session.SendPacketAndFlushAsync(frame).AsTask(); await WaitUntilAsync(() => session.QueuedSendBytes > 0); - await output.Writer.CompleteAsync(new IOException("output fault")); + // A real transport output fault is delivered twice: the transport completes its + // output pipe with the fault, and then notifies the session so the session + // cancellation tears the pump down. A faulted pipe alone never completes a pending + // FlushAsync (the pipe surfaces writer faults only to the reader), so the pump + // relies on the session cancellation to wake it from a paused flush. + var fault = new IOException("output fault"); + await output.Writer.CompleteAsync(fault); + session.NotifyDisconnected(fault); - var fault = await CaptureCompletionExceptionAsync(flush, TimeSpan.FromSeconds(5)); - Ensure(fault is SharpLinkException { Code: SharpLinkErrorCode.ConnectionClosed } or + var completionException = await CaptureCompletionExceptionAsync(flush, TimeSpan.FromSeconds(5)); + Ensure(completionException is SharpLinkException { Code: SharpLinkErrorCode.ConnectionClosed } or OperationCanceledException, "a transport output fault must fault the pending flush completion"); await session.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5)); From eb948c7837c025016a6afc88d9a82487432b66e7 Mon Sep 17 00:00:00 2001 From: sunsi Date: Sat, 15 Aug 2026 19:37:56 +0800 Subject: [PATCH 6/6] test: publish timer-arm hooks only after the timer is actually armed (issue-204 codex review round 3) --- test/SharpLink.UnitTests/Runtime/SendPumpTests.cs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs b/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs index d0638449..737e59fe 100644 --- a/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs +++ b/test/SharpLink.UnitTests/Runtime/SendPumpTests.cs @@ -672,8 +672,13 @@ private sealed class HookedTimer( { public bool Change(TimeSpan dueTime, TimeSpan period) { - onChangedDueTime(dueTime); - return inner.Change(dueTime, period); + var changed = inner.Change(dueTime, period); + // Publish the observation only after the timer is actually armed: the tests + // advance the manual clock once the arm is observed, and the due time must be + // relative to the clock position at arm time. + if (changed) + onChangedDueTime(dueTime); + return changed; } public void Dispose() => inner.Dispose();