From 6d9018cb11c3739c66e227b11e6b17ea849555fb Mon Sep 17 00:00:00 2001 From: Richard Ramos Date: Fri, 30 Jun 2023 11:26:35 -0400 Subject: [PATCH] refactor: connect to discovered peers linearly --- waku/v2/discovery_connector.go | 27 +++++++-------------------- 1 file changed, 7 insertions(+), 20 deletions(-) diff --git a/waku/v2/discovery_connector.go b/waku/v2/discovery_connector.go index 9459dcb7..aaed82e7 100644 --- a/waku/v2/discovery_connector.go +++ b/waku/v2/discovery_connector.go @@ -250,13 +250,6 @@ func (c *PeerConnectionStrategy) workPublisher(ctx context.Context) { func (c *PeerConnectionStrategy) dialPeers(ctx context.Context) { defer c.wg.Done() - - maxGoRoutines := c.minPeers - if maxGoRoutines > 15 { - maxGoRoutines = 15 - } - - sem := make(chan struct{}, maxGoRoutines) for { select { case pi, ok := <-c.dialCh: @@ -291,20 +284,14 @@ func (c *PeerConnectionStrategy) dialPeers(ctx context.Context) { continue } - sem <- struct{}{} - c.wg.Add(1) - go func(pi peer.AddrInfo) { - defer c.wg.Done() + dialCtx, dialCtxCancel := context.WithTimeout(c.workerCtx, c.dialTimeout) + err := c.host.Connect(dialCtx, pi) + if err != nil && !errors.Is(err, context.Canceled) { + c.host.Peerstore().(peers.WakuPeerstore).AddConnFailure(pi) + c.logger.Info("connecting to peer", logging.HostID("peerID", pi.ID), zap.Error(err)) + } + dialCtxCancel() - ctx, cancel := context.WithTimeout(c.workerCtx, c.dialTimeout) - defer cancel() - err := c.host.Connect(ctx, pi) - if err != nil && !errors.Is(err, context.Canceled) { - c.host.Peerstore().(peers.WakuPeerstore).AddConnFailure(pi) - c.logger.Info("connecting to peer", logging.HostID("peerID", pi.ID), zap.Error(err)) - } - <-sem - }(pi) case <-ctx.Done(): return }