eth: rewrite sync controller

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
-	txsyncCh  chan *txsync
+	quitSync chan struct{}
-	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)
+			return pm.runPeer(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
-			}
 		},
 		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)
-
+			pm.chainSync.handlePeerEvent(p)
-			// 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)
-			}
 		}
 
 	case msg.Code == NewPooledTransactionHashesMsg && p.version >= eth65:

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() {
+	go func() { errc <- pm.runPeer(peer) }()
-		select {
-		case pm.newPeerCh <- peer:
-			errc <- pm.handle(peer)
-		case <-pm.quitSync:
-			errc <- p2p.DiscQuitting
-		}
-	}()
 	tp := &testPeer{app: app, net: net, peer: peer}
+
 	// Execute any implicitly requested handshakes and return
 	if shake {
 		var (

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)

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.
+// chainSyncer coordinates blockchain sync components.
-func (pm *ProtocolManager) syncer() {
+type chainSyncer struct {
-	defer pm.wg.Done()
+	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()
+type chainSyncOp struct {
-	pm.txFetcher.Start()
+	mode downloader.SyncMode
-	defer pm.blockFetcher.Stop()
+	peer *peer
-	defer pm.txFetcher.Stop()
+	td   *big.Int
-	defer pm.downloader.Terminate()
+	head common.Hash
+}
 
-	for pm.syncTriggerWait() {
+func newChainSyncer(pm *ProtocolManager) *chainSyncer {
-		pm.synchronise(pm.peers.BestPeer())
+	return &chainSyncer{
+		pm:          pm,
+		peerEventCh: make(chan struct{}),
 	}
 }
 
-// syncTriggerWait waits for sync start conditions to be met.
+// handlePeerEvent notifies the syncer about a change in the peer set.
-func (pm *ProtocolManager) syncTriggerWait() bool {
+// This is called for new peers and every time a peer announces a new
-	force := time.NewTimer(forceSyncCycle)
+// chain head.
-	defer force.Stop()
+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 {
+		if op := cs.nextSyncOp(); op != nil {
-		case <-pm.quitSync:
+			log.Trace("Starting chain sync", "mode", op.mode, "peercount", cs.pm.peers.Len(), "id", op.peer.id)
-			return false
+			cs.startSync(op)
-		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
-			}
 		}
-	}
-}
 
-// synchronise tries to sync up our local block chain with a remote peer.
+		select {
-func (pm *ProtocolManager) synchronise(peer *peer) {
+		case <-cs.peerEventCh:
-	// Short circuit if no peers are available
+			// Peer information changed, recheck.
-	if peer == nil {
+		case <-cs.doneCh:
-		return
+			cs.doneCh = nil
-	}
+			cs.force.Reset(forceSyncCycle)
-	// Make sure the peer's TD is higher than our own
+			cs.forced = false
-	currentHeader := pm.blockchain.CurrentHeader()
+		case <-cs.force.C:
-	td := pm.blockchain.GetTd(currentHeader.Hash(), currentHeader.Number.Uint64())
+			cs.forced = true
 
-	pHead, pTd := peer.Head()
+		case <-cs.pm.quitSync:
-	if pTd.Cmp(td) <= 0 {
+			if cs.doneCh != nil {
-		return
+				<-cs.doneCh
-	}
+			}
-	// 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 {
 			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
 }

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 {