Merge pull request #767 from iotaledger/fix/peering-lockups

Fix peering lockups
iotaledger · Feb 21, 2024 · 8e97dc4 · 8e97dc4
2 parents 09347be + 9f0eeb9
commit 8e97dc4
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 12 deletions.
diff --git a/pkg/network/p2p/autopeering/autopeering.go b/pkg/network/p2p/autopeering/autopeering.go
@@ -213,6 +213,11 @@ func (m *Manager) discoverAndDialPeers() {
 	}
 
 	for peerAddrInfo := range peerChan {
+		if m.ctx.Err() != nil {
+			m.logger.LogDebug("Context is done, stopping dialing new peers")
+			return
+		}
+
 		if peersToFind <= 0 {
 			m.logger.LogDebug("Enough new autopeering peers connected")
 			return

diff --git a/pkg/network/p2p/manager.go b/pkg/network/p2p/manager.go
@@ -38,6 +38,8 @@ type Manager struct {
 	libp2pHost host.Host
 	peerDB     *network.DB
 
+	ctx context.Context
+
 	logger log.Logger
 
 	shutdownMutex syncutils.RWMutex
@@ -50,8 +52,6 @@ type Manager struct {
 
 	autoPeering   *autopeering.Manager
 	manualPeering *manualpeering.Manager
-
-	allowPeerMutex syncutils.Mutex
 }
 
 var _ network.Manager = (*Manager)(nil)
@@ -116,9 +116,6 @@ func (m *Manager) DialPeer(ctx context.Context, peer *network.Peer) error {
 		return ierrors.Wrapf(network.ErrDuplicatePeer, "peer %s already exists", peer.ID)
 	}
 
-	m.allowPeerMutex.Lock()
-	defer m.allowPeerMutex.Unlock()
-
 	if !m.allowPeer(peer.ID) {
 		return ierrors.Wrapf(network.ErrMaxAutopeeringPeersReached, "peer %s is not allowed", peer.ID)
 	}
@@ -158,6 +155,8 @@ func (m *Manager) DialPeer(ctx context.Context, peer *network.Peer) error {
 
 // Start starts the manager and initiates manual- and autopeering.
 func (m *Manager) Start(ctx context.Context, networkID string) error {
+	m.ctx = ctx
+
 	m.manualPeering.Start()
 
 	if m.autoPeering.MaxNeighbors() > 0 {
@@ -281,8 +280,12 @@ func (m *Manager) handleStream(stream p2pnetwork.Stream) {
 		return
 	}
 
-	m.allowPeerMutex.Lock()
-	defer m.allowPeerMutex.Unlock()
+	if m.ctx.Err() != nil {
+		m.logger.LogDebugf("aborting handling stream, context is done")
+		m.closeStream(stream)
+
+		return
+	}
 
 	peerID := stream.Conn().RemotePeer()
 
@@ -314,7 +317,7 @@ func (m *Manager) handleStream(stream p2pnetwork.Stream) {
 		return
 	}
 
-	if err := m.addNeighbor(context.Background(), networkPeer, ps); err != nil {
+	if err := m.addNeighbor(m.ctx, networkPeer, ps); err != nil {
 		m.logger.LogErrorf("failed to add neighbor, peerID: %s, error: %s", peerID, err)
 		m.closeStream(stream)
 
@@ -352,7 +355,9 @@ func (m *Manager) addNeighbor(ctx context.Context, peer *network.Peer, ps *Packe
 	}
 
 	firstPacketReceivedCtx, firstPacketReceivedCancel := context.WithDeadline(ctx, time.Now().Add(5*time.Second))
-	// create and add the neighbor
+	defer firstPacketReceivedCancel()
+
+	var innerErr error
 	nbr := newNeighbor(m.logger, peer, ps, func(nbr *neighbor, packet proto.Message) {
 		m.protocolHandlerMutex.RLock()
 		defer m.protocolHandlerMutex.RUnlock()
@@ -392,7 +397,7 @@ func (m *Manager) addNeighbor(ctx context.Context, peer *network.Peer, ps *Packe
 		return ierrors.WithStack(network.ErrFirstPacketNotReceived)
 	}
 
-	return nil
+	return innerErr
 }
 
 func (m *Manager) NeighborExists(id peer.ID) bool {
@@ -430,8 +435,6 @@ func (m *Manager) dropAllNeighbors() {
 }
 
 func (m *Manager) allowPeer(id peer.ID) (allow bool) {
-	// This should always be called from within the allowPeerMutex lock
-
 	// Always allow manual peers
 	if m.manualPeering.IsPeerKnown(id) {
 		m.logger.LogDebugf("Allow manual peer %s", id)