Skip to content

refactor(runtime): collapse nine PooledAsyncStreamDispatcher test hook fields (#801) #809

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SunSi12138 merged 4 commits into from
Oct 11, 2026
Merged

refactor(runtime): collapse nine PooledAsyncStreamDispatcher test hook fields (#801) #809

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4f16d1d
chore(issue-801) stage dispatcher-hooks cleanup
SunSi12138 Oct 11, 2026
536d69c
refactor(runtime): lazily compact pooled stream dispatcher test callb…
github-actions[bot] Oct 11, 2026
e5220a7
style(issue-801): keep compact dispatcher hooks within reviewed LOC b…
SunSi12138 Oct 11, 2026
31b77a0
Merge latest dev into cleanup/issue-801-dispatcher-test-hooks
SunSi12138 Oct 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
107 changes: 53 additions & 54 deletions src/SharpLink.Runtime/PooledAsyncStreamDispatcher.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,15 +50,8 @@ internal sealed class PooledAsyncStreamDispatcher<T> :
private long _disposeFinalizedLeaseState;
private DisposeCompletion? _disposeCompletion;
private TaskCompletionSource? _remoteTerminalPublication;
private Action? _beforeConcurrentDisposeCompletionInstallForTests;
private Action? _beforeRemoteTerminalPublicationPublishForTests;
private Action? _beforeRemoteTerminalPublicationCompletionInstallForTests;
private Action? _afterRemoteTerminalPublicationCompletionInstallForTests;
private Action? _beforeProducerOperationAcquireForTests;
private Action? _beforeConsumerWaitOwnerAcquireForTests;
private Action? _afterConsumerWaitResultForTests;
private Action? _beforeReturnTransitionForTests;
private Action? _afterReturnTransitionForTests;
// Test hooks allocate storage only when a test installs a callback.
private Action?[]? _testHooks;
private int _terminalDispatchStateClosed;
// 0 = no remote terminal, 1 = terminal publication still owns dispatch-state close,
// 2 = remote terminal publication is complete.
Expand Down Expand Up @@ -241,15 +234,7 @@ private void Reset(
Volatile.Write(ref _consumerTerminal, 0);
Volatile.Write(ref _terminalDispatchStateClosed, 0);
Volatile.Write(ref _remoteTerminalPublication, null);
Volatile.Write(ref _beforeConcurrentDisposeCompletionInstallForTests, null);
Volatile.Write(ref _beforeRemoteTerminalPublicationPublishForTests, null);
Volatile.Write(ref _beforeRemoteTerminalPublicationCompletionInstallForTests, null);
Volatile.Write(ref _afterRemoteTerminalPublicationCompletionInstallForTests, null);
Volatile.Write(ref _beforeProducerOperationAcquireForTests, null);
Volatile.Write(ref _beforeConsumerWaitOwnerAcquireForTests, null);
Volatile.Write(ref _afterConsumerWaitResultForTests, null);
Volatile.Write(ref _beforeReturnTransitionForTests, null);
Volatile.Write(ref _afterReturnTransitionForTests, null);
Volatile.Write(ref _testHooks, null);
Volatile.Write(ref _remoteTerminalPublicationState, 0);
}

Expand Down Expand Up @@ -343,7 +328,7 @@ private bool TryAcquireDispatch(out long leaseState)
if (GetLeaseStatus(leaseState) != LeaseActive)
return false;

Volatile.Read(ref _beforeProducerOperationAcquireForTests)?.Invoke();
InvokeTestHook(4);
Interlocked.Increment(ref _producerOperations);
var currentLeaseState = Volatile.Read(ref _leaseState);
if (IsSameLeaseGeneration(currentLeaseState, leaseState) &&
Expand Down Expand Up @@ -413,33 +398,63 @@ internal long RetainForRegistration()
// boundary. It gates a rare second-dispose CAS so the old-generation/re-rent handoff can be
// tested deterministically without reflection; it is never read by a first disposal.
internal void SetBeforeConcurrentDisposeCompletionInstallForTests(Action? callback)
=> Volatile.Write(ref _beforeConcurrentDisposeCompletionInstallForTests, callback);
=> SetTestHook(0, callback);

// Gates the remote-terminal publication in Complete's finally so a racing DisposeAsync can
// deterministically install its completion while the publication state is still 1 (issue #206).
internal void SetBeforeRemoteTerminalPublicationPublishForTests(Action? callback)
=> Volatile.Write(ref _beforeRemoteTerminalPublicationPublishForTests, callback);
=> SetTestHook(1, callback);

internal void SetBeforeRemoteTerminalPublicationCompletionInstallForTests(Action? callback)
=> Volatile.Write(ref _beforeRemoteTerminalPublicationCompletionInstallForTests, callback);
=> SetTestHook(2, callback);

internal void SetAfterRemoteTerminalPublicationCompletionInstallForTests(Action? callback)
=> Volatile.Write(ref _afterRemoteTerminalPublicationCompletionInstallForTests, callback);
=> SetTestHook(3, callback);

internal void SetBeforeProducerOperationAcquireForTests(Action? callback)
=> Volatile.Write(ref _beforeProducerOperationAcquireForTests, callback);
=> SetTestHook(4, callback);

internal void SetBeforeConsumerWaitOwnerAcquireForTests(Action? callback)
=> Volatile.Write(ref _beforeConsumerWaitOwnerAcquireForTests, callback);
=> SetTestHook(5, callback);

internal void SetAfterConsumerWaitResultForTests(Action? callback)
=> Volatile.Write(ref _afterConsumerWaitResultForTests, callback);
=> SetTestHook(6, callback);

internal void SetBeforeReturnTransitionForTests(Action? callback)
=> Volatile.Write(ref _beforeReturnTransitionForTests, callback);
=> SetTestHook(7, callback);

internal void SetAfterReturnTransitionForTests(Action? callback)
=> Volatile.Write(ref _afterReturnTransitionForTests, callback);
=> SetTestHook(8, callback);
private void SetTestHook(int index, Action? callback)
{
var hooks = Volatile.Read(ref _testHooks);
if (hooks is null)
{
if (callback is null)
return;
var candidate = new Action?[9];
hooks = Interlocked.CompareExchange(ref _testHooks, candidate, null) ?? candidate;
}
Volatile.Write(ref hooks[index], callback);
}

private void InvokeTestHook(int index)
{
var hooks = Volatile.Read(ref _testHooks);
if (hooks is not null)
Volatile.Read(ref hooks[index])?.Invoke();
}

private bool HasTestHooks()
{
var hooks = Volatile.Read(ref _testHooks);
if (hooks is null)
return false;
foreach (var callback in hooks)
if (callback is not null)
return true;
return false;
}

private static ValueTask RejectedDispatch()
{
Expand Down Expand Up @@ -723,7 +738,7 @@ public void Complete(Exception? exception)
}
finally
{
Volatile.Read(ref _beforeRemoteTerminalPublicationPublishForTests)?.Invoke();
InvokeTestHook(1);

// Publish only after the first remote Close and Signal are complete. A racing
// consumer DisposeAsync then performs its own second Close in sequence.
Expand Down Expand Up @@ -873,7 +888,7 @@ private async ValueTask<bool> SlowMoveNextAsync(long consumerLeaseState)
ThrowIfEnumerationCanceled();
if (!ownsConsumerOperation)
{
Volatile.Read(ref _beforeConsumerWaitOwnerAcquireForTests)?.Invoke();
InvokeTestHook(5);
Interlocked.Increment(ref _consumerOperations);
var currentLeaseState = Volatile.Read(ref _leaseState);
if (!IsSameLeaseGeneration(currentLeaseState, consumerLeaseState) ||
Expand All @@ -889,7 +904,7 @@ private async ValueTask<bool> SlowMoveNextAsync(long consumerLeaseState)
ownsConsumerOperation = true;
}
await WaitForDataAsync().ConfigureAwait(false);
Volatile.Read(ref _afterConsumerWaitResultForTests)?.Invoke();
InvokeTestHook(6);
}
}
finally
Expand Down Expand Up @@ -991,7 +1006,7 @@ private ValueTask AwaitConcurrentDisposeAsync(long disposingLeaseState)
}

var created = new DisposeCompletion(disposingLeaseState);
Volatile.Read(ref _beforeConcurrentDisposeCompletionInstallForTests)?.Invoke();
InvokeTestHook(0);
if (Interlocked.CompareExchange(ref _disposeCompletion, created, null) is not null)
continue;

Expand Down Expand Up @@ -1020,7 +1035,7 @@ private ValueTask WaitForRemoteTerminalPublicationAsync()
if (Volatile.Read(ref _remoteTerminalPublicationState) == 2)
return ValueTask.CompletedTask;

Volatile.Read(ref _beforeRemoteTerminalPublicationCompletionInstallForTests)?.Invoke();
InvokeTestHook(2);

var completion = Volatile.Read(ref _remoteTerminalPublication);
if (completion is null)
Expand All @@ -1029,7 +1044,7 @@ private ValueTask WaitForRemoteTerminalPublicationAsync()
completion = Interlocked.CompareExchange(ref _remoteTerminalPublication, created, null) ?? created;
}

Volatile.Read(ref _afterRemoteTerminalPublicationCompletionInstallForTests)?.Invoke();
InvokeTestHook(3);

// Fenced re-read (issue #206): mirrors the publisher-side fence in Complete. The
// install CAS above is already a full barrier, but a fenced read keeps the two sides
Expand Down Expand Up @@ -1423,13 +1438,13 @@ private void TryReturnToPool(long disposingLeaseState)
return;

var returnedState = unchecked(disposingLeaseState + 2);
Volatile.Read(ref _beforeReturnTransitionForTests)?.Invoke();
InvokeTestHook(7);
if (Interlocked.CompareExchange(
ref _leaseState,
returnedState,
disposingLeaseState) != disposingLeaseState)
return;
Volatile.Read(ref _afterReturnTransitionForTests)?.Invoke();
InvokeTestHook(8);

// Close the acquire-vs-return race before clearing lease state.
var dispatchState = Volatile.Read(ref _dispatchState);
Expand Down Expand Up @@ -1472,15 +1487,7 @@ private void TryReturnToPool(long disposingLeaseState)
Volatile.Write(ref _dispatchState, null);
Volatile.Write(ref _disposeCompletion, null);
Volatile.Write(ref _remoteTerminalPublication, null);
Volatile.Write(ref _beforeConcurrentDisposeCompletionInstallForTests, null);
Volatile.Write(ref _beforeRemoteTerminalPublicationPublishForTests, null);
Volatile.Write(ref _beforeRemoteTerminalPublicationCompletionInstallForTests, null);
Volatile.Write(ref _afterRemoteTerminalPublicationCompletionInstallForTests, null);
Volatile.Write(ref _beforeProducerOperationAcquireForTests, null);
Volatile.Write(ref _beforeConsumerWaitOwnerAcquireForTests, null);
Volatile.Write(ref _afterConsumerWaitResultForTests, null);
Volatile.Write(ref _beforeReturnTransitionForTests, null);
Volatile.Write(ref _afterReturnTransitionForTests, null);
Volatile.Write(ref _testHooks, null);

// 复位枚举器占用标记
Volatile.Write(ref _enumeratorTaken, 0);
Expand Down Expand Up @@ -1557,15 +1564,7 @@ _current is not null || _enumerationToken.CanBeCanceled ||
!_additionalEnumerationCancellationRegistration.Equals(default) ||
Volatile.Read(ref _disposeCompletion) is not null ||
Volatile.Read(ref _remoteTerminalPublication) is not null ||
Volatile.Read(ref _beforeConcurrentDisposeCompletionInstallForTests) is not null ||
Volatile.Read(ref _beforeRemoteTerminalPublicationPublishForTests) is not null ||
Volatile.Read(ref _beforeRemoteTerminalPublicationCompletionInstallForTests) is not null ||
Volatile.Read(ref _afterRemoteTerminalPublicationCompletionInstallForTests) is not null ||
Volatile.Read(ref _beforeProducerOperationAcquireForTests) is not null ||
Volatile.Read(ref _beforeConsumerWaitOwnerAcquireForTests) is not null ||
Volatile.Read(ref _afterConsumerWaitResultForTests) is not null ||
Volatile.Read(ref _beforeReturnTransitionForTests) is not null ||
Volatile.Read(ref _afterReturnTransitionForTests) is not null)
HasTestHooks())
{
return true;
}
Expand Down
Loading