evmonly/evmonlyapp: revert the execution optimizations (#4254, #4258, #4260, #4261) ahead of a reviewable stack

giga/evmonly/README.md

@@ -217,21 +217,10 @@ 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 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.
+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.
 
 - transactions with no dependency on newly accepted prior writes are retained
   and accepted in block order without rerunning

giga/evmonly/account_reader_test.go

@@ -1,7 +1,6 @@
 package evmonly
 
 import (
-	"bytes"
 	"math/big"
 	"sync/atomic"
 	"testing"
@@ -88,36 +87,48 @@ func TestSnapshotReaderFetchesCodeOnlyWhenTheAccountHasSome(t *testing.T) {
 	require.Equal(t, []byte{0x60, 0x00}, account.Code)
 }
 
-// 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) {
+// 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) {
 	snapshot := newMemoryGigaSnapshot(7)
-	addrs := make([]common.Address, 0, 512)
-	for i := range 512 {
-		addr := common.BigToAddress(new(big.Int).Lsh(big.NewInt(int64(i)+1), 150))
+	addrs := make([]common.Address, 0, minPrefetchedAccounts+8)
+	for i := range minPrefetchedAccounts + 8 {
+		addr := common.BigToAddress(big.NewInt(int64(i) + 1))
 		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.shard(addr).balances[addr] = big.NewInt(int64(i) + 100)
-		state.shard(addr).nonces[addr] = uint64(i) + 1 //nolint:gosec // i is non-negative.
+		state.balances[addr] = big.NewInt(int64(i) + 100)
 	}
-	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[:]))
+	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)
 	}
+	require.Equal(t, before, reading.reads.Load(), "the merge must not re-read what was prefetched")
+}
+
+// Below the threshold the pool costs more than the reads it saves, so the merge reads them itself.
+func TestPrefetchIsSkippedForASmallBlock(t *testing.T) {
+	snapshot := newMemoryGigaSnapshot(7)
+	addr := testAddress(0xd4)
+	snapshot.setBalance(addr, big.NewInt(5))
+	reading := &accountReadingSnapshot{memoryGigaSnapshot: snapshot}
+
+	state := newBlockSTMState(gigaSnapshotStateReader{snapshot: reading})
+	state.balances[addr] = big.NewInt(6)
+	state.prefetchBaseAccounts(t.Context(), newOCCWorkerPool(4))
+
+	require.Nil(t, state.prefetched)
 }

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

@@ -951,10 +951,6 @@ 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) {
@@ -969,7 +965,7 @@ func BenchmarkExecuteTransferBlock(b *testing.B) {
 			require.NoError(b, err)
 
 			state := newGeneratedState()
-			workload, err := scenarios.NewWorkload(cfg.workload, scenarioConfig(cfg), state)
+			workload, err := scenarios.NewTransferWorkload(scenarioConfig(cfg), state)
 			require.NoError(b, err)
 			request, err := workload.BuildBlock(b.Context(), 1)
 			require.NoError(b, err)

giga/evmonly/executor.go

@@ -52,9 +52,7 @@ type Executor struct {
 	pipelineMu      sync.Mutex
 	pipelineDone    chan struct{}
 	pipelineErr     error
-	pipelineChanges *pendingChanges
-	// Counts commits started, so a reader can tell that a block landed between two of its steps.
-	pipelineGeneration uint64
+	pipelineChanges *StateChangeSet
 	// The first commit that failed, kept so no caller can miss it.
 	pipelineFailure error
 }

giga/evmonly/executor_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"errors"
-	"math"
 	"math/big"
 	"sync"
 	"testing"
@@ -730,64 +729,6 @@ 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
@@ -2173,7 +2114,7 @@ func TestStateDBGetCodeHashTracksCodelessAccountExistenceReads(t *testing.T) {
 		{kind: stateAccessBalance, address: eoa}: {},
 	})
 	validation := occValidationResult{}
-	accepted := validateSTMResultAgainstPrefix(&validation, writes, occTxExecution{gasLimit: 1, readSet: readSet}, 0, 10, 0, math.MaxInt)
+	accepted := validateSTMResultAgainstPrefix(&validation, writes, occTxExecution{gasLimit: 1, readSet: readSet}, 0, 10, 0)
 	require.False(t, accepted)
 	require.Equal(t, occFallbackReasonConflict, validation.fallbackReason)
 }
@@ -2285,7 +2226,7 @@ func TestValidateSTMConflictMatrix(t *testing.T) {
 			t.Fatalf("unknown access mode %q", access)
 		}
 		validation := occValidationResult{}
-		accepted := validateSTMResultAgainstPrefix(&validation, writes, result, 0, 10, 0, math.MaxInt)
+		accepted := validateSTMResultAgainstPrefix(&validation, writes, result, 0, 10, 0)
 		return accepted, validation
 	}
 
@@ -2327,7 +2268,7 @@ func TestValidateSTMConflictSourcePrefix(t *testing.T) {
 			readSet:  map[stateAccessKey]struct{}{key: {}},
 			gasLimit: 1,
 			gasUsed:  1,
-		}, 0, 10, sourcePrefix, math.MaxInt)
+		}, 0, 10, sourcePrefix)
 		return accepted, validation
 	}
 	cases := []struct {

giga/evmonly/giga_store.go

@@ -92,7 +92,7 @@ func (e *Executor) executePreparedBlockWithStore(ctx context.Context, req Prepar
 		snapshot:     snapshot,
 		missingState: e.missingState,
 	}
-	source = pending.overlay(source)
+	source = newPendingOverlay(source, pending)
 
 	e.blockPhases.SetPhase("execute")
 	result, err := e.executePreparedBlock(ctx, req, source)
@@ -201,79 +201,12 @@ func (e *Executor) AwaitCommits() error {
 
 // pipelinePending returns the changes of a block whose commit has not been waited on yet, or nil
 // when the store is caught up.
-func (e *Executor) pipelinePending() *pendingChanges {
+func (e *Executor) pipelinePending() *StateChangeSet {
 	e.pipelineMu.Lock()
 	defer e.pipelineMu.Unlock()
 	return e.pipelineChanges
 }
 
-// LatestAccount is the balance and nonce of an account after the last block this executor ran.
-type LatestAccount struct {
-	Balance *big.Int
-	Nonce   uint64
-}
-
-// ReadLatestAccount returns addr's balance and nonce after the last block this executor ran,
-// without waiting for that block's commit to land. It reports the first failed commit instead of
-// state that lacks the failed block.
-func (e *Executor) ReadLatestAccount(addr common.Address) (LatestAccount, error) {
-	if e.stateStore == nil {
-		return LatestAccount{}, errMissingStateStore
-	}
-	for {
-		e.pipelineMu.Lock()
-		pending, generation, failure := e.pipelineChanges, e.pipelineGeneration, e.pipelineFailureLocked()
-		e.pipelineMu.Unlock()
-		if failure != nil {
-			return LatestAccount{}, failure
-		}
-		snapshot := e.stateStore.OpenView()
-		if snapshot == nil {
-			return LatestAccount{}, errors.New("giga store returned a nil snapshot")
-		}
-		account, ok := e.readLatestAccount(snapshot, pending, generation, addr)
-		snapshot.Close()
-		if ok {
-			return account, nil
-		}
-	}
-}
-
-// readLatestAccount reads addr through pending laid over snapshot. It reports false when another
-// commit started after generation was read, since the view may then hold a later block's writes and
-// pending would replay older values over them; the caller reads again.
-func (e *Executor) readLatestAccount(snapshot gigatypes.EVMStateView, pending *pendingChanges, generation uint64, addr common.Address) (LatestAccount, bool) {
-	e.pipelineMu.Lock()
-	moved := e.pipelineGeneration != generation
-	e.pipelineMu.Unlock()
-	if moved {
-		return LatestAccount{}, false
-	}
-	reader := pending.overlay(gigaSnapshotStateReader{snapshot: snapshot, missingState: e.missingState})
-	if rowReader, ok := reader.(accountSnapshotReader); ok {
-		if row, ok := rowReader.ReadAccount(addr); ok {
-			balance := row.Balance
-			if balance == nil {
-				balance = new(big.Int)
-			}
-			return LatestAccount{Balance: balance, Nonce: row.Nonce}, true
-		}
-	}
-	return LatestAccount{Balance: reader.GetBalance(addr), Nonce: reader.GetNonce(addr)}, true
-}
-
-// pipelineFailureLocked returns the first failed commit, whether or not a waiter has retired it yet.
-// Callers hold pipelineMu.
-func (e *Executor) pipelineFailureLocked() error {
-	if e.pipelineFailure != nil {
-		return e.pipelineFailure
-	}
-	if e.pipelineErr != nil {
-		return fmt.Errorf("commit state changes: %w", e.pipelineErr)
-	}
-	return nil
-}
-
 // awaitPipelineCommit blocks until the in-flight commit has landed, reporting the first commit that
 // failed. After it returns the store holds every block this executor has run, so the next view
 // opens on a known height and needs no overlay.
@@ -315,15 +248,14 @@ func (e *Executor) awaitPipelineCommit() error {
 // Commits stay ordered because only one is ever in flight: awaitPipelineCommit lands the previous
 // one before this is called.
 func (e *Executor) startPipelineCommit(blockNumber int64, changesets []*proto.NamedChangeSet, changes *StateChangeSet) error {
-	pending := newPendingChanges(changes.clone())
+	pending := changes.clone()
 	done := make(chan struct{})
 	e.pipelineMu.Lock()
 	if failure := e.pipelineFailure; failure != nil {
 		e.pipelineMu.Unlock()
 		return failure
 	}
 	e.pipelineChanges = pending
-	e.pipelineGeneration++
 	e.pipelineDone = done
 	e.pipelineErr = nil
 	e.pipelineMu.Unlock()