Extract commands info cache

cluster.go

@@ -485,20 +485,18 @@ type ClusterClient struct {
 
 	opt           *ClusterOptions
 	nodes         *clusterNodes
+	cmdsInfoCache *cmdsInfoCache
 
 	_state     atomic.Value
 	stateErrMu sync.RWMutex
 	stateErr   error
 
-	cmdsInfoOnce internal.Once
-	cmdsInfo     map[string]*CommandInfo
-
 	process           func(Cmder) error
 	processPipeline   func([]Cmder) error
 	processTxPipeline func([]Cmder) error
 
 	// Reports whether slots reloading is in progress.
-	reloading uint32
+	reloading uint32 // atomic
 }
 
 // NewClusterClient returns a Redis Cluster client as described in
@@ -509,6 +507,7 @@ func NewClusterClient(opt *ClusterOptions) *ClusterClient {
 	c := &ClusterClient{
 		opt:           opt,
 		nodes:         newClusterNodes(opt),
+		cmdsInfoCache: newCmdsInfoCache(),
 	}
 
 	c.process = c.defaultProcess
@@ -535,24 +534,17 @@ func (c *ClusterClient) retryBackoff(attempt int) time.Duration {
 }
 
 func (c *ClusterClient) cmdInfo(name string) *CommandInfo {
-	err := c.cmdsInfoOnce.Do(func() error {
+	cmdsInfo, err := c.cmdsInfoCache.Do(func() (map[string]*CommandInfo, error) {
 		node, err := c.nodes.Random()
 		if err != nil {
-			return err
+			return nil, err
 		}
-
+		return node.Client.Command().Result()
-		cmdsInfo, err := node.Client.Command().Result()
-		if err != nil {
-			return err
-		}
-
-		c.cmdsInfo = cmdsInfo
-		return nil
 	})
 	if err != nil {
 		return nil
 	}
-	info := c.cmdsInfo[name]
+	info := cmdsInfo[name]
 	if info == nil {
 		internal.Logf("info for cmd=%s not found", name)
 	}

command.go

@@ -1023,3 +1023,26 @@ func (cmd *CommandsInfoCmd) readReply(cn *pool.Conn) error {
 	cmd.val = v.(map[string]*CommandInfo)
 	return nil
 }
+
+//------------------------------------------------------------------------------
+
+type cmdsInfoCache struct {
+	once internal.Once
+	cmds map[string]*CommandInfo
+}
+
+func newCmdsInfoCache() *cmdsInfoCache {
+	return &cmdsInfoCache{}
+}
+
+func (c *cmdsInfoCache) Do(fn func() (map[string]*CommandInfo, error)) (map[string]*CommandInfo, error) {
+	err := c.once.Do(func() error {
+		cmds, err := fn()
+		if err != nil {
+			return err
+		}
+		c.cmds = cmds
+		return nil
+	})
+	return c.cmds, err
+}

redis.go

@@ -11,7 +11,7 @@ import (
 	"github.com/go-redis/redis/internal/proto"
 )
 
-// Nil reply redis returned when key does not exist.
+// Nil reply Redis returns when key does not exist.
 const Nil = proto.Nil
 
 func init() {
@@ -119,10 +119,7 @@ func (c *baseClient) initConn(cn *pool.Conn) error {
 	return nil
 }
 
-// WrapProcess replaces the process func. It takes a function createWrapper
+// WrapProcess wraps function that processes Redis commands.
-// which is supplied by the user. createWrapper takes the old process func as
-// an input and returns the new wrapper process func. createWrapper should
-// use call the old process func within the new process func.
 func (c *baseClient) WrapProcess(fn func(oldProcess func(cmd Cmder) error) func(cmd Cmder) error) {
 	c.process = fn(c.process)
 }

ring.go

@@ -15,6 +15,8 @@ import (
 	"github.com/go-redis/redis/internal/pool"
 )
 
+const nreplicas = 100
+
 var errRingShardsDown = errors.New("redis: all ring shards are down")
 
 // RingOptions are used to configure a ring client and should be
@@ -143,29 +145,24 @@ type Ring struct {
 	cmdable
 
 	opt           *RingOptions
-	nreplicas int
+	cmdsInfoCache *cmdsInfoCache
 
 	mu         sync.RWMutex
 	hash       *consistenthash.Map
-	shards     map[string]*ringShard
+	shards     map[string]*ringShard // read only
-	shardsList []*ringShard
+	shardsList []*ringShard          // read only
 
 	processPipeline func([]Cmder) error
 
-	cmdsInfoOnce internal.Once
-	cmdsInfo     map[string]*CommandInfo
-
 	closed bool
 }
 
 func NewRing(opt *RingOptions) *Ring {
-	const nreplicas = 100
-
 	opt.init()
 
 	ring := &Ring{
 		opt:           opt,
-		nreplicas: nreplicas,
+		cmdsInfoCache: newCmdsInfoCache(),
 
 		hash:   consistenthash.New(nreplicas, nil),
 		shards: make(map[string]*ringShard),
@@ -186,11 +183,9 @@ func NewRing(opt *RingOptions) *Ring {
 
 func (c *Ring) addShard(name string, cl *Client) {
 	shard := &ringShard{Client: cl}
-	c.mu.Lock()
 	c.hash.Add(name)
 	c.shards[name] = shard
 	c.shardsList = append(c.shardsList, shard)
-	c.mu.Unlock()
 }
 
 // Options returns read-only Options that were used to create the client.
@@ -285,31 +280,27 @@ func (c *Ring) ForEachShard(fn func(client *Client) error) error {
 }
 
 func (c *Ring) cmdInfo(name string) *CommandInfo {
-	err := c.cmdsInfoOnce.Do(func() error {
+	cmdsInfo, err := c.cmdsInfoCache.Do(func() (map[string]*CommandInfo, error) {
 		c.mu.RLock()
 		shards := c.shardsList
 		c.mu.RUnlock()
 
-		var firstErr error
+		firstErr := errRingShardsDown
 		for _, shard := range shards {
 			cmdsInfo, err := shard.Client.Command().Result()
 			if err == nil {
-				c.cmdsInfo = cmdsInfo
+				return cmdsInfo, nil
-				return nil
 			}
 			if firstErr == nil {
 				firstErr = err
 			}
 		}
-		return firstErr
+		return nil, firstErr
 	})
 	if err != nil {
 		return nil
 	}
-	if c.cmdsInfo == nil {
+	info := cmdsInfo[name]
-		return nil
-	}
-	info := c.cmdsInfo[name]
 	if info == nil {
 		internal.Logf("info for cmd=%s not found", name)
 	}
@@ -380,7 +371,7 @@ func (c *Ring) Process(cmd Cmder) error {
 
 // rebalance removes dead shards from the Ring.
 func (c *Ring) rebalance() {
-	hash := consistenthash.New(c.nreplicas, nil)
+	hash := consistenthash.New(nreplicas, nil)
 	for name, shard := range c.shards {
 		if shard.IsUp() {
 			hash.Add(name)