diff --git a/accounts/abi/bind/backends/simulated.go b/accounts/abi/bind/backends/simulated.go index e30572e0a1..ee81676c8d 100644 --- a/accounts/abi/bind/backends/simulated.go +++ b/accounts/abi/bind/backends/simulated.go @@ -70,7 +70,7 @@ type SimulatedBackend struct { func NewSimulatedBackendWithDatabase(database ethdb.Database, alloc core.GenesisAlloc, gasLimit uint64) *SimulatedBackend { genesis := core.Genesis{Config: params.AllEthashProtocolChanges, GasLimit: gasLimit, Alloc: alloc} genesis.MustCommit(database) - blockchain, _ := core.NewBlockChain(database, nil, genesis.Config, ethash.NewFaker(), vm.Config{}, nil) + blockchain, _ := core.NewBlockChain(database, nil, genesis.Config, ethash.NewFaker(), vm.Config{}, nil, 0) backend := &SimulatedBackend{ database: database, diff --git a/cmd/geth/main.go b/cmd/geth/main.go index 019eee605f..2a72f10f85 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -94,6 +94,7 @@ var ( utils.SyncModeFlag, utils.ExitWhenSyncedFlag, utils.GCModeFlag, + utils.TxLookupLimitFlag, utils.LightServeFlag, utils.LightLegacyServFlag, utils.LightIngressFlag, diff --git a/cmd/geth/retesteth.go b/cmd/geth/retesteth.go index bda19324d0..b608596898 100644 --- a/cmd/geth/retesteth.go +++ b/cmd/geth/retesteth.go @@ -398,7 +398,7 @@ func (api *RetestethAPI) SetChainParams(ctx context.Context, chainParams ChainPa } engine := &NoRewardEngine{inner: inner, rewardsOn: chainParams.SealEngine != "NoReward"} - blockchain, err := core.NewBlockChain(ethDb, nil, chainConfig, engine, vm.Config{}, nil) + blockchain, err := core.NewBlockChain(ethDb, nil, chainConfig, engine, vm.Config{}, nil, 0) if err != nil { return false, err } diff --git a/cmd/geth/usage.go b/cmd/geth/usage.go index af195425b1..d7aed0ce8c 100644 --- a/cmd/geth/usage.go +++ b/cmd/geth/usage.go @@ -80,6 +80,7 @@ var AppHelpFlagGroups = []flagGroup{ utils.SyncModeFlag, utils.ExitWhenSyncedFlag, utils.GCModeFlag, + utils.TxLookupLimitFlag, utils.EthStatsURLFlag, utils.IdentityFlag, utils.LightKDFFlag, diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index d553b662c8..45f4501183 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -215,6 +215,11 @@ var ( Usage: `Blockchain garbage collection mode ("full", "archive")`, Value: "full", } + TxLookupLimitFlag = cli.Int64Flag{ + Name: "txlookuplimit", + Usage: "The maximum number of blocks from head whose tx indices are reserved(0 means reserve all indices)", + Value: 0, + } LightKDFFlag = cli.BoolFlag{ Name: "lightkdf", Usage: "Reduce key-derivation RAM & CPU usage at some expense of KDF strength", @@ -1450,7 +1455,9 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *eth.Config) { } cfg.NoPruning = ctx.GlobalString(GCModeFlag.Name) == "archive" cfg.NoPrefetch = ctx.GlobalBool(CacheNoPrefetchFlag.Name) - + if ctx.GlobalIsSet(TxLookupLimitFlag.Name) { + cfg.TxLookupLimit = ctx.GlobalUint64(TxLookupLimitFlag.Name) + } if ctx.GlobalIsSet(CacheFlag.Name) || ctx.GlobalIsSet(CacheTrieFlag.Name) { cfg.TrieCleanCache = ctx.GlobalInt(CacheFlag.Name) * ctx.GlobalInt(CacheTrieFlag.Name) / 100 } @@ -1716,7 +1723,7 @@ func MakeChain(ctx *cli.Context, stack *node.Node) (chain *core.BlockChain, chai cache.TrieDirtyLimit = ctx.GlobalInt(CacheFlag.Name) * ctx.GlobalInt(CacheGCFlag.Name) / 100 } vmcfg := vm.Config{EnablePreimageRecording: ctx.GlobalBool(VMEnableDebugFlag.Name)} - chain, err = core.NewBlockChain(chainDb, cache, config, engine, vmcfg, nil) + chain, err = core.NewBlockChain(chainDb, cache, config, engine, vmcfg, nil, 0) if err != nil { Fatalf("Can't create BlockChain: %v", err) } diff --git a/consensus/clique/clique_test.go b/consensus/clique/clique_test.go index 710f448055..4a746db4bf 100644 --- a/consensus/clique/clique_test.go +++ b/consensus/clique/clique_test.go @@ -54,7 +54,7 @@ func TestReimportMirroredState(t *testing.T) { genesis := genspec.MustCommit(db) // Generate a batch of blocks, each properly signed - chain, _ := core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil) + chain, _ := core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil, 0) defer chain.Stop() blocks, _ := core.GenerateChain(params.AllCliqueProtocolChanges, genesis, engine, db, 3, func(i int, block *core.BlockGen) { @@ -88,7 +88,7 @@ func TestReimportMirroredState(t *testing.T) { db = rawdb.NewMemoryDatabase() genspec.MustCommit(db) - chain, _ = core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil) + chain, _ = core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil, 0) defer chain.Stop() if _, err := chain.InsertChain(blocks[:2]); err != nil { @@ -101,7 +101,7 @@ func TestReimportMirroredState(t *testing.T) { // Simulate a crash by creating a new chain on top of the database, without // flushing the dirty states out. Insert the last block, trigerring a sidechain // reimport. - chain, _ = core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil) + chain, _ = core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil, 0) defer chain.Stop() if _, err := chain.InsertChain(blocks[2:]); err != nil { diff --git a/consensus/clique/snapshot_test.go b/consensus/clique/snapshot_test.go index fc08722efd..52be035310 100644 --- a/consensus/clique/snapshot_test.go +++ b/consensus/clique/snapshot_test.go @@ -448,7 +448,7 @@ func TestClique(t *testing.T) { batches[len(batches)-1] = append(batches[len(batches)-1], block) } // Pass all the headers through clique and ensure tallying succeeds - chain, err := core.NewBlockChain(db, nil, &config, engine, vm.Config{}, nil) + chain, err := core.NewBlockChain(db, nil, &config, engine, vm.Config{}, nil, 0) if err != nil { t.Errorf("test %d: failed to create test chain: %v", i, err) continue diff --git a/core/bench_test.go b/core/bench_test.go index d7a5e11c2f..ae5a59b381 100644 --- a/core/bench_test.go +++ b/core/bench_test.go @@ -175,7 +175,7 @@ func benchInsertChain(b *testing.B, disk bool, gen func(int, *BlockGen)) { // Time the insertion of the new chain. // State and blocks are stored in the same DB. - chainman, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + chainman, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer chainman.Stop() b.ReportAllocs() b.ResetTimer() @@ -287,7 +287,7 @@ func benchReadChain(b *testing.B, full bool, count uint64) { if err != nil { b.Fatalf("error opening database at %v: %v", dir, err) } - chain, err := NewBlockChain(db, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil) + chain, err := NewBlockChain(db, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 0) if err != nil { b.Fatalf("error creating chain: %v", err) } diff --git a/core/block_validator_test.go b/core/block_validator_test.go index 06e2ba1a4f..1307b5d951 100644 --- a/core/block_validator_test.go +++ b/core/block_validator_test.go @@ -42,7 +42,7 @@ func TestHeaderVerification(t *testing.T) { headers[i] = block.Header() } // Run the header checker for blocks one-by-one, checking for both valid and invalid nonces - chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil) + chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 0) defer chain.Stop() for i := 0; i < len(blocks); i++ { @@ -106,11 +106,11 @@ func testHeaderConcurrentVerification(t *testing.T, threads int) { var results <-chan error if valid { - chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil) + chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 0) _, results = chain.engine.VerifyHeaders(chain, headers, seals) chain.Stop() } else { - chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFakeFailer(uint64(len(headers)-1)), vm.Config{}, nil) + chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFakeFailer(uint64(len(headers)-1)), vm.Config{}, nil, 0) _, results = chain.engine.VerifyHeaders(chain, headers, seals) chain.Stop() } @@ -173,7 +173,7 @@ func testHeaderConcurrentAbortion(t *testing.T, threads int) { defer runtime.GOMAXPROCS(old) // Start the verifications and immediately abort - chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFakeDelayer(time.Millisecond), vm.Config{}, nil) + chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFakeDelayer(time.Millisecond), vm.Config{}, nil, 0) defer chain.Stop() abort, results := chain.engine.VerifyHeaders(chain, headers, seals) diff --git a/core/blockchain.go b/core/blockchain.go index 8d8476df38..dbd472ea07 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -134,9 +134,10 @@ type BlockChain struct { chainConfig *params.ChainConfig // Chain & network configuration cacheConfig *CacheConfig // Cache configuration for pruning - db ethdb.Database // Low level persistent database to store final content in - triegc *prque.Prque // Priority queue mapping block numbers to tries to gc - gcproc time.Duration // Accumulates canonical block processing for trie dumping + db ethdb.Database // Low level persistent database to store final content in + triegc *prque.Prque // Priority queue mapping block numbers to tries to gc + gcproc time.Duration // Accumulates canonical block processing for trie dumping + txLookupLimit uint64 // The maximum number of blocks from head whose tx indices are reserved hc *HeaderChain rmLogsFeed event.Feed @@ -181,7 +182,7 @@ type BlockChain struct { // NewBlockChain returns a fully initialised block chain using information // available in the database. It initialises the default Ethereum Validator and // Processor. -func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config, shouldPreserve func(block *types.Block) bool) (*BlockChain, error) { +func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config, shouldPreserve func(block *types.Block) bool, txLookupLimit uint64) (*BlockChain, error) { if cacheConfig == nil { cacheConfig = &CacheConfig{ TrieCleanLimit: 256, @@ -200,6 +201,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par bc := &BlockChain{ chainConfig: chainConfig, cacheConfig: cacheConfig, + txLookupLimit: txLookupLimit, db: db, triegc: prque.New(nil), stateCache: state.NewDatabaseWithCache(db, cacheConfig.TrieCleanLimit), @@ -231,12 +233,6 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par // Initialize the chain with ancient data if it isn't empty. if bc.empty() { rawdb.InitBlockIndexFromFreezer(bc.db) - rawdb.WriteAncientTxLookupProgress(bc.db, 0) // Explicitly mark the missing of txlookup. - } - // Re-initialise all ancient txlookup indexes in the background. - if number := rawdb.ReadAncientTxLookupProgress(bc.db); number != nil { - // Genesis block doesn't have transaction, just ignore it. - go rawdb.InitTxsLookupFromFreezer(bc.db, *number+1) } if err := bc.loadLastState(); err != nil { return nil, err @@ -294,6 +290,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par } // Take ownership of this particular state go bc.update() + go bc.updateTxIndices() return bc, nil } @@ -2041,6 +2038,101 @@ func (bc *BlockChain) update() { } } +// updateTxIndices is responsible for the construction and deletion of the +// transaction index. +// +// User can use flag `txlookuplimit` to specify a "recentness" block, below +// which ancient tx indices get deleted. If `txlookuplimit` is 0, it means +// all tx indices will be reserved. +// +// The user can adjust the txlookuplimit value for each launch, Geth will +// automatically construct the missing indices and delete the extra indices. +func (bc *BlockChain) updateTxIndices() { + var ( + done chan struct{} // Non-nil if background unindexing or reindexing routine is active. + headCh = make(chan ChainHeadEvent) + ) + sub := bc.SubscribeChainHeadEvent(headCh) + defer func() { + if sub != nil { + sub.Unsubscribe() + } + }() + // initialiseIndices inits txlookup indices into the database. + // If there already exists some indices, this function will find + // the oldest block which has been indexed and start indexing from + // this point. + initialiseIndices := func(head uint64) { + defer func() { done <- struct{}{} }() + + from, to := uint64(0), head + if bc.txLookupLimit != 0 && head > bc.txLookupLimit { + from = head - bc.txLookupLimit + } + // Find oldest indexed block via binary search when we don't + // have this flag in database. + start := time.Now() + oldest := rawdb.FindOldestIndexedBlock(bc.db, from, to) + log.Debug("Find oldest indexed block", "oldest", oldest, "elapsed", common.PrettyDuration(time.Since(start))) + + // Re-construct missing tx indices. + if oldest == nil { + rawdb.IndexTxLookup(bc.db, from, to) // No block has been indexed. + } else { + rawdb.IndexTxLookup(bc.db, from, *oldest) + } + // Delete useless tx indices if user requires. + if from > 0 { + rawdb.RemoveTxsLookup(bc.db, 0, from) + } + log.Debug("Initialised transaction indices", "elapsed", common.PrettyDuration(time.Since(start))) + } + // indexBlocks reindex or unindex transaction indices depends + // on user's requirement. + indexBlocks := func(oldest uint64, head uint64) { + defer func() { done <- struct{}{} }() + + // All indices should be reserved. + if bc.txLookupLimit == 0 || head <= bc.txLookupLimit { + if oldest == 0 { + // Short circuit if nothing to delete. + return + } else { + // Reindex all indices if necessary + rawdb.IndexTxLookup(bc.db, 0, oldest) + return + } + } + if head-bc.txLookupLimit < oldest { + // Reindex a part of missing indices and rewind oldest indexed + // point to HEAD-limit + rawdb.IndexTxLookup(bc.db, head-bc.txLookupLimit, oldest) + } else { + // Unindex a part of stale indices and forward oldest indexed + // point to HEAD-limit + rawdb.RemoveTxsLookup(bc.db, oldest, head-bc.txLookupLimit) + } + } + + for { + select { + case head := <-headCh: + if done == nil { + done = make(chan struct{}) + if number := rawdb.ReadOldestIndexedBlock(bc.db); number == nil { + go initialiseIndices(head.Block.NumberU64()) + } else { + go indexBlocks(*number, head.Block.NumberU64()) + } + } + case <-done: + done = nil + case <-bc.quit: + return + } + } +} + // BadBlocks returns a list of the last 'bad blocks' that the client has seen on the network func (bc *BlockChain) BadBlocks() []*types.Block { blocks := make([]*types.Block, 0, bc.badBlocks.Len()) diff --git a/core/blockchain_test.go b/core/blockchain_test.go index db624c4dc0..161b3a015c 100644 --- a/core/blockchain_test.go +++ b/core/blockchain_test.go @@ -54,7 +54,7 @@ func newCanonical(engine consensus.Engine, n int, full bool) (ethdb.Database, *B ) // Initialize a fresh chain with only a genesis block - blockchain, _ := NewBlockChain(db, nil, params.AllEthashProtocolChanges, engine, vm.Config{}, nil) + blockchain, _ := NewBlockChain(db, nil, params.AllEthashProtocolChanges, engine, vm.Config{}, nil, 0) // Create and inject the requested chain if n == 0 { return db, blockchain, nil @@ -509,7 +509,7 @@ func testReorgBadHashes(t *testing.T, full bool) { blockchain.Stop() // Create a new BlockChain and check that it rolled back the state. - ncm, err := NewBlockChain(blockchain.db, nil, blockchain.chainConfig, ethash.NewFaker(), vm.Config{}, nil) + ncm, err := NewBlockChain(blockchain.db, nil, blockchain.chainConfig, ethash.NewFaker(), vm.Config{}, nil, 0) if err != nil { t.Fatalf("failed to create new chain manager: %v", err) } @@ -621,7 +621,7 @@ func TestFastVsFullChains(t *testing.T) { // Import the chain as an archive node for the comparison baseline archiveDb := rawdb.NewMemoryDatabase() gspec.MustCommit(archiveDb) - archive, _ := NewBlockChain(archiveDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + archive, _ := NewBlockChain(archiveDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer archive.Stop() if n, err := archive.InsertChain(blocks); err != nil { @@ -630,7 +630,7 @@ func TestFastVsFullChains(t *testing.T) { // Fast import the chain as a non-archive node to test fastDb := rawdb.NewMemoryDatabase() gspec.MustCommit(fastDb) - fast, _ := NewBlockChain(fastDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + fast, _ := NewBlockChain(fastDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer fast.Stop() headers := make([]*types.Header, len(blocks)) @@ -654,7 +654,7 @@ func TestFastVsFullChains(t *testing.T) { t.Fatalf("failed to create temp freezer db: %v", err) } gspec.MustCommit(ancientDb) - ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer ancient.Stop() if n, err := ancient.InsertHeaderChain(headers, 1); err != nil { @@ -750,7 +750,7 @@ func TestLightVsFastVsFullChainHeads(t *testing.T) { // Import the chain as an archive node and ensure all pointers are updated archiveDb, delfn := makeDb() defer delfn() - archive, _ := NewBlockChain(archiveDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + archive, _ := NewBlockChain(archiveDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) if n, err := archive.InsertChain(blocks); err != nil { t.Fatalf("failed to process block %d: %v", n, err) } @@ -763,7 +763,7 @@ func TestLightVsFastVsFullChainHeads(t *testing.T) { // Import the chain as a non-archive node and ensure all pointers are updated fastDb, delfn := makeDb() defer delfn() - fast, _ := NewBlockChain(fastDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + fast, _ := NewBlockChain(fastDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer fast.Stop() headers := make([]*types.Header, len(blocks)) @@ -783,7 +783,7 @@ func TestLightVsFastVsFullChainHeads(t *testing.T) { // Import the chain as a ancient-first node and ensure all pointers are updated ancientDb, delfn := makeDb() defer delfn() - ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer ancient.Stop() if n, err := ancient.InsertHeaderChain(headers, 1); err != nil { @@ -802,7 +802,7 @@ func TestLightVsFastVsFullChainHeads(t *testing.T) { // Import the chain as a light node and ensure all pointers are updated lightDb, delfn := makeDb() defer delfn() - light, _ := NewBlockChain(lightDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + light, _ := NewBlockChain(lightDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) if n, err := light.InsertHeaderChain(headers, 1); err != nil { t.Fatalf("failed to insert header %d: %v", n, err) } @@ -871,7 +871,7 @@ func TestChainTxReorgs(t *testing.T) { } }) // Import the chain. This runs all block validation rules. - blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) if i, err := blockchain.InsertChain(chain); err != nil { t.Fatalf("failed to insert original chain[%d]: %v", i, err) } @@ -941,7 +941,7 @@ func TestLogReorgs(t *testing.T) { signer = types.NewEIP155Signer(gspec.Config.ChainID) ) - blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer blockchain.Stop() rmLogsCh := make(chan RemovedLogsEvent) @@ -1018,7 +1018,7 @@ func TestLogRebirth(t *testing.T) { } } - blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer blockchain.Stop() logsCh := make(chan []*types.Log) @@ -1140,7 +1140,7 @@ func TestSideLogRebirth(t *testing.T) { } } - blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer blockchain.Stop() logsCh := make(chan []*types.Log) @@ -1195,7 +1195,7 @@ func TestReorgSideEvent(t *testing.T) { signer = types.NewEIP155Signer(gspec.Config.ChainID) ) - blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer blockchain.Stop() chain, _ := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, 3, func(i int, gen *BlockGen) {}) @@ -1324,7 +1324,7 @@ func TestEIP155Transition(t *testing.T) { genesis = gspec.MustCommit(db) ) - blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer blockchain.Stop() blocks, _ := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, 4, func(i int, block *BlockGen) { @@ -1431,7 +1431,7 @@ func TestEIP161AccountRemoval(t *testing.T) { } genesis = gspec.MustCommit(db) ) - blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer blockchain.Stop() blocks, _ := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, 3, func(i int, block *BlockGen) { @@ -1506,7 +1506,7 @@ func TestBlockchainHeaderchainReorgConsistency(t *testing.T) { diskdb := rawdb.NewMemoryDatabase() new(Genesis).MustCommit(diskdb) - chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil) + chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0) if err != nil { t.Fatalf("failed to create tester chain: %v", err) } @@ -1550,7 +1550,7 @@ func TestTrieForkGC(t *testing.T) { diskdb := rawdb.NewMemoryDatabase() new(Genesis).MustCommit(diskdb) - chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil) + chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0) if err != nil { t.Fatalf("failed to create tester chain: %v", err) } @@ -1589,7 +1589,7 @@ func TestLargeReorgTrieGC(t *testing.T) { diskdb := rawdb.NewMemoryDatabase() new(Genesis).MustCommit(diskdb) - chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil) + chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0) if err != nil { t.Fatalf("failed to create tester chain: %v", err) } @@ -1649,7 +1649,7 @@ func TestBlockchainRecovery(t *testing.T) { t.Fatalf("failed to create temp freezer db: %v", err) } gspec.MustCommit(ancientDb) - ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) headers := make([]*types.Header, len(blocks)) for i, block := range blocks { @@ -1668,7 +1668,7 @@ func TestBlockchainRecovery(t *testing.T) { rawdb.WriteHeadFastBlockHash(ancientDb, midBlock.Hash()) // Reopen broken blockchain again - ancient, _ = NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + ancient, _ = NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer ancient.Stop() if num := ancient.CurrentBlock().NumberU64(); num != 0 { t.Errorf("head block mismatch: have #%v, want #%v", num, 0) @@ -1705,7 +1705,7 @@ func TestIncompleteAncientReceiptChainInsertion(t *testing.T) { t.Fatalf("failed to create temp freezer db: %v", err) } gspec.MustCommit(ancientDb) - ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer ancient.Stop() headers := make([]*types.Header, len(blocks)) @@ -1762,7 +1762,7 @@ func TestLowDiffLongChain(t *testing.T) { diskdb := rawdb.NewMemoryDatabase() new(Genesis).MustCommit(diskdb) - chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil) + chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0) if err != nil { t.Fatalf("failed to create tester chain: %v", err) } @@ -1809,7 +1809,7 @@ func testSideImport(t *testing.T, numCanonBlocksInSidechain, blocksBetweenCommon blocks, _ := GenerateChain(params.TestChainConfig, genesis, engine, db, 2*TriesInMemory, nil) diskdb := rawdb.NewMemoryDatabase() new(Genesis).MustCommit(diskdb) - chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil) + chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0) if err != nil { t.Fatalf("failed to create tester chain: %v", err) } @@ -1906,7 +1906,7 @@ func testInsertKnownChainData(t *testing.T, typ string) { new(Genesis).MustCommit(chaindb) defer os.RemoveAll(dir) - chain, err := NewBlockChain(chaindb, nil, params.TestChainConfig, engine, vm.Config{}, nil) + chain, err := NewBlockChain(chaindb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0) if err != nil { t.Fatalf("failed to create tester chain: %v", err) } @@ -2029,7 +2029,7 @@ func getLongAndShortChains() (*BlockChain, []*types.Block, []*types.Block, error diskdb := rawdb.NewMemoryDatabase() new(Genesis).MustCommit(diskdb) - chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil) + chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0) if err != nil { return nil, nil, nil, fmt.Errorf("failed to create tester chain: %v", err) } @@ -2133,6 +2133,140 @@ func TestReorgToShorterRemovesCanonMappingHeaderChain(t *testing.T) { } } +func TestTransactionIndices(t *testing.T) { + // Configure and generate a sample block chain + var ( + gendb = rawdb.NewMemoryDatabase() + key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + address = crypto.PubkeyToAddress(key.PublicKey) + funds = big.NewInt(1000000000) + gspec = &Genesis{Config: params.TestChainConfig, Alloc: GenesisAlloc{address: {Balance: funds}}} + genesis = gspec.MustCommit(gendb) + signer = types.NewEIP155Signer(gspec.Config.ChainID) + ) + height := uint64(128) + blocks, receipts := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), gendb, int(height), func(i int, block *BlockGen) { + tx, err := types.SignTx(types.NewTransaction(block.TxNonce(address), common.Address{0x00}, big.NewInt(1000), params.TxGas, nil, nil), signer, key) + if err != nil { + panic(err) + } + block.AddTx(tx) + }) + blocks2, _ := GenerateChain(gspec.Config, blocks[len(blocks)-1], ethash.NewFaker(), gendb, 10, nil) + + check := func(oldest *uint64, chain *BlockChain) { + indexed := rawdb.ReadOldestIndexedBlock(chain.db) + if oldest == nil && indexed != nil { + t.Fatalf("Oldest indexded block mismatch, want nil, have %d", *indexed) + } + if oldest != nil && *indexed != *oldest { + t.Fatalf("Oldest indexded block mismatch, want %d, have %d", *oldest, *indexed) + } + if oldest != nil { + for i := *oldest; i <= chain.CurrentBlock().NumberU64(); i++ { + block := rawdb.ReadBlock(chain.db, rawdb.ReadCanonicalHash(chain.db, i), i) + if block.Transactions().Len() == 0 { + continue + } + for _, tx := range block.Transactions() { + if index := rawdb.ReadTxLookupEntry(chain.db, tx.Hash()); index == nil { + t.Fatalf("Miss transaction indice, number %d hash %s", i, tx.Hash().Hex()) + } + } + } + for i := uint64(0); i < *oldest; i++ { + block := rawdb.ReadBlock(chain.db, rawdb.ReadCanonicalHash(chain.db, i), i) + if block.Transactions().Len() == 0 { + continue + } + for _, tx := range block.Transactions() { + if index := rawdb.ReadTxLookupEntry(chain.db, tx.Hash()); index != nil { + t.Fatalf("Transaction indice should be deleted, number %d hash %s", i, tx.Hash().Hex()) + } + } + } + } + } + // Freezer style fast import the chain. + frdir, err := ioutil.TempDir("", "") + if err != nil { + t.Fatalf("failed to create temp freezer dir: %v", err) + } + defer os.Remove(frdir) + ancientDb, err := rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), frdir, "") + if err != nil { + t.Fatalf("failed to create temp freezer db: %v", err) + } + gspec.MustCommit(ancientDb) + + // Import all blocks into ancient db + chain, err := NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 0) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + headers := make([]*types.Header, len(blocks)) + for i, block := range blocks { + headers[i] = block.Header() + } + if n, err := chain.InsertHeaderChain(headers, 0); err != nil { + t.Fatalf("failed to insert header %d: %v", n, err) + } + if n, err := chain.InsertReceiptChain(blocks, receipts, 128); err != nil { + t.Fatalf("block %d: failed to insert into chain: %v", n, err) + } + chain.Stop() + ancientDb.Close() + + // Reconstruct an ancient db with inserted ancient blocks. + ancientDb, err = rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), frdir, "") + if err != nil { + t.Fatalf("failed to create temp freezer db: %v", err) + } + gspec.MustCommit(ancientDb) + + var oldest uint64 + chain, err = NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 0) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + chain.InsertChain(blocks2[:1]) // Feed chain a higher block to trigger indices updater. + time.Sleep(50 * time.Millisecond) // Wait for indices initialisation + check(&oldest, chain) + chain.Stop() + + // Reconstruct a blockchain which only reserves HEAD-64 tx indices + chain, err = NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 64) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + chain.InsertChain(blocks2[1:2]) // Feed chain a higher block to trigger indices updater. + time.Sleep(50 * time.Millisecond) // Wait for indices initialisation + oldest = chain.CurrentBlock().NumberU64() - 64 + check(&oldest, chain) + chain.Stop() + + // Reconstruct a block which only reserves HEAD-32 tx indices, shorten the indices history. + chain, err = NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 32) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + chain.InsertChain(blocks2[2:3]) // Feed chain a higher block to trigger indices updater. + time.Sleep(50 * time.Millisecond) // Wait for indices initialisation + oldest = chain.CurrentBlock().NumberU64() - 32 + check(&oldest, chain) + chain.Stop() + + // Reconstruct a block which only reserves all tx indices, extends the indices history + chain, err = NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, 0) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + chain.InsertChain(blocks2[3:4]) // Feed chain a higher block to trigger indices updater. + time.Sleep(50 * time.Millisecond) // Wait for indices initialisation + oldest = 0 + check(&oldest, chain) +} + // Benchmarks large blocks with value transfers to non-existing accounts func benchmarkLargeNumberOfValueToNonexisting(b *testing.B, numTxs, numBlocks int, recipientFn func(uint64) common.Address, dataFn func(uint64) []byte) { var ( @@ -2178,7 +2312,7 @@ func benchmarkLargeNumberOfValueToNonexisting(b *testing.B, numTxs, numBlocks in diskdb := rawdb.NewMemoryDatabase() gspec.MustCommit(diskdb) - chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil) + chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, 0) if err != nil { b.Fatalf("failed to create tester chain: %v", err) } diff --git a/core/chain_makers_test.go b/core/chain_makers_test.go index 32e3888d55..63d853327f 100644 --- a/core/chain_makers_test.go +++ b/core/chain_makers_test.go @@ -79,7 +79,7 @@ func ExampleGenerateChain() { }) // Import the chain. This runs all block validation rules. - blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil) + blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, 0) defer blockchain.Stop() if i, err := blockchain.InsertChain(chain); err != nil { diff --git a/core/dao_test.go b/core/dao_test.go index 4e8dba9e84..94d0b077d1 100644 --- a/core/dao_test.go +++ b/core/dao_test.go @@ -45,7 +45,7 @@ func TestDAOForkRangeExtradata(t *testing.T) { proConf.DAOForkBlock = forkBlock proConf.DAOForkSupport = true - proBc, _ := NewBlockChain(proDb, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil) + proBc, _ := NewBlockChain(proDb, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil, 0) defer proBc.Stop() conDb := rawdb.NewMemoryDatabase() @@ -55,7 +55,7 @@ func TestDAOForkRangeExtradata(t *testing.T) { conConf.DAOForkBlock = forkBlock conConf.DAOForkSupport = false - conBc, _ := NewBlockChain(conDb, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil) + conBc, _ := NewBlockChain(conDb, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil, 0) defer conBc.Stop() if _, err := proBc.InsertChain(prefix); err != nil { @@ -69,7 +69,7 @@ func TestDAOForkRangeExtradata(t *testing.T) { // Create a pro-fork block, and try to feed into the no-fork chain db = rawdb.NewMemoryDatabase() gspec.MustCommit(db) - bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil) + bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil, 0) defer bc.Stop() blocks := conBc.GetBlocksFromHash(conBc.CurrentBlock().Hash(), int(conBc.CurrentBlock().NumberU64())) @@ -94,7 +94,7 @@ func TestDAOForkRangeExtradata(t *testing.T) { // Create a no-fork block, and try to feed into the pro-fork chain db = rawdb.NewMemoryDatabase() gspec.MustCommit(db) - bc, _ = NewBlockChain(db, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil) + bc, _ = NewBlockChain(db, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil, 0) defer bc.Stop() blocks = proBc.GetBlocksFromHash(proBc.CurrentBlock().Hash(), int(proBc.CurrentBlock().NumberU64())) @@ -120,7 +120,7 @@ func TestDAOForkRangeExtradata(t *testing.T) { // Verify that contra-forkers accept pro-fork extra-datas after forking finishes db = rawdb.NewMemoryDatabase() gspec.MustCommit(db) - bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil) + bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil, 0) defer bc.Stop() blocks := conBc.GetBlocksFromHash(conBc.CurrentBlock().Hash(), int(conBc.CurrentBlock().NumberU64())) @@ -140,7 +140,7 @@ func TestDAOForkRangeExtradata(t *testing.T) { // Verify that pro-forkers accept contra-fork extra-datas after forking finishes db = rawdb.NewMemoryDatabase() gspec.MustCommit(db) - bc, _ = NewBlockChain(db, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil) + bc, _ = NewBlockChain(db, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil, 0) defer bc.Stop() blocks = proBc.GetBlocksFromHash(proBc.CurrentBlock().Hash(), int(proBc.CurrentBlock().NumberU64())) diff --git a/core/genesis_test.go b/core/genesis_test.go index c6bcd0aa54..9ae64e8a38 100644 --- a/core/genesis_test.go +++ b/core/genesis_test.go @@ -120,7 +120,7 @@ func TestSetupGenesis(t *testing.T) { // Advance to block #4, past the homestead transition block of customg. genesis := oldcustomg.MustCommit(db) - bc, _ := NewBlockChain(db, nil, oldcustomg.Config, ethash.NewFullFaker(), vm.Config{}, nil) + bc, _ := NewBlockChain(db, nil, oldcustomg.Config, ethash.NewFullFaker(), vm.Config{}, nil, 0) defer bc.Stop() blocks, _ := GenerateChain(oldcustomg.Config, genesis, ethash.NewFaker(), db, 4, nil) diff --git a/core/rawdb/accessors_chain.go b/core/rawdb/accessors_chain.go index 50df9adc79..e546af5bba 100644 --- a/core/rawdb/accessors_chain.go +++ b/core/rawdb/accessors_chain.go @@ -171,10 +171,11 @@ func WriteFastTrieProgress(db ethdb.KeyValueWriter, count uint64) { } } -// ReadAncientTxLookupProgress retrieves the number of ancient blocks which -// txlookup has been inserted to allow reporting correct numbers across restarts. -func ReadAncientTxLookupProgress(db ethdb.KeyValueReader) *uint64 { - data, _ := db.Get(ancientTxLookupProgressKey) +// ReadOldestIndexedBlock retrieves the number of oldest indexed block +// whose transaction indices has been indexed. If the corresponding entry +// is non-existent in database it means the indexing has been finished. +func ReadOldestIndexedBlock(db ethdb.KeyValueReader) *uint64 { + data, _ := db.Get(oldestIndexedBlockKey) if len(data) != 8 { return nil } @@ -182,18 +183,11 @@ func ReadAncientTxLookupProgress(db ethdb.KeyValueReader) *uint64 { return &number } -// WriteAncientTxLookupProgress stores the ancient txlookup process counter to support -// retrieving it across restarts. -func WriteAncientTxLookupProgress(db ethdb.KeyValueWriter, number uint64) { - if err := db.Put(ancientTxLookupProgressKey, encodeBlockNumber(number)); err != nil { - log.Crit("Failed to store head number of txlookup", "err", err) - } -} - -// DeleteAncientTxLookupProgress deletes the ancient txlookup progress. -func DeleteAncientTxLookupProgress(db ethdb.KeyValueWriter) { - if err := db.Delete(ancientTxLookupProgressKey); err != nil { - log.Crit("Failed to delete ancient txlookup progress entry", "err", err) +// WriteOldestIndexedBlock stores the number of oldest indexed block +// into database. +func WriteOldestIndexedBlock(db ethdb.KeyValueWriter, number uint64) { + if err := db.Put(oldestIndexedBlockKey, encodeBlockNumber(number)); err != nil { + log.Crit("Failed to store the number of oldest indexed block", "err", err) } } @@ -584,3 +578,40 @@ func FindCommonAncestor(db ethdb.Reader, a, b *types.Header) *types.Header { } return a } + +// FindOldestIndexedBlock binary searches the oldest block which has been indexed. +// We will always ensures that if Bi is indexed, then Bi+1 must has been indexed. +// +// If no block has been indexed, then the returned value is to+1. +// +// The block doesn't contain any transaction will be regarded as unindexed. It can +// cause the blocks before this block can be reindexed. +func FindOldestIndexedBlock(db ethdb.Reader, from uint64, to uint64) *uint64 { + low, high := from, to+1 + + check := func(number uint64) bool { + block := ReadBlock(db, ReadCanonicalHash(db, number), number) + if block == nil { + log.Crit("Failed to retrieve block from database", "number", number) + } + if block.Transactions().Len() == 0 { + return false + } + if ReadTxLookupEntry(db, block.Transactions()[0].Hash()) == nil { + return false + } + return true + } + for low != high { + mid := (low + high) / 2 + if !check(mid) { + low = mid + 1 + } else { + high = mid + } + } + if low == to+1 { + return nil + } + return &low +} diff --git a/core/rawdb/accessors_chain_test.go b/core/rawdb/accessors_chain_test.go index 8c8affffd9..aacbe32216 100644 --- a/core/rawdb/accessors_chain_test.go +++ b/core/rawdb/accessors_chain_test.go @@ -358,3 +358,52 @@ func checkReceiptsRLP(have, want types.Receipts) error { } return nil } + +func TestFindOldestIndexedBlock(t *testing.T) { + var cases = []struct { + empty bool + oldest uint64 + height uint64 + nilBlocks map[uint64]bool + expect uint64 + }{ + {true, 0, 10, nil, 0}, // No block has been indexed + {false, 0, 10, nil, 1}, // Genesis block doesn't have indices + {false, 1, 10, nil, 1}, + {false, 4, 10, nil, 4}, + {false, 5, 10, nil, 5}, + {false, 6, 10, nil, 6}, + {false, 10, 10, nil, 10}, + {false, 3, 10, map[uint64]bool{4: true, 6: true, 8: true}, 5}, + } + for cid, c := range cases { + var ( + db = NewMemoryDatabase() + block *types.Block + ) + for i := uint64(0); i <= c.height; i++ { + if i == 0 { + block = types.NewBlock(&types.Header{Number: big.NewInt(int64(i))}, nil, nil, nil) // Empty genesis block + } else { + tx := types.NewTransaction(i, common.BytesToAddress([]byte{0x11}), big.NewInt(111), 1111, big.NewInt(11111), []byte{0x11, 0x11, 0x11}) + txset := []*types.Transaction{tx} + if c.nilBlocks != nil && c.nilBlocks[i] { + txset = nil + } + block = types.NewBlock(&types.Header{Number: big.NewInt(int64(i))}, txset, nil, nil) + } + WriteBlock(db, block) + WriteCanonicalHash(db, block.Hash(), block.NumberU64()) + if !c.empty && block.NumberU64() >= c.oldest { + WriteTxLookupEntries(db, block) + } + } + res := FindOldestIndexedBlock(db, 0, c.height) + if c.empty && res != nil { + t.Fatalf("Case %d failed, oldest block mismatch, want nil, have %d", cid, *res) + } + if !c.empty && *res != c.expect { + t.Fatalf("Case %d failed, oldest block mismatch, want %d, have %d", cid, c.expect, *res) + } + } +} diff --git a/core/rawdb/accessors_indexes.go b/core/rawdb/accessors_indexes.go index 38f8fe10ea..4f99436bf3 100644 --- a/core/rawdb/accessors_indexes.go +++ b/core/rawdb/accessors_indexes.go @@ -63,6 +63,16 @@ func WriteTxLookupEntries(db ethdb.KeyValueWriter, block *types.Block) { } } +// DeleteTxLookupEntries removes all transaction lookup indices contained in +// given block. +func DeleteTxLookupEntries(db ethdb.KeyValueWriter, block *types.Block) { + for _, tx := range block.Transactions() { + if err := db.Delete(txLookupKey(tx.Hash())); err != nil { + log.Crit("Failed to delete transaction lookup entry", "err", err) + } + } +} + // DeleteTxLookupEntry removes all transaction data associated with a hash. func DeleteTxLookupEntry(db ethdb.KeyValueWriter, hash common.Hash) { db.Delete(txLookupKey(hash)) diff --git a/core/rawdb/chain_iterator.go b/core/rawdb/chain_iterator.go new file mode 100644 index 0000000000..e54217c15d --- /dev/null +++ b/core/rawdb/chain_iterator.go @@ -0,0 +1,230 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package rawdb + +import ( + "errors" + "math" + "runtime" + "sync/atomic" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/prque" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/log" +) + +type ( + prepareCallback func(*types.Block) // The callback for customized prepare operation. + actionCallback func(ethdb.Batch, *types.Block) // The callback for customized action. +) + +// iterateCanonicalChain iterates the specified range canonical chain and then apply +// the given action callback. +// Note for forward iteration, the range is [from, to), otherwise the range is (from, to]. +func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, typ string, prepare prepareCallback, action actionCallback, reverse bool, report bool) error { + // Short circuit if the action is nil. + if action == nil { + return nil + } + // Short circuit if the iteration range is invalid. + if from >= to { + return nil + } + // Spawn multi-routines, iterate over the specified blocks and invoke prepare + // callback concurrently. + var ( + number int64 + results = make(chan *types.Block, 4*runtime.NumCPU()) + ) + if !reverse { + number = int64(from - 1) + } else { + number = int64(to + 1) + } + abort := make(chan struct{}) + defer close(abort) + + for i := 0; i < runtime.NumCPU(); i++ { + go func() { + for { + // Fetch the next task number, terminating if everything's done + var n int64 + if !reverse { + n = atomic.AddInt64(&number, 1) + if n >= int64(to) { + return + } + } else { + n = atomic.AddInt64(&number, -1) + if n <= int64(from) { + return + } + } + block := ReadBlock(db, ReadCanonicalHash(db, uint64(n)), uint64(n)) + if prepare != nil && block != nil { + prepare(block) + } + // Feed the block to the aggregator, or abort on interrupt + select { + case results <- block: + case <-abort: + return + } + } + }() + } + // Reassemble the blocks into a contiguous stream and apply the action callback. + var ( + next, first, last int64 + queue = prque.New(nil) + + batch = db.NewBatch() + start = time.Now() + logged time.Time + ) + if !reverse { + next, first, last = int64(from), int64(from), int64(to) + } else { + next, first, last = int64(to), int64(to), int64(from) + } + logFn := log.Debug + if report { + logFn = log.Info + } + for i := from; i < to; i++ { + // Retrieve the next result and bail if it's nil + block := <-results + if block == nil { + return errors.New("broken database") + } + // Push the block into the import queue and process contiguous ranges + priority := -int64(block.NumberU64()) + if reverse { + priority = int64(block.NumberU64()) + } + queue.Push(block, priority) + for !queue.Empty() { + // If the next available item is gapped, return + if _, priority := queue.Peek(); !reverse && -priority != next || reverse && priority != next { + break + } + // Next block available, pop it off and index it + block = queue.PopItem().(*types.Block) + + if !reverse { + next++ + } else { + next-- + } + // Invoke action to inject specified data into key-value database. + action(batch, block) + + // If enough data was accumulated in memory or we're at the last block, dump to disk + if batch.ValueSize() > ethdb.IdealBatchSize || next == last { + if err := batch.Write(); err != nil { + return err + } + batch.Reset() + } + // If we've spent too much time already, notify the user of what we're doing + if time.Since(logged) > 8*time.Second { + logFn("Iterating canonical chain", "type", typ, "reserve", reverse, "number", block.Number(), "hash", block.Hash(), "total", int64(math.Abs(float64(next-first))), "elapsed", common.PrettyDuration(time.Since(start))) + logged = time.Now() + } + } + } + logFn("Iterated canonical chain", "type", typ, "reverse", reverse, "total", to-from, "elapsed", common.PrettyDuration(time.Since(start))) + return nil +} + +// InitBlockIndexFromFreezer reinitializes an empty database from a previous batch +// of frozen ancient blocks. The method iterates over all the frozen blocks and +// injects into the database the block hash->number mappings and the transaction +// lookup entries. +func InitBlockIndexFromFreezer(db ethdb.Database) { + // If we can't access the freezer or it's empty, abort + frozen, err := db.Ancients() + if err != nil || frozen == 0 { + return + } + // hashBlock calculates block hash in advance using the multi-routine's concurrent + // computing power. + hashBlock := func(block *types.Block) { block.Hash() } + + // writeIndex injects hash <-> number mapping into the database. + writeIndex := func(batch ethdb.Batch, block *types.Block) { WriteHeaderNumber(batch, block.Hash(), block.NumberU64()) } + + if err := iterateCanonicalChain(db, 0, frozen, "blocks", hashBlock, writeIndex, false, true); err != nil { + log.Crit("Failed to iterate canonical chain", "err", err) + } + hash := ReadCanonicalHash(db, frozen-1) + WriteHeadHeaderHash(db, hash) + WriteHeadFastBlockHash(db, hash) + log.Info("Initialized chain from ancient data", "number", frozen-1, "hash", hash) +} + +// IndexTxLookup initializes txlookup indices of the specified range blocks into the database. +// +// This function iterates canonical chain in reverse order, it has two advantages: +// * If Geth crashes during the indexing without writing the oldest flag, we can +// binary search to quickly locate the oldest indexed block +// * We can write oldest indexed block flag periodically even without the whole +// indexing procedure is finished. So that we can resume indexing procedure next +// time quickly. +func IndexTxLookup(db ethdb.Database, from uint64, to uint64) { + // hashTxs calculates transaction hash in advance using the multi-routine's + // concurrent computing power. + hashTxs := func(block *types.Block) { + for _, tx := range block.Transactions() { + tx.Hash() + } + } + // writeIndices injects txlookup indices into the database. + writeIndices := func(batch ethdb.Batch, block *types.Block) { + WriteTxLookupEntries(batch, block) + if block.NumberU64()%1000000 == 0 { + WriteOldestIndexedBlock(batch, block.NumberU64()) + } + } + if err := iterateCanonicalChain(db, from, to, "txlookup", hashTxs, writeIndices, true, true); err != nil { + log.Crit("Failed to iterate canonical chain", "err", err) + } + WriteOldestIndexedBlock(db, from) + log.Info("Constructed transaction indices", "from", from, "to", to, "count", to-from) +} + +// RemoveTxsLookup removes txlookup indices of the specified range blocks. +func RemoveTxsLookup(db ethdb.Database, from uint64, to uint64) { + // Write flag first and then unindex the transaction indices. Some indices + // will be left in the database if crash happens but it's fine. + WriteOldestIndexedBlock(db, to) + + if from+1 == to { + hash := ReadCanonicalHash(db, from) + DeleteTxLookupEntries(db, ReadBlock(db, hash, from)) + log.Debug("Removed transaction indices", "number", from, "hash", hash) + } else { + deleteIndices := func(batch ethdb.Batch, block *types.Block) { DeleteTxLookupEntries(batch, block) } + if err := iterateCanonicalChain(db, from, to, "txlookup", nil, deleteIndices, false, false); err != nil { + log.Crit("Failed to iterate canonical chain", "err", err) + } + log.Debug("Removed transaction indices", "from", from, "to", to, "count", to-from) + } +} diff --git a/core/rawdb/initer.go b/core/rawdb/initer.go deleted file mode 100644 index d5116d558d..0000000000 --- a/core/rawdb/initer.go +++ /dev/null @@ -1,180 +0,0 @@ -// Copyright 2019 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package rawdb - -import ( - "errors" - "runtime" - "sync/atomic" - "time" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/common/prque" - "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/ethdb" - "github.com/ethereum/go-ethereum/log" -) - -type ( - initPrepare func(*types.Block) // The callback for customized prepare operation. - initAction func(ethdb.Batch, *types.Block) // The callback for customized initialisation action. -) - -// iterateAncient iterates the specified range blocks from ancient database -// and then apply initialisation action. -func iterateAncient(db ethdb.Database, from uint64, typ string, prepare initPrepare, action initAction) error { - // Short circuit if the init action is nil. - if action == nil { - return nil - } - // If we can't access the freezer or it's empty, abort - frozen, err := db.Ancients() - if err != nil || frozen == 0 { - return err - } - // Spawn multi-routines, iterate over the specified blocks and invoke prepare - // callback concurrently. - var ( - number uint64 - results = make(chan *types.Block, 4*runtime.NumCPU()) - ) - if from == 0 { - number = ^uint64(0) // -1 - } else { - number = from - 1 - } - abort := make(chan struct{}) - defer close(abort) - - for i := 0; i < runtime.NumCPU(); i++ { - go func() { - for { - // Fetch the next task number, terminating if everything's done - n := atomic.AddUint64(&number, 1) - if n >= frozen { - return - } - // Retrieve the block from the freezer (no need for the hash, we pull by - // number from the freezer). - block := ReadBlock(db, common.Hash{}, n) - if prepare != nil && block != nil { - prepare(block) - } - // Feed the block to the aggregator, or abort on interrupt - select { - case results <- block: - case <-abort: - return - } - } - }() - } - // Reassemble the blocks into a contiguous stream and apply the action callback. - var ( - queue = prque.New(nil) - next = int64(from) - - batch = db.NewBatch() - start = time.Now() - logged time.Time - ) - for i := from; i < frozen; i++ { - // Retrieve the next result and bail if it's nil - block := <-results - if block == nil { - return errors.New("broken database") - } - // Push the block into the import queue and process contiguous ranges - queue.Push(block, -int64(block.NumberU64())) - for !queue.Empty() { - // If the next available item is gapped, return - if _, priority := queue.Peek(); -priority != next { - break - } - // Next block available, pop it off and index it - block = queue.PopItem().(*types.Block) - next++ - - // Invoke action to inject specified data into key-value database. - action(batch, block) - - // If enough data was accumulated in memory or we're at the last block, dump to disk - if batch.ValueSize() > ethdb.IdealBatchSize || uint64(next) == frozen { - if err := batch.Write(); err != nil { - return err - } - batch.Reset() - } - // If we've spent too much time already, notify the user of what we're doing - if time.Since(logged) > 8*time.Second { - log.Info("Initializing chain from ancient data", "type", typ, "number", block.Number(), "hash", block.Hash(), "total", uint64(next)-from, "elapsed", common.PrettyDuration(time.Since(start))) - logged = time.Now() - } - } - } - log.Info("Initialized chain from ancient data", "type", typ, "number", frozen-from, "elapsed", common.PrettyDuration(time.Since(start))) - return nil -} - -// InitBlockIndexFromFreezer reinitializes an empty database from a previous batch -// of frozen ancient blocks. The method iterates over all the frozen blocks and -// injects into the database the block hash->number mappings and the transaction -// lookup entries. -func InitBlockIndexFromFreezer(db ethdb.Database) error { - // If we can't access the freezer or it's empty, abort - frozen, err := db.Ancients() - if err != nil || frozen == 0 { - return err - } - // hashBlock calculates block hash in advance using the multi-routine's concurrent - // computing power. - hashBlock := func(block *types.Block) { block.Hash() } - - // writeIndex injects hash <-> number mapping into the database. - writeIndex := func(batch ethdb.Batch, block *types.Block) { WriteHeaderNumber(batch, block.Hash(), block.NumberU64()) } - - if err := iterateAncient(db, 0, "blocks", hashBlock, writeIndex); err != nil { - return err - } - hash := ReadCanonicalHash(db, frozen-1) - WriteHeadHeaderHash(db, hash) - WriteHeadFastBlockHash(db, hash) - return nil -} - -// InitTxsLookupFromFreezer initializes txlookup indexes in the database. -func InitTxsLookupFromFreezer(db ethdb.Database, from uint64) error { - // hashTxs calculates transaction hash in advance using the multi-routine's - // concurrent computing power. - hashTxs := func(block *types.Block) { - for _, tx := range block.Transactions() { - tx.Hash() - } - } - // writeIndex injects txlookup indexes into the database. - writeIndex := func(batch ethdb.Batch, block *types.Block) { - WriteTxLookupEntries(batch, block) - if block.NumberU64()%10000 == 0 { - WriteAncientTxLookupProgress(batch, block.NumberU64()) - } - } - if err := iterateAncient(db, from, "txlookup", hashTxs, writeIndex); err != nil { - return err - } - DeleteAncientTxLookupProgress(db) // Mark all txlookup indexes of ancient blocks have been inserted. - return nil -} diff --git a/core/rawdb/schema.go b/core/rawdb/schema.go index 34f47c6422..097b04fcac 100644 --- a/core/rawdb/schema.go +++ b/core/rawdb/schema.go @@ -41,8 +41,8 @@ var ( // fastTrieProgressKey tracks the number of trie entries imported during fast sync. fastTrieProgressKey = []byte("TrieSync") - // ancientTxLookupProgressKey tracks the progress of ancient txs lookup insertion. - ancientTxLookupProgressKey = []byte("AncientTxsLookup") + // oldestIndexedBlockKey tracks the oldest block whose transaction indices(txlookup) has been indexed. + oldestIndexedBlockKey = []byte("OldestIndexedBlock") // Data item prefixes (use single byte to avoid mixing data types, avoid `i`, used for indexes). headerPrefix = []byte("h") // headerPrefix + num (uint64 big endian) + hash -> header diff --git a/eth/backend.go b/eth/backend.go index ce37541f4d..05518969f9 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -186,7 +186,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { TrieTimeLimit: config.TrieTimeout, } ) - eth.blockchain, err = core.NewBlockChain(chainDb, cacheConfig, chainConfig, eth.engine, vmConfig, eth.shouldPreserve) + eth.blockchain, err = core.NewBlockChain(chainDb, cacheConfig, chainConfig, eth.engine, vmConfig, eth.shouldPreserve, config.TxLookupLimit) if err != nil { return nil, err } diff --git a/eth/config.go b/eth/config.go index 5094a533bf..fe846c0546 100644 --- a/eth/config.go +++ b/eth/config.go @@ -98,6 +98,8 @@ type Config struct { NoPruning bool // Whether to disable pruning and flush everything to disk NoPrefetch bool // Whether to disable prefetching and only load state on demand + TxLookupLimit uint64 `toml:",omitempty"` // The maximum number of blocks from head whose tx indices are reserved. + // Whitelist of required block number -> hash values to accept Whitelist map[uint64]common.Hash `toml:"-"` diff --git a/eth/gen_config.go b/eth/gen_config.go index bc4b55b120..1a19314dc2 100644 --- a/eth/gen_config.go +++ b/eth/gen_config.go @@ -23,6 +23,7 @@ func (c Config) MarshalTOML() (interface{}, error) { SyncMode downloader.SyncMode NoPruning bool NoPrefetch bool + TxLookupLimit uint64 `toml:",omitempty"` Whitelist map[uint64]common.Hash `toml:"-"` LightServ int `toml:",omitempty"` LightIngress int `toml:",omitempty"` @@ -56,6 +57,7 @@ func (c Config) MarshalTOML() (interface{}, error) { enc.SyncMode = c.SyncMode enc.NoPruning = c.NoPruning enc.NoPrefetch = c.NoPrefetch + enc.TxLookupLimit = c.TxLookupLimit enc.Whitelist = c.Whitelist enc.LightServ = c.LightServ enc.LightIngress = c.LightIngress @@ -93,6 +95,7 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error { SyncMode *downloader.SyncMode NoPruning *bool NoPrefetch *bool + TxLookupLimit *uint64 `toml:",omitempty"` Whitelist map[uint64]common.Hash `toml:"-"` LightServ *int `toml:",omitempty"` LightIngress *int `toml:",omitempty"` @@ -139,6 +142,9 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error { if dec.NoPrefetch != nil { c.NoPrefetch = *dec.NoPrefetch } + if dec.TxLookupLimit != nil { + c.TxLookupLimit = *dec.TxLookupLimit + } if dec.Whitelist != nil { c.Whitelist = dec.Whitelist } diff --git a/eth/handler_test.go b/eth/handler_test.go index 0f1672fd44..a694842b43 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -489,7 +489,7 @@ func testCheckpointChallenge(t *testing.T, syncmode downloader.SyncMode, checkpo } } // Create a checkpoint aware protocol manager - blockchain, err := core.NewBlockChain(db, nil, config, ethash.NewFaker(), vm.Config{}, nil) + blockchain, err := core.NewBlockChain(db, nil, config, ethash.NewFaker(), vm.Config{}, nil, 0) if err != nil { t.Fatalf("failed to create new blockchain: %v", err) } @@ -576,7 +576,7 @@ func testBroadcastBlock(t *testing.T, totalPeers, broadcastExpected int) { gspec = &core.Genesis{Config: config} genesis = gspec.MustCommit(db) ) - blockchain, err := core.NewBlockChain(db, nil, config, pow, vm.Config{}, nil) + blockchain, err := core.NewBlockChain(db, nil, config, pow, vm.Config{}, nil, 0) if err != nil { t.Fatalf("failed to create new blockchain: %v", err) } diff --git a/eth/helper_test.go b/eth/helper_test.go index 1482e99c4e..b8990f4a40 100644 --- a/eth/helper_test.go +++ b/eth/helper_test.go @@ -60,7 +60,7 @@ func newTestProtocolManager(mode downloader.SyncMode, blocks int, generator func Alloc: core.GenesisAlloc{testBank: {Balance: big.NewInt(1000000)}}, } genesis = gspec.MustCommit(db) - blockchain, _ = core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}, nil) + blockchain, _ = core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}, nil, 0) ) chain, _ := core.GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, blocks, generator) if _, err := blockchain.InsertChain(chain); err != nil { diff --git a/light/odr_test.go b/light/odr_test.go index debd5544c3..c6c8d25634 100644 --- a/light/odr_test.go +++ b/light/odr_test.go @@ -257,7 +257,7 @@ func testChainOdr(t *testing.T, protocol int, fn odrTestFn) { ) gspec.MustCommit(ldb) // Assemble the test environment - blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil) + blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil, 0) gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), sdb, 4, testChainGen) if _, err := blockchain.InsertChain(gchain); err != nil { t.Fatal(err) diff --git a/light/trie_test.go b/light/trie_test.go index 4919f89641..6d960642e6 100644 --- a/light/trie_test.go +++ b/light/trie_test.go @@ -40,7 +40,7 @@ func TestNodeIterator(t *testing.T) { genesis = gspec.MustCommit(fulldb) ) gspec.MustCommit(lightdb) - blockchain, _ := core.NewBlockChain(fulldb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil) + blockchain, _ := core.NewBlockChain(fulldb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil, 0) gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), fulldb, 4, testChainGen) if _, err := blockchain.InsertChain(gchain); err != nil { panic(err) diff --git a/light/txpool_test.go b/light/txpool_test.go index 0996bd7c9c..e8441ab5f5 100644 --- a/light/txpool_test.go +++ b/light/txpool_test.go @@ -88,7 +88,7 @@ func TestTxPool(t *testing.T) { ) gspec.MustCommit(ldb) // Assemble the test environment - blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil) + blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil, 0) gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), sdb, poolTestBlocks, txPoolTestChainGen) if _, err := blockchain.InsertChain(gchain); err != nil { panic(err) diff --git a/miner/worker_test.go b/miner/worker_test.go index 1604e988dd..3d2372648f 100644 --- a/miner/worker_test.go +++ b/miner/worker_test.go @@ -103,7 +103,7 @@ func newTestWorkerBackend(t *testing.T, chainConfig *params.ChainConfig, engine } genesis := gspec.MustCommit(db) - chain, _ := core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}, nil) + chain, _ := core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}, nil, 0) txpool := core.NewTxPool(testTxPoolConfig, chainConfig, chain) // Generate a small n-block chain and an uncle block for it diff --git a/tests/block_test_util.go b/tests/block_test_util.go index 81dd7b1d04..441f0af21b 100644 --- a/tests/block_test_util.go +++ b/tests/block_test_util.go @@ -118,7 +118,7 @@ func (t *BlockTest) Run() error { } else { engine = ethash.NewShared() } - chain, err := core.NewBlockChain(db, &core.CacheConfig{TrieCleanLimit: 0}, config, engine, vm.Config{}, nil) + chain, err := core.NewBlockChain(db, &core.CacheConfig{TrieCleanLimit: 0}, config, engine, vm.Config{}, nil, 0) if err != nil { return err }