aboutsummaryrefslogtreecommitdiffstats
path: root/eth/downloader
diff options
context:
space:
mode:
authorPéter Szilágyi <peterke@gmail.com>2015-06-03 19:39:21 +0800
committerPéter Szilágyi <peterke@gmail.com>2015-06-03 19:40:11 +0800
commitc9a546c310d82eb00e0e76a5e73d2ff7d601f8f0 (patch)
tree763680b8b27c61352ad53a34a0f865aac9e6d126 /eth/downloader
parent9085b10508f1a3a5830549037f033ca58d184a0e (diff)
downloaddexon-c9a546c310d82eb00e0e76a5e73d2ff7d601f8f0.tar
dexon-c9a546c310d82eb00e0e76a5e73d2ff7d601f8f0.tar.gz
dexon-c9a546c310d82eb00e0e76a5e73d2ff7d601f8f0.tar.bz2
dexon-c9a546c310d82eb00e0e76a5e73d2ff7d601f8f0.tar.lz
dexon-c9a546c310d82eb00e0e76a5e73d2ff7d601f8f0.tar.xz
dexon-c9a546c310d82eb00e0e76a5e73d2ff7d601f8f0.tar.zst
dexon-c9a546c310d82eb00e0e76a5e73d2ff7d601f8f0.zip
eth/downloader: add a basic block download congestion control
Diffstat (limited to 'eth/downloader')
-rw-r--r--eth/downloader/downloader.go14
-rw-r--r--eth/downloader/downloader_test.go22
-rw-r--r--eth/downloader/peer.go45
-rw-r--r--eth/downloader/queue.go10
-rw-r--r--eth/downloader/queue_test.go31
5 files changed, 67 insertions, 55 deletions
diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index 85531ce15..f52a97610 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -27,9 +27,10 @@ const (
)
var (
- blockTTL = 5 * time.Second // Time it takes for a block request to time out
- crossCheckCycle = time.Second // Period after which to check for expired cross checks
- minDesiredPeerCount = 5 // Amount of peers desired to start syncing
+ blockSoftTTL = 3 * time.Second // Request completion threshold for increasing or decreasing a peer's bandwidth
+ blockHardTTL = 3 * blockSoftTTL // Maximum time allowance before a block request is considered expired
+ crossCheckCycle = time.Second // Period after which to check for expired cross checks
+ minDesiredPeerCount = 5 // Amount of peers desired to start syncing
)
var (
@@ -324,7 +325,7 @@ func (d *Downloader) fetchHashes(p *peer, h common.Hash) error {
glog.V(logger.Detail).Infof("Cross checking (%s) with %x/%x", active.id, origin, parent)
d.checks[origin] = &crossCheck{
- expire: time.Now().Add(blockTTL),
+ expire: time.Now().Add(blockSoftTTL),
parent: parent,
}
active.getBlocks([]common.Hash{origin})
@@ -429,6 +430,7 @@ out:
// Peer did deliver, but some blocks were off, penalize
glog.V(logger.Debug).Infof("Failed delivery for peer %s: %v\n", blockPack.peerId, err)
peer.Demote()
+ peer.SetIdle()
break
}
if glog.V(logger.Debug) && len(blockPack.blocks) > 0 {
@@ -444,7 +446,7 @@ out:
// that badly or poorly behave are removed from the peer set (not banned).
// Bad peers are excluded from the available peer set and therefor won't be
// reused. XXX We could re-introduce peers after X time.
- badPeers := d.queue.Expire(blockTTL)
+ badPeers := d.queue.Expire(blockHardTTL)
for _, pid := range badPeers {
// XXX We could make use of a reputation system here ranking peers
// in their performance
@@ -475,7 +477,7 @@ out:
}
// Get a possible chunk. If nil is returned no chunk
// could be returned due to no hashes available.
- request := d.queue.Reserve(peer, MaxBlockFetch)
+ request := d.queue.Reserve(peer)
if request == nil {
continue
}
diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go
index 66be1ca18..ef94ddbab 100644
--- a/eth/downloader/downloader_test.go
+++ b/eth/downloader/downloader_test.go
@@ -191,7 +191,7 @@ func (dl *downloadTester) badBlocksPeer(id string, td *big.Int, hash common.Hash
func TestDownload(t *testing.T) {
minDesiredPeerCount = 4
- blockTTL = 1 * time.Second
+ blockHardTTL = 1 * time.Second
targetBlocks := 1000
hashes := createHashes(0, targetBlocks)
@@ -240,7 +240,7 @@ func TestMissing(t *testing.T) {
func TestTaking(t *testing.T) {
minDesiredPeerCount = 4
- blockTTL = 1 * time.Second
+ blockHardTTL = 1 * time.Second
targetBlocks := 1000
hashes := createHashes(0, targetBlocks)
@@ -281,7 +281,7 @@ func TestInactiveDownloader(t *testing.T) {
func TestCancel(t *testing.T) {
minDesiredPeerCount = 4
- blockTTL = 1 * time.Second
+ blockHardTTL = 1 * time.Second
targetBlocks := 1000
hashes := createHashes(0, targetBlocks)
@@ -307,7 +307,7 @@ func TestCancel(t *testing.T) {
func TestThrottling(t *testing.T) {
minDesiredPeerCount = 4
- blockTTL = 1 * time.Second
+ blockHardTTL = 1 * time.Second
targetBlocks := 16 * blockCacheLimit
hashes := createHashes(0, targetBlocks)
@@ -461,7 +461,7 @@ func TestInvalidHashOrderAttack(t *testing.T) {
// Tests that if a malicious peer makes up a random hash chain and tries to push
// indefinitely, it actually gets caught with it.
func TestMadeupHashChainAttack(t *testing.T) {
- blockTTL = 100 * time.Millisecond
+ blockSoftTTL = 100 * time.Millisecond
crossCheckCycle = 25 * time.Millisecond
// Create a long chain of hashes without backing blocks
@@ -495,10 +495,10 @@ func TestMadeupHashChainDrippingAttack(t *testing.T) {
// Tests that if a malicious peer makes up a random block chain, and tried to
// push indefinitely, it actually gets caught with it.
func TestMadeupBlockChainAttack(t *testing.T) {
- defaultBlockTTL := blockTTL
+ defaultBlockTTL := blockSoftTTL
defaultCrossCheckCycle := crossCheckCycle
- blockTTL = 100 * time.Millisecond
+ blockSoftTTL = 100 * time.Millisecond
crossCheckCycle = 25 * time.Millisecond
// Create a long chain of blocks and simulate an invalid chain by dropping every second
@@ -516,7 +516,7 @@ func TestMadeupBlockChainAttack(t *testing.T) {
t.Fatalf("synchronisation error mismatch: have %v, want %v", err, ErrCrossCheckFailed)
}
// Ensure that a valid chain can still pass sync
- blockTTL = defaultBlockTTL
+ blockSoftTTL = defaultBlockTTL
crossCheckCycle = defaultCrossCheckCycle
tester.hashes = hashes
@@ -530,10 +530,10 @@ func TestMadeupBlockChainAttack(t *testing.T) {
// attacker make up a valid hashes for random blocks, but also forges the block
// parents to point to existing hashes.
func TestMadeupParentBlockChainAttack(t *testing.T) {
- defaultBlockTTL := blockTTL
+ defaultBlockTTL := blockSoftTTL
defaultCrossCheckCycle := crossCheckCycle
- blockTTL = 100 * time.Millisecond
+ blockSoftTTL = 100 * time.Millisecond
crossCheckCycle = 25 * time.Millisecond
// Create a long chain of blocks and simulate an invalid chain by dropping every second
@@ -550,7 +550,7 @@ func TestMadeupParentBlockChainAttack(t *testing.T) {
t.Fatalf("synchronisation error mismatch: have %v, want %v", err, ErrCrossCheckFailed)
}
// Ensure that a valid chain can still pass sync
- blockTTL = defaultBlockTTL
+ blockSoftTTL = defaultBlockTTL
crossCheckCycle = defaultCrossCheckCycle
tester.blocks = blocks
diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go
index 4abae8d5e..df54eecbd 100644
--- a/eth/downloader/peer.go
+++ b/eth/downloader/peer.go
@@ -5,10 +5,14 @@ package downloader
import (
"errors"
+ "math"
"sync"
"sync/atomic"
+ "time"
"github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/logger"
+ "github.com/ethereum/go-ethereum/logger/glog"
"gopkg.in/fatih/set.v0"
)
@@ -27,14 +31,15 @@ type peer struct {
head common.Hash // Hash of the peers latest known block
idle int32 // Current activity state of the peer (idle = 0, active = 1)
- rep int32 // Simple peer reputation (not used currently)
+ rep int32 // Simple peer reputation
- mu sync.RWMutex
+ capacity int32 // Number of blocks allowed to fetch per request
+ started time.Time // Time instance when the last fetch was started
- ignored *set.Set
+ ignored *set.Set // Set of hashes not to request (didn't have previously)
- getHashes hashFetcherFn
- getBlocks blockFetcherFn
+ getHashes hashFetcherFn // Method to retrieve a batch of hashes (mockable for testing)
+ getBlocks blockFetcherFn // Method to retrieve a batch of blocks (mockable for testing)
}
// newPeer create a new downloader peer, with specific hash and block retrieval
@@ -43,6 +48,7 @@ func newPeer(id string, head common.Hash, getHashes hashFetcherFn, getBlocks blo
return &peer{
id: id,
head: head,
+ capacity: 1,
getHashes: getHashes,
getBlocks: getBlocks,
ignored: set.New(),
@@ -52,6 +58,7 @@ func newPeer(id string, head common.Hash, getHashes hashFetcherFn, getBlocks blo
// Reset clears the internal state of a peer entity.
func (p *peer) Reset() {
atomic.StoreInt32(&p.idle, 0)
+ atomic.StoreInt32(&p.capacity, 1)
p.ignored.Clear()
}
@@ -61,6 +68,8 @@ func (p *peer) Fetch(request *fetchRequest) error {
if !atomic.CompareAndSwapInt32(&p.idle, 0, 1) {
return errAlreadyFetching
}
+ p.started = time.Now()
+
// Convert the hash set to a retrievable slice
hashes := make([]common.Hash, 0, len(request.Hashes))
for hash, _ := range request.Hashes {
@@ -72,10 +81,36 @@ func (p *peer) Fetch(request *fetchRequest) error {
}
// SetIdle 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() {
+ // Update the peer's download allowance based on previous performance
+ scale := 2.0
+ if time.Since(p.started) > blockSoftTTL {
+ scale = 0.5
+ }
+ for {
+ // Calculate the new download bandwidth allowance
+ prev := atomic.LoadInt32(&p.capacity)
+ next := int32(math.Max(1, math.Min(MaxBlockFetch, float64(prev)*scale)))
+ if scale < 1 {
+ glog.V(logger.Detail).Infof("%s: reducing block allowance from %d to %d", p.id, prev, next)
+ }
+ // Try to update the old value
+ if atomic.CompareAndSwapInt32(&p.capacity, prev, next) {
+ 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 {
+ return int(atomic.LoadInt32(&p.capacity))
+}
+
// Promote increases the peer's reputation.
func (p *peer) Promote() {
atomic.AddInt32(&p.rep, 1)
diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go
index 7ea400dc4..69d91512a 100644
--- a/eth/downloader/queue.go
+++ b/eth/downloader/queue.go
@@ -203,7 +203,7 @@ func (q *queue) TakeBlocks() []*Block {
// Reserve reserves a set of hashes for the given peer, skipping any previously
// failed download.
-func (q *queue) Reserve(p *peer, max int) *fetchRequest {
+func (q *queue) Reserve(p *peer) *fetchRequest {
q.lock.Lock()
defer q.lock.Unlock()
@@ -215,11 +215,17 @@ func (q *queue) Reserve(p *peer, max int) *fetchRequest {
if _, ok := q.pendPool[p.id]; ok {
return nil
}
+ // Calculate an upper limit on the hashes we might fetch (i.e. throttling)
+ space := len(q.blockCache) - len(q.blockPool)
+ for _, request := range q.pendPool {
+ space -= len(request.Hashes)
+ }
// Retrieve a batch of hashes, skipping previously failed ones
send := make(map[common.Hash]int)
skip := make(map[common.Hash]int)
- for len(send) < max && !q.hashQueue.Empty() {
+ capacity := p.Capacity()
+ for len(send) < space && len(send) < capacity && !q.hashQueue.Empty() {
hash, priority := q.hashQueue.Pop()
if p.ignored.Has(hash) {
skip[hash.(common.Hash)] = int(priority)
diff --git a/eth/downloader/queue_test.go b/eth/downloader/queue_test.go
index b1f3591f3..ee6141f71 100644
--- a/eth/downloader/queue_test.go
+++ b/eth/downloader/queue_test.go
@@ -1,8 +1,6 @@
package downloader
import (
- "testing"
-
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"gopkg.in/fatih/set.v0"
@@ -30,32 +28,3 @@ func createBlocksFromHashSet(hashes *set.Set) []*types.Block {
return blocks
}
-
-func TestChunking(t *testing.T) {
- queue := newQueue()
- peer1 := newPeer("peer1", common.Hash{}, nil, nil)
- peer2 := newPeer("peer2", common.Hash{}, nil, nil)
-
- // 99 + 1 (1 == known genesis hash)
- hashes := createHashes(0, 99)
- queue.Insert(hashes)
-
- chunk1 := queue.Reserve(peer1, 99)
- if chunk1 == nil {
- t.Errorf("chunk1 is nil")
- t.FailNow()
- }
- chunk2 := queue.Reserve(peer2, 99)
- if chunk2 == nil {
- t.Errorf("chunk2 is nil")
- t.FailNow()
- }
-
- if len(chunk1.Hashes) != 99 {
- t.Error("expected chunk1 hashes to be 99, got", len(chunk1.Hashes))
- }
-
- if len(chunk2.Hashes) != 1 {
- t.Error("expected chunk1 hashes to be 1, got", len(chunk2.Hashes))
- }
-}