From 15d3b0a05c97de456def36623caf611e41751323 Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Mon, 24 Feb 2020 10:45:53 +0800 Subject: [PATCH] les: address comments --- les/client.go | 7 +- les/commons.go | 1 - les/distributor.go | 2 +- les/peer.go | 371 ++++++++++++++++++++++-------------------- les/peer_test.go | 4 +- les/retrieve.go | 4 +- les/server.go | 5 +- les/server_handler.go | 12 +- les/test_helper.go | 20 +-- les/txrelay.go | 3 +- 10 files changed, 228 insertions(+), 201 deletions(-) diff --git a/les/client.go b/les/client.go index 392e53ad5e..dfd0909778 100644 --- a/les/client.go +++ b/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 diff --git a/les/commons.go b/les/commons.go index b402c51769..29b5b7660e 100644 --- a/les/commons.go +++ b/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 diff --git a/les/distributor.go b/les/distributor.go index bc9550614d..4d2be1b8f9 100644 --- a/les/distributor.go +++ b/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(), diff --git a/les/peer.go b/les/peer.go index 20ca0f6374..28ec201bc9 100644 --- a/les/peer.go +++ b/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 - - closed bool - client bool - lock sync.RWMutex +// 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 + 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 { - sub.registerPeer(p) - } - } else { - sub := s.(clientPeerSubscriber) - ps.cSubs = append(ps.cSubs, sub) - for _, p := range ps.clientPeers { - sub.registerPeer(p) - } + 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 *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 { - return errAlreadyRegistered - } - ps.serverPeers[peer.id] = peer - for _, sub := range ps.sSubs { - 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) - } + if _, exist := ps.peers[peer.id]; exist { + return errAlreadyRegistered + } + ps.peers[peer.id] = peer + for _, sub := range ps.subscribers { + 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] - if !ok { - return errNotRegistered - } - delete(ps.serverPeers, id) - for _, sub := range ps.sSubs { - 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, 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 } -// 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 { - ids = append(ids, id) - } - } else { - for id := range ps.clientPeers { - ids = append(ids, id) - } + for id := range ps.peers { + 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 nil - } - return ps.serverPeers[id] + return ps.peers[id] } -// Peer retrieves the registered peer with the given id. -func (ps *peerSet) clientPeer(id string) *clientPeer { +// len returns if the current number of peers in the set. +func (ps *clientPeerSet) len() int { ps.lock.RLock() defer ps.lock.RUnlock() - if ps.client { - return nil - } - return ps.clientPeers[id] + return len(ps.peers) } -// Len returns if the current number of peers in the set. -func (ps *peerSet) len() int { +// allClientPeers returns all client peers in a list. +func (ps *clientPeerSet) allPeers() []*clientPeer { ps.lock.RLock() defer ps.lock.RUnlock() - if ps.client { - return len(ps.serverPeers) - } else { - return len(ps.clientPeers) + 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) } } -// BestPeer retrieves the known peer with the currently highest total difficulty. +// 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 +} + +// 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() + + 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 +} + +// ids returns a list of all registered peer IDs +func (ps *serverPeerSet) ids() []string { + ps.lock.RLock() + defer ps.lock.RUnlock() + + var ids []string + for id := range ps.peers { + ids = append(ids, id) + } + return ids +} + +// 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 { - p.Disconnect(p2p.DiscQuitting) - } - } else { - for _, p := range ps.clientPeers { - p.Disconnect(p2p.DiscQuitting) - } + for _, p := range ps.peers { + p.Disconnect(p2p.DiscQuitting) } ps.closed = true } diff --git a/les/peer_test.go b/les/peer_test.go index c3127c62c1..59a2ad7009 100644 --- a/les/peer_test.go +++ b/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 } diff --git a/les/retrieve.go b/les/retrieve.go index 560e49afcb..5fa68b7456 100644 --- a/les/retrieve.go +++ b/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, diff --git a/les/server.go b/les/server.go index 880960330d..f72f31321a 100644 --- a/les/server.go +++ b/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 diff --git a/les/server_handler.go b/les/server_handler.go index 294d0877cf..186bdcbb03 100644 --- a/les/server_handler.go +++ b/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: diff --git a/les/test_helper.go b/les/test_helper.go index 51791d1093..d9ffe32db2 100644 --- a/les/test_helper.go +++ b/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) diff --git a/les/txrelay.go b/les/txrelay.go index 005797f5bd..595c4d5808 100644 --- a/les/txrelay.go +++ b/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 } } }