From ace22f8e013faf49d41ff1607eaeb4bfd70d2935 Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Sun, 28 Jul 2019 22:09:22 +0800 Subject: [PATCH] cmd, core, eth: add comments --- cmd/utils/flags.go | 6 +- core/blockchain.go | 157 +++++++++++++++-------------- core/blockchain_test.go | 96 ++++++++---------- core/rawdb/accessors_chain.go | 54 ++++------ core/rawdb/accessors_chain_test.go | 53 ---------- core/rawdb/chain_iterator.go | 29 +++--- core/rawdb/schema.go | 3 + eth/downloader/downloader.go | 8 +- eth/downloader/downloader_test.go | 2 +- eth/handler.go | 2 + eth/sync.go | 19 ++++ 11 files changed, 181 insertions(+), 248 deletions(-) diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 8739d49899..fd8c2cbb30 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -1727,11 +1727,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)} - txLookupLimit := uint64(0) - if ctx.GlobalIsSet(TxLookupLimitFlag.Name) { - txLookupLimit = ctx.GlobalUint64(TxLookupLimitFlag.Name) - } - chain, err = core.NewBlockChain(chainDb, cache, config, engine, vmcfg, nil, txLookupLimit) + chain, err = core.NewBlockChain(chainDb, cache, config, engine, vmcfg, nil, 0) if err != nil { Fatalf("Can't create BlockChain: %v", err) } diff --git a/core/blockchain.go b/core/blockchain.go index 9f9fec0936..cb03095bac 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -134,10 +134,15 @@ 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 - txLookupLimit uint64 // The maximum number of blocks from head whose tx indices are reserved + 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 is the maximum number of blocks from head whose tx indices + // are reserved. + // * 0 means no limit (and regenerate any missing) + // * N means N blocks limit [HEAD-N, HEAD] + txLookupLimit uint64 hc *HeaderChain rmLogsFeed event.Feed @@ -231,8 +236,15 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par return nil, ErrNoGenesis } // Initialize the chain with ancient data if it isn't empty. + var ancients uint64 if bc.empty() { rawdb.InitBlockIndexFromFreezer(bc.db) + // If ancient database is not empty, reconstruct all missing + // indices in the background. + frozen, _ := bc.db.Ancients() + if frozen > 0 { + ancients = frozen + } } if err := bc.loadLastState(); err != nil { return nil, err @@ -290,7 +302,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par } // Take ownership of this particular state go bc.update() - go bc.maintainTxIndex() + go bc.maintainTxIndex(ancients) return bc, nil } @@ -457,7 +469,7 @@ func (bc *BlockChain) SetHead(head uint64) error { // FastSyncCommitHead sets the current head block to the one defined by the hash // irrelevant what the chain contents were prior. -func (bc *BlockChain) FastSyncCommitHead(hash common.Hash, from uint64, ancient uint64) error { +func (bc *BlockChain) FastSyncCommitHead(hash common.Hash) error { // Make sure that both the block as well at its state trie exists block := bc.GetBlockByHash(hash) if block == nil { @@ -472,22 +484,6 @@ func (bc *BlockChain) FastSyncCommitHead(hash common.Hash, from uint64, ancient headBlockGauge.Update(int64(block.NumberU64())) bc.chainmu.Unlock() - // Write tx indices tail if it doesn't exist in database. - // The tx index tail can only be one of the following two options: - // * the start point of fast sync: - // in this case all blocks imported during fast sync have been indexed - // * ancient - limit: - // in this case the block before ancient-limit won't be indexed - if tail := rawdb.ReadTxIndexTail(bc.db); tail == nil { - if bc.txLookupLimit != 0 && ancient >= bc.txLookupLimit && ancient-bc.txLookupLimit > from { - rawdb.WriteTxIndexTail(bc.db, ancient-bc.txLookupLimit) - } else { - if from == 1 { - from = 0 - } - rawdb.WriteTxIndexTail(bc.db, from) - } - } log.Info("Committed new head block", "number", block.Number(), "hash", hash) return nil } @@ -1218,6 +1214,19 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ return n, err } } + // Write tx indices tail before write any live block. + if len(ancientBlocks) > 0 && len(liveBlocks) > 0 { + // The tx index tail can only be one of the following two options: + // * 0: all ancient blocks have been indexed. + // * ancient-limit: the indices of blocks before ancient-limit are ignored + if tail := rawdb.ReadTxIndexTail(bc.db); tail == nil { + if bc.txLookupLimit == 0 || ancientLimit <= bc.txLookupLimit { + rawdb.WriteTxIndexTail(bc.db, 0) + } else { + rawdb.WriteTxIndexTail(bc.db, ancientLimit-bc.txLookupLimit) + } + } + } if len(liveBlocks) > 0 { if n, err := writeLive(liveBlocks, liveReceipts); err != nil { if err == errInsertionInterrupted { @@ -1241,6 +1250,19 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ return 0, nil } +// AdjustTxLookupLimit is responsible for updating the txlookup limit +// to the original one stored in db if the new old mismatch with the old +// one. +func (bc *BlockChain) AdjustTxLookupLimit(limit uint64) { + bc.txLookupLimit = limit +} + +// TxLookupLimit retrieves the txlookup limit used by blockchain to prune +// stale tx indices. +func (bc *BlockChain) TxLookupLimit() uint64 { + return bc.txLookupLimit +} + var lastWrite uint64 // writeBlockWithoutState writes only the block and its metadata to the database, @@ -2069,74 +2091,58 @@ func (bc *BlockChain) update() { // 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) maintainTxIndex() { - // initialiseIndices inits tx indices into the database if `TxIndexTail` - // is missing in database. - // - // Note for archive sync or full sync, this code path will only be triggered - // after importing the first batch of blocks(e.g. 1024). But these block - // actually are already indexed. So a binary search will be performed to - // skip reindexing. - // - // Besides for old node database which actually contains all indices but - // without `TxIndexTail` in database, binary search can also help to skip - // reindexing. - initialiseIndices := func(head uint64, done chan struct{}) { - defer func() { done <- struct{}{} }() - - from, to := uint64(0), head - if bc.txLookupLimit != 0 && head > bc.txLookupLimit { - from = head - bc.txLookupLimit - } - if tail := rawdb.FindTxIndexTail(bc.db, from, to); tail != nil { - // Special case here is genesis block doesn't contain any transaction - // that will be regarded as unindexed. - if *tail == from || (from == 0 && *tail == 1) { - rawdb.WriteTxIndexTail(bc.db, from) - - // Drop all useless tx indices below the HEAD-limit. - if from > 0 { - rawdb.RemoveTxsLookup(bc.db, 0, from) - } - return - } - } - // Re-construct missing tx indices. - rawdb.IndexTxLookup(bc.db, from, to) - - // Drop all useless tx indices below the HEAD-limit. - if from > 0 { - rawdb.RemoveTxsLookup(bc.db, 0, from) - } - } +// The user can adjust the txlookuplimit value for each launch after fast +// sync, Geth will automatically construct the missing indices and delete +// the extra indices. +func (bc *BlockChain) maintainTxIndex(ancients uint64) { // indexBlocks reindex or unindex transaction indices depends // on user's requirement. - indexBlocks := func(oldest uint64, head uint64, done chan struct{}) { + indexBlocks := func(tail *uint64, head uint64, done chan struct{}) { defer func() { done <- struct{}{} }() + if tail == nil { + // This is a special case that user upgrades Geth to a new version + // which supports tx indices pruning feature but the tx index tail + // is missing. So that we can assume all blocks in db are indexed. + if bc.txLookupLimit == 0 || head <= bc.txLookupLimit { + // Nothing to delete, write the tail and return. + rawdb.WriteTxIndexTail(bc.db, 0) + } else { + // Prune all stale tx indices and record the tx index tail. + rawdb.RemoveTxsLookup(bc.db, 0, head-bc.txLookupLimit) + } + return + } // All indices should be reserved. if bc.txLookupLimit == 0 || head <= bc.txLookupLimit { - if oldest == 0 { + if *tail == 0 { // Short circuit if nothing to delete. - return } else { // Reindex all indices if necessary - rawdb.IndexTxLookup(bc.db, 0, oldest) - return + rawdb.IndexTxLookup(bc.db, 0, *tail) } + return } - if head-bc.txLookupLimit < oldest { + if head-bc.txLookupLimit < *tail { // Reindex a part of missing indices and rewind oldest indexed // point to HEAD-limit - rawdb.IndexTxLookup(bc.db, head-bc.txLookupLimit, oldest) + rawdb.IndexTxLookup(bc.db, head-bc.txLookupLimit, *tail) } else { // Unindex a part of stale indices and forward oldest indexed // point to HEAD-limit - rawdb.RemoveTxsLookup(bc.db, oldest, head-bc.txLookupLimit) + rawdb.RemoveTxsLookup(bc.db, *tail, head-bc.txLookupLimit) } } + // Special case here: user might init Geth with an external ancient database. + // If so, we should reindex all necessary indices before start processing any + // indices pruning requests. + if ancients > 0 { + var from = uint64(0) + if bc.txLookupLimit != 0 && ancients > bc.txLookupLimit { + from = ancients - bc.txLookupLimit - 1 + } + rawdb.IndexTxLookup(bc.db, from, ancients) + } var ( done chan struct{} // Non-nil if background unindexing or reindexing routine is active. headCh = make(chan ChainHeadEvent) @@ -2147,17 +2153,12 @@ func (bc *BlockChain) maintainTxIndex() { sub.Unsubscribe() } }() - for { select { case head := <-headCh: if done == nil { done = make(chan struct{}) - if tail := rawdb.ReadTxIndexTail(bc.db); tail == nil { - go initialiseIndices(head.Block.NumberU64(), done) - } else { - go indexBlocks(*tail, head.Block.NumberU64(), done) - } + go indexBlocks(rawdb.ReadTxIndexTail(bc.db), head.Block.NumberU64(), done) } case <-done: done = nil diff --git a/core/blockchain_test.go b/core/blockchain_test.go index c1943118a9..34e8695e4e 100644 --- a/core/blockchain_test.go +++ b/core/blockchain_test.go @@ -2154,16 +2154,16 @@ func TestTransactionIndices(t *testing.T) { }) blocks2, _ := GenerateChain(gspec.Config, blocks[len(blocks)-1], ethash.NewFaker(), gendb, 10, nil) - check := func(oldest *uint64, chain *BlockChain) { - indexed := rawdb.ReadTxIndexTail(chain.db) - if oldest == nil && indexed != nil { - t.Fatalf("Oldest indexded block mismatch, want nil, have %d", *indexed) + check := func(tail *uint64, chain *BlockChain) { + stored := rawdb.ReadTxIndexTail(chain.db) + if tail == nil && stored != nil { + t.Fatalf("Oldest indexded block mismatch, want nil, have %d", *stored) } - if oldest != nil && *indexed != *oldest { - t.Fatalf("Oldest indexded block mismatch, want %d, have %d", *oldest, *indexed) + if tail != nil && *stored != *tail { + t.Fatalf("Oldest indexded block mismatch, want %d, have %d", *tail, *stored) } - if oldest != nil { - for i := *oldest; i <= chain.CurrentBlock().NumberU64(); i++ { + if tail != nil { + for i := *tail; i <= chain.CurrentBlock().NumberU64(); i++ { block := rawdb.ReadBlock(chain.db, rawdb.ReadCanonicalHash(chain.db, i), i) if block.Transactions().Len() == 0 { continue @@ -2174,20 +2174,19 @@ func TestTransactionIndices(t *testing.T) { } } } - for i := uint64(0); i < *oldest; i++ { + for i := uint64(0); i < *tail; 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()) + t.Logf("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) @@ -2217,54 +2216,47 @@ func TestTransactionIndices(t *testing.T) { chain.Stop() ancientDb.Close() - // Reconstruct an ancient db with inserted ancient blocks. + // Init block chain with external ancients, check all needed indices has been indexed. + limit := []uint64{0, 32, 64, 128} + for _, l := range limit { + ancientDb, err = rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), frdir, "") + if err != nil { + t.Fatalf("failed to create temp freezer db: %v", err) + } + gspec.MustCommit(ancientDb) + chain, err = NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, l) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + time.Sleep(50 * time.Millisecond) // Wait for indices initialisation + var tail uint64 + if l != 0 { + tail = uint64(128) - l + } + check(&tail, chain) + chain.Stop() + ancientDb.Close() + } + + // Reconstruct a block chain which only reserves HEAD-64 tx indices 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) + limit = []uint64{0, 64 /* drop stale */, 32 /* shorten history */, 64 /* extend history */, 0 /* restore all */} + tails := []uint64{0, 66 /* 130 - 64 */, 99 /* 131 - 32 */, 68 /* 132 - 64 */, 0} + for i, l := range limit { + chain, err = NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, l) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + chain.InsertChain(blocks2[i : i+1]) // Feed chain a higher block to trigger indices updater. + time.Sleep(50 * time.Millisecond) // Wait for indices initialisation + check(&tails[i], chain) + chain.Stop() } - 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 diff --git a/core/rawdb/accessors_chain.go b/core/rawdb/accessors_chain.go index 927f049880..1963728d28 100644 --- a/core/rawdb/accessors_chain.go +++ b/core/rawdb/accessors_chain.go @@ -191,6 +191,23 @@ func WriteTxIndexTail(db ethdb.KeyValueWriter, number uint64) { } } +// ReadFastTxLookupLimit retrieves the tx lookup limit used in fast sync. +func ReadFastTxLookupLimit(db ethdb.KeyValueReader) *uint64 { + data, _ := db.Get(fastTxLookupLimitKey) + if len(data) != 8 { + return nil + } + number := binary.BigEndian.Uint64(data) + return &number +} + +// WriteFastTxLookupLimit stores the txlookup limit used in fast sync into database. +func WriteFastTxLookupLimit(db ethdb.KeyValueWriter, number uint64) { + if err := db.Put(fastTxLookupLimitKey, encodeBlockNumber(number)); err != nil { + log.Crit("Failed to store txlookup limit for fast sync", "err", err) + } +} + // ReadHeaderRLP retrieves a block header in its raw RLP database encoding. func ReadHeaderRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue { data, _ := db.Ancient(freezerHeaderTable, number) @@ -578,40 +595,3 @@ func FindCommonAncestor(db ethdb.Reader, a, b *types.Header) *types.Header { } return a } - -// FindTxIndexTail 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 will be reindexed. -func FindTxIndexTail(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 094296c440..8c8affffd9 100644 --- a/core/rawdb/accessors_chain_test.go +++ b/core/rawdb/accessors_chain_test.go @@ -358,56 +358,3 @@ func checkReceiptsRLP(have, want types.Receipts) error { } return nil } - -func TestFindOldestIndexedBlock(t *testing.T) { - var cases = []struct { - oldest uint64 - height uint64 - nilBlocks map[uint64]bool - start uint64 - end uint64 - expect uint64 - expectNil bool - }{ - {11, 10, nil, 0, 10, 0, true}, // No block has been indexed - {0, 10, nil, 0, 10, 1, false}, // Genesis block doesn't have indices - {1, 10, nil, 0, 10, 1, false}, - {5, 10, nil, 0, 10, 5, false}, - {5, 10, nil, 0, 5, 5, false}, - {5, 10, nil, 0, 4, 0, true}, - {5, 10, nil, 5, 5, 5, false}, - {5, 10, nil, 10, 10, 10, false}, - {10, 10, nil, 0, 10, 10, false}, - {3, 10, map[uint64]bool{4: true, 6: true, 8: true}, 0, 10, 5, false}, - } - 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 block.NumberU64() >= c.oldest { - WriteTxLookupEntries(db, block) - } - } - res := FindTxIndexTail(db, c.start, c.end) - if c.expectNil && res != nil { - t.Fatalf("Case %d failed, oldest block mismatch, want nil, have %d", cid, *res) - } - if !c.expectNil && *res != c.expect { - t.Fatalf("Case %d failed, oldest block mismatch, want %d, have %d", cid, c.expect, *res) - } - } -} diff --git a/core/rawdb/chain_iterator.go b/core/rawdb/chain_iterator.go index ba955ac92e..cab1ea6f9c 100644 --- a/core/rawdb/chain_iterator.go +++ b/core/rawdb/chain_iterator.go @@ -35,9 +35,9 @@ type ( 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]. +// iterateCanonicalChain iterates the specified range canonical chain and then +// apply the given action callback. +// Note both for forward and backward iteration, 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 { @@ -56,7 +56,7 @@ func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, typ string if !reverse { number = int64(from - 1) } else { - number = int64(to + 1) + number = int64(to) } abort := make(chan struct{}) defer close(abort) @@ -73,7 +73,7 @@ func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, typ string } } else { n = atomic.AddInt64(&number, -1) - if n <= int64(from) { + if n < int64(from) { return } } @@ -102,7 +102,7 @@ func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, typ string if !reverse { next, first, last = int64(from), int64(from), int64(to) } else { - next, first, last = int64(to), int64(to), int64(from) + next, first, last = int64(to-1), int64(to-1), int64(from-1) } logFn := log.Debug if report { @@ -156,8 +156,7 @@ func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, typ string // 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. +// injects into the database the block hash->number mappings. func InitBlockIndexFromFreezer(db ethdb.Database) { // If we can't access the freezer or it's empty, abort frozen, err := db.Ancients() @@ -180,14 +179,12 @@ func InitBlockIndexFromFreezer(db ethdb.Database) { log.Info("Initialized chain from ancient data", "number", frozen-1, "hash", hash) } -// IndexTxLookup initializes txlookup indices of the specified range blocks into the database. +// 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. +// This function iterates canonical chain in reverse order, it has one main advantage: +// We can write tx index tail 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. @@ -199,7 +196,7 @@ func IndexTxLookup(db ethdb.Database, from uint64, to uint64) { // writeIndices injects txlookup indices into the database. writeIndices := func(batch ethdb.Batch, block *types.Block) { WriteTxLookupEntries(batch, block) - if block.NumberU64()%100000 == 0 { + if block.NumberU64() == to-1 || block.NumberU64()%10000 == 0 { WriteTxIndexTail(batch, block.NumberU64()) } } diff --git a/core/rawdb/schema.go b/core/rawdb/schema.go index 372b6d2312..c38260aa4c 100644 --- a/core/rawdb/schema.go +++ b/core/rawdb/schema.go @@ -44,6 +44,9 @@ var ( // txIndexTailKey tracks the oldest block whose transaction indices(txlookup) has been indexed. txIndexTailKey = []byte("TxIndexTail") + // fastTxLookupLimitKey tracks the tx indices limit during fast sync + fastTxLookupLimitKey = []byte("FSTxLookupLimit") + // 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 headerTDSuffix = []byte("t") // headerPrefix + num (uint64 big endian) + hash + headerTDSuffix -> td diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index fa0d3f2340..edd0eb4d95 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -200,7 +200,7 @@ type BlockChain interface { CurrentFastBlock() *types.Block // FastSyncCommitHead directly commits the head block to a certain entity. - FastSyncCommitHead(common.Hash, uint64, uint64) error + FastSyncCommitHead(common.Hash) error // InsertChain inserts a batch of blocks into the local chain. InsertChain(types.Blocks) (int, error) @@ -1721,11 +1721,7 @@ func (d *Downloader) commitPivotBlock(result *fetchResult) error { if _, err := d.blockchain.InsertReceiptChain([]*types.Block{block}, []types.Receipts{result.Receipts}, d.ancientLimit); err != nil { return err } - // Use origin block number + 1 as the first inserted block number - d.syncStatsLock.RLock() - from := d.syncStatsChainOrigin + 1 - d.syncStatsLock.RUnlock() - if err := d.blockchain.FastSyncCommitHead(block.Hash(), from, d.ancientLimit); err != nil { + if err := d.blockchain.FastSyncCommitHead(block.Hash()); err != nil { return err } atomic.StoreInt32(&d.committed, 1) diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index 7fec1925c4..b23043b1c0 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -218,7 +218,7 @@ func (dl *downloadTester) CurrentFastBlock() *types.Block { } // FastSyncCommitHead manually sets the head block to a given hash. -func (dl *downloadTester) FastSyncCommitHead(hash common.Hash, from uint64, ancient uint64) error { +func (dl *downloadTester) FastSyncCommitHead(hash common.Hash) error { // For now only check that the state trie is correct if block := dl.GetBlockByHash(hash); block != nil { _, err := trie.NewSecure(block.Root(), trie.NewDatabase(dl.stateDb)) diff --git a/eth/handler.go b/eth/handler.go index 2e4c887bda..05fa4077a5 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -73,6 +73,7 @@ type ProtocolManager struct { txpool txPool blockchain *core.BlockChain + chaindb ethdb.Database maxPeers int downloader *downloader.Downloader @@ -106,6 +107,7 @@ func NewProtocolManager(config *params.ChainConfig, checkpoint *params.TrustedCh eventMux: mux, txpool: txpool, blockchain: blockchain, + chaindb: chaindb, peers: newPeerSet(), whitelist: whitelist, newPeerCh: make(chan *peer), diff --git a/eth/sync.go b/eth/sync.go index 9e180ee200..12bef2bdf6 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -22,6 +22,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth/downloader" "github.com/ethereum/go-ethereum/log" @@ -186,6 +187,24 @@ func (pm *ProtocolManager) synchronise(peer *peer) { return } } + if mode == downloader.FastSync { + // Before launch the fast sync, we have to ensure user uses the same + // txlookup limit. + // The main concern here is: during the fast sync Geth won't index the + // block(generate tx indices) before the HEAD-limit. But if user changes + // the limit in the next fast sync(e.g. user kill Geth manually and + // restart) then it will be hard for Geth to figure out the oldest block + // has been indexed. So here for the user-experience wise, it's non-optimal + // that user can't change limit during the fast sync. If changed, Geth + // will just blindly use the original one. + limit := pm.blockchain.TxLookupLimit() + if stored := rawdb.ReadFastTxLookupLimit(pm.chaindb); stored == nil { + rawdb.WriteFastTxLookupLimit(pm.chaindb, limit) + } else if *stored != limit { + pm.blockchain.AdjustTxLookupLimit(*stored) + log.Warn("Update txLookup limit", "provided", limit, "updated", *stored) + } + } // Run the sync cycle, and disable fast sync if we've went past the pivot block if err := pm.downloader.Synchronise(peer.id, pHead, pTd, mode); err != nil { return