aboutsummaryrefslogtreecommitdiffstats
path: root/eth/handler.go
diff options
context:
space:
mode:
Diffstat (limited to 'eth/handler.go')
-rw-r--r--eth/handler.go65
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()