diff --git a/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs b/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs index 443c63be..b5af1941 100644 --- a/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs +++ b/src/ReactiveUI.Binding.Shared/Observables/PropertyObservable.cs @@ -70,10 +70,16 @@ internal sealed class Subscription : IDisposable private readonly EqualityComparer _comparer; /// - /// Counts the emits asked for and not yet served. The thread that raises it from zero serves every - /// emit asked for while it runs, so emits never overlap and no thread waits for another. + /// Serializes emits across threads. A change raised on another thread waits for the running emit and is + /// then delivered on the thread that raised it. /// - private int _pendingEmits; + private readonly Lock _gate = new(); + + /// Whether an emit is running. Read and written only under . + private bool _emitting; + + /// Whether the emitting thread raised another change from inside its own emit. + private bool _changedDuringEmit; /// The downstream observer. Set to on disposal. private IObserver? _observer; @@ -146,46 +152,49 @@ private void OnPropertyChanged(object? sender, PropertyChangedEventArgs e) /// Reads the current property value and forwards it downstream when the distinct gate allows. /// - /// A call made while another emit runs returns at once. The running emit reads the property again - /// before it stops, so the value is never lost and never stale. A call from inside the downstream - /// observer is delivered after that observer returns. A throw clears the count, so the next change - /// still emits. + /// A call from another thread waits for the running emit, then emits on its own thread. A call the + /// emitting thread makes from inside its own emit, from the getter or the downstream observer, returns + /// at once; the running emit reads the property again after the observer returns. /// private void EmitCurrent() { - if (Interlocked.Increment(ref _pendingEmits) != 1) - { - return; - } - - var unserved = 1; - try + lock (_gate) { - do + if (_emitting) { - var observer = Volatile.Read(ref _observer); - if (observer is null) - { - return; - } - - var value = _parent._getter(_parent._source); + _changedDuringEmit = true; + return; + } - if (!_parent._distinctUntilChanged || !_hasValue || !_comparer.Equals(value!, _lastValue!)) + _emitting = true; + try + { + do { - _lastValue = value; - _hasValue = true; - observer.OnNext(value!); + _changedDuringEmit = false; + + var observer = Volatile.Read(ref _observer); + if (observer is null) + { + return; + } + + var value = _parent._getter(_parent._source); + + if (!_parent._distinctUntilChanged || !_hasValue || !_comparer.Equals(value!, _lastValue!)) + { + _lastValue = value; + _hasValue = true; + observer.OnNext(value!); + } } - - unserved = Interlocked.Add(ref _pendingEmits, -unserved); + while (_changedDuringEmit); + } + finally + { + _emitting = false; + _changedDuringEmit = false; } - while (unserved != 0); - } - catch - { - Volatile.Write(ref _pendingEmits, 0); - throw; } } } diff --git a/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs b/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs index 5c4f95b2..4811b66d 100644 --- a/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs +++ b/src/tests/ReactiveUI.Binding.Tests/Observables/PropertyObservableInitialEmitSerializationTests.cs @@ -32,11 +32,16 @@ public class PropertyObservableInitialEmitSerializationTests private const string ThirdName = "Carol"; /// - /// How long the initial emit waits for a competing thread to finish its write. The subscription never - /// makes that thread wait, so the join ends as soon as the write does; the bound only turns a deadlock - /// into a failure. + /// How long the initial emit gives a competing thread to finish its write. The competing thread waits + /// for the initial emit, so the join always expires; one that did not wait finishes in microseconds. /// - private const int CompetitorTimeoutMilliseconds = 10_000; + private const int InterleaveWindowMilliseconds = 500; + + /// + /// How long a competing emit waits for the subscribing thread to return from subscribe. The bound only + /// turns a subscribing thread held by that emit into a failure rather than a hang. + /// + private const int SubscribeReturnTimeoutMilliseconds = 10_000; /// /// Subscriptions the unforced sweep builds. Sized from measurement: against unserialized code this @@ -79,7 +84,7 @@ public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadWhileAttaching_E /// constructor performs for its initial emit itself raises /// , so the handler runs part-way through /// construction on the subscribing thread. This also pins that a notification raised on the emitting - /// thread is handed to the running emit, since waiting for that emit would deadlock here rather than fail. + /// thread is handed to the running emit rather than nested inside it. /// /// A representing the asynchronous unit test. [Test] @@ -114,18 +119,33 @@ public async Task Subscribe_PropertyChangedRaisedReentrantlyDuringInitialRead_Em } /// - /// A competing handler never runs inside the initial emit and never waits for it. A thread that - /// writes while the initial emit is on the stack finishes at once, and the running emit delivers its - /// value after the initial one. Waiting on the observer for such a thread is therefore not a deadlock. + /// A thread that writes while the initial emit is on the stack waits for that emit, then delivers its + /// value itself. The subscribing thread returns from subscribe without delivering the other thread's + /// value, so a subscriber runs on the thread that raised the change. /// /// A representing the asynchronous unit test. [Test] - public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadDuringInitialEmit_NeitherOverlapsNorBlocksThatThread() + public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadDuringInitialEmit_WaitsThenEmitsOnThatThread() { var source = new HookedViewModel { Name = InitialName }; using var competitorStarted = new ManualResetEventSlim(false); + using var subscribeReturned = new ManualResetEventSlim(false); Thread? competitor = null; - var competitorFinished = false; + var competitorFinishedDuringInitialEmit = true; + var replacementReadThreadId = 0; + var subscribeReturnedBeforeReplacementRead = false; + + string? ReadAndRecordThread(INotifyPropertyChanged instance) + { + var name = ((HookedViewModel)instance).Name; + if (name == ReplacementName) + { + replacementReadThreadId = Environment.CurrentManagedThreadId; + subscribeReturnedBeforeReplacementRead = subscribeReturned.Wait(SubscribeReturnTimeoutMilliseconds); + } + + return name; + } // Runs from inside the downstream call of the initial emit, which is the window no other emit may enter. var recorder = new EmissionRecorder @@ -140,22 +160,25 @@ public async Task Subscribe_PropertyChangedRaisedOnAnotherThreadDuringInitialEmi competitor.Start(); competitorStarted.Wait(); - competitorFinished = competitor.Join(CompetitorTimeoutMilliseconds); + competitorFinishedDuringInitialEmit = competitor.Join(InterleaveWindowMilliseconds); }, }; var observable = new PropertyObservable( source, nameof(HookedViewModel.Name), - static x => ((HookedViewModel)x).Name, + ReadAndRecordThread, distinctUntilChanged: true); using (observable.Subscribe(recorder)) { + subscribeReturned.Set(); competitor!.Join(); await AssertNoErrors(recorder); - await Assert.That(competitorFinished).IsTrue(); + await Assert.That(competitorFinishedDuringInitialEmit).IsFalse(); + await Assert.That(subscribeReturnedBeforeReplacementRead).IsTrue(); + await Assert.That(replacementReadThreadId).IsEqualTo(competitor.ManagedThreadId); await Assert.That(recorder.MaxConcurrentEmissions).IsEqualTo(1); await AssertSequence(recorder.Snapshot(), InitialName, ReplacementName); }