From 8639d929d1758db8f5c1b6142698ba3883c3b164 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 13 May 2019 12:08:55 +0300 Subject: [PATCH] eth/downloader: initialize fast sync bloom on startup --- cmd/geth/chaincmd.go | 12 ++++++++--- eth/downloader/downloader.go | 34 ++++++++++++++++++++----------- eth/downloader/downloader_test.go | 2 +- eth/downloader/statesync.go | 12 +++-------- eth/handler.go | 23 +++++++++++++-------- les/handler.go | 2 +- 6 files changed, 50 insertions(+), 35 deletions(-) diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index b69cfbcfad..582f0b768e 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -35,6 +35,7 @@ import ( "github.com/ethereum/go-ethereum/eth/downloader" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/trie" "gopkg.in/urfave/cli.v1" ) @@ -375,8 +376,13 @@ func copyDb(ctx *cli.Context) error { defer stack.Close() chain, chainDb := utils.MakeChain(ctx, stack) - syncmode := *utils.GlobalTextMarshaler(ctx, utils.SyncModeFlag.Name).(*downloader.SyncMode) - dl := downloader.New(syncmode, 0, chainDb, uint64(ctx.GlobalInt(utils.CacheFlag.Name)/2), new(event.TypeMux), chain, nil, nil) + syncMode := *utils.GlobalTextMarshaler(ctx, utils.SyncModeFlag.Name).(*downloader.SyncMode) + + var syncBloom *trie.SyncBloom + if syncMode == downloader.FastSync { + syncBloom = trie.NewSyncBloom(uint64(ctx.GlobalInt(utils.CacheFlag.Name)/2), chainDb) + } + dl := downloader.New(0, chainDb, syncBloom, new(event.TypeMux), chain, nil, nil) // Create a source peer to satisfy downloader requests from db, err := rawdb.NewLevelDBDatabase(ctx.Args().First(), ctx.GlobalInt(utils.CacheFlag.Name)/2, 256, "") @@ -395,7 +401,7 @@ func copyDb(ctx *cli.Context) error { start := time.Now() currentHeader := hc.CurrentHeader() - if err = dl.Synchronise("local", currentHeader.Hash(), hc.GetTd(currentHeader.Hash(), currentHeader.Number.Uint64()), syncmode); err != nil { + if err = dl.Synchronise("local", currentHeader.Hash(), hc.GetTd(currentHeader.Hash(), currentHeader.Number.Uint64()), syncMode); err != nil { return err } for dl.Synchronising() { diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index d3462754c6..0e19fe9e65 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -106,9 +106,8 @@ type Downloader struct { queue *queue // Scheduler for selecting the hashes to download peers *peerSet // Set of active peers from which download can proceed - stateDatabase ethdb.Database // Database to state sync into (and deduplicate via) - stateBloomSize uint64 // Number of bytes (in MB) to allocate for the bloom - stateBloom *trie.SyncBloom // Bloom filter for fast trie node existence checks + stateDB ethdb.Database // Database to state sync into (and deduplicate via) + stateBloom *trie.SyncBloom // Bloom filter for fast trie node existence checks rttEstimate uint64 // Round trip time to target for download requests rttConfidence uint64 // Confidence in the estimated RTT (unit: millionths to allow atomic ops) @@ -211,14 +210,13 @@ type BlockChain interface { } // New creates a new downloader to fetch hashes and blocks from remote peers. -func New(mode SyncMode, checkpoint uint64, stateDb ethdb.Database, stateBloomSize uint64, mux *event.TypeMux, chain BlockChain, lightchain LightChain, dropPeer peerDropFn) *Downloader { +func New(checkpoint uint64, stateDb ethdb.Database, stateBloom *trie.SyncBloom, mux *event.TypeMux, chain BlockChain, lightchain LightChain, dropPeer peerDropFn) *Downloader { if lightchain == nil { lightchain = chain } dl := &Downloader{ - mode: mode, - stateDatabase: stateDb, - stateBloomSize: stateBloomSize, + stateDB: stateDb, + stateBloom: stateBloom, mux: mux, checkpoint: checkpoint, queue: newQueue(), @@ -260,13 +258,15 @@ func (d *Downloader) Progress() ethereum.SyncProgress { defer d.syncStatsLock.RUnlock() current := uint64(0) - switch d.mode { - case FullSync: + switch { + case d.blockchain != nil && d.mode == FullSync: current = d.blockchain.CurrentBlock().NumberU64() - case FastSync: + case d.blockchain != nil && d.mode == FastSync: current = d.blockchain.CurrentFastBlock().NumberU64() - case LightSync: + case d.lightchain != nil: current = d.lightchain.CurrentHeader().Number.Uint64() + default: + log.Error("Unknown downloader chain/mode combo", "light", d.lightchain != nil, "full", d.blockchain != nil, "mode", d.mode) } return ethereum.SyncProgress{ StartingBlock: d.syncStatsChainOrigin, @@ -368,6 +368,12 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int, mode if atomic.CompareAndSwapInt32(&d.notified, 0, 1) { log.Info("Block synchronisation started") } + // If we are already full syncing, but have a fast-sync bloom filter laying + // around, make sure it does't use memory any more. This is a special case + // when the user attempts to fast sync a new empty network. + if mode == FullSync && d.stateBloom != nil { + d.stateBloom.Close() + } // Reset the queue, peer set and wake channels to clean any internal leftover state d.queue.Reset() d.peers.Reset() @@ -1677,7 +1683,11 @@ func (d *Downloader) commitPivotBlock(result *fetchResult) error { } atomic.StoreInt32(&d.committed, 1) - // If we had a bloom filter for the state sync, deallocate it now + // If we had a bloom filter for the state sync, deallocate it now. Note, we only + // deallocate internally, but keep the empty wrapper. This ensures that if we do + // a rollback after committing the pivot and restarting fast sync, we don't end + // up using a nil bloom. Empty bloom is fine, it just returns that it does not + // have the info we need, so reach down to the database instead. if d.stateBloom != nil { d.stateBloom.Close() } diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index 09248d5f79..ac7f48abdb 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -75,7 +75,7 @@ func newTester() *downloadTester { tester.stateDb = rawdb.NewMemoryDatabase() tester.stateDb.Put(testGenesis.Root().Bytes(), []byte{0x00}) - tester.downloader = New(FullSync, 0, tester.stateDb, 1, new(event.TypeMux), tester, nil, tester.dropPeer) + tester.downloader = New(0, tester.stateDb, trie.NewSyncBloom(1, tester.stateDb), new(event.TypeMux), tester, nil, tester.dropPeer) return tester } diff --git a/eth/downloader/statesync.go b/eth/downloader/statesync.go index e1cf6efc56..7e156f46e7 100644 --- a/eth/downloader/statesync.go +++ b/eth/downloader/statesync.go @@ -59,12 +59,6 @@ type stateSyncStats struct { // syncState starts downloading state with the given root hash. func (d *Downloader) syncState(root common.Hash) *stateSync { - // The downloader was requested to start a state sync. If the state sync bloom - // filter was not yet initialized, create it now. This lazy creation ensures we - // only allocate if really really really needed. - if d.stateBloom == nil { - d.stateBloom = trie.NewSyncBloom(d.stateBloomSize, d.stateDatabase) - } // Create the state sync s := newStateSync(d, root) select { @@ -246,7 +240,7 @@ type stateTask struct { func newStateSync(d *Downloader, root common.Hash) *stateSync { return &stateSync{ d: d, - sched: state.NewStateSync(root, d.stateDatabase, d.stateBloom), + sched: state.NewStateSync(root, d.stateDB, d.stateBloom), keccak: sha3.NewLegacyKeccak256(), tasks: make(map[common.Hash]*stateTask), deliver: make(chan *stateReq), @@ -342,7 +336,7 @@ func (s *stateSync) commit(force bool) error { return nil } start := time.Now() - b := s.d.stateDatabase.NewBatch() + b := s.d.stateDB.NewBatch() if written, err := s.sched.Commit(b); written == 0 || err != nil { return err } @@ -489,6 +483,6 @@ func (s *stateSync) updateStats(written, duplicate, unexpected int, duration tim log.Info("Imported new state entries", "count", written, "elapsed", common.PrettyDuration(duration), "processed", s.d.syncStatsState.processed, "pending", s.d.syncStatsState.pending, "retry", len(s.tasks), "duplicate", s.d.syncStatsState.duplicate, "unexpected", s.d.syncStatsState.unexpected) } if written > 0 { - rawdb.WriteFastTrieProgress(s.d.stateDatabase, s.d.syncStatsState.processed) + rawdb.WriteFastTrieProgress(s.d.stateDB, s.d.syncStatsState.processed) } } diff --git a/eth/handler.go b/eth/handler.go index f61c0f23e7..e200991d33 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -39,6 +39,7 @@ import ( "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rlp" + "github.com/ethereum/go-ethereum/trie" ) const ( @@ -120,12 +121,8 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne txsyncCh: make(chan *txsync), quitSync: make(chan struct{}), } - // Figure out whether to allow fast sync or not - if mode == downloader.FastSync && blockchain.CurrentBlock().NumberU64() > 0 { - log.Warn("Blockchain not empty, fast sync disabled") - mode = downloader.FullSync - } - if mode == downloader.FastSync { + // If fast sync was requested and our database is empty, grant it + if mode == downloader.FastSync && blockchain.CurrentBlock().NumberU64() == 0 { manager.fastSync = uint32(1) } // If we have trusted checkpoints, enforce them on the chain @@ -137,7 +134,8 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne manager.SubProtocols = make([]p2p.Protocol, 0, len(ProtocolVersions)) for i, version := range ProtocolVersions { // Skip protocol version if incompatible with the mode of operation - if mode == downloader.FastSync && version < eth63 { + // TODO(karalabe): hard-drop eth/62 from the code base + if version < eth63 { continue } // Compatible; initialise the sub-protocol @@ -171,9 +169,16 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne if len(manager.SubProtocols) == 0 { return nil, errIncompatibleConfig } - // Construct the different synchronisation mechanisms - manager.downloader = downloader.New(mode, manager.checkpointNumber, chaindb, uint64(cacheLimit), manager.eventMux, blockchain, nil, manager.removePeer) + // Construct the downloader (long sync) and its backing state bloom if fast + // sync is requested. The downloader is responsible for deallocating the state + // bloom when it's done. + var stateBloom *trie.SyncBloom + if atomic.LoadUint32(&manager.fastSync) == 1 { + stateBloom = trie.NewSyncBloom(uint64(cacheLimit), chaindb) + } + manager.downloader = downloader.New(manager.checkpointNumber, chaindb, stateBloom, manager.eventMux, blockchain, nil, manager.removePeer) + // Construct the fetcher (short sync) validator := func(header *types.Header) error { return engine.VerifyHeader(blockchain, header, true) } diff --git a/les/handler.go b/les/handler.go index 9339ab7a27..d51e0e7716 100644 --- a/les/handler.go +++ b/les/handler.go @@ -179,7 +179,7 @@ func NewProtocolManager( if cht, ok := params.TrustedCheckpoints[blockchain.Genesis().Hash()]; ok { checkpoint = (cht.SectionIndex+1)*params.CHTFrequency - 1 } - manager.downloader = downloader.New(downloader.LightSync, checkpoint, chainDb, 0, manager.eventMux, nil, blockchain, removePeer) + manager.downloader = downloader.New(checkpoint, chainDb, nil, manager.eventMux, nil, blockchain, removePeer) manager.peers.notify((*downloaderPeerNotify)(manager)) manager.fetcher = newLightFetcher(manager) }