les: code refactoring (#14416)

This commit does various code refactorings:

- generalizes and moves the request retrieval/timeout/resend logic out of LesOdr
  (will be used by a subsequent PR)
- reworks the peer management logic so that all services can register with
  peerSet to get notified about added/dropped peers (also gets rid of the ugly
  getAllPeers callback in requestDistributor)
- moves peerSet, LesOdr, requestDistributor and retrieveManager initialization
  out of ProtocolManager because I believe they do not really belong there and the
  whole init process was ugly and ad-hoc

1 files changed, 32 insertions, 6 deletions

diff --git a/les/peer.go b/les/peer.go
index ab55bafe3..791d0da24 100644
--- a/les/peer.go
+++ b/les/peer.go
@@ -166,9 +166,9 @@ func (p *peer) GetRequestCost(msgcode uint64, amount int) uint64 {
 // HasBlock checks if the peer has a given block
 func (p *peer) HasBlock(hash common.Hash, number uint64) bool {
 	p.lock.RLock()
-	hashBlock := p.hasBlock
+	hasBlock := p.hasBlock
 	p.lock.RUnlock()
-	return hashBlock != nil && hashBlock(hash, number)
+	return hasBlock != nil && hasBlock(hash, number)
 }
 
 // SendAnnounce announces the availability of a number of blocks through
@@ -433,12 +433,20 @@ func (p *peer) String() string {
 	)
 }
 
+// peerSetNotify is a callback interface to notify services about added or
+// removed peers
+type peerSetNotify interface {
+	registerPeer(*peer)
+	unregisterPeer(*peer)
+}
+
 // peerSet represents the collection of active peers currently participating in
 // the Light Ethereum sub-protocol.
 type peerSet struct {
-	peers  map[string]*peer
-	lock   sync.RWMutex
-	closed bool
+	peers      map[string]*peer
+	lock       sync.RWMutex
+	notifyList []peerSetNotify
+	closed     bool
 }
 
 // newPeerSet creates a new peer set to track the active participants.
@@ -448,6 +456,17 @@ func newPeerSet() *peerSet {
 	}
 }
 
+// notify adds a service to be notified about added or removed peers
+func (ps *peerSet) notify(n peerSetNotify) {
+	ps.lock.Lock()
+	defer ps.lock.Unlock()
+
+	ps.notifyList = append(ps.notifyList, n)
+	for _, p := range ps.peers {
+		go n.registerPeer(p)
+	}
+}
+
 // Register injects a new peer into the working set, or returns an error if the
 // peer is already known.
 func (ps *peerSet) Register(p *peer) error {
@@ -462,11 +481,14 @@ func (ps *peerSet) Register(p *peer) error {
 	}
 	ps.peers[p.id] = p
 	p.sendQueue = newExecQueue(100)
+	for _, n := range ps.notifyList {
+		go n.registerPeer(p)
+	}
 	return nil
 }
 
 // Unregister removes a remote peer from the active set, disabling any further
-// actions to/from that particular entity.
+// actions to/from that particular entity. It also initiates disconnection at the networking layer.
 func (ps *peerSet) Unregister(id string) error {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
@@ -474,7 +496,11 @@ func (ps *peerSet) Unregister(id string) error {
 	if p, ok := ps.peers[id]; !ok {
 		return errNotRegistered
 	} else {
+		for _, n := range ps.notifyList {
+			go n.unregisterPeer(p)
+		}
 		p.sendQueue.quit()
+		p.Peer.Disconnect(p2p.DiscUselessPeer)
 	}
 	delete(ps.peers, id)
 	return nil