func (c *Ring) Do(args ...interface{}) *Cmd {
	cmd := NewCmd(args...)
	c.Process(cmd)
	return cmd
}
Do creates a Cmd from the args and processes the cmd.
package redis
import (
	"context"
	"errors"
	"fmt"
	"math/rand"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
	"github.com/go-redis/redis/internal"
	"github.com/go-redis/redis/internal/consistenthash"
	"github.com/go-redis/redis/internal/hashtag"
	"github.com/go-redis/redis/internal/pool"
)
type Hash consistenthash.Hash
var errRingShardsDown = errors.New("redis: all ring shards are down")
type RingOptions struct {
	Addrs map[string]string
	HeartbeatFrequency time.Duration
	Hash Hash
	HashReplicas int
	OnConnect func(*Conn) error
	DB       int
	Password string
	MaxRetries      int
	MinRetryBackoff time.Duration
	MaxRetryBackoff time.Duration
	DialTimeout  time.Duration
	ReadTimeout  time.Duration
	WriteTimeout time.Duration
	PoolSize           int
	MinIdleConns       int
	MaxConnAge         time.Duration
	PoolTimeout        time.Duration
	IdleTimeout        time.Duration
	IdleCheckFrequency time.Duration
}
func (opt *RingOptions) init() {
	if opt.HeartbeatFrequency == 0 {
		opt.HeartbeatFrequency = 500 * time.Millisecond
	}
	if opt.HashReplicas == 0 {
		opt.HashReplicas = 100
	}
	switch opt.MinRetryBackoff {
	case -1:
		opt.MinRetryBackoff = 0
	case 0:
		opt.MinRetryBackoff = 8 * time.Millisecond
	}
	switch opt.MaxRetryBackoff {
	case -1:
		opt.MaxRetryBackoff = 0
	case 0:
		opt.MaxRetryBackoff = 512 * time.Millisecond
	}
}
func (opt *RingOptions) clientOptions() *Options {
	return &Options{
		OnConnect: opt.OnConnect,
		DB:       opt.DB,
		Password: opt.Password,
		DialTimeout:  opt.DialTimeout,
		ReadTimeout:  opt.ReadTimeout,
		WriteTimeout: opt.WriteTimeout,
		PoolSize:           opt.PoolSize,
		MinIdleConns:       opt.MinIdleConns,
		MaxConnAge:         opt.MaxConnAge,
		PoolTimeout:        opt.PoolTimeout,
		IdleTimeout:        opt.IdleTimeout,
		IdleCheckFrequency: opt.IdleCheckFrequency,
	}
}
type ringShard struct {
	Client *Client
	down   int32
}
func (shard *ringShard) String() string {
	var state string
	if shard.IsUp() {
		state = "up"
	} else {
		state = "down"
	}
	return fmt.Sprintf("%s is %s", shard.Client, state)
}
func (shard *ringShard) IsDown() bool {
	const threshold = 3
	return atomic.LoadInt32(&shard.down) >= threshold
}
func (shard *ringShard) IsUp() bool {
	return !shard.IsDown()
}
func (shard *ringShard) Vote(up bool) bool {
	if up {
		changed := shard.IsDown()
		atomic.StoreInt32(&shard.down, 0)
		return changed
	}
	if shard.IsDown() {
		return false
	}
	atomic.AddInt32(&shard.down, 1)
	return shard.IsDown()
}
type ringShards struct {
	opt *RingOptions
	mu     sync.RWMutex
	hash   *consistenthash.Map
	shards map[string]*ringShard  
	list   []*ringShard           
	len    int
	closed bool
}
func newRingShards(opt *RingOptions) *ringShards {
	return &ringShards{
		opt: opt,
		hash:   newConsistentHash(opt),
		shards: make(map[string]*ringShard),
	}
}
func (c *ringShards) Add(name string, cl *Client) {
	shard := &ringShard{Client: cl}
	c.hash.Add(name)
	c.shards[name] = shard
	c.list = append(c.list, shard)
}
func (c *ringShards) List() []*ringShard {
	c.mu.RLock()
	list := c.list
	c.mu.RUnlock()
	return list
}
func (c *ringShards) Hash(key string) string {
	c.mu.RLock()
	hash := c.hash.Get(key)
	c.mu.RUnlock()
	return hash
}
func (c *ringShards) GetByKey(key string) (*ringShard, error) {
	key = hashtag.Key(key)
	c.mu.RLock()
	if c.closed {
		c.mu.RUnlock()
		return nil, pool.ErrClosed
	}
	hash := c.hash.Get(key)
	if hash == "" {
		c.mu.RUnlock()
		return nil, errRingShardsDown
	}
	shard := c.shards[hash]
	c.mu.RUnlock()
	return shard, nil
}
func (c *ringShards) GetByHash(name string) (*ringShard, error) {
	if name == "" {
		return c.Random()
	}
	c.mu.RLock()
	shard := c.shards[name]
	c.mu.RUnlock()
	return shard, nil
}
func (c *ringShards) Random() (*ringShard, error) {
	return c.GetByKey(strconv.Itoa(rand.Int()))
}
func (c *ringShards) Heartbeat(frequency time.Duration) {
	ticker := time.NewTicker(frequency)
	defer ticker.Stop()
	for range ticker.C {
		var rebalance bool
		c.mu.RLock()
		if c.closed {
			c.mu.RUnlock()
			break
		}
		shards := c.list
		c.mu.RUnlock()
		for _, shard := range shards {
			err := shard.Client.Ping().Err()
			if shard.Vote(err == nil || err == pool.ErrPoolTimeout) {
				internal.Logf("ring shard state changed: %s", shard)
				rebalance = true
			}
		}
		if rebalance {
			c.rebalance()
		}
	}
}
func (c *ringShards) rebalance() {
	c.mu.RLock()
	shards := c.shards
	c.mu.RUnlock()
	hash := newConsistentHash(c.opt)
	var shardsNum int
	for name, shard := range shards {
		if shard.IsUp() {
			hash.Add(name)
			shardsNum++
		}
	}
	c.mu.Lock()
	c.hash = hash
	c.len = shardsNum
	c.mu.Unlock()
}
func (c *ringShards) Len() int {
	c.mu.RLock()
	l := c.len
	c.mu.RUnlock()
	return l
}
func (c *ringShards) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closed {
		return nil
	}
	c.closed = true
	var firstErr error
	for _, shard := range c.shards {
		if err := shard.Client.Close(); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	c.hash = nil
	c.shards = nil
	c.list = nil
	return firstErr
}
type Ring struct {
	cmdable
	ctx context.Context
	opt           *RingOptions
	shards        *ringShards
	cmdsInfoCache *cmdsInfoCache
	process         func(Cmder) error
	processPipeline func([]Cmder) error
}
func NewRing(opt *RingOptions) *Ring {
	opt.init()
	ring := &Ring{
		opt:    opt,
		shards: newRingShards(opt),
	}
	ring.cmdsInfoCache = newCmdsInfoCache(ring.cmdsInfo)
	ring.process = ring.defaultProcess
	ring.processPipeline = ring.defaultProcessPipeline
	ring.init()
	for name, addr := range opt.Addrs {
		clopt := opt.clientOptions()
		clopt.Addr = addr
		ring.shards.Add(name, NewClient(clopt))
	}
	go ring.shards.Heartbeat(opt.HeartbeatFrequency)
	return ring
}
func (c *Ring) init() {
	c.cmdable.setProcessor(c.Process)
}
func (c *Ring) Context() context.Context {
	if c.ctx != nil {
		return c.ctx
	}
	return context.Background()
}
func (c *Ring) WithContext(ctx context.Context) *Ring {
	if ctx == nil {
		panic("nil context")
	}
	c2 := c.clone()
	c2.ctx = ctx
	return c2
}
func (c *Ring) clone() *Ring {
	cp := *c
	cp.init()
	return &cp
}
func (c *Ring) Options() *RingOptions {
	return c.opt
}
func (c *Ring) retryBackoff(attempt int) time.Duration {
	return internal.RetryBackoff(attempt, c.opt.MinRetryBackoff, c.opt.MaxRetryBackoff)
}
func (c *Ring) PoolStats() *PoolStats {
	shards := c.shards.List()
	var acc PoolStats
	for _, shard := range shards {
		s := shard.Client.connPool.Stats()
		acc.Hits += s.Hits
		acc.Misses += s.Misses
		acc.Timeouts += s.Timeouts
		acc.TotalConns += s.TotalConns
		acc.IdleConns += s.IdleConns
	}
	return &acc
}
func (c *Ring) Len() int {
	return c.shards.Len()
}
func (c *Ring) Subscribe(channels ...string) *PubSub {
	if len(channels) == 0 {
		panic("at least one channel is required")
	}
	shard, err := c.shards.GetByKey(channels[0])
	if err != nil {
		panic(err)
	}
	return shard.Client.Subscribe(channels...)
}
func (c *Ring) PSubscribe(channels ...string) *PubSub {
	if len(channels) == 0 {
		panic("at least one channel is required")
	}
	shard, err := c.shards.GetByKey(channels[0])
	if err != nil {
		panic(err)
	}
	return shard.Client.PSubscribe(channels...)
}
func (c *Ring) ForEachShard(fn func(client *Client) error) error {
	shards := c.shards.List()
	var wg sync.WaitGroup
	errCh := make(chan error, 1)
	for _, shard := range shards {
		if shard.IsDown() {
			continue
		}
		wg.Add(1)
		go func(shard *ringShard) {
			defer wg.Done()
			err := fn(shard.Client)
			if err != nil {
				select {
				case errCh <- err:
				default:
				}
			}
		}(shard)
	}
	wg.Wait()
	select {
	case err := <-errCh:
		return err
	default:
		return nil
	}
}
func (c *Ring) cmdsInfo() (map[string]*CommandInfo, error) {
	shards := c.shards.List()
	firstErr := errRingShardsDown
	for _, shard := range shards {
		cmdsInfo, err := shard.Client.Command().Result()
		if err == nil {
			return cmdsInfo, nil
		}
		if firstErr == nil {
			firstErr = err
		}
	}
	return nil, firstErr
}
func (c *Ring) cmdInfo(name string) *CommandInfo {
	cmdsInfo, err := c.cmdsInfoCache.Get()
	if err != nil {
		return nil
	}
	info := cmdsInfo[name]
	if info == nil {
		internal.Logf("info for cmd=%s not found", name)
	}
	return info
}
func (c *Ring) cmdShard(cmd Cmder) (*ringShard, error) {
	cmdInfo := c.cmdInfo(cmd.Name())
	pos := cmdFirstKeyPos(cmd, cmdInfo)
	if pos == 0 {
		return c.shards.Random()
	}
	firstKey := cmd.stringArg(pos)
	return c.shards.GetByKey(firstKey)
}