diff --git a/src/DynamicData/Internal/CacheParentSubscription.cs b/src/DynamicData/Internal/CacheParentSubscription.cs index 3a33143df..2960cd974 100644 --- a/src/DynamicData/Internal/CacheParentSubscription.cs +++ b/src/DynamicData/Internal/CacheParentSubscription.cs @@ -13,7 +13,8 @@ namespace DynamicData.Internal; /// when either the parent or child gets a new value. /// Uses a for serialization and lock-free delivery. /// Same-thread reentrant delivery preserves child-during-parent ordering. -/// OnDrainComplete calls EmitChanges after the outermost delivery, outside the lock. +/// Accumulated changes are emitted once per delivery frame, where a frame is one +/// notification plus anything delivered synchronously beneath it on the same thread. /// /// Type of the Parent ChangeSet. /// Type for the Parent ChangeSet Key. @@ -29,6 +30,7 @@ internal abstract class CacheParentSubscription _observer; private int _subscriptionCounter = 1; // Starts at 1 for the parent subscription + private int _frameDepth; private bool _isCompleted; private bool _hasTerminated; private bool _disposedValue; @@ -40,7 +42,7 @@ internal abstract class CacheParentSubscription observer) { _observer = observer; - _queue = new SharedDeliveryQueue(onDrainComplete: OnDrainComplete); + _queue = new SharedDeliveryQueue(); } /// @@ -76,9 +78,9 @@ protected void AddChildSubscription(IObservable observable, TKey parentK disposableContainer.Disposable = observable .Finally(CheckCompleted) .SubscribeSafe( - onNext: val => ChildOnNext(val, parentKey), + onNext: val => DeliverChild(val, parentKey), onError: TerminalError, - onCompleted: () => RemoveChildSubscription(parentKey)); + onCompleted: () => CompleteChild(parentKey)); } protected void RemoveChildSubscription(TKey parentKey) => _childSubscriptions.Remove(parentKey); @@ -88,9 +90,9 @@ protected void CreateParentSubscription(IObservable> s source .SynchronizeSafe(_queue) .SubscribeSafe( - onNext: ParentOnNext, + onNext: DeliverParent, onError: TerminalError, - onCompleted: CheckCompleted); + onCompleted: CompleteParent); protected virtual void Dispose(bool disposing) { @@ -116,8 +118,72 @@ protected virtual void Dispose(bool disposing) protected IObservable MakeChildObservable(IObservable observable) => observable.SynchronizeSafe(_queue); - private void OnDrainComplete() + private void DeliverParent(IChangeSet changes) { + ++_frameDepth; + try + { + ParentOnNext(changes); + } + finally + { + EndFrame(); + } + } + + private void DeliverChild(TChild child, TKey parentKey) + { + ++_frameDepth; + try + { + ChildOnNext(child, parentKey); + } + finally + { + EndFrame(); + } + } + + private void CompleteParent() + { + ++_frameDepth; + try + { + CheckCompleted(); + } + finally + { + EndFrame(); + } + } + + private void CompleteChild(TKey parentKey) + { + ++_frameDepth; + try + { + RemoveChildSubscription(parentKey); + } + finally + { + EndFrame(); + } + } + + /// + /// Closes the current delivery frame. Deliveries nested beneath this one, which the queue + /// runs inline on the same thread, close their own frame first and leave the emit to the + /// outermost, so one upstream notification and everything it triggers synchronously produce + /// a single downstream changeset. No lock is needed around the depth because the queue has + /// already serialized delivery. + /// + private void EndFrame() + { + if (--_frameDepth != 0) + { + return; + } + EmitChanges(_observer); if (Volatile.Read(ref _isCompleted) && !_hasTerminated) diff --git a/src/DynamicData/Internal/SharedDeliveryQueue.cs b/src/DynamicData/Internal/SharedDeliveryQueue.cs index 6eab28894..46545c8af 100644 --- a/src/DynamicData/Internal/SharedDeliveryQueue.cs +++ b/src/DynamicData/Internal/SharedDeliveryQueue.cs @@ -1,4 +1,4 @@ -// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved. +// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved. // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. @@ -16,7 +16,6 @@ namespace DynamicData.Internal; internal sealed class SharedDeliveryQueue : IDisposable { private readonly List _sources = []; - private readonly Action? _onDrainComplete; #if NET9_0_OR_GREATER private readonly Lock _gate; @@ -31,22 +30,12 @@ internal sealed class SharedDeliveryQueue : IDisposable /// Initializes a new instance of the class with its own internal lock. public SharedDeliveryQueue() - : this(onDrainComplete: null) - { - } - - /// - /// Initializes a new instance of the class with its own internal lock - /// and a callback that fires outside the lock after each drain cycle completes. - /// - public SharedDeliveryQueue(Action? onDrainComplete) { #if NET9_0_OR_GREATER _gate = new Lock(); #else _gate = new object(); #endif - _onDrainComplete = onDrainComplete; } #if NET9_0_OR_GREATER @@ -198,11 +187,6 @@ private void DrainAll() return; } - if (_onDrainComplete is not null) - { - _onDrainComplete(); - } - // Atomically check for pending items and release drain ownership // if empty. This closes the TOCTOU window: if we checked and released // in separate lock scopes, Thread B could enqueue between them, @@ -212,7 +196,7 @@ private void DrainAll() if (_activeBits.HasAny() && !_isTerminated) { - // Items arrived during _onDrainComplete. Loop back to drain them. + // Items arrived while we were delivering. Loop back to drain them. ExitLock(); continue; }