evmonly: parallelize OCC validation and merge behind the serial acceptance barrier

giga/evmonly/README.md

@@ -217,10 +217,21 @@ are not enabled, the executor attempts optimistic parallel execution. Initial
 incarnations are split into execution ranges and run through the shared OCC
 worker pool against the base state. Worker fan-out is clamped to the amount of
 available work, so small blocks do not spawn idle workers and can still split
-down to one transaction per range. Validation then walks transaction order,
-comparing each incarnation's recorded balance, nonce, code, account, and
-`(address, slot)` storage reads/writes against writes accepted after that
-incarnation's source prefix.
+down to one transaction per range. Validation then accepts transactions in
+block order, comparing each incarnation's recorded balance, nonce, code,
+account, and `(address, slot)` storage reads/writes against writes recorded
+between that incarnation's source prefix and its own index.
+
+Acceptance is the serial barrier, but the work behind it is spread across the
+pool: every incarnation's writes are indexed by transaction index up front, a
+parallel pass checks the pending run of transactions against that index and
+reports the first one the frontier would not accept, the accepted run is folded
+into the prefix shard by shard (shards are contiguous address ranges), and the
+frontier handles only the reported transaction on the calling goroutine. Each
+incarnation records the set of shards it touched, so a worker skips whole
+results that hold nothing of its own. The merge emits the changeset one shard at
+a time on the pool and concatenates the shards in order, which is canonical
+address order; a prefix with few keys is merged on the calling goroutine instead.
 
 - transactions with no dependency on newly accepted prior writes are retained
   and accepted in block order without rerunning

giga/evmonly/account_reader_test.go

@@ -1,6 +1,7 @@
 package evmonly
 
 import (
+	"bytes"
 	"math/big"
 	"sync/atomic"
 	"testing"
@@ -87,48 +88,36 @@ func TestSnapshotReaderFetchesCodeOnlyWhenTheAccountHasSome(t *testing.T) {
 	require.Equal(t, []byte{0x60, 0x00}, account.Code)
 }
 
-// The merge reads every account it is about to compare against through the pool, then the serial
-// comparison finds them already resolved.
-func TestPrefetchResolvesEveryTouchedAccountOnce(t *testing.T) {
+// The merge compares balance, nonce and code against the same row, and resolves that row once per
+// touched account across the pool's workers.
+func TestParallelMergeResolvesEveryTouchedAccountOnce(t *testing.T) {
 	snapshot := newMemoryGigaSnapshot(7)
-	addrs := make([]common.Address, 0, minPrefetchedAccounts+8)
-	for i := range minPrefetchedAccounts + 8 {
-		addr := common.BigToAddress(big.NewInt(int64(i) + 1))
+	addrs := make([]common.Address, 0, 512)
+	for i := range 512 {
+		addr := common.BigToAddress(new(big.Int).Lsh(big.NewInt(int64(i)+1), 150))
 		snapshot.setBalance(addr, big.NewInt(int64(i)+1))
+		snapshot.nonces[addr] = uint64(i) //nolint:gosec // i is non-negative.
 		addrs = append(addrs, addr)
 	}
 	reading := &accountReadingSnapshot{memoryGigaSnapshot: snapshot}
 
 	state := newBlockSTMState(gigaSnapshotStateReader{snapshot: reading})
 	for i, addr := range addrs {
-		state.balances[addr] = big.NewInt(int64(i) + 100)
+		state.shard(addr).balances[addr] = big.NewInt(int64(i) + 100)
+		state.shard(addr).nonces[addr] = uint64(i) + 1 //nolint:gosec // i is non-negative.
 	}
-	state.prefetchBaseAccounts(t.Context(), newOCCWorkerPool(4))
-
-	require.Len(t, state.prefetched, len(addrs))
-	for i, addr := range addrs {
-		require.Equal(t, big.NewInt(int64(i)+1), state.prefetched[addr].Balance)
-	}
-
-	// The comparison that follows reads the prefetched rows rather than the view.
-	before := reading.reads.Load()
-	base := newBaseAccounts(state.source, state.prefetched)
-	for _, addr := range addrs {
-		base.balance(addr)
+	pool := newOCCWorkerPool(4)
+	defer pool.Close()
+
+	var changes StateChangeSet
+	require.NoError(t, state.changeSetIntoParallel(t.Context(), pool, &changes))
+
+	require.Equal(t, int64(len(addrs)), reading.reads.Load(), "each row must be read once")
+	require.Equal(t, state.ChangeSet(), changes, "the parallel merge must match the serial one")
+	require.Equal(t, int64(2*len(addrs)), reading.reads.Load(), "the serial merge reads each row once too")
+	require.Len(t, changes.Balances, len(addrs))
+	require.Len(t, changes.Nonces, len(addrs))
+	for i := 1; i < len(changes.Balances); i++ {
+		require.Negative(t, bytes.Compare(changes.Balances[i-1].Address[:], changes.Balances[i].Address[:]))
 	}
-	require.Equal(t, before, reading.reads.Load(), "the merge must not re-read what was prefetched")
-}
-
-// Below the threshold the pool costs more than the reads it saves, so the merge reads them itself.
-func TestPrefetchIsSkippedForASmallBlock(t *testing.T) {
-	snapshot := newMemoryGigaSnapshot(7)
-	addr := testAddress(0xd4)
-	snapshot.setBalance(addr, big.NewInt(5))
-	reading := &accountReadingSnapshot{memoryGigaSnapshot: snapshot}
-
-	state := newBlockSTMState(gigaSnapshotStateReader{snapshot: reading})
-	state.balances[addr] = big.NewInt(6)
-	state.prefetchBaseAccounts(t.Context(), newOCCWorkerPool(4))
-
-	require.Nil(t, state.prefetched)
 }

giga/evmonly/cmd/evmonly-loadtest/main_test.go

@@ -951,6 +951,10 @@ func BenchmarkExecuteTransferBlock(b *testing.B) {
 			name: "same_sender_nonce_chain",
 			args: []string{"--same-sender"},
 		},
+		{
+			name: "erc20_single_contract",
+			args: []string{"--workload=" + workloadERC20Transfer},
+		},
 	}
 	for _, tc := range tests {
 		b.Run(tc.name, func(b *testing.B) {
@@ -965,7 +969,7 @@ func BenchmarkExecuteTransferBlock(b *testing.B) {
 			require.NoError(b, err)
 
 			state := newGeneratedState()
-			workload, err := scenarios.NewTransferWorkload(scenarioConfig(cfg), state)
+			workload, err := scenarios.NewWorkload(cfg.workload, scenarioConfig(cfg), state)
 			require.NoError(b, err)
 			request, err := workload.BuildBlock(b.Context(), 1)
 			require.NoError(b, err)

giga/evmonly/executor_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"errors"
+	"math"
 	"math/big"
 	"sync"
 	"testing"
@@ -729,6 +730,64 @@ func TestExecutorOCCConflictingTransfersMatchSequential(t *testing.T) {
 	require.Equal(t, big.NewInt(int64(txCount*3)), occState.GetBalance(recipient))
 }
 
+// A block large enough for the parallel validation pass and the sharded merge, mixing independent
+// transfers, nonce chains from one sender, and a hot recipient, must produce the sequential result
+// exactly: transaction results, receipts, cumulative gas and the canonical changeset.
+func TestExecutorOCCLargeMixedBlockMatchesSequential(t *testing.T) {
+	chainID := big.NewInt(testChainID)
+	hot := testAddress(0xaa)
+	seqState := NewMemoryState()
+	occState := NewMemoryState()
+	var rawTxs [][]byte
+	sign := func(key *ecdsa.PrivateKey, nonce uint64, to common.Address, value int64) {
+		rawTxs = append(rawTxs, signLegacyTxWithGasPrice(t, key, chainID, nonce, &to, big.NewInt(value), nil, 100_000, big.NewInt(1)))
+	}
+	fund := func(key *ecdsa.PrivateKey) {
+		sender := crypto.PubkeyToAddress(key.PublicKey)
+		seqState.SetBalance(sender, big.NewInt(1_000_000_000))
+		occState.SetBalance(sender, big.NewInt(1_000_000_000))
+	}
+	for i := range 300 {
+		key, err := crypto.GenerateKey()
+		require.NoError(t, err)
+		fund(key)
+		switch {
+		case i%7 == 0:
+			sign(key, 0, hot, 3)
+		case i%29 == 0:
+			for nonce := range 3 {
+				sign(key, uint64(nonce), common.BigToAddress(big.NewInt(int64(50_000+i))), 5) //nolint:gosec // nonce is non-negative.
+			}
+		default:
+			sign(key, 0, common.BigToAddress(big.NewInt(int64(50_000+i))), 7)
+		}
+	}
+	// A late transaction reads the hot balance that the earlier ones credit, so it is rerun after the
+	// parallel pass has accepted the run before it.
+	hotKey, err := crypto.GenerateKey()
+	require.NoError(t, err)
+	hotSender := crypto.PubkeyToAddress(hotKey.PublicKey)
+	seqState.SetBalance(hotSender, big.NewInt(1_000_000_000))
+	occState.SetBalance(hotSender, big.NewInt(1_000_000_000))
+	seqState.SetBalance(hot, big.NewInt(1))
+	occState.SetBalance(hot, big.NewInt(1))
+	sign(hotKey, 0, hot, 1)
+
+	req := BlockRequest{Context: blockContext(chainID), Txs: rawTxs}
+	seqResult, err := NewExecutor(Config{MinGasPrice: big.NewInt(0)}, withTestState(seqState)).ExecuteBlock(t.Context(), req)
+	require.NoError(t, err)
+	occResult, err := NewExecutor(Config{MinGasPrice: big.NewInt(0), OCCWorkers: 4}, withTestState(occState)).ExecuteBlock(t.Context(), req)
+	require.NoError(t, err)
+
+	require.True(t, occResult.OCCStats.Attempted)
+	require.False(t, occResult.OCCStats.Fallback, occResult.OCCStats.FallbackReason)
+	require.Greater(t, occResult.OCCStats.RerunCount, uint64(0))
+	require.Equal(t, seqResult.GasUsed, occResult.GasUsed)
+	require.Equal(t, seqResult.Txs, occResult.Txs)
+	require.Equal(t, seqResult.Receipts, occResult.Receipts)
+	require.Equal(t, seqResult.ChangeSet, occResult.ChangeSet)
+}
+
 func TestExecutorOCCFeePayingTransfersDoNotConflictOnCoinbase(t *testing.T) {
 	chainID := big.NewInt(testChainID)
 	txCount := 4
@@ -2114,7 +2173,7 @@ func TestStateDBGetCodeHashTracksCodelessAccountExistenceReads(t *testing.T) {
 		{kind: stateAccessBalance, address: eoa}: {},
 	})
 	validation := occValidationResult{}
-	accepted := validateSTMResultAgainstPrefix(&validation, writes, occTxExecution{gasLimit: 1, readSet: readSet}, 0, 10, 0)
+	accepted := validateSTMResultAgainstPrefix(&validation, writes, occTxExecution{gasLimit: 1, readSet: readSet}, 0, 10, 0, math.MaxInt)
 	require.False(t, accepted)
 	require.Equal(t, occFallbackReasonConflict, validation.fallbackReason)
 }
@@ -2226,7 +2285,7 @@ func TestValidateSTMConflictMatrix(t *testing.T) {
 			t.Fatalf("unknown access mode %q", access)
 		}
 		validation := occValidationResult{}
-		accepted := validateSTMResultAgainstPrefix(&validation, writes, result, 0, 10, 0)
+		accepted := validateSTMResultAgainstPrefix(&validation, writes, result, 0, 10, 0, math.MaxInt)
 		return accepted, validation
 	}
 
@@ -2268,7 +2327,7 @@ func TestValidateSTMConflictSourcePrefix(t *testing.T) {
 			readSet:  map[stateAccessKey]struct{}{key: {}},
 			gasLimit: 1,
 			gasUsed:  1,
-		}, 0, 10, sourcePrefix)
+		}, 0, 10, sourcePrefix, math.MaxInt)
 		return accepted, validation
 	}
 	cases := []struct {