eth, eth/downloader: moved peer selection to protocol handler

1 files changed, 61 insertions, 4 deletions

diff --git a/eth/handler.go b/eth/handler.go
index a634b5bfd..a1b03f57c 100644
--- a/eth/handler.go
+++ b/eth/handler.go
@@ -39,6 +39,7 @@ import (
 	"math"
 	"math/big"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
@@ -51,6 +52,11 @@ import (
 	"github.com/ethereum/go-ethereum/rlp"
 )
 
+const (
+	peerCountTimeout    = 12 * time.Second // Amount of time it takes for the peer handler to ignore minDesiredPeerCount
+	minDesiredPeerCount = 5                // Amount of peers desired to start syncing
+)
+
 func errResp(code errCode, format string, v ...interface{}) error {
 	return fmt.Errorf("%v - %v", code, fmt.Sprintf(format, v...))
 }
@@ -82,6 +88,9 @@ type ProtocolManager struct {
 	eventMux      *event.TypeMux
 	txSub         event.Subscription
 	minedBlockSub event.Subscription
+
+	newPeerCh chan *peer
+	quit      chan struct{}
 }
 
 // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable
@@ -93,7 +102,10 @@ func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpo
 		chainman:   chainman,
 		downloader: downloader,
 		peers:      make(map[string]*peer),
+		newPeerCh:  make(chan *peer, 1),
+		quit:       make(chan struct{}),
 	}
+	go manager.peerHandler()
 
 	manager.SubProtocol = p2p.Protocol{
 		Name:    "eth",
@@ -101,16 +113,61 @@ func NewProtocolManager(protocolVersion, networkId int, mux *event.TypeMux, txpo
 		Length:  ProtocolLength,
 		Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 			peer := manager.newPeer(protocolVersion, networkId, p, rw)
-			err := manager.handle(peer)
-			//glog.V(logger.Detail).Infof("[%s]: %v\n", peer.id, err)
 
-			return err
+			manager.newPeerCh <- peer
+
+			return manager.handle(peer)
 		},
 	}
 
 	return manager
 }
 
+func (pm *ProtocolManager) peerHandler() {
+	// itimer is used to determine when to start ignoring `minDesiredPeerCount`
+	itimer := time.NewTimer(peerCountTimeout)
+out:
+	for {
+		select {
+		case <-pm.newPeerCh:
+			// Meet the `minDesiredPeerCount` before we select our best peer
+			if len(pm.peers) < minDesiredPeerCount {
+				break
+			}
+			itimer.Stop()
+
+			// Find the best peer
+			peer := getBestPeer(pm.peers)
+			if peer == nil {
+				glog.V(logger.Debug).Infoln("Sync attempt cancelled. No peers available")
+				return
+			}
+			go pm.synchronise(peer)
+		case <-itimer.C:
+			// The timer will make sure that the downloader keeps an active state
+			// in which it attempts to always check the network for highest td peers
+			// Either select the peer or restart the timer if no peers could
+			// be selected.
+			if peer := getBestPeer(pm.peers); peer != nil {
+				go pm.synchronise(peer)
+			} else {
+				itimer.Reset(5 * time.Second)
+			}
+		case <-pm.quit:
+			break out
+		}
+	}
+}
+
+func (pm *ProtocolManager) synchronise(peer *peer) {
+	// Get the hashes from the peer (synchronously)
+	_, err := pm.downloader.Synchronise(peer.id, peer.recentHash)
+	if err != nil {
+		// handle error
+		glog.V(logger.Debug).Infoln("error downloading:", err)
+	}
+}
+
 func (pm *ProtocolManager) Start() {
 	// broadcast transactions
 	pm.txSub = pm.eventMux.Subscribe(core.TxPreEvent{})
@@ -141,7 +198,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
 	pm.peers[p.id] = p
 	pm.pmu.Unlock()
 
-	pm.downloader.RegisterPeer(p.id, p.td, p.currentHash, p.requestHashes, p.requestBlocks)
+	pm.downloader.RegisterPeer(p.id, p.td, p.recentHash, p.requestHashes, p.requestBlocks)
 	defer func() {
 		pm.pmu.Lock()
 		defer pm.pmu.Unlock()