core: database commit

Author: rjl493456442

core/blockchain.go

@@ -73,9 +73,10 @@ var (
 	accountUpdateTimer = metrics.NewRegisteredResettingTimer("chain/account/updates", nil)
 	accountCommitTimer = metrics.NewRegisteredResettingTimer("chain/account/commits", nil)
 
-	storageReadTimer   = metrics.NewRegisteredResettingTimer("chain/storage/reads", nil)
-	storageUpdateTimer = metrics.NewRegisteredResettingTimer("chain/storage/updates", nil)
-	storageCommitTimer = metrics.NewRegisteredResettingTimer("chain/storage/commits", nil)
+	storageReadTimer    = metrics.NewRegisteredResettingTimer("chain/storage/reads", nil)
+	storageUpdateTimer  = metrics.NewRegisteredResettingTimer("chain/storage/updates", nil)
+	storageCommitTimer  = metrics.NewRegisteredResettingTimer("chain/storage/commits", nil)
+	databaseCommitTimer = metrics.NewRegisteredResettingTimer("chain/database/commits", nil)
 
 	accountCacheHitMeter  = metrics.NewRegisteredMeter("chain/account/reads/cache/process/hit", nil)
 	accountCacheMissMeter = metrics.NewRegisteredMeter("chain/account/reads/cache/process/miss", nil)
@@ -90,9 +91,6 @@ var (
 	accountReadSingleTimer = metrics.NewRegisteredResettingTimer("chain/account/single/reads", nil)
 	storageReadSingleTimer = metrics.NewRegisteredResettingTimer("chain/storage/single/reads", nil)
 
-	snapshotCommitTimer = metrics.NewRegisteredResettingTimer("chain/snapshot/commits", nil)
-	triedbCommitTimer   = metrics.NewRegisteredResettingTimer("chain/triedb/commits", nil)
-
 	blockInsertTimer          = metrics.NewRegisteredResettingTimer("chain/inserts", nil)
 	blockValidationTimer      = metrics.NewRegisteredResettingTimer("chain/validation", nil)
 	blockCrossValidationTimer = metrics.NewRegisteredResettingTimer("chain/crossvalidation", nil)
@@ -2191,11 +2189,10 @@ func (bc *BlockChain) ProcessBlock(parentRoot common.Hash, block *types.Block, s
 	}
 
 	// Update the metrics touched during block commit
-	stats.AccountCommits = statedb.AccountCommits  // Account commits are complete, we can mark them
-	stats.StorageCommits = statedb.StorageCommits  // Storage commits are complete, we can mark them
-	stats.SnapshotCommit = statedb.SnapshotCommits // Snapshot commits are complete, we can mark them
-	stats.TrieDBCommit = statedb.TrieDBCommits     // Trie database commits are complete, we can mark them
-	stats.BlockWrite = time.Since(wstart) - max(statedb.AccountCommits, statedb.StorageCommits) /* concurrent */ - statedb.SnapshotCommits - statedb.TrieDBCommits
+	stats.AccountCommits = statedb.AccountCommits // Account commits are complete, we can mark them
+	stats.StorageCommits = statedb.StorageCommits // Storage commits are complete, we can mark them
+	stats.DBCommit = statedb.DatabaseCommits      // Trie database commits are complete, we can mark them
+	stats.BlockWrite = time.Since(wstart) - max(statedb.AccountCommits, statedb.StorageCommits) /* concurrent */ - statedb.DatabaseCommits
 
 	elapsed := time.Since(startTime) + 1 // prevent zero division
 	stats.TotalTime = elapsed

core/blockchain_stats.go

@@ -48,8 +48,7 @@ type ExecuteStats struct {
 	Execution       time.Duration // Time spent on the EVM execution
 	Validation      time.Duration // Time spent on the block validation
 	CrossValidation time.Duration // Optional, time spent on the block cross validation
-	SnapshotCommit  time.Duration // Time spent on snapshot commit
-	TrieDBCommit    time.Duration // Time spent on database commit
+	DBCommit        time.Duration // Time spent on database commit
 	BlockWrite      time.Duration // Time spent on block write
 	TotalTime       time.Duration // The total time spent on block execution
 	MgasPerSecond   float64       // The million gas processed per second
@@ -80,8 +79,7 @@ func (s *ExecuteStats) reportMetrics() {
 	blockExecutionTimer.Update(s.Execution)                 // The time spent on EVM processing
 	blockValidationTimer.Update(s.Validation)               // The time spent on block validation
 	blockCrossValidationTimer.Update(s.CrossValidation)     // The time spent on stateless cross validation
-	snapshotCommitTimer.Update(s.SnapshotCommit)            // Snapshot commits are complete, we can mark them
-	triedbCommitTimer.Update(s.TrieDBCommit)                // Trie database commits are complete, we can mark them
+	databaseCommitTimer.Update(s.DBCommit)                  // Trie database commits are complete, we can mark them
 	blockWriteTimer.Update(s.BlockWrite)                    // The time spent on block write
 	blockInsertTimer.Update(s.TotalTime)                    // The total time spent on block execution
 	chainMgaspsMeter.Update(time.Duration(s.MgasPerSecond)) // TODO(rjl493456442) generalize the ResettingTimer
@@ -127,7 +125,7 @@ Block write: %v
 		common.PrettyDuration(s.StorageReads), s.StorageLoaded,
 		common.PrettyDuration(s.AccountHashes+s.AccountCommits+s.AccountUpdates),
 		common.PrettyDuration(s.StorageCommits+s.StorageUpdates),
-		common.PrettyDuration(s.TrieDBCommit+s.SnapshotCommit), common.PrettyDuration(s.BlockWrite),
+		common.PrettyDuration(s.DBCommit), common.PrettyDuration(s.BlockWrite),
 		s.StateReadCacheStats)
 	for _, line := range strings.Split(msg, "\n") {
 		if line == "" {

core/state/database.go

@@ -28,6 +28,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/trie"
 	"github.com/ethereum/go-ethereum/trie/bintrie"
 	"github.com/ethereum/go-ethereum/trie/transitiontrie"
@@ -60,6 +61,11 @@ type Database interface {
 
 	// Snapshot returns the underlying state snapshot.
 	Snapshot() *snapshot.Tree
+
+	// Commit flushes all pending writes and finalizes the state transition,
+	// committing the changes to the underlying storage. It returns an error
+	// if the commit fails.
+	Commit(update *stateUpdate) error
 }
 
 // Trie is a Ethereum Merkle Patricia trie.
@@ -280,6 +286,40 @@ func (db *CachingDB) Snapshot() *snapshot.Tree {
 	return db.snap
 }
 
+// Commit flushes all pending writes and finalizes the state transition,
+// committing the changes to the underlying storage. It returns an error
+// if the commit fails.
+func (db *CachingDB) Commit(update *stateUpdate) error {
+	// Short circuit if nothing to commit
+	if update.empty() {
+		return nil
+	}
+	// Commit dirty contract code if any exists
+	if len(update.codes) > 0 {
+		writer := db.codedb.Writer()
+		for _, code := range update.codes {
+			writer.Put(code.hash, code.blob)
+		}
+		if err := writer.Commit(); err != nil {
+			return err
+		}
+	}
+	// If snapshotting is enabled, update the snapshot tree with this new version
+	if db.snap != nil && db.snap.Snapshot(update.originRoot) != nil {
+		if err := db.snap.Update(update.root, update.originRoot, update.accounts, update.storages); err != nil {
+			log.Warn("Failed to update snapshot tree", "from", update.originRoot, "to", update.root, "err", err)
+		}
+		// Keep 128 diff layers in the memory, persistent layer is 129th.
+		// - head layer is paired with HEAD state
+		// - head-1 layer is paired with HEAD-1 state
+		// - head-127 layer(bottom-most diff layer) is paired with HEAD-127 state
+		if err := db.snap.Cap(update.root, TriesInMemory); err != nil {
+			log.Warn("Failed to cap snapshot tree", "root", update.root, "layers", TriesInMemory, "err", err)
+		}
+	}
+	return db.triedb.Update(update.root, update.originRoot, update.blockNumber, update.nodes, update.stateSet())
+}
+
 // mustCopyTrie returns a deep-copied trie.
 func mustCopyTrie(t Trie) Trie {
 	switch t := t.(type) {

core/state/database_history.go

@@ -148,3 +148,10 @@ func (db *HistoricDB) TrieDB() *triedb.Database {
 func (db *HistoricDB) Snapshot() *snapshot.Tree {
 	return nil
 }
+
+// Commit flushes all pending writes and finalizes the state transition,
+// committing the changes to the underlying storage. It returns an error
+// if the commit fails.
+func (db *HistoricDB) Commit(update *stateUpdate) error {
+	return errors.New("not implemented")
+}

core/state/statedb.go

@@ -28,7 +28,6 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/state/snapshot"
 	"github.com/ethereum/go-ethereum/core/stateless"
 	"github.com/ethereum/go-ethereum/core/tracing"
@@ -149,8 +148,7 @@ type StateDB struct {
 	StorageReads    time.Duration
 	StorageUpdates  time.Duration
 	StorageCommits  time.Duration
-	SnapshotCommits time.Duration
-	TrieDBCommits   time.Duration
+	DatabaseCommits time.Duration
 
 	AccountLoaded  int          // Number of accounts retrieved from the database during the state transition
 	AccountUpdated int          // Number of accounts updated during the state transition
@@ -1160,7 +1158,7 @@ func (s *StateDB) GetTrie() Trie {
 
 // commit gathers the state mutations accumulated along with the associated
 // trie changes, resetting all internal flags with the new state as the base.
-func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNumber uint64) (*stateUpdate, error) {
+func (s *StateDB) commit(deleteEmptyObjects bool, rawStorageKey bool, blockNumber uint64) (*stateUpdate, error) {
 	// Short circuit in case any database failure occurred earlier.
 	if s.dbErr != nil {
 		return nil, fmt.Errorf("commit aborted due to earlier error: %v", s.dbErr)
@@ -1214,7 +1212,7 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNum
 	// the same block, account deletions must be processed first. This ensures
 	// that the storage trie nodes deleted during destruction and recreated
 	// during subsequent resurrection can be combined correctly.
-	deletes, delNodes, err := s.handleDestruction(noStorageWiping)
+	deletes, delNodes, err := s.handleDestruction(rawStorageKey)
 	if err != nil {
 		return nil, err
 	}
@@ -1308,57 +1306,19 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNum
 	// Clear all internal flags and update state root at the end.
 	s.mutations = make(map[common.Address]*mutation)
 	s.stateObjectsDestruct = make(map[common.Address]*stateObject)
+	s.reader, _ = s.db.Reader(root)
 
 	origin := s.originalRoot
 	s.originalRoot = root
 
-	return newStateUpdate(noStorageWiping, origin, root, blockNumber, deletes, updates, nodes), nil
-}
+	update := newStateUpdate(rawStorageKey, origin, root, blockNumber, deletes, updates, nodes)
 
-// commitAndFlush is a wrapper of commit which also commits the state mutations
-// to the configured data stores.
-func (s *StateDB) commitAndFlush(block uint64, deleteEmptyObjects bool, noStorageWiping bool) (*stateUpdate, error) {
-	ret, err := s.commit(deleteEmptyObjects, noStorageWiping, block)
-	if err != nil {
+	start = time.Now()
+	if err := s.db.Commit(update); err != nil {
 		return nil, err
 	}
-	// Commit dirty contract code if any exists
-	if db := s.db.TrieDB().Disk(); db != nil && len(ret.codes) > 0 {
-		batch := db.NewBatch()
-		for _, code := range ret.codes {
-			rawdb.WriteCode(batch, code.hash, code.blob)
-		}
-		if err := batch.Write(); err != nil {
-			return nil, err
-		}
-	}
-	if !ret.empty() {
-		// If snapshotting is enabled, update the snapshot tree with this new version
-		if snap := s.db.Snapshot(); snap != nil && snap.Snapshot(ret.originRoot) != nil {
-			start := time.Now()
-			if err := snap.Update(ret.root, ret.originRoot, ret.accounts, ret.storages); err != nil {
-				log.Warn("Failed to update snapshot tree", "from", ret.originRoot, "to", ret.root, "err", err)
-			}
-			// Keep 128 diff layers in the memory, persistent layer is 129th.
-			// - head layer is paired with HEAD state
-			// - head-1 layer is paired with HEAD-1 state
-			// - head-127 layer(bottom-most diff layer) is paired with HEAD-127 state
-			if err := snap.Cap(ret.root, TriesInMemory); err != nil {
-				log.Warn("Failed to cap snapshot tree", "root", ret.root, "layers", TriesInMemory, "err", err)
-			}
-			s.SnapshotCommits += time.Since(start)
-		}
-		// If trie database is enabled, commit the state update as a new layer
-		if db := s.db.TrieDB(); db != nil {
-			start := time.Now()
-			if err := db.Update(ret.root, ret.originRoot, block, ret.nodes, ret.stateSet()); err != nil {
-				return nil, err
-			}
-			s.TrieDBCommits += time.Since(start)
-		}
-	}
-	s.reader, _ = s.db.Reader(s.originalRoot)
-	return ret, err
+	s.DatabaseCommits = time.Since(start)
+	return update, nil
 }
 
 // Commit writes the state mutations into the configured data stores.
@@ -1375,8 +1335,8 @@ func (s *StateDB) commitAndFlush(block uint64, deleteEmptyObjects bool, noStorag
 // Since self-destruction was deprecated with the Cancun fork and there are
 // no empty accounts left that could be deleted by EIP-158, storage wiping
 // should not occur.
-func (s *StateDB) Commit(block uint64, deleteEmptyObjects bool, noStorageWiping bool) (common.Hash, error) {
-	ret, err := s.commitAndFlush(block, deleteEmptyObjects, noStorageWiping)
+func (s *StateDB) Commit(blockNumber uint64, deleteEmptyObjects bool, rawStorageKey bool) (common.Hash, error) {
+	ret, err := s.commit(deleteEmptyObjects, rawStorageKey, blockNumber)
 	if err != nil {
 		return common.Hash{}, err
 	}
@@ -1385,8 +1345,8 @@ func (s *StateDB) Commit(block uint64, deleteEmptyObjects bool, noStorageWiping
 
 // CommitWithUpdate writes the state mutations and returns both the root hash and the state update.
 // This is useful for tracking state changes at the blockchain level.
-func (s *StateDB) CommitWithUpdate(block uint64, deleteEmptyObjects bool, noStorageWiping bool) (common.Hash, *stateUpdate, error) {
-	ret, err := s.commitAndFlush(block, deleteEmptyObjects, noStorageWiping)
+func (s *StateDB) CommitWithUpdate(blockNumber uint64, deleteEmptyObjects bool, rawStorageKey bool) (common.Hash, *stateUpdate, error) {
+	ret, err := s.commit(deleteEmptyObjects, rawStorageKey, blockNumber)
 	if err != nil {
 		return common.Hash{}, nil, err
 	}

core/state/statedb_fuzz_test.go

@@ -229,7 +229,7 @@ func (test *stateTest) run() bool {
 		} else {
 			state.IntermediateRoot(true) // call intermediateRoot at the transaction boundary
 		}
-		ret, err := state.commitAndFlush(0, true, false) // call commit at the block boundary
+		ret, err := state.commit(true, false, 0) // call commit at the block boundary
 		if err != nil {
 			panic(err)
 		}