Parallel transaction execution with BALs

src/Nethermind/Ethereum.Test.Base/BlockchainTestBase.cs

@@ -36,6 +36,7 @@
 using Nethermind.Merge.Plugin;
 using Nethermind.JsonRpc;
 using System.Reflection;
+using Nethermind.State;
 
 namespace Ethereum.Test.Base;
 
@@ -144,7 +145,7 @@ protected async Task<EthereumTestResult> RunTest(BlockchainTest test, Stopwatch?
         await using IContainer container = containerBuilder.Build();
 
         IMainProcessingContext mainBlockProcessingContext = container.Resolve<IMainProcessingContext>();
-        IWorldState stateProvider = mainBlockProcessingContext.WorldState;
+        IWorldState stateProvider = (mainBlockProcessingContext.WorldState as ParallelWorldState).Inner; // directly access underlying state
         BlockchainProcessor blockchainProcessor = (BlockchainProcessor)mainBlockProcessingContext.BlockchainProcessor;
         IBlockTree blockTree = container.Resolve<IBlockTree>();
         IBlockValidator blockValidator = container.Resolve<IBlockValidator>();

src/Nethermind/Nethermind.AuRa.Test/AuRaBlockProducerTests.cs

@@ -70,7 +70,7 @@ public Context()
                 {
                     Block block = c.Arg<Block>();
                     block.TrySetTransactions(TransactionSource.GetTransactions(BlockTree.Head!.Header, block.GasLimit).ToArray());
-                    return block;
+                    return (block, null);
                 });
                 StateProvider.HasStateForBlock(Arg.Any<BlockHeader>()).Returns(x => true);
                 InitProducer();
@@ -208,7 +208,7 @@ public async Task Does_not_produce_block_when_head_is_null()
         public async Task Does_not_produce_block_when_processing_fails()
         {
             Context context = new();
-            context.BlockchainProcessor.Process(Arg.Any<Block>(), ProcessingOptions.ProducingBlock, Arg.Any<IBlockTracer>(), Arg.Any<CancellationToken>()).Returns((Block)null);
+            context.BlockchainProcessor.Process(Arg.Any<Block>(), ProcessingOptions.ProducingBlock, Arg.Any<IBlockTracer>(), Arg.Any<CancellationToken>()).Returns((null, null));
             (await StartStop(context)).ShouldProduceBlocks(Quantity.None());
         }
 

src/Nethermind/Nethermind.AuRa.Test/AuraBlockProcessorTests.cs

@@ -3,13 +3,15 @@
 
 using System;
 using System.Collections.Generic;
+using System.Threading.Tasks;
 using FluentAssertions;
 using Nethermind.Blockchain;
 using Nethermind.Blockchain.BeaconBlockRoot;
 using Nethermind.Blockchain.Receipts;
 using Nethermind.Blockchain.Test.Validators;
 using Nethermind.Blockchain.Tracing;
 using Nethermind.Consensus.AuRa;
+using Nethermind.Consensus.AuRa.Config;
 using Nethermind.Consensus.ExecutionRequests;
 using Nethermind.Consensus.Processing;
 using Nethermind.Consensus.Rewards;
@@ -37,13 +39,13 @@ namespace Nethermind.AuRa.Test
     public class AuraBlockProcessorTests
     {
         [Test]
-        public void Prepared_block_contains_author_field()
+        public async Task Prepared_block_contains_author_field()
         {
             BranchProcessor processor = CreateProcessor().Processor;
 
             BlockHeader header = Build.A.BlockHeader.WithAuthor(TestItem.AddressD).TestObject;
             Block block = Build.A.Block.WithHeader(header).TestObject;
-            Block[] processedBlocks = processor.Process(
+            Block[] processedBlocks = await processor.Process(
                 null,
                 new List<Block> { block },
                 ProcessingOptions.None,
@@ -83,17 +85,17 @@ public void For_normal_processing_it_should_not_fail_with_gas_remaining_rules()
                 .SignedAndResolved().WithChainId(105).WithGasPrice(0).WithValue(0).WithGasLimit(gasLimit + 1).TestObject;
             Block block = Build.A.Block.WithHeader(header).WithTransactions(new Transaction[] { tx })
                 .WithGasLimit(gasLimit).TestObject;
-            Assert.DoesNotThrow(() => processor.Process(
+            Assert.DoesNotThrowAsync(() => processor.Process(
                 null,
                 new List<Block> { block },
                 ProcessingOptions.None,
                 NullBlockTracer.Instance));
         }
 
         [Test]
-        public void Should_rewrite_contracts([Values] bool isPostMerge)
+        public async Task Should_rewrite_contracts([Values] bool isPostMerge)
         {
-            static BlockHeader Process(BranchProcessor auRaBlockProcessor, BlockHeader parent, IBlockTree blockTree, bool isPostMerge)
+            static async Task<BlockHeader> Process(BranchProcessor auRaBlockProcessor, BlockHeader parent, IBlockTree blockTree, bool isPostMerge)
             {
                 BlockHeader header = Build.A.BlockHeader
                     .WithAuthor(TestItem.AddressD)
@@ -102,11 +104,11 @@ static BlockHeader Process(BranchProcessor auRaBlockProcessor, BlockHeader paren
                     .WithTotalDifficulty(0).TestObject;
                 header.IsPostMerge = isPostMerge;
                 Block block = Build.A.Block.WithHeader(header).TestObject;
-                BlockHeader res = auRaBlockProcessor.Process(
+                BlockHeader res = (await auRaBlockProcessor.Process(
                     parent,
                     new List<Block> { block },
                     ProcessingOptions.None,
-                    NullBlockTracer.Instance)[0].Header;
+                    NullBlockTracer.Instance))[0].Header;
                 blockTree.Insert(res);
                 return res;
             }
@@ -156,7 +158,7 @@ static BlockHeader Process(BranchProcessor auRaBlockProcessor, BlockHeader paren
             }
 
             BlockHeader currentBlock = Build.A.BlockHeader.WithNumber(0).WithStateRoot(stateRoot).TestObject;
-            currentBlock = Process(processor, currentBlock, blockTree, isPostMerge);
+            currentBlock = await Process(processor, currentBlock, blockTree, isPostMerge);
 
             using (stateProvider.BeginScope(currentBlock))
             {
@@ -166,7 +168,7 @@ static BlockHeader Process(BranchProcessor auRaBlockProcessor, BlockHeader paren
                 stateProvider.GetCode(TestItem.AddressD).Should().BeEquivalentTo(Array.Empty<byte>());
             }
 
-            currentBlock = Process(processor, currentBlock, blockTree, isPostMerge);
+            currentBlock = await Process(processor, currentBlock, blockTree, isPostMerge);
 
             using (stateProvider.BeginScope(currentBlock))
             {
@@ -176,7 +178,7 @@ static BlockHeader Process(BranchProcessor auRaBlockProcessor, BlockHeader paren
                 stateProvider.GetCode(TestItem.AddressD).Should().BeEquivalentTo(Bytes.FromHexString("0x321"));
             }
 
-            currentBlock = Process(processor, currentBlock, blockTree, isPostMerge);
+            currentBlock = await Process(processor, currentBlock, blockTree, isPostMerge);
 
             using (stateProvider.BeginScope(currentBlock))
             {
@@ -189,14 +191,22 @@ static BlockHeader Process(BranchProcessor auRaBlockProcessor, BlockHeader paren
 
         private (BranchProcessor Processor, IWorldState StateProvider, IBlockTree blockTree) CreateProcessor(ITxFilter? txFilter = null, ContractRewriter? contractRewriter = null)
         {
-            IWorldState stateProvider = TestWorldStateFactory.CreateForTest();
+            IWorldState stateProvider = TestWorldStateFactory.CreateForTest(parallel: true);
             IBlockTree blockTree = Build.A.BlockTree(GnosisSpecProvider.Instance).TestObject;
             ITransactionProcessor transactionProcessor = Substitute.For<ITransactionProcessor>();
             AuRaBlockProcessor processor = new(
                 GnosisSpecProvider.Instance,
+                new AuRaChainSpecEngineParameters(),
                 TestBlockValidator.AlwaysValid,
                 NoBlockRewards.Instance,
-                new BlockProcessor.BlockValidationTransactionsExecutor(new ExecuteTransactionProcessorAdapter(transactionProcessor), stateProvider),
+                new BlockProcessor.BlockValidationTransactionsExecutor(
+                    stateProvider,
+                    new ExecuteTransactionProcessorAdapter(transactionProcessor),
+                    new BlobBaseFeeCalculator(),
+                    HoodiSpecProvider.Instance,
+                    Substitute.For<IBlockhashProvider>(),
+                    Substitute.For<ICodeInfoRepository>(),
+                    LimboLogs.Instance),
                 stateProvider,
                 NullReceiptStorage.Instance,
                 new BeaconBlockRootHandler(transactionProcessor, stateProvider),

src/Nethermind/Nethermind.Benchmark/Scheduler/BackgroundTaskSchedulerBenchmarks.cs

@@ -149,9 +149,9 @@ private sealed class StubBranchProcessor : IBranchProcessor
         public event EventHandler<BlockEventArgs>? BlockProcessing;
 #pragma warning restore CS0067
 
-        public Block[] Process(BlockHeader? baseBlock, IReadOnlyList<Block> suggestedBlocks,
+        public Task<Block[]> Process(BlockHeader? baseBlock, IReadOnlyList<Block> suggestedBlocks,
             ProcessingOptions processingOptions, IBlockTracer blockTracer, CancellationToken token = default)
-            => [];
+            => Task.FromResult(Array.Empty<Block>());
 
         public void RaiseBlocksProcessing() =>
             BlocksProcessing?.Invoke(this, new BlocksProcessingEventArgs([]));

src/Nethermind/Nethermind.Blockchain.Test/BeaconBlockRootHandlerTests.cs

@@ -94,7 +94,7 @@ public void Test_BeaconRootsAccessList_canInsertBeaconRootIsTrue_AccountExists()
     {
         BlockHeader header = Build.A.BlockHeader.WithNumber(1).WithParentBeaconBlockRoot(Hash256.Zero).TestObject;
         Block block = Build.A.Block.WithHeader(header).TestObject;
-        _worldState.AccountExists(Arg.Any<Address>()).Returns(true);
+        _worldState.AccountExists(Arg.Any<Address>(), Arg.Any<int?>()).Returns(true);
         (_, AccessList? accessList) = _beaconBlockRootHandler
             .BeaconRootsAccessList(block, Cancun.Instance, includeStorageCells: true);
 
@@ -108,7 +108,7 @@ public void Test_BeaconRootsAccessList_canInsertBeaconRootIsTrue_AccountExists_I
     {
         BlockHeader header = Build.A.BlockHeader.WithNumber(1).WithParentBeaconBlockRoot(Hash256.Zero).TestObject;
         Block block = Build.A.Block.WithHeader(header).TestObject;
-        _worldState.AccountExists(Arg.Any<Address>()).Returns(true);
+        _worldState.AccountExists(Arg.Any<Address>(), Arg.Any<int?>()).Returns(true);
         (_, AccessList? accessList) = _beaconBlockRootHandler
             .BeaconRootsAccessList(block, Cancun.Instance, false);
 
@@ -134,7 +134,7 @@ public void Test_StoreBeaconRoot_AccessListNotNull()
         BlockHeader header = Build.A.BlockHeader.WithNumber(1).WithParentBeaconBlockRoot(Hash256.Zero).TestObject;
         Block block = Build.A.Block.WithHeader(header).TestObject;
 
-        _worldState.AccountExists(Arg.Any<Address>()).Returns(true);
+        _worldState.AccountExists(Arg.Any<Address>(), Arg.Any<int?>()).Returns(true);
 
         _beaconBlockRootHandler.StoreBeaconRoot(block, Cancun.Instance, NullTxTracer.Instance);
 

src/Nethermind/Nethermind.Blockchain.Test/BlockProcessorTests.cs

@@ -45,12 +45,19 @@ private static (BlockProcessor processor, BranchProcessor branchProcessor, IWorl
         IRewardCalculator? rewardCalculator = null,
         IBlockCachePreWarmer? preWarmer = null)
     {
-        IWorldState stateProvider = TestWorldStateFactory.CreateForTest();
+        IWorldState stateProvider = TestWorldStateFactory.CreateForTest(parallel: true);
         ITransactionProcessor transactionProcessor = Substitute.For<ITransactionProcessor>();
         BlockProcessor processor = new(HoodiSpecProvider.Instance,
             TestBlockValidator.AlwaysValid,
             rewardCalculator ?? NoBlockRewards.Instance,
-            new BlockProcessor.BlockValidationTransactionsExecutor(new ExecuteTransactionProcessorAdapter(transactionProcessor), stateProvider),
+            new BlockProcessor.BlockValidationTransactionsExecutor(
+                    stateProvider,
+                    new ExecuteTransactionProcessorAdapter(transactionProcessor),
+                    new BlobBaseFeeCalculator(),
+                    HoodiSpecProvider.Instance,
+                    Substitute.For<IBlockhashProvider>(),
+                    Substitute.For<ICodeInfoRepository>(),
+                    LimboLogs.Instance),
             stateProvider,
             NullReceiptStorage.Instance,
             new BeaconBlockRootHandler(transactionProcessor, stateProvider),
@@ -71,13 +78,13 @@ private static (BlockProcessor processor, BranchProcessor branchProcessor, IWorl
     }
 
     [Test, MaxTime(Timeout.MaxTestTime)]
-    public void Prepared_block_contains_author_field()
+    public async Task Prepared_block_contains_author_field()
     {
         (_, BranchProcessor branchProcessor, _) = CreateProcessorAndBranch();
 
         BlockHeader header = Build.A.BlockHeader.WithAuthor(TestItem.AddressD).TestObject;
         Block block = Build.A.Block.WithHeader(header).TestObject;
-        Block[] processedBlocks = branchProcessor.Process(
+        Block[] processedBlocks = await branchProcessor.Process(
             null,
             new List<Block> { block },
             ProcessingOptions.None,
@@ -87,20 +94,20 @@ public void Prepared_block_contains_author_field()
     }
 
     [Test, MaxTime(Timeout.MaxTestTime)]
-    public void Recovers_state_on_cancel()
+    public async Task Recovers_state_on_cancel()
     {
         (_, BranchProcessor branchProcessor, _) = CreateProcessorAndBranch(
             rewardCalculator: new RewardCalculator(MainnetSpecProvider.Instance));
 
         BlockHeader header = Build.A.BlockHeader.WithNumber(1).WithAuthor(TestItem.AddressD).TestObject;
         Block block = Build.A.Block.WithTransactions(1, MuirGlacier.Instance).WithHeader(header).TestObject;
-        Assert.Throws<OperationCanceledException>(() => branchProcessor.Process(
+        Assert.ThrowsAsync<OperationCanceledException>(() => branchProcessor.Process(
             null,
             new List<Block> { block },
             ProcessingOptions.None,
             AlwaysCancelBlockTracer.Instance));
 
-        Assert.Throws<OperationCanceledException>(() => branchProcessor.Process(
+        Assert.ThrowsAsync<OperationCanceledException>(() => branchProcessor.Process(
             null,
             new List<Block> { block },
             ProcessingOptions.None,
@@ -141,7 +148,7 @@ public async Task Process_long_running_branch(int blocksAmount)
 
 
     [Test, MaxTime(Timeout.MaxTestTime)]
-    public void TransactionsExecuted_event_fires_during_ProcessOne()
+    public async Task TransactionsExecuted_event_fires_during_ProcessOne()
     {
         (BlockProcessor processor, _, IWorldState stateProvider) = CreateProcessorAndBranch();
 
@@ -153,21 +160,21 @@ public void TransactionsExecuted_event_fires_during_ProcessOne()
         Block block = Build.A.Block.WithHeader(header).TestObject;
         IReleaseSpec spec = HoodiSpecProvider.Instance.GetSpec(block.Header);
 
-        processor.ProcessOne(block, ProcessingOptions.NoValidation, NullBlockTracer.Instance, spec, CancellationToken.None);
+        await processor.ProcessOne(block, ProcessingOptions.NoValidation, NullBlockTracer.Instance, spec, CancellationToken.None);
 
         eventFired.Should().BeTrue("TransactionsExecuted should fire after ProcessTransactions completes");
     }
 
     [Test, MaxTime(Timeout.MaxTestTime)]
-    public void BranchProcessor_cancels_prewarmer_via_TransactionsExecuted_event()
+    public async Task BranchProcessor_cancels_prewarmer_via_TransactionsExecuted_event()
     {
         TokenCapturingPreWarmer preWarmer = new();
         (_, BranchProcessor branchProcessor, _) = CreateProcessorAndBranch(preWarmer: preWarmer);
 
         BlockHeader header = Build.A.BlockHeader.WithAuthor(TestItem.AddressD).TestObject;
         Block block = Build.A.Block.WithHeader(header).WithTransactions(3, MuirGlacier.Instance).TestObject;
 
-        branchProcessor.Process(
+        await branchProcessor.Process(
             null,
             new List<Block> { block },
             ProcessingOptions.NoValidation,
@@ -178,14 +185,14 @@ public void BranchProcessor_cancels_prewarmer_via_TransactionsExecuted_event()
     }
 
     [Test, MaxTime(Timeout.MaxTestTime)]
-    public void BranchProcessor_unsubscribes_from_TransactionsExecuted_after_processing()
+    public async Task BranchProcessor_unsubscribes_from_TransactionsExecuted_after_processing()
     {
         (BlockProcessor processor, BranchProcessor branchProcessor, IWorldState stateProvider) = CreateProcessorAndBranch();
 
         BlockHeader header = Build.A.BlockHeader.WithAuthor(TestItem.AddressD).TestObject;
         Block block = Build.A.Block.WithHeader(header).TestObject;
 
-        branchProcessor.Process(
+        await branchProcessor.Process(
             null,
             new List<Block> { block },
             ProcessingOptions.NoValidation,
@@ -201,20 +208,20 @@ public void BranchProcessor_unsubscribes_from_TransactionsExecuted_after_process
         using IDisposable scope = stateProvider.BeginScope(null);
         Block block2 = Build.A.Block.WithHeader(Build.A.BlockHeader.WithAuthor(TestItem.AddressD).TestObject).TestObject;
         IReleaseSpec spec = HoodiSpecProvider.Instance.GetSpec(block2.Header);
-        processor.ProcessOne(block2, ProcessingOptions.NoValidation, NullBlockTracer.Instance, spec, CancellationToken.None);
+        await processor.ProcessOne(block2, ProcessingOptions.NoValidation, NullBlockTracer.Instance, spec, CancellationToken.None);
 
         externalHandlerCallCount.Should().Be(1, "only the externally subscribed handler should fire, BranchProcessor should have unsubscribed");
     }
 
     [Test, MaxTime(Timeout.MaxTestTime)]
-    public void BranchProcessor_no_prewarmer_still_processes_successfully()
+    public async Task BranchProcessor_no_prewarmer_still_processes_successfully()
     {
         (_, BranchProcessor branchProcessor, _) = CreateProcessorAndBranch(preWarmer: null);
 
         BlockHeader header = Build.A.BlockHeader.WithAuthor(TestItem.AddressD).TestObject;
         Block block = Build.A.Block.WithHeader(header).WithTransactions(3, MuirGlacier.Instance).TestObject;
 
-        Block[] processedBlocks = branchProcessor.Process(
+        Block[] processedBlocks = await branchProcessor.Process(
             null,
             new List<Block> { block },
             ProcessingOptions.NoValidation,

src/Nethermind/Nethermind.Blockchain.Test/BlockchainProcessorTests.cs

@@ -77,7 +77,7 @@ public void AllowToFail(Hash256 hash)
                 }
             }
 
-            public Block[] Process(BlockHeader? baseBlock, IReadOnlyList<Block> suggestedBlocks, ProcessingOptions processingOptions, IBlockTracer blockTracer, CancellationToken token)
+            public Task<Block[]> Process(BlockHeader? baseBlock, IReadOnlyList<Block> suggestedBlocks, ProcessingOptions processingOptions, IBlockTracer blockTracer, CancellationToken token)
             {
                 if (blockTracer != NullBlockTracer.Instance)
                 {
@@ -123,7 +123,7 @@ public Block[] Process(BlockHeader? baseBlock, IReadOnlyList<Block> suggestedBlo
                         else
                         {
                             _rootProcessed.Add(suggestedBlocks.Last().StateRoot!);
-                            return suggestedBlocks.ToArray();
+                            return Task.FromResult(suggestedBlocks.ToArray());
                         }
                     }
                 }

src/Nethermind/Nethermind.Blockchain.Test/Filters/TestProcessingContext.cs

@@ -6,6 +6,7 @@
 using System;
 using System.Collections.Generic;
 using System.Threading;
+using System.Threading.Tasks;
 using Nethermind.Consensus.Processing;
 using Nethermind.Core;
 using Nethermind.Evm.State;
@@ -23,9 +24,9 @@ internal class TestBranchProcessor : IBranchProcessor
     public event EventHandler<BlocksProcessingEventArgs>? BlocksProcessing { add { } remove { } }
     public event EventHandler<BlockEventArgs>? BlockProcessing { add { } remove { } }
 
-    public Block[] Process(BlockHeader? baseBlock, IReadOnlyList<Block> suggestedBlocks,
+    public Task<Block[]> Process(BlockHeader? baseBlock, IReadOnlyList<Block> suggestedBlocks,
         ProcessingOptions processingOptions, IBlockTracer blockTracer, CancellationToken token = default)
-        => [];
+        => Task.FromResult<Block[]>([]);
 
     public void RaiseBlockProcessed(BlockProcessedEventArgs args)
         => BlockProcessed?.Invoke(this, args);