diff options
| -rw-r--r-- | eth/downloader/downloader.go | 63 | ||||
| -rw-r--r-- | eth/handler.go | 11 |
2 files changed, 54 insertions, 20 deletions
diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index c71cfa684..41484e927 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -2,6 +2,7 @@ package downloader import ( "errors" + "fmt" "math" "math/big" "sync" @@ -18,8 +19,10 @@ import ( const ( maxBlockFetch = 256 // Amount of max blocks to be fetched per chunk - minDesiredPeerCount = 3 // Amount of peers desired to start syncing - blockTtl = 15 * time.Second // The amount of time it takes for a request to time out + minDesiredPeerCount = 5 // Amount of peers desired to start syncing + 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 ( @@ -34,9 +37,10 @@ type hashIterFn func() (common.Hash, error) type currentTdFn func() *big.Int type Downloader struct { - mu sync.RWMutex - queue *queue - peers peers + mu sync.RWMutex + queue *queue + peers peers + activePeer string // Callbacks hasBlock hashCheckFn @@ -51,7 +55,7 @@ type Downloader struct { // Channels newPeerCh chan *peer syncCh chan syncPack - HashCh chan []common.Hash + hashCh chan []common.Hash blockCh chan blockPack quit chan struct{} } @@ -76,7 +80,7 @@ func New(hasBlock hashCheckFn, insertChain chainInsertFn, currentTd currentTdFn) currentTd: currentTd, newPeerCh: make(chan *peer, 1), syncCh: make(chan syncPack, 1), - HashCh: make(chan []common.Hash, 1), + hashCh: make(chan []common.Hash, 1), blockCh: make(chan blockPack, 1), quit: make(chan struct{}), } @@ -181,8 +185,7 @@ func (d *Downloader) getFromPeer(p *peer, hash common.Hash, ignoreInitial bool) func (d *Downloader) peerHandler() { // itimer is used to determine when to start ignoring `minDesiredPeerCount` - //itimer := time.NewTicker(5 * time.Second) - itimer := time.NewTimer(5 * time.Second) + itimer := time.NewTimer(peerCountTimeout) out: for { select { @@ -233,12 +236,16 @@ out: for { select { case sync := <-d.syncCh: - selectedPeer := sync.peer - glog.V(logger.Detail).Infoln("Synchronising with the network using:", selectedPeer.id) + start := time.Now() + + 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 // interupts need to be send to the appropriate channels // 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 glog.V(logger.Debug).Infoln("Error fetching hashes:", err) // XXX Reset @@ -249,13 +256,13 @@ out: // 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(selectedPeer); err != nil { + if err := d.startFetchingBlocks(peer); err != nil { glog.V(logger.Debug).Infoln("Error downloading blocks:", err) // XXX reset break } - glog.V(logger.Detail).Infoln("Sync completed") + glog.V(logger.Detail).Infoln("Network sync completed in", time.Since(start)) d.process() case <-d.quit: @@ -282,10 +289,12 @@ func (d *Downloader) startFetchingHashes(p *peer, hash common.Hash, ignoreInitia // Get the first batch of hashes p.getHashes(hash) + failureResponse := time.NewTimer(hashTtl) + out: for { select { - case hashes := <-d.HashCh: + case hashes := <-d.hashCh: var done bool // determines whether we're done fetching hashes (i.e. common hash found) hashSet := set.New() for _, hash := range hashes { @@ -313,15 +322,20 @@ out: } else { // we're done 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 } 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) 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 } diff --git a/eth/handler.go b/eth/handler.go index f3fad68b7..3aa9815f1 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -194,7 +194,10 @@ func (self *ProtocolManager) handleMsg(p *peer) error { if err := msgStream.Decode(&hashes); err != nil { break } - self.downloader.HashCh <- hashes + err := self.downloader.AddHashes(p.id, hashes) + if err != nil { + glog.V(logger.Debug).Infoln(err) + } case GetBlocksMsg: 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 // 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. - 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 } |