Use atomics for HEALTHZ command

The HEALTHZ command is used to detect if there a leader or follower
are ready for commands. All that it cares about is if the leader
has it's data fully loaded in memory, and that a follower is
caught up to the leader.

Previously there was a read lock around the 'caught up' flags,
which could potentially be blocked by other operations such as
SET and INTERSECTS. This commit removes the shared read lock when
running the HEALTHZ command. It now uses atomics bit flags for the
'caught up' flags.

The result is that the HEALTHZ command should have little-to-no
noticable contention, always appearing instant to the user.

Author: tidwall

internal/server/follow.go

@@ -203,13 +203,37 @@ func (s *Server) followDoLeaderAuth(conn *RESPConn, auth string) error {
 	return nil
 }
 
+// bit flags for the fcupflags server field
+const (
+	bitCaughtUpOnce int32 = 1 // follower caught up at least once in the past
+	bitCaughtUp     int32 = 2 // follower is fully caught up to leader
+)
+
+func (s *Server) setCaughtUp(caughtUp bool) {
+	var flags int32
+	if caughtUp {
+		flags = bitCaughtUp | bitCaughtUpOnce
+	} else {
+		flags = s.fcupflags.Load() & bitCaughtUpOnce
+	}
+	s.fcupflags.Store(flags)
+}
+
+func (s *Server) caughtUp() bool {
+	return (s.fcupflags.Load() & bitCaughtUp) == bitCaughtUp
+}
+
+func (s *Server) caughtUpOnce() bool {
+	return (s.fcupflags.Load() & bitCaughtUpOnce) == bitCaughtUpOnce
+}
+
 func (s *Server) followStep(host string, port int, followc int) error {
 	if int(s.followc.Load()) != followc {
 		return errNoLongerFollowing
 	}
 	s.mu.Lock()
 	s.faofsz = 0
-	s.fcup = false
+	s.setCaughtUp(false)
 	auth := s.config.leaderAuth()
 	s.mu.Unlock()
 	addr := fmt.Sprintf("%s:%d", host, port)
@@ -305,10 +329,7 @@ func (s *Server) followStep(host string, port int, followc int) error {
 
 	caughtUp := pos >= aofSize
 	if caughtUp {
-		s.mu.Lock()
-		s.fcup = true
-		s.fcuponce = true
-		s.mu.Unlock()
+		s.setCaughtUp(true)
 		log.Info("caught up")
 	}
 
@@ -339,8 +360,7 @@ func (s *Server) followStep(host string, port int, followc int) error {
 				caughtUp = true
 				s.mu.Lock()
 				s.flushAOF(false)
-				s.fcup = true
-				s.fcuponce = true
+				s.setCaughtUp(true)
 				s.mu.Unlock()
 				log.Info("caught up")
 			}

internal/server/metrics.go

@@ -111,7 +111,7 @@ func (s *Server) Collect(ch chan<- prometheus.Metric) {
 	if s.config.followHost() != "" {
 		replLbls = []string{"follower",
 			fmt.Sprintf("%s:%d", s.config.followHost(), s.config.followPort()),
-			fmt.Sprintf("%t", s.fcup), fmt.Sprintf("%t", s.fcuponce)}
+			fmt.Sprintf("%t", s.caughtUp()), fmt.Sprintf("%t", s.caughtUpOnce())}
 	}
 	ch <- prometheus.MustNewConstMetric(
 		metricDescriptions["replication"],

internal/server/scripts.go

@@ -743,7 +743,7 @@ func (s *Server) luaTile38AtomicRW(msg *Message) (resp.Value, error) {
 	case "get", "keys", "scan", "nearby", "within", "intersects", "hooks", "search",
 		"ttl", "bounds", "server", "info", "type", "jget", "fget", "exists", "fexists", "test":
 		// read operations
-		if s.config.followHost() != "" && !s.fcuponce {
+		if s.config.followHost() != "" && !s.caughtUpOnce() {
 			return resp.NullValue(), errCatchingUp
 		}
 	}
@@ -796,7 +796,7 @@ func (s *Server) luaTile38AtomicRO(msg *Message) (resp.Value, error) {
 	case "get", "keys", "scan", "nearby", "within", "intersects", "hooks", "search",
 		"ttl", "bounds", "server", "info", "type", "jget", "fget", "exists", "fexists", "test":
 		// read operations
-		if s.config.followHost() != "" && !s.fcuponce {
+		if s.config.followHost() != "" && !s.caughtUpOnce() {
 			return resp.NullValue(), errCatchingUp
 		}
 	}
@@ -849,7 +849,7 @@ func (s *Server) luaTile38NonAtomic(msg *Message) (resp.Value, error) {
 		// read operations
 		s.mu.RLock()
 		defer s.mu.RUnlock()
-		if s.config.followHost() != "" && !s.fcuponce {
+		if s.config.followHost() != "" && !s.caughtUpOnce() {
 			return resp.NullValue(), errCatchingUp
 		}
 	}

internal/server/server.go

@@ -131,16 +131,15 @@ type Server struct {
 	hookExpires  *btree.BTree // queue of all hooks marked for expiration
 
 	// followers (external aof readers)
-	follows  map[*bytes.Buffer]bool
-	fcond    *sync.Cond
-	lstack   []*commandDetails
-	lives    map[*liveBuffer]bool
-	lcond    *sync.Cond // live geofence signal
-	faofsz   int        // last reported aofsize
-	fcup     bool       // follow caught up
-	fcuponce bool       // follow caught up once
-	aofconnM map[net.Conn]io.Closer
-	pubq     pubQueue
+	follows   map[*bytes.Buffer]bool
+	fcond     *sync.Cond
+	lstack    []*commandDetails
+	lives     map[*liveBuffer]bool
+	lcond     *sync.Cond   // live geofence signal
+	faofsz    int          // last reported aofsize
+	fcupflags atomic.Int32 // follow caught up (caughtUp and caughtUpOnce)
+	aofconnM  map[net.Conn]io.Closer
+	pubq      pubQueue
 
 	// lua scripts
 	luascripts *lScriptMap
@@ -1029,12 +1028,16 @@ func (s *Server) handleInputCommand(client *Client, msg *Message) error {
 		}
 	case "get", "keys", "scan", "nearby", "within", "intersects", "hooks",
 		"chans", "search", "ttl", "bounds", "server", "info", "type", "jget",
-		"evalro", "evalrosha", "healthz", "role", "fget", "exists", "fexists":
+		"evalro", "evalrosha", "role", "fget", "exists", "fexists":
 		// read operations
-
 		s.mu.RLock()
 		defer s.mu.RUnlock()
-		if s.config.followHost() != "" && !s.fcuponce {
+		// fallthrough to perform a "catching up to leader" check
+		fallthrough
+	case "healthz":
+		// healthz operation does not require a read lock. It only needs
+		// a caughtup once check to leader, which is atomic.
+		if s.config.followHost() != "" && !s.caughtUpOnce() {
 			return writeErr("catching up to leader")
 		}
 	case "follow", "slaveof", "replconf", "readonly", "config":

internal/server/stats.go

@@ -109,9 +109,7 @@ func (s *Server) cmdHEALTHZ(msg *Message) (resp.Value, error) {
 	// >> Operation
 
 	if s.config.followHost() != "" {
-		m := make(map[string]interface{})
-		s.basicStats(m)
-		if fmt.Sprintf("%v", m["caught_up"]) != "true" {
+		if !s.caughtUp() {
 			return retrerr(errors.New("not caught up"))
 		}
 	}
@@ -168,8 +166,9 @@ func (s *Server) basicStats(m map[string]interface{}) {
 	if s.config.followHost() != "" {
 		m["following"] = fmt.Sprintf("%s:%d", s.config.followHost(),
 			s.config.followPort())
-		m["caught_up"] = s.fcup
-		m["caught_up_once"] = s.fcuponce
+
+		m["caught_up"] = s.caughtUp()
+		m["caught_up_once"] = s.caughtUpOnce()
 	}
 	m["http_transport"] = s.http
 	m["pid"] = os.Getpid()