From 87981b230ceef87f7ed55ba4e5b88cd0ee0425c4 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Wed, 5 Aug 2026 22:10:07 -0700 Subject: [PATCH] Move change batching out of the delivery queue SharedDeliveryQueue took an optional callback that fired once per drain cycle, and CacheParentSubscription used it as its only emit point. That put a batching policy inside a class whose job is serialization, and it batched more than it should: DrainPending drains whatever is queued, so work another thread enqueued mid-drain landed in the same emission. CacheParentSubscription now tracks its own delivery frame. Each notification increments a depth counter, and the accumulated changes are emitted when it returns to zero. A child that emits synchronously during parent processing is delivered inline by the queue's reentrant path, so it nests inside the parent's frame rather than emitting separately. One upstream notification plus everything it triggers synchronously still produces one downstream changeset, which the six operators built on this class already relied on. What changes is that a second thread's work is no longer folded into the same emission. It gets its own frame. No lock is needed around the depth counter. The queue has already serialized delivery, so only one thread is ever inside these methods, and the queue's lock provides the barrier between drains on different threads. With that moved, the callback has no consumers, so the field, the overload that took it and the invoke site are all gone. SharedDeliveryQueue now only serializes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 9582bb33-26d3-4aa5-8dd7-57dc55304680 --- .../Internal/CacheParentSubscription.cs | 80 +++++++++++++++++-- .../Internal/SharedDeliveryQueue.cs | 20 +---- 2 files changed, 75 insertions(+), 25 deletions(-) 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; }