eth, les: Refactor downloader peer to use structs

1 files changed, 58 insertions, 72 deletions

diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go
index dc8b09772..71a692688 100644
--- a/eth/downloader/peer.go
+++ b/eth/downloader/peer.go
@@ -39,16 +39,6 @@ const (
 	measurementImpact = 0.1  // The impact a single measurement has on a peer's final throughput value.
 )
 
-// Head hash and total difficulty retriever for
-type currentHeadRetrievalFn func() (common.Hash, *big.Int)
-
-// Block header and body fetchers 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
-type receiptFetcherFn func([]common.Hash) error
-type stateFetcherFn func([]common.Hash) error
-
 var (
 	errAlreadyFetching   = errors.New("already fetching blocks from peer")
 	errAlreadyRegistered = errors.New("peer is already registered")
@@ -56,7 +46,7 @@ var (
 )
 
 // peer represents an active peer from which hashes and blocks are retrieved.
-type peer struct {
+type peerConnection struct {
 	id string // Unique identifier of the peer
 
 	headerIdle  int32 // Current header activity state of the peer (idle = 0, active = 1)
@@ -78,37 +68,31 @@ type peer struct {
 
 	lacking map[common.Hash]struct{} // Set of hashes not to request (didn't have previously)
 
-	currentHead currentHeadRetrievalFn // Method to fetch the currently known head of the peer
-
-	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
-
-	getReceipts receiptFetcherFn // [eth/63] Method to retrieve a batch of block transaction receipts
-	getNodeData stateFetcherFn   // [eth/63] Method to retrieve a batch of state trie data
+	peer Peer
 
 	version int        // Eth protocol version number to switch strategies
 	log     log.Logger // Contextual logger to add extra infos to peer logs
 	lock    sync.RWMutex
 }
 
+type Peer interface {
+	Head() (common.Hash, *big.Int)
+	RequestHeadersByHash(common.Hash, int, int, bool) error
+	RequestHeadersByNumber(uint64, int, int, bool) error
+	RequestBodies([]common.Hash) error
+	RequestReceipts([]common.Hash) error
+	RequestNodeData([]common.Hash) error
+}
+
 // newPeer create a new downloader peer, with specific hash and block retrieval
 // mechanisms.
-func newPeer(id string, version int, currentHead currentHeadRetrievalFn,
-	getRelHeaders relativeHeaderFetcherFn, getAbsHeaders absoluteHeaderFetcherFn, getBlockBodies blockBodyFetcherFn,
-	getReceipts receiptFetcherFn, getNodeData stateFetcherFn, logger log.Logger) *peer {
+func newPeerConnection(id string, version int, peer Peer, logger log.Logger) *peerConnection {
 
-	return &peer{
+	return &peerConnection{
 		id:      id,
 		lacking: make(map[common.Hash]struct{}),
 
-		currentHead:    currentHead,
-		getRelHeaders:  getRelHeaders,
-		getAbsHeaders:  getAbsHeaders,
-		getBlockBodies: getBlockBodies,
-
-		getReceipts: getReceipts,
-		getNodeData: getNodeData,
+		peer: peer,
 
 		version: version,
 		log:     logger,
@@ -116,7 +100,7 @@ func newPeer(id string, version int, currentHead currentHeadRetrievalFn,
 }
 
 // Reset clears the internal state of a peer entity.
-func (p *peer) Reset() {
+func (p *peerConnection) Reset() {
 	p.lock.Lock()
 	defer p.lock.Unlock()
 
@@ -134,7 +118,7 @@ func (p *peer) Reset() {
 }
 
 // FetchHeaders sends a header retrieval request to the remote peer.
-func (p *peer) FetchHeaders(from uint64, count int) error {
+func (p *peerConnection) FetchHeaders(from uint64, count int) error {
 	// Sanity check the protocol version
 	if p.version < 62 {
 		panic(fmt.Sprintf("header fetch [eth/62+] requested on eth/%d", p.version))
@@ -146,13 +130,13 @@ func (p *peer) FetchHeaders(from uint64, count int) error {
 	p.headerStarted = time.Now()
 
 	// Issue the header retrieval request (absolut upwards without gaps)
-	go p.getAbsHeaders(from, count, 0, false)
+	go p.peer.RequestHeadersByNumber(from, count, 0, false)
 
 	return nil
 }
 
 // FetchBodies sends a block body retrieval request to the remote peer.
-func (p *peer) FetchBodies(request *fetchRequest) error {
+func (p *peerConnection) FetchBodies(request *fetchRequest) error {
 	// Sanity check the protocol version
 	if p.version < 62 {
 		panic(fmt.Sprintf("body fetch [eth/62+] requested on eth/%d", p.version))
@@ -168,13 +152,13 @@ func (p *peer) FetchBodies(request *fetchRequest) error {
 	for _, header := range request.Headers {
 		hashes = append(hashes, header.Hash())
 	}
-	go p.getBlockBodies(hashes)
+	go p.peer.RequestBodies(hashes)
 
 	return nil
 }
 
 // FetchReceipts sends a receipt retrieval request to the remote peer.
-func (p *peer) FetchReceipts(request *fetchRequest) error {
+func (p *peerConnection) FetchReceipts(request *fetchRequest) error {
 	// Sanity check the protocol version
 	if p.version < 63 {
 		panic(fmt.Sprintf("body fetch [eth/63+] requested on eth/%d", p.version))
@@ -190,13 +174,13 @@ func (p *peer) FetchReceipts(request *fetchRequest) error {
 	for _, header := range request.Headers {
 		hashes = append(hashes, header.Hash())
 	}
-	go p.getReceipts(hashes)
+	go p.peer.RequestReceipts(hashes)
 
 	return nil
 }
 
 // FetchNodeData sends a node state data retrieval request to the remote peer.
-func (p *peer) FetchNodeData(hashes []common.Hash) error {
+func (p *peerConnection) FetchNodeData(hashes []common.Hash) error {
 	// Sanity check the protocol version
 	if p.version < 63 {
 		panic(fmt.Sprintf("node data fetch [eth/63+] requested on eth/%d", p.version))
@@ -206,48 +190,50 @@ func (p *peer) FetchNodeData(hashes []common.Hash) error {
 		return errAlreadyFetching
 	}
 	p.stateStarted = time.Now()
-	go p.getNodeData(hashes)
+
+	go p.peer.RequestNodeData(hashes)
+
 	return nil
 }
 
 // SetHeadersIdle sets the peer to idle, allowing it to execute new header retrieval
 // requests. Its estimated header retrieval throughput is updated with that measured
 // just now.
-func (p *peer) SetHeadersIdle(delivered int) {
+func (p *peerConnection) SetHeadersIdle(delivered int) {
 	p.setIdle(p.headerStarted, delivered, &p.headerThroughput, &p.headerIdle)
 }
 
 // SetBlocksIdle sets the peer to idle, allowing it to execute new block retrieval
 // requests. Its estimated block retrieval throughput is updated with that measured
 // just now.
-func (p *peer) SetBlocksIdle(delivered int) {
+func (p *peerConnection) SetBlocksIdle(delivered int) {
 	p.setIdle(p.blockStarted, delivered, &p.blockThroughput, &p.blockIdle)
 }
 
 // SetBodiesIdle sets the peer to idle, allowing it to execute block body retrieval
 // requests. Its estimated body retrieval throughput is updated with that measured
 // just now.
-func (p *peer) SetBodiesIdle(delivered int) {
+func (p *peerConnection) SetBodiesIdle(delivered int) {
 	p.setIdle(p.blockStarted, delivered, &p.blockThroughput, &p.blockIdle)
 }
 
 // SetReceiptsIdle sets the peer to idle, allowing it to execute new receipt
 // retrieval requests. Its estimated receipt retrieval throughput is updated
 // with that measured just now.
-func (p *peer) SetReceiptsIdle(delivered int) {
+func (p *peerConnection) SetReceiptsIdle(delivered int) {
 	p.setIdle(p.receiptStarted, delivered, &p.receiptThroughput, &p.receiptIdle)
 }
 
 // SetNodeDataIdle sets the peer to idle, allowing it to execute new state trie
 // data retrieval requests. Its estimated state retrieval throughput is updated
 // with that measured just now.
-func (p *peer) SetNodeDataIdle(delivered int) {
+func (p *peerConnection) SetNodeDataIdle(delivered int) {
 	p.setIdle(p.stateStarted, delivered, &p.stateThroughput, &p.stateIdle)
 }
 
 // setIdle sets the peer to idle, allowing it to execute new retrieval requests.
 // Its estimated retrieval throughput is updated with that measured just now.
-func (p *peer) setIdle(started time.Time, delivered int, throughput *float64, idle *int32) {
+func (p *peerConnection) setIdle(started time.Time, delivered int, throughput *float64, idle *int32) {
 	// Irrelevant of the scaling, make sure the peer ends up idle
 	defer atomic.StoreInt32(idle, 0)
 
@@ -274,7 +260,7 @@ func (p *peer) setIdle(started time.Time, delivered int, throughput *float64, id
 
 // HeaderCapacity retrieves the peers header download allowance based on its
 // previously discovered throughput.
-func (p *peer) HeaderCapacity(targetRTT time.Duration) int {
+func (p *peerConnection) HeaderCapacity(targetRTT time.Duration) int {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
@@ -283,7 +269,7 @@ func (p *peer) HeaderCapacity(targetRTT time.Duration) int {
 
 // BlockCapacity retrieves the peers block download allowance based on its
 // previously discovered throughput.
-func (p *peer) BlockCapacity(targetRTT time.Duration) int {
+func (p *peerConnection) BlockCapacity(targetRTT time.Duration) int {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
@@ -292,7 +278,7 @@ func (p *peer) BlockCapacity(targetRTT time.Duration) int {
 
 // ReceiptCapacity retrieves the peers receipt download allowance based on its
 // previously discovered throughput.
-func (p *peer) ReceiptCapacity(targetRTT time.Duration) int {
+func (p *peerConnection) ReceiptCapacity(targetRTT time.Duration) int {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
@@ -301,7 +287,7 @@ func (p *peer) ReceiptCapacity(targetRTT time.Duration) int {
 
 // NodeDataCapacity retrieves the peers state download allowance based on its
 // previously discovered throughput.
-func (p *peer) NodeDataCapacity(targetRTT time.Duration) int {
+func (p *peerConnection) NodeDataCapacity(targetRTT time.Duration) int {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
@@ -311,7 +297,7 @@ func (p *peer) NodeDataCapacity(targetRTT time.Duration) int {
 // MarkLacking appends a new entity to the set of items (blocks, receipts, states)
 // that a peer is known not to have (i.e. have been requested before). If the
 // set reaches its maximum allowed capacity, items are randomly dropped off.
-func (p *peer) MarkLacking(hash common.Hash) {
+func (p *peerConnection) MarkLacking(hash common.Hash) {
 	p.lock.Lock()
 	defer p.lock.Unlock()
 
@@ -326,7 +312,7 @@ func (p *peer) MarkLacking(hash common.Hash) {
 
 // Lacks retrieves whether the hash of a blockchain item is on the peers lacking
 // list (i.e. whether we know that the peer does not have it).
-func (p *peer) Lacks(hash common.Hash) bool {
+func (p *peerConnection) Lacks(hash common.Hash) bool {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
@@ -337,7 +323,7 @@ func (p *peer) Lacks(hash common.Hash) bool {
 // peerSet represents the collection of active peer participating in the chain
 // download procedure.
 type peerSet struct {
-	peers       map[string]*peer
+	peers       map[string]*peerConnection
 	newPeerFeed event.Feed
 	lock        sync.RWMutex
 }
@@ -345,11 +331,11 @@ type peerSet struct {
 // newPeerSet creates a new peer set top track the active download sources.
 func newPeerSet() *peerSet {
 	return &peerSet{
-		peers: make(map[string]*peer),
+		peers: make(map[string]*peerConnection),
 	}
 }
 
-func (ps *peerSet) SubscribeNewPeers(ch chan<- *peer) event.Subscription {
+func (ps *peerSet) SubscribeNewPeers(ch chan<- *peerConnection) event.Subscription {
 	return ps.newPeerFeed.Subscribe(ch)
 }
 
@@ -370,7 +356,7 @@ func (ps *peerSet) Reset() {
 // The method also sets the starting throughput values of the new peer to the
 // average of all existing peers, to give it a realistic chance of being used
 // for data retrievals.
-func (ps *peerSet) Register(p *peer) error {
+func (ps *peerSet) Register(p *peerConnection) error {
 	// Retrieve the current median RTT as a sane default
 	p.rtt = ps.medianRTT()
 
@@ -417,7 +403,7 @@ func (ps *peerSet) Unregister(id string) error {
 }
 
 // Peer retrieves the registered peer with the given id.
-func (ps *peerSet) Peer(id string) *peer {
+func (ps *peerSet) Peer(id string) *peerConnection {
 	ps.lock.RLock()
 	defer ps.lock.RUnlock()
 
@@ -433,11 +419,11 @@ func (ps *peerSet) Len() int {
 }
 
 // AllPeers retrieves a flat list of all the peers within the set.
-func (ps *peerSet) AllPeers() []*peer {
+func (ps *peerSet) AllPeers() []*peerConnection {
 	ps.lock.RLock()
 	defer ps.lock.RUnlock()
 
-	list := make([]*peer, 0, len(ps.peers))
+	list := make([]*peerConnection, 0, len(ps.peers))
 	for _, p := range ps.peers {
 		list = append(list, p)
 	}
@@ -446,11 +432,11 @@ func (ps *peerSet) AllPeers() []*peer {
 
 // HeaderIdlePeers retrieves a flat list of all the currently header-idle peers
 // within the active peer set, ordered by their reputation.
-func (ps *peerSet) HeaderIdlePeers() ([]*peer, int) {
-	idle := func(p *peer) bool {
+func (ps *peerSet) HeaderIdlePeers() ([]*peerConnection, int) {
+	idle := func(p *peerConnection) bool {
 		return atomic.LoadInt32(&p.headerIdle) == 0
 	}
-	throughput := func(p *peer) float64 {
+	throughput := func(p *peerConnection) float64 {
 		p.lock.RLock()
 		defer p.lock.RUnlock()
 		return p.headerThroughput
@@ -460,11 +446,11 @@ func (ps *peerSet) HeaderIdlePeers() ([]*peer, int) {
 
 // BodyIdlePeers retrieves a flat list of all the currently body-idle peers within
 // the active peer set, ordered by their reputation.
-func (ps *peerSet) BodyIdlePeers() ([]*peer, int) {
-	idle := func(p *peer) bool {
+func (ps *peerSet) BodyIdlePeers() ([]*peerConnection, int) {
+	idle := func(p *peerConnection) bool {
 		return atomic.LoadInt32(&p.blockIdle) == 0
 	}
-	throughput := func(p *peer) float64 {
+	throughput := func(p *peerConnection) float64 {
 		p.lock.RLock()
 		defer p.lock.RUnlock()
 		return p.blockThroughput
@@ -474,11 +460,11 @@ func (ps *peerSet) BodyIdlePeers() ([]*peer, int) {
 
 // ReceiptIdlePeers retrieves a flat list of all the currently receipt-idle peers
 // within the active peer set, ordered by their reputation.
-func (ps *peerSet) ReceiptIdlePeers() ([]*peer, int) {
-	idle := func(p *peer) bool {
+func (ps *peerSet) ReceiptIdlePeers() ([]*peerConnection, int) {
+	idle := func(p *peerConnection) bool {
 		return atomic.LoadInt32(&p.receiptIdle) == 0
 	}
-	throughput := func(p *peer) float64 {
+	throughput := func(p *peerConnection) float64 {
 		p.lock.RLock()
 		defer p.lock.RUnlock()
 		return p.receiptThroughput
@@ -488,11 +474,11 @@ func (ps *peerSet) ReceiptIdlePeers() ([]*peer, int) {
 
 // NodeDataIdlePeers retrieves a flat list of all the currently node-data-idle
 // peers within the active peer set, ordered by their reputation.
-func (ps *peerSet) NodeDataIdlePeers() ([]*peer, int) {
-	idle := func(p *peer) bool {
+func (ps *peerSet) NodeDataIdlePeers() ([]*peerConnection, int) {
+	idle := func(p *peerConnection) bool {
 		return atomic.LoadInt32(&p.stateIdle) == 0
 	}
-	throughput := func(p *peer) float64 {
+	throughput := func(p *peerConnection) float64 {
 		p.lock.RLock()
 		defer p.lock.RUnlock()
 		return p.stateThroughput
@@ -503,11 +489,11 @@ func (ps *peerSet) NodeDataIdlePeers() ([]*peer, int) {
 // idlePeers retrieves a flat list of all currently idle peers satisfying the
 // protocol version constraints, using the provided function to check idleness.
 // The resulting set of peers are sorted by their measure throughput.
-func (ps *peerSet) idlePeers(minProtocol, maxProtocol int, idleCheck func(*peer) bool, throughput func(*peer) float64) ([]*peer, int) {
+func (ps *peerSet) idlePeers(minProtocol, maxProtocol int, idleCheck func(*peerConnection) bool, throughput func(*peerConnection) float64) ([]*peerConnection, int) {
 	ps.lock.RLock()
 	defer ps.lock.RUnlock()
 
-	idle, total := make([]*peer, 0, len(ps.peers)), 0
+	idle, total := make([]*peerConnection, 0, len(ps.peers)), 0
 	for _, p := range ps.peers {
 		if p.version >= minProtocol && p.version <= maxProtocol {
 			if idleCheck(p) {