Merge branch 'NethermindEth:master' into master

CONTRIBUTING.md

@@ -1,6 +1,6 @@
 # Contributing to Nethermind
 
-The Nethermind team maintains guidelines for contributing to the Nethermind repos. Check out our [docs page](https://docs.nethermind.io/nethermind/) for more info about us.
+The Nethermind team maintains guidelines for contributing to the Nethermind repos. Check out our [docs page](https://docs.nethermind.io/) for more info about us.
 
 ### Code of Conduct
 

src/Nethermind/Nethermind.Analytics/AnalyticsWebSocketsModule.cs

@@ -30,7 +30,7 @@ public AnalyticsWebSocketsModule(IJsonSerializer jsonSerializer, ILogManager log
 
         public ISocketsClient CreateClient(WebSocket webSocket, string clientName, HttpContext httpContext)
         {
-            SocketClient socketsClient = new SocketClient(clientName, new WebSocketHandler(webSocket, _logManager), _jsonSerializer);
+            SocketClient<WebSocketMessageStream> socketsClient = new(clientName, new WebSocketMessageStream(webSocket, _logManager), _jsonSerializer);
             _clients.TryAdd(socketsClient.Id, socketsClient);
 
             return socketsClient;

src/Nethermind/Nethermind.Api/IApiWithBlockchain.cs

@@ -12,6 +12,7 @@
 using Nethermind.Consensus.Processing;
 using Nethermind.Consensus.Producers;
 using Nethermind.Consensus.Rewards;
+using Nethermind.Consensus.Scheduler;
 using Nethermind.Consensus.Validators;
 using Nethermind.Core;
 using Nethermind.Evm.TransactionProcessing;
@@ -94,5 +95,6 @@ public interface IApiWithBlockchain : IApiWithStores, IBlockchainBridgeFactory
         CompositePruningTrigger PruningTrigger { get; }
 
         IBlockProductionPolicy? BlockProductionPolicy { get; set; }
+        BackgroundTaskScheduler BackgroundTaskScheduler { get; set; }
     }
 }

src/Nethermind/Nethermind.Api/IInitConfig.cs

@@ -85,6 +85,9 @@ public interface IInitConfig : IConfig
 
     [ConfigItem(Description = "[TECHNICAL] Exit when block number is reached. Useful for scripting and testing.", DefaultValue = "null", HiddenFromDocs = true)]
     long? ExitOnBlockNumber { get; set; }
+
+    [ConfigItem(Description = "[TECHNICAL] Specify concurrency limit for background task.", DefaultValue = "1", HiddenFromDocs = true)]
+    int BackgroundTaskConcurrency { get; set; }
 }
 
 public enum DiagnosticMode

src/Nethermind/Nethermind.Api/InitConfig.cs

@@ -35,6 +35,7 @@ public class InitConfig : IInitConfig
         public bool DisableGcOnNewPayload { get; set; } = true;
         public bool DisableMallocOpts { get; set; } = false;
         public long? ExitOnBlockNumber { get; set; } = null;
+        public int BackgroundTaskConcurrency { get; set; } = 1;
 
         [Obsolete("Use DiagnosticMode with MemDb instead")]
         public bool UseMemDb

src/Nethermind/Nethermind.Api/NethermindApi.cs

@@ -19,6 +19,7 @@
 using Nethermind.Consensus.Processing;
 using Nethermind.Consensus.Producers;
 using Nethermind.Consensus.Rewards;
+using Nethermind.Consensus.Scheduler;
 using Nethermind.Consensus.Validators;
 using Nethermind.Core;
 using Nethermind.Core.Authentication;
@@ -209,6 +210,7 @@ public ISealEngine SealEngine
 
         public IEthSyncingInfo? EthSyncingInfo { get; set; }
         public IBlockProductionPolicy? BlockProductionPolicy { get; set; }
+        public BackgroundTaskScheduler BackgroundTaskScheduler { get; set; } = null!;
         public IWallet? Wallet { get; set; }
         public IBlockStore? BadBlocksStore { get; set; }
         public ITransactionComparerProvider? TransactionComparerProvider { get; set; }

src/Nethermind/Nethermind.Blockchain.Test/FullPruning/CopyTreeVisitorTests.cs

@@ -1,8 +1,10 @@
 // SPDX-FileCopyrightText: 2022 Demerzel Solutions Limited
 // SPDX-License-Identifier: LGPL-3.0-only
 
+using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using FluentAssertions;
 using Nethermind.Blockchain.FullPruning;
@@ -17,6 +19,7 @@
 using Nethermind.Trie;
 using Nethermind.Trie.Pruning;
 using NSubstitute;
+using NSubstitute.ExceptionExtensions;
 using NUnit.Framework;
 
 namespace Nethermind.Blockchain.Test.FullPruning
@@ -41,7 +44,7 @@ public void copies_state_between_dbs(int fullPruningMemoryBudgetMb, int maxDegre
             };
 
             IPruningContext ctx = StartPruning(trieDb, clonedDb);
-            CopyDb(ctx, trieDb, visitingOptions, writeFlags: WriteFlags.LowPriority);
+            CopyDb(ctx, CancellationToken.None, trieDb, visitingOptions, writeFlags: WriteFlags.LowPriority);
 
             List<byte[]> keys = trieDb.Keys.ToList();
             List<byte[]> values = trieDb.Values.ToList();
@@ -57,28 +60,29 @@ public void copies_state_between_dbs(int fullPruningMemoryBudgetMb, int maxDegre
         }
 
         [Test, Timeout(Timeout.MaxTestTime)]
-        public async Task cancel_coping_state_between_dbs()
+        public void cancel_coping_state_between_dbs()
         {
             MemDb trieDb = new();
             MemDb clonedDb = new();
             IPruningContext pruningContext = StartPruning(trieDb, clonedDb);
-            Task task = Task.Run(() => CopyDb(pruningContext, trieDb));
 
-            pruningContext.CancellationTokenSource.Cancel();
+            CancellationTokenSource cts = new CancellationTokenSource();
+            cts.Cancel();
 
-            await task;
+            CopyDb(pruningContext, cts.Token, trieDb);
 
             clonedDb.Count.Should().BeLessThan(trieDb.Count);
         }
 
-        private static IPruningContext CopyDb(IPruningContext pruningContext, MemDb trieDb, VisitingOptions? visitingOptions = null, WriteFlags writeFlags = WriteFlags.None)
+        private static IPruningContext CopyDb(IPruningContext pruningContext, CancellationToken cancellationToken, MemDb trieDb, VisitingOptions? visitingOptions = null, WriteFlags writeFlags = WriteFlags.None)
         {
             LimboLogs logManager = LimboLogs.Instance;
             PatriciaTree trie = Build.A.Trie(trieDb).WithAccountsByIndex(0, 100).TestObject;
             IStateReader stateReader = new StateReader(new TrieStore(trieDb, logManager), new MemDb(), logManager);
 
-            using CopyTreeVisitor copyTreeVisitor = new(pruningContext, writeFlags, logManager);
+            using CopyTreeVisitor copyTreeVisitor = new(pruningContext, writeFlags, logManager, cancellationToken);
             stateReader.RunTreeVisitor(copyTreeVisitor, trie.RootHash, visitingOptions);
+            copyTreeVisitor.Finish();
             return pruningContext;
         }
 

src/Nethermind/Nethermind.Blockchain.Test/FullPruning/FullPrunerTests.cs

@@ -30,7 +30,7 @@ namespace Nethermind.Blockchain.Test.FullPruning
     [TestFixture(0, 4)]
     [TestFixture(1, 1)]
     [TestFixture(1, 4)]
-    [Parallelizable(ParallelScope.All)]
+    [Parallelizable(ParallelScope.Children)]
     public class FullPrunerTests
     {
         private readonly int _fullPrunerMemoryBudgetMb;
@@ -46,16 +46,15 @@ public FullPrunerTests(int fullPrunerMemoryBudgetMb, int degreeOfParallelism)
         public async Task can_prune()
         {
             TestContext test = CreateTest();
-            bool contextDisposed = await test.WaitForPruning();
-            contextDisposed.Should().BeTrue();
+            await test.RunFullPruning();
             test.ShouldCopyAllValues();
         }
 
         [Test, Timeout(Timeout.MaxTestTime)]
         public async Task pruning_deletes_old_db_on_success()
         {
             TestContext test = CreateTest(clearPrunedDb: true);
-            await test.WaitForPruning();
+            await test.RunFullPruning();
             test.TrieDb.Count.Should().Be(0);
         }
 
@@ -64,15 +63,15 @@ public async Task pruning_keeps_old_db_on_fail()
         {
             TestContext test = CreateTest(false);
             int count = test.TrieDb.Count;
-            await test.WaitForPruning();
+            await test.RunFullPruning();
             test.TrieDb.Count.Should().Be(count);
         }
 
         [Test, Timeout(Timeout.MaxTestTime)]
         public async Task pruning_deletes_new_db_on_fail()
         {
             TestContext test = CreateTest(false);
-            await test.WaitForPruning();
+            await test.RunFullPruning();
             test.CopyDb.Count.Should().Be(0);
         }
 
@@ -81,8 +80,7 @@ public async Task pruning_keeps_new_db_on_success()
         {
             TestContext test = CreateTest();
             int count = test.TrieDb.Count;
-            bool result = await test.WaitForPruning();
-            result.Should().BeTrue();
+            await test.RunFullPruning();
             test.CopyDb.Count.Should().Be(count);
         }
 
@@ -96,7 +94,7 @@ public async Task pruning_keeps_new_db_on_success()
         public async Task pruning_shuts_down_node(bool success, FullPruningCompletionBehavior behavior, bool expectedShutdown)
         {
             TestContext test = CreateTest(successfulPruning: success, completionBehavior: behavior);
-            await test.WaitForPruning();
+            await test.RunFullPruning();
 
             if (expectedShutdown)
             {
@@ -114,7 +112,8 @@ public async Task can_not_start_pruning_when_other_is_in_progress()
             TestContext test = CreateTest();
             test.FullPruningDb.CanStartPruning.Should().BeTrue();
 
-            var pruningContext = test.WaitForPruningStart();
+            test.TriggerPruningViaEvent();
+            TestFullPruningDb.TestPruningContext pruningContext = await test.WaitForPruningStart();
             test.FullPruningDb.CanStartPruning.Should().BeFalse();
             await test.WaitForPruningEnd(pruningContext);
 
@@ -125,21 +124,23 @@ public async Task can_not_start_pruning_when_other_is_in_progress()
         public async Task should_not_start_multiple_pruning()
         {
             TestContext test = CreateTest();
-            test.PruningTrigger.Prune += Raise.Event<EventHandler<PruningTriggerEventArgs>>();
-            await test.WaitForPruning();
+            test.TriggerPruningViaEvent();
+            TestFullPruningDb.TestPruningContext ctx = await test.WaitForPruningStart();
+            test.TriggerPruningViaEvent();
+            await test.WaitForPruningEnd(ctx);
             test.FullPruningDb.PruningStarted.Should().Be(1);
         }
 
         [Test, Timeout(Timeout.MaxTestTime)]
         public async Task should_duplicate_writes_while_pruning()
         {
             TestContext test = CreateTest();
-            test.WaitForPruningStart();
+            TestFullPruningDb.TestPruningContext ctx = await test.WaitForPruningStart();
             byte[] key = { 1, 2, 3 };
             test.FullPruningDb[key] = key;
             test.FullPruningDb.Context.WaitForFinish.Set();
-            await test.FullPruningDb.Context.DisposeEvent.WaitOneAsync(TimeSpan.FromMilliseconds(Timeout.MaxWaitTime), CancellationToken.None);
 
+            await test.WaitForPruningEnd(ctx);
             test.FullPruningDb[key].Should().BeEquivalentTo(key);
         }
 
@@ -148,7 +149,7 @@ public async Task should_duplicate_writes_to_batches_while_pruning()
         {
             TestContext test = CreateTest();
             byte[] key = { 0, 1, 2 };
-            TestFullPruningDb.TestPruningContext context = test.WaitForPruningStart();
+            TestFullPruningDb.TestPruningContext context = await test.WaitForPruningStart();
 
             using (IWriteBatch writeBatch = test.FullPruningDb.StartWriteBatch())
             {
@@ -204,33 +205,52 @@ public TestContext(
                     FullPruningMaxDegreeOfParallelism = degreeOfParallelism,
                     FullPruningMemoryBudgetMb = fullScanMemoryBudgetMb,
                     FullPruningCompletionBehavior = completionBehavior
-                }, BlockTree, StateReader, ProcessExitSource, _chainEstimations, DriveInfo, LimboLogs.Instance);
+                }, BlockTree, StateReader, ProcessExitSource, _chainEstimations, DriveInfo, Substitute.For<IPruningTrieStore>(), LimboLogs.Instance);
             }
 
-            public async Task<bool> WaitForPruning()
+            public async Task RunFullPruning()
             {
-                TestFullPruningDb.TestPruningContext context = WaitForPruningStart();
-                bool result = await WaitForPruningEnd(context);
-                if (result && _clearPrunedDb)
-                {
-                    await FullPruningDb.WaitForClearDb.WaitOneAsync(TimeSpan.FromMilliseconds(Timeout.MaxWaitTime * 3), CancellationToken.None);
-                }
+                TestFullPruningDb.TestPruningContext ctx = await WaitForPruningStart();
+                await WaitForPruningEnd(ctx);
+            }
 
-                return result;
+            public void TriggerPruningViaEvent()
+            {
+                PruningTrigger.Prune += Raise.Event<EventHandler<PruningTriggerEventArgs>>();
             }
 
             public async Task<bool> WaitForPruningEnd(TestFullPruningDb.TestPruningContext context)
             {
-                await Task.Yield();
-                await context.WaitForFinish.WaitOneAsync(TimeSpan.FromMilliseconds(Timeout.MaxWaitTime * 5), CancellationToken.None);
+                while (!await context.WaitForFinish.WaitOneAsync(TimeSpan.FromMilliseconds(1), CancellationToken.None))
+                {
+                    AddBlocks(1);
+                }
                 AddBlocks(1);
                 return await context.DisposeEvent.WaitOneAsync(TimeSpan.FromMilliseconds(Timeout.MaxWaitTime * 5), CancellationToken.None);
             }
 
-            public TestFullPruningDb.TestPruningContext WaitForPruningStart()
+            public async Task<TestFullPruningDb.TestPruningContext> WaitForPruningStart()
             {
-                PruningTrigger.Prune += Raise.Event<EventHandler<PruningTriggerEventArgs>>();
-                AddBlocks(Reorganization.MaxDepth + 2);
+                TriggerPruningViaEvent();
+                using CancellationTokenSource cts = new CancellationTokenSource();
+                Task addBlockTasks = Task.Run(() =>
+                {
+                    while (!cts.IsCancellationRequested)
+                    {
+                        AddBlocks(1);
+                    }
+                });
+
+                try
+                {
+                    Assert.That(() => FullPruningDb.Context, Is.Not.Null.After(1000, 1));
+                }
+                finally
+                {
+                    await cts.CancelAsync();
+                    await addBlockTasks;
+                }
+
                 TestFullPruningDb.TestPruningContext context = FullPruningDb.Context;
                 return context;
             }
@@ -245,6 +265,7 @@ public void AddBlocks(long count)
                     BlockTree.Head.Returns(head);
                     BlockTree.FindHeader(number).Returns(head.Header);
                     BlockTree.OnUpdateMainChain += Raise.EventWith(new OnUpdateMainChainArgs(new List<Block>() { head }, true));
+                    Thread.Sleep(1); // Need to add a little sleep as the wait for event in full pruner is async.
                 }
             }
 
@@ -321,6 +342,11 @@ public byte[]? this[ReadOnlySpan<byte> key]
                     set => _context[key] = value;
                 }
 
+                public IWriteBatch StartWriteBatch()
+                {
+                    return _context.StartWriteBatch();
+                }
+
                 public void Set(ReadOnlySpan<byte> key, byte[]? value, WriteFlags flags = WriteFlags.None)
                 {
                     _context.Set(key, value, flags);