From c9a546c310d82eb00e0e76a5e73d2ff7d601f8f0 Mon Sep 17 00:00:00 2001 From: Péter Szilágyi Date: Wed, 3 Jun 2015 14:39:21 +0300 Subject: eth/downloader: add a basic block download congestion control --- eth/downloader/downloader.go | 14 ++++++------ eth/downloader/downloader_test.go | 22 +++++++++---------- eth/downloader/peer.go | 45 ++++++++++++++++++++++++++++++++++----- eth/downloader/queue.go | 10 +++++++-- eth/downloader/queue_test.go | 31 --------------------------- 5 files changed, 67 insertions(+), 55 deletions(-) (limited to 'eth/downloader') 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)) - } -} -- cgit