From b0ea067acb8854f5d3feba063bb6daa79052fcd9 Mon Sep 17 00:00:00 2001 From: Dave Collins Date: Thu, 9 Jul 2020 13:51:30 -0500 Subject: [PATCH] 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. --- connmgr/connmanager.go | 26 +++++++------- connmgr/connmanager_test.go | 72 +++++++++++++++++++++++++++++++++++++ server.go | 6 +++- 3 files changed, 91 insertions(+), 13 deletions(-) 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/connmgr/connmanager_test.go b/connmgr/connmanager_test.go index 0045b5ec..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" @@ -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. // 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.