Merge pull request #1701 from karalabe/eth62-sync-rebase

eth: implement eth/62 synchronization logic

1 files changed, 79 insertions, 13 deletions

diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go
index 4273b9168..8fd1f9a99 100644
--- a/eth/downloader/peer.go
+++ b/eth/downloader/peer.go
@@ -31,10 +31,16 @@ import (
 	"gopkg.in/fatih/set.v0"
 )
 
+// Hash and block fetchers belonging to eth/61 and below
 type relativeHashFetcherFn func(common.Hash) error
 type absoluteHashFetcherFn func(uint64, int) error
 type blockFetcherFn func([]common.Hash) error
 
+// Block header and body fethers belonging to eth/62 and above
+type relativeHeaderFetcherFn func(common.Hash, int, int, bool) error
+type absoluteHeaderFetcherFn func(uint64, int, int, bool) error
+type blockBodyFetcherFn func([]common.Hash) error
+
 var (
 	errAlreadyFetching   = errors.New("already fetching blocks from peer")
 	errAlreadyRegistered = errors.New("peer is already registered")
@@ -54,25 +60,37 @@ type peer struct {
 
 	ignored *set.Set // Set of hashes not to request (didn't have previously)
 
-	getRelHashes relativeHashFetcherFn // Method to retrieve a batch of hashes from an origin hash
-	getAbsHashes absoluteHashFetcherFn // Method to retrieve a batch of hashes from an absolute position
-	getBlocks    blockFetcherFn        // Method to retrieve a batch of blocks
+	getRelHashes relativeHashFetcherFn // [eth/61] Method to retrieve a batch of hashes from an origin hash
+	getAbsHashes absoluteHashFetcherFn // [eth/61] Method to retrieve a batch of hashes from an absolute position
+	getBlocks    blockFetcherFn        // [eth/61] Method to retrieve a batch of blocks
+
+	getRelHeaders  relativeHeaderFetcherFn // [eth/62] Method to retrieve a batch of headers from an origin hash
+	getAbsHeaders  absoluteHeaderFetcherFn // [eth/62] Method to retrieve a batch of headers from an absolute position
+	getBlockBodies blockBodyFetcherFn      // [eth/62] Method to retrieve a batch of block bodies
 
 	version int // Eth protocol version number to switch strategies
 }
 
 // newPeer create a new downloader peer, with specific hash and block retrieval
 // mechanisms.
-func newPeer(id string, version int, head common.Hash, getRelHashes relativeHashFetcherFn, getAbsHashes absoluteHashFetcherFn, getBlocks blockFetcherFn) *peer {
+func newPeer(id string, version int, head common.Hash,
+	getRelHashes relativeHashFetcherFn, getAbsHashes absoluteHashFetcherFn, getBlocks blockFetcherFn, // eth/61 callbacks, remove when upgrading
+	getRelHeaders relativeHeaderFetcherFn, getAbsHeaders absoluteHeaderFetcherFn, getBlockBodies blockBodyFetcherFn) *peer {
 	return &peer{
-		id:           id,
-		head:         head,
-		capacity:     1,
+		id:       id,
+		head:     head,
+		capacity: 1,
+		ignored:  set.New(),
+
 		getRelHashes: getRelHashes,
 		getAbsHashes: getAbsHashes,
 		getBlocks:    getBlocks,
-		ignored:      set.New(),
-		version:      version,
+
+		getRelHeaders:  getRelHeaders,
+		getAbsHeaders:  getAbsHeaders,
+		getBlockBodies: getBlockBodies,
+
+		version: version,
 	}
 }
 
@@ -83,8 +101,8 @@ func (p *peer) Reset() {
 	p.ignored.Clear()
 }
 
-// Fetch sends a block retrieval request to the remote peer.
-func (p *peer) Fetch(request *fetchRequest) error {
+// Fetch61 sends a block retrieval request to the remote peer.
+func (p *peer) Fetch61(request *fetchRequest) error {
 	// Short circuit if the peer is already fetching
 	if !atomic.CompareAndSwapInt32(&p.idle, 0, 1) {
 		return errAlreadyFetching
@@ -101,10 +119,28 @@ func (p *peer) Fetch(request *fetchRequest) error {
 	return nil
 }
 
-// SetIdle sets the peer to idle, allowing it to execute new retrieval requests.
+// Fetch sends a block body retrieval request to the remote peer.
+func (p *peer) Fetch(request *fetchRequest) error {
+	// Short circuit if the peer is already fetching
+	if !atomic.CompareAndSwapInt32(&p.idle, 0, 1) {
+		return errAlreadyFetching
+	}
+	p.started = time.Now()
+
+	// Convert the header set to a retrievable slice
+	hashes := make([]common.Hash, 0, len(request.Headers))
+	for _, header := range request.Headers {
+		hashes = append(hashes, header.Hash())
+	}
+	go p.getBlockBodies(hashes)
+
+	return nil
+}
+
+// SetIdle61 sets the peer to idle, allowing it to execute new retrieval requests.
 // Its block retrieval allowance will also be updated either up- or downwards,
 // depending on whether the previous fetch completed in time or not.
-func (p *peer) SetIdle() {
+func (p *peer) SetIdle61() {
 	// Update the peer's download allowance based on previous performance
 	scale := 2.0
 	if time.Since(p.started) > blockSoftTTL {
@@ -131,6 +167,36 @@ func (p *peer) SetIdle() {
 	atomic.StoreInt32(&p.idle, 0)
 }
 
+// SetIdle sets the peer to idle, allowing it to execute new retrieval requests.
+// Its block body retrieval allowance will also be updated either up- or downwards,
+// depending on whether the previous fetch completed in time or not.
+func (p *peer) SetIdle() {
+	// Update the peer's download allowance based on previous performance
+	scale := 2.0
+	if time.Since(p.started) > bodySoftTTL {
+		scale = 0.5
+		if time.Since(p.started) > bodyHardTTL {
+			scale = 1 / float64(MaxBodyFetch) // reduces capacity to 1
+		}
+	}
+	for {
+		// Calculate the new download bandwidth allowance
+		prev := atomic.LoadInt32(&p.capacity)
+		next := int32(math.Max(1, math.Min(float64(MaxBodyFetch), float64(prev)*scale)))
+
+		// Try to update the old value
+		if atomic.CompareAndSwapInt32(&p.capacity, prev, next) {
+			// If we're having problems at 1 capacity, try to find better peers
+			if next == 1 {
+				p.Demote()
+			}
+			break
+		}
+	}
+	// Set the peer to idle to allow further block requests
+	atomic.StoreInt32(&p.idle, 0)
+}
+
 // Capacity retrieves the peers block download allowance based on its previously
 // discovered bandwidth capacity.
 func (p *peer) Capacity() int {