downloader: make sure that hashes are only accepted from the active peer

This commit is contained in:
obscuren 2015-04-18 17:35:03 +02:00
parent ff67fbf964
commit 60613b57d1
2 changed files with 54 additions and 20 deletions

View File

@ -2,6 +2,7 @@ package downloader
import ( import (
"errors" "errors"
"fmt"
"math" "math"
"math/big" "math/big"
"sync" "sync"
@ -18,8 +19,10 @@ import (
const ( const (
maxBlockFetch = 256 // Amount of max blocks to be fetched per chunk maxBlockFetch = 256 // Amount of max blocks to be fetched per chunk
minDesiredPeerCount = 3 // Amount of peers desired to start syncing minDesiredPeerCount = 5 // Amount of peers desired to start syncing
blockTtl = 15 * time.Second // The amount of time it takes for a request to time out peerCountTimeout = 12 * time.Second // Amount of time it takes for the peer handler to ignore minDesiredPeerCount
blockTtl = 15 * time.Second // The amount of time it takes for a block request to time out
hashTtl = 20 * time.Second // The amount of time it takes for a hash request to time out
) )
var ( var (
@ -34,9 +37,10 @@ type hashIterFn func() (common.Hash, error)
type currentTdFn func() *big.Int type currentTdFn func() *big.Int
type Downloader struct { type Downloader struct {
mu sync.RWMutex mu sync.RWMutex
queue *queue queue *queue
peers peers peers peers
activePeer string
// Callbacks // Callbacks
hasBlock hashCheckFn hasBlock hashCheckFn
@ -51,7 +55,7 @@ type Downloader struct {
// Channels // Channels
newPeerCh chan *peer newPeerCh chan *peer
syncCh chan syncPack syncCh chan syncPack
HashCh chan []common.Hash hashCh chan []common.Hash
blockCh chan blockPack blockCh chan blockPack
quit chan struct{} quit chan struct{}
} }
@ -76,7 +80,7 @@ func New(hasBlock hashCheckFn, insertChain chainInsertFn, currentTd currentTdFn)
currentTd: currentTd, currentTd: currentTd,
newPeerCh: make(chan *peer, 1), newPeerCh: make(chan *peer, 1),
syncCh: make(chan syncPack, 1), syncCh: make(chan syncPack, 1),
HashCh: make(chan []common.Hash, 1), hashCh: make(chan []common.Hash, 1),
blockCh: make(chan blockPack, 1), blockCh: make(chan blockPack, 1),
quit: make(chan struct{}), quit: make(chan struct{}),
} }
@ -181,8 +185,7 @@ func (d *Downloader) getFromPeer(p *peer, hash common.Hash, ignoreInitial bool)
func (d *Downloader) peerHandler() { func (d *Downloader) peerHandler() {
// itimer is used to determine when to start ignoring `minDesiredPeerCount` // itimer is used to determine when to start ignoring `minDesiredPeerCount`
//itimer := time.NewTicker(5 * time.Second) itimer := time.NewTimer(peerCountTimeout)
itimer := time.NewTimer(5 * time.Second)
out: out:
for { for {
select { select {
@ -233,12 +236,16 @@ out:
for { for {
select { select {
case sync := <-d.syncCh: case sync := <-d.syncCh:
selectedPeer := sync.peer start := time.Now()
glog.V(logger.Detail).Infoln("Synchronising with the network using:", selectedPeer.id)
var peer *peer = sync.peer
d.activePeer = peer.id
glog.V(logger.Detail).Infoln("Synchronising with the network using:", peer.id)
// Start the fetcher. This will block the update entirely // Start the fetcher. This will block the update entirely
// interupts need to be send to the appropriate channels // interupts need to be send to the appropriate channels
// respectively. // respectively.
if err := d.startFetchingHashes(selectedPeer, sync.hash, sync.ignoreInitial); err != nil { if err := d.startFetchingHashes(peer, sync.hash, sync.ignoreInitial); err != nil {
// handle error // handle error
glog.V(logger.Debug).Infoln("Error fetching hashes:", err) glog.V(logger.Debug).Infoln("Error fetching hashes:", err)
// XXX Reset // XXX Reset
@ -249,13 +256,13 @@ out:
// take any available peers, seserve a chunk for each peer available, // take any available peers, seserve a chunk for each peer available,
// let the peer deliver the chunkn and periodically check if a peer // let the peer deliver the chunkn and periodically check if a peer
// has timedout. When done downloading, process blocks. // has timedout. When done downloading, process blocks.
if err := d.startFetchingBlocks(selectedPeer); err != nil { if err := d.startFetchingBlocks(peer); err != nil {
glog.V(logger.Debug).Infoln("Error downloading blocks:", err) glog.V(logger.Debug).Infoln("Error downloading blocks:", err)
// XXX reset // XXX reset
break break
} }
glog.V(logger.Detail).Infoln("Sync completed") glog.V(logger.Detail).Infoln("Network sync completed in", time.Since(start))
d.process() d.process()
case <-d.quit: case <-d.quit:
@ -282,10 +289,12 @@ func (d *Downloader) startFetchingHashes(p *peer, hash common.Hash, ignoreInitia
// Get the first batch of hashes // Get the first batch of hashes
p.getHashes(hash) p.getHashes(hash)
failureResponse := time.NewTimer(hashTtl)
out: out:
for { for {
select { select {
case hashes := <-d.HashCh: case hashes := <-d.hashCh:
var done bool // determines whether we're done fetching hashes (i.e. common hash found) var done bool // determines whether we're done fetching hashes (i.e. common hash found)
hashSet := set.New() hashSet := set.New()
for _, hash := range hashes { for _, hash := range hashes {
@ -313,15 +322,20 @@ out:
} else { // we're done } else { // we're done
break out break out
} }
case <-failureResponse.C:
glog.V(logger.Debug).Infof("Peer (%s) didn't respond in time for hash request\n", p.id)
d.queue.reset()
break out
} }
} }
glog.V(logger.Detail).Infof("Downloaded hashes (%d). Took %v\n", d.queue.hashPool.Size(), time.Since(start)) glog.V(logger.Detail).Infof("Downloaded hashes (%d) in %v\n", d.queue.hashPool.Size(), time.Since(start))
return nil return nil
} }
func (d *Downloader) startFetchingBlocks(p *peer) error { func (d *Downloader) startFetchingBlocks(p *peer) error {
glog.V(logger.Detail).Infoln("Downloading", d.queue.hashPool.Size(), "blocks") glog.V(logger.Detail).Infoln("Downloading", d.queue.hashPool.Size(), "block(s)")
atomic.StoreInt32(&d.downloadingBlocks, 1) atomic.StoreInt32(&d.downloadingBlocks, 1)
defer atomic.StoreInt32(&d.downloadingBlocks, 0) defer atomic.StoreInt32(&d.downloadingBlocks, 0)
@ -407,7 +421,20 @@ out:
} }
} }
glog.V(logger.Detail).Infoln("Download blocks: done. Took", time.Since(start)) glog.V(logger.Detail).Infoln("Downloaded block(s) in", time.Since(start))
return nil
}
func (d *Downloader) AddHashes(id string, hashes []common.Hash) error {
// make sure that the hashes that are being added are actually from the peer
// that's the current active peer. hashes that have been received from other
// peers are dropped and ignored.
if d.activePeer != id {
return fmt.Errorf("received hashes from %s while active peer is %s", id, d.activePeer)
}
d.hashCh <- hashes
return nil return nil
} }

View File

@ -194,7 +194,10 @@ func (self *ProtocolManager) handleMsg(p *peer) error {
if err := msgStream.Decode(&hashes); err != nil { if err := msgStream.Decode(&hashes); err != nil {
break break
} }
self.downloader.HashCh <- hashes err := self.downloader.AddHashes(p.id, hashes)
if err != nil {
glog.V(logger.Debug).Infoln(err)
}
case GetBlocksMsg: case GetBlocksMsg:
msgStream := rlp.NewStream(msg.Payload) msgStream := rlp.NewStream(msg.Payload)
@ -259,7 +262,11 @@ func (self *ProtocolManager) handleMsg(p *peer) error {
// Make sure the block isn't already known. If this is the case simply drop // Make sure the block isn't already known. If this is the case simply drop
// the message and move on. If the TD is < currentTd; drop it as well. If this // the message and move on. If the TD is < currentTd; drop it as well. If this
// chain at some point becomes canonical, the downloader will fetch it. // chain at some point becomes canonical, the downloader will fetch it.
if self.chainman.HasBlock(hash) && self.chainman.Td().Cmp(request.TD) > 0 { if self.chainman.HasBlock(hash) {
break
}
if self.chainman.Td().Cmp(request.TD) > 0 {
glog.V(logger.Debug).Infoln("dropped block", request.Block.Number(), "due to low TD", request.TD)
break break
} }