diff --git a/beacon/blsync/engineclient.go b/beacon/blsync/engineclient.go index 77bee83f5b..f9821fc6f3 100644 --- a/beacon/blsync/engineclient.go +++ b/beacon/blsync/engineclient.go @@ -26,6 +26,7 @@ import ( "github.com/ethereum/go-ethereum/beacon/params" "github.com/ethereum/go-ethereum/beacon/types" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" ctypes "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/rpc" @@ -104,7 +105,11 @@ func (ec *engineClient) callNewPayload(fork string, event types.ChainHeadEvent) method = "engine_newPayloadV4" parentBeaconRoot := event.BeaconHead.ParentRoot blobHashes := collectBlobHashes(event.Block) - params = append(params, blobHashes, parentBeaconRoot, event.ExecRequests) + hexRequests := make([]hexutil.Bytes, len(event.ExecRequests)) + for i := range event.ExecRequests { + hexRequests[i] = hexutil.Bytes(event.ExecRequests[i]) + } + params = append(params, blobHashes, parentBeaconRoot, hexRequests) case "deneb": method = "engine_newPayloadV3" parentBeaconRoot := event.BeaconHead.ParentRoot diff --git a/cmd/geth/consolecmd_test.go b/cmd/geth/consolecmd_test.go index ca8efb5f98..4e1f6340a0 100644 --- a/cmd/geth/consolecmd_test.go +++ b/cmd/geth/consolecmd_test.go @@ -102,17 +102,17 @@ func TestAttachWelcome(t *testing.T) { "--http", "--http.port", httpPort, "--ws", "--ws.port", wsPort) t.Run("ipc", func(t *testing.T) { - waitForEndpoint(t, ipc, 4*time.Second) + waitForEndpoint(t, ipc, 2*time.Minute) testAttachWelcome(t, geth, "ipc:"+ipc, ipcAPIs) }) t.Run("http", func(t *testing.T) { endpoint := "http://127.0.0.1:" + httpPort - waitForEndpoint(t, endpoint, 4*time.Second) + waitForEndpoint(t, endpoint, 2*time.Minute) testAttachWelcome(t, geth, endpoint, httpAPIs) }) t.Run("ws", func(t *testing.T) { endpoint := "ws://127.0.0.1:" + wsPort - waitForEndpoint(t, endpoint, 4*time.Second) + waitForEndpoint(t, endpoint, 2*time.Minute) testAttachWelcome(t, geth, endpoint, httpAPIs) }) geth.Kill() diff --git a/core/bench_test.go b/core/bench_test.go index 00f924076a..155fa6c3b5 100644 --- a/core/bench_test.go +++ b/core/bench_test.go @@ -183,7 +183,7 @@ func benchInsertChain(b *testing.B, disk bool, gen func(int, *BlockGen)) { if !disk { db = rawdb.NewMemoryDatabase() } else { - pdb, err := pebble.New(b.TempDir(), 128, 128, "", false, true) + pdb, err := pebble.New(b.TempDir(), 128, 128, "", false) if err != nil { b.Fatalf("cannot create temporary database: %v", err) } @@ -303,7 +303,7 @@ func makeChainForBench(db ethdb.Database, genesis *Genesis, full bool, count uin func benchWriteChain(b *testing.B, full bool, count uint64) { genesis := &Genesis{Config: params.AllEthashProtocolChanges} for i := 0; i < b.N; i++ { - pdb, err := pebble.New(b.TempDir(), 1024, 128, "", false, true) + pdb, err := pebble.New(b.TempDir(), 1024, 128, "", false) if err != nil { b.Fatalf("error opening database: %v", err) } @@ -316,7 +316,7 @@ func benchWriteChain(b *testing.B, full bool, count uint64) { func benchReadChain(b *testing.B, full bool, count uint64) { dir := b.TempDir() - pdb, err := pebble.New(dir, 1024, 128, "", false, true) + pdb, err := pebble.New(dir, 1024, 128, "", false) if err != nil { b.Fatalf("error opening database: %v", err) } @@ -332,7 +332,7 @@ func benchReadChain(b *testing.B, full bool, count uint64) { b.ResetTimer() for i := 0; i < b.N; i++ { - pdb, err = pebble.New(dir, 1024, 128, "", false, true) + pdb, err = pebble.New(dir, 1024, 128, "", false) if err != nil { b.Fatalf("error opening database: %v", err) } diff --git a/core/blockchain.go b/core/blockchain.go index 6667f64911..302ab14cf0 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -65,7 +65,8 @@ var ( headFinalizedBlockGauge = metrics.NewRegisteredGauge("chain/head/finalized", nil) headSafeBlockGauge = metrics.NewRegisteredGauge("chain/head/safe", nil) - chainInfoGauge = metrics.NewRegisteredGaugeInfo("chain/info", nil) + chainInfoGauge = metrics.NewRegisteredGaugeInfo("chain/info", nil) + chainMgaspsGauge = metrics.NewRegisteredGauge("chain/mgasps", nil) accountReadTimer = metrics.NewRegisteredResettingTimer("chain/account/reads", nil) accountHashTimer = metrics.NewRegisteredResettingTimer("chain/account/hashes", nil) @@ -92,8 +93,10 @@ var ( blockReorgAddMeter = metrics.NewRegisteredMeter("chain/reorg/add", nil) blockReorgDropMeter = metrics.NewRegisteredMeter("chain/reorg/drop", nil) - blockPrefetchExecuteTimer = metrics.NewRegisteredTimer("chain/prefetch/executes", nil) - blockPrefetchInterruptMeter = metrics.NewRegisteredMeter("chain/prefetch/interrupts", nil) + blockPrefetchExecuteTimer = metrics.NewRegisteredResettingTimer("chain/prefetch/executes", nil) + blockPrefetchInterruptMeter = metrics.NewRegisteredMeter("chain/prefetch/interrupts", nil) + blockPrefetchTxsInvalidMeter = metrics.NewRegisteredMeter("chain/prefetch/txs/invalid", nil) + blockPrefetchTxsValidMeter = metrics.NewRegisteredMeter("chain/prefetch/txs/valid", nil) errInsertionInterrupted = errors.New("insertion is interrupted") errChainStopped = errors.New("blockchain is stopped") @@ -979,17 +982,16 @@ func (bc *BlockChain) setHeadBeyondRoot(head uint64, time uint64, root common.Ha // Ignore the error here since light client won't hit this path frozen, _ := bc.db.Ancients() if num+1 <= frozen { - // Truncate all relative data(header, total difficulty, body, receipt - // and canonical hash) from ancient store. - if _, err := bc.db.TruncateHead(num); err != nil { - log.Crit("Failed to truncate ancient data", "number", num, "err", err) - } - // Remove the hash <-> number mapping from the active store. - rawdb.DeleteHeaderNumber(db, hash) + // The chain segment, such as the block header, canonical hash, + // body, and receipt, will be removed from the ancient store + // in one go. + // + // The hash-to-number mapping in the key-value store will be + // removed by the hc.SetHead function. } else { - // Remove relative body and receipts from the active store. - // The header, total difficulty and canonical hash will be - // removed in the hc.SetHead function. + // Remove the associated body and receipts from the key-value store. + // The header, hash-to-number mapping, and canonical hash will be + // removed by the hc.SetHead function. rawdb.DeleteBody(db, hash, num) rawdb.DeleteReceipts(db, hash, num) } @@ -1361,7 +1363,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ size += writeSize // Sync the ancient store explicitly to ensure all data has been flushed to disk. - if err := bc.db.Sync(); err != nil { + if err := bc.db.SyncAncient(); err != nil { return 0, err } // Write hash to number mappings @@ -1759,18 +1761,6 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool, makeWitness bc.reportBlock(block, nil, err) return nil, it.index, err } - // No validation errors for the first block (or chain prefix skipped) - var activeState *state.StateDB - defer func() { - // The chain importer is starting and stopping trie prefetchers. If a bad - // block or other error is hit however, an early return may not properly - // terminate the background threads. This defer ensures that we clean up - // and dangling prefetcher, without deferring each and holding on live refs. - if activeState != nil { - activeState.StopPrefetcher() - } - }() - // Track the singleton witness from this chain insertion (if any) var witness *stateless.Witness @@ -1826,63 +1816,20 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool, makeWitness continue } // Retrieve the parent block and it's state to execute on top - start := time.Now() parent := it.previous() if parent == nil { parent = bc.GetHeader(block.ParentHash(), block.NumberU64()-1) } - statedb, err := state.New(parent.Root, bc.statedb) - if err != nil { - return nil, it.index, err - } - - // If we are past Byzantium, enable prefetching to pull in trie node paths - // while processing transactions. Before Byzantium the prefetcher is mostly - // useless due to the intermediate root hashing after each transaction. - if bc.chainConfig.IsByzantium(block.Number()) { - // Generate witnesses either if we're self-testing, or if it's the - // only block being inserted. A bit crude, but witnesses are huge, - // so we refuse to make an entire chain of them. - if bc.vmConfig.StatelessSelfValidation || (makeWitness && len(chain) == 1) { - witness, err = stateless.NewWitness(block.Header(), bc) - if err != nil { - return nil, it.index, err - } - } - statedb.StartPrefetcher("chain", witness) - } - activeState = statedb - - // If we have a followup block, run that against the current state to pre-cache - // transactions and probabilistically some of the account/storage trie nodes. - var followupInterrupt atomic.Bool - if !bc.cacheConfig.TrieCleanNoPrefetch { - if followup, err := it.peek(); followup != nil && err == nil { - throwaway, _ := state.New(parent.Root, bc.statedb) - - go func(start time.Time, followup *types.Block, throwaway *state.StateDB) { - // Disable tracing for prefetcher executions. - vmCfg := bc.vmConfig - vmCfg.Tracer = nil - bc.prefetcher.Prefetch(followup, throwaway, vmCfg, &followupInterrupt) - - blockPrefetchExecuteTimer.Update(time.Since(start)) - if followupInterrupt.Load() { - blockPrefetchInterruptMeter.Mark(1) - } - }(time.Now(), followup, throwaway) - } - } - // The traced section of block import. - res, err := bc.processBlock(block, statedb, start, setHead) - followupInterrupt.Store(true) + start := time.Now() + res, err := bc.processBlock(parent.Root, block, setHead, makeWitness && len(chain) == 1) if err != nil { return nil, it.index, err } // Report the import stats before returning the various results stats.processed++ stats.usedGas += res.usedGas + witness = res.witness var snapDiffItems, snapBufItems common.StorageSize if bc.snaps != nil { @@ -1938,11 +1885,74 @@ type blockProcessingResult struct { usedGas uint64 procTime time.Duration status WriteStatus + witness *stateless.Witness } // processBlock executes and validates the given block. If there was no error // it writes the block and associated state to database. -func (bc *BlockChain) processBlock(block *types.Block, statedb *state.StateDB, start time.Time, setHead bool) (_ *blockProcessingResult, blockEndErr error) { +func (bc *BlockChain) processBlock(parentRoot common.Hash, block *types.Block, setHead bool, makeWitness bool) (_ *blockProcessingResult, blockEndErr error) { + var ( + err error + startTime = time.Now() + statedb *state.StateDB + interrupt atomic.Bool + ) + defer interrupt.Store(true) // terminate the prefetch at the end + + if bc.cacheConfig.TrieCleanNoPrefetch { + statedb, err = state.New(parentRoot, bc.statedb) + if err != nil { + return nil, err + } + } else { + // If prefetching is enabled, run that against the current state to pre-cache + // transactions and probabilistically some of the account/storage trie nodes. + // + // Note: the main processor and prefetcher share the same reader with a local + // cache for mitigating the overhead of state access. + reader, err := bc.statedb.ReaderWithCache(parentRoot) + if err != nil { + return nil, err + } + throwaway, err := state.NewWithReader(parentRoot, bc.statedb, reader) + if err != nil { + return nil, err + } + statedb, err = state.NewWithReader(parentRoot, bc.statedb, reader) + if err != nil { + return nil, err + } + go func(start time.Time, throwaway *state.StateDB, block *types.Block) { + // Disable tracing for prefetcher executions. + vmCfg := bc.vmConfig + vmCfg.Tracer = nil + bc.prefetcher.Prefetch(block, throwaway, vmCfg, &interrupt) + + blockPrefetchExecuteTimer.Update(time.Since(start)) + if interrupt.Load() { + blockPrefetchInterruptMeter.Mark(1) + } + }(time.Now(), throwaway, block) + } + + // If we are past Byzantium, enable prefetching to pull in trie node paths + // while processing transactions. Before Byzantium the prefetcher is mostly + // useless due to the intermediate root hashing after each transaction. + var witness *stateless.Witness + if bc.chainConfig.IsByzantium(block.Number()) { + // Generate witnesses either if we're self-testing, or if it's the + // only block being inserted. A bit crude, but witnesses are huge, + // so we refuse to make an entire chain of them. + if bc.vmConfig.StatelessSelfValidation || makeWitness { + witness, err = stateless.NewWitness(block.Header(), bc) + if err != nil { + return nil, err + } + } + statedb.StartPrefetcher("chain", witness) + defer statedb.StopPrefetcher() + } + if bc.logger != nil && bc.logger.OnBlockStart != nil { bc.logger.OnBlockStart(tracing.BlockEvent{ Block: block, @@ -2001,7 +2011,7 @@ func (bc *BlockChain) processBlock(block *types.Block, statedb *state.StateDB, s } } xvtime := time.Since(xvstart) - proctime := time.Since(start) // processing + validation + cross validation + proctime := time.Since(startTime) // processing + validation + cross validation // Update the metrics touched during block processing and validation accountReadTimer.Update(statedb.AccountReads) // Account reads are complete(in processing) @@ -2042,9 +2052,14 @@ func (bc *BlockChain) processBlock(block *types.Block, statedb *state.StateDB, s triedbCommitTimer.Update(statedb.TrieDBCommits) // Trie database commits are complete, we can mark them blockWriteTimer.Update(time.Since(wstart) - max(statedb.AccountCommits, statedb.StorageCommits) /* concurrent */ - statedb.SnapshotCommits - statedb.TrieDBCommits) - blockInsertTimer.UpdateSince(start) + blockInsertTimer.UpdateSince(startTime) - return &blockProcessingResult{usedGas: res.GasUsed, procTime: proctime, status: status}, nil + return &blockProcessingResult{ + usedGas: res.GasUsed, + procTime: proctime, + status: status, + witness: witness, + }, nil } // insertSideChain is called when an import batch hits upon a pruned ancestor @@ -2627,7 +2642,8 @@ func (bc *BlockChain) InsertHeadersBeforeCutoff(headers []*types.Header) (int, e if err != nil { return 0, err } - if err := bc.db.Sync(); err != nil { + // Sync the ancient store explicitly to ensure all data has been flushed to disk. + if err := bc.db.SyncAncient(); err != nil { return 0, err } // Write hash to number mappings diff --git a/core/blockchain_insert.go b/core/blockchain_insert.go index ec3f771818..695aa6679b 100644 --- a/core/blockchain_insert.go +++ b/core/blockchain_insert.go @@ -43,8 +43,12 @@ func (st *insertStats) report(chain []*types.Block, index int, snapDiffItems, sn // Fetch the timings for the batch var ( now = mclock.Now() - elapsed = now.Sub(st.startTime) + elapsed = now.Sub(st.startTime) + 1 // prevent zero division + mgasps = float64(st.usedGas) * 1000 / float64(elapsed) ) + // Update the Mgas per second gauge + chainMgaspsGauge.Update(int64(mgasps)) + // If we're at the last block of the batch or report period reached, log if index == len(chain)-1 || elapsed >= statsReportLimit { // Count the number of transactions in this segment @@ -58,7 +62,7 @@ func (st *insertStats) report(chain []*types.Block, index int, snapDiffItems, sn context := []interface{}{ "number", end.Number(), "hash", end.Hash(), "blocks", st.processed, "txs", txs, "mgas", float64(st.usedGas) / 1000000, - "elapsed", common.PrettyDuration(elapsed), "mgasps", float64(st.usedGas) * 1000 / float64(elapsed), + "elapsed", common.PrettyDuration(elapsed), "mgasps", mgasps, } if timestamp := time.Unix(int64(end.Time()), 0); time.Since(timestamp) > time.Minute { context = append(context, []interface{}{"age", common.PrettyAge(timestamp)}...) @@ -138,6 +142,7 @@ func (it *insertIterator) next() (*types.Block, error) { // // Both header and body validation errors (nil too) is cached into the iterator // to avoid duplicating work on the following next() call. +// nolint:unused func (it *insertIterator) peek() (*types.Block, error) { // If we reached the end of the chain, abort if it.index+1 >= len(it.chain) { diff --git a/core/blockchain_repair_test.go b/core/blockchain_repair_test.go index 3ff1d77fc8..6c52d057ad 100644 --- a/core/blockchain_repair_test.go +++ b/core/blockchain_repair_test.go @@ -1765,7 +1765,7 @@ func testRepairWithScheme(t *testing.T, tt *rewindTest, snapshots bool, scheme s datadir := t.TempDir() ancient := filepath.Join(datadir, "ancient") - pdb, err := pebble.New(datadir, 0, 0, "", false, true) + pdb, err := pebble.New(datadir, 0, 0, "", false) if err != nil { t.Fatalf("Failed to create persistent key-value database: %v", err) } @@ -1850,7 +1850,7 @@ func testRepairWithScheme(t *testing.T, tt *rewindTest, snapshots bool, scheme s chain.stopWithoutSaving() // Start a new blockchain back up and see where the repair leads us - pdb, err = pebble.New(datadir, 0, 0, "", false, true) + pdb, err = pebble.New(datadir, 0, 0, "", false) if err != nil { t.Fatalf("Failed to reopen persistent key-value database: %v", err) } @@ -1915,7 +1915,7 @@ func testIssue23496(t *testing.T, scheme string) { datadir := t.TempDir() ancient := filepath.Join(datadir, "ancient") - pdb, err := pebble.New(datadir, 0, 0, "", false, true) + pdb, err := pebble.New(datadir, 0, 0, "", false) if err != nil { t.Fatalf("Failed to create persistent key-value database: %v", err) } @@ -1973,7 +1973,7 @@ func testIssue23496(t *testing.T, scheme string) { chain.stopWithoutSaving() // Start a new blockchain back up and see where the repair leads us - pdb, err = pebble.New(datadir, 0, 0, "", false, true) + pdb, err = pebble.New(datadir, 0, 0, "", false) if err != nil { t.Fatalf("Failed to reopen persistent key-value database: %v", err) } diff --git a/core/blockchain_sethead_test.go b/core/blockchain_sethead_test.go index e998b510df..424854b2bf 100644 --- a/core/blockchain_sethead_test.go +++ b/core/blockchain_sethead_test.go @@ -1969,7 +1969,7 @@ func testSetHeadWithScheme(t *testing.T, tt *rewindTest, snapshots bool, scheme datadir := t.TempDir() ancient := filepath.Join(datadir, "ancient") - pdb, err := pebble.New(datadir, 0, 0, "", false, true) + pdb, err := pebble.New(datadir, 0, 0, "", false) if err != nil { t.Fatalf("Failed to create persistent key-value database: %v", err) } diff --git a/core/blockchain_snapshot_test.go b/core/blockchain_snapshot_test.go index 23effea15e..1a6fe38af6 100644 --- a/core/blockchain_snapshot_test.go +++ b/core/blockchain_snapshot_test.go @@ -66,7 +66,7 @@ func (basic *snapshotTestBasic) prepare(t *testing.T) (*BlockChain, []*types.Blo datadir := t.TempDir() ancient := filepath.Join(datadir, "ancient") - pdb, err := pebble.New(datadir, 0, 0, "", false, true) + pdb, err := pebble.New(datadir, 0, 0, "", false) if err != nil { t.Fatalf("Failed to create persistent key-value database: %v", err) } @@ -257,7 +257,7 @@ func (snaptest *crashSnapshotTest) test(t *testing.T) { chain.triedb.Close() // Start a new blockchain back up and see where the repair leads us - pdb, err := pebble.New(snaptest.datadir, 0, 0, "", false, true) + pdb, err := pebble.New(snaptest.datadir, 0, 0, "", false) if err != nil { t.Fatalf("Failed to create persistent key-value database: %v", err) } diff --git a/core/blockchain_test.go b/core/blockchain_test.go index 134deee237..b981c33f21 100644 --- a/core/blockchain_test.go +++ b/core/blockchain_test.go @@ -2492,7 +2492,7 @@ func testSideImportPrunedBlocks(t *testing.T, scheme string) { datadir := t.TempDir() ancient := path.Join(datadir, "ancient") - pdb, err := pebble.New(datadir, 0, 0, "", false, true) + pdb, err := pebble.New(datadir, 0, 0, "", false) if err != nil { t.Fatalf("Failed to create persistent key-value database: %v", err) } diff --git a/core/headerchain.go b/core/headerchain.go index f7acc49bef..6e70dfa865 100644 --- a/core/headerchain.go +++ b/core/headerchain.go @@ -591,17 +591,50 @@ func (hc *HeaderChain) setHead(headBlock uint64, headTime uint64, updateFn Updat hashes = append(hashes, hdr.Hash()) } for _, hash := range hashes { + // Remove the associated block body and receipts if required. + // + // If the block is in the chain freezer, then this delete operation + // is actually ineffective. if delFn != nil { delFn(batch, hash, num) } + // Remove the hash->number mapping along with the header itself rawdb.DeleteHeader(batch, hash, num) } + // Remove the number->hash mapping rawdb.DeleteCanonicalHash(batch, num) } } // Flush all accumulated deletions. if err := batch.Write(); err != nil { - log.Crit("Failed to rewind block", "error", err) + log.Crit("Failed to commit batch in setHead", "err", err) + } + // Explicitly flush the pending writes in the key-value store to disk, ensuring + // data durability of the previous deletions. + if err := hc.chainDb.SyncKeyValue(); err != nil { + log.Crit("Failed to sync the key-value store in setHead", "err", err) + } + // Truncate the excessive chain segments in the ancient store. + // These are actually deferred deletions from the loop above. + // + // This step must be performed after synchronizing the key-value store; + // otherwise, in the event of a panic, it's theoretically possible to + // lose recent key-value store writes while the ancient store deletions + // remain, leading to data inconsistency, e.g., the gap between the key + // value store and ancient can be created due to unclean shutdown. + if delFn != nil { + // Ignore the error here since light client won't hit this path + frozen, _ := hc.chainDb.Ancients() + header := hc.CurrentHeader() + + // Truncate the excessive chain segment above the current chain head + // in the ancient store. + if header.Number.Uint64()+1 < frozen { + _, err := hc.chainDb.TruncateHead(header.Number.Uint64() + 1) + if err != nil { + log.Crit("Failed to truncate head block", "err", err) + } + } } // Clear out any stale content from the caches hc.headerCache.Purge() diff --git a/core/rawdb/accessors_state.go b/core/rawdb/accessors_state.go index adc77fae83..41e15debe9 100644 --- a/core/rawdb/accessors_state.go +++ b/core/rawdb/accessors_state.go @@ -258,13 +258,21 @@ func ReadStateHistory(db ethdb.AncientReaderOp, id uint64) ([]byte, []byte, []by // WriteStateHistory writes the provided state history to database. Compute the // position of state history in freezer by minus one since the id of first state // history starts from one(zero for initial state). -func WriteStateHistory(db ethdb.AncientWriter, id uint64, meta []byte, accountIndex []byte, storageIndex []byte, accounts []byte, storages []byte) { - db.ModifyAncients(func(op ethdb.AncientWriteOp) error { - op.AppendRaw(stateHistoryMeta, id-1, meta) - op.AppendRaw(stateHistoryAccountIndex, id-1, accountIndex) - op.AppendRaw(stateHistoryStorageIndex, id-1, storageIndex) - op.AppendRaw(stateHistoryAccountData, id-1, accounts) - op.AppendRaw(stateHistoryStorageData, id-1, storages) - return nil +func WriteStateHistory(db ethdb.AncientWriter, id uint64, meta []byte, accountIndex []byte, storageIndex []byte, accounts []byte, storages []byte) error { + _, err := db.ModifyAncients(func(op ethdb.AncientWriteOp) error { + if err := op.AppendRaw(stateHistoryMeta, id-1, meta); err != nil { + return err + } + if err := op.AppendRaw(stateHistoryAccountIndex, id-1, accountIndex); err != nil { + return err + } + if err := op.AppendRaw(stateHistoryStorageIndex, id-1, storageIndex); err != nil { + return err + } + if err := op.AppendRaw(stateHistoryAccountData, id-1, accounts); err != nil { + return err + } + return op.AppendRaw(stateHistoryStorageData, id-1, storages) }) + return err } diff --git a/core/rawdb/accessors_trie.go b/core/rawdb/accessors_trie.go index 8bd6b71eee..e154ab527b 100644 --- a/core/rawdb/accessors_trie.go +++ b/core/rawdb/accessors_trie.go @@ -18,7 +18,6 @@ package rawdb import ( "fmt" - "sync" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" @@ -45,25 +44,6 @@ const HashScheme = "hash" // on extra state diffs to survive deep reorg. const PathScheme = "path" -// hasher is used to compute the sha256 hash of the provided data. -type hasher struct{ sha crypto.KeccakState } - -var hasherPool = sync.Pool{ - New: func() interface{} { return &hasher{sha: crypto.NewKeccakState()} }, -} - -func newHasher() *hasher { - return hasherPool.Get().(*hasher) -} - -func (h *hasher) hash(data []byte) common.Hash { - return crypto.HashData(h.sha, data) -} - -func (h *hasher) release() { - hasherPool.Put(h) -} - // ReadAccountTrieNode retrieves the account trie node with the specified node path. func ReadAccountTrieNode(db ethdb.KeyValueReader, path []byte) []byte { data, _ := db.Get(accountTrieNodeKey(path)) @@ -170,9 +150,7 @@ func HasTrieNode(db ethdb.KeyValueReader, owner common.Hash, path []byte, hash c if len(blob) == 0 { return false } - h := newHasher() - defer h.release() - return h.hash(blob) == hash // exists but not match + return crypto.Keccak256Hash(blob) == hash // exists but not match default: panic(fmt.Sprintf("Unknown scheme %v", scheme)) } @@ -194,9 +172,7 @@ func ReadTrieNode(db ethdb.KeyValueReader, owner common.Hash, path []byte, hash if len(blob) == 0 { return nil } - h := newHasher() - defer h.release() - if h.hash(blob) != hash { + if crypto.Keccak256Hash(blob) != hash { return nil // exists but not match } return blob diff --git a/core/rawdb/chain_freezer.go b/core/rawdb/chain_freezer.go index f3c671f45a..cc7a62df32 100644 --- a/core/rawdb/chain_freezer.go +++ b/core/rawdb/chain_freezer.go @@ -205,7 +205,7 @@ func (f *chainFreezer) freeze(db ethdb.KeyValueStore) { continue } // Batch of blocks have been frozen, flush them before wiping from key-value store - if err := f.Sync(); err != nil { + if err := f.SyncAncient(); err != nil { log.Crit("Failed to flush frozen tables", "err", err) } // Wipe out all data from the active database diff --git a/core/rawdb/database.go b/core/rawdb/database.go index 2a50e3f9ee..a03dbafb1f 100644 --- a/core/rawdb/database.go +++ b/core/rawdb/database.go @@ -131,8 +131,8 @@ func (db *nofreezedb) TruncateTail(items uint64) (uint64, error) { return 0, errNotSupported } -// Sync returns an error as we don't have a backing chain freezer. -func (db *nofreezedb) Sync() error { +// SyncAncient returns an error as we don't have a backing chain freezer. +func (db *nofreezedb) SyncAncient() error { return errNotSupported } diff --git a/core/rawdb/freezer.go b/core/rawdb/freezer.go index 105d3af934..1e5b98c7fe 100644 --- a/core/rawdb/freezer.go +++ b/core/rawdb/freezer.go @@ -325,8 +325,8 @@ func (f *Freezer) TruncateTail(tail uint64) (uint64, error) { return old, nil } -// Sync flushes all data tables to disk. -func (f *Freezer) Sync() error { +// SyncAncient flushes all data tables to disk. +func (f *Freezer) SyncAncient() error { var errs []error for _, table := range f.tables { if err := table.Sync(); err != nil { diff --git a/core/rawdb/freezer_memory.go b/core/rawdb/freezer_memory.go index 4274546de5..bd286f45f5 100644 --- a/core/rawdb/freezer_memory.go +++ b/core/rawdb/freezer_memory.go @@ -395,8 +395,8 @@ func (f *MemoryFreezer) TruncateTail(tail uint64) (uint64, error) { return old, nil } -// Sync flushes all data tables to disk. -func (f *MemoryFreezer) Sync() error { +// SyncAncient flushes all data tables to disk. +func (f *MemoryFreezer) SyncAncient() error { return nil } diff --git a/core/rawdb/freezer_resettable.go b/core/rawdb/freezer_resettable.go index 2e64e6074c..01df2877d9 100644 --- a/core/rawdb/freezer_resettable.go +++ b/core/rawdb/freezer_resettable.go @@ -194,12 +194,12 @@ func (f *resettableFreezer) TruncateTail(tail uint64) (uint64, error) { return f.freezer.TruncateTail(tail) } -// Sync flushes all data tables to disk. -func (f *resettableFreezer) Sync() error { +// SyncAncient flushes all data tables to disk. +func (f *resettableFreezer) SyncAncient() error { f.lock.RLock() defer f.lock.RUnlock() - return f.freezer.Sync() + return f.freezer.SyncAncient() } // AncientDatadir returns the path of the ancient store. diff --git a/core/rawdb/freezer_test.go b/core/rawdb/freezer_test.go index 150734d3ac..a7a3559ec4 100644 --- a/core/rawdb/freezer_test.go +++ b/core/rawdb/freezer_test.go @@ -392,7 +392,7 @@ func TestFreezerCloseSync(t *testing.T) { if err := f.Close(); err != nil { t.Fatal(err) } - if err := f.Sync(); err == nil { + if err := f.SyncAncient(); err == nil { t.Fatalf("want error, have nil") } else if have, want := err.Error(), "[closed closed]"; have != want { t.Fatalf("want %v, have %v", have, want) diff --git a/core/rawdb/table.go b/core/rawdb/table.go index 1a9060b636..9a342a8217 100644 --- a/core/rawdb/table.go +++ b/core/rawdb/table.go @@ -107,10 +107,10 @@ func (t *table) TruncateTail(items uint64) (uint64, error) { return t.db.TruncateTail(items) } -// Sync is a noop passthrough that just forwards the request to the underlying +// SyncAncient is a noop passthrough that just forwards the request to the underlying // database. -func (t *table) Sync() error { - return t.db.Sync() +func (t *table) SyncAncient() error { + return t.db.SyncAncient() } // AncientDatadir returns the ancient datadir of the underlying database. @@ -188,6 +188,12 @@ func (t *table) Compact(start []byte, limit []byte) error { return t.db.Compact(start, limit) } +// SyncKeyValue ensures that all pending writes are flushed to disk, +// guaranteeing data durability up to the point. +func (t *table) SyncKeyValue() error { + return t.db.SyncKeyValue() +} + // NewBatch creates a write-only database that buffers changes to its host db // until a final write is called, each operation prefixing all keys with the // pre-configured string. diff --git a/core/state/database.go b/core/state/database.go index faf4954650..cef59cccfb 100644 --- a/core/state/database.go +++ b/core/state/database.go @@ -34,10 +34,10 @@ import ( const ( // Number of codehash->size associations to keep. - codeSizeCacheSize = 100000 + codeSizeCacheSize = 1_000_000 // 4 megabytes in total // Cache size granted for caching clean code. - codeCacheSize = 64 * 1024 * 1024 + codeCacheSize = 256 * 1024 * 1024 // Number of address->curve point associations to keep. pointCacheSize = 4096 @@ -208,6 +208,15 @@ func (db *CachingDB) Reader(stateRoot common.Hash) (Reader, error) { return newReader(newCachingCodeReader(db.disk, db.codeCache, db.codeSizeCache), combined), nil } +// ReaderWithCache creates a state reader with internal local cache. +func (db *CachingDB) ReaderWithCache(stateRoot common.Hash) (Reader, error) { + reader, err := db.Reader(stateRoot) + if err != nil { + return nil, err + } + return newReaderWithCache(reader), nil +} + // OpenTrie opens the main account trie at a specific root hash. func (db *CachingDB) OpenTrie(root common.Hash) (Trie, error) { if db.triedb.IsVerkle() { diff --git a/core/state/reader.go b/core/state/reader.go index a0f15dfcc8..09edf6ab8d 100644 --- a/core/state/reader.go +++ b/core/state/reader.go @@ -18,6 +18,7 @@ package state import ( "errors" + "sync" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/lru" @@ -51,6 +52,9 @@ type ContractCodeReader interface { // StateReader defines the interface for accessing accounts and storage slots // associated with a specific state. +// +// StateReader is assumed to be thread-safe and implementation must take care +// of the concurrency issue by themselves. type StateReader interface { // Account retrieves the account associated with a particular address. // @@ -70,6 +74,9 @@ type StateReader interface { // Reader defines the interface for accessing accounts, storage slots and contract // code associated with a specific state. +// +// Reader is assumed to be thread-safe and implementation must take care of the +// concurrency issue by themselves. type Reader interface { ContractCodeReader StateReader @@ -77,6 +84,8 @@ type Reader interface { // cachingCodeReader implements ContractCodeReader, accessing contract code either in // local key-value store or the shared code cache. +// +// cachingCodeReader is safe for concurrent access. type cachingCodeReader struct { db ethdb.KeyValueReader @@ -123,18 +132,14 @@ func (r *cachingCodeReader) CodeSize(addr common.Address, codeHash common.Hash) return len(code), nil } -// flatReader wraps a database state reader. +// flatReader wraps a database state reader and is safe for concurrent access. type flatReader struct { reader database.StateReader - buff crypto.KeccakState } // newFlatReader constructs a state reader with on the given state root. func newFlatReader(reader database.StateReader) *flatReader { - return &flatReader{ - reader: reader, - buff: crypto.NewKeccakState(), - } + return &flatReader{reader: reader} } // Account implements StateReader, retrieving the account specified by the address. @@ -144,7 +149,7 @@ func newFlatReader(reader database.StateReader) *flatReader { // // The returned account might be nil if it's not existent. func (r *flatReader) Account(addr common.Address) (*types.StateAccount, error) { - account, err := r.reader.Account(crypto.HashData(r.buff, addr.Bytes())) + account, err := r.reader.Account(crypto.Keccak256Hash(addr.Bytes())) if err != nil { return nil, err } @@ -174,8 +179,8 @@ func (r *flatReader) Account(addr common.Address) (*types.StateAccount, error) { // // The returned storage slot might be empty if it's not existent. func (r *flatReader) Storage(addr common.Address, key common.Hash) (common.Hash, error) { - addrHash := crypto.HashData(r.buff, addr.Bytes()) - slotHash := crypto.HashData(r.buff, key.Bytes()) + addrHash := crypto.Keccak256Hash(addr.Bytes()) + slotHash := crypto.Keccak256Hash(key.Bytes()) ret, err := r.reader.Storage(addrHash, slotHash) if err != nil { return common.Hash{}, err @@ -196,13 +201,20 @@ func (r *flatReader) Storage(addr common.Address, key common.Hash) (common.Hash, // trieReader implements the StateReader interface, providing functions to access // state from the referenced trie. +// +// trieReader is safe for concurrent read. type trieReader struct { - root common.Hash // State root which uniquely represent a state - db *triedb.Database // Database for loading trie - buff crypto.KeccakState // Buffer for keccak256 hashing - mainTrie Trie // Main trie, resolved in constructor + root common.Hash // State root which uniquely represent a state + db *triedb.Database // Database for loading trie + buff crypto.KeccakState // Buffer for keccak256 hashing + + // Main trie, resolved in constructor. Note either the Merkle-Patricia-tree + // or Verkle-tree is not safe for concurrent read. + mainTrie Trie + subRoots map[common.Address]common.Hash // Set of storage roots, cached when the account is resolved subTries map[common.Address]Trie // Group of storage tries, cached when it's resolved + lock sync.Mutex // Lock for protecting concurrent read } // trieReader constructs a trie reader of the specific state. An error will be @@ -230,11 +242,8 @@ func newTrieReader(root common.Hash, db *triedb.Database, cache *utils.PointCach }, nil } -// Account implements StateReader, retrieving the account specified by the address. -// -// An error will be returned if the trie state is corrupted. An nil account -// will be returned if it's not existent in the trie. -func (r *trieReader) Account(addr common.Address) (*types.StateAccount, error) { +// account is the inner version of Account and assumes the r.lock is already held. +func (r *trieReader) account(addr common.Address) (*types.StateAccount, error) { account, err := r.mainTrie.GetAccount(addr) if err != nil { return nil, err @@ -247,12 +256,26 @@ func (r *trieReader) Account(addr common.Address) (*types.StateAccount, error) { return account, nil } +// Account implements StateReader, retrieving the account specified by the address. +// +// An error will be returned if the trie state is corrupted. An nil account +// will be returned if it's not existent in the trie. +func (r *trieReader) Account(addr common.Address) (*types.StateAccount, error) { + r.lock.Lock() + defer r.lock.Unlock() + + return r.account(addr) +} + // Storage implements StateReader, retrieving the storage slot specified by the // address and slot key. // // An error will be returned if the trie state is corrupted. An empty storage // slot will be returned if it's not existent in the trie. func (r *trieReader) Storage(addr common.Address, key common.Hash) (common.Hash, error) { + r.lock.Lock() + defer r.lock.Unlock() + var ( tr Trie found bool @@ -268,7 +291,7 @@ func (r *trieReader) Storage(addr common.Address, key common.Hash) (common.Hash, // The storage slot is accessed without account caching. It's unexpected // behavior but try to resolve the account first anyway. if !ok { - _, err := r.Account(addr) + _, err := r.account(addr) if err != nil { return common.Hash{}, err } @@ -293,6 +316,9 @@ func (r *trieReader) Storage(addr common.Address, key common.Hash) (common.Hash, // multiStateReader is the aggregation of a list of StateReader interface, // providing state access by leveraging all readers. The checking priority // is determined by the position in the reader list. +// +// multiStateReader is safe for concurrent read and assumes all underlying +// readers are thread-safe as well. type multiStateReader struct { readers []StateReader // List of state readers, sorted by checking priority } @@ -358,3 +384,95 @@ func newReader(codeReader ContractCodeReader, stateReader StateReader) *reader { StateReader: stateReader, } } + +// readerWithCache is a wrapper around Reader that maintains additional state caches +// to support concurrent state access. +type readerWithCache struct { + Reader // safe for concurrent read + + // Previously resolved state entries. + accounts map[common.Address]*types.StateAccount + accountLock sync.RWMutex + + // List of storage buckets, each of which is thread-safe. + // This reader is typically used in scenarios requiring concurrent + // access to storage. Using multiple buckets helps mitigate + // the overhead caused by locking. + storageBuckets [16]struct { + lock sync.RWMutex + storages map[common.Address]map[common.Hash]common.Hash + } +} + +// newReaderWithCache constructs the reader with local cache. +func newReaderWithCache(reader Reader) *readerWithCache { + r := &readerWithCache{ + Reader: reader, + accounts: make(map[common.Address]*types.StateAccount), + } + for i := range r.storageBuckets { + r.storageBuckets[i].storages = make(map[common.Address]map[common.Hash]common.Hash) + } + return r +} + +// Account implements StateReader, retrieving the account specified by the address. +// The returned account might be nil if it's not existent. +// +// An error will be returned if the state is corrupted in the underlying reader. +func (r *readerWithCache) Account(addr common.Address) (*types.StateAccount, error) { + // Try to resolve the requested account in the local cache + r.accountLock.RLock() + acct, ok := r.accounts[addr] + r.accountLock.RUnlock() + if ok { + return acct, nil + } + // Try to resolve the requested account from the underlying reader + acct, err := r.Reader.Account(addr) + if err != nil { + return nil, err + } + r.accountLock.Lock() + r.accounts[addr] = acct + r.accountLock.Unlock() + return acct, nil +} + +// Storage implements StateReader, retrieving the storage slot specified by the +// address and slot key. The returned storage slot might be empty if it's not +// existent. +// +// An error will be returned if the state is corrupted in the underlying reader. +func (r *readerWithCache) Storage(addr common.Address, slot common.Hash) (common.Hash, error) { + var ( + value common.Hash + ok bool + bucket = &r.storageBuckets[addr[0]&0x0f] + ) + // Try to resolve the requested storage slot in the local cache + bucket.lock.RLock() + slots, ok := bucket.storages[addr] + if ok { + value, ok = slots[slot] + } + bucket.lock.RUnlock() + if ok { + return value, nil + } + // Try to resolve the requested storage slot from the underlying reader + value, err := r.Reader.Storage(addr, slot) + if err != nil { + return common.Hash{}, err + } + bucket.lock.Lock() + slots, ok = bucket.storages[addr] + if !ok { + slots = make(map[common.Hash]common.Hash) + bucket.storages[addr] = slots + } + slots[slot] = value + bucket.lock.Unlock() + + return value, nil +} diff --git a/core/state/statedb.go b/core/state/statedb.go index e3f5b9e1a0..9378cae7de 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -159,11 +159,17 @@ type StateDB struct { // New creates a new state from a given trie. func New(root common.Hash, db Database) (*StateDB, error) { - tr, err := db.OpenTrie(root) + reader, err := db.Reader(root) if err != nil { return nil, err } - reader, err := db.Reader(root) + return NewWithReader(root, db, reader) +} + +// NewWithReader creates a new state for the specified state root. Unlike New, +// this function accepts an additional Reader which is bound to the given root. +func NewWithReader(root common.Hash, db Database, reader Reader) (*StateDB, error) { + tr, err := db.OpenTrie(root) if err != nil { return nil, err } @@ -392,6 +398,12 @@ func (s *StateDB) Database() Database { return s.db } +// Reader retrieves the low level database reader supporting the +// lower level operations. +func (s *StateDB) Reader() Reader { + return s.reader +} + func (s *StateDB) HasSelfDestructed(addr common.Address) bool { stateObject := s.getStateObject(addr) if stateObject != nil { @@ -650,11 +662,10 @@ func (s *StateDB) CreateContract(addr common.Address) { // Snapshots of the copied state cannot be applied to the copy. func (s *StateDB) Copy() *StateDB { // Copy all the basic fields, initialize the memory ones - reader, _ := s.db.Reader(s.originalRoot) // impossible to fail state := &StateDB{ db: s.db, trie: mustCopyTrie(s.trie), - reader: reader, + reader: s.reader, originalRoot: s.originalRoot, stateObjects: make(map[common.Address]*stateObject, len(s.stateObjects)), stateObjectsDestruct: make(map[common.Address]*stateObject, len(s.stateObjectsDestruct)), diff --git a/core/state_prefetcher.go b/core/state_prefetcher.go index 805df5ef62..f3129f57cd 100644 --- a/core/state_prefetcher.go +++ b/core/state_prefetcher.go @@ -17,17 +17,22 @@ package core import ( + "bytes" + "runtime" "sync/atomic" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/params" + "golang.org/x/sync/errgroup" ) -// statePrefetcher is a basic Prefetcher, which blindly executes a block on top -// of an arbitrary state with the goal of prefetching potentially useful state -// data from disk before the main block processor start executing. +// statePrefetcher is a basic Prefetcher that executes transactions from a block +// on top of the parent state, aiming to prefetch potentially useful state data +// from disk. Transactions are executed in parallel to fully leverage the +// SSD's read performance. type statePrefetcher struct { config *params.ChainConfig // Chain configuration options chain *HeaderChain // Canonical block chain @@ -43,41 +48,81 @@ func newStatePrefetcher(config *params.ChainConfig, chain *HeaderChain) *statePr // Prefetch processes the state changes according to the Ethereum rules by running // the transaction messages using the statedb, but any changes are discarded. The -// only goal is to pre-cache transaction signatures and state trie nodes. +// only goal is to warm the state caches. func (p *statePrefetcher) Prefetch(block *types.Block, statedb *state.StateDB, cfg vm.Config, interrupt *atomic.Bool) { var ( - header = block.Header() - gaspool = new(GasPool).AddGas(block.GasLimit()) - blockContext = NewEVMBlockContext(header, p.chain, nil) - evm = vm.NewEVM(blockContext, statedb, p.config, cfg) - signer = types.MakeSigner(p.config, header.Number, header.Time) + fails atomic.Int64 + header = block.Header() + signer = types.MakeSigner(p.config, header.Number, header.Time) + workers errgroup.Group + reader = statedb.Reader() ) - // Iterate over and process the individual transactions - byzantium := p.config.IsByzantium(block.Number()) - for i, tx := range block.Transactions() { - // If block precaching was interrupted, abort - if interrupt != nil && interrupt.Load() { - return - } - // Convert the transaction into an executable message and pre-cache its sender - msg, err := TransactionToMessage(tx, signer, header.BaseFee) - if err != nil { - return // Also invalid block, bail out - } - statedb.SetTxContext(tx.Hash(), i) + workers.SetLimit(runtime.NumCPU() / 2) - // We attempt to apply a transaction. The goal is not to execute - // the transaction successfully, rather to warm up touched data slots. - if _, err := ApplyMessage(evm, msg, gaspool); err != nil { - return // Ugh, something went horribly wrong, bail out - } - // If we're pre-byzantium, pre-load trie nodes for the intermediate root - if !byzantium { - statedb.IntermediateRoot(true) - } - } - // If were post-byzantium, pre-load trie nodes for the final root hash - if byzantium { - statedb.IntermediateRoot(true) + // Iterate over and process the individual transactions + for i, tx := range block.Transactions() { + stateCpy := statedb.Copy() // closure + workers.Go(func() error { + // If block precaching was interrupted, abort + if interrupt != nil && interrupt.Load() { + return nil + } + // Preload the touched accounts and storage slots in advance + sender, err := types.Sender(signer, tx) + if err != nil { + fails.Add(1) + return nil + } + reader.Account(sender) + + if tx.To() != nil { + account, _ := reader.Account(*tx.To()) + + // Preload the contract code if the destination has non-empty code + if account != nil && !bytes.Equal(account.CodeHash, types.EmptyCodeHash.Bytes()) { + reader.Code(*tx.To(), common.BytesToHash(account.CodeHash)) + } + } + for _, list := range tx.AccessList() { + reader.Account(list.Address) + if len(list.StorageKeys) > 0 { + for _, slot := range list.StorageKeys { + reader.Storage(list.Address, slot) + } + } + } + // Execute the message to preload the implicit touched states + evm := vm.NewEVM(NewEVMBlockContext(header, p.chain, nil), stateCpy, p.config, cfg) + + // Convert the transaction into an executable message and pre-cache its sender + msg, err := TransactionToMessage(tx, signer, header.BaseFee) + if err != nil { + fails.Add(1) + return nil // Also invalid block, bail out + } + // Disable the nonce check + msg.SkipNonceChecks = true + + stateCpy.SetTxContext(tx.Hash(), i) + + // We attempt to apply a transaction. The goal is not to execute + // the transaction successfully, rather to warm up touched data slots. + if _, err := ApplyMessage(evm, msg, new(GasPool).AddGas(block.GasLimit())); err != nil { + fails.Add(1) + return nil // Ugh, something went horribly wrong, bail out + } + // Pre-load trie nodes for the intermediate root. + // + // This operation incurs significant memory allocations due to + // trie hashing and node decoding. TODO(rjl493456442): investigate + // ways to mitigate this overhead. + stateCpy.IntermediateRoot(true) + return nil + }) } + workers.Wait() + + blockPrefetchTxsValidMeter.Mark(int64(len(block.Transactions())) - fails.Load()) + blockPrefetchTxsInvalidMeter.Mark(fails.Load()) + return } diff --git a/core/state_transition.go b/core/state_transition.go index f9c9a2ab5a..ff2051ddd2 100644 --- a/core/state_transition.go +++ b/core/state_transition.go @@ -159,7 +159,9 @@ type Message struct { // When SkipNonceChecks is true, the message nonce is not checked against the // account nonce in state. - // This field will be set to true for operations like RPC eth_call. + // + // This field will be set to true for operations like RPC eth_call + // or the state prefetching. SkipNonceChecks bool // When SkipFromEOACheck is true, the message sender is not checked to be an EOA. diff --git a/core/types/account.go b/core/types/account.go index 52ce184cda..bcfb83418c 100644 --- a/core/types/account.go +++ b/core/types/account.go @@ -38,17 +38,13 @@ type Account struct { Storage map[common.Hash]common.Hash `json:"storage,omitempty"` Balance *big.Int `json:"balance" gencodec:"required"` Nonce uint64 `json:"nonce,omitempty"` - - // used in tests - PrivateKey []byte `json:"secretKey,omitempty"` } type accountMarshaling struct { - Code hexutil.Bytes - Balance *math.HexOrDecimal256 - Nonce math.HexOrDecimal64 - Storage map[storageJSON]storageJSON - PrivateKey hexutil.Bytes + Code hexutil.Bytes + Balance *math.HexOrDecimal256 + Nonce math.HexOrDecimal64 + Storage map[storageJSON]storageJSON } // storageJSON represents a 256 bit byte array, but allows less than 256 bits when diff --git a/core/types/gen_account.go b/core/types/gen_account.go index 4e475896a7..89165ee3ad 100644 --- a/core/types/gen_account.go +++ b/core/types/gen_account.go @@ -17,11 +17,10 @@ var _ = (*accountMarshaling)(nil) // MarshalJSON marshals as JSON. func (a Account) MarshalJSON() ([]byte, error) { type Account struct { - Code hexutil.Bytes `json:"code,omitempty"` - Storage map[storageJSON]storageJSON `json:"storage,omitempty"` - Balance *math.HexOrDecimal256 `json:"balance" gencodec:"required"` - Nonce math.HexOrDecimal64 `json:"nonce,omitempty"` - PrivateKey hexutil.Bytes `json:"secretKey,omitempty"` + Code hexutil.Bytes `json:"code,omitempty"` + Storage map[storageJSON]storageJSON `json:"storage,omitempty"` + Balance *math.HexOrDecimal256 `json:"balance" gencodec:"required"` + Nonce math.HexOrDecimal64 `json:"nonce,omitempty"` } var enc Account enc.Code = a.Code @@ -33,18 +32,16 @@ func (a Account) MarshalJSON() ([]byte, error) { } enc.Balance = (*math.HexOrDecimal256)(a.Balance) enc.Nonce = math.HexOrDecimal64(a.Nonce) - enc.PrivateKey = a.PrivateKey return json.Marshal(&enc) } // UnmarshalJSON unmarshals from JSON. func (a *Account) UnmarshalJSON(input []byte) error { type Account struct { - Code *hexutil.Bytes `json:"code,omitempty"` - Storage map[storageJSON]storageJSON `json:"storage,omitempty"` - Balance *math.HexOrDecimal256 `json:"balance" gencodec:"required"` - Nonce *math.HexOrDecimal64 `json:"nonce,omitempty"` - PrivateKey *hexutil.Bytes `json:"secretKey,omitempty"` + Code *hexutil.Bytes `json:"code,omitempty"` + Storage map[storageJSON]storageJSON `json:"storage,omitempty"` + Balance *math.HexOrDecimal256 `json:"balance" gencodec:"required"` + Nonce *math.HexOrDecimal64 `json:"nonce,omitempty"` } var dec Account if err := json.Unmarshal(input, &dec); err != nil { @@ -66,8 +63,5 @@ func (a *Account) UnmarshalJSON(input []byte) error { if dec.Nonce != nil { a.Nonce = uint64(*dec.Nonce) } - if dec.PrivateKey != nil { - a.PrivateKey = *dec.PrivateKey - } return nil } diff --git a/core/types/hashing.go b/core/types/hashing.go index 224d7a87ea..3cc22d50d1 100644 --- a/core/types/hashing.go +++ b/core/types/hashing.go @@ -25,12 +25,11 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/rlp" - "golang.org/x/crypto/sha3" ) // hasherPool holds LegacyKeccak256 hashers for rlpHash. var hasherPool = sync.Pool{ - New: func() interface{} { return sha3.NewLegacyKeccak256() }, + New: func() interface{} { return crypto.NewKeccakState() }, } // encodeBufferPool holds temporary encoder buffers for DeriveSha and TX encoding. diff --git a/core/types/transaction.go b/core/types/transaction.go index a2f4104635..934feb7353 100644 --- a/core/types/transaction.go +++ b/core/types/transaction.go @@ -355,28 +355,31 @@ func (tx *Transaction) GasTipCapIntCmp(other *big.Int) int { // Note: if the effective gasTipCap is negative, this method returns both error // the actual negative value, _and_ ErrGasFeeCapTooLow func (tx *Transaction) EffectiveGasTip(baseFee *big.Int) (*big.Int, error) { + dst := new(big.Int) + err := tx.calcEffectiveGasTip(dst, baseFee) + return dst, err +} + +// calcEffectiveGasTip calculates the effective gas tip of the transaction and +// saves the result to dst. +func (tx *Transaction) calcEffectiveGasTip(dst *big.Int, baseFee *big.Int) error { if baseFee == nil { - return tx.GasTipCap(), nil + dst.Set(tx.inner.gasTipCap()) + return nil } + var err error - gasFeeCap := tx.GasFeeCap() + gasFeeCap := tx.inner.gasFeeCap() if gasFeeCap.Cmp(baseFee) < 0 { err = ErrGasFeeCapTooLow } - gasFeeCap = gasFeeCap.Sub(gasFeeCap, baseFee) - gasTipCap := tx.GasTipCap() - if gasTipCap.Cmp(gasFeeCap) < 0 { - return gasTipCap, err + dst.Sub(gasFeeCap, baseFee) + gasTipCap := tx.inner.gasTipCap() + if gasTipCap.Cmp(dst) < 0 { + dst.Set(gasTipCap) } - return gasFeeCap, err -} - -// EffectiveGasTipValue is identical to EffectiveGasTip, but does not return an -// error in case the effective gasTipCap is negative -func (tx *Transaction) EffectiveGasTipValue(baseFee *big.Int) *big.Int { - effectiveTip, _ := tx.EffectiveGasTip(baseFee) - return effectiveTip + return err } // EffectiveGasTipCmp compares the effective gasTipCap of two transactions assuming the given base fee. @@ -384,7 +387,11 @@ func (tx *Transaction) EffectiveGasTipCmp(other *Transaction, baseFee *big.Int) if baseFee == nil { return tx.GasTipCapCmp(other) } - return tx.EffectiveGasTipValue(baseFee).Cmp(other.EffectiveGasTipValue(baseFee)) + // Use more efficient internal method. + txTip, otherTip := new(big.Int), new(big.Int) + tx.calcEffectiveGasTip(txTip, baseFee) + other.calcEffectiveGasTip(otherTip, baseFee) + return txTip.Cmp(otherTip) } // EffectiveGasTipIntCmp compares the effective gasTipCap of a transaction to the given gasTipCap. @@ -392,7 +399,9 @@ func (tx *Transaction) EffectiveGasTipIntCmp(other *big.Int, baseFee *big.Int) i if baseFee == nil { return tx.GasTipCapIntCmp(other) } - return tx.EffectiveGasTipValue(baseFee).Cmp(other) + txTip := new(big.Int) + tx.calcEffectiveGasTip(txTip, baseFee) + return txTip.Cmp(other) } // BlobGas returns the blob gas limit of the transaction for blob transactions, 0 otherwise. diff --git a/core/types/transaction_test.go b/core/types/transaction_test.go index 8922448d97..7d5e2f058a 100644 --- a/core/types/transaction_test.go +++ b/core/types/transaction_test.go @@ -29,6 +29,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rlp" ) @@ -593,3 +594,101 @@ func BenchmarkHash(b *testing.B) { signer.Hash(tx) } } + +func BenchmarkEffectiveGasTip(b *testing.B) { + signer := LatestSigner(params.TestChainConfig) + key, _ := crypto.GenerateKey() + txdata := &DynamicFeeTx{ + ChainID: big.NewInt(1), + Nonce: 0, + GasTipCap: big.NewInt(2000000000), + GasFeeCap: big.NewInt(3000000000), + Gas: 21000, + To: &common.Address{}, + Value: big.NewInt(0), + Data: nil, + } + tx, _ := SignNewTx(key, signer, txdata) + baseFee := big.NewInt(1000000000) // 1 gwei + + b.Run("Original", func(b *testing.B) { + b.ReportAllocs() + for i := 0; i < b.N; i++ { + _, err := tx.EffectiveGasTip(baseFee) + if err != nil { + b.Fatal(err) + } + } + }) + + b.Run("IntoMethod", func(b *testing.B) { + b.ReportAllocs() + dst := new(big.Int) + for i := 0; i < b.N; i++ { + err := tx.calcEffectiveGasTip(dst, baseFee) + if err != nil { + b.Fatal(err) + } + } + }) +} + +func TestEffectiveGasTipInto(t *testing.T) { + signer := LatestSigner(params.TestChainConfig) + key, _ := crypto.GenerateKey() + + testCases := []struct { + tipCap int64 + feeCap int64 + baseFee *int64 + }{ + {tipCap: 1, feeCap: 100, baseFee: intPtr(50)}, + {tipCap: 10, feeCap: 100, baseFee: intPtr(50)}, + {tipCap: 50, feeCap: 100, baseFee: intPtr(50)}, + {tipCap: 100, feeCap: 100, baseFee: intPtr(50)}, + {tipCap: 1, feeCap: 50, baseFee: intPtr(50)}, + {tipCap: 1, feeCap: 20, baseFee: intPtr(50)}, // Base fee higher than fee cap + {tipCap: 50, feeCap: 100, baseFee: intPtr(0)}, + {tipCap: 50, feeCap: 100, baseFee: nil}, // nil base fee + } + + for i, tc := range testCases { + txdata := &DynamicFeeTx{ + ChainID: big.NewInt(1), + Nonce: 0, + GasTipCap: big.NewInt(tc.tipCap), + GasFeeCap: big.NewInt(tc.feeCap), + Gas: 21000, + To: &common.Address{}, + Value: big.NewInt(0), + Data: nil, + } + tx, _ := SignNewTx(key, signer, txdata) + + var baseFee *big.Int + if tc.baseFee != nil { + baseFee = big.NewInt(*tc.baseFee) + } + + // Get result from original method + orig, origErr := tx.EffectiveGasTip(baseFee) + + // Get result from new method + dst := new(big.Int) + newErr := tx.calcEffectiveGasTip(dst, baseFee) + + // Compare results + if (origErr != nil) != (newErr != nil) { + t.Fatalf("case %d: error mismatch: orig %v, new %v", i, origErr, newErr) + } + + if orig.Cmp(dst) != 0 { + t.Fatalf("case %d: result mismatch: orig %v, new %v", i, orig, dst) + } + } +} + +// Helper function to create integer pointer +func intPtr(i int64) *int64 { + return &i +} diff --git a/core/vm/evm.go b/core/vm/evm.go index c28dcb2554..ecb0f118ec 100644 --- a/core/vm/evm.go +++ b/core/vm/evm.go @@ -555,7 +555,8 @@ func (evm *EVM) Create(caller common.Address, code []byte, gas uint64, value *ui // The different between Create2 with Create is Create2 uses keccak256(0xff ++ msg.sender ++ salt ++ keccak256(init_code))[12:] // instead of the usual sender-and-nonce-hash as the address where the contract is initialized at. func (evm *EVM) Create2(caller common.Address, code []byte, gas uint64, endowment *uint256.Int, salt *uint256.Int) (ret []byte, contractAddr common.Address, leftOverGas uint64, err error) { - contractAddr = crypto.CreateAddress2(caller, salt.Bytes32(), crypto.Keccak256(code)) + inithash := crypto.HashData(evm.interpreter.hasher, code) + contractAddr = crypto.CreateAddress2(caller, salt.Bytes32(), inithash[:]) return evm.create(caller, code, gas, endowment, contractAddr, CREATE2) } diff --git a/core/vm/instructions.go b/core/vm/instructions.go index 0b3b1d1569..63bb6d2d51 100644 --- a/core/vm/instructions.go +++ b/core/vm/instructions.go @@ -22,7 +22,6 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/tracing" "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/params" "github.com/holiman/uint256" ) @@ -234,11 +233,7 @@ func opKeccak256(pc *uint64, interpreter *EVMInterpreter, scope *ScopeContext) ( offset, size := scope.Stack.pop(), scope.Stack.peek() data := scope.Memory.GetPtr(offset.Uint64(), size.Uint64()) - if interpreter.hasher == nil { - interpreter.hasher = crypto.NewKeccakState() - } else { - interpreter.hasher.Reset() - } + interpreter.hasher.Reset() interpreter.hasher.Write(data) interpreter.hasher.Read(interpreter.hasherBuf[:]) diff --git a/core/vm/interpreter.go b/core/vm/interpreter.go index a0038d1aa8..a62c3c843d 100644 --- a/core/vm/interpreter.go +++ b/core/vm/interpreter.go @@ -150,7 +150,7 @@ func NewEVMInterpreter(evm *EVM) *EVMInterpreter { } } evm.Config.ExtraEips = extraEips - return &EVMInterpreter{evm: evm, table: table} + return &EVMInterpreter{evm: evm, table: table, hasher: crypto.NewKeccakState()} } // Run loops and evaluates the contract's code with the given input data and returns diff --git a/crypto/crypto.go b/crypto/crypto.go index 13e9b134f0..09596c05ce 100644 --- a/crypto/crypto.go +++ b/crypto/crypto.go @@ -28,6 +28,7 @@ import ( "io" "math/big" "os" + "sync" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/math" @@ -73,6 +74,12 @@ func NewKeccakState() KeccakState { return sha3.NewLegacyKeccak256().(KeccakState) } +var hasherPool = sync.Pool{ + New: func() any { + return sha3.NewLegacyKeccak256().(KeccakState) + }, +} + // HashData hashes the provided data using the KeccakState and returns a 32 byte hash func HashData(kh KeccakState, data []byte) (h common.Hash) { kh.Reset() @@ -84,22 +91,26 @@ func HashData(kh KeccakState, data []byte) (h common.Hash) { // Keccak256 calculates and returns the Keccak256 hash of the input data. func Keccak256(data ...[]byte) []byte { b := make([]byte, 32) - d := NewKeccakState() + d := hasherPool.Get().(KeccakState) + d.Reset() for _, b := range data { d.Write(b) } d.Read(b) + hasherPool.Put(d) return b } // Keccak256Hash calculates and returns the Keccak256 hash of the input data, // converting it to an internal Hash data structure. func Keccak256Hash(data ...[]byte) (h common.Hash) { - d := NewKeccakState() + d := hasherPool.Get().(KeccakState) + d.Reset() for _, b := range data { d.Write(b) } d.Read(h[:]) + hasherPool.Put(d) return h } diff --git a/eth/api_backend_test.go b/eth/api_backend_test.go index 049f68d827..dfca24aba7 100644 --- a/eth/api_backend_test.go +++ b/eth/api_backend_test.go @@ -134,13 +134,17 @@ func TestSendTx(t *testing.T) { func testSendTx(t *testing.T, withLocal bool) { b := initBackend(withLocal) - txA := pricedSetCodeTx(0, 250000, uint256.NewInt(params.GWei), uint256.NewInt(params.GWei), key, []unsignedAuth{ - { - nonce: 0, - key: key, - }, - }) - b.SendTx(context.Background(), txA) + txA := pricedSetCodeTx(0, 250000, uint256.NewInt(params.GWei), uint256.NewInt(params.GWei), key, []unsignedAuth{{nonce: 0, key: key}}) + if err := b.SendTx(context.Background(), txA); err != nil { + t.Fatalf("Failed to submit tx: %v", err) + } + for { + pending, _ := b.TxPool().ContentFrom(address) + if len(pending) == 1 { + break + } + time.Sleep(100 * time.Millisecond) + } txB := makeTx(1, nil, nil, key) err := b.SendTx(context.Background(), txB) diff --git a/eth/tracers/js/goja.go b/eth/tracers/js/goja.go index 227ea57226..d1e65bf7f4 100644 --- a/eth/tracers/js/goja.go +++ b/eth/tracers/js/goja.go @@ -260,7 +260,8 @@ func (t *jsTracer) OnTxStart(env *tracing.VMContext, tx *types.Transaction, from t.activePrecompiles = vm.ActivePrecompiles(rules) t.ctx["block"] = t.vm.ToValue(t.env.BlockNumber.Uint64()) t.ctx["gas"] = t.vm.ToValue(tx.Gas()) - gasPriceBig, err := t.toBig(t.vm, tx.EffectiveGasTipValue(env.BaseFee).String()) + gasTip, _ := tx.EffectiveGasTip(env.BaseFee) + gasPriceBig, err := t.toBig(t.vm, gasTip.String()) if err != nil { t.err = err return diff --git a/ethdb/database.go b/ethdb/database.go index f2d458b85f..7f421752c4 100644 --- a/ethdb/database.go +++ b/ethdb/database.go @@ -57,6 +57,13 @@ type KeyValueStater interface { Stat() (string, error) } +// KeyValueSyncer wraps the SyncKeyValue method of a backing data store. +type KeyValueSyncer interface { + // SyncKeyValue ensures that all pending writes are flushed to disk, + // guaranteeing data durability up to the point. + SyncKeyValue() error +} + // Compacter wraps the Compact method of a backing data store. type Compacter interface { // Compact flattens the underlying data store for the given key range. In essence, @@ -75,6 +82,7 @@ type KeyValueStore interface { KeyValueReader KeyValueWriter KeyValueStater + KeyValueSyncer KeyValueRangeDeleter Batcher Iteratee @@ -126,6 +134,9 @@ type AncientWriter interface { // The integer return value is the total size of the written data. ModifyAncients(func(AncientWriteOp) error) (int64, error) + // SyncAncient flushes all in-memory ancient store data to disk. + SyncAncient() error + // TruncateHead discards all but the first n ancient data from the ancient store. // After the truncation, the latest item can be accessed it item_n-1(start from 0). TruncateHead(n uint64) (uint64, error) @@ -138,9 +149,6 @@ type AncientWriter interface { // // Note that data marked as non-prunable will still be retained and remain accessible. TruncateTail(n uint64) (uint64, error) - - // Sync flushes all in-memory ancient store data to disk. - Sync() error } // AncientWriteOp is given to the function argument of ModifyAncients. diff --git a/ethdb/leveldb/leveldb.go b/ethdb/leveldb/leveldb.go index ef02e91822..223d01aff6 100644 --- a/ethdb/leveldb/leveldb.go +++ b/ethdb/leveldb/leveldb.go @@ -324,6 +324,22 @@ func (db *Database) Path() string { return db.fn } +// SyncKeyValue flushes all pending writes in the write-ahead-log to disk, +// ensuring data durability up to that point. +func (db *Database) SyncKeyValue() error { + // In theory, the WAL (Write-Ahead Log) can be explicitly synchronized using + // a write operation with SYNC=true. However, there is no dedicated key reserved + // for this purpose, and even a nil key (key=nil) is considered a valid + // database entry. + // + // In LevelDB, writes are blocked until the data is written to the WAL, meaning + // recent writes won't be lost unless a power failure or system crash occurs. + // Additionally, LevelDB is no longer the default database engine and is likely + // only used by hash-mode archive nodes. Given this, the durability guarantees + // without explicit sync are acceptable in the context of LevelDB. + return nil +} + // meter periodically retrieves internal leveldb counters and reports them to // the metrics subsystem. func (db *Database) meter(refresh time.Duration, namespace string) { diff --git a/ethdb/memorydb/memorydb.go b/ethdb/memorydb/memorydb.go index a797275e92..f56727cf4a 100644 --- a/ethdb/memorydb/memorydb.go +++ b/ethdb/memorydb/memorydb.go @@ -199,6 +199,12 @@ func (db *Database) Compact(start []byte, limit []byte) error { return nil } +// SyncKeyValue ensures that all pending writes are flushed to disk, +// guaranteeing data durability up to the point. +func (db *Database) SyncKeyValue() error { + return nil +} + // Len returns the number of entries currently present in the memory database. // // Note, this method is only used for testing (i.e. not public in general) and diff --git a/ethdb/pebble/pebble.go b/ethdb/pebble/pebble.go index 969e67af5a..9ece995655 100644 --- a/ethdb/pebble/pebble.go +++ b/ethdb/pebble/pebble.go @@ -144,7 +144,7 @@ func (l panicLogger) Fatalf(format string, args ...interface{}) { // New returns a wrapped pebble DB object. The namespace is the prefix that the // metrics reporting should use for surfacing internal stats. -func New(file string, cache int, handles int, namespace string, readonly bool, ephemeral bool) (*Database, error) { +func New(file string, cache int, handles int, namespace string, readonly bool) (*Database, error) { // Ensure we have some minimal caching and file guarantees if cache < minCache { cache = minCache @@ -182,10 +182,18 @@ func New(file string, cache int, handles int, namespace string, readonly bool, e memTableSize = maxMemTableSize - 1 } db := &Database{ - fn: file, - log: logger, - quitChan: make(chan chan error), - writeOptions: &pebble.WriteOptions{Sync: !ephemeral}, + fn: file, + log: logger, + quitChan: make(chan chan error), + + // Use asynchronous write mode by default. Otherwise, the overhead of frequent fsync + // operations can be significant, especially on platforms with slow fsync performance + // (e.g., macOS) or less capable SSDs. + // + // Note that enabling async writes means recent data may be lost in the event of an + // application-level panic (writes will also be lost on a machine-level failure, + // of course). Geth is expected to handle recovery from an unclean shutdown. + writeOptions: pebble.NoSync, } opt := &pebble.Options{ // Pebble has a single combined cache area and the write @@ -228,6 +236,15 @@ func New(file string, cache int, handles int, namespace string, readonly bool, e WriteStallEnd: db.onWriteStallEnd, }, Logger: panicLogger{}, // TODO(karalabe): Delete when this is upstreamed in Pebble + + // Pebble is configured to use asynchronous write mode, meaning write operations + // return as soon as the data is cached in memory, without waiting for the WAL + // to be written. This mode offers better write performance but risks losing + // recent writes if the application crashes or a power failure/system crash occurs. + // + // By setting the WALBytesPerSync, the cached WAL writes will be periodically + // flushed at the background if the accumulated size exceeds this threshold. + WALBytesPerSync: 5 * ethdb.IdealBatchSize, } // Disable seek compaction explicitly. Check https://github.com/ethereum/go-ethereum/pull/20130 // for more details. @@ -414,6 +431,18 @@ func (d *Database) Path() string { return d.fn } +// SyncKeyValue flushes all pending writes in the write-ahead-log to disk, +// ensuring data durability up to that point. +func (d *Database) SyncKeyValue() error { + // The entry (value=nil) is not written to the database; it is only + // added to the WAL. Writing this special log entry in sync mode + // automatically flushes all previous writes, ensuring database + // durability up to this point. + b := d.db.NewBatch() + b.LogData(nil, nil) + return d.db.Apply(b, pebble.Sync) +} + // meter periodically retrieves internal pebble counters and reports them to // the metrics subsystem. func (d *Database) meter(refresh time.Duration, namespace string) { diff --git a/ethdb/pebble/pebble_test.go b/ethdb/pebble/pebble_test.go index 3265491d4a..e703a8d0ce 100644 --- a/ethdb/pebble/pebble_test.go +++ b/ethdb/pebble/pebble_test.go @@ -17,6 +17,7 @@ package pebble import ( + "errors" "testing" "github.com/cockroachdb/pebble" @@ -54,3 +55,26 @@ func BenchmarkPebbleDB(b *testing.B) { } }) } + +func TestPebbleLogData(t *testing.T) { + db, err := pebble.Open("", &pebble.Options{ + FS: vfs.NewMem(), + }) + if err != nil { + t.Fatal(err) + } + + _, _, err = db.Get(nil) + if !errors.Is(err, pebble.ErrNotFound) { + t.Fatal("Unknown database entry") + } + + b := db.NewBatch() + b.LogData(nil, nil) + db.Apply(b, pebble.Sync) + + _, _, err = db.Get(nil) + if !errors.Is(err, pebble.ErrNotFound) { + t.Fatal("Unknown database entry") + } +} diff --git a/ethdb/remotedb/remotedb.go b/ethdb/remotedb/remotedb.go index 8a91fdbcf2..a417a25854 100644 --- a/ethdb/remotedb/remotedb.go +++ b/ethdb/remotedb/remotedb.go @@ -110,7 +110,7 @@ func (db *Database) TruncateTail(n uint64) (uint64, error) { panic("not supported") } -func (db *Database) Sync() error { +func (db *Database) SyncAncient() error { return nil } @@ -138,6 +138,10 @@ func (db *Database) Compact(start []byte, limit []byte) error { return nil } +func (db *Database) SyncKeyValue() error { + return nil +} + func (db *Database) Close() error { db.remote.Close() return nil diff --git a/internal/cmdtest/test_cmd.go b/internal/cmdtest/test_cmd.go index 4890d0b7c6..f6f0425598 100644 --- a/internal/cmdtest/test_cmd.go +++ b/internal/cmdtest/test_cmd.go @@ -237,7 +237,7 @@ func (tt *TestCmd) Kill() { } func (tt *TestCmd) withKillTimeout(fn func()) { - timeout := time.AfterFunc(30*time.Second, func() { + timeout := time.AfterFunc(2*time.Minute, func() { tt.Log("killing the child process (timeout)") tt.Kill() }) diff --git a/internal/web3ext/web3ext.go b/internal/web3ext/web3ext.go index aed6bbbdb9..a6d93fc1c5 100644 --- a/internal/web3ext/web3ext.go +++ b/internal/web3ext/web3ext.go @@ -247,11 +247,6 @@ web3._extend({ call: 'debug_vmodule', params: 1 }), - new web3._extend.Method({ - name: 'backtraceAt', - call: 'debug_backtraceAt', - params: 1, - }), new web3._extend.Method({ name: 'stacks', call: 'debug_stacks', diff --git a/node/database.go b/node/database.go index b7d0d856cb..e3ccb91066 100644 --- a/node/database.go +++ b/node/database.go @@ -36,11 +36,6 @@ type openOptions struct { Cache int // the capacity(in megabytes) of the data caching Handles int // number of files to be open simultaneously ReadOnly bool - - // Ephemeral means that filesystem sync operations should be avoided: - // data integrity in the face of a crash is not important. This option - // should typically be used in tests. - Ephemeral bool } // openDatabase opens both a disk-based key-value database such as leveldb or pebble, but also @@ -83,7 +78,7 @@ func openKeyValueDatabase(o openOptions) (ethdb.Database, error) { } if o.Type == rawdb.DBPebble || existingDb == rawdb.DBPebble { log.Info("Using pebble as the backing database") - return newPebbleDBDatabase(o.Directory, o.Cache, o.Handles, o.Namespace, o.ReadOnly, o.Ephemeral) + return newPebbleDBDatabase(o.Directory, o.Cache, o.Handles, o.Namespace, o.ReadOnly) } if o.Type == rawdb.DBLeveldb || existingDb == rawdb.DBLeveldb { log.Info("Using leveldb as the backing database") @@ -91,7 +86,7 @@ func openKeyValueDatabase(o openOptions) (ethdb.Database, error) { } // No pre-existing database, no user-requested one either. Default to Pebble. log.Info("Defaulting to pebble as the backing database") - return newPebbleDBDatabase(o.Directory, o.Cache, o.Handles, o.Namespace, o.ReadOnly, o.Ephemeral) + return newPebbleDBDatabase(o.Directory, o.Cache, o.Handles, o.Namespace, o.ReadOnly) } // newLevelDBDatabase creates a persistent key-value database without a freezer @@ -107,8 +102,8 @@ func newLevelDBDatabase(file string, cache int, handles int, namespace string, r // newPebbleDBDatabase creates a persistent key-value database without a freezer // moving immutable chain segments into cold storage. -func newPebbleDBDatabase(file string, cache int, handles int, namespace string, readonly bool, ephemeral bool) (ethdb.Database, error) { - db, err := pebble.New(file, cache, handles, namespace, readonly, ephemeral) +func newPebbleDBDatabase(file string, cache int, handles int, namespace string, readonly bool) (ethdb.Database, error) { + db, err := pebble.New(file, cache, handles, namespace, readonly) if err != nil { return nil, err } diff --git a/p2p/metrics.go b/p2p/metrics.go index 29c2acb0cb..c44a878aa4 100644 --- a/p2p/metrics.go +++ b/p2p/metrics.go @@ -66,6 +66,18 @@ var ( // capture the rest of errors that are not handled by the above meters dialOtherError = metrics.NewRegisteredMeter("p2p/dials/error/other", nil) + + // handshake error meters for inbound connections + serveTooManyPeers = metrics.NewRegisteredMeter("p2p/serves/error/saturated", nil) + serveAlreadyConnected = metrics.NewRegisteredMeter("p2p/serves/error/known", nil) + serveSelf = metrics.NewRegisteredMeter("p2p/serves/error/self", nil) + serveUselessPeer = metrics.NewRegisteredMeter("p2p/serves/error/useless", nil) + serveUnexpectedIdentity = metrics.NewRegisteredMeter("p2p/serves/error/id/unexpected", nil) + serveEncHandshakeError = metrics.NewRegisteredMeter("p2p/serves/error/rlpx/enc", nil) //EOF; connection reset during handshake; (message too big?) + serveProtoHandshakeError = metrics.NewRegisteredMeter("p2p/serves/error/rlpx/proto", nil) + + // capture the rest of errors that are not handled by the above meters + serveOtherError = metrics.NewRegisteredMeter("p2p/serves/error/other", nil) ) // markDialError matches errors that occur while setting up a dial connection to the @@ -99,6 +111,36 @@ func markDialError(err error) { } } +// markServeError matches errors that occur while serving an inbound connection +// to the corresponding meter. +func markServeError(err error) { + if !metrics.Enabled() { + return + } + + var reason DiscReason + var handshakeErr *protoHandshakeError + d := errors.As(err, &reason) + switch { + case d && reason == DiscTooManyPeers: + serveTooManyPeers.Mark(1) + case d && reason == DiscAlreadyConnected: + serveAlreadyConnected.Mark(1) + case d && reason == DiscSelf: + serveSelf.Mark(1) + case d && reason == DiscUselessPeer: + serveUselessPeer.Mark(1) + case d && reason == DiscUnexpectedIdentity: + serveUnexpectedIdentity.Mark(1) + case errors.As(err, &handshakeErr): + serveProtoHandshakeError.Mark(1) + case errors.Is(err, errEncHandshakeError): + serveEncHandshakeError.Mark(1) + default: + serveOtherError.Mark(1) + } +} + // meteredConn is a wrapper around a net.Conn that meters both the // inbound and outbound network traffic. type meteredConn struct { diff --git a/p2p/server.go b/p2p/server.go index d9105976dd..f3a58bba29 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -864,6 +864,8 @@ func (srv *Server) SetupConn(fd net.Conn, flags connFlag, dialDest *enode.Node) if err != nil { if !c.is(inboundConn) { markDialError(err) + } else { + markServeError(err) } c.close(err) } diff --git a/trie/hasher.go b/trie/hasher.go index 614640ae3a..393cb0bd4d 100644 --- a/trie/hasher.go +++ b/trie/hasher.go @@ -34,7 +34,7 @@ type hasher struct { // hasherPool holds pureHashers var hasherPool = sync.Pool{ - New: func() interface{} { + New: func() any { return &hasher{ tmp: make([]byte, 0, 550), // cap is as large as a full fullNode. sha: crypto.NewKeccakState(), diff --git a/trie/sync.go b/trie/sync.go index 3b7caae5b1..8d0ce6901c 100644 --- a/trie/sync.go +++ b/trie/sync.go @@ -729,9 +729,7 @@ func (s *Sync) hasNode(owner common.Hash, path []byte, hash common.Hash) (exists } else { blob = rawdb.ReadStorageTrieNode(s.database, owner, path) } - h := newBlobHasher() - defer h.release() - exists = hash == h.hash(blob) + exists = hash == crypto.Keccak256Hash(blob) inconsistent = !exists && len(blob) != 0 return exists, inconsistent } @@ -746,23 +744,3 @@ func ResolvePath(path []byte) (common.Hash, []byte) { } return owner, path } - -// blobHasher is used to compute the sha256 hash of the provided data. -type blobHasher struct{ state crypto.KeccakState } - -// blobHasherPool is the pool for reusing pre-allocated hash state. -var blobHasherPool = sync.Pool{ - New: func() interface{} { return &blobHasher{state: crypto.NewKeccakState()} }, -} - -func newBlobHasher() *blobHasher { - return blobHasherPool.Get().(*blobHasher) -} - -func (h *blobHasher) hash(data []byte) common.Hash { - return crypto.HashData(h.state, data) -} - -func (h *blobHasher) release() { - blobHasherPool.Put(h) -} diff --git a/trie/trie_test.go b/trie/trie_test.go index 54d1b083d8..91fde6dbf2 100644 --- a/trie/trie_test.go +++ b/trie/trie_test.go @@ -830,6 +830,7 @@ func (s *spongeDb) NewBatch() ethdb.Batch { return &spongeBat func (s *spongeDb) NewBatchWithSize(size int) ethdb.Batch { return &spongeBatch{s} } func (s *spongeDb) Stat() (string, error) { panic("implement me") } func (s *spongeDb) Compact(start []byte, limit []byte) error { panic("implement me") } +func (s *spongeDb) SyncKeyValue() error { return nil } func (s *spongeDb) Close() error { return nil } func (s *spongeDb) Put(key []byte, value []byte) error { var ( diff --git a/triedb/pathdb/buffer.go b/triedb/pathdb/buffer.go index dea8875bda..c4e081b973 100644 --- a/triedb/pathdb/buffer.go +++ b/triedb/pathdb/buffer.go @@ -135,10 +135,13 @@ func (b *buffer) flush(db ethdb.KeyValueStore, freezer ethdb.AncientWriter, node start = time.Now() batch = db.NewBatchWithSize(b.nodes.dbsize() * 11 / 10) // extra 10% for potential pebble internal stuff ) - // Explicitly sync the state freezer, ensuring that all written - // data is transferred to disk before updating the key-value store. + // Explicitly sync the state freezer to ensure all written data is persisted to disk + // before updating the key-value store. + // + // This step is crucial to guarantee that the corresponding state history remains + // available for state rollback. if freezer != nil { - if err := freezer.Sync(); err != nil { + if err := freezer.SyncAncient(); err != nil { return err } } diff --git a/triedb/pathdb/database.go b/triedb/pathdb/database.go index d48850c102..155e28543d 100644 --- a/triedb/pathdb/database.go +++ b/triedb/pathdb/database.go @@ -454,6 +454,15 @@ func (db *Database) Recover(root common.Hash) error { db.tree.reset(dl) } rawdb.DeleteTrieJournal(db.diskdb) + + // Explicitly sync the key-value store to ensure all recent writes are + // flushed to disk. This step is crucial to prevent a scenario where + // recent key-value writes are lost due to an application panic, while + // the associated state histories have already been removed, resulting + // in the inability to perform a state rollback. + if err := db.diskdb.SyncKeyValue(); err != nil { + return err + } _, err := truncateFromHead(db.diskdb, db.freezer, dl.stateID()) if err != nil { return err diff --git a/triedb/pathdb/disklayer.go b/triedb/pathdb/disklayer.go index 184f6430a2..b8869888d9 100644 --- a/triedb/pathdb/disklayer.go +++ b/triedb/pathdb/disklayer.go @@ -115,15 +115,12 @@ func (dl *diskLayer) node(owner common.Hash, path []byte, depth int) ([]byte, co dirtyNodeMissMeter.Mark(1) // Try to retrieve the trie node from the clean memory cache - h := newHasher() - defer h.release() - key := nodeCacheKey(owner, path) if dl.nodes != nil { if blob := dl.nodes.Get(nil, key); len(blob) > 0 { cleanNodeHitMeter.Mark(1) cleanNodeReadMeter.Mark(int64(len(blob))) - return blob, h.hash(blob), &nodeLoc{loc: locCleanCache, depth: depth}, nil + return blob, crypto.Keccak256Hash(blob), &nodeLoc{loc: locCleanCache, depth: depth}, nil } cleanNodeMissMeter.Mark(1) } @@ -138,7 +135,7 @@ func (dl *diskLayer) node(owner common.Hash, path []byte, depth int) ([]byte, co dl.nodes.Set(key, blob) cleanNodeWriteMeter.Mark(int64(len(blob))) } - return blob, h.hash(blob), &nodeLoc{loc: locDiskLayer, depth: depth}, nil + return blob, crypto.Keccak256Hash(blob), &nodeLoc{loc: locDiskLayer, depth: depth}, nil } // account directly retrieves the account RLP associated with a particular @@ -231,6 +228,8 @@ func (dl *diskLayer) commit(bottom *diffLayer, force bool) (*diskLayer, error) { oldest uint64 ) if dl.db.freezer != nil { + // Bail out with an error if writing the state history fails. + // This can happen, for example, if the device is full. err := writeHistory(dl.db.freezer, bottom) if err != nil { return nil, err @@ -357,22 +356,3 @@ func (dl *diskLayer) resetCache() { dl.nodes.Reset() } } - -// hasher is used to compute the sha256 hash of the provided data. -type hasher struct{ sha crypto.KeccakState } - -var hasherPool = sync.Pool{ - New: func() interface{} { return &hasher{sha: crypto.NewKeccakState()} }, -} - -func newHasher() *hasher { - return hasherPool.Get().(*hasher) -} - -func (h *hasher) hash(data []byte) common.Hash { - return crypto.HashData(h.sha, data) -} - -func (h *hasher) release() { - hasherPool.Put(h) -} diff --git a/triedb/pathdb/execute.go b/triedb/pathdb/execute.go index 2400f280a3..db1e679277 100644 --- a/triedb/pathdb/execute.go +++ b/triedb/pathdb/execute.go @@ -22,6 +22,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/trie" "github.com/ethereum/go-ethereum/trie/trienode" @@ -85,10 +86,9 @@ func apply(db database.NodeDatabase, prevRoot common.Hash, postRoot common.Hash, func updateAccount(ctx *context, db database.NodeDatabase, addr common.Address) error { // The account was present in prev-state, decode it from the // 'slim-rlp' format bytes. - h := newHasher() - defer h.release() + h := crypto.NewKeccakState() - addrHash := h.hash(addr.Bytes()) + addrHash := crypto.HashData(h, addr.Bytes()) prev, err := types.FullAccount(ctx.accounts[addr]) if err != nil { return err @@ -113,7 +113,7 @@ func updateAccount(ctx *context, db database.NodeDatabase, addr common.Address) for key, val := range ctx.storages[addr] { tkey := key if ctx.rawStorageKey { - tkey = h.hash(key.Bytes()) + tkey = crypto.HashData(h, key.Bytes()) } var err error if len(val) == 0 { @@ -149,10 +149,9 @@ func updateAccount(ctx *context, db database.NodeDatabase, addr common.Address) // account and storage is wiped out correctly. func deleteAccount(ctx *context, db database.NodeDatabase, addr common.Address) error { // The account must be existent in post-state, load the account. - h := newHasher() - defer h.release() + h := crypto.NewKeccakState() - addrHash := h.hash(addr.Bytes()) + addrHash := crypto.HashData(h, addr.Bytes()) blob, err := ctx.accountTrie.Get(addrHash.Bytes()) if err != nil { return err @@ -174,7 +173,7 @@ func deleteAccount(ctx *context, db database.NodeDatabase, addr common.Address) } tkey := key if ctx.rawStorageKey { - tkey = h.hash(key.Bytes()) + tkey = crypto.HashData(h, key.Bytes()) } if err := st.Delete(tkey.Bytes()); err != nil { return err diff --git a/triedb/pathdb/history.go b/triedb/pathdb/history.go index c063e45371..aed0296da5 100644 --- a/triedb/pathdb/history.go +++ b/triedb/pathdb/history.go @@ -542,8 +542,9 @@ func writeHistory(writer ethdb.AncientWriter, dl *diffLayer) error { indexSize := common.StorageSize(len(accountIndex) + len(storageIndex)) // Write history data into five freezer table respectively. - rawdb.WriteStateHistory(writer, dl.stateID(), history.meta.encode(), accountIndex, storageIndex, accountData, storageData) - + if err := rawdb.WriteStateHistory(writer, dl.stateID(), history.meta.encode(), accountIndex, storageIndex, accountData, storageData); err != nil { + return err + } historyDataBytesMeter.Mark(int64(dataSize)) historyIndexBytesMeter.Mark(int64(indexSize)) historyBuildTimeMeter.UpdateSince(start) diff --git a/triedb/pathdb/metrics.go b/triedb/pathdb/metrics.go index 45dad6f1ae..abe2dfe1f6 100644 --- a/triedb/pathdb/metrics.go +++ b/triedb/pathdb/metrics.go @@ -46,7 +46,7 @@ var ( nodeDiskFalseMeter = metrics.NewRegisteredMeter("pathdb/disk/false", nil) nodeDiffFalseMeter = metrics.NewRegisteredMeter("pathdb/diff/false", nil) - commitTimeTimer = metrics.NewRegisteredTimer("pathdb/commit/time", nil) + commitTimeTimer = metrics.NewRegisteredResettingTimer("pathdb/commit/time", nil) commitNodesMeter = metrics.NewRegisteredMeter("pathdb/commit/nodes", nil) commitBytesMeter = metrics.NewRegisteredMeter("pathdb/commit/bytes", nil) @@ -57,7 +57,7 @@ var ( gcStorageMeter = metrics.NewRegisteredMeter("pathdb/gc/storage/count", nil) gcStorageBytesMeter = metrics.NewRegisteredMeter("pathdb/gc/storage/bytes", nil) - historyBuildTimeMeter = metrics.NewRegisteredTimer("pathdb/history/time", nil) + historyBuildTimeMeter = metrics.NewRegisteredResettingTimer("pathdb/history/time", nil) historyDataBytesMeter = metrics.NewRegisteredMeter("pathdb/history/bytes/data", nil) historyIndexBytesMeter = metrics.NewRegisteredMeter("pathdb/history/bytes/index", nil) )