diff --git a/src/DynamicData/Internal/CacheParentSubscription.cs b/src/DynamicData/Internal/CacheParentSubscription.cs index 3a33143d..e817c9fb 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,55 @@ protected virtual void Dispose(bool disposing) protected IObservable MakeChildObservable(IObservable observable) => observable.SynchronizeSafe(_queue); - private void OnDrainComplete() + private void DeliverParent(IChangeSet changes) { + using var frame = BeginFrame(); + ParentOnNext(changes); + } + + private void DeliverChild(TChild child, TKey parentKey) + { + using var frame = BeginFrame(); + ChildOnNext(child, parentKey); + } + + private void CompleteParent() + { + using var frame = BeginFrame(); + CheckCompleted(); + } + + private void CompleteChild(TKey parentKey) + { + using var frame = BeginFrame(); + RemoveChildSubscription(parentKey); + } + + /// + /// Opens a delivery frame that stays open until the returned is disposed. + /// Deliveries nested beneath this one, which the queue runs inline on the same thread, open and close + /// their own frame 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. + /// + /// A tracker that closes the frame when disposed. + private FrameTracker BeginFrame() + { + ++_frameDepth; + return new FrameTracker(this); + } + + /// + /// Closes the current delivery frame, emitting the accumulated changes only when the outermost + /// frame closes. + /// + private void EndFrame() + { + if (--_frameDepth != 0) + { + return; + } + EmitChanges(_observer); if (Volatile.Read(ref _isCompleted) && !_hasTerminated) @@ -142,4 +191,15 @@ private void CheckCompleted() Debug.Assert(_subscriptionCounter >= 0, "Should never be negative"); } + + /// + /// Closes the delivery frame opened by when disposed, so a frame can be + /// scoped with instead of pairing the calls by hand. + /// + /// The subscription whose frame is being tracked. + private readonly struct FrameTracker(CacheParentSubscription owner) : IDisposable + { + /// + public void Dispose() => owner.EndFrame(); + } } diff --git a/src/DynamicData/Internal/SharedDeliveryQueue.cs b/src/DynamicData/Internal/SharedDeliveryQueue.cs index 3ca1f2b2..cdcd762f 100644 --- a/src/DynamicData/Internal/SharedDeliveryQueue.cs +++ b/src/DynamicData/Internal/SharedDeliveryQueue.cs @@ -29,8 +29,6 @@ internal sealed class SharedDeliveryQueue : IDisposable /// private readonly Queue _order = new(); - private readonly Action? _onDrainComplete; - #if NET9_0_OR_GREATER private readonly Lock _gate; #else @@ -42,22 +40,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 @@ -167,8 +155,6 @@ private void DrainAll() return; } - _onDrainComplete?.Invoke(); - // Atomically re-check for work and release ownership if there is none. Checking // and releasing in separate lock scopes would let a producer enqueue in between, // see that a drain is in progress, and rely on us to deliver an item we never saw.