From b20da06268a9b51ca4a767f308ad43bcf45c6460 Mon Sep 17 00:00:00 2001 From: sakno Date: Fri, 8 Mar 2024 21:19:02 +0200 Subject: [PATCH] Release 5.2.0 --- CHANGELOG.md | 25 ++ README.md | 31 +-- src/DotNext.IO/DotNext.IO.csproj | 2 +- .../DotNext.Metaprogramming.csproj | 2 +- .../Collections/Concurrent/IndexPoolTests.cs | 107 ++++++++ .../Consensus/Raft/LeaderStateContextTests.cs | 15 ++ src/DotNext.Tests/Numerics/NumberTests.cs | 11 + .../Collections/Concurrent/IndexPool.cs | 252 ++++++++++++++++++ .../DotNext.Threading.csproj | 2 +- src/DotNext.Unsafe/DotNext.Unsafe.csproj | 2 +- src/DotNext/DotNext.csproj | 2 +- src/DotNext/Func.cs | 15 +- src/DotNext/Numerics/Number.cs | 122 +++++++++ .../DotNext.AspNetCore.Cluster.csproj | 2 +- .../Consensus/Raft/Http/RaftHttpCluster.cs | 21 +- .../DotNext.Net.Cluster.csproj | 2 +- .../Cluster/Consensus/Raft/FollowerState.cs | 1 - .../Consensus/Raft/LeaderState.Context.cs | 70 +++-- .../Consensus/Raft/LeaderState.Replication.cs | 2 +- .../Net/Cluster/Consensus/Raft/LeaderState.cs | 6 +- .../Raft/PersistentState.LockManager.cs | 4 +- .../Raft/PersistentState.SessionManagement.cs | 36 +-- .../Cluster/Consensus/Raft/PersistentState.cs | 2 +- .../Consensus/Raft/RaftCluster.DefaultImpl.cs | 21 +- .../Raft/RaftCluster.TermTracking.cs | 14 +- 25 files changed, 632 insertions(+), 137 deletions(-) create mode 100644 src/DotNext.Tests/Collections/Concurrent/IndexPoolTests.cs create mode 100644 src/DotNext.Threading/Collections/Concurrent/IndexPool.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 0f715832e..b56d4a9fe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,31 @@ Release Notes ==== +# 03-08-2024 +DotNext 5.2.0 +* Added `Number.IsPrime` static method that allows to check whether the specified number is a prime number +* Fixed AOT compatibility issues + +DotNext.Metaprogramming 5.2.0 +* Updated dependencies + +DotNext.Unsafe 5.2.0 +* Updated dependencies + +DotNext.Threading 5.2.0 +* Added specialized `IndexPool` data type that can be useful for implementing fast object pools + +DotNext.IO 5.2.0 +* Updated dependencies + +DotNext.Net.Cluster 5.2.0 +* Fixed [226](https://github.com/dotnet/dotNext/issues/226) +* Fixed [221](https://github.com/dotnet/dotNext/issues/221) + +DotNext.AspNetCore.Cluster 5.2.0 +* Fixed [226](https://github.com/dotnet/dotNext/issues/226) +* Fixed [221](https://github.com/dotnet/dotNext/issues/221) + # 02-28-2024 DotNext 5.1.0 * Added `Span.Advance` extension method for spans diff --git a/README.md b/README.md index e06c40b66..6e06f5798 100644 --- a/README.md +++ b/README.md @@ -44,30 +44,31 @@ All these things are implemented in 100% managed code on top of existing .NET AP * [NuGet Packages](https://www.nuget.org/profiles/rvsakno) # What's new -Release Date: 02-28-2024 +Release Date: 03-08-2024 -DotNext 5.1.0 -* Added `Span.Advance` extension method for spans -* `CollectionType.GetItemType` now correctly recognizes enumerable pattern even if target type doesn't implement `IEnumerable` +DotNext 5.2.0 +* Added `Number.IsPrime` static method that allows to check whether the specified number is a prime number +* Fixed AOT compatibility issues -DotNext.Metaprogramming 5.1.0 +DotNext.Metaprogramming 5.2.0 * Updated dependencies -DotNext.Unsafe 5.1.0 -* Added `UnmanagedMemory.AsMemory` static method that allows to wrap unmanaged pointer into [Memory<T>](https://learn.microsoft.com/en-us/dotnet/api/system.memory-1) - -DotNext.Threading 5.1.0 +DotNext.Unsafe 5.2.0 * Updated dependencies -DotNext.IO 5.1.0 -* Merged [225](https://github.com/dotnet/dotNext/pull/225) -* Added `AsUnbufferedStream` extension method for [SafeFileHandle](https://learn.microsoft.com/en-us/dotnet/api/microsoft.win32.safehandles.safefilehandle) class +DotNext.Threading 5.2.0 +* Added specialized `IndexPool` data type that can be useful for implementing fast object pools -DotNext.Net.Cluster 5.1.0 +DotNext.IO 5.2.0 * Updated dependencies -DotNext.AspNetCore.Cluster 5.1.0 -* Updated dependencies +DotNext.Net.Cluster 5.2.0 +* Fixed [226](https://github.com/dotnet/dotNext/issues/226) +* Fixed [221](https://github.com/dotnet/dotNext/issues/221) + +DotNext.AspNetCore.Cluster 5.2.0 +* Fixed [226](https://github.com/dotnet/dotNext/issues/226) +* Fixed [221](https://github.com/dotnet/dotNext/issues/221) Changelog for previous versions located [here](./CHANGELOG.md). diff --git a/src/DotNext.IO/DotNext.IO.csproj b/src/DotNext.IO/DotNext.IO.csproj index 3608bef3f..829ed2814 100644 --- a/src/DotNext.IO/DotNext.IO.csproj +++ b/src/DotNext.IO/DotNext.IO.csproj @@ -11,7 +11,7 @@ .NET Foundation and Contributors .NEXT Family of Libraries - 5.1.0 + 5.2.0 DotNext.IO MIT diff --git a/src/DotNext.Metaprogramming/DotNext.Metaprogramming.csproj b/src/DotNext.Metaprogramming/DotNext.Metaprogramming.csproj index 9b8e2ea7e..c8b5b134a 100644 --- a/src/DotNext.Metaprogramming/DotNext.Metaprogramming.csproj +++ b/src/DotNext.Metaprogramming/DotNext.Metaprogramming.csproj @@ -8,7 +8,7 @@ true false nullablePublicOnly - 5.1.0 + 5.2.0 .NET Foundation .NEXT Family of Libraries diff --git a/src/DotNext.Tests/Collections/Concurrent/IndexPoolTests.cs b/src/DotNext.Tests/Collections/Concurrent/IndexPoolTests.cs new file mode 100644 index 000000000..ddad312dc --- /dev/null +++ b/src/DotNext.Tests/Collections/Concurrent/IndexPoolTests.cs @@ -0,0 +1,107 @@ +namespace DotNext.Collections.Concurrent; + +public sealed class IndexPoolTests : Test +{ + [Fact] + public static void EmptyPool() + { + var pool = default(IndexPool); + False(pool.TryPeek(out _)); + False(pool.TryTake(out _)); + DoesNotContain(10, pool); + Empty(pool); + } + + [Fact] + public static void TakeAll() + { + var pool = new IndexPool(); + NotEmpty(pool); + + for (var i = 0; i <= IndexPool.MaxValue; i++) + { + Equal(i, pool.Take()); + } + + Throws(() => pool.Take()); + } + + [Fact] + public static void ContainsAll() + { + var pool = new IndexPool(); + for (var i = 0; i <= IndexPool.MaxValue; i++) + { + True(pool.Contains(i)); + } + + for (var i = 0; i <= IndexPool.MaxValue; i++) + { + Equal(i, pool.Take()); + } + + for (var i = 0; i <= IndexPool.MaxValue; i++) + { + False(pool.Contains(i)); + } + } + + [Fact] + public static void Enumerator() + { + var pool = new IndexPool(); + var expected = new int[pool.Count]; + Span.ForEach(expected, static (ref int value, int index) => value = index); + + Equal(expected, pool.ToArray()); + + while (pool.TryTake(out _)) + { + // take all indicies + } + + Equal(Array.Empty(), pool.ToArray()); + } + + [Fact] + public static void CustomMaxValue() + { + var pool = new IndexPool(maxValue: 2); + Equal(3, pool.Count); + + Equal(0, pool.Take()); + Equal(1, pool.Take()); + Equal(2, pool.Take()); + + False(pool.TryTake(out _)); + Empty(pool); + } + + [Fact] + public static void Consistency() + { + var pool = new IndexPool(); + Equal(0, pool.Take()); + + Equal(1, pool.Take()); + pool.Return(1); + + Equal(1, pool.Take()); + pool.Return(1); + + pool.Return(0); + } + + [Fact] + public static void TakeReturnMany() + { + var pool = new IndexPool(); + Span indicies = stackalloc int[IndexPool.Capacity]; + + Equal(IndexPool.Capacity, pool.Take(indicies)); + Empty(pool); + + pool.Return(indicies); + NotEmpty(pool); + } +} \ No newline at end of file diff --git a/src/DotNext.Tests/Net/Cluster/Consensus/Raft/LeaderStateContextTests.cs b/src/DotNext.Tests/Net/Cluster/Consensus/Raft/LeaderStateContextTests.cs index f93f2f1bd..6fe3cfd6c 100644 --- a/src/DotNext.Tests/Net/Cluster/Consensus/Raft/LeaderStateContextTests.cs +++ b/src/DotNext.Tests/Net/Cluster/Consensus/Raft/LeaderStateContextTests.cs @@ -96,4 +96,19 @@ public static void Resize() private static LeaderState.Replicator CreateReplicator(DummyRaftClusterMember member) => new(member, NullLogger.Instance); + + [Fact] + public static void RegressionIssue221() + { + const int length = 16; + using var context = new LeaderState.Context(length); + + var keys = new DummyRaftClusterMember[length]; + Span.Initialize(keys); + + for (var i = 0; i < length; i++) + { + var ctx = context.GetOrCreate(keys[i], CreateReplicator); + } + } } \ No newline at end of file diff --git a/src/DotNext.Tests/Numerics/NumberTests.cs b/src/DotNext.Tests/Numerics/NumberTests.cs index 7a9deb575..22431799a 100644 --- a/src/DotNext.Tests/Numerics/NumberTests.cs +++ b/src/DotNext.Tests/Numerics/NumberTests.cs @@ -63,4 +63,15 @@ public static void BinarySize() Equal(sizeof(int), Number.GetMaxByteCount()); Equal(sizeof(long), Number.GetMaxByteCount()); } + + [Fact] + public static void IsPrime() + { + False(Number.IsPrime(1L)); + True(Number.IsPrime(2L)); + True(Number.IsPrime(3)); + False(Number.IsPrime(4)); + + True(Number.IsPrime(1669)); + } } \ No newline at end of file diff --git a/src/DotNext.Threading/Collections/Concurrent/IndexPool.cs b/src/DotNext.Threading/Collections/Concurrent/IndexPool.cs new file mode 100644 index 000000000..b8e3e5bcc --- /dev/null +++ b/src/DotNext.Threading/Collections/Concurrent/IndexPool.cs @@ -0,0 +1,252 @@ +using System.Collections; +using System.ComponentModel; +using System.Diagnostics; +using System.Diagnostics.CodeAnalysis; +using System.Numerics; +using System.Runtime.InteropServices; + +namespace DotNext.Collections.Concurrent; + +/// +/// Represents a pool of integer values. +/// +/// +/// This type is thread-safe. +/// +[EditorBrowsable(EditorBrowsableState.Advanced)] +[StructLayout(LayoutKind.Auto)] +public struct IndexPool : ISupplier, IConsumer, IReadOnlyCollection +{ + private readonly int maxValue; + private ulong bitmask; + + /// + /// Initializes a new pool that can return an integer value within the range [0..]. + /// + public IndexPool() + { + bitmask = ulong.MaxValue; + maxValue = MaxValue; + } + + /// + /// Initializes a new pool that can return an integer within the range [0..]. + /// + /// The maximum possible value to return, inclusive. + /// + /// is less than zero; + /// or greater than . + /// + public IndexPool(int maxValue) + { + if (maxValue < 0 || maxValue > MaxValue) + throw new ArgumentOutOfRangeException(nameof(maxValue)); + + bitmask = ulong.MaxValue; + this.maxValue = maxValue; + } + + /// + /// Gets the maximum number that can be returned by the pool. + /// + /// Always returns 63. + public static int MaxValue => Capacity - 1; + + /// + /// Gets the maximum capacity of the pool. + /// + public static int Capacity => sizeof(ulong) * 8; + + /// + /// Tries to peek the next available index from the pool, without acquiring it. + /// + /// The index which is greater than or equal to zero. + /// if the index is available for rent; otherwise, . + public readonly bool TryPeek(out int result) + => (result = BitOperations.TrailingZeroCount(Volatile.Read(in bitmask))) <= maxValue; + + /// + /// Returns the available index from the pool. + /// + /// The index which is greater than or equal to zero. + /// if the index is successfully rented from the pool; otherwise, . + /// + public bool TryTake(out int result) + { + ulong current, newValue = Volatile.Read(in bitmask); + do + { + result = BitOperations.TrailingZeroCount(current = newValue); + + if (result > maxValue) + return false; + + newValue = current ^ (1UL << result); + } + while ((newValue = Interlocked.CompareExchange(ref bitmask, newValue, current)) != current); + + return true; + } + + /// + /// Returns the available index from the pool. + /// + /// The index which is greater than or equal to zero. + /// There is no available index to return. + /// + public int Take() + { + if (!TryTake(out var result)) + ThrowOverflowException(); + + return result; + + [DoesNotReturn] + [StackTraceHidden] + static void ThrowOverflowException() => throw new OverflowException(); + } + + /// + /// Takes all available indicies, atomically. + /// + /// + /// The buffer to be modified with the indicies taken from the pool. + /// The size of the buffer should not be less than . + /// + /// The number of indicies written to the buffer. + /// is too small to place indicies. + /// + public int Take(Span indicies) + { + if (indicies.Length < Capacity) + throw new ArgumentOutOfRangeException(nameof(indicies)); + + var oldValue = Interlocked.Exchange(ref bitmask, 0UL); + var bufferOffset = 0; + + for (int bitPosition = 0; bitPosition < Capacity; bitPosition++) + { + if (Contains(oldValue, bitPosition)) + { + indicies[bufferOffset++] = bitPosition; + } + } + + return bufferOffset; + } + + /// + int ISupplier.Invoke() => Take(); + + /// + /// Returns an index previously obtained using back to the pool. + /// + /// + /// is less than zero or greater than the maximum + /// value specified for this pool. + public void Return(int value) + { + if (value < 0 || value > maxValue) + ThrowArgumentOutOfRangeException(); + + Interlocked.Or(ref bitmask, 1UL << value); + + [DoesNotReturn] + [StackTraceHidden] + static void ThrowArgumentOutOfRangeException() + => throw new ArgumentOutOfRangeException(nameof(value)); + } + + /// + /// Returns multiple indicies, atomically. + /// + /// The buffer of indicies to return back to the pool. + public void Return(ReadOnlySpan indicies) + { + var newValue = 0UL; + + foreach (var index in indicies) + { + newValue |= 1UL << index; + } + + Interlocked.Or(ref bitmask, newValue); + } + + /// + void IConsumer.Invoke(int value) => Return(value); + + /// + /// Determines whether the specified index is available for rent. + /// + /// The value to check. + /// if is available for rent; otherwise, . + public readonly bool Contains(int value) + => value >= 0 && value <= maxValue && Contains(Volatile.Read(in bitmask), value); + + private static bool Contains(ulong bitmask, int index) + => (bitmask & (1UL << index)) is not 0UL; + + /// + /// Gets the number of available indicies. + /// + public readonly int Count => Math.Min(BitOperations.PopCount(bitmask), maxValue + 1); + + /// + /// Gets an enumerator over available indicies in the pool. + /// + /// The enumerator over available indicies in this pool. + public readonly Enumerator GetEnumerator() => new(Volatile.Read(in bitmask), maxValue); + + /// + readonly IEnumerator IEnumerable.GetEnumerator() => GetEnumerator().AsClassicEnumerator(); + + /// + readonly IEnumerator IEnumerable.GetEnumerator() => GetEnumerator().AsClassicEnumerator(); + + /// + /// Represents an enumerator over available indicies in the pool. + /// + [StructLayout(LayoutKind.Auto)] + public struct Enumerator + { + private readonly ulong bitmask; + private readonly int maxValue; + private int current; + + internal Enumerator(ulong bitmask, int maxValue) + { + this.bitmask = bitmask; + this.maxValue = maxValue; + current = -1; + } + + /// + /// Gets the current index. + /// + public readonly int Current => current; + + /// + /// Advances to the next available index. + /// + /// if enumerator advanced successfully; otherwise, . + public bool MoveNext() + { + while (++current <= maxValue) + { + if (Contains(bitmask, current)) + { + return true; + } + } + + return false; + } + + internal IEnumerator AsClassicEnumerator() + { + while (MoveNext()) + yield return Current; + } + } +} \ No newline at end of file diff --git a/src/DotNext.Threading/DotNext.Threading.csproj b/src/DotNext.Threading/DotNext.Threading.csproj index 04e4f78ad..45872754c 100644 --- a/src/DotNext.Threading/DotNext.Threading.csproj +++ b/src/DotNext.Threading/DotNext.Threading.csproj @@ -7,7 +7,7 @@ true true nullablePublicOnly - 5.1.0 + 5.2.0 .NET Foundation and Contributors .NEXT Family of Libraries diff --git a/src/DotNext.Unsafe/DotNext.Unsafe.csproj b/src/DotNext.Unsafe/DotNext.Unsafe.csproj index af05c5a80..abad20ae6 100644 --- a/src/DotNext.Unsafe/DotNext.Unsafe.csproj +++ b/src/DotNext.Unsafe/DotNext.Unsafe.csproj @@ -7,7 +7,7 @@ enable true true - 5.1.0 + 5.2.0 nullablePublicOnly .NET Foundation and Contributors diff --git a/src/DotNext/DotNext.csproj b/src/DotNext/DotNext.csproj index 2ae09a6a1..7f376f80a 100644 --- a/src/DotNext/DotNext.csproj +++ b/src/DotNext/DotNext.csproj @@ -11,7 +11,7 @@ .NET Foundation and Contributors .NEXT Family of Libraries - 5.1.0 + 5.2.0 DotNext MIT diff --git a/src/DotNext/Func.cs b/src/DotNext/Func.cs index ec5448d0b..4ea604ecb 100644 --- a/src/DotNext/Func.cs +++ b/src/DotNext/Func.cs @@ -92,21 +92,16 @@ public static Func Constant(T obj) if (typeof(T) == typeof(bool)) return Unsafe.As>(Constant(Unsafe.As(ref obj))); - // cache nulls - if (obj is null) - return Default!; - // slow path - allocates a new delegate - unsafe - { - return DelegateHelpers.CreateDelegate(&ConstantCore, obj); - } - - static T ConstantCore(object? obj) => (T)obj!; + return obj is null + ? Default! + : obj.UnboxAny; static T? Default() => default; } + private static T UnboxAny(this object obj) => (T)obj; + private static Func Constant(bool value) { return value ? True : False; diff --git a/src/DotNext/Numerics/Number.cs b/src/DotNext/Numerics/Number.cs index 968f04528..e17f345b0 100644 --- a/src/DotNext/Numerics/Number.cs +++ b/src/DotNext/Numerics/Number.cs @@ -1,3 +1,4 @@ +using System.ComponentModel; using System.Numerics; using System.Runtime.CompilerServices; @@ -99,4 +100,125 @@ public static float Normalize(this uint value) /// The normalized value in range [0..1). public static float Normalize(this int value) => Normalize(unchecked((uint)value)); + + /// + /// Determines whether the specified value is a prime number. + /// + /// The integer type. + /// The value to check. + /// if is a prime number; otherwise, . + /// is negative or zero. + public static bool IsPrime(T value) + where T : struct, IBinaryInteger, ISignedNumber + { + ArgumentOutOfRangeException.ThrowIfNegativeOrZero(value); + + if (value == T.One) + return false; + + var two = T.One << 1; + + if ((value & T.One) != T.Zero) + { + for (T divisor = two + T.One, limit = Sqrt(value); divisor <= limit; divisor += two) + { + if ((value % divisor) == T.Zero) + return false; + } + + return true; + } + + return value == two; + + // https://math.stackexchange.com/questions/2469446/what-is-a-fast-algorithm-for-finding-the-integer-square-root + static T Sqrt(T value) + { + var log2x = T.Log2(value) - T.One; + var log2y = int.CreateChecked(log2x >> 1); + + var y = T.One << log2y; + var y_squared = T.One << (2 * log2y); + + var sqr_diff = value - y_squared; + + // Perform lerp between powers of four + y += (sqr_diff / (T.One + T.One + T.One)) >> log2y; + + // The estimate is probably too low, refine it upward + y_squared = y * y; + sqr_diff = value - y_squared; + + y += sqr_diff / (y << 1); + + // The estimate may be too high. If so, refine it downward + y_squared = y * y; + sqr_diff = value - y_squared; + if (sqr_diff >= T.Zero) + { + return y; + } + + y -= (-sqr_diff / (y << 1)) + T.One; + + // The estimate may still be 1 too high + y_squared = y * y; + sqr_diff = value - y_squared; + if (sqr_diff < T.Zero) + { + --y; + } + + return y; + } + } + + /// + /// Gets a prime number which is greater than the specified value. + /// + /// The type of the value. + /// The value which is smaller than the requested prime number. + /// The table with cached prime numbers sorted in ascending order. + /// The prime number which is greater than . + /// There is no prime number that is greater than and less than . + [EditorBrowsable(EditorBrowsableState.Never)] + public static T GetPrime(T lowerBound, ReadOnlySpan cachedPrimes = default) + where T : struct, IBinaryInteger, ISignedNumber, IMinMaxValue + { + ArgumentOutOfRangeException.ThrowIfNegativeOrZero(lowerBound); + + if (TryGetFromTable(cachedPrimes, lowerBound, out T result)) + return result; + + //outside of predefined table + for (result = lowerBound | T.One; result < T.MaxValue; result += T.One + T.One) + { + if (IsPrime(result)) + return result; + } + + throw new OverflowException(); + + static bool TryGetFromTable(ReadOnlySpan cachedPrimes, T value, out T result) + { + var low = 0; + for (var high = cachedPrimes.Length; low < high;) + { + var mid = (low + high) / 2; + result = cachedPrimes[mid]; + var cmp = result.CompareTo(value); + if (cmp > 0) + high = mid; + else + low = mid + 1; + } + + bool success; + result = (success = low < cachedPrimes.Length) + ? T.CreateChecked(cachedPrimes[low]) + : default; + + return success; + } + } } \ No newline at end of file diff --git a/src/cluster/DotNext.AspNetCore.Cluster/DotNext.AspNetCore.Cluster.csproj b/src/cluster/DotNext.AspNetCore.Cluster/DotNext.AspNetCore.Cluster.csproj index d159df9bf..c781a5356 100644 --- a/src/cluster/DotNext.AspNetCore.Cluster/DotNext.AspNetCore.Cluster.csproj +++ b/src/cluster/DotNext.AspNetCore.Cluster/DotNext.AspNetCore.Cluster.csproj @@ -8,7 +8,7 @@ true true nullablePublicOnly - 5.1.0 + 5.2.0 .NET Foundation and Contributors .NEXT Family of Libraries diff --git a/src/cluster/DotNext.AspNetCore.Cluster/Net/Cluster/Consensus/Raft/Http/RaftHttpCluster.cs b/src/cluster/DotNext.AspNetCore.Cluster/Net/Cluster/Consensus/Raft/Http/RaftHttpCluster.cs index 3fe0c08e8..96bfd4c70 100644 --- a/src/cluster/DotNext.AspNetCore.Cluster/Net/Cluster/Consensus/Raft/Http/RaftHttpCluster.cs +++ b/src/cluster/DotNext.AspNetCore.Cluster/Net/Cluster/Consensus/Raft/Http/RaftHttpCluster.cs @@ -150,7 +150,7 @@ HttpMessageHandler IHostingContext.CreateHttpHandler() public override async Task StartAsync(CancellationToken token) { configurator?.OnStart(this, metadata); - ConfigurationStorage.ActiveConfigurationChanged += GetConfigurationEventHandler(configurationEvents.Writer); + ConfigurationStorage.ActiveConfigurationChanged += configurationEvents.Writer.WriteConfigurationEvent; if (coldStart) { @@ -193,7 +193,7 @@ async Task StopAsync() { configurator?.OnStop(this); duplicationDetector.Trim(100); - ConfigurationStorage.ActiveConfigurationChanged -= GetConfigurationEventHandler(configurationEvents.Writer); + ConfigurationStorage.ActiveConfigurationChanged -= configurationEvents.Writer.WriteConfigurationEvent; configurationEvents.Writer.TryComplete(); await pollingLoopTask.ConfigureAwait(false); pollingLoopTask = Task.CompletedTask; @@ -205,17 +205,6 @@ async Task StopAsync() } } - private static Func GetConfigurationEventHandler(ChannelWriter<(UriEndPoint, bool)> writer) - { - unsafe - { - return DelegateHelpers.CreateDelegate, UriEndPoint, bool, CancellationToken, ValueTask>(&WriteConfigurationEvent, writer); - } - - static ValueTask WriteConfigurationEvent(ChannelWriter<(UriEndPoint, bool)> writer, UriEndPoint address, bool isAdded, CancellationToken token) - => writer.WriteAsync(new(address, isAdded), token); - } - /// ISubscriber? IPeerMesh.TryGetPeer(EndPoint peer) { @@ -243,4 +232,10 @@ protected override void Dispose(bool disposing) base.Dispose(disposing); } +} + +file static class RaftHttpClusterHelpers +{ + internal static ValueTask WriteConfigurationEvent(this ChannelWriter<(UriEndPoint, bool)> writer, UriEndPoint address, bool isAdded, CancellationToken token) + => writer.WriteAsync(new(address, isAdded), token); } \ No newline at end of file diff --git a/src/cluster/DotNext.Net.Cluster/DotNext.Net.Cluster.csproj b/src/cluster/DotNext.Net.Cluster/DotNext.Net.Cluster.csproj index e4ae7cfb7..c776fc7cf 100644 --- a/src/cluster/DotNext.Net.Cluster/DotNext.Net.Cluster.csproj +++ b/src/cluster/DotNext.Net.Cluster/DotNext.Net.Cluster.csproj @@ -8,7 +8,7 @@ enable true nullablePublicOnly - 5.1.0 + 5.2.0 .NET Foundation and Contributors .NEXT Family of Libraries diff --git a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/FollowerState.cs b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/FollowerState.cs index 3065fc652..eea246a75 100644 --- a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/FollowerState.cs +++ b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/FollowerState.cs @@ -1,6 +1,5 @@ using System.Diagnostics.Metrics; using System.Runtime.InteropServices; -using Debug = System.Diagnostics.Debug; namespace DotNext.Net.Cluster.Consensus.Raft; diff --git a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.Context.cs b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.Context.cs index 28a272bdb..78328c3ad 100644 --- a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.Context.cs +++ b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.Context.cs @@ -3,10 +3,11 @@ using System.Runtime; using System.Runtime.CompilerServices; using System.Runtime.InteropServices; -using Microsoft.Extensions.Logging; namespace DotNext.Net.Cluster.Consensus.Raft; +using static Numerics.Number; + internal partial class LeaderState { private sealed class ContextEntry : Disposable @@ -23,13 +24,11 @@ internal ContextEntry(TMember member, int hashCode, Func fa internal int HashCode => hashCode; - [DisallowNull] internal TMember? Key { get => Unsafe.As(handle.Target); } - [DisallowNull] internal Replicator? Value { get => Unsafe.As(handle.Dependent); @@ -42,13 +41,6 @@ internal void Reuse(TMember key, int hashCode, Func factory this.hashCode = hashCode; } - internal void Deconstruct(out TMember? key, out Replicator? context) - { - var pair = handle.TargetAndDependent; - key = Unsafe.As(pair.Target); - context = Unsafe.As(pair.Dependent); - } - protected override void Dispose(bool disposing) { if (disposing) @@ -70,14 +62,33 @@ protected override void Dispose(bool disposing) #endif struct Context : IDisposable { - private static readonly int HalfMaxSize = Array.MaxLength >> 1; + private static ReadOnlySpan Primes => [ + 3, 7, 11, 17, 23, 29, 37, 47, 59, 71, 89, 107, 131, 163, 197, 239, 293, 353, 431, 521, 631, 761, 919, + 1103, 1327, 1597, 1931, 2333, 2801, 3371, 4049, 4861, 5839, 7013, 8419, 10103, 12143, 14591, + 17519, 21023, 25229, 30293, 36353, 43627, 52361, 62851, 75431, 90523, 108631, 130363, 156437, + 187751, 225307, 270371, 324449, 389357, 467237, 560689, 672827, 807403, 968897, 1162687, 1395263, + 1674319, 2009191, 2411033, 2893249, 3471899, 4166287, 4999559, 5999471, 7199369]; + private ContextEntry?[] entries; public Context(int sizeHint) { Debug.Assert(sizeHint > 0); - entries = new ContextEntry?[sizeHint <= HalfMaxSize ? sizeHint << 1 : sizeHint]; + entries = new ContextEntry?[GetPrime(sizeHint, Primes)]; + } + + private static int Grow(int size) + { + // This is the maximum prime smaller than Array.MaxLength + const int maxPrimeLength = 0x7FEFFFFD; + + int newSize; + return size is maxPrimeLength + ? throw new InsufficientMemoryException() + : (uint)(newSize = size << 1) > maxPrimeLength && maxPrimeLength > size + ? maxPrimeLength + : GetPrime(newSize, Primes); } public Context() => entries = []; @@ -91,21 +102,19 @@ private readonly int GetIndex(TMember member, out int hashCode) private readonly ref ContextEntry? GetEntry(TMember member, out int hashCode) => ref Unsafe.Add(ref MemoryMarshal.GetArrayDataReference(entries), GetIndex(member, out hashCode)); - private void ResizeAndRemoveDeadEntries() + private void ResizeAndRemoveDeadEntries(CancellationToken token) { - if (entries.Length == Array.MaxLength) - throw new InsufficientMemoryException(); - var oldEntries = entries; - entries = new ContextEntry?[entries.Length <= HalfMaxSize ? entries.Length << 1 : entries.Length + 1]; + entries = new ContextEntry?[Grow(oldEntries.Length)]; // copy elements from old array to a new one - for (var i = 0; i < oldEntries.Length; i++) + for (var i = 0; i < oldEntries.Length; i++, token.ThrowIfCancellationRequested()) { ref var oldEntry = ref Unsafe.Add(ref MemoryMarshal.GetArrayDataReference(oldEntries), i); for (ContextEntry? current = oldEntry, next; current is not null; current = next) { - next = current.Next; // make a copy because Next can be modified by Insert operation + next = current.Next; + current.Next = null; if (current.Key is { } key) { @@ -124,17 +133,22 @@ private void ResizeAndRemoveDeadEntries() } } - private readonly void Insert(ContextEntry entry) + private readonly bool Insert(ContextEntry entry) { + Debug.Assert(entry.Next is null); + + const int maxCollisions = 3; ref var location = ref Unsafe.Add(ref MemoryMarshal.GetArrayDataReference(entries), GetIndex(entry.HashCode, entries.Length)); - while (location is not null) + int collisions; + for (collisions = 0; location is not null; collisions++) location = ref location.Next; location = entry; + return collisions <= maxCollisions; } - public Replicator GetOrCreate(TMember key, Func factory) + public Replicator GetOrCreate(TMember key, Func factory, CancellationToken token = default) { Debug.Assert(key is not null); @@ -151,10 +165,9 @@ public Replicator GetOrCreate(TMember key, Func factory) ContextEntry? entryToReuse = null; // try to get from dictionary - for (var current = entry; current is not null; current = current.Next) + for (var current = entry; current is not null; current = current.Next, token.ThrowIfCancellationRequested()) { - var tmp = current.Key; - if (tmp is null) + if (current.Key is not { } tmp) { entryToReuse ??= current; } @@ -174,11 +187,10 @@ public Replicator GetOrCreate(TMember key, Func factory) { entryToReuse.Reuse(key, hashCode, factory, out result); } - else + else if (!Insert(new(key, hashCode, factory, out result))) { - // failed to reuse, add a new element - ResizeAndRemoveDeadEntries(); - Insert(new(key, hashCode, factory, out result)); + // too many collisions, resize + ResizeAndRemoveDeadEntries(token); } } diff --git a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.Replication.cs b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.Replication.cs index b23d2301b..10896c634 100644 --- a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.Replication.cs +++ b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.Replication.cs @@ -178,7 +178,7 @@ private ConfiguredTaskAwaitable>.ConfiguredTaskAwaiter R where TEntry : notnull, IRaftLogEntry where TList : notnull, IReadOnlyList { - logger.ReplicaSize(Member.EndPoint, entries.Count, replicationIndex, precedingTerm, Member.State.ConfigurationFingerprint, fingerprint, applyConfig); + logger.ReplicaSize(Member.EndPoint, entries.Count, replicationIndex, precedingTerm, fingerprint, Member.State.ConfigurationFingerprint, applyConfig); var result = Member.AppendEntriesAsync(term, entries, replicationIndex, precedingTerm, commitIndex, configuration, applyConfig, token).ConfigureAwait(false).GetAwaiter(); replicationIndex += entries.Count; return result; diff --git a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.cs b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.cs index 9b567b918..f14fe1567 100644 --- a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.cs +++ b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/LeaderState.cs @@ -56,13 +56,13 @@ internal LeaderState(IRaftStateMachine stateMachine, long term, TimeSpa var precedingIndex = member.State.PrecedingIndex; // fork replication procedure - replicator = context.GetOrCreate(member, replicatorFactory); + replicator = context.GetOrCreate(member, replicatorFactory, LeadershipToken); replicator.Initialize(activeConfig, proposedConfig, commitIndex, currentTerm, precedingIndex); response = SpawnReplicationAsync(replicator, auditTrail, currentIndex, LeadershipToken); } else { - replicator = context.GetOrCreate(member, localReplicatorFactory); + replicator = context.GetOrCreate(member, localReplicatorFactory, LeadershipToken); response = localMemberResponse; } } @@ -148,7 +148,7 @@ private async Task DoHeartbeats(TimeSpan period, IAuditTrail audi int quorum = 0, commitQuorum = 0, majority; (long currentIndex, long commitIndex, majority) = ForkHeartbeats(responsePipe, auditTrail, configurationStorage, enumerator); - while (await responsePipe.WaitToReadAsync(LeadershipToken).ConfigureAwait(false)) + while (await responsePipe.WaitToReadAsync().ConfigureAwait(false)) { while (responsePipe.TryRead(out var response, out var replicator)) { diff --git a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentState.LockManager.cs b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentState.LockManager.cs index 032a5d4d7..40450fded 100644 --- a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentState.LockManager.cs +++ b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentState.LockManager.cs @@ -33,8 +33,10 @@ sealed class LockManager : QueuedSynchronizer private uint readerCount; private bool allowWrite; + // maximum possible number of concurrent locks is max readers count + 1 because WriteLock can be acquired + // when all readers are active internal LockManager(int concurrencyLevel) - : base(concurrencyLevel + 2) // + write lock + compaction lock + : base(concurrencyLevel + 1) // + write lock or compaction lock { maxReadCount = (uint)concurrencyLevel; allowWrite = true; diff --git a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentState.SessionManagement.cs b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentState.SessionManagement.cs index 764a90533..ac78f8aab 100644 --- a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentState.SessionManagement.cs +++ b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentState.SessionManagement.cs @@ -1,10 +1,10 @@ -using System.Numerics; -using System.Runtime.CompilerServices; +using System.Runtime.CompilerServices; using System.Runtime.InteropServices; namespace DotNext.Net.Cluster.Consensus.Raft; using AtomicBoolean = Threading.Atomic.Boolean; +using IndexPool = Collections.Concurrent.IndexPool; public partial class PersistentState { @@ -21,37 +21,13 @@ private protected abstract class SessionIdPool private sealed class FastSessionIdPool : SessionIdPool { - internal const int MaxReadersCount = 63; + private IndexPool indicies = new(); - // all bits are set to 1 - // if bit at position N is 1 then N is available session identifier; - // otherwise, session identifier N is acquired by another thread - private ulong control = ulong.MaxValue; + internal static int MaxReadersCount => IndexPool.Capacity; - internal override int Take() - { - int sessionId; - ulong current, newValue = Volatile.Read(in control); - do - { - sessionId = BitOperations.TrailingZeroCount(current = newValue); - newValue = current ^ (1UL << sessionId); - } - while ((newValue = Interlocked.CompareExchange(ref control, newValue, current)) != current); - - return sessionId; - } + internal override int Take() => indicies.Take(); - internal override void Return(int sessionId) - { - ulong current, newValue = Volatile.Read(ref control); - do - { - current = newValue; - newValue = current | (1UL << sessionId); - } - while ((newValue = Interlocked.CompareExchange(ref control, newValue, current)) != current); - } + internal override void Return(int sessionId) => indicies.Return(sessionId); } private sealed class SlowSessionIdPool : SessionIdPool diff --git a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentState.cs b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentState.cs index 4f308074b..065623be3 100644 --- a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentState.cs +++ b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/PersistentState.cs @@ -53,7 +53,7 @@ private protected PersistentState(DirectoryInfo path, int recordsPerPartition, O commitEvent = new() { MeasurementTags = configuration.MeasurementTags }; bufferManager = new(configuration); concurrentReads = configuration.MaxConcurrentReads; - sessionManager = concurrentReads < FastSessionIdPool.MaxReadersCount + sessionManager = concurrentReads < FastSessionIdPool.MaxReadersCount // not <=, see LockManager for more info ? new FastSessionIdPool() : new SlowSessionIdPool(concurrentReads); parallelIO = configuration.ParallelIO; diff --git a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/RaftCluster.DefaultImpl.cs b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/RaftCluster.DefaultImpl.cs index b978016b2..145f7ba8f 100644 --- a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/RaftCluster.DefaultImpl.cs +++ b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/RaftCluster.DefaultImpl.cs @@ -121,7 +121,7 @@ public RaftCluster(NodeConfiguration configuration) /// The task representing asynchronous execution of the method. public override async Task StartAsync(CancellationToken token = default) { - ConfigurationStorage.ActiveConfigurationChanged += GetConfigurationEventHandler(configurationEvents.Writer); + ConfigurationStorage.ActiveConfigurationChanged += configurationEvents.Writer.WriteConfigurationEvent; if (coldStart) { @@ -151,17 +151,6 @@ public override async Task StartAsync(CancellationToken token = default) await announcer(LocalMemberAddress, metadata, token).ConfigureAwait(false); } - private static Func GetConfigurationEventHandler(ChannelWriter<(EndPoint, bool)> writer) - { - unsafe - { - return DelegateHelpers.CreateDelegate, EndPoint, bool, CancellationToken, ValueTask>(&WriteConfigurationEvent, writer); - } - - static ValueTask WriteConfigurationEvent(ChannelWriter<(EndPoint, bool)> writer, EndPoint address, bool isAdded, CancellationToken token) - => writer.WriteAsync(new(address, isAdded), token); - } - /// protected override ValueTask DetectLocalMemberAsync(RaftClusterMember candidate, CancellationToken token) => new(EndPointComparer.Equals(LocalMemberAddress, candidate.EndPoint)); @@ -181,7 +170,7 @@ async Task StopAsync() { await (server?.DisposeAsync() ?? ValueTask.CompletedTask).ConfigureAwait(false); server = null; - ConfigurationStorage.ActiveConfigurationChanged -= GetConfigurationEventHandler(configurationEvents.Writer); + ConfigurationStorage.ActiveConfigurationChanged -= configurationEvents.Writer.WriteConfigurationEvent; configurationEvents.Writer.TryComplete(); await pollingLoopTask.ConfigureAwait(false); pollingLoopTask = Task.CompletedTask; @@ -355,4 +344,10 @@ protected override void Dispose(bool disposing) base.Dispose(disposing); } +} + +file static class RaftClusterHelpers +{ + internal static ValueTask WriteConfigurationEvent(this ChannelWriter<(EndPoint, bool)> writer, EndPoint address, bool isAdded, CancellationToken token) + => writer.WriteAsync(new(address, isAdded), token); } \ No newline at end of file diff --git a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/RaftCluster.TermTracking.cs b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/RaftCluster.TermTracking.cs index 4ccbe9d4d..53e8e5ac7 100644 --- a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/RaftCluster.TermTracking.cs +++ b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/RaftCluster.TermTracking.cs @@ -1,26 +1,14 @@ -using Debug = System.Diagnostics.Debug; - namespace DotNext.Net.Cluster.Consensus.Raft; using IO.Log; public partial class RaftCluster { - private sealed class ReplicationWithSenderTermDetector : ILogEntryProducer + private sealed class ReplicationWithSenderTermDetector(ILogEntryProducer entries, long expectedTerm) : ILogEntryProducer where TEntry : notnull, IRaftLogEntry { - private readonly ILogEntryProducer entries; - private readonly long expectedTerm; private bool replicatedWithExpectedTerm; - internal ReplicationWithSenderTermDetector(ILogEntryProducer entries, long expectedTerm) - { - Debug.Assert(entries is not null); - - this.entries = entries; - this.expectedTerm = expectedTerm; - } - internal bool IsReplicatedWithExpectedTerm => replicatedWithExpectedTerm; TEntry IAsyncEnumerator.Current