diff --git a/eth/handler.go b/eth/handler.go index 9d2dce6e90..eb1943031d 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -87,9 +87,10 @@ type ProtocolManager struct { whitelist map[uint64]common.Hash // channels for fetcher, syncer, txsyncLoop - newPeerCh chan *peer - txsyncCh chan *txsync - quitSync chan struct{} + txsyncCh chan *txsync + quitSync chan struct{} + + chainSync *chainSyncer wg sync.WaitGroup peerWG sync.WaitGroup @@ -109,10 +110,10 @@ func NewProtocolManager(config *params.ChainConfig, checkpoint *params.TrustedCh blockchain: blockchain, peers: newPeerSet(), whitelist: whitelist, - newPeerCh: make(chan *peer), txsyncCh: make(chan *txsync), quitSync: make(chan struct{}), } + if mode == downloader.FullSync { // The database seems empty as the current block is the genesis. Yet the fast // block is ahead, so fast sync was enabled for this node at a certain point. @@ -136,6 +137,7 @@ func NewProtocolManager(config *params.ChainConfig, checkpoint *params.TrustedCh manager.fastSync = uint32(1) } } + // If we have trusted checkpoints, enforce them on the chain if checkpoint != nil { manager.checkpointNumber = (checkpoint.SectionIndex+1)*params.CHTFrequency - 1 @@ -195,6 +197,8 @@ func NewProtocolManager(config *params.ChainConfig, checkpoint *params.TrustedCh } manager.txFetcher = fetcher.NewTxFetcher(txpool.Has, txpool.AddRemotes, fetchTx) + manager.chainSync = newChainSyncer(manager) + return manager, nil } @@ -209,15 +213,7 @@ func (pm *ProtocolManager) makeProtocol(version uint) p2p.Protocol { Version: version, Length: length, Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - peer := pm.newPeer(int(version), p, rw, pm.txpool.Get) - select { - case pm.newPeerCh <- peer: - pm.peerWG.Add(1) - defer pm.peerWG.Done() - return pm.handle(peer) - case <-pm.quitSync: - return p2p.DiscQuitting - } + return pm.runPeer(pm.newPeer(int(version), p, rw, pm.txpool.Get)) }, NodeInfo: func() interface{} { return pm.NodeInfo() @@ -268,7 +264,7 @@ func (pm *ProtocolManager) Start(maxPeers int) { // start sync handlers pm.wg.Add(2) - go pm.syncer() + go pm.chainSync.loop() go pm.txsyncLoop64() // TODO(karalabe): Legacy initial tx echange, drop with eth/64. } @@ -276,7 +272,7 @@ func (pm *ProtocolManager) Stop() { pm.txsSub.Unsubscribe() // quits txBroadcastLoop pm.minedBlockSub.Unsubscribe() // quits blockBroadcastLoop - // Quit syncer and txsync64. + // Quit chainSync and txsync64. // After this is done, no new peers will be accepted. close(pm.quitSync) pm.downloader.Cancel() @@ -296,6 +292,15 @@ func (pm *ProtocolManager) newPeer(pv int, p *p2p.Peer, rw p2p.MsgReadWriter, ge return newPeer(pv, p, rw, getPooledTx) } +func (pm *ProtocolManager) runPeer(p *peer) error { + if !pm.chainSync.handlePeerEvent(p) { + return p2p.DiscQuitting + } + pm.peerWG.Add(1) + defer pm.peerWG.Done() + return pm.handle(p) +} + // handle is the callback invoked to manage the life cycle of an eth peer. When // this function terminates, the peer is disconnected. func (pm *ProtocolManager) handle(p *peer) error { @@ -317,6 +322,7 @@ func (pm *ProtocolManager) handle(p *peer) error { p.Log().Debug("Ethereum handshake failed", "err", err) return err } + // Register the peer locally if err := pm.peers.Register(p); err != nil { p.Log().Error("Ethereum peer registration failed", "err", err) @@ -717,14 +723,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { // Update the peer's total difficulty if better than the previous if _, td := p.Head(); trueTD.Cmp(td) > 0 { p.SetHead(trueHead, trueTD) - - // Schedule a sync if above ours. Note, this will not fire a sync for a gap of - // a single block (as the true TD is below the propagated block), however this - // scenario should easily be covered by the fetcher. - currentHeader := pm.blockchain.CurrentHeader() - if trueTD.Cmp(pm.blockchain.GetTd(currentHeader.Hash(), currentHeader.Number.Uint64())) > 0 { - go pm.synchronise(p) - } + pm.chainSync.handlePeerEvent(p) } case msg.Code == NewPooledTransactionHashesMsg && p.version >= eth65: diff --git a/eth/helper_test.go b/eth/helper_test.go index bec37e16cb..3338af71d5 100644 --- a/eth/helper_test.go +++ b/eth/helper_test.go @@ -170,23 +170,14 @@ func newTestPeer(name string, version int, pm *ProtocolManager, shake bool) (*te // Create a message pipe to communicate through app, net := p2p.MsgPipe() - // Generate a random id and create the peer + // Start the peer on a new thread var id enode.ID rand.Read(id[:]) - peer := pm.newPeer(version, p2p.NewPeer(id, name, nil), net, pm.txpool.Get) - - // Start the peer on a new thread errc := make(chan error, 1) - go func() { - select { - case pm.newPeerCh <- peer: - errc <- pm.handle(peer) - case <-pm.quitSync: - errc <- p2p.DiscQuitting - } - }() + go func() { errc <- pm.runPeer(peer) }() tp := &testPeer{app: app, net: net, peer: peer} + // Execute any implicitly requested handshakes and return if shake { var ( diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 4bbfe9bd3c..a313e4e6cf 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -385,7 +385,7 @@ func testSyncTransaction(t *testing.T, propagtion bool) { go pmFetcher.handle(pmFetcher.newPeer(65, p2p.NewPeer(enode.ID{}, "fetcher", nil), io1, pmFetcher.txpool.Get)) time.Sleep(250 * time.Millisecond) - pmFetcher.synchronise(pmFetcher.peers.BestPeer()) + pmFetcher.doSync(peerToSyncOp(downloader.FullSync, pmFetcher.peers.BestPeer())) atomic.StoreUint32(&pmFetcher.acceptTxs, 1) newTxs := make(chan core.NewTxsEvent, 1024) diff --git a/eth/sync.go b/eth/sync.go index 72f9960e83..6e77930921 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -17,6 +17,7 @@ package eth import ( + "math/big" "math/rand" "sync/atomic" "time" @@ -30,7 +31,7 @@ import ( const ( forceSyncCycle = 10 * time.Second // Time interval to force syncs, even if few peers are available - minDesiredPeerCount = 5 // Amount of peers desired to start syncing + defaultMinSyncPeers = 5 // Amount of peers desired to start syncing // This is the target size for the packs of transactions sent by txsyncLoop64. // A pack can get larger than this if a single transactions exceeds this size. @@ -152,79 +153,146 @@ func (pm *ProtocolManager) txsyncLoop64() { } } -// syncer runs in its own goroutine and coordinates sync-related components. -func (pm *ProtocolManager) syncer() { - defer pm.wg.Done() +// chainSyncer coordinates blockchain sync components. +type chainSyncer struct { + pm *ProtocolManager + force *time.Timer + forced bool // true when force timer fired + peerEventCh chan struct{} + doneCh chan error // non-nil when sync is running +} - pm.blockFetcher.Start() - pm.txFetcher.Start() - defer pm.blockFetcher.Stop() - defer pm.txFetcher.Stop() - defer pm.downloader.Terminate() +type chainSyncOp struct { + mode downloader.SyncMode + peer *peer + td *big.Int + head common.Hash +} - for pm.syncTriggerWait() { - pm.synchronise(pm.peers.BestPeer()) +func newChainSyncer(pm *ProtocolManager) *chainSyncer { + return &chainSyncer{ + pm: pm, + peerEventCh: make(chan struct{}), } } -// syncTriggerWait waits for sync start conditions to be met. -func (pm *ProtocolManager) syncTriggerWait() bool { - force := time.NewTimer(forceSyncCycle) - defer force.Stop() +// handlePeerEvent notifies the syncer about a change in the peer set. +// This is called for new peers and every time a peer announces a new +// chain head. +func (cs *chainSyncer) handlePeerEvent(p *peer) bool { + select { + case cs.peerEventCh <- struct{}{}: + return true + case <-cs.pm.quitSync: + return false + } +} + +// loop runs in its own goroutine and launches the sync when necessary. +func (cs *chainSyncer) loop() { + defer cs.pm.wg.Done() + + cs.pm.blockFetcher.Start() + cs.pm.txFetcher.Start() + defer cs.pm.blockFetcher.Stop() + defer cs.pm.txFetcher.Stop() + defer cs.pm.downloader.Terminate() + + // The force timer lowers the peer count threshold down to one when it fires. + // This ensures we'll always start sync even if there aren't enough peers. + cs.force = time.NewTimer(forceSyncCycle) + defer cs.force.Stop() + for { - select { - case <-pm.quitSync: - return false - default: - if pm.peers.Len() >= minDesiredPeerCount { - return true - } - select { - case <-pm.newPeerCh: - // Go round and check the peer count again. - case <-force.C: - return true - case <-pm.quitSync: - return false - } + if op := cs.nextSyncOp(); op != nil { + log.Trace("Starting chain sync", "mode", op.mode, "peercount", cs.pm.peers.Len(), "id", op.peer.id) + cs.startSync(op) } - } -} -// synchronise tries to sync up our local block chain with a remote peer. -func (pm *ProtocolManager) synchronise(peer *peer) { - // Short circuit if no peers are available - if peer == nil { - return - } - // Make sure the peer's TD is higher than our own - currentHeader := pm.blockchain.CurrentHeader() - td := pm.blockchain.GetTd(currentHeader.Hash(), currentHeader.Number.Uint64()) + select { + case <-cs.peerEventCh: + // Peer information changed, recheck. + case <-cs.doneCh: + cs.doneCh = nil + cs.force.Reset(forceSyncCycle) + cs.forced = false + case <-cs.force.C: + cs.forced = true - pHead, pTd := peer.Head() - if pTd.Cmp(td) <= 0 { - return - } - // Otherwise try to sync with the downloader - mode := downloader.FullSync - if atomic.LoadUint32(&pm.fastSync) == 1 { - // Fast sync was explicitly requested, and explicitly granted - mode = downloader.FastSync - } - if mode == downloader.FastSync { - // Make sure the peer's total difficulty we are synchronizing is higher. - if pm.blockchain.GetTdByHash(pm.blockchain.CurrentFastBlock().Hash()).Cmp(pTd) >= 0 { + case <-cs.pm.quitSync: + if cs.doneCh != nil { + <-cs.doneCh + } return } } - // 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 +} + +// nextSyncOp determines whether sync is required at this time. +func (cs *chainSyncer) nextSyncOp() *chainSyncOp { + if cs.doneCh != nil { + return nil // Sync already running. + } + + // Ensure we're at mininum peer count. + minPeers := defaultMinSyncPeers + if cs.forced { + minPeers = 1 + } else if minPeers > cs.pm.maxPeers { + minPeers = cs.pm.maxPeers + } + if cs.pm.peers.Len() < minPeers { + return nil + } + + // We have enough peers, check TD. + peer := cs.pm.peers.BestPeer() + if peer == nil { + return nil + } + mode, ourTD := cs.modeAndLocalHead() + op := peerToSyncOp(mode, peer) + if op.td.Cmp(ourTD) <= 0 { + return nil // We're in sync. + } + return op +} + +func peerToSyncOp(mode downloader.SyncMode, p *peer) *chainSyncOp { + peerHead, peerTD := p.Head() + return &chainSyncOp{mode: mode, peer: p, td: peerTD, head: peerHead} +} + +func (cs *chainSyncer) modeAndLocalHead() (downloader.SyncMode, *big.Int) { + if atomic.LoadUint32(&cs.pm.fastSync) == 1 { + block := cs.pm.blockchain.CurrentFastBlock() + td := cs.pm.blockchain.GetTdByHash(block.Hash()) + return downloader.FastSync, td + } else { + head := cs.pm.blockchain.CurrentHeader() + td := cs.pm.blockchain.GetTd(head.Hash(), head.Number.Uint64()) + return downloader.FullSync, td + } +} + +// startSync launches doSync in a new goroutine. +func (cs *chainSyncer) startSync(op *chainSyncOp) { + cs.doneCh = make(chan error, 1) + go func() { cs.doneCh <- cs.pm.doSync(op) }() +} + +// doSync synchronizes the local blockchain with a remote peer. +func (pm *ProtocolManager) doSync(op *chainSyncOp) error { + // Run the sync cycle, and disable fast sync if we're past the pivot block + err := pm.downloader.Synchronise(op.peer.id, op.head, op.td, op.mode) + if err != nil { + return err } if atomic.LoadUint32(&pm.fastSync) == 1 { log.Info("Fast sync complete, auto disabling") atomic.StoreUint32(&pm.fastSync, 0) } + // If we've successfully finished a sync cycle and passed any required checkpoint, // enable accepting transactions from the network. head := pm.blockchain.CurrentBlock() @@ -235,6 +303,7 @@ func (pm *ProtocolManager) synchronise(peer *peer) { atomic.StoreUint32(&pm.acceptTxs, 1) } } + if head.NumberU64() > 0 { // We've completed a sync cycle, notify all peers of new state. This path is // essential in star-topology networks where a gateway node needs to notify @@ -244,4 +313,6 @@ func (pm *ProtocolManager) synchronise(peer *peer) { // more reliably update peers or the local TD state. go pm.BroadcastBlock(head, false) } + + return nil } diff --git a/eth/sync_test.go b/eth/sync_test.go index d02bc57108..ac1e5fad1b 100644 --- a/eth/sync_test.go +++ b/eth/sync_test.go @@ -33,6 +33,8 @@ func TestFastSyncDisabling65(t *testing.T) { testFastSyncDisabling(t, 65) } // Tests that fast sync gets disabled as soon as a real block is successfully // imported into the blockchain. func testFastSyncDisabling(t *testing.T, protocol int) { + t.Parallel() + // Create a pristine protocol manager, check that fast sync is left enabled pmEmpty, _ := newTestProtocolManagerMust(t, downloader.FastSync, 0, nil, nil) if atomic.LoadUint32(&pmEmpty.fastSync) == 0 { @@ -43,14 +45,17 @@ func testFastSyncDisabling(t *testing.T, protocol int) { if atomic.LoadUint32(&pmFull.fastSync) == 1 { t.Fatalf("fast sync not disabled on non-empty blockchain") } + // Sync up the two peers io1, io2 := p2p.MsgPipe() - go pmFull.handle(pmFull.newPeer(protocol, p2p.NewPeer(enode.ID{}, "empty", nil), io2, pmFull.txpool.Get)) go pmEmpty.handle(pmEmpty.newPeer(protocol, p2p.NewPeer(enode.ID{}, "full", nil), io1, pmEmpty.txpool.Get)) time.Sleep(250 * time.Millisecond) - pmEmpty.synchronise(pmEmpty.peers.BestPeer()) + op := peerToSyncOp(downloader.FastSync, pmEmpty.peers.BestPeer()) + if err := pmEmpty.doSync(op); err != nil { + t.Fatal("sync failed:", err) + } // Check that fast sync was disabled if atomic.LoadUint32(&pmEmpty.fastSync) == 1 {