diff --git a/giga/evmonly/README.md b/giga/evmonly/README.md index adbfecc2e8..36c6f0c70e 100644 --- a/giga/evmonly/README.md +++ b/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 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/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/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..e9c924a364 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" @@ -27,6 +26,7 @@ type occTxExecution struct { gasUsed uint64 gasLimit uint64 commutativeBalanceDeltas map[common.Address]*big.Int + shards occShardSet incarnation int sourcePrefix int err error @@ -90,6 +90,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 } @@ -196,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 } @@ -299,8 +303,23 @@ 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 + } + var backoff serialBackoff for state.nextToValidate < len(results) { - rerun, err := validateBlockSTMFrontier(ctx, runner, results, state, &validation) + if backoff.parallelPassDue(state.nextToValidate) { + accepted, err := e.acceptValidatedPrefix(ctx, runner, pool, results, state, &validation) + if err != nil { + return nil, nil, validation, err + } + if state.nextToValidate == len(results) { + break + } + backoff.record(state.nextToValidate, accepted) + } + end := backoff.serialEnd(state.nextToValidate, len(results)) + rerun, err := validateBlockSTMFrontier(ctx, runner, results, state, &validation, end) if err != nil { return nil, nil, validation, err } @@ -310,18 +329,58 @@ 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 } +// 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( 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 +421,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 +448,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 +520,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 +530,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 +546,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 +644,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,102 +662,132 @@ 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) { - for _, change := range result.changeSet.Balances { + 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 occShardSet) { + for change := range ownedChanges(owns, result.changeSet.Balances, BalanceChange.addr) { + 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 + for change := range ownedChanges(owns, result.changeSet.Nonces, NonceChange.addr) { + s.shard(change.Address).nonces[change.Address] = change.Nonce } - for _, change := range result.changeSet.Code { + for change := range ownedChanges(owns, result.changeSet.Code, CodeChange.addr) { 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 { + for addr := range ownedChanges(owns, result.changeSet.StorageClears, sameAddress) { + 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 + for change := range ownedChanges(owns, result.changeSet.Storage, StorageChange.addr) { + s.shard(change.Address).storage[storageChangeKey{address: change.Address, key: change.Key}] = change.Value } } +func (c BalanceChange) addr() common.Address { return c.Address } +func (c NonceChange) addr() common.Address { return c.Address } +func (c CodeChange) addr() common.Address { return c.Address } +func (c StorageChange) addr() common.Address { return c.Address } +func sameAddress(a common.Address) common.Address { return a } + func (s *blockSTMState) ChangeSet() StateChangeSet { var changes StateChangeSet s.ChangeSetInto(&changes) @@ -780,19 +795,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 +839,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 +891,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 +908,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 +969,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 occShardSet) { for key := range set { - i.recordWrite(i.exact, key, txIndex) + if !owns.has(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 occShardSet) { for addr, delta := range deltas { - if delta == nil || delta.Sign() == 0 { + if delta == nil || delta.Sign() == 0 || !owns.has(occShardOf(addr)) { continue } - i.recordAddressWrite(i.commutativeBalance, addr, txIndex) + recordSpan(i.shard(addr).commutativeBalance, addr, txIndexSpan{first: txIndex, last: 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 - } - 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..5e1b0e6fbf --- /dev/null +++ b/giga/evmonly/occ_shards.go @@ -0,0 +1,303 @@ +package evmonly + +import ( + "context" + "iter" + "math" + "sync" + "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 + +// 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 + +// 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 (s occShardSet) has(shard int) bool { return s&(1< 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 + } + } + if result.shards.intersects(owns) { + s.applyOwned(result, owns) + } + } + return nil + }) +} + +// occChangeSetFragments holds one changeset per shard for a parallel merge to fill. +type occChangeSetFragments [occStateShards]StateChangeSet + +// occFragmentPool recycles fragment arrays so their capacity survives across blocks. +var occFragmentPool = sync.Pool{New: func() any { return new(occChangeSetFragments) }} + +// 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.keyCount() < occMinParallelMergeKeys { + s.ChangeSetInto(changes) + return ctx.Err() + } + changes.resetForReuse() + fragments := occFragmentPool.Get().(*occChangeSetFragments) + defer occFragmentPool.Put(fragments) + 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 + } + fragments[shard].resetForReuse() + 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 +} + +// keyCount returns how many keys the accepted prefix holds across all shards. +func (s *blockSTMState) keyCount() int { + count := 0 + for i := range s.shards { + shard := &s.shards[i] + count += len(shard.balances) + len(shard.nonces) + len(shard.code) + len(shard.storageClears) + len(shard.storage) + } + return count +} diff --git a/giga/evmonly/occ_shards_test.go b/giga/evmonly/occ_shards_test.go new file mode 100644 index 0000000000..5dd9069dac --- /dev/null +++ b/giga/evmonly/occ_shards_test.go @@ -0,0 +1,190 @@ +package evmonly + +import ( + "context" + "math" + "math/big" + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/require" +) + +func TestOCCShardOfIsMonotonicAndCoversEveryShard(t *testing.T) { + seen := map[int]bool{} + previous := 0 + for first := range 256 { + addr := common.Address{byte(first), 0xff} + shard := occShardOf(addr) + require.GreaterOrEqual(t, shard, previous, "shards must follow address order") + require.Less(t, shard, occStateShards) + seen[shard] = true + previous = shard + } + require.Len(t, seen, occStateShards) +} + +func TestOCCShardsOwnedByPartitionsTheShards(t *testing.T) { + for _, workers := range []int{1, 3, 8, 64, 100} { + var union occShardSet + for workerID := range workers { + owned := occShardsOwnedBy(workerID, workers) + require.False(t, union.intersects(owned), "workers=%d worker %d overlaps another", workers, workerID) + union |= owned + } + require.Equal(t, occAllShards, union, "workers=%d", workers) + } +} + +func TestConflictsWithinHonoursBothBounds(t *testing.T) { + addr := testAddress(0xa1) + key := stateAccessKey{kind: stateAccessBalance, address: addr} + writes := newStateAccessIndex() + writes.addAllAt(5, map[stateAccessKey]struct{}{key: {}}) + + require.False(t, writes.conflictsWithin(key, 0, 5), "a write at the reader's own index is not a conflict") + require.False(t, writes.conflictsWithin(key, 0, 3), "a write above the reader is not a conflict") + require.False(t, writes.conflictsWithin(key, 6, 10), "a write below the source prefix is not a conflict") + require.True(t, writes.conflictsWithin(key, 0, 6)) + require.True(t, writes.conflictsWithin(key, 5, 6)) + require.False(t, writes.conflictsWithin(key, 6, 6), "an empty range holds no writes") + require.False(t, writes.conflictsWithin(stateAccessKey{kind: stateAccessBalance, address: testAddress(0xa2)}, 0, 10)) +} + +func TestConflictsWithinSpanOnlyOverApproximates(t *testing.T) { + addr := testAddress(0xa1) + key := stateAccessKey{kind: stateAccessBalance, address: addr} + writes := newStateAccessIndex() + writes.addAllAt(2, map[stateAccessKey]struct{}{key: {}}) + writes.addAllAt(9, map[stateAccessKey]struct{}{key: {}}) + + require.True(t, writes.conflictsWithin(key, 4, 7), "a gap inside the span is reported as a conflict") + require.False(t, writes.conflictsWithin(key, 0, 2)) + require.False(t, writes.conflictsWithin(key, 10, 20)) +} + +func TestCumulativeGasFromStopsBeforeOverflow(t *testing.T) { + results := []occTxExecution{{gasUsed: 10}, {gasUsed: 20}, {gasUsed: math.MaxUint64}, {gasUsed: 1}} + + cumulative, to := cumulativeGasFrom(results, 1, 100) + require.Equal(t, 2, to) + require.Equal(t, []uint64{100, 120}, cumulative) + + cumulative, to = cumulativeGasFrom(results[:2], 0, 0) + require.Equal(t, 2, to) + require.Equal(t, []uint64{0, 10, 30}, cumulative) +} + +func TestFirstUnacceptedResultReturnsTheLowestRejection(t *testing.T) { + pool := newOCCWorkerPool(8) + defer pool.Close() + const count = 1000 + results := make([]occTxExecution, count) + for i := range results { + results[i] = occTxExecution{gasLimit: 1, gasUsed: 1} + } + cumulative, to := cumulativeGasFrom(results, 0, 0) + require.Equal(t, count, to) + writes := newStateAccessIndex() + + stop, err := firstUnacceptedResult(context.Background(), pool, results, writes, math.MaxUint64, 0, to, cumulative) + require.NoError(t, err) + require.Equal(t, count, stop, "every result is acceptable") + + for _, rejected := range []int{999, 640, 333, 1} { + results[rejected].err = errOCCMaxIncarnation + stop, err = firstUnacceptedResult(context.Background(), pool, results, writes, math.MaxUint64, 0, to, cumulative) + require.NoError(t, err) + require.Equal(t, rejected, stop) + } + + results[0].err = errOCCMaxIncarnation + stop, err = firstUnacceptedResult(context.Background(), pool, results, writes, math.MaxUint64, 0, to, cumulative) + require.NoError(t, err) + 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) { + pool := newOCCWorkerPool(4) + defer pool.Close() + addr := testAddress(0xb1) + key := stateAccessKey{kind: stateAccessNonce, address: addr} + results := make([]occTxExecution, 200) + for i := range results { + results[i] = occTxExecution{gasLimit: 10, gasUsed: 10} + } + results[150].readSet = map[stateAccessKey]struct{}{key: {}} + writes := newStateAccessIndex() + writes.addAllAt(20, map[stateAccessKey]struct{}{key: {}}) + cumulative, to := cumulativeGasFrom(results, 0, 0) + + stop, err := firstUnacceptedResult(context.Background(), pool, results, writes, math.MaxUint64, 0, to, cumulative) + require.NoError(t, err) + require.Equal(t, 150, stop, "a read of a write inside [sourcePrefix, txIndex) is rejected") + + results[150].sourcePrefix = 21 + stop, err = firstUnacceptedResult(context.Background(), pool, results, writes, math.MaxUint64, 0, to, cumulative) + require.NoError(t, err) + require.Equal(t, len(results), stop, "a write below the source prefix is already accounted for") + + stop, err = firstUnacceptedResult(context.Background(), pool, results, writes, 10*100, 0, to, cumulative) + require.NoError(t, err) + require.Equal(t, 100, stop, "the first result over the block gas limit is rejected") +} + +func TestTouchedShardsCoversEveryAddressTheResultChanged(t *testing.T) { + written := common.Address{0x00} + credited := common.Address{0x40} + changed := common.Address{0x80} + cleared := common.Address{0xc0} + result := occTxExecution{ + writeSet: map[stateAccessKey]struct{}{{kind: stateAccessNonce, address: written}: {}}, + commutativeBalanceDeltas: map[common.Address]*big.Int{credited: big.NewInt(1)}, + } + result.changeSet.Storage = append(result.changeSet.Storage, StorageChange{Address: changed}) + result.changeSet.StorageClears = append(result.changeSet.StorageClears, cleared) + + touched := result.touchedShards() + for _, addr := range []common.Address{written, credited, changed, cleared} { + require.True(t, touched.has(occShardOf(addr)), "%s", addr) + } + require.False(t, touched.has(occShardOf(common.Address{0x20}))) +} + +func TestSmallMergeStaysOnTheCallingGoroutine(t *testing.T) { + pool := newOCCWorkerPool(4) + pool.Close() + state := newBlockSTMState(NewMemoryState()) + state.shards[0].nonces[common.Address{0x00}] = 1 + require.Less(t, state.keyCount(), occMinParallelMergeKeys) + + var changes StateChangeSet + require.NoError(t, state.changeSetIntoParallel(context.Background(), pool, &changes), "a closed pool is never touched below the threshold") + require.Equal(t, state.ChangeSet(), changes) +}