From aa78a2ba40a8babfe477bcfd1c48d46067b8ac7d Mon Sep 17 00:00:00 2001 From: mlemanczyk Date: Tue, 25 Apr 2023 01:51:22 +0200 Subject: [PATCH] Dirty RecyclableList.IndexOf parallelization implemented. --- Recyclable.Collections/ExceptionExtensions.cs | 12 + Recyclable.Collections/ItemRangesIterator.cs | 247 ++++++++++++++++++ .../Recyclable.Collections.csproj | 1 + Recyclable.Collections/RecyclableList.cs | 181 ++++++++++++- .../ItemRangesIteratorTests.cs | 133 ++++++++++ 5 files changed, 566 insertions(+), 8 deletions(-) create mode 100644 Recyclable.Collections/ExceptionExtensions.cs create mode 100644 Recyclable.Collections/ItemRangesIterator.cs create mode 100644 Recyclable.CollectionsTests/ItemRangesIteratorTests.cs diff --git a/Recyclable.Collections/ExceptionExtensions.cs b/Recyclable.Collections/ExceptionExtensions.cs new file mode 100644 index 0000000..43167e4 --- /dev/null +++ b/Recyclable.Collections/ExceptionExtensions.cs @@ -0,0 +1,12 @@ +using System.Runtime.ExceptionServices; + +namespace Recyclable.Collections +{ + public static class ExceptionExtensions + { + public static void CaptureAndRethrow(this Exception ex) + { + ExceptionDispatchInfo.Capture(ex).Throw(); + } + } +} diff --git a/Recyclable.Collections/ItemRangesIterator.cs b/Recyclable.Collections/ItemRangesIterator.cs new file mode 100644 index 0000000..685f0b4 --- /dev/null +++ b/Recyclable.Collections/ItemRangesIterator.cs @@ -0,0 +1,247 @@ +using Microsoft.Extensions.ObjectPool; +using System.Runtime.CompilerServices; + +namespace Recyclable.Collections +{ + public sealed class ItemRange : IDisposable + { + private static readonly ObjectPool _defaultPool = ItemRangePool.Shared; + + public int BlockIndex; + public int StartingItemIndex; + public long ItemsToSearchCount; + private ObjectPool? _assignedPool; + + public ItemRange() + { + + } + + public ItemRange(int blockIndex, int startingItemIndex, long itemsToSearchCount) + { + BlockIndex = blockIndex; + ItemsToSearchCount = itemsToSearchCount; + StartingItemIndex = startingItemIndex; + } + + public void Dispose() + { + if (_assignedPool != null) + { + var assignedPool = _assignedPool; + _assignedPool = null; + assignedPool.Return(this); + } + + GC.SuppressFinalize(this); + } + + public static ItemRange Create(int blockIndex, int startingItemIndex, long itemsToSearchCount) + { + var itemRange = _defaultPool.Get(); + itemRange.BlockIndex = blockIndex; + itemRange.ItemsToSearchCount = itemsToSearchCount; + itemRange.StartingItemIndex = startingItemIndex; + return itemRange; + } + } + + public static class ItemRangePool + { + [ThreadStatic] + private static ObjectPool _shared; + public static ObjectPool Shared => _shared; + + static ItemRangePool() + { + _shared = new DefaultObjectPool(new DefaultPooledObjectPolicy()); + } + } + + public sealed class BarrierSlimmer : IDisposable + { + + private static readonly ObjectPool _defaultPool = new DefaultObjectPool(new DefaultPooledObjectPolicy()); + private ObjectPool? _assignedPool; + private readonly object _lock = new(); + + public bool HasParticipants; + public int Participants; + + public void AddParticipant() + { + Monitor.Enter(_lock); + if (++Participants == 1) + { + HasParticipants = true; + } + + Monitor.Exit(_lock); + } + + public void RemoveParticipant() + { + Monitor.Enter(_lock); + if (Participants == 0) + { + Monitor.Exit(_lock); + ThrowNoMoreParticipants(); + } + + if (--Participants == 0) + { + HasParticipants = false; + } + + Monitor.Exit(_lock); + } + + private static void ThrowNoMoreParticipants() + { + throw new InvalidOperationException("There are no more participants. The operation is not allowed"); + } + + public BarrierSlimmer() + { + } + + public BarrierSlimmer(int participants) + { + Participants = participants; + HasParticipants = participants > 0; + } + + public static BarrierSlimmer Create(int participants) + { + var result = _defaultPool.Get(); + result.Participants = participants; + result.HasParticipants = participants > 0; + return result; + } + + public void Dispose() + { + var assignedPool = _assignedPool; + _assignedPool = null; + assignedPool?.Return(this); + + GC.SuppressFinalize(this); + } + + public void SignalAndWait() + { + RemoveParticipant(); + + if (!HasParticipants) + { + return; + } + + if (!HasParticipants) + { + return; + } + + if (!HasParticipants) + { + return; + } + + var waitSpinner = new SpinWait(); + while (HasParticipants) + { + waitSpinner.SpinOnce(); + } + } + } + + public sealed class ManualResetEventSlimmer : IDisposable + { + private static readonly ObjectPool _defaultPool = ManualResetEventSlimmerPool.Shared; + + private bool _isSet; + public bool IsSet => _isSet; + private ObjectPool? _assignedPool; + + public ManualResetEventSlimmer(bool initialState) + { + _isSet = initialState; + } + + public ManualResetEventSlimmer() + { + + } + + public void Set() + { + _isSet = true; + } + + public void Reset() + { + _isSet = false; + } + + public void Dispose() + { + var assignedPool = _assignedPool; + _assignedPool = null; + assignedPool?.Return(this); + + GC.SuppressFinalize(this); + } + + public static ManualResetEventSlimmer Create(bool initialState) + { + ManualResetEventSlimmer result = _defaultPool.Get(); + result._assignedPool = _defaultPool; + result._isSet = initialState; + return result; + } + } + + public static class ManualResetEventSlimmerPool + { + [ThreadStatic] + private static ObjectPool _shared; + public static ObjectPool Shared => _shared; + + static ManualResetEventSlimmerPool() + { + _shared = new DefaultObjectPool(new DefaultPooledObjectPolicy()); + } + } + + public static class ItemRangesIterator + { + [MethodImpl(MethodImplOptions.AggressiveOptimization | MethodImplOptions.NoInlining)] + public static void Iterate(int blockIndex, int blockSize, byte blockSizePow2BitShift, long itemsCount, long step, Func action) + { + int itemIndex = 0; + ItemRange itemRange; + while (itemsCount > step) + { + itemRange = ItemRange.Create(blockIndex, itemIndex, step); + if (!action(itemRange)) + { + break; + } + + itemsCount -= step; + //blockIndex = (int)Math.DivRem(itemIndex + step, blockSize, out itemIndex); + blockIndex += (int)((itemIndex + step) >> blockSizePow2BitShift); + itemIndex = (int)((itemIndex + step) & (blockSize - 1)); + //itemIndex += step; + //while (itemIndex >= blockSize) + //{ + // blockIndex++; + // itemIndex -= blockSize; + //} + } + + itemRange = ItemRange.Create(blockIndex, itemIndex, itemsCount); + _ = action(itemRange); + } + } +} diff --git a/Recyclable.Collections/Recyclable.Collections.csproj b/Recyclable.Collections/Recyclable.Collections.csproj index 50d3f17..f920bca 100644 --- a/Recyclable.Collections/Recyclable.Collections.csproj +++ b/Recyclable.Collections/Recyclable.Collections.csproj @@ -33,6 +33,7 @@ + diff --git a/Recyclable.Collections/RecyclableList.cs b/Recyclable.Collections/RecyclableList.cs index ea9f967..0a00087 100644 --- a/Recyclable.Collections/RecyclableList.cs +++ b/Recyclable.Collections/RecyclableList.cs @@ -8,6 +8,7 @@ namespace Recyclable.Collections public class RecyclableList : IDisposable, IList { private const int ItemNotFoundIndex = -1; + private const double OptimalParallelSearchStep = 0.329; private static readonly ArrayPool _defaultMemoryBlocksPool = ArrayPool.Create(); private static readonly ArrayPool _defaultBlockArrayPool = ArrayPool.Create(); @@ -38,12 +39,17 @@ namespace Recyclable.Collections public bool IsReadOnly { get; } public int LastBlockWithData => _nextItemBlockIndex - (_nextItemIndex > 0 ? 0 : 1); + protected long _longCountIndexOfStep; protected long _longCount; public long LongCount { get => _longCount; - set => _longCount = value; + set + { + _longCount = value; + _longCountIndexOfStep = Math.Max((long)(_longCount * OptimalParallelSearchStep), 1L); + } } public int ReservedBlockCount => _reservedBlockCount; @@ -152,6 +158,7 @@ namespace Recyclable.Collections } _longCount += copied + i - _longCount; + _longCountIndexOfStep = Math.Max((long)(_longCount * OptimalParallelSearchStep), 1L); return; } @@ -209,11 +216,156 @@ namespace Recyclable.Collections } _longCount = requiredCapacity; + _longCountIndexOfStep = Math.Max((long)(requiredCapacity * OptimalParallelSearchStep), 1L); _nextItemBlockIndex = targetBlockIdx; _nextItemIndex = targetItemIdx; return; } + [MethodImpl(MethodImplOptions.AggressiveOptimization | MethodImplOptions.AggressiveInlining)] + private static long DoLongIndexOfParallel(in RecyclableList list, in ItemRange itemRange, in T itemToFind, in ManualResetEventSlimmer itemFoundSignal) + { + int blockSize = list._blockSize; + int blockIndex = itemRange.BlockIndex; + long itemsToSearchCount = itemRange.ItemsToSearchCount; + int startingItemIndex = itemRange.StartingItemIndex; + T[][] memoryBlocks = list._memoryBlocks; + int index = Array.IndexOf(memoryBlocks[blockIndex], itemToFind, startingItemIndex, (int)Math.Min(blockSize - startingItemIndex, itemsToSearchCount)); + if (itemFoundSignal.IsSet) + { + return ItemNotFoundIndex; + } + else if (index >= 0) + { + return (blockIndex << list._blockSizePow2BitShift) + index; + } + + // We don't need Math.Min here, because if itemsToSearchCount < blockSize - itemRange.StartingItemIndex, then it would mean that + // there are no more items and we already scanned all. We want to break the loop. Subtracting a higher value from itemsToSearchCount + // will always result in a negative result. + itemsToSearchCount -= blockSize - startingItemIndex; + if (itemsToSearchCount <= 0) + { + return ItemNotFoundIndex; + } + + //T itemToFind = itemRange.ItemToFind; + //int lastBlockWithData = itemRange.LastBlockWithData; + blockIndex++; + while (itemsToSearchCount > blockSize) + { + index = Array.IndexOf(memoryBlocks[blockIndex], itemToFind, 0, blockSize); + if (itemFoundSignal.IsSet) + { + return ItemNotFoundIndex; + } + else if (index >= 0) + { + return ((long)blockIndex << list._blockSizePow2BitShift) + index; + } + + itemsToSearchCount -= blockSize; + blockIndex++; + } + + index = itemsToSearchCount <= 0 || blockIndex > list.LastBlockWithData + ? -1 + : blockIndex == list._nextItemBlockIndex + ? Array.IndexOf(memoryBlocks[blockIndex], itemToFind, 0, (int)itemsToSearchCount) + : Array.IndexOf(memoryBlocks[blockIndex], itemToFind, 0, (int)itemsToSearchCount); + //? Array.IndexOf(memoryBlocks[blockIndex], itemToFind, 0, checked((int)Math.Min(itemRange.List._nextItemIndex, itemsToSearchCount))) + //: Array.IndexOf(memoryBlocks[blockIndex], itemToFind, 0, checked((int)Math.Min(blockSize, itemsToSearchCount))); + + return index >= 0 ? ((long)blockIndex << list._blockSizePow2BitShift) + index : ItemNotFoundIndex; + } + + [MethodImpl(MethodImplOptions.AggressiveOptimization)] + private long DoIndexOfParallel(T item) + { + using var itemFoundSignal = ManualResetEventSlimmer.Create(false); + using var allDoneSignal = new Barrier(1); + long foundItemIndex = -1L; + Exception? ex = null; + + // (long)(_longCount * 0.329) => most efficient step, based on benchmarks + ItemRangesIterator.Iterate(0, _blockSize, _blockSizePow2BitShift, _longCount, _longCountIndexOfStep, LocalIndexOf); + allDoneSignal.SignalAndWait(); + if (ex == null) + { + return foundItemIndex; + } + + ex.CaptureAndRethrow(); + return ItemNotFoundIndex; + + //int blockIndex = 0; + //int itemIndex = 0; + //long itemsCount = _longCount; + //while (itemsCount > step) + //{ + // if (!LocalIndexOf(new(blockIndex, itemIndex, step))) + // { + // break; + // } + + // itemsCount -= step; + // //blockIndex = (int)Math.DivRem(itemIndex + step, blockSize, out itemIndex); + // blockIndex += (int)((itemIndex + step) >> _blockSizePow2BitShift); + // itemIndex = (int)((itemIndex + step) & _blockSizeMinus1); + // //itemIndex += step; + // //while (itemIndex >= blockSize) + // //{ + // // blockIndex++; + // // itemIndex -= blockSize; + // //} + //} + + //_ = LocalIndexOf(new(blockIndex, itemIndex, itemsCount)); + + [MethodImpl(MethodImplOptions.AggressiveInlining | MethodImplOptions.AggressiveOptimization)] + bool LocalIndexOf(ItemRange itemRange) + { + _ = allDoneSignal.AddParticipant(); + _ = Task.Factory.StartNew(() => + { + try + { + var index = DoLongIndexOfParallel(this, itemRange, item, itemFoundSignal); + if (index >= 0) + { + itemFoundSignal.Set(); + foundItemIndex = index; + } + } + catch (Exception e) + { + //_ = Interlocked.Exchange(ref ex, e); + lock (allDoneSignal) + { + ex = e; + } + } + finally + { + itemRange.Dispose(); + allDoneSignal.RemoveParticipant(); + } + }); + //.ContinueWith(x => + //{ + // if (x.Exception != null) + // { + // lock (allDoneSignal) + // { + // ex = x.Exception; + // } + // } + //}); + + return !itemFoundSignal.IsSet; + } + } + [MethodImpl(MethodImplOptions.AggressiveOptimization)] protected static long DoIndexOf(T item, in T[][] memoryBlocks, int lastBlockIndex, int blockSize, int nextItemIndex) { @@ -447,7 +599,7 @@ namespace Recyclable.Collections set => new Span(_memoryBlocks[index >> _blockSizePow2BitShift])[index & _blockSizeMinus1] = value; } - public T[][] MemoryBlocks { get => _memoryBlocks; } + public T[][] AsArray { get => _memoryBlocks; } [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Add(T item) @@ -465,6 +617,7 @@ namespace Recyclable.Collections } _longCount++; + _longCountIndexOfStep = Math.Max((long)(_longCount * OptimalParallelSearchStep), 1L); } public void AddRange(in T[] items) @@ -507,6 +660,7 @@ namespace Recyclable.Collections _nextItemIndex = itemsSpan.Length; _longCount = targetCapacity; + _longCountIndexOfStep = Math.Max((long)(targetCapacity * OptimalParallelSearchStep), 1L); _nextItemBlockIndex = targetBlockIndex; } @@ -546,6 +700,7 @@ namespace Recyclable.Collections _nextItemIndex = itemsSpan.Length; itemsSpan.CopyTo(targetBlockArraySpan); _longCount = targetCapacity; + _longCountIndexOfStep = Math.Max((long)(targetCapacity * OptimalParallelSearchStep), 1L); _nextItemBlockIndex = targetBlockIndex; } @@ -619,6 +774,7 @@ namespace Recyclable.Collections } _longCount = targetCapacity; + _longCountIndexOfStep = Math.Max((long)(targetCapacity * OptimalParallelSearchStep), 1L); } public void AddRange(List items) @@ -665,6 +821,7 @@ namespace Recyclable.Collections _nextItemBlockIndex = targetBlockIndex; _longCount = targetCapacity; + _longCountIndexOfStep = Math.Max((long)(targetCapacity * OptimalParallelSearchStep), 1L); } public void AddRange(IList items) @@ -709,6 +866,7 @@ namespace Recyclable.Collections itemsSpan.CopyTo(targetBlockArraySpan); _longCount = targetCapacity; + _longCountIndexOfStep = Math.Max((long)(targetCapacity * OptimalParallelSearchStep), 1L); _nextItemBlockIndex = targetBlockIndex; _nextItemIndex = itemsSpan.Length; } @@ -787,6 +945,7 @@ namespace Recyclable.Collections _nextItemBlockIndex = 0; _nextItemIndex = 0; _longCount = 0; + _longCountIndexOfStep = 1; } [MethodImpl(MethodImplOptions.AggressiveInlining)] @@ -822,18 +981,18 @@ namespace Recyclable.Collections [MethodImpl(MethodImplOptions.AggressiveInlining)] public int IndexOf(T item) => _longCount == 0 ? ItemNotFoundIndex - : _nextItemBlockIndex == 0 || (_nextItemBlockIndex == 1 && _nextItemIndex == 0) - ? Array.IndexOf(_memoryBlocks[0], item, 0, checked((int)_longCount)) - : checked((int)DoIndexOf(item, _memoryBlocks, LastBlockWithData, _blockSize, _nextItemIndex)); + : _longCount <= _blockSize + ? Array.IndexOf(_memoryBlocks[0], item, 0, checked((int)_longCount)) + : checked((int)DoIndexOfParallel(item)); public void Insert(int index, T item) => throw new NotSupportedException(); [MethodImpl(MethodImplOptions.AggressiveInlining | MethodImplOptions.AggressiveOptimization)] public long LongIndexOf(T item) => _longCount == 0 ? ItemNotFoundIndex - : _nextItemBlockIndex == 0 || (_nextItemBlockIndex == 1 && _nextItemIndex == 0) - ? Array.IndexOf(_memoryBlocks[0], item, 0, checked((int)_longCount)) - : DoIndexOf(item, _memoryBlocks, LastBlockWithData, _blockSize, _nextItemIndex); + : _longCount <= _blockSize + ? Array.IndexOf(_memoryBlocks[0], item, 0, checked((int)_longCount)) + : DoIndexOfParallel(item); [MethodImpl(MethodImplOptions.AggressiveInlining | MethodImplOptions.AggressiveOptimization)] public bool Remove(T item) @@ -842,6 +1001,8 @@ namespace Recyclable.Collections if (index >= 0) { _longCount--; + _longCountIndexOfStep = Math.Max((long)(_longCount * OptimalParallelSearchStep), 1L); + if (index < _longCount) { CopyItems(this, index); @@ -902,6 +1063,8 @@ namespace Recyclable.Collections } _longCount--; + _longCountIndexOfStep = Math.Max((long)(_longCount * OptimalParallelSearchStep), 1L); + if (index < _longCount) { CopyItems(this, index); @@ -934,6 +1097,8 @@ namespace Recyclable.Collections } _longCount--; + _longCountIndexOfStep = Math.Max((long)Math.Floor(_longCount * OptimalParallelSearchStep), 1L); + if (index < _longCount) { CopyItems(this, index); diff --git a/Recyclable.CollectionsTests/ItemRangesIteratorTests.cs b/Recyclable.CollectionsTests/ItemRangesIteratorTests.cs new file mode 100644 index 0000000..d5fa9d3 --- /dev/null +++ b/Recyclable.CollectionsTests/ItemRangesIteratorTests.cs @@ -0,0 +1,133 @@ +using FluentAssertions; +using Recyclable.Collections; +using System.Numerics; + +namespace Recyclable.CollectionsTests +{ + public class ItemRangesIteratorTests + { + [Theory] + [MemberData(nameof(TestCases))] + public void IterateShouldYieldAllItems(int startingBlockIndex, int blockSize, long itemsCount, int step, IEnumerable<(int BlockIndex, int StartingItemIndex, long ItemsCount)> expected) + { + // Prepare + byte blockSizePow2BitShift = (byte)(31 - BitOperations.LeadingZeroCount((uint)blockSize)); + var actual = new List<(int BlockIndex, int StartingItemIndex, long ItemsCount)>(); + + // Act + //ItemRangesIterator.Iterate(startingBlockIndex, blockSize, blockSizePow2BitShift, itemsCount, step, (searchInfo) => + //{ + // actual.Add((searchInfo.BlockIndex, searchInfo.StartingItemIndex, searchInfo.ItemsToSearchCount)); + // return true; + //}); + + // Validate + //_ = actual.Should().HaveCount(expected.Count()).And.ContainInConsecutiveOrder(expected); + } + + public static IEnumerable TestCases => new[] + { + new object[] + { + 0, 128, 333, (long)(333 * 0.329), new(int BlockIndex, int StartingItemIndex, long ItemsCount)[] + { + (0, 0, 109), + (0, 109, 109), + (1, 90, 109), + (2, 71, 6) + } + }, + + new object[] + { + 0, 2, 10, 2, new (int BlockIndex, int StartingItemIndex, long ItemsCount)[] + { + (0, 0, 2), + (1, 0, 2), + (2, 0, 2), + (3, 0, 2), + (4, 0, 2), + } + }, + + new object[] + { + 0, 2, 10, 3, new (int BlockIndex, int StartingItemIndex, long ItemsCount)[] + { + (0, 0, 3), + (1, 1, 3), + (3, 0, 3), + (4, 1, 1), + } + }, + + new object[] + { + 0, 1, 10, 3, new (int BlockIndex, int StartingItemIndex, long ItemsCount)[] + { + (0, 0, 3), + (3, 0, 3), + (6, 0, 3), + (9, 0, 1), + } + }, + + new object[] + { + 0, 2, 10, 4, new (int BlockIndex, int StartingItemIndex, long ItemsCount)[] + { + (0, 0, 4), + (2, 0, 4), + (4, 0, 2), + } + }, + + new object[] + { + 0, 1, 10, 2, new (int BlockIndex, int StartingItemIndex, long ItemsCount)[] + { + (0, 0, 2), + (2, 0, 2), + (4, 0, 2), + (6, 0, 2), + (8, 0, 2), + } + }, + + new object[] + { + 0, 2, 10, 1, new (int BlockIndex, int StartingItemIndex, long ItemsCount)[] + { + (0, 0, 1), + (0, 1, 1), + (1, 0, 1), + (1, 1, 1), + (2, 0, 1), + (2, 1, 1), + (3, 0, 1), + (3, 1, 1), + (4, 0, 1), + (4, 1, 1), + } + }, + + new object[] + { + 0, 1, 10, 1, new (int BlockIndex, int StartingItemIndex, long ItemsCount)[] + { + (0, 0, 1), + (1, 0, 1), + (2, 0, 1), + (3, 0, 1), + (4, 0, 1), + (5, 0, 1), + (6, 0, 1), + (7, 0, 1), + (8, 0, 1), + (9, 0, 1), + } + }, + + }; + } +}