diff options
Diffstat (limited to 'eth')
| -rw-r--r-- | eth/downloader/downloader.go | 102 | ||||
| -rw-r--r-- | eth/downloader/peer.go | 20 |
2 files changed, 117 insertions, 5 deletions
diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 1707e1395..5f9d9ed74 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -1,6 +1,7 @@ package downloader import ( + "errors" "math" "math/big" "sync" @@ -20,6 +21,12 @@ const ( minDesiredPeerCount = 3 // Amount of peers desired to start syncing ) +var ( + errLowTd = errors.New("peer's TD is too low") + errBusy = errors.New("busy") + errUnknownPeer = errors.New("peer's unknown or unhealthy") +) + type hashCheckFn func(common.Hash) bool type chainInsertFn func(types.Blocks) error type hashIterFn func() (common.Hash, error) @@ -82,18 +89,19 @@ func (d *Downloader) RegisterPeer(id string, td *big.Int, hash common.Hash, getH d.mu.Lock() defer d.mu.Unlock() - glog.V(logger.Detail).Infoln("Register peer", id) + glog.V(logger.Detail).Infoln("Register peer", id, "TD =", td) // Create a new peer and add it to the list of known peers peer := newPeer(id, td, hash, getHashes, getBlocks) // add peer to our peer set d.peers[id] = peer // broadcast new peer - d.newPeerCh <- peer + //d.newPeerCh <- peer return nil } +// UnregisterPeer unregister's a peer. This will prevent any action from the specified peer. func (d *Downloader) UnregisterPeer(id string) { d.mu.Lock() defer d.mu.Unlock() @@ -103,6 +111,73 @@ func (d *Downloader) UnregisterPeer(id string) { delete(d.peers, id) } +// SynchroniseWithPeer will select the peer and use it for synchronising. If an empty string is given +// it will use the best peer possible and synchronise if it's TD is higher than our own. If any of the +// checks fail an error will be returned. This method is synchronous +func (d *Downloader) SynchroniseWithPeer(id string) (types.Blocks, error) { + // Check if we're busy + if d.isFetchingHashes() || d.isDownloadingBlocks() || d.isProcessing() { + return nil, errBusy + } + + // Attempt to select a peer. This can either be nothing, which returns, best peer + // or selected peer. If no peer could be found an error will be returned + var p *peer + if len(id) == 0 { + p = d.peers[id] + if p == nil { + return nil, errUnknownPeer + } + } else { + p = d.peers.bestPeer() + } + + // Make sure our td is lower than the peer's td + if p.td.Cmp(d.currentTd()) <= 0 || d.hasBlock(p.recentHash) { + return nil, errLowTd + } + + // Get the hash from the peer and initiate the downloading progress. + err := d.getFromPeer(p, p.recentHash, false) + if err != nil { + return nil, err + } + + return d.queue.blocks, nil +} + +// Synchronise will synchronise using the best peer. +func (d *Downloader) Synchronise() (types.Blocks, error) { + return d.SynchroniseWithPeer("") +} + +func (d *Downloader) getFromPeer(p *peer, hash common.Hash, ignoreInitial bool) error { + glog.V(logger.Detail).Infoln("Synchronising with the network using:", p.id) + // Start the fetcher. This will block the update entirely + // interupts need to be send to the appropriate channels + // respectively. + if err := d.startFetchingHashes(p, hash, ignoreInitial); err != nil { + // handle error + glog.V(logger.Debug).Infoln("Error fetching hashes:", err) + // XXX Reset + return err + } + + // Start fetching blocks in paralel. The strategy is simple + // take any available peers, seserve a chunk for each peer available, + // let the peer deliver the chunkn and periodically check if a peer + // has timedout. When done downloading, process blocks. + if err := d.startFetchingBlocks(p); err != nil { + glog.V(logger.Debug).Infoln("Error downloading blocks:", err) + // XXX reset + return err + } + + glog.V(logger.Detail).Infoln("Sync completed") + + return nil +} + func (d *Downloader) peerHandler() { // itimer is used to determine when to start ignoring `minDesiredPeerCount` //itimer := time.NewTicker(5 * time.Second) @@ -116,11 +191,18 @@ out: if len(d.peers) < minDesiredPeerCount { break } + d.selectPeer(d.peers.bestPeer()) 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 - d.selectPeer(d.peers.bestPeer()) + // Either select the peer or restart the timer if no peers could + // be selected. + if peer := d.peers.bestPeer(); peer != nil { + d.selectPeer(d.peers.bestPeer()) + } else { + itimer.Reset(5 * time.Second) + } case <-d.quit: break out } @@ -142,6 +224,7 @@ func (d *Downloader) selectPeer(p *peer) { glog.V(logger.Detail).Infoln("New peer with highest TD =", p.td) d.syncCh <- syncPack{p, p.recentHash, false} } + } func (d *Downloader) update() { @@ -245,8 +328,13 @@ out: for { select { case blockPack := <-d.blockCh: - d.queue.deliver(blockPack.peerId, blockPack.blocks) - d.peers.setState(blockPack.peerId, idleState) + // If the peer was previously banned and failed to deliver it's pack + // in a reasonable time frame, ignore it's message. + if d.peers[blockPack.peerId] != nil { + d.peers[blockPack.peerId].promote() + d.queue.deliver(blockPack.peerId, blockPack.blocks) + d.peers.setState(blockPack.peerId, idleState) + } case <-ticker.C: // If there are unrequested hashes left start fetching // from the available peers. @@ -310,6 +398,9 @@ out: // 2) Measure their speed; // 3) Amount and availability. d.queue.deliver(pid, nil) + if peer := d.peers[pid]; peer != nil { + peer.demote() + } } } @@ -343,6 +434,7 @@ func (d *Downloader) AddBlock(id string, block *types.Block, td *big.Int) { peer.td = td peer.recentHash = block.Hash() peer.mu.Unlock() + peer.promote() glog.V(logger.Detail).Infoln("Inserting new block from:", id) d.queue.addBlock(id, block, td) diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index f66e5afd8..4cd306a05 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -95,3 +95,23 @@ func (p *peer) fetch(chunk *chunk) error { return nil } + +// promote increases the peer's reputation +func (p *peer) promote() { + p.mu.Lock() + defer p.mu.Unlock() + + p.rep++ +} + +// demote decreases the peer's reputation or leaves it at 0 +func (p *peer) demote() { + p.mu.Lock() + defer p.mu.Unlock() + + if p.rep > 1 { + p.rep -= 2 + } else { + p.rep = 0 + } +} |