eth/connmanager: monitor sync status

Signed-off-by: Csaba Kiraly <csaba.kiraly@gmail.com>

eth/backend.go

@@ -405,7 +405,7 @@ func (s *Ethereum) Start() error {
 	s.handler.Start(s.p2pServer.MaxPeers)
 
 	// Start the connection manager
-	s.connman.Start(s.p2pServer, s.Synced)
+	s.connman.Start(s.p2pServer, func() bool { return !s.Synced() })
 
 	// start log indexer
 	s.filterMaps.Start()

eth/connmanager.go

@@ -40,15 +40,18 @@ const (
 	// dropping when no more peers can be added. Larger numbers result in more
 	// aggressive drop behavior.
 	peerDropThreshold = 0
+	// Sync status poll interval (no need to be too reactive here)
+	syncCheckInterval = 60 * time.Second
 )
 
 // connManager monitors the state of the peer pool and makes changes as follows:
-//   - if the peer count is close to the limit, it drops peers randomly every
+//   - during sync the Downloader handles peer connections co connManager is disabled
-//     peerDropInterval to make space for new peers
+//   - if not syncing and the peer count is close to the limit, it drops peers
+//     randomly every peerDropInterval to make space for new peers
 type connManager struct {
 	connmanConfig
-	peersFunc getPeersFunc
+	peersFunc   getPeersFunc
-	syncFunc  getSyncFunc
+	syncingFunc getSyncingFunc
 
 	// the peerDrop timer introduces churn if we are close to limit capacity
 	peerDropTimer *mclock.Alarm
@@ -62,8 +65,9 @@ type connManager struct {
 // callback type to get the list of connected peers.
 type getPeersFunc func() []*p2p.Peer
 
-// callback type to get sync status.
+// Callback type to get syncing status.
-type getSyncFunc func() bool
+// Returns true while syncing, false when synced.
+type getSyncingFunc func() bool
 
 type connmanConfig struct {
 	maxDialPeers int // maximum number of dialed peers
@@ -100,15 +104,16 @@ func newConnManager(config *connmanConfig) *connManager {
 	return cm
 }
 
-func (cm *connManager) Start(srv *p2p.Server, syncFunc getSyncFunc) {
+// Start the connection manager.
+func (cm *connManager) Start(srv *p2p.Server, syncingFunc getSyncingFunc) {
 	cm.wg.Add(1)
 	cm.peersFunc = srv.Peers
-	cm.syncFunc = syncFunc
+	cm.syncingFunc = syncingFunc
 	cm.sub = srv.SubscribeEvents(cm.peerEventCh)
 	go cm.loop()
 }
 
-// stop the connection manager.
+// Stop the connection manager.
 func (cm *connManager) Stop() {
 	cm.sub.Unsubscribe()
 	cm.peerDropTimer.Stop()
@@ -149,32 +154,59 @@ func (cm *connManager) dropRandomPeer() bool {
 	return false
 }
 
+// updatePeerDropTimer checks and starts/stops the timer for peer drop.
+func (cm *connManager) updatePeerDropTimer(syncing bool) {
+
+	numpeers, out, in := cm.numPeers()
+	cm.log.Trace("ConnManager status", "syncing", syncing,
+		"peers", numpeers, "out", out, "in", in, "maxout", cm.maxDialPeers)
+
+	if !syncing {
+		// If a drop was already scheduled, Schedule does nothing.
+		if cm.maxDialPeers-cm.numDialPeers() <= peerDropThreshold {
+			cm.peerDropTimer.Schedule(cm.clock.Now().Add(peerDropInterval))
+		} else {
+			cm.peerDropTimer.Stop()
+		}
+	} else {
+		// Downloader is managing connections while syncing.
+		cm.peerDropTimer.Stop()
+	}
+}
+
 // loop is the main loop of the connection manager.
 func (cm *connManager) loop() {
+
 	defer cm.wg.Done()
 
+	// Set up periodic timer to pull syncing status.
+	// We could get syncing status in a few ways:
+	// - poll the sync status (we use this for now)
+	// - subscribe to Downloader.mux
+	// - subscribe to DownloaderAPI (which itself polls the sync status)
+	syncing := cm.syncingFunc()
+	cm.log.Trace("Sync status", "syncing", syncing)
+	syncCheckTimer := mclock.NewAlarm(cm.connmanConfig.clock)
+	syncCheckTimer.Schedule(cm.clock.Now().Add(syncCheckInterval))
+	defer syncCheckTimer.Stop()
+
 	for {
 		select {
-		case ev := <-cm.peerEventCh:
+		case <-syncCheckTimer.C():
-			switch ev.Type {
+			// Update info about syncing status, and rearm the timer.
-			case p2p.PeerEventTypeAdd:
+			syncingNew := cm.syncingFunc()
-				// check and start timer for peer drop
+			if syncing != syncingNew {
-				// If a drop was already scheduled, Schedule does nothing.
+				// Syncing status changed, we might need to update the timer.
-				numpeers, out, in := cm.numPeers()
+				cm.log.Trace("Sync status changed", "syncing", syncingNew)
-				cm.log.Trace("addPeerCh", "peers", numpeers, "out", out, "in", in, "maxout", cm.maxDialPeers)
+				syncing = syncingNew
-				if cm.maxDialPeers-cm.numDialPeers() <= peerDropThreshold {
+				cm.updatePeerDropTimer(syncing)
-					cm.peerDropTimer.Schedule(cm.clock.Now().Add(peerDropInterval))
+			}
-				}
+			syncCheckTimer.Schedule(cm.clock.Now().Add(syncCheckInterval))
-
+		case ev := <-cm.peerEventCh:
-			case p2p.PeerEventTypeDrop:
+			if ev.Type == p2p.PeerEventTypeAdd || ev.Type == p2p.PeerEventTypeDrop {
-				// check and stop timer for peer drop
+				// Number of peers changed, we might need to start the timer.
-				numpeers, out, in := cm.numPeers()
+				cm.updatePeerDropTimer(syncing)
-				cm.log.Trace("remPeerCh", "peers", numpeers, "out", out, "in", in, "maxout", cm.maxDialPeers)
-				if cm.maxDialPeers-cm.numDialPeers() > peerDropThreshold {
-					cm.peerDropTimer.Stop()
-				}
 			}
-
 		case <-cm.peerDropTimer.C():
 			cm.dropRandomPeer()
 		case <-cm.shutdownCh: