diff options
Diffstat (limited to 'eth/downloader/peer.go')
| -rw-r--r-- | eth/downloader/peer.go | 92 |
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 { |