From 9dadc4fdeb265aff4c4a80181bb422a4021542b1 Mon Sep 17 00:00:00 2001 From: Brandon Chatham Date: Sat, 19 Sep 2026 17:08:38 +0000 Subject: [PATCH 1/4] evmonly: parallelize OCC validation and merge behind the serial acceptance barrier Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- giga/evmonly/README.md | 17 +- giga/evmonly/account_reader_test.go | 57 ++-- giga/evmonly/executor_test.go | 65 ++++- giga/evmonly/occ.go | 415 +++++++++++++++------------- giga/evmonly/occ_shards.go | 223 +++++++++++++++ 5 files changed, 547 insertions(+), 230 deletions(-) create mode 100644 giga/evmonly/occ_shards.go diff --git a/giga/evmonly/README.md b/giga/evmonly/README.md index 77bf3ee23d..e3a712d9e3 100644 --- a/giga/evmonly/README.md +++ b/giga/evmonly/README.md @@ -211,10 +211,19 @@ 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. The +merge emits the changeset one shard at a time on the pool and concatenates the +shards in order, which is canonical address order. - transactions with no dependency on newly accepted prior writes are retained and accepted in block order without rerunning diff --git a/giga/evmonly/account_reader_test.go b/giga/evmonly/account_reader_test.go index c5a4b739a1..e10c28a371 100644 --- a/giga/evmonly/account_reader_test.go +++ b/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) } diff --git a/giga/evmonly/executor_test.go b/giga/evmonly/executor_test.go index e2a109f633..87aaac9178 100644 --- a/giga/evmonly/executor_test.go +++ b/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 { diff --git a/giga/evmonly/occ.go b/giga/evmonly/occ.go index 19ce7b6e1e..ffffcd1e07 100644 --- a/giga/evmonly/occ.go +++ b/giga/evmonly/occ.go @@ -8,7 +8,6 @@ import ( "math" "math/big" "sort" - "sync/atomic" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" @@ -90,6 +89,9 @@ func (e *Executor) executeBlockOCC(ctx context.Context, req PreparedBlock, sourc } e.blockPhases.SetPhase("occ_merge") result, err := e.mergeOCCResults(ctx, results, finalState) + if errors.Is(err, errOCCWorkerPoolClosed) { + return e.executeBlockOCCSequentialFallback(ctx, req, source, validation, occFallbackReasonWorkerPoolClosed) + } if err != nil { return nil, err } @@ -299,8 +301,34 @@ func (e *Executor) validateBlockSTM( ) ([]occTxExecution, *blockSTMState, occValidationResult, error) { state := newBlockSTMValidationState(source) validation := occValidationResult{} + if err := state.writes.indexResults(ctx, pool, results); err != nil { + return nil, nil, validation, err + } + // The frontier alternates between a parallel pass, which accepts every result up to the first + // one that needs attention, and the serial frontier, which handles that one. A block whose + // transactions depend on each other one after another would make each parallel pass accept + // nothing, so after such a pass the serial frontier keeps going for a stretch that doubles each + // time it happens again. + serialUntil := 0 + serialStretch := occMinParallelValidation for state.nextToValidate < len(results) { - rerun, err := validateBlockSTMFrontier(ctx, runner, results, state, &validation) + if state.nextToValidate >= serialUntil { + accepted, err := e.acceptValidatedPrefix(ctx, runner, pool, results, state, &validation) + if err != nil { + return nil, nil, validation, err + } + if state.nextToValidate == len(results) { + break + } + if accepted < occMinParallelValidation { + serialUntil = state.nextToValidate + serialStretch + serialStretch *= 2 + } else { + serialStretch = occMinParallelValidation + } + } + end := min(len(results), max(serialUntil, state.nextToValidate+1)) + rerun, err := validateBlockSTMFrontier(ctx, runner, results, state, &validation, end) if err != nil { return nil, nil, validation, err } @@ -310,18 +338,25 @@ func (e *Executor) validateBlockSTM( if err := runner.runTasks(ctx, pool, []occExecutionTask{*rerun}, results); err != nil { return nil, nil, validation, err } + // The rerun's writes join the index; the previous incarnation's stay, which can only cost a + // later transaction a rerun it did not need, never miss a conflict. + state.writes.addAllAt(rerun.txIndex, results[rerun.txIndex].writeSet) + state.writes.addCommutativeBalanceDeltasAt(rerun.txIndex, results[rerun.txIndex].commutativeBalanceDeltas) } return results, state.prefix, validation, nil } +// validateBlockSTMFrontier accepts results in block order on the calling goroutine until it reaches +// end or a result that has to be rerun, which it returns as a task. func validateBlockSTMFrontier( ctx context.Context, runner occSpeculativeRunner, results []occTxExecution, state *blockSTMValidationState, validation *occValidationResult, + end int, ) (*occExecutionTask, error) { - for state.nextToValidate < len(results) { + for state.nextToValidate < end { if err := ctx.Err(); err != nil { return nil, err } @@ -362,8 +397,6 @@ func validateBlockSTMFrontier( } state.cumulativeGasUsed += result.gasUsed state.prefix.apply(result) - state.writes.addAllAt(txIndex, result.writeSet) - state.writes.addCommutativeBalanceDeltasAt(txIndex, result.commutativeBalanceDeltas) state.nextToValidate++ } return nil, nil @@ -391,7 +424,7 @@ func needsSTMRerun( } return nextToValidate > sourcePrefix, nil } - return !validateSTMResultAgainstPrefix(validation, writes, result, cumulativeGasUsed, gasLimit, sourcePrefix), nil + return !validateSTMResultAgainstPrefix(validation, writes, result, cumulativeGasUsed, gasLimit, sourcePrefix, txIndex), nil } func newSTMRerunTask( @@ -463,6 +496,9 @@ const ( occFallbackReasonWorkerPoolClosed = "worker_pool_closed" ) +// validateSTMResultAgainstPrefix reports whether the result at txIndex, executed against the prefix +// [0, sourcePrefix), still holds once the writes at [sourcePrefix, txIndex) are accepted, recording +// every conflict it finds. func validateSTMResultAgainstPrefix( validation *occValidationResult, writes *stateAccessIndex, @@ -470,13 +506,14 @@ func validateSTMResultAgainstPrefix( cumulativeGasUsed uint64, gasLimit uint64, sourcePrefix int, + txIndex int, ) bool { if err := stmGasValidationError(validation, result, cumulativeGasUsed, gasLimit); err != nil { return false } conflictsBefore := validation.conflictCount - validation.addConflicts("read", writes, result.readSet, sourcePrefix) - validation.addConflicts("write", writes, result.writeSet, sourcePrefix) + validation.addConflicts("read", writes, result.readSet, sourcePrefix, txIndex) + validation.addConflicts("write", writes, result.writeSet, sourcePrefix, txIndex) if validation.conflictCount == conflictsBefore { return true } @@ -485,20 +522,28 @@ func validateSTMResultAgainstPrefix( } func stmGasValidationError(validation *occValidationResult, result occTxExecution, cumulativeGasUsed uint64, gasLimit uint64) error { + reason, err := stmGasFailure(result, cumulativeGasUsed, gasLimit) + if err != nil { + validation.fallbackReason = reason + } + return err +} + +// stmGasFailure returns the fallback reason and error for a result that does not fit the block gas +// accounting after cumulativeGasUsed, or an empty reason and nil error when it does. +func stmGasFailure(result occTxExecution, cumulativeGasUsed uint64, gasLimit uint64) (string, error) { if result.gasUsed > math.MaxUint64-cumulativeGasUsed { - validation.fallbackReason = occFallbackReasonGasOverflow - return errors.New(occFallbackReasonGasOverflow) + return occFallbackReasonGasOverflow, errors.New(occFallbackReasonGasOverflow) } if cumulativeGasUsed > gasLimit || result.gasLimit > gasLimit-cumulativeGasUsed { - validation.fallbackReason = occFallbackReasonGasLimit - return core.ErrGasLimitReached + return occFallbackReasonGasLimit, core.ErrGasLimitReached } - return nil + return "", nil } -func (r *occValidationResult) addConflicts(access string, writes *stateAccessIndex, set map[stateAccessKey]struct{}, sourcePrefix int) { +func (r *occValidationResult) addConflicts(access string, writes *stateAccessIndex, set map[stateAccessKey]struct{}, sourcePrefix int, txIndex int) { for key := range set { - if !writes.conflictsWithAfter(key, sourcePrefix) { + if !writes.conflictsWithin(key, sourcePrefix, txIndex) { continue } if r.conflicts == nil { @@ -575,91 +620,7 @@ func (k stateAccessKind) String() string { } } -// minPrefetchedAccounts is the point below which resolving rows across the pool costs more in -// waking workers than the serial reads it saves. -const minPrefetchedAccounts = 256 - -// prefetchBaseAccounts resolves, across the worker pool, the account rows the merge will compare -// against, leaving them for ChangeSetInto to find already read. -// -// The merge is the block's largest serial phase and most of it is these reads, one address at a -// time. They are independent and read-only, and the OCC workers already read this view -// concurrently during speculation. -func (s *blockSTMState) prefetchBaseAccounts(ctx context.Context, pool *occWorkerPool) { - if pool == nil { - return - } - reader, ok := s.source.(accountSnapshotReader) - if !ok { - return - } - addrs := s.touchedAccounts() - if len(addrs) < minPrefetchedAccounts { - return - } - - snapshots := make([]accountSnapshot, len(addrs)) - served := make([]bool, len(addrs)) - var next atomic.Int64 - // A failure here only leaves rows unread, which the merge then reads itself. - _ = pool.Run(ctx, len(addrs), func(workerCtx context.Context, _ int, _ int) error { - for { - i := int(next.Add(1)) - 1 - if i >= len(addrs) { - return nil - } - if err := workerCtx.Err(); err != nil { - return err - } - if snapshot, hit := reader.ReadAccount(addrs[i]); hit { - snapshots[i] = snapshot - served[i] = true - } - } - }) - - s.prefetched = make(map[common.Address]accountSnapshot, len(addrs)) - for i, addr := range addrs { - if served[i] { - s.prefetched[addr] = snapshots[i] - } - } -} - -// touchedAccounts returns each address the block wrote a balance, nonce, or code for, once. -func (s *blockSTMState) touchedAccounts() []common.Address { - addrs := make([]common.Address, 0, len(s.balances)+len(s.nonces)+len(s.code)) - seen := make(map[common.Address]struct{}, len(s.balances)+len(s.nonces)+len(s.code)) - for _, set := range []func(func(common.Address)){ - func(yield func(common.Address)) { - for addr := range s.balances { - yield(addr) - } - }, - func(yield func(common.Address)) { - for addr := range s.nonces { - yield(addr) - } - }, - func(yield func(common.Address)) { - for addr := range s.code { - yield(addr) - } - }, - } { - set(func(addr common.Address) { - if _, dup := seen[addr]; dup { - return - } - seen[addr] = struct{}{} - addrs = append(addrs, addr) - }) - } - return addrs -} - func (e *Executor) mergeOCCResults(ctx context.Context, results []occTxExecution, finalState *blockSTMState) (*BlockResult, error) { - finalState.prefetchBaseAccounts(ctx, e.occPool) blockResult, err := e.acquireBlockResult(ctx, len(results)) if err != nil { return nil, err @@ -677,99 +638,136 @@ func (e *Executor) mergeOCCResults(ctx context.Context, results []occTxExecution blockResult.Txs[i] = result.txResult blockResult.Receipts[i] = result.receipt } - finalState.ChangeSetInto(&blockResult.ChangeSet) + if err := finalState.changeSetIntoParallel(ctx, e.occPool, &blockResult.ChangeSet); err != nil { + blockResult.Release() + return nil, err + } return blockResult, nil } +// blockSTMState is the accepted prefix of a block's state, split into address shards so that +// applying results and emitting the changeset can proceed shard by shard on different workers. type blockSTMState struct { - source StateReader + source StateReader + shards [occStateShards]blockSTMShard +} + +// blockSTMShard holds the accepted writes for the addresses of one shard. +type blockSTMShard struct { balances map[common.Address]*big.Int nonces map[common.Address]uint64 code map[common.Address][]byte storageClears map[common.Address]struct{} storage map[storageChangeKey]common.Hash - - // Account rows resolved ahead of the merge by prefetchBaseAccounts, or nil when it did not run. - prefetched map[common.Address]accountSnapshot } func newBlockSTMState(source StateReader) *blockSTMState { if source == nil { source = NewMemoryState() } - return &blockSTMState{ - source: source, - balances: map[common.Address]*big.Int{}, - nonces: map[common.Address]uint64{}, - code: map[common.Address][]byte{}, - storageClears: map[common.Address]struct{}{}, - storage: map[storageChangeKey]common.Hash{}, + state := &blockSTMState{source: source} + for i := range state.shards { + state.shards[i] = blockSTMShard{ + balances: map[common.Address]*big.Int{}, + nonces: map[common.Address]uint64{}, + code: map[common.Address][]byte{}, + storageClears: map[common.Address]struct{}{}, + storage: map[storageChangeKey]common.Hash{}, + } } + return state +} + +func (s *blockSTMState) shard(addr common.Address) *blockSTMShard { + return &s.shards[occShardOf(addr)] } func (s *blockSTMState) GetBalance(addr common.Address) *big.Int { - if balance, ok := s.balances[addr]; ok { + if balance, ok := s.shard(addr).balances[addr]; ok { return cloneBig(balance) } return s.source.GetBalance(addr) } func (s *blockSTMState) GetNonce(addr common.Address) uint64 { - if nonce, ok := s.nonces[addr]; ok { + if nonce, ok := s.shard(addr).nonces[addr]; ok { return nonce } return s.source.GetNonce(addr) } func (s *blockSTMState) GetCode(addr common.Address) []byte { - if code, ok := s.code[addr]; ok { + if code, ok := s.shard(addr).code[addr]; ok { return cloneBytes(code) } return s.source.GetCode(addr) } func (s *blockSTMState) GetState(addr common.Address, key common.Hash) common.Hash { - if value, ok := s.storage[storageChangeKey{address: addr, key: key}]; ok { + shard := s.shard(addr) + if value, ok := shard.storage[storageChangeKey{address: addr, key: key}]; ok { return value } - if _, ok := s.storageClears[addr]; ok { + if _, ok := shard.storageClears[addr]; ok { return common.Hash{} } return s.source.GetState(addr, key) } +// apply folds one accepted result into the prefix. func (s *blockSTMState) apply(result occTxExecution) { + s.applyOwned(result, occAllShards) +} + +// applyOwned folds the parts of one accepted result whose addresses fall in the shards owns +// reports true for. Two callers with disjoint ownership can run concurrently. +func (s *blockSTMState) applyOwned(result occTxExecution, owns occShardOwnership) { for _, change := range result.changeSet.Balances { + if !owns(occShardOf(change.Address)) { + continue + } + shard := s.shard(change.Address) delta := result.commutativeBalanceDeltas[change.Address] _, normalWrite := result.writeSet[stateAccessKey{kind: stateAccessBalance, address: change.Address}] if delta != nil && !normalWrite { balance := cloneBig(s.GetBalance(change.Address)) balance.Add(balance, delta) - s.balances[change.Address] = balance + shard.balances[change.Address] = balance continue } - s.balances[change.Address] = cloneBig(change.Balance) + shard.balances[change.Address] = cloneBig(change.Balance) } for _, change := range result.changeSet.Nonces { - s.nonces[change.Address] = change.Nonce + if owns(occShardOf(change.Address)) { + s.shard(change.Address).nonces[change.Address] = change.Nonce + } } for _, change := range result.changeSet.Code { + if !owns(occShardOf(change.Address)) { + continue + } if change.Delete { - s.code[change.Address] = nil + s.shard(change.Address).code[change.Address] = nil } else { - s.code[change.Address] = cloneBytes(change.Code) + s.shard(change.Address).code[change.Address] = cloneBytes(change.Code) } } for _, addr := range result.changeSet.StorageClears { - s.storageClears[addr] = struct{}{} - for key := range s.storage { + if !owns(occShardOf(addr)) { + continue + } + shard := s.shard(addr) + shard.storageClears[addr] = struct{}{} + for key := range shard.storage { if key.address == addr { - delete(s.storage, key) + delete(shard.storage, key) } } } for _, change := range result.changeSet.Storage { - s.storage[storageChangeKey{address: change.Address, key: change.Key}] = change.Value + if owns(occShardOf(change.Address)) { + s.shard(change.Address).storage[storageChangeKey{address: change.Address, key: change.Key}] = change.Value + } } } @@ -780,19 +778,15 @@ func (s *blockSTMState) ChangeSet() StateChangeSet { } // baseAccounts serves an account's pre-block fields, reading the row once however many fields a -// caller asks for. It is scoped to one merge and is not safe for concurrent use. +// caller asks for. It is scoped to one shard of one merge and is not safe for concurrent use. type baseAccounts struct { source StateReader reader accountSnapshotReader seen map[common.Address]accountSnapshot } -func newBaseAccounts(source StateReader, prefetched map[common.Address]accountSnapshot) *baseAccounts { - seen := prefetched - if seen == nil { - seen = map[common.Address]accountSnapshot{} - } - b := &baseAccounts{source: source, seen: seen} +func newBaseAccounts(source StateReader) *baseAccounts { + b := &baseAccounts{source: source, seen: map[common.Address]accountSnapshot{}} b.reader, _ = source.(accountSnapshotReader) return b } @@ -828,40 +822,49 @@ func (b *baseAccounts) balance(addr common.Address) *big.Int { func (b *baseAccounts) nonce(addr common.Address) uint64 { return b.get(addr).Nonce } func (b *baseAccounts) code(addr common.Address) []byte { return b.get(addr).Code } +// ChangeSetInto writes the block's net state changes, in canonical order, on the calling goroutine. func (s *blockSTMState) ChangeSetInto(changes *StateChangeSet) { changes.resetForReuse() + for i := range s.shards { + s.shards[i].changeSetInto(s.source, changes) + } +} + +// changeSetInto appends the shard's net changes, in canonical order, to changes. Shards partition +// the address space in canonical order, so appending shard by shard yields a canonically ordered +// changeset. +func (h *blockSTMShard) changeSetInto(source StateReader, changes *StateChangeSet) { // The three loops below each compare against the same accounts, and balance, nonce and code hash - // share one row. Reading per field would resolve that row three times per address, on the one - // goroutine a block's merge runs on. - base := newBaseAccounts(s.source, s.prefetched) - balanceAddrs := sortedAddressesFromBigMap(s.balances) + // share one row. Reading per field would resolve that row three times per address. + base := newBaseAccounts(source) + balanceAddrs := sortedAddressesFromBigMap(h.balances) for _, addr := range balanceAddrs { - balance := cloneBig(s.balances[addr]) + balance := cloneBig(h.balances[addr]) if balance.Cmp(base.balance(addr)) == 0 { continue } changes.Balances = append(changes.Balances, BalanceChange{Address: addr, Balance: balance}) } - nonceAddrs := sortedAddressesFromUint64Map(s.nonces) + nonceAddrs := sortedAddressesFromUint64Map(h.nonces) for _, addr := range nonceAddrs { - if s.nonces[addr] == base.nonce(addr) { + if h.nonces[addr] == base.nonce(addr) { continue } - changes.Nonces = append(changes.Nonces, NonceChange{Address: addr, Nonce: s.nonces[addr]}) + changes.Nonces = append(changes.Nonces, NonceChange{Address: addr, Nonce: h.nonces[addr]}) } - codeAddrs := sortedAddressesFromBytesMap(s.code) + codeAddrs := sortedAddressesFromBytesMap(h.code) for _, addr := range codeAddrs { - code := cloneBytes(s.code[addr]) + code := cloneBytes(h.code[addr]) if bytes.Equal(code, base.code(addr)) { continue } changes.Code = append(changes.Code, CodeChange{Address: addr, Code: code, Delete: len(code) == 0}) } - storageClearAddrs := sortedAddressesFromSet(s.storageClears) + storageClearAddrs := sortedAddressesFromSet(h.storageClears) changes.StorageClears = append(changes.StorageClears, storageClearAddrs...) - storageKeys := make([]storageChangeKey, 0, len(s.storage)) - for key := range s.storage { + storageKeys := make([]storageChangeKey, 0, len(h.storage)) + for key := range h.storage { storageKeys = append(storageKeys, key) } sort.Slice(storageKeys, func(i, j int) bool { @@ -871,9 +874,9 @@ func (s *blockSTMState) ChangeSetInto(changes *StateChangeSet) { return bytes.Compare(storageKeys[i].key[:], storageKeys[j].key[:]) < 0 }) for _, key := range storageKeys { - value := s.storage[key] - baseValue := s.source.GetState(key.address, key.key) - if _, cleared := s.storageClears[key.address]; cleared { + value := h.storage[key] + baseValue := source.GetState(key.address, key.key) + if _, cleared := h.storageClears[key.address]; cleared { baseValue = common.Hash{} } if value == baseValue { @@ -888,31 +891,58 @@ func (s *blockSTMState) ChangeSetInto(changes *StateChangeSet) { } } +// stateAccessIndex records, per state key and per address, the range of transaction indexes that +// wrote it. It is split into address shards so that disjoint shards can be filled concurrently. type stateAccessIndex struct { - exact map[stateAccessKey]int - account map[common.Address]int - touched map[common.Address]int - commutativeBalance map[common.Address]int + shards [occStateShards]stateAccessShard +} + +// stateAccessShard indexes the writes to the addresses of one shard. +type stateAccessShard struct { + exact map[stateAccessKey]txIndexSpan + account map[common.Address]txIndexSpan + touched map[common.Address]txIndexSpan + commutativeBalance map[common.Address]txIndexSpan +} + +// txIndexSpan is the lowest and highest transaction index recorded for one key. +type txIndexSpan struct { + first int + last int } func newStateAccessIndex() *stateAccessIndex { - return &stateAccessIndex{ - exact: map[stateAccessKey]int{}, - account: map[common.Address]int{}, - touched: map[common.Address]int{}, - commutativeBalance: map[common.Address]int{}, + index := &stateAccessIndex{} + for i := range index.shards { + index.shards[i] = stateAccessShard{ + exact: map[stateAccessKey]txIndexSpan{}, + account: map[common.Address]txIndexSpan{}, + touched: map[common.Address]txIndexSpan{}, + commutativeBalance: map[common.Address]txIndexSpan{}, + } } + return index +} + +func (i *stateAccessIndex) shard(addr common.Address) *stateAccessShard { + return &i.shards[occShardOf(addr)] } -func (i *stateAccessIndex) conflictsWithAfter(key stateAccessKey, sourcePrefix int) bool { - if i.hasWriteAtOrAfter(i.exact, key, sourcePrefix) { +// conflictsWithin reports whether a write recorded for key would invalidate a read or write of it +// by a transaction that executed against the prefix [0, lo) and sits at index hi, i.e. whether the +// key was written at an index in [lo, hi). Only the first and last write of a key are recorded, so +// the answer can be a false positive when both fall outside the range with a gap across it; it is +// never a false negative, and a false positive costs one rerun, not correctness. +func (i *stateAccessIndex) conflictsWithin(key stateAccessKey, lo int, hi int) bool { + shard := i.shard(key.address) + if writtenWithin(shard.exact, key, lo, hi) { return true } - if i.hasAddressWriteAtOrAfter(i.account, key.address, sourcePrefix) { + if writtenWithin(shard.account, key.address, lo, hi) { return true } if key.kind == stateAccessAccount { - if i.hasAddressWriteAtOrAfter(i.touched, key.address, sourcePrefix) { + if writtenWithin(shard.touched, key.address, lo, hi) { return true } } @@ -922,57 +952,64 @@ func (i *stateAccessIndex) conflictsWithAfter(key stateAccessKey, sourcePrefix i if key.kind != stateAccessAccount && key.kind != stateAccessBalance { return false } - return i.hasAddressWriteAtOrAfter(i.commutativeBalance, key.address, sourcePrefix) + return writtenWithin(shard.commutativeBalance, key.address, lo, hi) } +// writtenWithin reports whether the span recorded for key may contain an index in [lo, hi). +func writtenWithin[K comparable](writes map[K]txIndexSpan, key K, lo int, hi int) bool { + span, ok := writes[key] + return ok && lo < hi && span.last >= lo && span.first < hi +} + +// addAll records the set as written by transactions at every index. func (i *stateAccessIndex) addAll(set map[stateAccessKey]struct{}) { - i.addAllAt(math.MaxInt, set) + i.addSpan(txIndexSpan{first: 0, last: math.MaxInt}, set, occAllShards) } +// addAllAt records the set as written by the transaction at txIndex. func (i *stateAccessIndex) addAllAt(txIndex int, set map[stateAccessKey]struct{}) { + i.addSpan(txIndexSpan{first: txIndex, last: txIndex}, set, occAllShards) +} + +func (i *stateAccessIndex) addSpan(span txIndexSpan, set map[stateAccessKey]struct{}, owns occShardOwnership) { for key := range set { - i.recordWrite(i.exact, key, txIndex) + if !owns(occShardOf(key.address)) { + continue + } + shard := i.shard(key.address) + recordSpan(shard.exact, key, span) // Exist/Empty account reads depend on account metadata, not storage slots. if key.kind != stateAccessStorage { - i.recordAddressWrite(i.touched, key.address, txIndex) + recordSpan(shard.touched, key.address, span) } if key.kind == stateAccessAccount { - i.recordAddressWrite(i.account, key.address, txIndex) + recordSpan(shard.account, key.address, span) } } } +// addCommutativeBalanceDeltasAt records the non-zero deltas as balance credits by the transaction +// at txIndex. func (i *stateAccessIndex) addCommutativeBalanceDeltasAt(txIndex int, deltas map[common.Address]*big.Int) { + i.addCommutativeBalanceDeltas(txIndex, deltas, occAllShards) +} + +func (i *stateAccessIndex) addCommutativeBalanceDeltas(txIndex int, deltas map[common.Address]*big.Int, owns occShardOwnership) { for addr, delta := range deltas { - if delta == nil || delta.Sign() == 0 { + if delta == nil || delta.Sign() == 0 || !owns(occShardOf(addr)) { continue } - i.recordAddressWrite(i.commutativeBalance, addr, txIndex) - } -} - -func (i *stateAccessIndex) hasWriteAtOrAfter(writes map[stateAccessKey]int, key stateAccessKey, sourcePrefix int) bool { - txIndex, ok := writes[key] - return ok && txIndex >= sourcePrefix -} - -func (i *stateAccessIndex) hasAddressWriteAtOrAfter(writes map[common.Address]int, addr common.Address, sourcePrefix int) bool { - txIndex, ok := writes[addr] - return ok && txIndex >= sourcePrefix -} - -func (i *stateAccessIndex) recordWrite(writes map[stateAccessKey]int, key stateAccessKey, txIndex int) { - if existing, ok := writes[key]; ok && existing >= txIndex { - return + recordSpan(i.shard(addr).commutativeBalance, addr, txIndexSpan{first: txIndex, last: txIndex}) } - writes[key] = txIndex } -func (i *stateAccessIndex) recordAddressWrite(writes map[common.Address]int, addr common.Address, txIndex int) { - if existing, ok := writes[addr]; ok && existing >= txIndex { - return +// recordSpan widens the span recorded for key to include span. +func recordSpan[K comparable](writes map[K]txIndexSpan, key K, span txIndexSpan) { + if existing, ok := writes[key]; ok { + span.first = min(span.first, existing.first) + span.last = max(span.last, existing.last) } - writes[addr] = txIndex + writes[key] = span } type storageChangeKey struct { diff --git a/giga/evmonly/occ_shards.go b/giga/evmonly/occ_shards.go new file mode 100644 index 0000000000..eeb5468a00 --- /dev/null +++ b/giga/evmonly/occ_shards.go @@ -0,0 +1,223 @@ +package evmonly + +import ( + "context" + "math" + "sync/atomic" + + "github.com/ethereum/go-ethereum/common" +) + +// occStateShards is the number of address shards the accepted prefix and the write index are split +// into. Shards are contiguous address ranges, so concatenating per-shard output in shard order keeps +// it in canonical address order. +const occStateShards = 64 + +// occMinParallelValidation is the fewest results a parallel validation pass is worth running for; +// below it, waking the pool costs more than validating on the calling goroutine. +const occMinParallelValidation = 64 + +// occCancellationCheckInterval is how many items a worker handles between context checks. +const occCancellationCheckInterval = 64 + +// occShardOwnership tells a worker which shards it may write to. +type occShardOwnership func(shard int) bool + +func occShardOf(addr common.Address) int { + return int(addr[0]) * occStateShards / 256 +} + +func occAllShards(int) bool { return true } + +// occShardsOwnedBy assigns the shards round-robin across the pool's workers. +func occShardsOwnedBy(workerID int, workers int) occShardOwnership { + return func(shard int) bool { return shard%workers == workerID } +} + +// indexResults records every result's writes, each worker filling the shards it owns. +func (i *stateAccessIndex) indexResults(ctx context.Context, pool *occWorkerPool, results []occTxExecution) error { + if len(results) < occMinParallelValidation { + for txIndex, result := range results { + i.addAllAt(txIndex, result.writeSet) + i.addCommutativeBalanceDeltasAt(txIndex, result.commutativeBalanceDeltas) + } + return ctx.Err() + } + return pool.Run(ctx, occStateShards, func(workerCtx context.Context, workerID int, workers int) error { + owns := occShardsOwnedBy(workerID, workers) + for txIndex, result := range results { + if txIndex%occCancellationCheckInterval == 0 { + if err := workerCtx.Err(); err != nil { + return err + } + } + i.addSpan(txIndexSpan{first: txIndex, last: txIndex}, result.writeSet, owns) + i.addCommutativeBalanceDeltas(txIndex, result.commutativeBalanceDeltas, owns) + } + return nil + }) +} + +// acceptValidatedPrefix validates the results from the frontier onward across the pool, stopping at +// the first one the serial frontier would not accept as it stands, and folds the accepted run into +// the prefix. It returns how many results it accepted. +func (e *Executor) acceptValidatedPrefix( + ctx context.Context, + runner occSpeculativeRunner, + pool *occWorkerPool, + results []occTxExecution, + state *blockSTMValidationState, + validation *occValidationResult, +) (int, error) { + from := state.nextToValidate + if len(results)-from < occMinParallelValidation { + return 0, nil + } + cumulative, to := cumulativeGasFrom(results, from, state.cumulativeGasUsed) + stop, err := firstUnacceptedResult(ctx, pool, results, state.writes, runner.blockGasLimit, from, to, cumulative) + if err != nil { + return 0, err + } + if stop == from { + return 0, nil + } + if err := state.prefix.applyRange(ctx, pool, results[from:stop]); err != nil { + return 0, err + } + validation.validationCount += uint64(stop - from) //nolint:gosec // stop > from. + state.cumulativeGasUsed = cumulative[stop-from] + state.nextToValidate = stop + return stop - from, nil +} + +// cumulativeGasFrom returns, for each result at or after from, the block gas used before it, plus one +// trailing entry for the gas used after the last. It stops at the first result whose gas would +// overflow the counter and returns that index, or len(results) when none does. +func cumulativeGasFrom(results []occTxExecution, from int, gasUsedBefore uint64) ([]uint64, int) { + cumulative := make([]uint64, 1, len(results)-from+1) + cumulative[0] = gasUsedBefore + for i := from; i < len(results); i++ { + gasUsed := results[i].gasUsed + if gasUsed > math.MaxUint64-cumulative[i-from] { + return cumulative, i + } + cumulative = append(cumulative, cumulative[i-from]+gasUsed) + } + return cumulative, len(results) +} + +// firstUnacceptedResult returns the lowest index in [from, to) whose result the serial frontier +// would not accept given every result before it accepted, or to when it would accept them all. +func firstUnacceptedResult( + ctx context.Context, + pool *occWorkerPool, + results []occTxExecution, + writes *stateAccessIndex, + blockGasLimit uint64, + from int, + to int, + cumulative []uint64, +) (int, error) { + var stop atomic.Int64 + stop.Store(int64(to)) + err := pool.Run(ctx, to-from, func(workerCtx context.Context, workerID int, workers int) error { + for i := from + workerID; i < to; i += workers { + if int64(i) >= stop.Load() { + return nil + } + if (i-from)%occCancellationCheckInterval < workers { + if err := workerCtx.Err(); err != nil { + return err + } + } + if stmFrontierAccepts(results[i], writes, cumulative[i-from], blockGasLimit, i) { + continue + } + lowerStop(&stop, int64(i)) + return nil + } + return nil + }) + return int(stop.Load()), err +} + +func lowerStop(stop *atomic.Int64, index int64) { + for { + current := stop.Load() + if current <= index || stop.CompareAndSwap(current, index) { + return + } + } +} + +// stmFrontierAccepts mirrors the serial frontier's decision for a result at txIndex, without +// recording anything: the frontier records the outcome itself when it reaches the result. +func stmFrontierAccepts(result occTxExecution, writes *stateAccessIndex, cumulativeGasUsed uint64, blockGasLimit uint64, txIndex int) bool { + if result.err != nil { + return false + } + if _, err := stmGasFailure(result, cumulativeGasUsed, blockGasLimit); err != nil { + return false + } + for key := range result.readSet { + if writes.conflictsWithin(key, result.sourcePrefix, txIndex) { + return false + } + } + for key := range result.writeSet { + if writes.conflictsWithin(key, result.sourcePrefix, txIndex) { + return false + } + } + return true +} + +// applyRange folds a run of accepted results into the prefix in block order, each worker applying +// the shards it owns. +func (s *blockSTMState) applyRange(ctx context.Context, pool *occWorkerPool, results []occTxExecution) error { + return pool.Run(ctx, occStateShards, func(workerCtx context.Context, workerID int, workers int) error { + owns := occShardsOwnedBy(workerID, workers) + for txIndex, result := range results { + if txIndex%occCancellationCheckInterval == 0 { + if err := workerCtx.Err(); err != nil { + return err + } + } + s.applyOwned(result, owns) + } + return nil + }) +} + +// changeSetIntoParallel writes the block's net state changes, in canonical order, computing each +// shard's part on the pool. The shards' base-state reads are what the merge mostly spends its time +// on, and they are independent of each other. +func (s *blockSTMState) changeSetIntoParallel(ctx context.Context, pool *occWorkerPool, changes *StateChangeSet) error { + if pool == nil { + s.ChangeSetInto(changes) + return ctx.Err() + } + changes.resetForReuse() + var fragments [occStateShards]StateChangeSet + err := pool.Run(ctx, occStateShards, func(workerCtx context.Context, workerID int, workers int) error { + for shard := workerID; shard < occStateShards; shard += workers { + if err := workerCtx.Err(); err != nil { + return err + } + s.shards[shard].changeSetInto(s.source, &fragments[shard]) + } + return nil + }) + if err != nil { + return err + } + for i := range fragments { + fragment := &fragments[i] + changes.Balances = append(changes.Balances, fragment.Balances...) + changes.Nonces = append(changes.Nonces, fragment.Nonces...) + changes.Code = append(changes.Code, fragment.Code...) + changes.StorageClears = append(changes.StorageClears, fragment.StorageClears...) + changes.Storage = append(changes.Storage, fragment.Storage...) + } + return nil +} From 96cca066272914c1e03bc1f4bc9220b9b879bbd3 Mon Sep 17 00:00:00 2001 From: Brandon Chatham Date: Sat, 19 Sep 2026 17:30:44 +0000 Subject: [PATCH 2/4] evmonly: skip results by touched-shard mask; serial merge below a key threshold; shard helper tests Each incarnation records the shards its writes, credits and changes touch, so index/apply workers skip results that hold nothing of theirs instead of scanning every result's maps. The changeset merge reuses pooled per-shard fragments and stays on the calling goroutine when the prefix holds fewer than 256 keys. Adds direct tests for conflictsWithin bounds, occShardOf, cumulativeGasFrom, firstUnacceptedResult and touchedShards, and an ERC-20 single-contract case to the block benchmark. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- giga/evmonly/README.md | 8 +- .../evmonly/cmd/evmonly-loadtest/main_test.go | 6 +- giga/evmonly/occ.go | 22 +-- giga/evmonly/occ_shards.go | 84 ++++++++- giga/evmonly/occ_shards_test.go | 164 ++++++++++++++++++ 5 files changed, 262 insertions(+), 22 deletions(-) create mode 100644 giga/evmonly/occ_shards_test.go diff --git a/giga/evmonly/README.md b/giga/evmonly/README.md index e3a712d9e3..7a656d5299 100644 --- a/giga/evmonly/README.md +++ b/giga/evmonly/README.md @@ -221,9 +221,11 @@ 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. The -merge emits the changeset one shard at a time on the pool and concatenates the -shards in order, which is canonical address order. +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 diff --git a/giga/evmonly/cmd/evmonly-loadtest/main_test.go b/giga/evmonly/cmd/evmonly-loadtest/main_test.go index 64c924516b..0f5a002709 100644 --- a/giga/evmonly/cmd/evmonly-loadtest/main_test.go +++ b/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) diff --git a/giga/evmonly/occ.go b/giga/evmonly/occ.go index ffffcd1e07..8569ad414a 100644 --- a/giga/evmonly/occ.go +++ b/giga/evmonly/occ.go @@ -26,6 +26,7 @@ type occTxExecution struct { gasUsed uint64 gasLimit uint64 commutativeBalanceDeltas map[common.Address]*big.Int + shards occShardSet incarnation int sourcePrefix int err error @@ -198,6 +199,7 @@ func (r occSpeculativeRunner) executeTaskInto(ctx context.Context, task occExecu } result.incarnation = task.incarnation result.sourcePrefix = task.sourcePrefix + result.shards = result.touchedShards() results[task.txIndex] = result return nil } @@ -721,9 +723,9 @@ func (s *blockSTMState) apply(result occTxExecution) { // applyOwned folds the parts of one accepted result whose addresses fall in the shards owns // reports true for. Two callers with disjoint ownership can run concurrently. -func (s *blockSTMState) applyOwned(result occTxExecution, owns occShardOwnership) { +func (s *blockSTMState) applyOwned(result occTxExecution, owns occShardSet) { for _, change := range result.changeSet.Balances { - if !owns(occShardOf(change.Address)) { + if !owns.has(occShardOf(change.Address)) { continue } shard := s.shard(change.Address) @@ -738,12 +740,12 @@ func (s *blockSTMState) applyOwned(result occTxExecution, owns occShardOwnership shard.balances[change.Address] = cloneBig(change.Balance) } for _, change := range result.changeSet.Nonces { - if owns(occShardOf(change.Address)) { + if owns.has(occShardOf(change.Address)) { s.shard(change.Address).nonces[change.Address] = change.Nonce } } for _, change := range result.changeSet.Code { - if !owns(occShardOf(change.Address)) { + if !owns.has(occShardOf(change.Address)) { continue } if change.Delete { @@ -753,7 +755,7 @@ func (s *blockSTMState) applyOwned(result occTxExecution, owns occShardOwnership } } for _, addr := range result.changeSet.StorageClears { - if !owns(occShardOf(addr)) { + if !owns.has(occShardOf(addr)) { continue } shard := s.shard(addr) @@ -765,7 +767,7 @@ func (s *blockSTMState) applyOwned(result occTxExecution, owns occShardOwnership } } for _, change := range result.changeSet.Storage { - if owns(occShardOf(change.Address)) { + if owns.has(occShardOf(change.Address)) { s.shard(change.Address).storage[storageChangeKey{address: change.Address, key: change.Key}] = change.Value } } @@ -971,9 +973,9 @@ func (i *stateAccessIndex) addAllAt(txIndex int, set map[stateAccessKey]struct{} i.addSpan(txIndexSpan{first: txIndex, last: txIndex}, set, occAllShards) } -func (i *stateAccessIndex) addSpan(span txIndexSpan, set map[stateAccessKey]struct{}, owns occShardOwnership) { +func (i *stateAccessIndex) addSpan(span txIndexSpan, set map[stateAccessKey]struct{}, owns occShardSet) { for key := range set { - if !owns(occShardOf(key.address)) { + if !owns.has(occShardOf(key.address)) { continue } shard := i.shard(key.address) @@ -994,9 +996,9 @@ func (i *stateAccessIndex) addCommutativeBalanceDeltasAt(txIndex int, deltas map i.addCommutativeBalanceDeltas(txIndex, deltas, occAllShards) } -func (i *stateAccessIndex) addCommutativeBalanceDeltas(txIndex int, deltas map[common.Address]*big.Int, owns occShardOwnership) { +func (i *stateAccessIndex) addCommutativeBalanceDeltas(txIndex int, deltas map[common.Address]*big.Int, owns occShardSet) { for addr, delta := range deltas { - if delta == nil || delta.Sign() == 0 || !owns(occShardOf(addr)) { + if delta == nil || delta.Sign() == 0 || !owns.has(occShardOf(addr)) { continue } recordSpan(i.shard(addr).commutativeBalance, addr, txIndexSpan{first: txIndex, last: txIndex}) diff --git a/giga/evmonly/occ_shards.go b/giga/evmonly/occ_shards.go index eeb5468a00..10ced961d2 100644 --- a/giga/evmonly/occ_shards.go +++ b/giga/evmonly/occ_shards.go @@ -3,6 +3,7 @@ package evmonly import ( "context" "math" + "sync" "sync/atomic" "github.com/ethereum/go-ethereum/common" @@ -17,21 +18,65 @@ const occStateShards = 64 // below it, waking the pool costs more than validating on the calling goroutine. const occMinParallelValidation = 64 +// occMinParallelMergeKeys is the fewest accepted keys a parallel merge is worth running for; below +// it, the merge runs on the calling goroutine and appends straight into the pooled changeset. +const occMinParallelMergeKeys = 256 + // occCancellationCheckInterval is how many items a worker handles between context checks. const occCancellationCheckInterval = 64 -// occShardOwnership tells a worker which shards it may write to. -type occShardOwnership func(shard int) bool +// occShardSet is a set of shards, one bit per shard. +type occShardSet uint64 + +var _ = [1]struct{}{}[occStateShards-64] // occShardSet has exactly one bit per shard. + +const occAllShards = occShardSet(math.MaxUint64) func occShardOf(addr common.Address) int { return int(addr[0]) * occStateShards / 256 } -func occAllShards(int) bool { return true } +func (s occShardSet) has(shard int) bool { return s&(1< Date: Sat, 19 Sep 2026 17:45:09 +0000 Subject: [PATCH 3/4] evmonly: name the serial backoff step; pin the rejection-at-from case Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- giga/evmonly/occ.go | 52 +++++++++++++++++++++++---------- giga/evmonly/occ_shards_test.go | 30 +++++++++++++++++-- 2 files changed, 65 insertions(+), 17 deletions(-) diff --git a/giga/evmonly/occ.go b/giga/evmonly/occ.go index 8569ad414a..f3379ea42e 100644 --- a/giga/evmonly/occ.go +++ b/giga/evmonly/occ.go @@ -306,15 +306,9 @@ func (e *Executor) validateBlockSTM( if err := state.writes.indexResults(ctx, pool, results); err != nil { return nil, nil, validation, err } - // The frontier alternates between a parallel pass, which accepts every result up to the first - // one that needs attention, and the serial frontier, which handles that one. A block whose - // transactions depend on each other one after another would make each parallel pass accept - // nothing, so after such a pass the serial frontier keeps going for a stretch that doubles each - // time it happens again. - serialUntil := 0 - serialStretch := occMinParallelValidation + var backoff serialBackoff for state.nextToValidate < len(results) { - if state.nextToValidate >= serialUntil { + if backoff.parallelPassDue(state.nextToValidate) { accepted, err := e.acceptValidatedPrefix(ctx, runner, pool, results, state, &validation) if err != nil { return nil, nil, validation, err @@ -322,14 +316,9 @@ func (e *Executor) validateBlockSTM( if state.nextToValidate == len(results) { break } - if accepted < occMinParallelValidation { - serialUntil = state.nextToValidate + serialStretch - serialStretch *= 2 - } else { - serialStretch = occMinParallelValidation - } + backoff.record(state.nextToValidate, accepted) } - end := min(len(results), max(serialUntil, state.nextToValidate+1)) + end := backoff.serialEnd(state.nextToValidate, len(results)) rerun, err := validateBlockSTMFrontier(ctx, runner, results, state, &validation, end) if err != nil { return nil, nil, validation, err @@ -348,6 +337,39 @@ func (e *Executor) validateBlockSTM( return results, state.prefix, validation, nil } +// serialBackoff decides how far the serial frontier runs before the next parallel pass. +// +// The frontier alternates between a parallel pass, which accepts every result up to the first one +// that needs attention, and the serial frontier, which handles that one. A block whose transactions +// depend on each other one after another makes every parallel pass accept nothing, so after a pass +// that accepts too little to pay for itself the serial frontier keeps going for a stretch that +// doubles each time it happens again, and resets once a pass accepts enough. +type serialBackoff struct { + until int + stretch int +} + +// parallelPassDue reports whether the frontier at nextToValidate has left the serial stretch. +func (b *serialBackoff) parallelPassDue(nextToValidate int) bool { + return nextToValidate >= b.until +} + +// record sets the serial stretch that follows a parallel pass which accepted accepted results. +func (b *serialBackoff) record(nextToValidate int, accepted int) { + if accepted >= occMinParallelValidation { + b.stretch = 0 + return + } + b.stretch = max(2*b.stretch, occMinParallelValidation) + b.until = nextToValidate + b.stretch +} + +// serialEnd returns the index the serial frontier runs up to, at least one past nextToValidate and +// never past n. +func (b *serialBackoff) serialEnd(nextToValidate int, n int) int { + return min(n, max(b.until, nextToValidate+1)) +} + // validateBlockSTMFrontier accepts results in block order on the calling goroutine until it reaches // end or a result that has to be rerun, which it returns as a task. func validateBlockSTMFrontier( diff --git a/giga/evmonly/occ_shards_test.go b/giga/evmonly/occ_shards_test.go index 5888c8c45b..5dd9069dac 100644 --- a/giga/evmonly/occ_shards_test.go +++ b/giga/evmonly/occ_shards_test.go @@ -98,10 +98,36 @@ func TestFirstUnacceptedResultReturnsTheLowestRejection(t *testing.T) { require.Equal(t, rejected, stop) } - results[0].err = nil + results[0].err = errOCCMaxIncarnation stop, err = firstUnacceptedResult(context.Background(), pool, results, writes, math.MaxUint64, 0, to, cumulative) require.NoError(t, err) - require.Equal(t, 1, stop, "the lowest rejection wins even when higher ones are found first") + require.Equal(t, 0, stop, "a rejection at from itself stops the pass before it accepts anything") +} + +func TestSerialBackoffDoublesWhileParallelPassesAcceptTooLittle(t *testing.T) { + const n = 10_000 + var backoff serialBackoff + require.True(t, backoff.parallelPassDue(0)) + require.Equal(t, 1, backoff.serialEnd(0, n), "without a stretch the serial frontier takes one result") + + backoff.record(0, 0) + require.False(t, backoff.parallelPassDue(occMinParallelValidation-1)) + require.True(t, backoff.parallelPassDue(occMinParallelValidation)) + require.Equal(t, occMinParallelValidation, backoff.serialEnd(0, n)) + + next := occMinParallelValidation + backoff.record(next, 1) + require.Equal(t, next+2*occMinParallelValidation, backoff.serialEnd(next, n)) + next = backoff.serialEnd(next, n) + backoff.record(next, 1) + require.Equal(t, next+4*occMinParallelValidation, backoff.serialEnd(next, n)) + require.Equal(t, 100, backoff.serialEnd(next, 100), "the stretch never runs past the block") + + next = backoff.serialEnd(next, n) + backoff.record(next, occMinParallelValidation) + require.True(t, backoff.parallelPassDue(next)) + backoff.record(next, 0) + require.Equal(t, next+occMinParallelValidation, backoff.serialEnd(next, n), "a pass that accepts enough resets the stretch") } func TestFirstUnacceptedResultRejectsGasAndConflicts(t *testing.T) { From 592e38f54164181ca154ef66a902a2fa45550ca6 Mon Sep 17 00:00:00 2001 From: Brandon Chatham Date: Sat, 19 Sep 2026 19:06:41 +0000 Subject: [PATCH 4/4] ci: rerun (TestListenerTimeoutReadWrite 10ms read timeout tripped on the shard-3 runner; privval untouched here) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>