From c4678ffd77a18a9d03c888fdf242c9e5915b9f5f Mon Sep 17 00:00:00 2001
From: obscuren <geffobscura@gmail.com>
Date: Thu, 16 Apr 2015 00:14:31 +0200
Subject: downloader: updated downloader and fixed issues with catch up

Properly ignore blocks coming from peers not in our peer list (blocked)
and do never request anything from bad peers. Added some checks to
account for blocks known when requesting hashes (missing parents).
---
 eth/downloader/downloader.go | 70 ++++++++++++++++++++++++++++++--------------
 1 file changed, 48 insertions(+), 22 deletions(-)

(limited to 'eth/downloader')

diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 83e6b8d32..1707e1395 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -54,8 +54,9 @@ type blockPack struct {
 }
 
 type syncPack struct {
-	peer *peer
-	hash common.Hash
+	peer          *peer
+	hash          common.Hash
+	ignoreInitial bool
 }
 
 func New(hasBlock hashCheckFn, insertChain chainInsertFn, currentTd currentTdFn) *Downloader {
@@ -104,11 +105,13 @@ func (d *Downloader) UnregisterPeer(id string) {
 
 func (d *Downloader) peerHandler() {
 	// itimer is used to determine when to start ignoring `minDesiredPeerCount`
-	itimer := time.NewTicker(5 * time.Second)
+	//itimer := time.NewTicker(5 * time.Second)
+	itimer := time.NewTimer(5 * time.Second)
 out:
 	for {
 		select {
 		case <-d.newPeerCh:
+			itimer.Stop()
 			// Meet the `minDesiredPeerCount` before we select our best peer
 			if len(d.peers) < minDesiredPeerCount {
 				break
@@ -137,7 +140,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}
+		d.syncCh <- syncPack{p, p.recentHash, false}
 	}
 }
 
@@ -147,11 +150,11 @@ out:
 		select {
 		case sync := <-d.syncCh:
 			selectedPeer := sync.peer
-			glog.V(logger.Detail).Infoln("Synchronising with network using:", selectedPeer.id)
+			glog.V(logger.Detail).Infoln("Synchronising with the network using:", selectedPeer.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); err != nil {
+			if err := d.startFetchingHashes(selectedPeer, sync.hash, sync.ignoreInitial); err != nil {
 				// handle error
 				glog.V(logger.Debug).Infoln("Error fetching hashes:", err)
 				// XXX Reset
@@ -178,11 +181,18 @@ out:
 }
 
 // XXX Make synchronous
-func (d *Downloader) startFetchingHashes(p *peer, hash common.Hash) error {
-	glog.V(logger.Debug).Infoln("Downloading hashes")
+func (d *Downloader) startFetchingHashes(p *peer, hash common.Hash, ignoreInitial bool) error {
+	glog.V(logger.Debug).Infof("Downloading hashes (%x) from %s", hash.Bytes()[:4], p.id)
 
 	start := time.Now()
 
+	// We ignore the initial hash in some cases (e.g. we received a block without it's parent)
+	// In such circumstances we don't need to download the block so don't add it to the queue.
+	if !ignoreInitial {
+		// Add the hash to the queue first
+		d.queue.hashPool.Add(hash)
+	}
+
 	// Get the first batch of hashes
 	p.getHashes(hash)
 	atomic.StoreInt32(&d.fetchingHashes, 1)
@@ -195,7 +205,7 @@ out:
 			hashSet := set.New()
 			for _, hash := range hashes {
 				if d.hasBlock(hash) {
-					glog.V(logger.Debug).Infof("Found common hash %x\n", hash)
+					glog.V(logger.Debug).Infof("Found common hash %x\n", hash[:4])
 
 					done = true
 					break
@@ -207,7 +217,7 @@ out:
 
 			// Add hashes to the chunk set
 			// Check if we're done fetching
-			if !done {
+			if !done && len(hashes) > 0 {
 				//fmt.Println("re-fetch. current =", d.queue.hashPool.Size())
 				// Get the next set of hashes
 				p.getHashes(hashes[len(hashes)-1])
@@ -218,7 +228,7 @@ out:
 			}
 		}
 	}
-	glog.V(logger.Detail).Infoln("Download hashes: done. Took", time.Since(start))
+	glog.V(logger.Detail).Infof("Downloaded hashes (%d). Took %v\n", d.queue.hashPool.Size(), time.Since(start))
 
 	return nil
 }
@@ -242,6 +252,10 @@ out:
 			// from the available peers.
 			if d.queue.hashPool.Size() > 0 {
 				availablePeers := d.peers.get(idleState)
+				if len(availablePeers) == 0 {
+					glog.V(logger.Detail).Infoln("No peers available out of", len(d.peers))
+				}
+
 				for _, peer := range availablePeers {
 					// Get a possible chunk. If nil is returned no chunk
 					// could be returned due to no hashes available.
@@ -317,21 +331,33 @@ func (d *Downloader) AddBlock(id string, block *types.Block, td *big.Int) {
 		return
 	}
 
-	glog.V(logger.Detail).Infoln("Inserting new block from:", id)
-	d.queue.addBlock(id, block, td)
-
+	peer := d.peers.getPeer(id)
 	// if the peer is in our healthy list of peers; update the td
-	// here is a good chance to add the peer back to the list
-	if peer := d.peers.getPeer(id); peer != nil {
-		peer.mu.Lock()
-		peer.td = td
-		peer.recentHash = block.Hash()
-		peer.mu.Unlock()
+	// and add the block. Otherwise just ignore it
+	if peer == nil {
+		glog.V(logger.Detail).Infof("Ignored block from bad peer %s\n", id)
+		return
 	}
 
+	peer.mu.Lock()
+	peer.td = td
+	peer.recentHash = block.Hash()
+	peer.mu.Unlock()
+
+	glog.V(logger.Detail).Infoln("Inserting new block from:", id)
+	d.queue.addBlock(id, block, td)
+
 	// if neither go ahead to process
 	if !(d.isFetchingHashes() || d.isDownloadingBlocks()) {
-		d.process()
+		// Check if the parent of the received block is known.
+		// If the block is not know, request it otherwise, request.
+		phash := block.ParentHash()
+		if !d.hasBlock(phash) {
+			glog.V(logger.Detail).Infof("Missing parent %x, requires fetching\n", phash.Bytes()[:4])
+			d.syncCh <- syncPack{peer, peer.recentHash, true}
+		} else {
+			d.process()
+		}
 	}
 }
 
@@ -369,7 +395,7 @@ func (d *Downloader) process() error {
 			// TODO change this. This shite
 			for i, block := range blocks[:max] {
 				if !d.hasBlock(block.ParentHash()) {
-					d.syncCh <- syncPack{d.peers.bestPeer(), block.Hash()}
+					d.syncCh <- syncPack{d.peers.bestPeer(), block.Hash(), true}
 					// remove processed blocks
 					blocks = blocks[i:]
 
-- 
cgit v1.2.3


From 8f873b762b54a033e891df03175a26cbfb582c43 Mon Sep 17 00:00:00 2001
From: obscuren <geffobscura@gmail.com>
Date: Sat, 18 Apr 2015 03:15:26 +0200
Subject: downloader: all handlers check for isBusy

---
 eth/downloader/downloader.go | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

(limited to 'eth/downloader')

diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 5f9d9ed74..c5b951344 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -116,7 +116,7 @@ func (d *Downloader) UnregisterPeer(id string) {
 // 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() {
+	if d.isBusy() {
 		return nil, errBusy
 	}
 
@@ -213,7 +213,7 @@ func (d *Downloader) selectPeer(p *peer) {
 	// Make sure it's doing neither. Once done we can restart the
 	// downloading process if the TD is higher. For now just get on
 	// with whatever is going on. This prevents unecessary switching.
-	if !(d.isFetchingHashes() || d.isDownloadingBlocks() || d.isProcessing()) {
+	if !d.isBusy() {
 		// selected peer must be better than our own
 		// XXX we also check the peer's recent hash to make sure we
 		// don't have it. Some peers report (i think) incorrect TD.
@@ -340,10 +340,6 @@ out:
 			// from the available peers.
 			if d.queue.hashPool.Size() > 0 {
 				availablePeers := d.peers.get(idleState)
-				if len(availablePeers) == 0 {
-					glog.V(logger.Detail).Infoln("No peers available out of", len(d.peers))
-				}
-
 				for _, peer := range availablePeers {
 					// Get a possible chunk. If nil is returned no chunk
 					// could be returned due to no hashes available.
@@ -440,7 +436,7 @@ func (d *Downloader) AddBlock(id string, block *types.Block, td *big.Int) {
 	d.queue.addBlock(id, block, td)
 
 	// if neither go ahead to process
-	if !(d.isFetchingHashes() || d.isDownloadingBlocks()) {
+	if !d.isBusy() {
 		// Check if the parent of the received block is known.
 		// If the block is not know, request it otherwise, request.
 		phash := block.ParentHash()
@@ -519,3 +515,7 @@ func (d *Downloader) isDownloadingBlocks() bool {
 func (d *Downloader) isProcessing() bool {
 	return atomic.LoadInt32(&d.processingBlocks) == 1
 }
+
+func (d *Downloader) isBusy() bool {
+	return d.isFetchingHashes() || d.isDownloadingBlocks() || d.isProcessing()
+}
-- 
cgit v1.2.3


From 8244825bbf9ca7342c052508f50a56b16c979a1e Mon Sep 17 00:00:00 2001
From: obscuren <geffobscura@gmail.com>
Date: Sat, 18 Apr 2015 15:14:12 +0200
Subject: downloader: reset the queue if a peer response with an empty hash set

---
 eth/downloader/downloader.go | 39 +++++++++++++++++++++++++--------------
 eth/downloader/queue.go      | 11 +++++++++++
 2 files changed, 36 insertions(+), 14 deletions(-)

(limited to 'eth/downloader')

diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index c5b951344..c71cfa684 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -17,8 +17,9 @@ import (
 )
 
 const (
-	maxBlockFetch       = 256 // Amount of max blocks to be fetched per chunk
-	minDesiredPeerCount = 3   // Amount of peers desired to start syncing
+	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
 )
 
 var (
@@ -96,7 +97,7 @@ func (d *Downloader) RegisterPeer(id string, td *big.Int, hash common.Hash, getH
 	// add peer to our peer set
 	d.peers[id] = peer
 	// broadcast new peer
-	//d.newPeerCh <- peer
+	d.newPeerCh <- peer
 
 	return nil
 }
@@ -265,6 +266,9 @@ out:
 
 // XXX Make synchronous
 func (d *Downloader) startFetchingHashes(p *peer, hash common.Hash, ignoreInitial bool) error {
+	atomic.StoreInt32(&d.fetchingHashes, 1)
+	defer atomic.StoreInt32(&d.fetchingHashes, 0)
+
 	glog.V(logger.Debug).Infof("Downloading hashes (%x) from %s", hash.Bytes()[:4], p.id)
 
 	start := time.Now()
@@ -275,10 +279,8 @@ func (d *Downloader) startFetchingHashes(p *peer, hash common.Hash, ignoreInitia
 		// Add the hash to the queue first
 		d.queue.hashPool.Add(hash)
 	}
-
 	// Get the first batch of hashes
 	p.getHashes(hash)
-	atomic.StoreInt32(&d.fetchingHashes, 1)
 
 out:
 	for {
@@ -299,14 +301,16 @@ out:
 			d.queue.put(hashSet)
 
 			// Add hashes to the chunk set
-			// Check if we're done fetching
-			if !done && len(hashes) > 0 {
+			if len(hashes) == 0 { // Make sure the peer actually gave you something valid
+				glog.V(logger.Debug).Infof("Peer (%s) responded with empty hash set\n", p.id)
+				d.queue.reset()
+
+				break out
+			} else if !done { // Check if we're done fetching
 				//fmt.Println("re-fetch. current =", d.queue.hashPool.Size())
 				// Get the next set of hashes
 				p.getHashes(hashes[len(hashes)-1])
-				atomic.StoreInt32(&d.fetchingHashes, 1)
-			} else {
-				atomic.StoreInt32(&d.fetchingHashes, 0)
+			} else { // we're done
 				break out
 			}
 		}
@@ -319,6 +323,7 @@ out:
 func (d *Downloader) startFetchingBlocks(p *peer) error {
 	glog.V(logger.Detail).Infoln("Downloading", d.queue.hashPool.Size(), "blocks")
 	atomic.StoreInt32(&d.downloadingBlocks, 1)
+	defer atomic.StoreInt32(&d.downloadingBlocks, 0)
 
 	start := time.Now()
 
@@ -364,8 +369,6 @@ out:
 				// When there are no more queue and no more `fetching`. We can
 				// safely assume we're done. Another part of the process will  check
 				// for parent errors and will re-request anything that's missing
-				atomic.StoreInt32(&d.downloadingBlocks, 0)
-				// Break out so that we can process with processing blocks
 				break out
 			} else {
 				// Check for bad peers. Bad peers may indicate a peer not responding
@@ -376,7 +379,7 @@ out:
 				d.queue.mu.Lock()
 				var badPeers []string
 				for pid, chunk := range d.queue.fetching {
-					if time.Since(chunk.itime) > 5*time.Second {
+					if time.Since(chunk.itime) > blockTtl {
 						badPeers = append(badPeers, pid)
 						// remove peer as good peer from peer list
 						d.UnregisterPeer(pid)
@@ -466,8 +469,11 @@ func (d *Downloader) process() error {
 	// to a seperate goroutine where it periodically checks for linked pieces.
 	types.BlockBy(types.Number).Sort(d.queue.blocks)
 	blocks := d.queue.blocks
+	if len(blocks) == 0 {
+		return nil
+	}
 
-	glog.V(logger.Debug).Infoln("Inserting chain with", len(blocks), "blocks")
+	glog.V(logger.Debug).Infof("Inserting chain with %d blocks (#%v - #%v)\n", len(blocks), blocks[0].Number(), blocks[len(blocks)-1].Number())
 
 	var err error
 	// Loop untill we're out of blocks
@@ -491,6 +497,11 @@ func (d *Downloader) process() error {
 				}
 			}
 			break
+		} else if err != nil {
+			// Reset chain completely. This needs much, much improvement.
+			// instead: check all blocks leading down to this block false block and remove it
+			blocks = nil
+			break
 		}
 		blocks = blocks[max:]
 	}
diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go
index 4d1aa4e93..df3bf7087 100644
--- a/eth/downloader/queue.go
+++ b/eth/downloader/queue.go
@@ -31,6 +31,17 @@ func newqueue() *queue {
 	}
 }
 
+func (c *queue) reset() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.hashPool.Clear()
+	c.fetchPool.Clear()
+	c.blockHashes.Clear()
+	c.blocks = nil
+	c.fetching = make(map[string]*chunk)
+}
+
 // reserve a `max` set of hashes for `p` peer.
 func (c *queue) get(p *peer, max int) *chunk {
 	c.mu.Lock()
-- 
cgit v1.2.3


From 60613b57d1956275bb475a53b5085c4ead4ceb2c Mon Sep 17 00:00:00 2001
From: obscuren <geffobscura@gmail.com>
Date: Sat, 18 Apr 2015 17:35:03 +0200
Subject: downloader: make sure that hashes are only accepted from the active
 peer

---
 eth/downloader/downloader.go | 63 +++++++++++++++++++++++++++++++-------------
 1 file changed, 45 insertions(+), 18 deletions(-)

(limited to 'eth/downloader')

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
 }
-- 
cgit v1.2.3


From c2c24b3bb419a8ffffb58ec25788b951bef779f9 Mon Sep 17 00:00:00 2001
From: obscuren <geffobscura@gmail.com>
Date: Sat, 18 Apr 2015 18:54:57 +0200
Subject: downloader: improved downloading and synchronisation

* Downloader's peers keeps track of peer's previously requested hashes
  so that we don't have to re-request
* Changed `AddBlock` to be fully synchronous
---
 eth/downloader/downloader.go  | 144 ++++++++++--------------------------------
 eth/downloader/peer.go        |  15 ++++-
 eth/downloader/queue.go       |   3 +
 eth/downloader/synchronous.go |  77 ++++++++++++++++++++++
 4 files changed, 129 insertions(+), 110 deletions(-)
 create mode 100644 eth/downloader/synchronous.go

(limited to 'eth/downloader')

diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 41484e927..810031c79 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -26,9 +26,12 @@ const (
 )
 
 var (
-	errLowTd       = errors.New("peer's TD is too low")
-	errBusy        = errors.New("busy")
-	errUnknownPeer = errors.New("peer's unknown or unhealthy")
+	errLowTd        = errors.New("peer's TD is too low")
+	errBusy         = errors.New("busy")
+	errUnknownPeer  = errors.New("peer's unknown or unhealthy")
+	errBadPeer      = errors.New("action from bad peer ignored")
+	errTimeout      = errors.New("timeout")
+	errEmptyHashSet = errors.New("empty hash set by peer")
 )
 
 type hashCheckFn func(common.Hash) bool
@@ -116,73 +119,6 @@ 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.isBusy() {
-		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.NewTimer(peerCountTimeout)
@@ -236,34 +172,14 @@ out:
 	for {
 		select {
 		case sync := <-d.syncCh:
-			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(peer, sync.hash, sync.ignoreInitial); err != nil {
-				// handle error
-				glog.V(logger.Debug).Infoln("Error fetching hashes:", err)
-				// XXX Reset
-				break
-			}
 
-			// 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(peer); err != nil {
-				glog.V(logger.Debug).Infoln("Error downloading blocks:", err)
-				// XXX reset
+			err := d.getFromPeer(peer, sync.hash, sync.ignoreInitial)
+			if err != nil {
 				break
 			}
 
-			glog.V(logger.Detail).Infoln("Network sync completed in", time.Since(start))
-
 			d.process()
 		case <-d.quit:
 			break out
@@ -314,9 +230,8 @@ out:
 				glog.V(logger.Debug).Infof("Peer (%s) responded with empty hash set\n", p.id)
 				d.queue.reset()
 
-				break out
+				return errEmptyHashSet
 			} else if !done { // Check if we're done fetching
-				//fmt.Println("re-fetch. current =", d.queue.hashPool.Size())
 				// Get the next set of hashes
 				p.getHashes(hashes[len(hashes)-1])
 			} else { // we're done
@@ -324,9 +239,12 @@ out:
 			}
 		case <-failureResponse.C:
 			glog.V(logger.Debug).Infof("Peer (%s) didn't respond in time for hash request\n", p.id)
+			// TODO instead of reseting the queue select a new peer from which we can start downloading hashes.
+			// 1. check for peer's best hash to be included in the current hash set;
+			// 2. resume from last point (hashes[len(hashes)-1]) using the newly selected peer.
 			d.queue.reset()
 
-			break out
+			return errTimeout
 		}
 	}
 	glog.V(logger.Detail).Infof("Downloaded hashes (%d) in %v\n", d.queue.hashPool.Size(), time.Since(start))
@@ -367,7 +285,6 @@ out:
 						continue
 					}
 
-					//fmt.Println("fetching for", peer.id)
 					// XXX make fetch blocking.
 					// Fetch the chunk and check for error. If the peer was somehow
 					// already fetching a chunk due to a bug, it will be returned to
@@ -417,7 +334,6 @@ out:
 				}
 
 			}
-			//fmt.Println(d.queue.hashPool.Size(), len(d.queue.fetching))
 		}
 	}
 
@@ -441,11 +357,14 @@ func (d *Downloader) AddHashes(id string, hashes []common.Hash) error {
 
 // Add an (unrequested) block to the downloader. This is usually done through the
 // NewBlockMsg by the protocol handler.
-func (d *Downloader) AddBlock(id string, block *types.Block, td *big.Int) {
+// Adding blocks is done synchronously. if there are missing blocks, blocks will be
+// fetched first. If the downloader is busy or if some other processed failed an error
+// will be returned.
+func (d *Downloader) AddBlock(id string, block *types.Block, td *big.Int) error {
 	hash := block.Hash()
 
 	if d.hasBlock(hash) {
-		return
+		return fmt.Errorf("known block %x", hash.Bytes()[:4])
 	}
 
 	peer := d.peers.getPeer(id)
@@ -453,7 +372,7 @@ func (d *Downloader) AddBlock(id string, block *types.Block, td *big.Int) {
 	// and add the block. Otherwise just ignore it
 	if peer == nil {
 		glog.V(logger.Detail).Infof("Ignored block from bad peer %s\n", id)
-		return
+		return errBadPeer
 	}
 
 	peer.mu.Lock()
@@ -466,17 +385,24 @@ func (d *Downloader) AddBlock(id string, block *types.Block, td *big.Int) {
 	d.queue.addBlock(id, block, td)
 
 	// if neither go ahead to process
-	if !d.isBusy() {
-		// Check if the parent of the received block is known.
-		// If the block is not know, request it otherwise, request.
-		phash := block.ParentHash()
-		if !d.hasBlock(phash) {
-			glog.V(logger.Detail).Infof("Missing parent %x, requires fetching\n", phash.Bytes()[:4])
-			d.syncCh <- syncPack{peer, peer.recentHash, true}
-		} else {
-			d.process()
+	if d.isBusy() {
+		return errBusy
+	}
+
+	// Check if the parent of the received block is known.
+	// If the block is not know, request it otherwise, request.
+	phash := block.ParentHash()
+	if !d.hasBlock(phash) {
+		glog.V(logger.Detail).Infof("Missing parent %x, requires fetching\n", phash.Bytes()[:4])
+
+		// Get the missing hashes from the peer (synchronously)
+		err := d.getFromPeer(peer, peer.recentHash, true)
+		if err != nil {
+			return err
 		}
 	}
+
+	return d.process()
 }
 
 // Deliver a chunk to the downloader. This is usually done through the BlocksMsg by
diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go
index 4cd306a05..5d5208e8e 100644
--- a/eth/downloader/peer.go
+++ b/eth/downloader/peer.go
@@ -6,6 +6,7 @@ import (
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
+	"gopkg.in/fatih/set.v0"
 )
 
 const (
@@ -64,13 +65,23 @@ type peer struct {
 	td         *big.Int
 	recentHash common.Hash
 
+	requested *set.Set
+
 	getHashes hashFetcherFn
 	getBlocks blockFetcherFn
 }
 
 // create a new peer
 func newPeer(id string, td *big.Int, hash common.Hash, getHashes hashFetcherFn, getBlocks blockFetcherFn) *peer {
-	return &peer{id: id, td: td, recentHash: hash, getHashes: getHashes, getBlocks: getBlocks, state: idleState}
+	return &peer{
+		id:         id,
+		td:         td,
+		recentHash: hash,
+		getHashes:  getHashes,
+		getBlocks:  getBlocks,
+		state:      idleState,
+		requested:  set.New(),
+	}
 }
 
 // fetch a chunk using the peer
@@ -82,6 +93,8 @@ func (p *peer) fetch(chunk *chunk) error {
 		return errors.New("peer already fetching chunk")
 	}
 
+	p.requested.Merge(chunk.hashes)
+
 	// set working state
 	p.state = workingState
 	// convert the set to a fetchable slice
diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go
index df3bf7087..5745bf1f8 100644
--- a/eth/downloader/queue.go
+++ b/eth/downloader/queue.go
@@ -65,6 +65,9 @@ func (c *queue) get(p *peer, max int) *chunk {
 
 		return true
 	})
+	// remove hashes that have previously been fetched
+	hashes.Separate(p.requested)
+
 	// remove the fetchable hashes from hash pool
 	c.hashPool.Separate(hashes)
 	c.fetchPool.Merge(hashes)
diff --git a/eth/downloader/synchronous.go b/eth/downloader/synchronous.go
new file mode 100644
index 000000000..0511533cf
--- /dev/null
+++ b/eth/downloader/synchronous.go
@@ -0,0 +1,77 @@
+package downloader
+
+import (
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
+)
+
+// THIS IS PENDING AND TO DO CHANGES FOR MAKING THE DOWNLOADER SYNCHRONOUS
+
+// 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.isBusy() {
+		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
+}
-- 
cgit v1.2.3


From 0d536734fe10e62dce86db1a6128b383ef66921d Mon Sep 17 00:00:00 2001
From: obscuren <geffobscura@gmail.com>
Date: Sat, 18 Apr 2015 18:55:50 +0200
Subject: eth: adapted to new synchronous api of downloader's AddBlock

---
 eth/downloader/downloader.go  | 2 --
 eth/downloader/synchronous.go | 2 ++
 2 files changed, 2 insertions(+), 2 deletions(-)

(limited to 'eth/downloader')

diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 810031c79..6dce40b04 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -173,8 +173,6 @@ out:
 		select {
 		case sync := <-d.syncCh:
 			var peer *peer = sync.peer
-			d.activePeer = peer.id
-
 			err := d.getFromPeer(peer, sync.hash, sync.ignoreInitial)
 			if err != nil {
 				break
diff --git a/eth/downloader/synchronous.go b/eth/downloader/synchronous.go
index 0511533cf..7bb49d24e 100644
--- a/eth/downloader/synchronous.go
+++ b/eth/downloader/synchronous.go
@@ -50,6 +50,8 @@ func (d *Downloader) Synchronise() (types.Blocks, error) {
 }
 
 func (d *Downloader) getFromPeer(p *peer, hash common.Hash, ignoreInitial bool) error {
+	d.activePeer = p.id
+
 	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
-- 
cgit v1.2.3


From 78e37e98e77b467e6950922da4ea99ff146ab21f Mon Sep 17 00:00:00 2001
From: obscuren <geffobscura@gmail.com>
Date: Sat, 18 Apr 2015 19:14:25 +0200
Subject: downloader: fixed a race condition for download status

---
 eth/downloader/downloader.go | 1 -
 1 file changed, 1 deletion(-)

(limited to 'eth/downloader')

diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 6dce40b04..290e3b474 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -293,7 +293,6 @@ out:
 						d.queue.put(chunk.hashes)
 					}
 				}
-				atomic.StoreInt32(&d.downloadingBlocks, 1)
 			} else if len(d.queue.fetching) == 0 {
 				// When there are no more queue and no more `fetching`. We can
 				// safely assume we're done. Another part of the process will  check
-- 
cgit v1.2.3


From 7c5d50f627b223a8b0217f6ca684b4c7d1d877ef Mon Sep 17 00:00:00 2001
From: obscuren <geffobscura@gmail.com>
Date: Sat, 18 Apr 2015 19:29:30 +0200
Subject: downloader: throw an error if there are no peers available for
 download

If all peers have been tried during the block download process and some
hashes are unfetchable (available peers > 0 and fetching == 0) throw an
error so the process can be aborted.
---
 eth/downloader/downloader.go | 22 ++++++++++++++++------
 1 file changed, 16 insertions(+), 6 deletions(-)

(limited to 'eth/downloader')

diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 290e3b474..2b5dbe952 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -26,12 +26,13 @@ const (
 )
 
 var (
-	errLowTd        = errors.New("peer's TD is too low")
-	errBusy         = errors.New("busy")
-	errUnknownPeer  = errors.New("peer's unknown or unhealthy")
-	errBadPeer      = errors.New("action from bad peer ignored")
-	errTimeout      = errors.New("timeout")
-	errEmptyHashSet = errors.New("empty hash set by peer")
+	errLowTd            = errors.New("peer's TD is too low")
+	errBusy             = errors.New("busy")
+	errUnknownPeer      = errors.New("peer's unknown or unhealthy")
+	errBadPeer          = errors.New("action from bad peer ignored")
+	errTimeout          = errors.New("timeout")
+	errEmptyHashSet     = errors.New("empty hash set by peer")
+	errPeersUnavailable = errors.New("no peers available or all peers tried for block download process")
 )
 
 type hashCheckFn func(common.Hash) bool
@@ -293,6 +294,15 @@ out:
 						d.queue.put(chunk.hashes)
 					}
 				}
+
+				// make sure that we have peers available for fetching. If all peers have been tried
+				// and all failed throw an error
+				if len(availablePeers) > 0 && d.queue.fetchPool.Size() == 0 {
+					d.queue.reset()
+
+					return errPeersUnavailable
+				}
+
 			} else if len(d.queue.fetching) == 0 {
 				// When there are no more queue and no more `fetching`. We can
 				// safely assume we're done. Another part of the process will  check
-- 
cgit v1.2.3


From 6830ddb659270b59b5a310fdc0e581b09fae5326 Mon Sep 17 00:00:00 2001
From: obscuren <geffobscura@gmail.com>
Date: Sat, 18 Apr 2015 20:25:55 +0200
Subject: downloader: free up peers from work when the downloader resets

---
 eth/downloader/downloader.go |  4 +++-
 eth/downloader/peer.go       | 10 ++++++++++
 eth/downloader/queue.go      | 13 +++++++++----
 3 files changed, 22 insertions(+), 5 deletions(-)

(limited to 'eth/downloader')

diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 2b5dbe952..2f98a1414 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -297,8 +297,9 @@ out:
 
 				// make sure that we have peers available for fetching. If all peers have been tried
 				// and all failed throw an error
-				if len(availablePeers) > 0 && d.queue.fetchPool.Size() == 0 {
+				if len(d.queue.fetching) == 0 {
 					d.queue.reset()
+					d.peers.reset()
 
 					return errPeersUnavailable
 				}
@@ -337,6 +338,7 @@ out:
 					d.queue.deliver(pid, nil)
 					if peer := d.peers[pid]; peer != nil {
 						peer.demote()
+						peer.reset()
 					}
 				}
 
diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go
index 5d5208e8e..ec2a61550 100644
--- a/eth/downloader/peer.go
+++ b/eth/downloader/peer.go
@@ -20,6 +20,12 @@ type blockFetcherFn func([]common.Hash) error
 // XXX make threadsafe!!!!
 type peers map[string]*peer
 
+func (p peers) reset() {
+	for _, peer := range p {
+		p.reset()
+	}
+}
+
 func (p peers) get(state int) []*peer {
 	var peers []*peer
 	for _, peer := range p {
@@ -128,3 +134,7 @@ func (p *peer) demote() {
 		p.rep = 0
 	}
 }
+
+func (p *peer) reset() {
+	p.state = idleState
+}
diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go
index 5745bf1f8..ce3aa9850 100644
--- a/eth/downloader/queue.go
+++ b/eth/downloader/queue.go
@@ -60,13 +60,18 @@ func (c *queue) get(p *peer, max int) *chunk {
 			return false
 		}
 
-		hashes.Add(v)
-		i++
+		// Skip any hashes that have previously been requested from the peer
+		if !p.requested.Has(v) {
+			hashes.Add(v)
+			i++
+		}
 
 		return true
 	})
-	// remove hashes that have previously been fetched
-	hashes.Separate(p.requested)
+	// if no hashes can be requested return a nil chunk
+	if hashes.Size() == 0 {
+		return nil
+	}
 
 	// remove the fetchable hashes from hash pool
 	c.hashPool.Separate(hashes)
-- 
cgit v1.2.3


From a1d97ea4dbb5b4462b898f32b1c55e925a465227 Mon Sep 17 00:00:00 2001
From: obscuren <geffobscura@gmail.com>
Date: Sat, 18 Apr 2015 20:35:49 +0200
Subject: typo

---
 eth/downloader/peer.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'eth/downloader')

diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go
index ec2a61550..88ede16f9 100644
--- a/eth/downloader/peer.go
+++ b/eth/downloader/peer.go
@@ -22,7 +22,7 @@ type peers map[string]*peer
 
 func (p peers) reset() {
 	for _, peer := range p {
-		p.reset()
+		peer.reset()
 	}
 }
 
-- 
cgit v1.2.3


From 50e096e627c8c07b4dda3a7221dda5f32dc5c5cb Mon Sep 17 00:00:00 2001
From: obscuren <geffobscura@gmail.com>
Date: Sat, 18 Apr 2015 23:56:08 +0200
Subject: downloader: don't remove peers. keep them around

---
 eth/downloader/downloader.go | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

(limited to 'eth/downloader')

diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 2f98a1414..8f955b483 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -176,6 +176,7 @@ out:
 			var peer *peer = sync.peer
 			err := d.getFromPeer(peer, sync.hash, sync.ignoreInitial)
 			if err != nil {
+				glog.V(logger.Detail).Infoln(err)
 				break
 			}
 
@@ -301,7 +302,7 @@ out:
 					d.queue.reset()
 					d.peers.reset()
 
-					return errPeersUnavailable
+					return fmt.Errorf("%v avaialable = %d. total = %d", errPeersUnavailable, len(availablePeers), len(d.peers))
 				}
 
 			} else if len(d.queue.fetching) == 0 {
@@ -321,7 +322,7 @@ out:
 					if time.Since(chunk.itime) > blockTtl {
 						badPeers = append(badPeers, pid)
 						// remove peer as good peer from peer list
-						d.UnregisterPeer(pid)
+						//d.UnregisterPeer(pid)
 					}
 				}
 				d.queue.mu.Unlock()
-- 
cgit v1.2.3