feat(primitives): add CancellationToken overloads for ToTask, First/Last terminals, and TakeUntil#111
Merged
Conversation
…st terminals, and TakeUntil Resolves #108. FirstAsync().ToTask(cts.Token) now compiles and cancels correctly. Adds Task<T>.ToTask(CancellationToken), FirstAsync(ct), FirstOrDefaultAsync(ct), LastAsync(ct), LastOrDefaultAsync(ct), and IObservable<T>.TakeUntil(CancellationToken) to both ReactiveUI.Primitives and ReactiveUI.Primitives.Reactive. The Async package already provided these overloads; TUnit coverage added for the new sync surface. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #111 +/- ##
==========================================
+ Coverage 90.96% 91.35% +0.38%
==========================================
Files 656 658 +2
Lines 20632 20564 -68
Branches 2489 2481 -8
==========================================
+ Hits 18768 18786 +18
+ Misses 1484 1441 -43
+ Partials 380 337 -43 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Restore the Codecov patch status configuration from main so PR #111 does not relax or mask coverage enforcement. Add focused TUnit coverage for the Codecov-reported SignalOperatorParityMixins cancellation gaps: FirstOrDefaultAsync pre-canceled token branches, First/FirstOrDefault range fast paths, synchronous first-value cleanup, LastOrDefault cancellation, task fault propagation, and LastOrDefaultAsync(CancellationToken) null validation. Verification: focused SignalOperatorParityMixinsTests passed 152 tests across net8.0/net9.0/net10.0/net11.0. Full ReactiveUI.Primitives.Tests passed 2340 tests across net8.0/net9.0/net10.0/net11.0. Net8 coverage output showed no missed coverage in SignalOperatorMixins.StatefulSignals.cs and no missed/partial coverage on the PR-added cancellation lines in SignalOperatorParityMixins.cs.
…tions - AwaitWithCancellationAsync abandoned the losing task on cancel; a source erroring afterwards faulted an unobserved task and raised UnobservedTaskException on the finalizer thread (5000/5000 in a stress run). It now rides Task.WaitAsync (BCL on net6+, repo polyfill on net4x) and attaches an OnlyOnFaulted observer to the abandoned task, which drops the hand-rolled TCS/Register/WhenAny machinery too. Post-fix stress: 0 unobserved exceptions. - FirstOrDefaultCoreAsync and Signal.ToTask registered the cancel callback before subscribing. Synchronous sources paid a pointless CTS register/dispose round trip per await, and the callback could read a stale null subscription on weak memory models because no happens-before edge covered the write, leaking the subscription. Both now subscribe first and only register while still pending, which also gives the callback a publication edge over the write. - The two duplicated First cores and the six copies of the RangeSignal fast path collapsed into one FirstCoreAsync; the token-less path had already drifted (plain TCS vs RunContinuationsAsynchronously). One core, async continuations, matching the 6-of-9 repo convention. - FirstOrDefaultAsync(CancellationToken) and LastOrDefaultAsync(CancellationToken) made existing '.FirstOrDefaultAsync(default!)' call sites ambiguous (CS0121, reproduced against the built package; the PR had to rewrite its own internal call site for the same reason). Marked both overloads with OverloadResolutionPriority(-1) so old code keeps binding to the defaultValue overloads. - CancellationSignal registered with a capturing closure through context-capturing Register; now UnsafeRegister with a static callback and the observer as state, matching the nine existing UnsafeRegister call sites. TakeUntil(token) drops from 472B to 384B per subscription, FirstAsync(token) from 520B to 456B and 128ns to 89ns. - Added a CancellationTokenRegistration.Unregister polyfill for net4x so pending-completion cleanup can release registrations without blocking on in-flight callbacks (CA1849).
- TerminalCancellationBenchmarks compares the token overloads (FirstAsync, LastOrDefaultAsync, ToTask shim, TakeUntil, cancel of a pending await) against System.Reactive and R3 with MemoryDiagnoser. - Rx rows call TaskObservableExtensions/Observable statics explicitly: instance-style calls resolved to the Primitives extension members and silently benchmarked us against ourselves.
Contributor
|
Reviewed this for concurrency/perf/allocs and pushed two commits with fixes and benchmarks. What changed and why:
Numbers (net10.0, short job, MemoryDiagnoser; before -> after for Primitives):
Concurrency stress (16 threads): 400k three-way FirstAsync/cancel/emit races - 0 wrong results, 0 hangs, 621 lock contentions total (all CTS/threadpool internals); 400k TakeUntil subscriptions on one shared long-lived token - 0 contentions, heap returns to baseline after GC, so registrations on an app-lifetime token don't accumulate. Remaining behavior note, no code change: All 10,324 tests pass including the new cancellation coverage. |
glennawatson
approved these changes
Jul 2, 2026
Add focused TUnit coverage for the Task<T>.ToTask() identity and null guard paths and the Task<T>.ToSignal() canceled and faulted branches. This improves coverage around the cancellation terminal work without touching codecov.yml or changing production behavior. Verification: affected SignalOperatorParityMixinsTests and SignalOperatorMixinsTests passed 156 tests across net8.0, net9.0, net10.0, and net11.0. Net8 coverage passed 586 tests. Full ReactiveUI.Primitives.Tests passed 2344 tests across net8.0, net9.0, net10.0, and net11.0.
- Signal.ToTask and FirstCoreAsync carried the same TCS/subscription/ registration scaffolding after the subscribe-first fix; SonarCloud flagged it at 19.1% duplication on new code. - TaskTerminalCompletion<T> now owns the task source, the subscription, the registration, and the subscribe-first/register-while-pending ordering; the terminals only supply their observer callbacks. The shared empty-source exception lives there too. - Costs one small state object on the pending path (456B -> 480B for FirstAsync(ct), still under the 520B pre-fix figure); the cancel path drops 40B.
- SyncLatest2 through SyncLatest16 are arity expansions of one shape and duplicative by construction; the existing exclusions covered the coordinator/signal/state/witness files but not the operators themselves.
Contributor
|
Follow-up push addressing the SonarCloud duplication gate (19.1% on new code):
|
glennawatson
changed the title
Add direct TUnit coverage for TaskTerminalCompletion<T> task exposure, already-resolved attachment disposal, and cancellation disposal behavior. This improves coverage for the refactored task-terminal helper without touching codecov.yml or adding timing-sensitive race tests. Verification: TaskTerminalCompletionTests plus affected terminal tests passed 12 focused tests across net8.0, net9.0, net10.0, and net11.0. Net8 coverage passed 589 tests and now leaves only the registration-race cleanup lines in TaskTerminalCompletion<T> missed. Full ReactiveUI.Primitives.Tests passed 2356 tests across net8.0, net9.0, net10.0, and net11.0.
… sinks - Empty, Return, Throw, and Defer each nested a byte-identical guard observer (~55 lines apiece, the top of the Sonar duplication list); one internal GuardedWitness<T> replaces all four. - CreateSignal and CreateSafeSignal nested the same sink differing only in whether a throwing downstream OnNext releases the subscription; one internal CreateSink<T> with a disposeOnNextThrow flag replaces both, delegating the cancel/stop plumbing to WitnessLifetime. - PublishSelectorWitness<T> was byte-identical to CreateWitness<T> and had a single internal call site; PublishSelectorSignal now uses CreateWitness and the duplicate type is removed with *REMOVED* PublicAPI entries (5.9 is still in alpha). Its behavior tests were retargeted to CreateWitness, which previously had no direct coverage. - Covers the extracted helpers directly: GuardedWitnessTests, CreateSinkTests, and TaskTerminalCompletion cases for the already-canceled-token, fault, empty-source, and resolve-releases-registration branches Chris flagged as missed.
- SynchronizeWitness (Lock gate) and SynchronizeObjectWitness (caller object gate) must stay separate types: lock over System.Threading.Lock and Monitor over a caller's object cannot share a generic gate without mixing lock kinds (CS9216 territory), so the three forwarding members are duplicative by construction.
Contributor
|
Pushed the pre-existing (main) duplication fixes here as well, since 5.9 is still in alpha:
2420 + 124 tests pass, zero warnings, full solution build clean. The |
- The extracted sinks stay useful beyond this package: downstream implementations (ReactiveUI itself among them) build the same terminate-and-release and create-factory observer shapes, so both types are public in Advanced alongside CreateWitness and the factory signals. - Public surface means real argument validation: both constructors now reject null observers (and CreateSink a null eager cancel), matching CreateWitness; PublicAPI entries added for all TFMs of both packages and null-argument tests added alongside.
- Releases here ship straight from main, so the Unshipped staging file is redundant bookkeeping; 2810 entries move into the Shipped baselines across all 44 TFM folders and the 252 *REMOVED* markers (PublishSelectorWitness and earlier removals) are applied to the Shipped lists they pointed at. - Unshipped files stay in place, empty, so the PublicApiAnalyzers wiring keeps working; the analyzer validated the merged baselines against the compiled surface via a clean full-solution build.
## Test coverage - Add primitive-vocabulary operator tests for cancelable enumerable conversion, loop Take(0), null/default wrapper branches, pending task conversion, and SwitchTo fallback paths. - Add parity operator tests for CancellationToken conversion branches, params Prepend, default comparer fallbacks, range timing shortcuts, and latest-fusion range branches. ## Verification - dotnet test "tests/ReactiveUI.Primitives.Tests/ReactiveUI.Primitives.Tests.csproj" -- --output Detailed - dotnet test "tests/ReactiveUI.Primitives.Tests/ReactiveUI.Primitives.Tests.csproj" -f net8.0 -- --coverage --coverage-output-format cobertura --coverage-output TestResults\pr111-after-public-branches-net8.cobertura.xml --output Detailed
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What kind of change does this PR introduce?
Feature (new cancellation-aware API surface), plus concurrency/allocation hardening of the new paths, benchmarks, duplication cleanup, and repo housekeeping (Sonar CPD config, PublicAPI baseline promotion).
What is the new behavior?
New API, compiled into both
ReactiveUI.Primitivesand theReactiveUI.Primitives.Reactiveshim (all 18 TFMs) viaPrimitives.Shared, bringing the sync package to parity withReactiveUI.Primitives.Async:Task<T>.ToTask(CancellationToken)- mirrors the task but transitions to canceled when the token fires first, so the Rx.NET migration patternFirstAsync().ToTask(cts.Token)compiles and behaves. Fast paths return the original task when the token cannot cancel or the task is already complete (BCLWaitAsyncsemantics). Built onTask.WaitAsync(net6+ BCL, repo polyfill on net4x), and a fault of the abandoned task after cancellation is observed so it can never surface asUnobservedTaskException.FirstAsync(ct),FirstOrDefaultAsync(ct),FirstOrDefaultAsync(T, ct)- cancellation disposes the subscription and cancels the task; pre-canceled tokens short-circuit. All six First overloads (token and token-less) run through oneFirstCoreAsyncholding theRangeSignalfast path once.LastAsync(ct),LastOrDefaultAsync(ct),LastOrDefaultAsync(T, ct)- route through the cancellation-awareSignal.ToTask.IObservable<T>.TakeUntil(CancellationToken)- completes gracefully on cancellation (Rx.NET parity); a non-cancelable token returns the source unchanged. The stop source registers withUnsafeRegisterand a static callback (no closure, no context capture).Cancellation plumbing shared by the terminals lives in one internal
TaskTerminalCompletion<T>: it subscribes before registering, so synchronous sources never pay for a CTS registration and the registration write is published to the thread that runs the cancel callback (closes a weak-memory-model window where a racing cancel could observe a stale null subscription and leak it).Duplication cleanup riding along (SonarCloud gated at 19.1% on new code, plus the pre-existing hotspots):
GuardedWitness<T>andCreateSink<T>are new public types inAdvanced- the byte-identical guard observers nested inEmptySignal/ReturnSignal/ThrowSignal/DeferSignaland the near-identical sinks inCreateSignal/CreateSafeSignalcollapse onto them. Public rather than internal so implementers layering on Primitives (ReactiveUI itself, Uno/Avalonia adapters) can reuse the sink shapes; going public brought full argument validation and direct unit tests.PublishSelectorWitness<T>is removed: it was byte-for-byte identical toCreateWitness<T>with one internal call site, which now usesCreateWitness. Its behavior tests were retargeted toCreateWitness, which previously had no direct coverage.Benchmarks (net10.0, MemoryDiagnoser) vs System.Reactive and R3:
Concurrency stress (16 threads): 400k FirstAsync/cancel/emit three-way races - 0 wrong results, 0 hangs, 621 lock contentions (all CTS/threadpool internals); 400k TakeUntil subscriptions on one shared long-lived token - 0 contentions and heap returns to baseline, so app-lifetime tokens do not accumulate registrations; 5000 cancel-then-fault iterations through the ToTask shim - 0 unobserved exceptions.
What is the current behavior?
FirstAsync()returnsTask<T>rather thanIObservable<T>, somessageStream.FirstAsync().ToTask(cts.Token)failed to compile and a pending first-value wait could not be canceled.TakeUntil(CancellationToken)was missing from the sync package entirely.Closes #108
What might this PR break?
FirstOrDefaultAsync(CancellationToken)/LastOrDefaultAsync(CancellationToken)overloads initially made existing.FirstOrDefaultAsync(default!)call sites ambiguous (CS0121); both are marked[OverloadResolutionPriority(-1)]so old code keeps binding to thedefaultValueoverloads.PublishSelectorWitness<T>is gone (shipped in 5.8; removed with*REMOVED*baseline entries while 5.9 is in alpha). Anyone who constructed it directly should switch to the identicalCreateWitness<T>.FirstAsync/FirstOrDefaultAsynccontinuations now run withRunContinuationsAsynchronously(previously inline on the producer's OnNext stack). This matches every other task terminal in the package and removes a reentrancy hazard, but continuation timing shifts for code that depended on inline resumption.completedTask.ToTask(canceledToken)returns the result (BCLWaitAsyncconvention) whileFirstAsync(canceledToken)cancels up front (Rx convention). The task-side shim also cannot unsubscribe the source on cancel - onlyFirstAsync(ct)can - so prefer passing the token into the terminal.Checklist
mainbranchAdditional information
TerminalCancellationBenchmarksadded to the benchmarks project; the Rx rows callObservable/TaskObservableExtensionsstatics explicitly because instance-style calls resolve to the Primitives extension members and would benchmark us against ourselves.GuardedWitnessTests,CreateSinkTests,TaskTerminalCompletionTests(including the already-canceled-token, fault, empty-source, and resolve-releases-registration branches).**/Operators/SyncLatest*.cs(arity expansions of one shape, not dedupable without codegen) and theSynchronizeWitness/SynchronizeObjectWitnesspair (a shared generic gate would mixlockoverSystem.Threading.LockwithMonitorover a caller's object).*REMOVED*markers applied): releases ship straight from main here, so the Unshipped staging step was redundant bookkeeping. Unshipped files remain, empty, so the analyzer wiring is unchanged; new API added after this PR stages there as usual.CancellationTokenRegistration.Unregisterpolyfill added for net4x so pending-completion cleanup can release registrations without blocking on in-flight callbacks.