Cluster 2.6.0

dotnet · May 30, 2024 · cacf3e5 · cacf3e5
1 parent d0a2508
commit cacf3e5
Show file tree

Hide file tree

Showing 26 changed files with 352 additions and 86 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,6 +1,16 @@
 Release Notes
 ====
 
+# 05-30-2024
+<a href="https://www.nuget.org/packages/dotnext.threading/5.4.1">DotNext.Metaprogramming 5.4.1</a>
+* Smallish performance improvements for all synchronization primitives
+
+<a href="https://www.nuget.org/packages/dotnext.net.cluster/5.6.0">DotNext.Net.Cluster 5.6.0</a>
+* Added support of custom data to be passed to `PersistentState.ApplyAsync` method through WAL processing pipeline
+
+<a href="https://www.nuget.org/packages/dotnext.aspnetcore.cluster/5.6.0">DotNext.AspNetCore.Cluster 5.6.0</a>
+* Updated dependencies
+
 # 05-21-2024
 <a href="https://www.nuget.org/packages/dotnext.threading/5.4.0">DotNext.Metaprogramming 5.4.0</a>
 * Smallish performance improvements of `IndexPool` instance methods

diff --git a/README.md b/README.md
@@ -44,11 +44,16 @@ 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: 05-21-2024
+Release Date: 05-30-2024
 
-<a href="https://www.nuget.org/packages/dotnext.threading/5.4.0">DotNext.Metaprogramming 5.4.0</a>
-* Smallish performance improvements of `IndexPool` instance methods
-* Added ability to instantiate empty `IndexPool`
+<a href="https://www.nuget.org/packages/dotnext.threading/5.4.1">DotNext.Metaprogramming 5.4.1</a>
+* Smallish performance improvements for all synchronization primitives
+
+<a href="https://www.nuget.org/packages/dotnext.net.cluster/5.6.0">DotNext.Net.Cluster 5.6.0</a>
+* Added support of custom data to be passed to `PersistentState.ApplyAsync` method through WAL processing pipeline
+
+<a href="https://www.nuget.org/packages/dotnext.aspnetcore.cluster/5.6.0">DotNext.AspNetCore.Cluster 5.6.0</a>
+* Updated dependencies
 
 Changelog for previous versions located [here](./CHANGELOG.md).
 

diff --git a/src/DotNext.Tests/Net/Cluster/Consensus/Raft/Commands/CommandInterpreterTests.cs b/src/DotNext.Tests/Net/Cluster/Consensus/Raft/Commands/CommandInterpreterTests.cs
@@ -127,8 +127,11 @@ internal static ValueTask DoBinaryOperation(ref int value, BinaryOperationComman
         }
 
         [CommandHandler]
-        public ValueTask DoBinaryOperation(BinaryOperationCommand command, CancellationToken token)
-            => DoBinaryOperation(ref Value, command, token);
+        public ValueTask DoBinaryOperation(BinaryOperationCommand command, object context, CancellationToken token)
+        {
+            Null(context);
+            return DoBinaryOperation(ref Value, command, token);
+        }
 
         internal static ValueTask DoUnaryOperation(ref int value, UnaryOperationCommand command, CancellationToken token)
         {
@@ -213,18 +216,11 @@ public static async Task MethodsAsHandlers()
     public static async Task DelegatesAsHandlers()
     {
         var state = new StrongBox<int>();
-        Func<BinaryOperationCommand, CancellationToken, ValueTask> binaryOp = (command, token) => CustomInterpreter.DoBinaryOperation(ref state.Value, command, token);
-        Func<UnaryOperationCommand, CancellationToken, ValueTask> unaryOp = (command, token) => CustomInterpreter.DoUnaryOperation(ref state.Value, command, token);
-        Func<AssignCommand, CancellationToken, ValueTask> assignOp = (command, token) =>
-        {
-            state.Value = command.Value;
-            return new ValueTask();
-        };
 
         var interpreter = new CommandInterpreter.Builder()
-            .Add(BinaryOperationCommand.Id, binaryOp)
-            .Add(UnaryOperationCommand.Id, unaryOp)
-            .Add(AssignCommand.Id, assignOp)
+            .Add(BinaryOperationCommand.Id, new Func<BinaryOperationCommand, CancellationToken, ValueTask>(BinaryOp))
+            .Add(UnaryOperationCommand.Id, new Func<UnaryOperationCommand, CancellationToken, ValueTask>(UnaryOp))
+            .Add(AssignCommand.Id, new Func<AssignCommand, object, CancellationToken, ValueTask>(AssignOp))
             .Build();
 
         var entry1 = interpreter.CreateLogEntry(new BinaryOperationCommand { X = 40, Y = 2, Type = BinaryOperation.Add }, 1L);
@@ -240,8 +236,19 @@ public static async Task DelegatesAsHandlers()
 
         var entry3 = interpreter.CreateLogEntry(new AssignCommand { Value = int.MaxValue }, 68L);
         Equal(68L, entry3.Term);
-        Equal(3, await interpreter.InterpretAsync(entry3));
+        Equal(3, await interpreter.InterpretAsync(entry3, string.Empty));
         Equal(int.MaxValue, state.Value);
+
+        ValueTask BinaryOp(BinaryOperationCommand command, CancellationToken token) => CustomInterpreter.DoBinaryOperation(ref state.Value, command, token);
+
+        ValueTask UnaryOp(UnaryOperationCommand command, CancellationToken token) => CustomInterpreter.DoUnaryOperation(ref state.Value, command, token);
+
+        ValueTask AssignOp(AssignCommand command, object context, CancellationToken token)
+        {
+            NotNull(context);
+            state.Value = command.Value;
+            return ValueTask.CompletedTask;
+        }
     }
 
     [Fact]

diff --git a/src/DotNext.Tests/Net/Cluster/Consensus/Raft/MemoryBasedStateMachineTests.cs b/src/DotNext.Tests/Net/Cluster/Consensus/Raft/MemoryBasedStateMachineTests.cs
@@ -41,6 +41,14 @@ protected override ValueTask ApplyAsync(LogEntry entry)
             False(entry.IsEmpty);
             True(entry.GetReader().TryGetRemainingBytesCount(out var length));
             NotEqual(0L, length);
+
+            switch (entry.Context)
+            {
+                case int value:
+                    Equal(56, value);
+                    break;
+            }
+
             return ValueTask.CompletedTask;
         }
 
@@ -96,9 +104,9 @@ public static async Task StateManipulations()
         {
             Equal(0, state.Term);
             Equal(1, await state.IncrementTermAsync(default));
-            True(state.IsVotedFor(default(ClusterMemberId)));
+            True(state.IsVotedFor(default));
             await state.UpdateVotedForAsync(member);
-            False(state.IsVotedFor(default(ClusterMemberId)));
+            False(state.IsVotedFor(default));
             True(state.IsVotedFor(member));
         }
         finally
@@ -111,7 +119,7 @@ public static async Task StateManipulations()
         try
         {
             Equal(1, state.Term);
-            False(state.IsVotedFor(default(ClusterMemberId)));
+            False(state.IsVotedFor(default));
             True(state.IsVotedFor(member));
         }
         finally
@@ -149,7 +157,7 @@ public static async Task EmptyLogEntry()
     [InlineData(1024, false, 65)]
     public static async Task QueryAppendEntries(long partitionSize, bool caching, int concurrentReads)
     {
-        var entry1 = new TestLogEntry("SET X = 0") { Term = 42L };
+        var entry1 = new TestLogEntry("SET X = 0") { Term = 42L, Context = 56 };
         var entry2 = new TestLogEntry("SET Y = 1") { Term = 43L };
         var dir = Path.Combine(Path.GetTempPath(), Path.GetRandomFileName());
         Func<IReadOnlyList<IRaftLogEntry>, long?, CancellationToken, ValueTask<Missing>> checker;
@@ -175,6 +183,7 @@ public static async Task QueryAppendEntries(long partitionSize, bool caching, in
                 Equal(0L, entries.First().Term);          // element 0
                 Equal(42L, entries.Skip(1).First().Term); // element 1
                 Equal(entry1.Content, await entries[1].ToStringAsync(Encoding.UTF8));
+                Equal(entry1.Context, IsAssignableFrom<IInputLogEntry>(entries[1]).Context);
                 return Missing.Value;
             };
 
@@ -460,7 +469,7 @@ public static async Task PartitionOverflow(bool useCaching)
     [InlineData(false)]
     public static async Task Commit(bool useCaching)
     {
-        var entry1 = new TestLogEntry("SET X = 0") { Term = 42L };
+        var entry1 = new TestLogEntry("SET X = 0") { Term = 42L, Context = 56 };
         var entry2 = new TestLogEntry("SET Y = 1") { Term = 43L };
         var entry3 = new TestLogEntry("SET Z = 2") { Term = 44L };
         var entry4 = new TestLogEntry("SET U = 3") { Term = 45L };

diff --git a/src/DotNext.Tests/Net/Cluster/Consensus/Raft/TestLogEntry.cs b/src/DotNext.Tests/Net/Cluster/Consensus/Raft/TestLogEntry.cs
@@ -6,7 +6,7 @@ namespace DotNext.Net.Cluster.Consensus.Raft;
 using TextMessage = Messaging.TextMessage;
 
 [ExcludeFromCodeCoverage]
-internal sealed class TestLogEntry : TextMessage, IRaftLogEntry
+internal sealed class TestLogEntry : TextMessage, IInputLogEntry
 {
     public TestLogEntry(string command)
         : base(command, "Entry")
@@ -19,4 +19,10 @@ public TestLogEntry(string command)
     public long Term { get; set; }
 
     bool ILogEntry.IsSnapshot => false;
+
+    public object Context
+    {
+        get;
+        init;
+    }
 }
diff --git a/src/DotNext.Threading/DotNext.Threading.csproj b/src/DotNext.Threading/DotNext.Threading.csproj
@@ -7,7 +7,7 @@
     <ImplicitUsings>true</ImplicitUsings>
     <IsTrimmable>true</IsTrimmable>
     <Features>nullablePublicOnly</Features>
-    <VersionPrefix>5.4.0</VersionPrefix>
+    <VersionPrefix>5.4.1</VersionPrefix>
     <VersionSuffix></VersionSuffix>
     <Authors>.NET Foundation and Contributors</Authors>
     <Product>.NEXT Family of Libraries</Product>

diff --git a/src/DotNext.Threading/Threading/AsyncCountdownEvent.cs b/src/DotNext.Threading/Threading/AsyncCountdownEvent.cs
@@ -293,7 +293,7 @@ internal ValueTask<bool> SignalAndWaitAsync(out bool completedSynchronously, Tim
                         goto resume_suspended_callers;
                     }
 
-                    factory = EnqueueNode(ref pool, ref manager, throwOnTimeout: false);
+                    factory = EnqueueNode<DefaultWaitNode, StateManager>(ref pool, WaitNodeFlags.None);
                 }
 
                 completedSynchronously = false;
@@ -335,7 +335,7 @@ internal ValueTask SignalAndWaitAsync(out bool completedSynchronously, Cancellat
                 goto resume_suspended_callers;
             }
 
-            factory = EnqueueNode(ref pool, ref manager, throwOnTimeout: true);
+            factory = EnqueueNode<DefaultWaitNode, StateManager>(ref pool, WaitNodeFlags.ThrowOnTimeout);
         }
 
         completedSynchronously = false;

diff --git a/src/DotNext.Threading/Threading/AsyncTrigger.cs b/src/DotNext.Threading/Threading/AsyncTrigger.cs
@@ -189,7 +189,7 @@ public ValueTask<bool> SignalAndWaitAsync(bool resumeAll, bool throwOnEmptyQueue
                     suspendedCallers = Detach(resumeAll)?.SetResult(true, out signaled);
                     factory = !signaled && throwOnEmptyQueue
                         ? EmptyWaitQueueExceptionFactory.Instance
-                        : EnqueueNode(ref pool, ref manager, throwOnTimeout: false);
+                        : EnqueueNode<DefaultWaitNode, LockManager>(ref pool, WaitNodeFlags.None);
                 }
 
                 suspendedCallers?.Unwind();
@@ -229,7 +229,7 @@ public ValueTask SignalAndWaitAsync(bool resumeAll, bool throwOnEmptyQueue, Canc
             suspendedCallers = Detach(resumeAll)?.SetResult(true, out signaled);
             factory = !signaled && throwOnEmptyQueue
                 ? EmptyWaitQueueExceptionFactory.Instance
-                : EnqueueNode(ref pool, ref manager, throwOnTimeout: true);
+                : EnqueueNode<DefaultWaitNode, LockManager>(ref pool, WaitNodeFlags.ThrowOnTimeout);
         }
 
         suspendedCallers?.Unwind();

diff --git a/src/DotNext.Threading/Threading/QueuedSynchronizer.cs b/src/DotNext.Threading/Threading/QueuedSynchronizer.cs
@@ -138,15 +138,15 @@ private protected void EnqueueNode(WaitNode node)
         SuspendedCallersMeter.Add(1, measurementTags);
     }
 
-    private protected TNode EnqueueNode<TNode, TLockManager>(ref ValueTaskPool<bool, TNode, Action<TNode>> pool, ref TLockManager manager, bool throwOnTimeout)
+    private protected TNode EnqueueNode<TNode, TLockManager>(ref ValueTaskPool<bool, TNode, Action<TNode>> pool, WaitNodeFlags flags)
         where TNode : WaitNode, IPooledManualResetCompletionSource<Action<TNode>>, new()
         where TLockManager : struct, ILockManager<TNode>
     {
         Debug.Assert(Monitor.IsEntered(SyncRoot));
 
         var node = pool.Get();
         TLockManager.InitializeNode(node);
-        node.Initialize(this, throwOnTimeout);
+        node.Initialize(this, flags);
         EnqueueNode(node);
         return node;
     }
@@ -226,7 +226,7 @@ private protected ValueTask AcquireAsync<TNode, TLockManager, TOptions>(ref Valu
                         break;
                     }
 
-                    factory = EnqueueNode(ref pool, ref manager, throwOnTimeout: true);
+                    factory = EnqueueNode<TNode, TLockManager>(ref pool, WaitNodeFlags.ThrowOnTimeout);
                 }
 
                 interruptedCallers?.Unwind();
@@ -296,7 +296,7 @@ private protected ValueTask<bool> TryAcquireAsync<TNode, TLockManager, TOptions>
                         break;
                     }
 
-                    factory = EnqueueNode(ref pool, ref manager, throwOnTimeout: false);
+                    factory = EnqueueNode<TNode, TLockManager>(ref pool, WaitNodeFlags.None);
                 }
 
                 interruptedCallers?.Unwind();
@@ -460,11 +460,18 @@ internal CallerInformationStorage(Func<object> callerInfoProvider)
         }
     }
 
+    [Flags]
+    internal enum WaitNodeFlags
+    {
+        None = 0,
+        ThrowOnTimeout = 1,
+    }
+
     private protected abstract class WaitNode : LinkedValueTaskCompletionSource<bool>
     {
         private readonly WeakReference<QueuedSynchronizer?> owner = new(target: null, trackResurrection: false);
         private Timestamp createdAt;
-        private bool throwOnTimeout;
+        private WaitNodeFlags flags;
 
         // stores information about suspended caller for debugging purposes
         internal object? CallerInfo
@@ -483,17 +490,18 @@ protected override void Cleanup()
         [DebuggerBrowsable(DebuggerBrowsableState.Never)]
         internal bool NeedsRemoval => CompletionData is null;
 
-        internal void Initialize(QueuedSynchronizer owner, bool throwOnTimeout)
+        internal void Initialize(QueuedSynchronizer owner, WaitNodeFlags flags)
         {
             Debug.Assert(owner is not null);
 
-            this.throwOnTimeout = throwOnTimeout;
+            this.flags = flags;
             this.owner.SetTarget(owner);
             CallerInfo = owner.callerInfo?.Capture();
             createdAt = new();
         }
 
-        protected sealed override Result<bool> OnTimeout() => throwOnTimeout ? base.OnTimeout() : false;
+        protected sealed override Result<bool> OnTimeout()
+            => (flags & WaitNodeFlags.ThrowOnTimeout) is not 0 ? base.OnTimeout() : false;
 
         private protected static void AfterConsumed<T>(T node)
             where T : WaitNode, IPooledManualResetCompletionSource<Action<T>>
@@ -806,13 +814,13 @@ private bool TryAcquireCore(TContext context)
         return false;
     }
 
-    private WaitNode EnqueueNode(TContext context, bool throwOnTimeout)
+    private WaitNode EnqueueNode(TContext context, WaitNodeFlags flags)
     {
         Debug.Assert(Monitor.IsEntered(SyncRoot));
 
         var node = pool.Get();
         node.Context = context;
-        node.Initialize(this, throwOnTimeout);
+        node.Initialize(this, flags);
         EnqueueNode(node);
         return node;
     }
@@ -868,7 +876,7 @@ protected ValueTask<bool> TryAcquireAsync(TContext context, TimeSpan timeout, Ca
                         break;
                     }
 
-                    factory = EnqueueNode(context, throwOnTimeout: false);
+                    factory = EnqueueNode(context, WaitNodeFlags.None);
                 }
 
                 task = factory.Invoke(timeout, token);
@@ -932,7 +940,7 @@ protected ValueTask AcquireAsync(TContext context, TimeSpan timeout, Cancellatio
                         break;
                     }
 
-                    factory = EnqueueNode(context, throwOnTimeout: true);
+                    factory = EnqueueNode(context, WaitNodeFlags.ThrowOnTimeout);
                 }
 
                 task = factory.Invoke(timeout, token);

diff --git a/src/cluster/DotNext.AspNetCore.Cluster/DotNext.AspNetCore.Cluster.csproj b/src/cluster/DotNext.AspNetCore.Cluster/DotNext.AspNetCore.Cluster.csproj
@@ -8,7 +8,7 @@
     <ImplicitUsings>true</ImplicitUsings>
     <IsTrimmable>true</IsTrimmable>
     <Features>nullablePublicOnly</Features>
-    <VersionPrefix>5.5.1</VersionPrefix>
+    <VersionPrefix>5.6.0</VersionPrefix>
     <VersionSuffix></VersionSuffix>
     <Authors>.NET Foundation and Contributors</Authors>
     <Product>.NEXT Family of Libraries</Product>

diff --git a/src/cluster/DotNext.Net.Cluster/DotNext.Net.Cluster.csproj b/src/cluster/DotNext.Net.Cluster/DotNext.Net.Cluster.csproj
@@ -8,7 +8,7 @@
     <Nullable>enable</Nullable>
     <IsTrimmable>true</IsTrimmable>
     <Features>nullablePublicOnly</Features>
-    <VersionPrefix>5.5.1</VersionPrefix>
+    <VersionPrefix>5.6.0</VersionPrefix>
     <VersionSuffix></VersionSuffix>
     <Authors>.NET Foundation and Contributors</Authors>
     <Product>.NEXT Family of Libraries</Product>