Repository: decred/dcrd
Pull request #2254 (2020-07-09 18:57:42)

connmgr: Remain responsive with simul failed conns.
This modifies the connection manager handler for failed connections so that it remains responsive to requests when there are multiple simultaneous failed connections in the retry state and adds a test to ensure proper functionality.

While here it also updates a few formatting nits in the server code and modifies the connection manager test code to use `t.Fatal` instead of `t.Fatalf` when the message has no parameters.



Patch:
diff --git a/connmgr/connmanager.go b/connmgr/connmanager.go
index 4b15e258..7efbcb14 100644
--- a/connmgr/connmanager.go
+++ b/connmgr/connmanager.go
@@ -250,24 +250,26 @@ func (cm *ConnManager) handleFailedConn(ctx context.Context, c *ConnReq) {
 			d = maxRetryDuration
 		}
 		log.Debugf("Retrying connection to %v in %v", c, d)
-		select {
-		case <-time.After(d):
-			go cm.Connect(ctx, c)
-		case <-cm.quit:
-			return
-		}
+		go func() {
+			select {
+			case <-time.After(d):
+				cm.Connect(ctx, c)
+			case <-cm.quit:
+			}
+		}()
 	} else if cm.cfg.GetNewAddress != nil {
 		cm.failedAttempts++
 		if cm.failedAttempts >= maxFailedAttempts {
 			log.Debugf("Max failed connection attempts reached: [%d] "+
 				"-- retrying connection in: %v", maxFailedAttempts,
 				cm.cfg.RetryDuration)
-			select {
-			case <-time.After(cm.cfg.RetryDuration):
-				go cm.newConnReq(ctx)
-			case <-cm.quit:
-				return
-			}
+			go func() {
+				select {
+				case <-time.After(cm.cfg.RetryDuration):
+					cm.newConnReq(ctx)
+				case <-cm.quit:
+				}
+			}()
 		} else {
 			go cm.newConnReq(ctx)
 		}
diff --git a/server.go b/server.go
index b51c8521..9ec45fa1 100644
--- a/server.go
+++ b/server.go
@@ -2024,6 +2024,7 @@ func (s *server) handleQuery(state *peerState, querymsg interface{}) {
 				Permanent: msg.permanent,
 			})
 		msg.reply <- nil
+
 	case removeNodeMsg:
 		found := disconnectPeer(state.persistentPeers, msg.cmp, func(sp *serverPeer) {
 			// Keep group counts ok since we remove from
@@ -2045,6 +2046,7 @@ func (s *server) handleQuery(state *peerState, querymsg interface{}) {
 		} else {
 			msg.reply <- errors.New("peer not found")
 		}
+
 	case cancelPendingMsg:
 		netAddr, err := addrStringToNetAddr(msg.addr)
 		if err != nil {
@@ -2052,6 +2054,7 @@ func (s *server) handleQuery(state *peerState, querymsg interface{}) {
 			return
 		}
 		msg.reply <- s.connManager.CancelPending(netAddr)
+
 	case getOutboundGroup:
 		count, ok := state.outboundGroups[msg.key]
 		if ok {
@@ -2059,7 +2062,7 @@ func (s *server) handleQuery(state *peerState, querymsg interface{}) {
 		} else {
 			msg.reply <- 0
 		}
-	// Request a list of the persistent (added) peers.
+
 	case getAddedNodesMsg:
 		// Respond with a slice of the relevant peers.
 		peers := make([]*serverPeer, 0, len(state.persistentPeers))
@@ -2067,6 +2070,7 @@ func (s *server) handleQuery(state *peerState, querymsg interface{}) {
 			peers = append(peers, sp)
 		}
 		msg.reply <- peers
+
 	case disconnectNodeMsg:
 		// Check inbound peers. We pass a nil callback since we don't
 		// require any additional actions on disconnect for inbound peers.


Test patch:
diff --git a/connmgr/connmanager_test.go b/connmgr/connmanager_test.go
index beb2d8c5..db705c32 100644
--- a/connmgr/connmanager_test.go
+++ b/connmgr/connmanager_test.go
@@ -8,6 +8,7 @@ package connmgr
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"sync"
@@ -99,7 +100,7 @@ func mockDialerAddr(ctx context.Context, addr net.Addr) (net.Conn, error) {
 func TestNewConfig(t *testing.T) {
 	_, err := New(&Config{})
 	if err == nil {
-		t.Fatalf("New expected error: 'Dial can't be nil', got nil")
+		t.Fatal("New expected error: 'Dial can't be nil', got nil")
 	}
 	_, err = New(&Config{
 		Dial: mockDialer,
@@ -113,7 +114,7 @@ func TestNewConfig(t *testing.T) {
 		DialAddr: mockDialerAddr,
 	})
 	if err == nil {
-		t.Fatalf("New expected error: 'Dial and DialAddr can't be both nil', got nil")
+		t.Fatal("New expected error: 'Dial and DialAddr can't be both nil', got nil")
 	}
 
 	_, err = New(&Config{
@@ -269,13 +270,13 @@ func TestPassAddrAlongDialAddr(t *testing.T) {
 	case c := <-connected:
 		receivedMock, isMockAddr := c.Addr.(mockAddr)
 		if !isMockAddr {
-			t.Fatalf("connected to an address that was not a mockAddr")
+			t.Fatal("connected to an address that was not a mockAddr")
 		}
 		if receivedMock != targetAddr {
-			t.Fatalf("connected to an address different than the expected target")
+			t.Fatal("connected to an address different than the expected target")
 		}
 	case <-time.After(time.Millisecond * 5):
-		t.Fatalf("did not get connection to target address before timeout")
+		t.Fatal("did not get connection to target address before timeout")
 	}
 
 	// Ensure clean shutdown of connection manager.
@@ -390,7 +391,7 @@ func TestMaxRetryDuration(t *testing.T) {
 	select {
 	case <-connected:
 	case <-time.After(20 * time.Millisecond):
-		t.Fatalf("max retry duration: connection timeout")
+		t.Fatal("max retry duration: connection timeout")
 	}
 
 	// Ensure clean shutdown of connection manager.
@@ -454,6 +455,77 @@ func TestNetworkFailure(t *testing.T) {
 	}
 }
 
+// TestMultipleFailedConns ensures that the connection manager remains
+// responsive when there are multiple simultaneous failed connections for
+// persistent peers in the retry state.
+func TestMultipleFailedConns(t *testing.T) {
+	// Override the max retry duration for this test since it relies on having
+	// multiple connections in the retry state.
+	curMaxRetryDuration := maxRetryDuration
+	maxRetryDuration = 500 * time.Millisecond
+	defer func() {
+		maxRetryDuration = curMaxRetryDuration
+	}()
+
+	const targetFailed = 5
+	var dials uint32
+	var closeOnce sync.Once
+	hitTargetFailed := make(chan struct{})
+	errDialer := func(ctx context.Context, network, addr string) (net.Conn, error) {
+		totalDials := atomic.AddUint32(&dials, 1)
+		if totalDials >= targetFailed {
+			closeOnce.Do(func() { close(hitTargetFailed) })
+		}
+		return nil, errors.New("network down")
+	}
+	cmgr, err := New(&Config{
+		RetryDuration: maxRetryDuration,
+		Dial:          errDialer,
+	})
+	if err != nil {
+		t.Fatalf("New error: %v", err)
+	}
+	ctx, shutdown, wg := runConnMgrAsync(context.Background(), cmgr)
+
+	// Establish several connection requests to localhost IPs.
+	for i := 0; i < targetFailed; i++ {
+		cr := &ConnReq{
+			Addr: &net.TCPAddr{
+				IP:   net.ParseIP(fmt.Sprintf("127.0.0.%d", i+1)),
+				Port: 18555,
+			},
+			Permanent: true,
+		}
+		go cmgr.Connect(ctx, cr)
+	}
+
+	// Wait for the target number of dials and ensure they happen simultaneously
+	// by checking it happens before the retry timeout.
+	select {
+	case <-hitTargetFailed:
+	case <-time.After(20 * time.Millisecond):
+		t.Fatal("did not reach target number of dials before timeout")
+	}
+
+	// Ensure that the connection manager still responds to requests while the
+	// failed connections are still retrying.
+	disconnected := make(chan struct{})
+	go func() {
+		const badID = ^uint64(0)
+		cmgr.Disconnect(badID)
+		close(disconnected)
+	}()
+	select {
+	case <-disconnected:
+	case <-time.After(20 * time.Millisecond):
+		t.Fatal("timeout servicing connmgr requests")
+	}
+
+	// Ensure clean shutdown of connection manager.
+	shutdown()
+	wg.Wait()
+}
+
 // TestShutdownFailedConns tests that failed connections are ignored after
 // connmgr is shutdown.
 //
@@ -619,7 +691,7 @@ func TestCancelIgnoreDelayedConnection(t *testing.T) {
 	// allowed to properly elapse.
 	select {
 	case <-connected:
-		t.Fatalf("on-connect should not be called for canceled req")
+		t.Fatal("on-connect should not be called for canceled req")
 	case <-time.After(5 * retryTimeout):
 	}