[IMPROVED] Reduce lock contention in saveClosedClient

Protect the closed connection ring buffer with a dedicated
mutex instead of the server lock. This allows saveClosedClient
to record closed connections concurrently with closeConnection
for the server lock.

Signed-off-by: Daniele Sciascia <daniele@nats.io>

Author: sciascid

locksordering.txt

@@ -65,3 +65,9 @@ while holding the Server lock, but the Server lock must not be acquired while
 holding eventIdsMu.
 
     Server -> eventIdsMu
+
+The "closedMu" lock protects the server's closed connection ring. It may be
+acquired while holding the Server lock, but the Server lock must not be acquired
+while holding closedMu.
+
+    Server -> closedMu

server/monitor.go

@@ -316,11 +316,11 @@ func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
 	case ConnOpen:
 		c.Total = len(clist)
 	case ConnClosed:
-		closedClients = s.closed.closedClients()
+		closedClients = s.closedClients()
 		c.Total = len(closedClients)
 	case ConnAll:
 		c.Total = len(clist)
-		closedClients = s.closed.closedClients()
+		closedClients = s.closedClients()
 		c.Total += len(closedClients)
 	}
 

server/monitor_test.go

@@ -1205,14 +1205,12 @@ func TestMonitorConnzSortedByStopTimeClosedConn(t *testing.T) {
 	checkClosedConns(t, s, 4, time.Second)
 
 	// Now adjust the Stop times for these with some random values.
-	s.mu.Lock()
 	now := time.Now().UTC()
-	ccs := s.closed.closedClients()
+	ccs := s.closedClients()
 	for _, cc := range ccs {
 		newStop := now.Add(time.Duration(rand.Int()%120) * -time.Minute)
 		cc.Stop = &newStop
 	}
-	s.mu.Unlock()
 
 	url = fmt.Sprintf("http://127.0.0.1:%d/", s.MonitorAddr().Port)
 	for mode := 0; mode < 2; mode++ {
@@ -1250,13 +1248,11 @@ func TestMonitorConnzSortedByReason(t *testing.T) {
 	checkClosedConns(t, s, 20, time.Second)
 
 	// Now adjust the Reasons for these with some random values.
-	s.mu.Lock()
-	ccs := s.closed.closedClients()
+	ccs := s.closedClients()
 	max := int(ServerShutdown)
 	for _, cc := range ccs {
 		cc.Reason = ClosedState(rand.Int() % max).String()
 	}
-	s.mu.Unlock()
 
 	url = fmt.Sprintf("http://127.0.0.1:%d/", s.MonitorAddr().Port)
 	for mode := 0; mode < 2; mode++ {
@@ -7008,9 +7004,9 @@ func TestConnzClosedSubsDetailNoSharedMutation(t *testing.T) {
 	cc.Cid = 1
 	cc.subs = []SubDetail{{Subject: "foo.bar"}}
 	cc.NumSubs = 1
-	s.mu.Lock()
+	s.closedMu.Lock()
 	s.closed.append(cc)
-	s.mu.Unlock()
+	s.closedMu.Unlock()
 
 	// Concurrently request closed connections with subscription detail.
 	var wg sync.WaitGroup

server/server.go

@@ -211,6 +211,7 @@ type Server struct {
 	users               map[string]*User
 	nkeys               map[string]*NkeyUser
 	totalClients        uint64
+	closedMu            sync.Mutex // Protects the closedRingBuffer
 	closed              *closedRingBuffer
 	done                chan bool
 	start               time.Time
@@ -3608,11 +3609,11 @@ func (s *Server) saveClosedClient(c *client, nc net.Conn, subs map[string]*subsc
 	c.mu.Unlock()
 
 	// Place in the ring buffer
-	s.mu.Lock()
+	s.closedMu.Lock()
 	if s.closed != nil {
 		s.closed.append(cc)
 	}
-	s.mu.Unlock()
+	s.closedMu.Unlock()
 }
 
 // Adds to the list of client and websocket clients connect URLs.
@@ -4102,20 +4103,20 @@ func (s *Server) startGoRoutine(f func(), tags ...pprofLabels) bool {
 }
 
 func (s *Server) numClosedConns() int {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
+	s.closedMu.Lock()
+	defer s.closedMu.Unlock()
 	return s.closed.len()
 }
 
 func (s *Server) totalClosedConns() uint64 {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
+	s.closedMu.Lock()
+	defer s.closedMu.Unlock()
 	return s.closed.totalConns()
 }
 
 func (s *Server) closedClients() []*closedClient {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
+	s.closedMu.Lock()
+	defer s.closedMu.Unlock()
 	return s.closed.closedClients()
 }