From b07c03c807f00c31d64914ecb9893ec56747e12f Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Tue, 4 Aug 2026 21:21:59 -0700 Subject: [PATCH] Deliver queued notifications in the order they were received SharedDeliveryQueue partitions pending notifications into a typed sub-queue per source, and the drain loop picked the next one with Bitset.FindHighest(). Selection was therefore by sub-queue index rather than by arrival, so notifications from two sources could be delivered in an order they were not received in. With receipt order A1, A2, Bx, delivery came out A1, Bx, A2. The sub-queues are still typed, so the payloads are still held as structs and queuing one still costs no allocation. What changes is that a second queue now records which source each pending notification came from, in arrival order, and the drain loop follows that instead of scanning a bitset. The entries are sub-queue references that already exist, so recording the order does not allocate either. Highest-index-first was there so a child sub-queue drained before a parent's delivery could dispose it. That is no longer needed. A disposed sub-queue drops its pending notifications, and its leftover order entries are skipped when the drain reaches them, which is the same outcome by a shorter route. This also removes Bitset, which had no other caller, along with the sub-queue index bookkeeping and the compaction pass that existed to keep the bitset dense. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 9582bb33-26d3-4aa5-8dd7-57dc55304680 --- .../Internal/BitsetFixture.cs | 449 ------------------ .../Internal/SharedDeliveryQueueFixture.cs | 153 ++++++ src/DynamicData/Internal/Bitset.cs | 242 ---------- .../Internal/SharedDeliveryQueue.cs | 329 ++++--------- 4 files changed, 257 insertions(+), 916 deletions(-) delete mode 100644 src/DynamicData.Tests/Internal/BitsetFixture.cs delete mode 100644 src/DynamicData/Internal/Bitset.cs diff --git a/src/DynamicData.Tests/Internal/BitsetFixture.cs b/src/DynamicData.Tests/Internal/BitsetFixture.cs deleted file mode 100644 index fd33ce09e..000000000 --- a/src/DynamicData.Tests/Internal/BitsetFixture.cs +++ /dev/null @@ -1,449 +0,0 @@ -// 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. - -using DynamicData.Internal; -using FluentAssertions; -using Xunit; - -namespace DynamicData.Tests.Internal; - -public class BitsetFixture -{ - [Fact] - public void NewBitset_HasNoActiveBits() - { - // Arrange - var bitset = new Bitset(); - - // Act (no action, testing initial state) - - // Assert - bitset.HasAny().Should().BeFalse(); - bitset.Count.Should().Be(0); - bitset.FindHighest().Should().Be(-1); - bitset.FindLowest().Should().Be(-1); - } - - [Fact] - public void Set_SingleBit_IsActiveAndQueryable() - { - // Arrange - var bitset = new Bitset(); - - // Act - bitset.Set(0); - - // Assert - bitset.IsSet(0).Should().BeTrue(); - bitset.HasAny().Should().BeTrue(); - bitset.Count.Should().Be(1); - bitset.FindHighest().Should().Be(0); - bitset.FindLowest().Should().Be(0); - } - - [Fact] - public void Set_MultipleBits_FindHighestReturnsLargest() - { - // Arrange - var bitset = new Bitset(); - - // Act - bitset.Set(3); - bitset.Set(17); - bitset.Set(42); - - // Assert - bitset.IsSet(3).Should().BeTrue(); - bitset.IsSet(17).Should().BeTrue(); - bitset.IsSet(42).Should().BeTrue(); - bitset.IsSet(4).Should().BeFalse(); - bitset.Count.Should().Be(3); - bitset.FindHighest().Should().Be(42); - bitset.FindLowest().Should().Be(3); - } - - [Fact] - public void Clear_RemovesBitAndUpdatesQueries() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(5); - bitset.Set(10); - - // Act - bitset.Clear(10); - - // Assert - bitset.IsSet(10).Should().BeFalse(); - bitset.IsSet(5).Should().BeTrue(); - bitset.HasAny().Should().BeTrue(); - bitset.Count.Should().Be(1); - bitset.FindHighest().Should().Be(5); - } - - [Fact] - public void Clear_LastBit_HasAnyReturnsFalse() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(7); - - // Act - bitset.Clear(7); - - // Assert - bitset.IsSet(7).Should().BeFalse(); - bitset.HasAny().Should().BeFalse(); - bitset.Count.Should().Be(0); - bitset.FindHighest().Should().Be(-1); - } - - [Fact] - public void Set_IsIdempotent() - { - // Arrange - var bitset = new Bitset(); - - // Act - bitset.Set(3); - bitset.Set(3); - bitset.Set(3); - - // Assert - bitset.IsSet(3).Should().BeTrue(); - bitset.FindHighest().Should().Be(3); - bitset.Clear(3); - bitset.HasAny().Should().BeFalse("single Clear should undo any number of Sets"); - } - - [Fact] - public void Clear_IsIdempotent() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(5); - - // Act - bitset.Clear(5); - bitset.Clear(5); - bitset.Clear(5); - - // Assert - bitset.IsSet(5).Should().BeFalse(); - bitset.HasAny().Should().BeFalse(); - } - - [Fact] - public void ClearAll_RemovesEverything() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(0); - bitset.Set(31); - bitset.Set(63); - - // Act - bitset.ClearAll(); - - // Assert - bitset.IsSet(0).Should().BeFalse(); - bitset.IsSet(31).Should().BeFalse(); - bitset.IsSet(63).Should().BeFalse(); - bitset.HasAny().Should().BeFalse(); - bitset.Count.Should().Be(0); - bitset.FindHighest().Should().Be(-1); - bitset.FindLowest().Should().Be(-1); - } - - [Fact] - public void Set_BeyondInitialCapacity_GrowsAutomatically() - { - // Arrange - var bitset = new Bitset(); - - // Act - bitset.Set(100); - - // Assert - bitset.IsSet(100).Should().BeTrue(); - bitset.HasAny().Should().BeTrue(); - bitset.Count.Should().Be(1); - bitset.FindHighest().Should().Be(100); - bitset.FindLowest().Should().Be(100); - } - - [Fact] - public void MultipleWords_FindHighest_ReturnsCorrectIndex() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(10); // word 0 - bitset.Set(70); // word 1 - bitset.Set(130); // word 2 - - // Act - var highest = bitset.FindHighest(); - var lowest = bitset.FindLowest(); - - // Assert - highest.Should().Be(130); - lowest.Should().Be(10); - } - - [Fact] - public void MultipleWords_ClearHighest_UpdatesFindHighest() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(10); - bitset.Set(70); - bitset.Set(130); - - // Act - bitset.Clear(130); - - // Assert - bitset.FindHighest().Should().Be(70); - bitset.IsSet(130).Should().BeFalse(); - } - - [Fact] - public void MultipleWords_ClearLowest_UpdatesFindLowest() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(10); - bitset.Set(70); - bitset.Set(130); - - // Act - bitset.Clear(10); - - // Assert - bitset.FindLowest().Should().Be(70); - bitset.IsSet(10).Should().BeFalse(); - } - - [Fact] - public void MultipleWords_ClearAll_ClearsAllWords() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(0); - bitset.Set(64); - bitset.Set(128); - bitset.Set(192); - - // Act - bitset.ClearAll(); - - // Assert - bitset.HasAny().Should().BeFalse(); - bitset.Count.Should().Be(0); - bitset.FindHighest().Should().Be(-1); - } - - [Fact] - public void WordBoundary_Bit63And64() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(63); // last bit of word 0 - bitset.Set(64); // first bit of word 1 - - // Act - bitset.Clear(64); - - // Assert - bitset.IsSet(63).Should().BeTrue(); - bitset.IsSet(64).Should().BeFalse(); - bitset.FindHighest().Should().Be(63); - bitset.FindLowest().Should().Be(63); - } - - [Fact] - public void FindHighest_WithOnlyLowBitsSet() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(200); // force multi-word allocation - bitset.Clear(200); - bitset.Set(0); - bitset.Set(1); - bitset.Set(2); - - // Act - var highest = bitset.FindHighest(); - var lowest = bitset.FindLowest(); - - // Assert - highest.Should().Be(2); - lowest.Should().Be(0); - } - - [Theory] - [InlineData(0)] - [InlineData(1)] - [InlineData(31)] - [InlineData(32)] - [InlineData(63)] - public void SingleBit_RoundTrips(int index) - { - // Arrange - var bitset = new Bitset(); - - // Act - bitset.Set(index); - - // Assert - bitset.IsSet(index).Should().BeTrue(); - bitset.HasAny().Should().BeTrue(); - bitset.FindHighest().Should().Be(index); - bitset.FindLowest().Should().Be(index); - - bitset.Clear(index); - bitset.IsSet(index).Should().BeFalse(); - bitset.HasAny().Should().BeFalse(); - } - - [Fact] - public void SetAndClear_ManyBits_CountStaysConsistent() - { - // Arrange - var bitset = new Bitset(); - for (var i = 0; i < 100; i++) - { - bitset.Set(i); - } - - // Act - for (var i = 0; i < 99; i++) - { - bitset.Clear(i); - } - - // Assert - bitset.HasAny().Should().BeTrue(); - bitset.Count.Should().Be(1); - bitset.FindHighest().Should().Be(99); - bitset.FindLowest().Should().Be(99); - bitset.IsSet(99).Should().BeTrue(); - bitset.IsSet(0).Should().BeFalse(); - - bitset.Clear(99); - bitset.HasAny().Should().BeFalse(); - bitset.Count.Should().Be(0); - } - - [Fact] - public void IsSet_BeyondCapacity_ReturnsFalse() - { - // Arrange - var bitset = new Bitset(); - - // Act - var result = bitset.IsSet(500); - - // Assert - result.Should().BeFalse(); - } - - [Fact] - public void Clear_BeyondCapacity_DoesNotThrow() - { - // Arrange - var bitset = new Bitset(); - - // Act - var act = () => bitset.Clear(500); - - // Assert - act.Should().NotThrow(); - bitset.HasAny().Should().BeFalse(); - } - - [Fact] - public void Compact_ShrinksThenSetGrowsBack() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(200); - bitset.Clear(200); - - // Act - bitset.Compact(); - bitset.Set(200); - - // Assert - bitset.IsSet(200).Should().BeTrue(); - bitset.FindHighest().Should().Be(200); - } - - [Fact] - public void Compact_RetainsActiveBits() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(5); - bitset.Set(200); - bitset.Clear(200); - - // Act - bitset.Compact(); - - // Assert - bitset.IsSet(5).Should().BeTrue(); - bitset.HasAny().Should().BeTrue(); - bitset.FindHighest().Should().Be(5); - } - - [Fact] - public void Compact_AllEmpty_ShrinkToMinimum() - { - // Arrange - var bitset = new Bitset(); - bitset.Set(200); - bitset.ClearAll(); - - // Act - bitset.Compact(); - - // Assert - bitset.HasAny().Should().BeFalse(); - bitset.FindHighest().Should().Be(-1); - } - - [Fact] - public void Count_TracksSetBitsAccurately() - { - // Arrange - var bitset = new Bitset(); - - // Act - bitset.Set(1); - bitset.Set(10); - bitset.Set(100); - - // Assert - bitset.Count.Should().Be(3); - - // Act (idempotent set) - bitset.Set(10); - - // Assert - bitset.Count.Should().Be(3, "idempotent Set should not increment"); - - // Act (clear one) - bitset.Clear(10); - - // Assert - bitset.Count.Should().Be(2); - - // Act (clear all) - bitset.ClearAll(); - - // Assert - bitset.Count.Should().Be(0); - } -} diff --git a/src/DynamicData.Tests/Internal/SharedDeliveryQueueFixture.cs b/src/DynamicData.Tests/Internal/SharedDeliveryQueueFixture.cs index e67444cb6..583005489 100644 --- a/src/DynamicData.Tests/Internal/SharedDeliveryQueueFixture.cs +++ b/src/DynamicData.Tests/Internal/SharedDeliveryQueueFixture.cs @@ -175,6 +175,159 @@ public async Task ConcurrentMultiSourceDelivery() } } + [Fact] + public void ReceiptOrderIsPreservedAcrossSubQueues() + { + var queue = new SharedDeliveryQueue(_gate); + var delivered = new List(); + var blockFirst = new ManualResetEventSlim(false); + var firstIsDelivering = new ManualResetEventSlim(false); + + var sub1 = queue.CreateQueue(new TestObserver(i => + { + lock (delivered) { delivered.Add($"int:{i}"); } + + if (i == 1) + { + firstIsDelivering.Set(); + blockFirst.Wait(); + } + })); + + var sub2 = queue.CreateQueue(new TestObserver(s => + { + lock (delivered) { delivered.Add($"str:{s}"); } + })); + + // Park a drain part-way through, so the notifications below get queued rather than + // delivered inline. + var drainer = Task.Run(() => + { + using var scope = sub1.AcquireLock(); + scope.EnqueueNext(1); + }); + + firstIsDelivering.Wait(TimeSpan.FromSeconds(5)); + + using (var scope = sub1.AcquireLock()) + { + scope.EnqueueNext(2); + } + + using (var scope = sub2.AcquireLock()) + { + scope.EnqueueNext("hello"); + } + + blockFirst.Set(); + drainer.Wait(TimeSpan.FromSeconds(5)); + + delivered.Should().Equal(new[] { "int:1", "int:2", "str:hello" }, "delivery should follow the order the notifications were received, not the order the sub-queues were created"); + } + + [Fact] + public void InterleavedSubQueuesDeliverInReceiptOrder() + { + var queue = new SharedDeliveryQueue(_gate); + var delivered = new List(); + var block = new ManualResetEventSlim(false); + var parked = new ManualResetEventSlim(false); + + var sub1 = queue.CreateQueue(new TestObserver(i => + { + lock (delivered) { delivered.Add($"int:{i}"); } + + if (i == 0) + { + parked.Set(); + block.Wait(); + } + })); + + var sub2 = queue.CreateQueue(new TestObserver(s => + { + lock (delivered) { delivered.Add($"str:{s}"); } + })); + + var drainer = Task.Run(() => + { + using var scope = sub1.AcquireLock(); + scope.EnqueueNext(0); + }); + + parked.Wait(TimeSpan.FromSeconds(5)); + + using (var scope = sub2.AcquireLock()) + { + scope.EnqueueNext("a"); + } + + using (var scope = sub1.AcquireLock()) + { + scope.EnqueueNext(2); + } + + using (var scope = sub2.AcquireLock()) + { + scope.EnqueueNext("b"); + } + + using (var scope = sub1.AcquireLock()) + { + scope.EnqueueNext(4); + } + + block.Set(); + drainer.Wait(TimeSpan.FromSeconds(5)); + + delivered.Should().Equal("int:0", "str:a", "int:2", "str:b", "int:4"); + } + + [Fact] + public void DisposedSubQueueDoesNotDeliverQueuedItems() + { + var queue = new SharedDeliveryQueue(_gate); + var delivered = new List(); + var block = new ManualResetEventSlim(false); + var parked = new ManualResetEventSlim(false); + + var sub1 = queue.CreateQueue(new TestObserver(i => + { + lock (delivered) { delivered.Add($"int:{i}"); } + + if (i == 0) + { + parked.Set(); + block.Wait(); + } + })); + + var sub2 = queue.CreateQueue(new TestObserver(s => + { + lock (delivered) { delivered.Add($"str:{s}"); } + })); + + var drainer = Task.Run(() => + { + using var scope = sub1.AcquireLock(); + scope.EnqueueNext(0); + }); + + parked.Wait(TimeSpan.FromSeconds(5)); + + using (var scope = sub2.AcquireLock()) + { + scope.EnqueueNext("dropped"); + } + + sub2.Dispose(); + + block.Set(); + drainer.Wait(TimeSpan.FromSeconds(5)); + + delivered.Should().Equal(new[] { "int:0" }, "a disposed sub-queue should not deliver what it had queued"); + } + private sealed class TestObserver(Action onNext) : IObserver { public Exception? Error { get; private set; } diff --git a/src/DynamicData/Internal/Bitset.cs b/src/DynamicData/Internal/Bitset.cs deleted file mode 100644 index d28e4a669..000000000 --- a/src/DynamicData/Internal/Bitset.cs +++ /dev/null @@ -1,242 +0,0 @@ -// 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. - -#if NETCOREAPP3_0_OR_GREATER -using System.Numerics; -#endif -using System.Runtime.CompilerServices; - -namespace DynamicData.Internal; - -/// -/// -/// A compact bitset backed by a [] array that tracks active slots -/// (e.g., sub-queues with pending items). Provides O(1) set/clear operations and fast -/// highest/lowest-bit lookup via hardware intrinsics (LZCNT/TZCNT) for LIFO/FIFO iteration. -/// -/// -/// Each holds 64 bits. A slot index maps to a word and bit position: -/// word = index / 64, bit = index % 64. The backing array grows automatically -/// via but never shrinks. Callers that need compaction should -/// create a new instance. -/// -/// -internal struct Bitset -{ - private const int BitsPerWord = 64; - private const int WordShift = 6; - private const int BitMask = BitsPerWord - 1; - - private long[] _words; - - /// Initializes a new instance of the struct with capacity for 64 slots. - public Bitset() => _words = [0]; - - /// Gets the number of bits currently set. - public int Count { get; private set; } - - /// Sets the bit at , marking the slot as active. - /// The zero-based slot index. - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Set(int index) - { - EnsureCapacity(index); - ref var word = ref _words[index >> WordShift]; - var mask = 1L << (index & BitMask); - if ((word & mask) == 0) - { - word |= mask; - Count++; - } - } - - /// Clears the bit at , marking the slot as inactive. - /// The zero-based slot index. - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Clear(int index) - { - if ((index >> WordShift) >= _words.Length) - { - return; - } - - ref var word = ref _words[index >> WordShift]; - var mask = 1L << (index & BitMask); - if ((word & mask) != 0) - { - word &= ~mask; - Count--; - } - } - - /// Returns if the bit at is set. - /// The zero-based slot index. - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public readonly bool IsSet(int index) - { - var wordIndex = index >> WordShift; - if (wordIndex >= _words.Length) - { - return false; - } - - return (_words[wordIndex] & (1L << (index & BitMask))) != 0; - } - - /// - /// - /// Finds the highest set bit (for LIFO iteration) and returns its index, - /// or -1 if no bits are set. - /// - /// - /// Scans words from highest to lowest. Within each word, the leading zero count - /// intrinsic (LZCNT on x86, CLZ on ARM) locates the most significant set bit - /// in a single CPU instruction. - /// - /// - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public readonly int FindHighest() - { - var words = _words; - if (words.Length == 1) - { - var w0 = words[0]; - if (w0 == 0) return -1; -#if NETCOREAPP3_0_OR_GREATER - return 63 - BitOperations.LeadingZeroCount((ulong)w0); -#else - return HighestSetBit(w0); -#endif - } - - for (var w = words.Length - 1; w >= 0; w--) - { - var word = words[w]; - if (word != 0) - { -#if NETCOREAPP3_0_OR_GREATER - var bitIndex = 63 - BitOperations.LeadingZeroCount((ulong)word); -#else - var bitIndex = HighestSetBit(word); -#endif - return (w << WordShift) | bitIndex; - } - } - - return -1; - } - - /// - /// - /// Finds the lowest set bit (for FIFO iteration) and returns its index, - /// or -1 if no bits are set. - /// - /// - /// Scans words from lowest to highest. Within each word, the trailing zero count - /// intrinsic (TZCNT on x86, CTZ on ARM) locates the least significant set bit - /// in a single CPU instruction. - /// - /// - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public readonly int FindLowest() - { - var words = _words; - for (var w = 0; w < words.Length; w++) - { - var word = words[w]; - if (word != 0) - { -#if NETCOREAPP3_0_OR_GREATER - var bitIndex = BitOperations.TrailingZeroCount((ulong)word); -#else - var bitIndex = LowestSetBit(word); -#endif - return (w << WordShift) | bitIndex; - } - } - - return -1; - } - - /// Returns if any bit is set. - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public readonly bool HasAny() => Count > 0; - - /// Clears all bits in every word. - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void ClearAll() - { - if (_words.Length == 1) - { - _words[0] = 0; - } - else - { - Array.Clear(_words, 0, _words.Length); - } - - Count = 0; - } - - /// Grows the backing array if needed so that is addressable. Called implicitly by . - /// The zero-based slot index that must be representable. - [MethodImpl(MethodImplOptions.AggressiveInlining)] - private void EnsureCapacity(int index) - { - var wordIndex = index >> WordShift; - if (wordIndex >= _words.Length) - { - Array.Resize(ref _words, wordIndex + 1); - } - } - - /// - /// Shrinks the backing array to the minimum size needed to represent all set bits. - /// Reclaims memory from words that are entirely zero at the end of the array. - /// Always retains at least one word. - /// - public void Compact() - { - var needed = 1; - for (var w = _words.Length - 1; w >= 1; w--) - { - if (_words[w] != 0) - { - needed = w + 1; - break; - } - } - - if (needed < _words.Length) - { - Array.Resize(ref _words, needed); - } - } - -#if !NETCOREAPP3_0_OR_GREATER - private static int HighestSetBit(long value) - { - var bit = 0; - for (var v = (ulong)value; v > 1; v >>= 1) - { - bit++; - } - - return bit; - } - - private static int LowestSetBit(long value) - { - var bit = 0; - var v = (ulong)value; - while ((v & 1) == 0) - { - v >>= 1; - bit++; - } - - return bit; - } -#endif -} diff --git a/src/DynamicData/Internal/SharedDeliveryQueue.cs b/src/DynamicData/Internal/SharedDeliveryQueue.cs index 6eab28894..3ca1f2b2c 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. @@ -7,15 +7,28 @@ namespace DynamicData.Internal; /// -/// A type-erased delivery queue that serializes delivery across multiple sources -/// with different item types. Each source gets a typed -/// via . A single drain loop delivers items from all -/// sub-queues outside the lock, one item per iteration. An -/// tracks which sub-queues have pending items, replacing O(N) scans with O(1) lookups. +/// A delivery queue that serializes delivery across multiple sources with different +/// item types. Each source gets a typed via +/// , which holds that source's notifications without +/// boxing them. A single order queue records which source each pending notification +/// came from, so delivery follows the order notifications were received rather than +/// the order the sources happen to be registered in. +/// +/// The lock is never held while an observer runs. A producer that arrives while +/// another thread is delivering enqueues and returns rather than blocking, so a +/// pipeline that crosses into another cache during delivery cannot deadlock against +/// a producer on this one. +/// /// internal sealed class SharedDeliveryQueue : IDisposable { - private readonly List _sources = []; + /// + /// One entry per pending notification, identifying the source it belongs to, + /// in the order the notifications were received. The payloads themselves stay + /// in their typed sub-queues, so recording the order costs no allocation. + /// + private readonly Queue _order = new(); + private readonly Action? _onDrainComplete; #if NET9_0_OR_GREATER @@ -24,8 +37,6 @@ internal sealed class SharedDeliveryQueue : IDisposable private readonly object _gate; #endif - private Bitset _activeBits = new(); - private int _deadCount; private int _drainThreadId = -1; private volatile bool _isTerminated; @@ -57,30 +68,31 @@ public SharedDeliveryQueue(Action? onDrainComplete) public SharedDeliveryQueue(object gate) => _gate = gate; #endif - /// Gets whether this queue has been terminated. + /// Gets a value indicating whether this queue has been terminated. public bool IsTerminated { [MethodImpl(MethodImplOptions.AggressiveInlining)] get => _isTerminated; } + /// Creates a typed sub-queue bound to the specified observer. + public DeliverySubQueue CreateQueue(IObserver observer) => new(this, observer); + + /// Acquires the gate for read-only inspection. Does not trigger delivery on dispose. + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public ReadOnlyScopedAccess AcquireReadLock() => new(this); + /// - /// Terminates the queue (rejecting further enqueues) and blocks until - /// any in-flight delivery has completed. After this returns, no more - /// observer callbacks will fire. Safe to call from within a delivery - /// callback (skips the spin-wait if the calling thread is the deliverer). + /// Terminates the queue, rejecting further enqueues, and blocks until any in-flight + /// delivery has completed. After this returns, no more observer callbacks will fire. + /// Safe to call from within a delivery callback, which skips the spin-wait. /// - private void EnsureDeliveryComplete() + public void Dispose() { EnterLock(); _isTerminated = true; - _activeBits.ClearAll(); - - foreach (var s in _sources) - { - s.Clear(); - } + _order.Clear(); if (_drainThreadId == Environment.CurrentManagedThreadId) { @@ -95,42 +107,9 @@ private void EnsureDeliveryComplete() spinner.SpinOnce(); } - /// Disposes the queue by calling . - public void Dispose() => EnsureDeliveryComplete(); - - /// Creates a typed sub-queue bound to the specified observer. - public DeliverySubQueue CreateQueue(IObserver observer) - { - EnterLock(); - try - { - var index = _sources.Count; - var queue = new DeliverySubQueue(this, observer, index); - _sources.Add(queue); - - return queue; - } - finally - { - ExitLock(); - } - } - - /// Acquires the gate for read-only inspection. Does not trigger delivery on dispose. + /// Records that the given source has one more notification pending. Must be called under the lock. [MethodImpl(MethodImplOptions.AggressiveInlining)] - public ReadOnlyScopedAccess AcquireReadLock() => new(this); - - /// Called by a sub-queue when it is disposed. Clears its active bit and tracks dead slots. - [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal void NotifyQueueRemoved(int index) - { - _activeBits.Clear(index); - _deadCount++; - } - - /// Sets the active bit for a sub-queue when an item is enqueued. - [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal void SetActive(int index) => _activeBits.Set(index); + internal void EnqueueOrder(DrainableBase source) => _order.Enqueue(source); #if NET9_0_OR_GREATER [MethodImpl(MethodImplOptions.AggressiveInlining)] @@ -150,10 +129,10 @@ internal void ExitLockAndDrain() { var currentThreadId = Environment.CurrentManagedThreadId; - // Same-thread reentrant: if we're already draining on this thread, - // deliver newly enqueued items inline. This preserves the same delivery - // order as Synchronize(lock): child items emitted synchronously during - // parent delivery are delivered immediately, not deferred. + // Same-thread reentrant: if we're already draining on this thread, deliver newly + // enqueued items inline. This preserves the same delivery order as Synchronize(lock): + // child items emitted synchronously during parent delivery are delivered immediately, + // not deferred. if (_drainThreadId == currentThreadId) { ExitLock(); @@ -162,7 +141,7 @@ internal void ExitLockAndDrain() } var shouldDrain = false; - if (_drainThreadId == -1 && !_isTerminated && _activeBits.HasAny()) + if (_drainThreadId == -1 && !_isTerminated && _order.Count != 0) { _drainThreadId = currentThreadId; shouldDrain = true; @@ -184,70 +163,48 @@ private void DrainAll() { if (!DrainPending()) { - EnterLock(); - try - { - _drainThreadId = -1; - CompactIfNeeded(); - } - finally - { - ExitLock(); - } - + ReleaseDrainOwnership(); return; } - if (_onDrainComplete is not null) - { - _onDrainComplete(); - } + _onDrainComplete?.Invoke(); - // 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, - // see _drainThreadId != -1, and rely on us to drain, but we'd exit - // without draining Thread B's item. + // 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. EnterLock(); - if (_activeBits.HasAny() && !_isTerminated) + if (_order.Count != 0 && !_isTerminated) { - // Items arrived during _onDrainComplete. Loop back to drain them. ExitLock(); continue; } - try - { - _drainThreadId = -1; - CompactIfNeeded(); - } - finally - { - ExitLock(); - } - + _drainThreadId = -1; + ExitLock(); return; } } catch { - EnterLock(); - _drainThreadId = -1; - ExitLock(); + ReleaseDrainOwnership(); throw; } } + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private void ReleaseDrainOwnership() + { + EnterLock(); + _drainThreadId = -1; + ExitLock(); + } + /// - /// Delivers all pending items from all sub-queues, one at a time. - /// Sub-queues are found via the active bitset using LZCNT (highest-index first - /// for LIFO ordering). When one sub-queue's delivery can dispose another - /// (parent disposing a child), the child must drain first to prevent pending - /// child notifications from being silently lost. Newer sub-queues are always - /// children of older ones, so LIFO provides this guarantee. + /// Delivers pending notifications, one at a time, in the order they were received. + /// Each is delivered outside the lock. /// - /// True if completed normally; false if an error terminated the queue. + /// True if the queue drained normally; false if it was terminated. private bool DrainPending() { while (true) @@ -260,83 +217,39 @@ private bool DrainPending() return false; } - var sourceIndex = _activeBits.FindHighest(); - if (sourceIndex < 0) + if (_order.Count == 0) { ExitLock(); return true; } - var active = _sources[sourceIndex]; - var isError = active.StageNext(); + var source = _order.Dequeue(); - // If sub-queue is now empty, clear its active bit immediately. - if (!active.HasItems) + // The source may have been disposed since this entry was recorded, which drops + // its pending notifications. Skip the stale entry and take the next one. + if (!source.TryStageNext()) { - _activeBits.Clear(sourceIndex); + ExitLock(); + continue; } + var isError = source.IsStagedError; + ExitLock(); - active.DeliverStaged(); + source.DeliverStaged(); if (isError) { EnterLock(); _isTerminated = true; - _activeBits.ClearAll(); - foreach (var s in _sources) - { - s.Clear(); - } - + _order.Clear(); ExitLock(); return false; } } } - /// - /// Compacts the source list when dead slots exceed 50% of capacity. - /// Rebuilds indices and the bitset atomically. Must be called under lock. - /// - private void CompactIfNeeded() - { - if (_deadCount == 0 || _deadCount <= _sources.Count / 2) - { - return; - } - - _deadCount = 0; - _activeBits.ClearAll(); - - var writeIndex = 0; - for (var readIndex = 0; readIndex < _sources.Count; readIndex++) - { - var source = _sources[readIndex]; - if (!source.IsRemoved) - { - source.Index = writeIndex; - _sources[writeIndex] = source; - - if (source.HasItems) - { - SetActive(writeIndex); - } - - writeIndex++; - } - } - - // Remove trailing dead entries - if (writeIndex < _sources.Count) - { - _sources.RemoveRange(writeIndex, _sources.Count - writeIndex); - } - - _activeBits.Compact(); - } - /// Read-only scoped access. Disposing releases the gate without triggering delivery. public ref struct ReadOnlyScopedAccess { @@ -349,19 +262,11 @@ internal ReadOnlyScopedAccess(SharedDeliveryQueue owner) owner.EnterLock(); } - /// Gets whether any sub-queue has pending items. + /// Gets a value indicating whether any notification is pending or in flight. public readonly bool HasPending { [MethodImpl(MethodImplOptions.AggressiveInlining)] - get - { - if (_owner is null) - { - return false; - } - - return _owner._drainThreadId != -1 || _owner._activeBits.HasAny(); - } + get => _owner is not null && (_owner._drainThreadId != -1 || _owner._order.Count != 0); } /// Releases the gate lock. @@ -380,32 +285,23 @@ public void Dispose() } } -/// Base class for typed sub-queues. Enables devirtualization in the drain loop. +/// Base class for typed sub-queues, so the drain loop can hold them without knowing their element type. internal abstract class DrainableBase { - /// Gets whether this sub-queue has items. - internal abstract bool HasItems { get; } - - /// Gets whether this sub-queue has been removed and should be skipped/compacted. - internal abstract bool IsRemoved { get; } - - /// Gets or sets the stable index in the parent's source list. - internal abstract int Index { get; set; } + /// Gets a value indicating whether the staged notification is an error. + internal abstract bool IsStagedError { get; } - /// Dequeues the next item into staging. Returns true if error (terminal). - /// True if the staged item is an error notification. - internal abstract bool StageNext(); + /// Moves the next pending notification into staging. Returns false if there is nothing to stage. + internal abstract bool TryStageNext(); - /// Delivers the staged item to the observer. + /// Delivers the staged notification to the observer. internal abstract void DeliverStaged(); - - /// Clears all pending items. - internal abstract void Clear(); } /// -/// A typed sub-queue. All enqueue access goes through -/// which acquires the parent's lock. +/// A typed sub-queue. Notifications are held as structs, so queuing one costs no +/// allocation. All enqueue access goes through , which +/// acquires the parent's lock. /// internal sealed class DeliverySubQueue : DrainableBase, IObserver, IDisposable { @@ -413,59 +309,40 @@ internal sealed class DeliverySubQueue : DrainableBase, IObserver, IDispos private readonly SharedDeliveryQueue _parent; private readonly IObserver _observer; private Notification _staged; - private int _index; private bool _isRemoved; - internal DeliverySubQueue(SharedDeliveryQueue parent, IObserver observer, int index) + internal DeliverySubQueue(SharedDeliveryQueue parent, IObserver observer) { _parent = parent; _observer = observer; - _index = index; - } - - /// - internal override bool HasItems - { - [MethodImpl(MethodImplOptions.AggressiveInlining)] - get => !_isRemoved && _items.Count > 0; } /// - internal override bool IsRemoved + internal override bool IsStagedError { [MethodImpl(MethodImplOptions.AggressiveInlining)] - get => _isRemoved; + get => _staged.IsError; } - /// - internal override int Index - { - [MethodImpl(MethodImplOptions.AggressiveInlining)] - get => _index; - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - set => _index = value; - } - - /// Acquires the parent gate. Disposing releases the lock and triggers drain. + /// Acquires the parent gate. Disposing releases the lock and triggers delivery. [MethodImpl(MethodImplOptions.AggressiveInlining)] public ScopedAccess AcquireLock() => new(this); - /// Enqueues an OnNext notification via the lock, then drains. + /// Enqueues an OnNext notification via the lock, then delivers. public void OnNext(T value) { using var scope = AcquireLock(); scope.EnqueueNext(value); } - /// Enqueues an OnError notification via the lock, then drains. + /// Enqueues an OnError notification via the lock, then delivers. public void OnError(Exception error) { using var scope = AcquireLock(); scope.EnqueueError(error); } - /// Enqueues an OnCompleted notification via the lock, then drains. + /// Enqueues an OnCompleted notification via the lock, then delivers. public void OnCompleted() { using var scope = AcquireLock(); @@ -473,8 +350,9 @@ public void OnCompleted() } /// - /// Marks this sub-queue as removed under the parent lock, clearing pending items - /// and notifying the parent for GC compaction. Idempotent. + /// Marks this sub-queue as removed under the parent lock and drops its pending + /// notifications. Any order entries left behind are skipped when the drain reaches + /// them. Idempotent. /// public void Dispose() { @@ -488,7 +366,6 @@ public void Dispose() _isRemoved = true; _items.Clear(); - _parent.NotifyQueueRemoved(_index); } finally { @@ -497,10 +374,15 @@ public void Dispose() } /// - internal override bool StageNext() + internal override bool TryStageNext() { + if (_isRemoved || _items.Count == 0) + { + return false; + } + _staged = _items.Dequeue(); - return _staged.IsError; + return true; } /// @@ -510,9 +392,6 @@ internal override void DeliverStaged() _staged = default; } - /// - internal override void Clear() => _items.Clear(); - [MethodImpl(MethodImplOptions.AggressiveInlining)] private void EnqueueItem(Notification item) { @@ -522,10 +401,10 @@ private void EnqueueItem(Notification item) } _items.Enqueue(item); - _parent.SetActive(_index); + _parent.EnqueueOrder(this); } - /// Scoped access for enqueueing items. Acquires the parent's gate lock. + /// Scoped access for enqueueing notifications. Acquires the parent's gate lock. public ref struct ScopedAccess { private DeliverySubQueue? _owner; @@ -537,7 +416,7 @@ internal ScopedAccess(DeliverySubQueue owner) owner._parent.EnterLock(); } - /// Enqueues an OnNext item. + /// Enqueues an OnNext notification. [MethodImpl(MethodImplOptions.AggressiveInlining)] public readonly void EnqueueNext(T item) => _owner?.EnqueueItem(Notification.CreateNext(item)); @@ -549,7 +428,7 @@ internal ScopedAccess(DeliverySubQueue owner) [MethodImpl(MethodImplOptions.AggressiveInlining)] public readonly void EnqueueCompleted() => _owner?.EnqueueItem(Notification.CreateCompleted()); - /// Releases the parent gate lock and delivers pending items. + /// Releases the parent gate lock and delivers pending notifications. public void Dispose() { var owner = _owner; @@ -562,4 +441,4 @@ public void Dispose() owner._parent.ExitLockAndDrain(); } } -} +} \ No newline at end of file