address pr comments

Author: ndyakov

internal/pool/conn.go

@@ -27,28 +27,6 @@ var (
 	errConnNotAvailableForWrite = errors.New("redis: connection not available for write operation")
 )
 
-// Global time cache updated every 50ms by background goroutine.
-// This avoids expensive time.Now() syscalls in hot paths like getEffectiveReadTimeout.
-// Max staleness: 50ms, which is acceptable for timeout deadline checks (timeouts are typically 3-30 seconds).
-var globalTimeCache struct {
-	nowNs atomic.Int64
-}
-
-func init() {
-	// Initialize immediately
-	globalTimeCache.nowNs.Store(time.Now().UnixNano())
-
-	// Start background updater
-	go func() {
-		ticker := time.NewTicker(50 * time.Millisecond)
-		defer ticker.Stop()
-
-		for range ticker.C {
-			globalTimeCache.nowNs.Store(time.Now().UnixNano())
-		}
-	}()
-}
-
 // getCachedTimeNs returns the current time in nanoseconds from the global cache.
 // This is updated every 50ms by a background goroutine, avoiding expensive syscalls.
 // Max staleness: 50ms.

internal/pool/global_time_cache.go

@@ -0,0 +1,74 @@
+package pool
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// Global time cache updated every 50ms by background goroutine.
+// This avoids expensive time.Now() syscalls in hot paths like getEffectiveReadTimeout.
+// Max staleness: 50ms, which is acceptable for timeout deadline checks (timeouts are typically 3-30 seconds).
+var globalTimeCache struct {
+	nowNs       atomic.Int64
+	lock        sync.Mutex
+	started     bool
+	stop        chan struct{}
+	subscribers int32
+}
+
+func subscribeToGlobalTimeCache() {
+	globalTimeCache.lock.Lock()
+	globalTimeCache.subscribers += 1
+	globalTimeCache.lock.Unlock()
+}
+
+func unsubscribeFromGlobalTimeCache() {
+	globalTimeCache.lock.Lock()
+	globalTimeCache.subscribers -= 1
+	globalTimeCache.lock.Unlock()
+}
+
+func startGlobalTimeCache() {
+	globalTimeCache.lock.Lock()
+	if globalTimeCache.started {
+		globalTimeCache.lock.Unlock()
+		return
+	}
+
+	globalTimeCache.started = true
+	globalTimeCache.nowNs.Store(time.Now().UnixNano())
+	globalTimeCache.stop = make(chan struct{})
+	globalTimeCache.lock.Unlock()
+	// Start background updater
+	go func() {
+		ticker := time.NewTicker(50 * time.Millisecond)
+		defer ticker.Stop()
+
+		for range ticker.C {
+			select {
+			case <-globalTimeCache.stop:
+				return
+			default:
+			}
+			globalTimeCache.nowNs.Store(time.Now().UnixNano())
+		}
+	}()
+}
+
+// stopGlobalTimeCache stops the global time cache if there are no subscribers.
+// This should only be called when the last subscriber is removed.
+func stopGlobalTimeCache() {
+	globalTimeCache.lock.Lock()
+	if !globalTimeCache.started || globalTimeCache.subscribers > 0 {
+		globalTimeCache.lock.Unlock()
+		return
+	}
+	globalTimeCache.started = false
+	close(globalTimeCache.stop)
+	globalTimeCache.lock.Unlock()
+}
+
+func init() {
+	startGlobalTimeCache()
+}

internal/pool/pool.go

@@ -177,6 +177,9 @@ func NewConnPool(opt *Options) *ConnPool {
 		p.connsMu.Unlock()
 	}
 
+	startGlobalTimeCache()
+	subscribeToGlobalTimeCache()
+
 	return p
 }
 
@@ -959,6 +962,9 @@ func (p *ConnPool) Close() error {
 		return ErrClosed
 	}
 
+	unsubscribeFromGlobalTimeCache()
+	stopGlobalTimeCache()
+
 	var firstErr error
 	p.connsMu.Lock()
 	for _, cn := range p.conns {

redis.go

@@ -1353,7 +1353,7 @@ func (c *Conn) TxPipeline() Pipeliner {
 
 // processPushNotifications processes all pending push notifications on a connection
 // This ensures that cluster topology changes are handled immediately before the connection is used
-// This method should be called by the client before using WithWriter for command execution
+// This method should be called by the client before using WithReader for command execution
 //
 // Performance optimization: Skip the expensive MaybeHasData() syscall if a health check
 // was performed recently (within 5 seconds). The health check already verified the connection