refactored backoffcache and increased an internal channel size.
@@ -100,8 +100,39 @@ func (d *BackoffDiscovery) FindPeers(ctx context.Context, ns string, opts ...dis
}
c.ongoing = true
+ go findPeerDispatcher(ctx, c, pch)
+ // If it's not yet time to search again then return cached peers
+ } else if !timeExpired {
+ chLen := options.Limit
+
+ if chLen == 0 {
+ chLen = len(c.prevPeers)
+ } else if chLen > len(c.prevPeers) {
+ chLen = len(c.prevPeers)
+ }
+ pch := make(chan peer.AddrInfo, chLen)
+ for _, ai := range c.prevPeers {
+ pch <- ai
+ }
+ close(pch)
+ return pch, nil
+ }
+
+ // Setup receiver channel for receiving peers from ongoing requests
+ evtCh := make(chan peer.AddrInfo, 32)
+ pch := make(chan peer.AddrInfo, 32)
+ rcvPeers := make([]peer.AddrInfo, 0, 32)
+ for _, ai := range c.peers {
+ rcvPeers = append(rcvPeers, ai)
+ }
+ c.sendingChs[evtCh] = options.Limit
+
+ go findPeerReceiver(ctx, pch, evtCh, rcvPeers)
+
+ return pch, nil
+}
- go func() {
+func findPeerDispatcher(ctx context.Context, c *backoffCache, pch <-chan peer.AddrInfo) {
defer func() {
c.mux.Lock()
@@ -161,35 +192,9 @@ func (d *BackoffDiscovery) FindPeers(ctx context.Context, ns string, opts ...dis
return
}
}
- }()
- // If it's not yet time to search again then return cached peers
- } else if !timeExpired {
- chLen := options.Limit
-
- if chLen == 0 {
- chLen = len(c.prevPeers)
- } else if chLen > len(c.prevPeers) {
- chLen = len(c.prevPeers)
- }
- pch := make(chan peer.AddrInfo, chLen)
- for _, ai := range c.prevPeers {
- pch <- ai
- }
- close(pch)
- return pch, nil
- }
-
- // Setup receiver channel for receiving peers from ongoing requests
-
- evtCh := make(chan peer.AddrInfo, 32)
- pch := make(chan peer.AddrInfo, 8)
- rcvPeers := make([]peer.AddrInfo, 0, 32)
- for _, ai := range c.peers {
- rcvPeers = append(rcvPeers, ai)
}
- c.sendingChs[evtCh] = options.Limit
- go func() {
+func findPeerReceiver(ctx context.Context, pch, evtCh chan peer.AddrInfo, rcvPeers []peer.AddrInfo) {
defer close(pch)
for {
@@ -226,9 +231,6 @@ func (d *BackoffDiscovery) FindPeers(ctx context.Context, ns string, opts ...dis
return
}
}
- }()
-
- return pch, nil
}
func mergeAddrInfos(prevAi, newAi peer.AddrInfo) *peer.AddrInfo {
