func NewClusterClient(opt *ClusterOptions) *ClusterClient {
	opt.init()
	c := &ClusterClient{
		opt:   opt,
		nodes: newClusterNodes(opt),
	}
	c.state = newClusterStateHolder(c.loadState)
	c.cmdsInfoCache = newCmdsInfoCache(c.cmdsInfo)
	c.process = c.defaultProcess
	c.processPipeline = c.defaultProcessPipeline
	c.processTxPipeline = c.defaultProcessTxPipeline
	c.init()
	_, _ = c.state.Reload()
	_, _ = c.cmdsInfoCache.Get()
	if opt.IdleCheckFrequency > 0 {
		go c.reaper(opt.IdleCheckFrequency)
	}
	return c
}
NewClusterClient returns a Redis Cluster client as described in
http://redis.io/topics/cluster-spec.
package redis
import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"math"
	"math/rand"
	"net"
	"sort"
	"sync"
	"sync/atomic"
	"time"
	"github.com/go-redis/redis/internal"
	"github.com/go-redis/redis/internal/hashtag"
	"github.com/go-redis/redis/internal/pool"
	"github.com/go-redis/redis/internal/proto"
	"github.com/go-redis/redis/internal/singleflight"
)
var errClusterNoNodes = fmt.Errorf("redis: cluster has no nodes")
type ClusterOptions struct {
	Addrs []string
	MaxRedirects int
	ReadOnly bool
	RouteByLatency bool
	RouteRandomly bool
	ClusterSlots func() ([]ClusterSlot, error)
	OnConnect func(*Conn) error
	MaxRetries      int
	MinRetryBackoff time.Duration
	MaxRetryBackoff time.Duration
	Password        string
	DialTimeout  time.Duration
	ReadTimeout  time.Duration
	WriteTimeout time.Duration
	PoolSize           int
	PoolTimeout        time.Duration
	IdleTimeout        time.Duration
	IdleCheckFrequency time.Duration
	TLSConfig *tls.Config
}
func (opt *ClusterOptions) init() {
	if opt.MaxRedirects == -1 {
		opt.MaxRedirects = 0
	} else if opt.MaxRedirects == 0 {
		opt.MaxRedirects = 8
	}
	if opt.RouteByLatency || opt.RouteRandomly {
		opt.ReadOnly = true
	}
	switch opt.ReadTimeout {
	case -1:
		opt.ReadTimeout = 0
	case 0:
		opt.ReadTimeout = 3 * time.Second
	}
	switch opt.WriteTimeout {
	case -1:
		opt.WriteTimeout = 0
	case 0:
		opt.WriteTimeout = opt.ReadTimeout
	}
	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 *ClusterOptions) clientOptions() *Options {
	const disableIdleCheck = -1
	return &Options{
		OnConnect: opt.OnConnect,
		MaxRetries:      opt.MaxRetries,
		MinRetryBackoff: opt.MinRetryBackoff,
		MaxRetryBackoff: opt.MaxRetryBackoff,
		Password:        opt.Password,
		readOnly:        opt.ReadOnly,
		DialTimeout:  opt.DialTimeout,
		ReadTimeout:  opt.ReadTimeout,
		WriteTimeout: opt.WriteTimeout,
		PoolSize:    opt.PoolSize,
		PoolTimeout: opt.PoolTimeout,
		IdleTimeout: opt.IdleTimeout,
		IdleCheckFrequency: disableIdleCheck,
		TLSConfig: opt.TLSConfig,
	}
}
type clusterNode struct {
	Client *Client
	latency    uint32  
	generation uint32  
	loading    uint32  
}
func newClusterNode(clOpt *ClusterOptions, addr string) *clusterNode {
	opt := clOpt.clientOptions()
	opt.Addr = addr
	node := clusterNode{
		Client: NewClient(opt),
	}
	node.latency = math.MaxUint32
	if clOpt.RouteByLatency {
		go node.updateLatency()
	}
	return &node
}
func (n *clusterNode) String() string {
	return n.Client.String()
}
func (n *clusterNode) Close() error {
	return n.Client.Close()
}
func (n *clusterNode) updateLatency() {
	const probes = 10
	var latency uint32
	for i := 0; i < probes; i++ {
		start := time.Now()
		n.Client.Ping()
		probe := uint32(time.Since(start) / time.Microsecond)
		latency = (latency + probe) / 2
	}
	atomic.StoreUint32(&n.latency, latency)
}
func (n *clusterNode) Latency() time.Duration {
	latency := atomic.LoadUint32(&n.latency)
	return time.Duration(latency) * time.Microsecond
}
func (n *clusterNode) MarkAsLoading() {
	atomic.StoreUint32(&n.loading, uint32(time.Now().Unix()))
}
func (n *clusterNode) Loading() bool {
	const minute = int64(time.Minute / time.Second)
	loading := atomic.LoadUint32(&n.loading)
	if loading == 0 {
		return false
	}
	if time.Now().Unix()-int64(loading) < minute {
		return true
	}
	atomic.StoreUint32(&n.loading, 0)
	return false
}
func (n *clusterNode) Generation() uint32 {
	return atomic.LoadUint32(&n.generation)
}
func (n *clusterNode) SetGeneration(gen uint32) {
	for {
		v := atomic.LoadUint32(&n.generation)
		if gen < v || atomic.CompareAndSwapUint32(&n.generation, v, gen) {
			break
		}
	}
}
type clusterNodes struct {
	opt *ClusterOptions
	mu           sync.RWMutex
	allAddrs     []string
	allNodes     map[string]*clusterNode
	clusterAddrs []string
	closed       bool
	nodeCreateGroup singleflight.Group
	_generation uint32  
}
func newClusterNodes(opt *ClusterOptions) *clusterNodes {
	return &clusterNodes{
		opt: opt,
		allAddrs: opt.Addrs,
		allNodes: make(map[string]*clusterNode),
	}
}
func (c *clusterNodes) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closed {
		return nil
	}
	c.closed = true
	var firstErr error
	for _, node := range c.allNodes {
		if err := node.Client.Close(); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	c.allNodes = nil
	c.clusterAddrs = nil
	return firstErr
}
func (c *clusterNodes) Addrs() ([]string, error) {
	var addrs []string
	c.mu.RLock()
	closed := c.closed
	if !closed {
		if len(c.clusterAddrs) > 0 {
			addrs = c.clusterAddrs
		} else {
			addrs = c.allAddrs
		}
	}
	c.mu.RUnlock()
	if closed {
		return nil, pool.ErrClosed
	}
	if len(addrs) == 0 {
		return nil, errClusterNoNodes
	}
	return addrs, nil
}
func (c *clusterNodes) NextGeneration() uint32 {
	return atomic.AddUint32(&c._generation, 1)
}
func (c *clusterNodes) GC(generation uint32) {
	var collected []*clusterNode
	c.mu.Lock()
	for addr, node := range c.allNodes {
		if node.Generation() >= generation {
			continue
		}
		c.clusterAddrs = remove(c.clusterAddrs, addr)
		delete(c.allNodes, addr)
		collected = append(collected, node)
	}
	c.mu.Unlock()
	for _, node := range collected {
		_ = node.Client.Close()
	}
}
func (c *clusterNodes) Get(addr string) (*clusterNode, error) {
	var node *clusterNode
	var err error
	c.mu.RLock()
	if c.closed {
		err = pool.ErrClosed
	} else {
		node = c.allNodes[addr]
	}
	c.mu.RUnlock()
	return node, err
}
func (c *clusterNodes) GetOrCreate(addr string) (*clusterNode, error) {
	node, err := c.Get(addr)
	if err != nil {
		return nil, err
	}
	if node != nil {
		return node, nil
	}
	v, err := c.nodeCreateGroup.Do(addr, func() (interface{}, error) {
		node := newClusterNode(c.opt, addr)
		return node, nil
	})
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closed {
		return nil, pool.ErrClosed
	}
	node, ok := c.allNodes[addr]
	if ok {
		_ = v.(*clusterNode).Close()
		return node, err
	}
	node = v.(*clusterNode)
	c.allAddrs = appendIfNotExists(c.allAddrs, addr)
	if err == nil {
		c.clusterAddrs = append(c.clusterAddrs, addr)
	}
	c.allNodes[addr] = node
	return node, err
}
func (c *clusterNodes) All() ([]*clusterNode, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()
	if c.closed {
		return nil, pool.ErrClosed
	}
	cp := make([]*clusterNode, 0, len(c.allNodes))
	for _, node := range c.allNodes {
		cp = append(cp, node)
	}
	return cp, nil
}
func (c *clusterNodes) Random() (*clusterNode, error) {
	addrs, err := c.Addrs()
	if err != nil {
		return nil, err
	}
	n := rand.Intn(len(addrs))
	return c.GetOrCreate(addrs[n])
}
type clusterSlot struct {
	start, end int
	nodes      []*clusterNode
}
type clusterSlotSlice []*clusterSlot
func (p clusterSlotSlice) Len() int {
	return len(p)
}
func (p clusterSlotSlice) Less(i, j int) bool {
	return p[i].start < p[j].start
}
func (p clusterSlotSlice) Swap(i, j int) {
	p[i], p[j] = p[j], p[i]
}
type clusterState struct {
	nodes   *clusterNodes
	Masters []*clusterNode
	Slaves  []*clusterNode
	slots []*clusterSlot
	generation uint32
	createdAt  time.Time
}
func newClusterState(
	nodes *clusterNodes, slots []ClusterSlot, origin string,
) (*clusterState, error) {
	c := clusterState{
		nodes: nodes,
		slots: make([]*clusterSlot, 0, len(slots)),
		generation: nodes.NextGeneration(),
		createdAt:  time.Now(),
	}
	isLoopbackOrigin := isLoopbackAddr(origin)
	for _, slot := range slots {
		var nodes []*clusterNode
		for i, slotNode := range slot.Nodes {
			addr := slotNode.Addr
			if !isLoopbackOrigin && useOriginAddr(origin, addr) {
				addr = origin
			}
			node, err := c.nodes.GetOrCreate(addr)
			if err != nil {
				return nil, err
			}
			node.SetGeneration(c.generation)
			nodes = append(nodes, node)
			if i == 0 {
				c.Masters = appendUniqueNode(c.Masters, node)
			} else {
				c.Slaves = appendUniqueNode(c.Slaves, node)
			}
		}
		c.slots = append(c.slots, &clusterSlot{
			start: slot.Start,
			end:   slot.End,
			nodes: nodes,
		})
	}
	sort.Sort(clusterSlotSlice(c.slots))
	time.AfterFunc(time.Minute, func() {
		nodes.GC(c.generation)
	})
	return &c, nil
}
func (c *clusterState) slotMasterNode(slot int) (*clusterNode, error) {
	nodes := c.slotNodes(slot)
	if len(nodes) > 0 {
		return nodes[0], nil
	}
	return c.nodes.Random()
}
func (c *clusterState) slotSlaveNode(slot int) (*clusterNode, error) {
	nodes := c.slotNodes(slot)
	switch len(nodes) {
	case 0:
		return c.nodes.Random()
	case 1:
		return nodes[0], nil
	case 2:
		if slave := nodes[1]; !slave.Loading() {
			return slave, nil
		}
		return nodes[0], nil
	default:
		var slave *clusterNode
		for i := 0; i < 10; i++ {
			n := rand.Intn(len(nodes)-1) + 1
			slave = nodes[n]
			if !slave.Loading() {
				break
			}
		}
		return slave, nil
	}
}
func (c *clusterState) slotClosestNode(slot int) (*clusterNode, error) {
	const threshold = time.Millisecond
	nodes := c.slotNodes(slot)
	if len(nodes) == 0 {
		return c.nodes.Random()
	}
	var node *clusterNode
	for _, n := range nodes {
		if n.Loading() {
			continue
		}
		if node == nil || node.Latency()-n.Latency() > threshold {
			node = n
		}
	}
	return node, nil
}
func (c *clusterState) slotRandomNode(slot int) *clusterNode {
	nodes := c.slotNodes(slot)
	n := rand.Intn(len(nodes))
	return nodes[n]
}
func (c *clusterState) slotNodes(slot int) []*clusterNode {
	i := sort.Search(len(c.slots), func(i int) bool {
		return c.slots[i].end >= slot
	})
	if i >= len(c.slots) {
		return nil
	}
	x := c.slots[i]
	if slot >= x.start && slot <= x.end {
		return x.nodes
	}
	return nil
}
func (c *clusterState) IsConsistent() bool {
	if c.nodes.opt.ClusterSlots != nil {
		return true
	}
	return len(c.Masters) <= len(c.Slaves)
}
type clusterStateHolder struct {
	load func() (*clusterState, error)
	state atomic.Value
	firstErrMu sync.RWMutex
	firstErr   error
	reloading uint32  
}
func newClusterStateHolder(fn func() (*clusterState, error)) *clusterStateHolder {
	return &clusterStateHolder{
		load: fn,
	}
}
func (c *clusterStateHolder) Reload() (*clusterState, error) {
	state, err := c.reload()
	if err != nil {
		return nil, err
	}
	if !state.IsConsistent() {
		time.AfterFunc(time.Second, c.LazyReload)
	}
	return state, nil
}
func (c *clusterStateHolder) reload() (*clusterState, error) {
	state, err := c.load()
	if err != nil {
		c.firstErrMu.Lock()
		if c.firstErr == nil {
			c.firstErr = err
		}
		c.firstErrMu.Unlock()
		return nil, err
	}
	c.state.Store(state)
	return state, nil
}
func (c *clusterStateHolder) LazyReload() {
	if !atomic.CompareAndSwapUint32(&c.reloading, 0, 1) {
		return
	}
	go func() {
		defer atomic.StoreUint32(&c.reloading, 0)
		for {
			state, err := c.reload()
			if err != nil {
				return
			}
			time.Sleep(100 * time.Millisecond)
			if state.IsConsistent() {
				return
			}
		}
	}()
}
func (c *clusterStateHolder) Get() (*clusterState, error) {
	v := c.state.Load()
	if v != nil {
		state := v.(*clusterState)
		if time.Since(state.createdAt) > time.Minute {
			c.LazyReload()
		}
		return state, nil
	}
	c.firstErrMu.RLock()
	err := c.firstErr
	c.firstErrMu.RUnlock()
	if err != nil {
		return nil, err
	}
	return nil, errors.New("redis: cluster has no state")
}
func (c *clusterStateHolder) ReloadOrGet() (*clusterState, error) {
	state, err := c.Reload()
	if err == nil {
		return state, nil
	}
	return c.Get()
}
type ClusterClient struct {
	cmdable
	ctx context.Context
	opt           *ClusterOptions
	nodes         *clusterNodes
	state         *clusterStateHolder
	cmdsInfoCache *cmdsInfoCache
	process           func(Cmder) error
	processPipeline   func([]Cmder) error
	processTxPipeline func([]Cmder) error
}