Merge pull request #8385 from Roasbeef/ping-async-dc

peer: make PingManager disconnect call async

docs/release-notes/release-notes-0.17.4.md

@@ -29,6 +29,11 @@
   FilterKnownChanIDs](https://github.com/lightningnetwork/lnd/pull/8400) by
   ensuring the `cacheMu` mutex is acquired before the main database lock.
 
+* [Prevent](https://github.com/lightningnetwork/lnd/pull/8385) ping failures
+  from [deadlocking](https://github.com/lightningnetwork/lnd/issues/8379)
+  the peer connection.
+
+
 # New Features
 ## Functional Enhancements
 ## RPC Additions
@@ -50,5 +55,8 @@
 ## Tooling and Documentation
 
 # Contributors (Alphabetical Order)
+
 * Elle Mouton
+* Keagan McClelland
+* Olaoluwa Osuntokun
 * ziggie1984

peer/brontide.go

@@ -577,7 +577,7 @@ func NewBrontide(cfg Config) *Brontide {
 			eStr := "pong response failure for %s: %v " +
 				"-- disconnecting"
 			p.log.Warnf(eStr, p, err)
-			p.Disconnect(fmt.Errorf(eStr, p, err))
+			go p.Disconnect(fmt.Errorf(eStr, p, err))
 		},
 	})
 

peer/ping_manager.go

@@ -13,7 +13,6 @@ import (
 // PingManagerConfig is a structure containing various parameters that govern
 // how the PingManager behaves.
 type PingManagerConfig struct {
-
 	// NewPingPayload is a closure that returns the payload to be packaged
 	// in the Ping message.
 	NewPingPayload func() []byte
@@ -99,16 +98,20 @@ func NewPingManager(cfg *PingManagerConfig) *PingManager {
 func (m *PingManager) Start() error {
 	var err error
 	m.started.Do(func() {
-		err = m.start()
+		m.pingTicker = time.NewTicker(m.cfg.IntervalDuration)
+		m.pingTimeout = time.NewTimer(0)
+
+		m.wg.Add(1)
+		go m.pingHandler()
 	})
 
 	return err
 }
 
-func (m *PingManager) start() error {
+// pingHandler is the main goroutine responsible for enforcing the ping/pong
-	m.pingTicker = time.NewTicker(m.cfg.IntervalDuration)
+// protocol.
-
+func (m *PingManager) pingHandler() {
-	m.pingTimeout = time.NewTimer(0)
+	defer m.wg.Done()
 	defer m.pingTimeout.Stop()
 
 	// Ensure that the pingTimeout channel is empty.
@@ -116,17 +119,13 @@ func (m *PingManager) start() error {
 		<-m.pingTimeout.C
 	}
 
-	m.wg.Add(1)
-	go func() {
-		defer m.wg.Done()
 	for {
 		select {
 		case <-m.pingTicker.C:
-				// If this occurs it means that the new ping
+			// If this occurs it means that the new ping cycle has
-				// cycle has begun while there is still an
+			// begun while there is still an outstanding ping
-				// outstanding ping awaiting a pong response.
+			// awaiting a pong response.  This should never occur,
-				// This should never occur, but if it does, it
+			// but if it does, it implies a timeout.
-				// implies a timeout.
 			if m.outstandingPongSize >= 0 {
 				e := errors.New("impossible: new ping" +
 					"in unclean state",
@@ -157,6 +156,7 @@ func (m *PingManager) start() error {
 			e := errors.New("timeout while waiting for " +
 				"pong response",
 			)
+
 			m.cfg.OnPongFailure(e)
 
 			return
@@ -164,19 +164,20 @@ func (m *PingManager) start() error {
 		case pong := <-m.pongChan:
 			pongSize := int32(len(pong.PongBytes))
 
-				// Save off values we are about to override
+			// Save off values we are about to override when we
-				// when we call resetPingState.
+			// call resetPingState.
 			expected := m.outstandingPongSize
 			lastPing := m.pingLastSend
 
 			m.resetPingState()
 
-				// If the pong we receive doesn't match the
+			// If the pong we receive doesn't match the ping we
-				// ping we sent out, then we fail out.
+			// sent out, then we fail out.
 			if pongSize != expected {
 				e := errors.New("pong response does " +
 					"not match expected size",
 				)
+
 				m.cfg.OnPongFailure(e)
 
 				return
@@ -188,13 +189,11 @@ func (m *PingManager) start() error {
 				rtt := time.Since(*lastPing)
 				m.pingTime.Store(&rtt)
 			}
+
 		case <-m.quit:
 			return
 		}
 	}
-	}()
-
-	return nil
 }
 
 // Stop interrupts the goroutines that the PingManager owns. Can only be called