les: address comments

les/client.go

@@ -49,6 +49,7 @@ import (
 type LightEthereum struct {
 	lesCommons
 
+	peers      *serverPeerSet
 	reqDist    *requestDistributor
 	retriever  *retrieveManager
 	odr        *LesOdr
@@ -80,7 +81,7 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
 	}
 	log.Info("Initialised chain configuration", "config", chainConfig)
 
-	peers := newPeerSet(true)
+	peers := newServerPeerSet()
 	leth := &LightEthereum{
 		lesCommons: lesCommons{
 			genesis:     genesisHash,
@@ -88,9 +89,9 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
 			chainConfig: chainConfig,
 			iConfig:     light.DefaultClientIndexerConfig,
 			chainDb:     chainDb,
-			peers:       peers,
 			closeCh:     make(chan struct{}),
 		},
+		peers:          peers,
 		eventMux:       ctx.EventMux,
 		reqDist:        newRequestDistributor(peers, &mclock.System{}),
 		accountManager: ctx.AccountManager,
@@ -225,7 +226,7 @@ func (s *LightEthereum) EventMux() *event.TypeMux           { return s.eventMux
 // network protocols to start.
 func (s *LightEthereum) Protocols() []p2p.Protocol {
 	return s.makeProtocols(ClientProtocolVersions, s.handler.runPeer, func(id enode.ID) interface{} {
-		if p := s.peers.serverPeer(peerIdToString(id)); p != nil {
+		if p := s.peers.peer(peerIdToString(id)); p != nil {
 			return p.Info()
 		}
 		return nil

les/commons.go

@@ -61,7 +61,6 @@ type lesCommons struct {
 	chainConfig                  *params.ChainConfig
 	iConfig                      *light.IndexerConfig
 	chainDb                      ethdb.Database
-	peers                        *peerSet
 	chainReader                  chainReader
 	chtIndexer, bloomTrieIndexer *core.ChainIndexer
 	oracle                       *checkpointoracle.CheckpointOracle

les/distributor.go

@@ -73,7 +73,7 @@ type distReq struct {
 }
 
 // newRequestDistributor creates a new request distributor
-func newRequestDistributor(peers *peerSet, clock mclock.Clock) *requestDistributor {
+func newRequestDistributor(peers *serverPeerSet, clock mclock.Clock) *requestDistributor {
 	d := &requestDistributor{
 		clock:    clock,
 		reqQueue: list.New(),

les/peer.go

@@ -163,11 +163,11 @@ func (p *peerCommons) queueSend(f func()) bool {
 
 // mustQueueSend starts a for loop and retry the caching if failed.
 // If the stopCh is closed, then it returns.
-func (p *peerCommons) mustQueueSend(f func(), stopCh chan struct{}) {
+func (p *peerCommons) mustQueueSend(f func()) {
 	for {
 		// Check whether the stopCh is closed.
 		select {
-		case <-stopCh:
+		case <-p.closeCh:
 			return
 		default:
 		}
@@ -198,8 +198,7 @@ func (p *peerCommons) Head() (hash common.Hash) {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
-	copy(hash[:], p.headInfo.Hash[:])
-	return hash
+	return p.headInfo.Hash
 }
 
 // Td retrieves the current total difficulty of a peer.
@@ -215,8 +214,7 @@ func (p *peerCommons) HeadAndTd() (hash common.Hash, td *big.Int) {
 	p.lock.RLock()
 	defer p.lock.RUnlock()
 
-	copy(hash[:], p.headInfo.Hash[:])
-	return hash, new(big.Int).Set(p.headInfo.Td)
+	return p.headInfo.Hash, new(big.Int).Set(p.headInfo.Td)
 }
 
 // sendReceiveHandshake exchanges handshake packet with remote peer and returns any error
@@ -805,9 +803,7 @@ func (p *clientPeer) updateCapacity(cap uint64) {
 	var kvList keyValueList
 	kvList = kvList.add("flowControl/MRR", cap)
 	kvList = kvList.add("flowControl/BL", cap*bufLimitRatio)
-
-	// todo(rjl493456442) please ensure the capacity upgrade function can be queued
-	p.mustQueueSend(func() { p.sendAnnounce(announceData{Update: kvList}) }, p.closeCh)
+	p.mustQueueSend(func() { p.sendAnnounce(announceData{Update: kvList}) })
 }
 
 // freezeClient temporarily puts the client in a frozen state which means all
@@ -900,223 +896,272 @@ func (p *clientPeer) Handshake(td *big.Int, head common.Hash, headNum uint64, ge
 	})
 }
 
-// serverPeerSubscriber is a callback interface to notify services about added or
+// serverPeerSubscriber is an interface to notify services about added or
 // removed server peers
 type serverPeerSubscriber interface {
 	registerPeer(*serverPeer)
 	unregisterPeer(*serverPeer)
 }
 
-// clientPeerSubscriber is a callback interface to notify services about added or
+// clientPeerSubscriber is an interface to notify services about added or
 // removed client peers
 type clientPeerSubscriber interface {
 	registerPeer(*clientPeer)
 	unregisterPeer(*clientPeer)
 }
 
-// peerSet represents the collection of active peers currently participating in
-// the Light Ethereum sub-protocol.
-type peerSet struct {
-	serverPeers map[string]*serverPeer
-	clientPeers map[string]*clientPeer
-
-	// sSubs is a batch of subscribers and peerset will notify these
-	// subscribers when the peerset changes(new server peer is added
-	// or removed)
-	sSubs []serverPeerSubscriber
-
-	// cSubs is a batch of subscribers and peerset will notify these
-	// subscribers when the peerset changes(new client peer is added
-	// or removed)
-	cSubs []clientPeerSubscriber
-
+// clientPeerSet represents the set of active client peers currently
+// participating in the Light Ethereum sub-protocol.
+type clientPeerSet struct {
+	peers map[string]*clientPeer
+	// subscribers is a batch of subscribers and peerset will notify
+	// these subscribers when the peerset changes(new client peer is
+	// added or removed)
+	subscribers []clientPeerSubscriber
 	closed      bool
-	client bool
 	lock        sync.RWMutex
 }
 
-// newPeerSet creates a new peer set to track the active participants.
-func newPeerSet(client bool) *peerSet {
-	set := &peerSet{client: client}
-	if client {
-		set.serverPeers = make(map[string]*serverPeer)
-	} else {
-		set.clientPeers = make(map[string]*clientPeer)
-	}
-	return set
+// newClientPeerSet creates a new peer set to track the client peers.
+func newClientPeerSet() *clientPeerSet {
+	return &clientPeerSet{peers: make(map[string]*clientPeer)}
 }
 
 // subscribe adds a service to be notified about added or removed
 // peers and also register all active peers into the given service.
-func (ps *peerSet) subscribe(s interface{}) {
+func (ps *clientPeerSet) subscribe(sub clientPeerSubscriber) {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
 
-	if ps.client {
-		sub := s.(serverPeerSubscriber)
-		ps.sSubs = append(ps.sSubs, sub)
-		for _, p := range ps.serverPeers {
+	ps.subscribers = append(ps.subscribers, sub)
+	for _, p := range ps.peers {
 		sub.registerPeer(p)
 	}
-	} else {
-		sub := s.(clientPeerSubscriber)
-		ps.cSubs = append(ps.cSubs, sub)
-		for _, p := range ps.clientPeers {
-			sub.registerPeer(p)
-		}
-	}
 }
 
 // unSubscribe removes the specified service from the subscriber pool.
-func (ps *peerSet) unSubscribe(s interface{}) {
+func (ps *clientPeerSet) unSubscribe(sub clientPeerSubscriber) {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
 
-	if ps.client {
-		sub := s.(serverPeerSubscriber)
-		for i, ss := range ps.sSubs {
-			if ss == sub {
-				ps.sSubs = append(ps.sSubs[:i], ps.sSubs[i+1:]...)
-			}
-		}
-	} else {
-		sub := s.(clientPeerSubscriber)
-		for i, cs := range ps.cSubs {
-			if cs == sub {
-				ps.cSubs = append(ps.cSubs[:i], ps.cSubs[i+1:]...)
-			}
+	for i, s := range ps.subscribers {
+		if s == sub {
+			ps.subscribers = append(ps.subscribers[:i], ps.subscribers[i+1:]...)
+			return
 		}
 	}
 }
 
-// Register injects a new peer into the working set, or returns an error if the
+// register adds a new peer into the peer set, or returns an error if the
 // peer is already known.
-func (ps *peerSet) register(p interface{}) error {
+func (ps *clientPeerSet) register(peer *clientPeer) error {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
 
 	if ps.closed {
 		return errClosed
 	}
-	if ps.client {
-		peer := p.(*serverPeer)
-		if _, exist := ps.serverPeers[peer.id]; exist {
+	if _, exist := ps.peers[peer.id]; exist {
 		return errAlreadyRegistered
 	}
-		ps.serverPeers[peer.id] = peer
-		for _, sub := range ps.sSubs {
+	ps.peers[peer.id] = peer
+	for _, sub := range ps.subscribers {
 		sub.registerPeer(peer)
 	}
-	} else {
-		peer := p.(*clientPeer)
-		if _, exist := ps.clientPeers[peer.id]; exist {
-			return errAlreadyRegistered
-		}
-		ps.clientPeers[peer.id] = peer
-		for _, sub := range ps.cSubs {
-			sub.registerPeer(peer)
-		}
-	}
 	return nil
 }
 
-// Unregister removes a remote peer from the active set, disabling any further
-// actions to/from that particular entity. It also initiates disconnection at the networking layer.
-func (ps *peerSet) unregister(id string) error {
+// unregister removes a remote peer from the peer set, disabling any further
+// actions to/from that particular entity. It also initiates disconnection
+// at the networking layer.
+func (ps *clientPeerSet) unregister(id string) error {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
 
-	if ps.client {
-		p, ok := ps.serverPeers[id]
+	p, ok := ps.peers[id]
 	if !ok {
 		return errNotRegistered
 	}
-		delete(ps.serverPeers, id)
-		for _, sub := range ps.sSubs {
+	delete(ps.peers, id)
+	for _, sub := range ps.subscribers {
 		sub.unregisterPeer(p)
 	}
-		p.Peer.Disconnect(p2p.DiscUselessPeer)
-	} else {
-		p, ok := ps.clientPeers[id]
-		if !ok {
-			return errNotRegistered
-		}
-		delete(ps.clientPeers, id)
-		for _, sub := range ps.cSubs {
-			sub.unregisterPeer(p)
-		}
-		p.Peer.Disconnect(p2p.DiscUselessPeer)
-	}
+	p.Peer.Disconnect(p2p.DiscRequested)
 	return nil
 }
 
-// AllPeerIDs returns a list of all registered peer IDs
-func (ps *peerSet) allPeerIds() []string {
+// ids returns a list of all registered peer IDs
+func (ps *clientPeerSet) ids() []string {
 	ps.lock.RLock()
 	defer ps.lock.RUnlock()
 
 	var ids []string
-	if ps.client {
-		for id := range ps.serverPeers {
+	for id := range ps.peers {
 		ids = append(ids, id)
 	}
-	} else {
-		for id := range ps.clientPeers {
-			ids = append(ids, id)
-		}
-	}
 	return ids
 }
 
-// Peer retrieves the registered peer with the given id.
-func (ps *peerSet) serverPeer(id string) *serverPeer {
+// peer retrieves the registered peer with the given id.
+func (ps *clientPeerSet) peer(id string) *clientPeer {
 	ps.lock.RLock()
 	defer ps.lock.RUnlock()
 
-	if !ps.client {
+	return ps.peers[id]
+}
+
+// len returns if the current number of peers in the set.
+func (ps *clientPeerSet) len() int {
+	ps.lock.RLock()
+	defer ps.lock.RUnlock()
+
+	return len(ps.peers)
+}
+
+// allClientPeers returns all client peers in a list.
+func (ps *clientPeerSet) allPeers() []*clientPeer {
+	ps.lock.RLock()
+	defer ps.lock.RUnlock()
+
+	list := make([]*clientPeer, 0, len(ps.peers))
+	for _, p := range ps.peers {
+		list = append(list, p)
+	}
+	return list
+}
+
+// close disconnects all peers. No new peers can be registered
+// after close has returned.
+func (ps *clientPeerSet) close() {
+	ps.lock.Lock()
+	defer ps.lock.Unlock()
+
+	for _, p := range ps.peers {
+		p.Disconnect(p2p.DiscQuitting)
+	}
+	ps.closed = true
+}
+
+// serverPeerSet represents the set of active server peers currently
+// participating in the Light Ethereum sub-protocol.
+type serverPeerSet struct {
+	peers map[string]*serverPeer
+	// subscribers is a batch of subscribers and peerset will notify
+	// these subscribers when the peerset changes(new server peer is
+	// added or removed)
+	subscribers []serverPeerSubscriber
+	closed      bool
+	lock        sync.RWMutex
+}
+
+// newServerPeerSet creates a new peer set to track the active server peers.
+func newServerPeerSet() *serverPeerSet {
+	return &serverPeerSet{peers: make(map[string]*serverPeer)}
+}
+
+// subscribe adds a service to be notified about added or removed
+// peers and also register all active peers into the given service.
+func (ps *serverPeerSet) subscribe(sub serverPeerSubscriber) {
+	ps.lock.Lock()
+	defer ps.lock.Unlock()
+
+	ps.subscribers = append(ps.subscribers, sub)
+	for _, p := range ps.peers {
+		sub.registerPeer(p)
+	}
+}
+
+// unSubscribe removes the specified service from the subscriber pool.
+func (ps *serverPeerSet) unSubscribe(sub serverPeerSubscriber) {
+	ps.lock.Lock()
+	defer ps.lock.Unlock()
+
+	for i, s := range ps.subscribers {
+		if s == sub {
+			ps.subscribers = append(ps.subscribers[:i], ps.subscribers[i+1:]...)
+			return
+		}
+	}
+}
+
+// register adds a new server peer into the set, or returns an error if the
+// peer is already known.
+func (ps *serverPeerSet) register(peer *serverPeer) error {
+	ps.lock.Lock()
+	defer ps.lock.Unlock()
+
+	if ps.closed {
+		return errClosed
+	}
+	if _, exist := ps.peers[peer.id]; exist {
+		return errAlreadyRegistered
+	}
+	ps.peers[peer.id] = peer
+	for _, sub := range ps.subscribers {
+		sub.registerPeer(peer)
+	}
 	return nil
-	}
-	return ps.serverPeers[id]
 }
 
-// Peer retrieves the registered peer with the given id.
-func (ps *peerSet) clientPeer(id string) *clientPeer {
-	ps.lock.RLock()
-	defer ps.lock.RUnlock()
+// unregister removes a remote peer from the active set, disabling any further
+// actions to/from that particular entity. It also initiates disconnection at
+// the networking layer.
+func (ps *serverPeerSet) unregister(id string) error {
+	ps.lock.Lock()
+	defer ps.lock.Unlock()
 
-	if ps.client {
+	p, ok := ps.peers[id]
+	if !ok {
+		return errNotRegistered
+	}
+	delete(ps.peers, id)
+	for _, sub := range ps.subscribers {
+		sub.unregisterPeer(p)
+	}
+	p.Peer.Disconnect(p2p.DiscRequested)
 	return nil
-	}
-	return ps.clientPeers[id]
 }
 
-// Len returns if the current number of peers in the set.
-func (ps *peerSet) len() int {
+// ids returns a list of all registered peer IDs
+func (ps *serverPeerSet) ids() []string {
 	ps.lock.RLock()
 	defer ps.lock.RUnlock()
 
-	if ps.client {
-		return len(ps.serverPeers)
-	} else {
-		return len(ps.clientPeers)
+	var ids []string
+	for id := range ps.peers {
+		ids = append(ids, id)
 	}
+	return ids
 }
 
-// BestPeer retrieves the known peer with the currently highest total difficulty.
+// peer retrieves the registered peer with the given id.
+func (ps *serverPeerSet) peer(id string) *serverPeer {
+	ps.lock.RLock()
+	defer ps.lock.RUnlock()
+
+	return ps.peers[id]
+}
+
+// len returns if the current number of peers in the set.
+func (ps *serverPeerSet) len() int {
+	ps.lock.RLock()
+	defer ps.lock.RUnlock()
+
+	return len(ps.peers)
+}
+
+// bestPeer retrieves the known peer with the currently highest total difficulty.
 // If the peerset is "client peer set", then nothing meaningful will return. The
 // reason is client peer never send back their latest status to server.
-func (ps *peerSet) bestPeer() *serverPeer {
+func (ps *serverPeerSet) bestPeer() *serverPeer {
 	ps.lock.RLock()
 	defer ps.lock.RUnlock()
 
-	if !ps.client {
-		return nil
-	}
 	var (
 		bestPeer *serverPeer
 		bestTd   *big.Int
 	)
-	for _, p := range ps.serverPeers {
+	for _, p := range ps.peers {
 		if td := p.Td(); bestTd == nil || td.Cmp(bestTd) > 0 {
 			bestPeer, bestTd = p, td
 		}
@@ -1125,49 +1170,25 @@ func (ps *peerSet) bestPeer() *serverPeer {
 }
 
 // allServerPeers returns all server peers in a list.
-func (ps *peerSet) allServerPeers() []*serverPeer {
+func (ps *serverPeerSet) allPeers() []*serverPeer {
 	ps.lock.RLock()
 	defer ps.lock.RUnlock()
 
-	if !ps.client {
-		return nil
-	}
-	list := make([]*serverPeer, 0, len(ps.serverPeers))
-	for _, p := range ps.serverPeers {
+	list := make([]*serverPeer, 0, len(ps.peers))
+	for _, p := range ps.peers {
 		list = append(list, p)
 	}
 	return list
 }
 
-// allClientPeers returns all client peers in a list.
-func (ps *peerSet) allClientPeers() []*clientPeer {
-	ps.lock.RLock()
-	defer ps.lock.RUnlock()
-
-	if ps.client {
-		return nil
-	}
-	list := make([]*clientPeer, 0, len(ps.clientPeers))
-	for _, p := range ps.clientPeers {
-		list = append(list, p)
-	}
-	return list
-}
-
-// Close disconnects all peers.
-// No new peers can be registered after Close has returned.
-func (ps *peerSet) close() {
+// close disconnects all peers. No new peers can be registered
+// after close has returned.
+func (ps *serverPeerSet) close() {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
 
-	if ps.client {
-		for _, p := range ps.serverPeers {
+	for _, p := range ps.peers {
 		p.Disconnect(p2p.DiscQuitting)
 	}
-	} else {
-		for _, p := range ps.clientPeers {
-			p.Disconnect(p2p.DiscQuitting)
-		}
-	}
 	ps.closed = true
 }

les/peer_test.go

@@ -45,11 +45,11 @@ func (t *testServerPeerSub) registerPeer(p *serverPeer)   { t.regCh <- p }
 func (t *testServerPeerSub) unregisterPeer(p *serverPeer) { t.unregCh <- p }
 
 func TestPeerSubscription(t *testing.T) {
-	peers := newPeerSet(true)
+	peers := newServerPeerSet()
 	defer peers.close()
 
 	checkIds := func(expect []string) {
-		given := peers.allPeerIds()
+		given := peers.ids()
 		if len(given) == 0 && len(expect) == 0 {
 			return
 		}

les/retrieve.go

@@ -38,7 +38,7 @@ var (
 // matching replies by request ID and handles timeouts and resends if necessary.
 type retrieveManager struct {
 	dist       *requestDistributor
-	peers      *peerSet
+	peers      *serverPeerSet
 	serverPool peerSelector
 
 	lock     sync.RWMutex
@@ -99,7 +99,7 @@ const (
 )
 
 // newRetrieveManager creates the retrieve manager
-func newRetrieveManager(peers *peerSet, dist *requestDistributor, serverPool peerSelector) *retrieveManager {
+func newRetrieveManager(peers *serverPeerSet, dist *requestDistributor, serverPool peerSelector) *retrieveManager {
 	return &retrieveManager{
 		peers:      peers,
 		dist:       dist,

les/server.go

@@ -40,6 +40,7 @@ type LesServer struct {
 	lesCommons
 
 	archiveMode bool // Flag whether the ethereum node runs in archive mode.
+	peers       *clientPeerSet
 	handler     *serverHandler
 	lesTopics   []discv5.Topic
 	privateKey  *ecdsa.PrivateKey
@@ -75,13 +76,13 @@ func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) {
 			chainConfig:      e.BlockChain().Config(),
 			iConfig:          light.DefaultServerIndexerConfig,
 			chainDb:          e.ChainDb(),
-			peers:            newPeerSet(false),
 			chainReader:      e.BlockChain(),
 			chtIndexer:       light.NewChtIndexer(e.ChainDb(), nil, params.CHTFrequency, params.HelperTrieProcessConfirmations),
 			bloomTrieIndexer: light.NewBloomTrieIndexer(e.ChainDb(), nil, params.BloomBitsBlocks, params.BloomTrieFrequency),
 			closeCh:          make(chan struct{}),
 		},
 		archiveMode:  e.ArchiveMode(),
+		peers:        newClientPeerSet(),
 		lesTopics:    lesTopics,
 		fcManager:    flowcontrol.NewClientManager(nil, &mclock.System{}),
 		servingQueue: newServingQueue(int64(time.Millisecond*10), float64(config.LightServ)/100),
@@ -152,7 +153,7 @@ func (s *LesServer) APIs() []rpc.API {
 
 func (s *LesServer) Protocols() []p2p.Protocol {
 	ps := s.makeProtocols(ServerProtocolVersions, s.handler.runPeer, func(id enode.ID) interface{} {
-		if p := s.peers.clientPeer(peerIdToString(id)); p != nil {
+		if p := s.peers.peer(peerIdToString(id)); p != nil {
 			return p.Info()
 		}
 		return nil

les/server_handler.go

@@ -266,7 +266,7 @@ func (h *serverHandler) handleMsg(p *clientPeer, wg *sync.WaitGroup) error {
 					default:
 					}
 				}
-			}, p.closeCh)
+			})
 		}
 	}
 	switch msg.Code {
@@ -914,7 +914,7 @@ func (h *serverHandler) broadcastHeaders() {
 	for {
 		select {
 		case ev := <-headCh:
-			peers := h.server.peers.allClientPeers()
+			peers := h.server.peers.allPeers()
 			if len(peers) == 0 {
 				continue
 			}
@@ -940,14 +940,18 @@ func (h *serverHandler) broadcastHeaders() {
 				p := p
 				switch p.announceType {
 				case announceTypeSimple:
-					p.mustQueueSend(func() { p.sendAnnounce(announce) }, p.closeCh)
+					if !p.queueSend(func() { p.sendAnnounce(announce) }) {
+						log.Debug("Drop announcement because queue is full", "number", number, "hash", hash)
+					}
 				case announceTypeSigned:
 					if !signed {
 						signedAnnounce = announce
 						signedAnnounce.sign(h.server.privateKey)
 						signed = true
 					}
-					p.mustQueueSend(func() { p.sendAnnounce(signedAnnounce) }, p.closeCh)
+					if !p.queueSend(func() { p.sendAnnounce(signedAnnounce) }) {
+						log.Debug("Drop announcement because queue is full", "number", number, "hash", hash)
+					}
 				}
 			}
 		case <-h.closeCh:

les/test_helper.go

@@ -166,7 +166,7 @@ func testIndexers(db ethdb.Database, odr light.OdrBackend, config *light.Indexer
 	return indexers[:]
 }
 
-func newTestClientHandler(backend *backends.SimulatedBackend, odr *LesOdr, indexers []*core.ChainIndexer, db ethdb.Database, peers *peerSet, ulcServers []string, ulcFraction int) *clientHandler {
+func newTestClientHandler(backend *backends.SimulatedBackend, odr *LesOdr, indexers []*core.ChainIndexer, db ethdb.Database, peers *serverPeerSet, ulcServers []string, ulcFraction int) *clientHandler {
 	var (
 		evmux  = new(event.TypeMux)
 		engine = ethash.NewFaker()
@@ -206,9 +206,9 @@ func newTestClientHandler(backend *backends.SimulatedBackend, odr *LesOdr, index
 			chainDb:     db,
 			oracle:      oracle,
 			chainReader: chain,
-			peers:       peers,
 			closeCh:     make(chan struct{}),
 		},
+		peers:      peers,
 		reqDist:    odr.retriever.dist,
 		retriever:  odr.retriever,
 		odr:        odr,
@@ -224,7 +224,7 @@ func newTestClientHandler(backend *backends.SimulatedBackend, odr *LesOdr, index
 	return client.handler
 }
 
-func newTestServerHandler(blocks int, indexers []*core.ChainIndexer, db ethdb.Database, peers *peerSet, clock mclock.Clock) (*serverHandler, *backends.SimulatedBackend) {
+func newTestServerHandler(blocks int, indexers []*core.ChainIndexer, db ethdb.Database, peers *clientPeerSet, clock mclock.Clock) (*serverHandler, *backends.SimulatedBackend) {
 	var (
 		gspec = core.Genesis{
 			Config:   params.AllEthashProtocolChanges,
@@ -269,9 +269,9 @@ func newTestServerHandler(blocks int, indexers []*core.ChainIndexer, db ethdb.Da
 			chainDb:     db,
 			chainReader: simulation.Blockchain(),
 			oracle:      oracle,
-			peers:       peers,
 			closeCh:     make(chan struct{}),
 		},
+		peers:        peers,
 		servingQueue: newServingQueue(int64(time.Millisecond*10), 1),
 		defParams: flowcontrol.ServerParams{
 			BufLimit:    testBufLimit,
@@ -446,7 +446,7 @@ func newServerEnv(t *testing.T, blocks int, protocol int, callback indexerCallba
 	if simClock {
 		clock = &mclock.Simulated{}
 	}
-	handler, b := newTestServerHandler(blocks, indexers, db, newPeerSet(false), clock)
+	handler, b := newTestServerHandler(blocks, indexers, db, newClientPeerSet(), clock)
 
 	var peer *testPeer
 	if newPeer {
@@ -485,14 +485,14 @@ func newServerEnv(t *testing.T, blocks int, protocol int, callback indexerCallba
 
 func newClientServerEnv(t *testing.T, blocks int, protocol int, callback indexerCallback, ulcServers []string, ulcFraction int, simClock bool, connect bool) (*testServer, *testClient, func()) {
 	sdb, cdb := rawdb.NewMemoryDatabase(), rawdb.NewMemoryDatabase()
-	speers, cPeers := newPeerSet(false), newPeerSet(true)
+	speers, cpeers := newServerPeerSet(), newClientPeerSet()
 
 	var clock mclock.Clock = &mclock.System{}
 	if simClock {
 		clock = &mclock.Simulated{}
 	}
-	dist := newRequestDistributor(cPeers, clock)
-	rm := newRetrieveManager(cPeers, dist, nil)
+	dist := newRequestDistributor(speers, clock)
+	rm := newRetrieveManager(speers, dist, nil)
 	odr := NewLesOdr(cdb, light.TestClientIndexerConfig, rm)
 
 	sindexers := testIndexers(sdb, nil, light.TestServerIndexerConfig)
@@ -502,8 +502,8 @@ func newClientServerEnv(t *testing.T, blocks int, protocol int, callback indexer
 	ccIndexer, cbIndexer, cbtIndexer := cIndexers[0], cIndexers[1], cIndexers[2]
 	odr.SetIndexers(ccIndexer, cbIndexer, cbtIndexer)
 
-	server, b := newTestServerHandler(blocks, sindexers, sdb, speers, clock)
-	client := newTestClientHandler(b, odr, cIndexers, cdb, cPeers, ulcServers, ulcFraction)
+	server, b := newTestServerHandler(blocks, sindexers, sdb, cpeers, clock)
+	client := newTestClientHandler(b, odr, cIndexers, cdb, speers, ulcServers, ulcFraction)
 
 	scIndexer.Start(server.blockchain)
 	sbIndexer.Start(server.blockchain)

les/txrelay.go

@@ -41,7 +41,7 @@ type lesTxRelay struct {
 	retriever *retrieveManager
 }
 
-func newLesTxRelay(ps *peerSet, retriever *retrieveManager) *lesTxRelay {
+func newLesTxRelay(ps *serverPeerSet, retriever *retrieveManager) *lesTxRelay {
 	r := &lesTxRelay{
 		txSent:    make(map[common.Hash]*ltrInfo),
 		txPending: make(map[common.Hash]struct{}),
@@ -75,6 +75,7 @@ func (ltrx *lesTxRelay) unregisterPeer(p *serverPeer) {
 		if peer == p {
 			// Remove from the peer list
 			ltrx.peerList = append(ltrx.peerList[:i], ltrx.peerList[i+1:]...)
+			return
 		}
 	}
 }