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