Skip to content

Commit 0d6d5eb

Browse files
MarchhillMarchhill
authored
Optimise parallel execution: reduce worldstate, txprocessor allocations (#11318)
* optimise allocations of worldstates, txprocessors, bals * address minor comments, add comments for thread safety --------- Co-authored-by: Marc Harvey-Hill <10379486+Marchhill@users.noreply.github.com>
1 parent c5551cc commit 0d6d5eb

11 files changed

Lines changed: 333 additions & 98 deletions

File tree

‎src/Nethermind/Nethermind.Consensus/Processing/BlockAccessListManager.cs‎

Lines changed: 216 additions & 42 deletions
Large diffs are not rendered by default.

‎src/Nethermind/Nethermind.Consensus/Processing/BlockProcessor.ParallelBlockValidationTransactionsExecutor.cs‎

Lines changed: 17 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -102,22 +102,29 @@ private TxReceipt[] ProcessTransactionsParallel(Block block, ProcessingOptions p
102102
}
103103

104104
int txIndex = i - 1;
105+
Transaction tx = state.txs[txIndex];
105106
try
106107
{
107-
Transaction tx = state.txs[txIndex];
108-
ProcessTransaction(
109-
state.balManager.GetTxProcessor(i),
110-
state.stateProvider,
111-
state.block,
112-
tx,
113-
txIndex,
114-
state.receiptsTracers[txIndex],
115-
state.processingOptions);
108+
// The using block detaches the worker's BAL into _perTxBal[i] and
109+
// recycles the pool slot via Dispose BEFORE we signal the gas result,
110+
// so the validator finds _perTxBal[i] populated when it awaits
111+
// gasResults[i-1] — even if ProcessTransaction throws.
112+
using (TxProcessorLease lease = state.balManager.RentTxProcessor(i))
113+
{
114+
ProcessTransaction(
115+
lease.Adapter,
116+
state.stateProvider,
117+
state.block,
118+
tx,
119+
txIndex,
120+
state.receiptsTracers[txIndex],
121+
state.processingOptions);
122+
}
116123
state.gasResults[txIndex].SetResult((tx.BlockGasUsed, state.receiptsTracers[txIndex].BlockStateGasUsed, null));
117124
}
118125
catch (InvalidBlockException ex)
119126
{
120-
state.gasResults[txIndex].SetResult((state.txs[txIndex].GasLimit, 0, ex));
127+
state.gasResults[txIndex].SetResult((tx.GasLimit, 0, ex));
121128
}
122129
catch
123130
{

‎src/Nethermind/Nethermind.Consensus/Processing/IBlockAccessListManager.cs‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,17 @@ public interface IBlockAccessListManager
2626
ITransactionProcessorAdapter GetTxProcessor(int? balIndex = null);
2727
void NextTransaction();
2828
void Rollback();
29+
void ReturnTxProcessor(int balIndex);
30+
31+
/// <summary>
32+
/// Acquires a tx processor for <paramref name="balIndex"/> and returns a stack-only lease
33+
/// whose <c>Dispose</c> calls <see cref="ReturnTxProcessor"/>. Use with a <c>using</c>
34+
/// block so the pool slot is recycled (and the worker's BAL captured into <c>_perTxBal</c>)
35+
/// even on exception, without an explicit try/finally.
36+
/// </summary>
37+
TxProcessorLease RentTxProcessor(int balIndex)
38+
=> new(GetTxProcessor(balIndex), this, balIndex);
39+
2940
void IncrementalValidation(Block block, TaskCompletionSource<(long BlockGasUsed, long BlockStateGasUsed, InvalidBlockException? Exception)>[] gasResults, BlockReceiptsTracer[] receiptsTracers, BlockProcessor.BlockValidationTransactionsExecutor.ITransactionProcessedEventHandler? transactionProcessedEventHandler, CancellationToken token);
3041
void SetBlockAccessList(Block block);
3142
void ValidateBlockAccessList(Block block, ushort index, bool validateStorageReads = true);
@@ -35,3 +46,24 @@ public interface IBlockAccessListManager
3546
void ProcessExecutionRequests(Block block, TxReceipt[] txReceipts, IReleaseSpec spec);
3647
void ApplyAuRaPreprocessingChanges(IReleaseSpec spec, Address withdrawalContractAddress);
3748
}
49+
50+
/// <summary>
51+
/// Stack-only handle representing a borrowed tx processor from the pool. Dispose returns
52+
/// the slot via <see cref="IBlockAccessListManager.ReturnTxProcessor"/>. <c>readonly ref
53+
/// struct</c> so it never heap-allocates and can't escape the worker's stack frame.
54+
/// </summary>
55+
public readonly ref struct TxProcessorLease
56+
{
57+
public readonly ITransactionProcessorAdapter Adapter;
58+
private readonly IBlockAccessListManager _manager;
59+
private readonly int _balIndex;
60+
61+
internal TxProcessorLease(ITransactionProcessorAdapter adapter, IBlockAccessListManager manager, int balIndex)
62+
{
63+
Adapter = adapter;
64+
_manager = manager;
65+
_balIndex = balIndex;
66+
}
67+
68+
public void Dispose() => _manager.ReturnTxProcessor(_balIndex);
69+
}

‎src/Nethermind/Nethermind.Consensus/Processing/NullBlockAccessListManager.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ public void SetBlockExecutionContext(in BlockExecutionContext blockExecutionCont
3030
public ITransactionProcessorAdapter GetTxProcessor(int? balIndex = null) => throw new InvalidOperationException("NullBlockAccessListManager does not provide transaction processors.");
3131
public void NextTransaction() { }
3232
public void Rollback() { }
33+
public void ReturnTxProcessor(int balIndex) { }
3334
public void IncrementalValidation(Block block, TaskCompletionSource<(long BlockGasUsed, long BlockStateGasUsed, InvalidBlockException? Exception)>[] gasResults, BlockReceiptsTracer[] receiptsTracers, BlockProcessor.BlockValidationTransactionsExecutor.ITransactionProcessedEventHandler? transactionProcessedEventHandler, CancellationToken token) { }
3435
public void SetBlockAccessList(Block block) { }
3536
public void ValidateBlockAccessList(Block block, ushort index, bool validateStorageReads = true) { }

‎src/Nethermind/Nethermind.Core/BlockAccessLists/BlockAccessList.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,13 +8,14 @@
88
using System.Text.Json.Serialization;
99
using Nethermind.Int256;
1010
using Nethermind.Core.Collections;
11+
using Nethermind.Core.Resettables;
1112

1213
[assembly: InternalsVisibleTo("Nethermind.Core.Test")]
1314
[assembly: InternalsVisibleTo("Nethermind.State.Test")]
1415

1516
namespace Nethermind.Core.BlockAccessLists;
1617

17-
public class BlockAccessList : IEquatable<BlockAccessList>, IJournal<int>
18+
public class BlockAccessList : IEquatable<BlockAccessList>, IJournal<int>, IResettable
1819
{
1920
[JsonIgnore]
2021
public int Index = 0;

‎src/Nethermind/Nethermind.Core/Caching/StaticPool.cs‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -16,11 +16,21 @@ namespace Nethermind.Core.Caching;
1616
/// </typeparam>
1717
public static class StaticPool<T> where T : class, IResettable, new()
1818
{
19+
private const int DefaultMaxPooledCount = 4096;
20+
1921
/// <summary>
2022
/// Hard cap for the total number of items that can be stored in the shared pool.
21-
/// Prevents unbounded growth under bursty workloads while still allowing reuse.
23+
/// Defaults to <see cref="DefaultMaxPooledCount"/>; override per-T at startup via
24+
/// <see cref="SetMaxPooledCount"/>.
25+
/// </summary>
26+
private static int _maxPooledCount = DefaultMaxPooledCount;
27+
28+
/// <summary>
29+
/// Override the pool's hard cap for this <typeparamref name="T"/>. Intended to be called
30+
/// once at startup before any Rent/Return; there is no synchronization against in-flight
31+
/// callers, so reservations may briefly exceed a smaller new cap.
2232
/// </summary>
23-
private const int MaxPooledCount = 4096;
33+
public static void SetMaxPooledCount(int maxPooledCount) => _maxPooledCount = maxPooledCount;
2434

2535
/// <summary>
2636
/// Global pool shared between threads.
@@ -78,7 +88,7 @@ public static void Return(T item)
7888
{
7989
// We use Interlocked.Increment to reserve a slot up front.
8090
// This guarantees a bounded queue length without relying on slow Count().
81-
if (Interlocked.Increment(ref _poolCount) > MaxPooledCount)
91+
if (Interlocked.Increment(ref _poolCount) > _maxPooledCount)
8292
{
8393
// Roll back reservation if we'd exceed the cap.
8494
Interlocked.Decrement(ref _poolCount);

‎src/Nethermind/Nethermind.Evm.Test/Eip7928Tests.cs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,7 @@ public class Eip7928Tests(bool parallel) : VirtualMachineTestsBase
6262
{
6363
bool useParallel = parallelOverride ?? parallel;
6464
TracedAccessWorldState tracedState = new(TestState, parallel: useParallel);
65+
tracedState.SetGeneratingBlockAccessList(new BlockAccessList());
6566
ILogManager logManager = LimboLogs.Instance;
6667
IBlockhashProvider blockhashProvider = new TestBlockhashProvider(SpecProvider);
6768
EthereumCodeInfoRepository codeInfoRepo = new(tracedState);
@@ -979,6 +980,7 @@ private static IEnumerable<TestCaseData> ExceptionTestSource
979980
public void CodeInfoRepository_getcachedcodeinfo_records_account_read_in_bal(string address)
980981
{
981982
TracedAccessWorldState tracedState = new(TestState, parallel: parallel);
983+
tracedState.SetGeneratingBlockAccessList(new BlockAccessList());
982984

983985
CodeInfoRepository repo = new(tracedState, new EthereumPrecompileProvider());
984986

@@ -1037,6 +1039,7 @@ public void CodeInfoRepository_getcachedcodeinfo_delegated_records_account_read_
10371039
TestState.CommitTree(0);
10381040

10391041
TracedAccessWorldState tracedState = new(TestState, parallel: parallel);
1042+
tracedState.SetGeneratingBlockAccessList(new BlockAccessList());
10401043

10411044
CodeInfoRepository repo = new(tracedState, new EthereumPrecompileProvider());
10421045
CodeInfo result = repo.GetCachedCodeInfo(delegatedAccount, true, Amsterdam.Instance, out Address? delegationAddress);
@@ -1065,6 +1068,7 @@ public void CacheCodeInfoRepository_tracing_records_account_read_in_bal()
10651068
TestState.CommitTree(0);
10661069

10671070
TracedAccessWorldState tracedState = new(TestState, parallel: parallel);
1071+
tracedState.SetGeneratingBlockAccessList(new BlockAccessList());
10681072

10691073
CacheCodeInfoRepository repo = new(tracedState, new EthereumPrecompileProvider());
10701074
CodeInfo result = repo.GetCachedCodeInfo(TestItem.AddressB, false, Amsterdam.Instance, out Address? delegationAddress);

‎src/Nethermind/Nethermind.State.Test/BlockAccessListBasedWorldStateTests.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,8 @@ private static (BlockAccessListBasedWorldState bws, IDisposable scope) CreateBlo
6060
BlockAccessList suggestedBal = new();
6161
balSetup(suggestedBal);
6262

63-
BlockAccessListBasedWorldState bws = new(inner, blockAccessIndex, Logger);
63+
BlockAccessListBasedWorldState bws = new(inner, Logger);
64+
bws.SetBlockAccessIndex(blockAccessIndex);
6465
Block block = Build.A.Block.WithHeader(baseBlock).WithBlockAccessList(suggestedBal).TestObject;
6566
bws.Setup(block);
6667
IDisposable scope = bws.BeginScope(baseBlock);

‎src/Nethermind/Nethermind.State.Test/TracedAccessWorldStateTests.cs‎

Lines changed: 18 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ public class TracedAccessWorldStateTests(bool parallel)
5151

5252
BlockHeader baseBlock = Build.A.BlockHeader.WithStateRoot(stateRoot).WithNumber(0).TestObject;
5353
TracedAccessWorldState tws = new(inner, parallel: parallel);
54+
tws.SetGeneratingBlockAccessList(new());
5455
IDisposable scope = tws.BeginScope(baseBlock);
5556
tws.SetIndex(0);
5657
return (tws, scope);
@@ -74,7 +75,7 @@ public void BalanceOp_RecordsBalanceChange(
7475
tws.SubtractFromBalance(TestItem.AddressA, delta, Spec, out _);
7576
}
7677

77-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA);
78+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA);
7879
using (Assert.EnterMultipleScope())
7980
{
8081
Assert.That(ac, Is.Not.Null);
@@ -102,7 +103,7 @@ public void NonceOp_RecordsNonceChange(
102103
tws.SetNonce(TestItem.AddressA, value);
103104
}
104105

105-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA);
106+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA);
106107
using (Assert.EnterMultipleScope())
107108
{
108109
Assert.That(ac, Is.Not.Null);
@@ -123,7 +124,7 @@ public void InsertCode_RecordsCodeChange()
123124
ValueHash256 codeHash = ValueKeccak.Compute(code);
124125
tws.InsertCode(TestItem.AddressA, codeHash, code, Spec);
125126

126-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA);
127+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA);
127128
using (Assert.EnterMultipleScope())
128129
{
129130
Assert.That(ac, Is.Not.Null);
@@ -143,7 +144,7 @@ public void Set_RecordsStorageChange()
143144
{
144145
tws.Set(cell, [0x01]);
145146

146-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA);
147+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA);
147148
using (Assert.EnterMultipleScope())
148149
{
149150
Assert.That(ac, Is.Not.Null);
@@ -168,7 +169,7 @@ public void StorageRead_RecordsStorageRead(bool useGetOriginal)
168169
_ = tws.GetOriginal(cell);
169170
}
170171

171-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA);
172+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA);
172173
Assert.That(ac, Is.Not.Null);
173174
Assert.That(ac!.StorageReads, Has.Count.EqualTo(1));
174175
Assert.That(ac.StorageReads.First(), Is.EqualTo((UInt256)2));
@@ -259,7 +260,7 @@ public void ReadOp_RecordsAccountRead(Action<IWorldState> setup, Action<TracedAc
259260
using (Assert.EnterMultipleScope())
260261
{
261262
readAndAssert(tws);
262-
Assert.That(tws.GetGeneratingBlockAccessList().HasAccount(TestItem.AddressA), Is.True);
263+
Assert.That(tws.GetGeneratingBlockAccessList()!.HasAccount(TestItem.AddressA), Is.True);
263264
}
264265
}
265266
}
@@ -274,7 +275,7 @@ public void CreateAccount_RecordsChanges(
274275
{
275276
tws.CreateAccount(TestItem.AddressA, balance, nonce);
276277

277-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA);
278+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA);
278279
using (Assert.EnterMultipleScope())
279280
{
280281
Assert.That(ac, Is.Not.Null);
@@ -302,7 +303,7 @@ public void DeleteAccount_RecordsBalanceZeroed()
302303
{
303304
tws.DeleteAccount(TestItem.AddressA);
304305

305-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA);
306+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA);
306307
Assert.That(ac, Is.Not.Null);
307308
using (Assert.EnterMultipleScope())
308309
{
@@ -321,8 +322,8 @@ public void AddAccountRead_AddsAccountToBAL()
321322
{
322323
tws.AddAccountRead(TestItem.AddressA);
323324

324-
Assert.That(tws.GetGeneratingBlockAccessList().HasAccount(TestItem.AddressA), Is.True);
325-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA);
325+
Assert.That(tws.GetGeneratingBlockAccessList()!.HasAccount(TestItem.AddressA), Is.True);
326+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA);
326327
using (Assert.EnterMultipleScope())
327328
{
328329
Assert.That(ac!.BalanceChanges, Is.Empty);
@@ -341,13 +342,13 @@ public void TakeSnapshot_Restore_RollsBackBalanceChange()
341342
Snapshot snap = tws.TakeSnapshot();
342343
tws.AddToBalance(TestItem.AddressA, 50, Spec, out _);
343344

344-
Assert.That(tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA)!
345+
Assert.That(tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA)!
345346
.BalanceChanges, Has.Count.EqualTo(1));
346347

347348
tws.Restore(snap);
348349

349350
// Balance change must be rolled back by the snapshot restore.
350-
Assert.That(tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA)!
351+
Assert.That(tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA)!
351352
.BalanceChanges, Is.Empty);
352353
}
353354
}
@@ -360,7 +361,7 @@ public void SubtractFromBalance_DoesNotRecordSystemUserZeroChange()
360361
{
361362
tws.SubtractFromBalance(Address.SystemUser, 0u, Spec, out _);
362363

363-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(Address.SystemUser);
364+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(Address.SystemUser);
364365
Assert.That(ac, Is.Null);
365366
}
366367
}
@@ -381,7 +382,7 @@ public void RepeatedBalanceChanges_SameTx_UsesLatestBalValue()
381382
tws.AddToBalance(TestItem.AddressA, 100, Spec, out UInt256 oldBalance1);
382383
tws.AddToBalance(TestItem.AddressA, 100, Spec, out UInt256 oldBalance2);
383384

384-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA);
385+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA);
385386
using (Assert.EnterMultipleScope())
386387
{
387388
Assert.That(oldBalance1, Is.EqualTo((UInt256)1000), "first old balance from inner state");
@@ -404,7 +405,7 @@ public void RepeatedNonceChanges_SameTx_UsesLatestNonceValue()
404405
tws.IncrementNonce(TestItem.AddressA, 1, out UInt256 oldNonce1);
405406
tws.IncrementNonce(TestItem.AddressA, 1, out UInt256 oldNonce2);
406407

407-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA);
408+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA);
408409
using (Assert.EnterMultipleScope())
409410
{
410411
Assert.That(oldNonce1, Is.EqualTo((UInt256)0), "first old nonce from inner");
@@ -429,7 +430,7 @@ public void RepeatedCodeChanges_SameTx_UsesLatestCode()
429430
// Second InsertCode should see code1 as old code (from BAL), not empty (from inner)
430431
tws.InsertCode(TestItem.AddressA, ValueKeccak.Compute(code2), code2, Spec);
431432

432-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA);
433+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA);
433434
using (Assert.EnterMultipleScope())
434435
{
435436
Assert.That(ac, Is.Not.Null);
@@ -453,7 +454,7 @@ public void RepeatedStorageWrites_SameTx_UsesLatestValue_InParallel()
453454
tws.Set(cell, [0x01]);
454455
tws.Set(cell, [0x02]);
455456

456-
AccountChanges? ac = tws.GetGeneratingBlockAccessList().GetAccountChanges(TestItem.AddressA);
457+
AccountChanges? ac = tws.GetGeneratingBlockAccessList()!.GetAccountChanges(TestItem.AddressA);
457458
using (Assert.EnterMultipleScope())
458459
{
459460
Assert.That(ac, Is.Not.Null);

0 commit comments

Comments
 (0)