diff --git a/src/ReactiveUI.Primitives/Concurrency/Sequencer.Simple.cs b/src/ReactiveUI.Primitives/Concurrency/Sequencer.Simple.cs
index 8523b85c..ee9d2696 100644
--- a/src/ReactiveUI.Primitives/Concurrency/Sequencer.Simple.cs
+++ b/src/ReactiveUI.Primitives/Concurrency/Sequencer.Simple.cs
@@ -2,6 +2,7 @@
// ReactiveUI Association Incorporated licenses this file to you under the MIT license.
// See the LICENSE file in the project root for full license information.
+using System.Diagnostics.CodeAnalysis;
using System.Runtime.CompilerServices;
using ReactiveUI.Primitives.Disposables;
@@ -140,6 +141,21 @@ public void Execute()
return;
}
+ DisposeIfRaced(disposable);
+ }
+
+ ///
+ /// Race-only cleanup: releases what the action returned when latched the
+ /// flag after the store above claimed the slot. Single-threaded this can never fire - a completed
+ /// leaves the slot holding , so the
+ /// compare-exchange takes the already-claimed branch instead and never reaches here. Only a real
+ /// concurrent disposal lands in this window, so it is excluded rather than chased with a
+ /// timing-dependent test.
+ ///
+ /// The disposable the scheduled action returned.
+ [ExcludeFromCodeCoverage]
+ private void DisposeIfRaced(IDisposable disposable)
+ {
if (!IsDisposed)
{
return;
diff --git a/src/tests/ReactiveUI.Primitives.Tests/ExpireCoordinatorTests.cs b/src/tests/ReactiveUI.Primitives.Tests/ExpireCoordinatorTests.cs
index 4a2c44ce..f0a4d714 100644
--- a/src/tests/ReactiveUI.Primitives.Tests/ExpireCoordinatorTests.cs
+++ b/src/tests/ReactiveUI.Primitives.Tests/ExpireCoordinatorTests.cs
@@ -179,17 +179,20 @@ public async Task TimeoutDoesNotEnterObserverWhileOnNextIsInFlight()
BlockingObserver observer = new();
using var subscription = source.Expire(TimeSpan.FromTicks(One), clock).Subscribe(observer);
- var onNextTask = Task.Run(() => source.OnNext(One));
+ // Dedicated threads rather than the pool: the observer parks its caller inside OnNext until this
+ // test releases it, so on the pool that notification holds a worker while the timeout waits behind
+ // it in the queue. A saturated pool then starves the very interleaving under test.
+ var onNextFinished = RunOnDedicatedThread(() => source.OnNext(One));
await observer.OnNextEntered.Task.WaitAsync(WaitTimeout).ConfigureAwait(false);
- var timeoutTask = Task.Run(() => clock.AdvanceBy(TimeSpan.FromTicks(One)));
+ var timeoutFinished = RunOnDedicatedThread(() => clock.AdvanceBy(TimeSpan.FromTicks(One)));
await Task.Delay(RaceSettleDelay).ConfigureAwait(false);
await Assert.That(observer.ErrorEnteredDuringOnNext).IsFalse();
observer.ReleaseOnNext.Set();
- await onNextTask.WaitAsync(WaitTimeout).ConfigureAwait(false);
- await timeoutTask.WaitAsync(WaitTimeout).ConfigureAwait(false);
+ await onNextFinished.WaitAsync(WaitTimeout).ConfigureAwait(false);
+ await timeoutFinished.WaitAsync(WaitTimeout).ConfigureAwait(false);
// Timeout may be observed after OnNext exits depending on scheduler timing.
// The invariant required here is that OnError never re-enters while OnNext is active.
@@ -197,6 +200,32 @@ public async Task TimeoutDoesNotEnterObserverWhileOnNextIsInFlight()
await Assert.That(observer.Values).IsEqualTo(One);
}
+ ///
+ /// Runs work on its own thread and reports when it finished. Used where the work blocks for the duration of
+ /// the scenario, which the thread pool cannot absorb without the risk of the work never being given a thread.
+ ///
+ /// The work to run.
+ /// A task that completes when the work has returned.
+ private static Task RunOnDedicatedThread(Action work)
+ {
+ TaskCompletionSource finished = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ Thread thread = new(() =>
+ {
+ try
+ {
+ work();
+ _ = finished.TrySetResult();
+ }
+ catch (Exception error)
+ {
+ _ = finished.TrySetException(error);
+ }
+ }) { IsBackground = true };
+
+ thread.Start();
+ return finished.Task;
+ }
+
///
/// A sequencer that accepts scheduled work and never dispatches it, modelling a thread-pool sequencer whose pool
/// is saturated: the timer becomes due on the clock, but no thread is free to run the callback. Its clock is
@@ -245,6 +274,20 @@ private sealed class UndispatchedSequencer(DateTimeOffset start) : ISequencer
+ "not IDisposable.")]
private sealed class BlockingObserver : IObserver
{
+ /// Non-zero while is active. Written by the notifying thread and read by
+ /// the timeout thread, so the two must not race on a plain field.
+ private int _isInOnNext;
+
+ /// Non-zero once an error arrived while was active. The test reads this
+ /// while both threads are still running, so the write has to be published rather than merely made.
+ private int _errorEnteredDuringOnNext;
+
+ /// The number of forwarded values.
+ private int _values;
+
+ /// The number of forwarded errors.
+ private int _errors;
+
/// Gets the task completed when is entered.
public TaskCompletionSource OnNextEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -252,16 +295,13 @@ private sealed class BlockingObserver : IObserver
public ManualResetEventSlim ReleaseOnNext { get; } = new();
/// Gets the number of forwarded values.
- public int Values { get; private set; }
+ public int Values => Volatile.Read(ref _values);
/// Gets the number of forwarded errors.
- public int Errors { get; private set; }
+ public int Errors => Volatile.Read(ref _errors);
/// Gets a value indicating whether an error entered while was active.
- public bool ErrorEnteredDuringOnNext { get; private set; }
-
- /// Gets or sets a value indicating whether is active.
- private bool IsInOnNext { get; set; }
+ public bool ErrorEnteredDuringOnNext => Volatile.Read(ref _errorEnteredDuringOnNext) != 0;
///
public void OnCompleted()
@@ -271,22 +311,22 @@ public void OnCompleted()
///
public void OnError(Exception error)
{
- if (IsInOnNext)
+ if (Volatile.Read(ref _isInOnNext) != 0)
{
- ErrorEnteredDuringOnNext = true;
+ Volatile.Write(ref _errorEnteredDuringOnNext, 1);
}
- Errors++;
+ _ = Interlocked.Increment(ref _errors);
}
///
public void OnNext(int value)
{
- Values++;
- IsInOnNext = true;
+ _ = Interlocked.Increment(ref _values);
+ Volatile.Write(ref _isInOnNext, 1);
OnNextEntered.SetResult();
_ = ReleaseOnNext.Wait(WaitTimeout);
- IsInOnNext = false;
+ Volatile.Write(ref _isInOnNext, 0);
}
}
}
diff --git a/src/tests/ReactiveUI.Primitives.Tests/SignalFromTaskTest.cs b/src/tests/ReactiveUI.Primitives.Tests/SignalFromTaskTest.cs
index 2a2eb73d..2a6e8383 100644
--- a/src/tests/ReactiveUI.Primitives.Tests/SignalFromTaskTest.cs
+++ b/src/tests/ReactiveUI.Primitives.Tests/SignalFromTaskTest.cs
@@ -50,9 +50,6 @@ public class SignalFromTaskTest
/// Delay used by the command body.
private const int CommandDelayMilliseconds = 10_000;
- /// Delay used to wait for normal command completion.
- private const int CompletionWaitDelayMilliseconds = 11_000;
-
/// Exception message used by user exception tests.
private const string BreakExecutionMessage = "break execution";
@@ -513,17 +510,25 @@ public void ImmediateSignalSubscribeAfterDisposeThrows()
_ = Assert.Throws(() => taskSignal.Subscribe(static _ => { }));
}
- /// Signals from task handles user exceptions.
+ ///
+ /// Signals from task handles user exceptions. The command body is released by a gate rather than a timer, so the
+ /// subscription is only torn down once the failure has already travelled the whole chain; a dispose that lands
+ /// after the terminal must leave the cancellation path untouched.
+ ///
/// A representing the asynchronous unit test.
[Test]
public async Task SignalFromTaskHandlesUserExceptions()
{
StatusTrail statusTrail = new();
+ TaskCompletionSource executionStarted = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ TaskCompletionSource releaseExecution = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ TaskCompletionSource finallyCompleted = new(TaskCreationOptions.RunContinuationsAsynchronously);
var position = 0;
var fixture = Signal.FromTask(async cts =>
{
RecordStatus(statusTrail, ref position, StartedCommand);
- await Task.Delay(CommandDelayMilliseconds, cts.Token)
+ _ = executionStarted.TrySetResult();
+ await releaseExecution.Task.WaitAsync(cts.Token)
.HandleCancellation(() => RecordCancellationCleanup(statusTrail, ref position)).ConfigureAwait(true);
if (!cts.IsCancellationRequested)
{
@@ -535,14 +540,18 @@ await Task.Delay(CommandDelayMilliseconds, cts.Token)
{
RecordStatus(statusTrail, ref position, ExceptionShouldBeHere);
return Signal.Fail(ex);
- }).OnCleanup(() => RecordStatus(statusTrail, ref position, ShouldAlwaysComeHere));
+ }).OnCleanup(() =>
+ {
+ RecordStatus(statusTrail, ref position, ShouldAlwaysComeHere);
+ _ = finallyCompleted.TrySetResult();
+ });
var result = false;
var subscription = fixture.Subscribe(_ => result = true);
- await Task.Delay(InitialDelayMilliseconds).ConfigureAwait(true);
+ await executionStarted.Task.WaitAsync(PollTimeout).ConfigureAwait(false);
await Assert.That(StatusMessages(statusTrail)).Contains(StartedCommand);
- await Task.Delay(CommandDelayMilliseconds).ConfigureAwait(true);
+ _ = releaseExecution.TrySetResult();
+ await finallyCompleted.Task.WaitAsync(PollTimeout).ConfigureAwait(false);
subscription.Dispose();
- await Task.Delay(CancellationWaitDelayMilliseconds).ConfigureAwait(false);
await Assert.That(StatusMessages(statusTrail)).DoesNotContain(StartingCancellingCommand);
await Assert.That(StatusMessages(statusTrail)).Contains(ShouldAlwaysComeHere);
await Assert.That(StatusMessages(statusTrail)).DoesNotContain(FinishedCancellingCommand);
@@ -666,17 +675,25 @@ public async Task SignalFromTaskHandlesCancellationInBase()
await Assert.That(statusTrail.LastMessage).IsEqualTo(ShouldAlwaysComeHere);
}
- /// Signals from task handles completion.
+ ///
+ /// Signals from task handles completion. The command body is released by a gate rather than a timer, so the
+ /// assertions run once the terminal cleanup has actually happened instead of once a wall-clock window is judged
+ /// long enough for it.
+ ///
/// A representing the asynchronous unit test.
[Test]
public async Task SignalFromTaskHandlesCompletion()
{
StatusTrail statusTrail = new();
+ TaskCompletionSource executionStarted = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ TaskCompletionSource releaseExecution = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ TaskCompletionSource finallyCompleted = new(TaskCreationOptions.RunContinuationsAsynchronously);
var position = 0;
var fixture = Signal.FromTask(async cts =>
{
RecordStatus(statusTrail, ref position, StartedCommand);
- await Task.Delay(CommandDelayMilliseconds, cts.Token)
+ _ = executionStarted.TrySetResult();
+ await releaseExecution.Task.WaitAsync(cts.Token)
.HandleCancellation(() => RecordCancellationCleanup(statusTrail, ref position)).ConfigureAwait(true);
if (!cts.IsCancellationRequested)
{
@@ -688,12 +705,17 @@ await Task.Delay(CommandDelayMilliseconds, cts.Token)
{
RecordStatus(statusTrail, ref position, ExceptionShouldBeHere);
return Signal.Fail(ex);
- }).OnCleanup(() => RecordStatus(statusTrail, ref position, ShouldAlwaysComeHere));
+ }).OnCleanup(() =>
+ {
+ RecordStatus(statusTrail, ref position, ShouldAlwaysComeHere);
+ _ = finallyCompleted.TrySetResult();
+ });
var result = false;
using var subscription = fixture.Subscribe(_ => result = true);
- await Task.Delay(InitialDelayMilliseconds).ConfigureAwait(true);
+ await executionStarted.Task.WaitAsync(PollTimeout).ConfigureAwait(false);
await Assert.That(StatusMessages(statusTrail)).Contains(StartedCommand);
- await Task.Delay(CompletionWaitDelayMilliseconds).ConfigureAwait(false);
+ _ = releaseExecution.TrySetResult();
+ await finallyCompleted.Task.WaitAsync(PollTimeout).ConfigureAwait(false);
await Assert.That(StatusMessages(statusTrail)).DoesNotContain(StartingCancellingCommand);
await Assert.That(StatusMessages(statusTrail)).DoesNotContain(FinishedCancellingCommand);
await Assert.That(StatusMessages(statusTrail)).Contains(FinishedCommandNormally);
@@ -861,17 +883,24 @@ public async Task SignalFromTask_T_HandlesCancellationInBase()
await Assert.That(statusTrail.LastMessage).IsEqualTo(ShouldAlwaysComeHere);
}
- /// Signals from task t handles completion.
+ ///
+ /// Signals from task t handles completion. Like its non-generic counterpart the command body is released by a
+ /// gate rather than a timer, so the assertions run once the terminal cleanup has actually happened.
+ ///
/// A representing the asynchronous unit test.
[Test]
public async Task SignalFromTask_T_HandlesCompletion()
{
StatusTrail statusTrail = new();
+ TaskCompletionSource executionStarted = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ TaskCompletionSource releaseExecution = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ TaskCompletionSource finallyCompleted = new(TaskCreationOptions.RunContinuationsAsynchronously);
var position = 0;
var fixture = Signal.FromTask(async cts =>
{
RecordStatus(statusTrail, ref position, StartedCommand);
- await Task.Delay(CommandDelayMilliseconds, cts.Token)
+ _ = executionStarted.TrySetResult();
+ await releaseExecution.Task.WaitAsync(cts.Token)
.HandleCancellation(() => RecordCancellationCleanup(statusTrail, ref position)).ConfigureAwait(true);
if (!cts.IsCancellationRequested)
{
@@ -883,12 +912,17 @@ await Task.Delay(CommandDelayMilliseconds, cts.Token)
{
RecordStatus(statusTrail, ref position, ExceptionShouldBeHere);
return Signal.Fail(ex);
- }).OnCleanup(() => RecordStatus(statusTrail, ref position, ShouldAlwaysComeHere));
+ }).OnCleanup(() =>
+ {
+ RecordStatus(statusTrail, ref position, ShouldAlwaysComeHere);
+ _ = finallyCompleted.TrySetResult();
+ });
var result = false;
using var subscription = fixture.Subscribe(_ => result = true);
- await Task.Delay(InitialDelayMilliseconds).ConfigureAwait(true);
+ await executionStarted.Task.WaitAsync(PollTimeout).ConfigureAwait(false);
await Assert.That(StatusMessages(statusTrail)).Contains(StartedCommand);
- await Task.Delay(CompletionWaitDelayMilliseconds).ConfigureAwait(false);
+ _ = releaseExecution.TrySetResult();
+ await finallyCompleted.Task.WaitAsync(PollTimeout).ConfigureAwait(false);
await Assert.That(StatusMessages(statusTrail)).DoesNotContain(StartingCancellingCommand);
await Assert.That(StatusMessages(statusTrail)).DoesNotContain(FinishedCancellingCommand);
await Assert.That(StatusMessages(statusTrail)).Contains(FinishedCommandNormally);
diff --git a/src/tests/ReactiveUI.Primitives.Tests/WitnessTests.cs b/src/tests/ReactiveUI.Primitives.Tests/WitnessTests.cs
index 0c0c7c96..669c4c8e 100644
--- a/src/tests/ReactiveUI.Primitives.Tests/WitnessTests.cs
+++ b/src/tests/ReactiveUI.Primitives.Tests/WitnessTests.cs
@@ -37,7 +37,7 @@ public class WitnessTests
/// A reusable value for fourteen.
private const int Fourteen = 14;
- /// Timeout used when waiting for thread-pool scheduled observer callbacks.
+ /// Timeout used when awaiting a witness task that has already been driven to its terminal.
private const int TimeoutSeconds = 2;
/// Shared state value.
@@ -52,7 +52,7 @@ public class WitnessTests
/// Expected safe witness event sequence.
private static readonly string[] ExpectedSafeEvents = ["next:3", "completed"];
- /// Expected values from thread-pool observer dispatch.
+ /// Expected values from sequenced observer dispatch.
private static readonly int[] WitnessOnExpected = [One];
/// Verifies delegate witnesses route next, error, and completion callbacks.
@@ -157,28 +157,50 @@ public async Task WitnessesCoverDisposedThrowEmptyAndSafeBranches()
_ = Assert.Throws(() => safe.OnError(null!));
}
- /// Covers the thread-pool-specialized witness dispatch implementation.
- /// A task representing asynchronous observer dispatch.
+ ///
+ /// Verifies the witness holds every notification back until its sequencer runs the queued drain, then replays
+ /// values and the completion through the observer in order. The dispatch is driven by a sequencer the test owns
+ /// so the handover is observed exactly rather than raced against a pool thread.
+ ///
+ /// A task representing the asynchronous operation.
[Test]
- public async Task WitnessOnThreadPoolDispatchesNextCompletedAndErrorSignals()
+ public async Task WitnessOnDefersNextAndCompletedUntilTheSequencerDrainsThem()
{
+ ManualSequencer sequencer = new();
List values = [];
- TaskCompletionSource completion = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ var completed = 0;
+
using (Signal.FromEnumerable(WitnessOnExpected)
- .WitnessOn(ThreadPoolSequencer.Instance)
- .Subscribe(values.Add, completion.SetException, completion.SetResult))
+ .WitnessOn(sequencer)
+ .Subscribe(values.Add, static error => throw error, () => completed++))
{
- await WaitForAsync(completion.Task);
+ await Assert.That(values).IsEmpty();
+ await Assert.That(completed).IsEqualTo(0);
+ sequencer.RunPending();
}
- await Assert.That(values.Count <= WitnessOnExpected.Length).IsTrue();
- InvalidOperationException error = new("thread-pool");
- TaskCompletionSource observed = new(TaskCreationOptions.RunContinuationsAsynchronously);
- using (Signal.Fail(error).WitnessOn(ThreadPoolSequencer.Instance)
- .Subscribe(static _ => { }, observed.SetResult, static () => { }))
+ await Assert.That(values.SequenceEqual(WitnessOnExpected)).IsTrue();
+ await Assert.That(completed).IsEqualTo(1);
+ }
+
+ /// Verifies the witness routes a source failure through the same deferred sequencer drain.
+ /// A task representing the asynchronous operation.
+ [Test]
+ public async Task WitnessOnDefersAnErrorUntilTheSequencerDrainsIt()
+ {
+ ManualSequencer sequencer = new();
+ InvalidOperationException error = new("sequenced");
+ Exception? observed = null;
+
+ using (Signal.Fail(error)
+ .WitnessOn(sequencer)
+ .Subscribe(static _ => { }, failure => observed = failure, static () => { }))
{
- await Assert.That(await WaitForAsync(observed.Task)).IsSameReferenceAs(error);
+ await Assert.That(observed).IsNull();
+ sequencer.RunPending();
}
+
+ await Assert.That(observed).IsSameReferenceAs(error);
}
/// Covers callback, forwarding, and stateful witness contracts.
@@ -1044,7 +1066,7 @@ private static async Task WaitForAsync(Task task)
var completed = await Task.WhenAny(task, timeout).ConfigureAwait(false);
if (completed == timeout)
{
- throw new TimeoutException("Timed out waiting for scheduled observer dispatch.");
+ throw new TimeoutException("Timed out waiting for the witness task to complete.");
}
await task.ConfigureAwait(false);