func (c *ClusterClient) ForEachMaster(fn func(client *Client) error) error {
	state := c.state()
	if state == nil {
		return errNilClusterState
	}
	var wg sync.WaitGroup
	visited := make(map[*clusterNode]struct{})
	errCh := make(chan error, 1)
	for _, nodes := range state.slots {
		if len(nodes) == 0 {
			continue
		}
		master := nodes[0]
		if _, ok := visited[master]; ok {
			continue
		}
		visited[master] = struct{}{}
		wg.Add(1)
		go func(node *clusterNode) {
			defer wg.Done()
			err := fn(node.Client)
			if err != nil {
				select {
				case errCh <- err:
				default:
				}
			}
		}(master)
	}
	wg.Wait()
	select {
	case err := <-errCh:
		return err
	default:
		return nil
	}
}
ForEachMaster concurrently calls the fn on each master node in the cluster.
It returns the first error if any.
package redis
import (
	"fmt"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
	"gopkg.in/redis.v5/internal"
	"gopkg.in/redis.v5/internal/hashtag"
	"gopkg.in/redis.v5/internal/pool"
	"gopkg.in/redis.v5/internal/proto"
)
var errClusterNoNodes = internal.RedisError("redis: cluster has no nodes")
var errNilClusterState = internal.RedisError("redis: cannot load cluster slots")
type ClusterOptions struct {
	Addrs []string
	MaxRedirects int
	ReadOnly bool
	RouteByLatency bool
	Password string
	DialTimeout  time.Duration
	ReadTimeout  time.Duration
	WriteTimeout time.Duration
	PoolSize           int
	PoolTimeout        time.Duration
	IdleTimeout        time.Duration
	IdleCheckFrequency time.Duration
}
func (opt *ClusterOptions) init() {
	if opt.MaxRedirects == -1 {
		opt.MaxRedirects = 0
	} else if opt.MaxRedirects == 0 {
		opt.MaxRedirects = 16
	}
	if opt.RouteByLatency {
		opt.ReadOnly = true
	}
}
func (opt *ClusterOptions) clientOptions() *Options {
	const disableIdleCheck = -1
	return &Options{
		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,
	}
}
type clusterNode struct {
	Client  *Client
	Latency time.Duration
	loading time.Time
}
func newClusterNode(clOpt *ClusterOptions, addr string) *clusterNode {
	opt := clOpt.clientOptions()
	opt.Addr = addr
	node := clusterNode{
		Client: NewClient(opt),
	}
	if clOpt.RouteByLatency {
		node.updateLatency()
	}
	return &node
}
func (n *clusterNode) updateLatency() {
	const probes = 10
	for i := 0; i < probes; i++ {
		start := time.Now()
		n.Client.Ping()
		n.Latency += time.Since(start)
	}
	n.Latency = n.Latency / probes
}
func (n *clusterNode) Loading() bool {
	return !n.loading.IsZero() && time.Since(n.loading) < time.Minute
}
type clusterNodes struct {
	opt *ClusterOptions
	mu     sync.RWMutex
	addrs  []string
	nodes  map[string]*clusterNode
	closed bool
}
func newClusterNodes(opt *ClusterOptions) *clusterNodes {
	return &clusterNodes{
		opt:   opt,
		nodes: 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.nodes {
		if err := node.Client.Close(); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	c.addrs = nil
	c.nodes = nil
	return firstErr
}
func (c *clusterNodes) All() ([]*clusterNode, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()
	if c.closed {
		return nil, pool.ErrClosed
	}
	nodes := make([]*clusterNode, 0, len(c.nodes))
	for _, node := range c.nodes {
		nodes = append(nodes, node)
	}
	return nodes, nil
}
func (c *clusterNodes) Get(addr string) (*clusterNode, error) {
	var node *clusterNode
	var ok bool
	c.mu.RLock()
	if !c.closed {
		node, ok = c.nodes[addr]
	}
	c.mu.RUnlock()
	if ok {
		return node, nil
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closed {
		return nil, pool.ErrClosed
	}
	node, ok = c.nodes[addr]
	if ok {
		return node, nil
	}
	c.addrs = append(c.addrs, addr)
	node = newClusterNode(c.opt, addr)
	c.nodes[addr] = node
	return node, nil
}
func (c *clusterNodes) Random() (*clusterNode, error) {
	c.mu.RLock()
	closed := c.closed
	addrs := c.addrs
	c.mu.RUnlock()
	if closed {
		return nil, pool.ErrClosed
	}
	if len(addrs) == 0 {
		return nil, errClusterNoNodes
	}
	var nodeErr error
	for i := 0; i <= c.opt.MaxRedirects; i++ {
		n := rand.Intn(len(addrs))
		node, err := c.Get(addrs[n])
		if err != nil {
			return nil, err
		}
		nodeErr = node.Client.ClusterInfo().Err()
		if nodeErr == nil {
			return node, nil
		}
	}
	return nil, nodeErr
}
type clusterState struct {
	nodes *clusterNodes
	slots [][]*clusterNode
}
func newClusterState(nodes *clusterNodes, slots []ClusterSlot) (*clusterState, error) {
	c := clusterState{
		nodes: nodes,
		slots: make([][]*clusterNode, hashtag.SlotNumber),
	}
	for _, slot := range slots {
		var nodes []*clusterNode
		for _, slotNode := range slot.Nodes {
			node, err := c.nodes.Get(slotNode.Addr)
			if err != nil {
				return nil, err
			}
			nodes = append(nodes, node)
		}
		for i := slot.Start; i <= slot.End; i++ {
			c.slots[i] = nodes
		}
	}
	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) slotNodes(slot int) []*clusterNode {
	if slot < len(c.slots) {
		return c.slots[slot]
	}
	return nil
}
type ClusterClient struct {
	cmdable
	opt    *ClusterOptions
	cmds   map[string]*CommandInfo
	nodes  *clusterNodes
	_state atomic.Value
	reloading uint32
	closed bool
}
func NewClusterClient(opt *ClusterOptions) *ClusterClient {
	opt.init()
	c := &ClusterClient{
		opt:   opt,
		nodes: newClusterNodes(opt),
	}
	c.cmdable.process = c.Process
	for _, addr := range opt.Addrs {
		_, _ = c.nodes.Get(addr)
	}
	for i := 0; i < 10; i++ {
		state, err := c.reloadSlots()
		if err == nil {
			c._state.Store(state)
			break
		}
	}
	if opt.IdleCheckFrequency > 0 {
		go c.reaper(opt.IdleCheckFrequency)
	}
	return c
}
func (c *ClusterClient) state() *clusterState {
	v := c._state.Load()
	if v != nil {
		return v.(*clusterState)
	}
	c.lazyReloadSlots()
	return nil
}
func (c *ClusterClient) cmdSlotAndNode(state *clusterState, cmd Cmder) (int, *clusterNode, error) {
	if state == nil {
		node, err := c.nodes.Random()
		return 0, node, err
	}
	cmdInfo := c.cmds[cmd.name()]
	firstKey := cmd.arg(cmdFirstKeyPos(cmd, cmdInfo))
	slot := hashtag.Slot(firstKey)
	if cmdInfo != nil && cmdInfo.ReadOnly && c.opt.ReadOnly {
		if c.opt.RouteByLatency {
			node, err := state.slotClosestNode(slot)
			return slot, node, err
		}
		node, err := state.slotSlaveNode(slot)
		return slot, node, err
	}
	node, err := state.slotMasterNode(slot)
	return slot, node, err
}
func (c *ClusterClient) Watch(fn func(*Tx) error, keys ...string) error {
	state := c.state()
	var node *clusterNode
	var err error
	if state != nil && len(keys) > 0 {
		node, err = state.slotMasterNode(hashtag.Slot(keys[0]))
	} else {
		node, err = c.nodes.Random()
	}
	if err != nil {
		return err
	}
	return node.Client.Watch(fn, keys...)
}
func (c *ClusterClient) Close() error {
	return c.nodes.Close()
}
func (c *ClusterClient) Process(cmd Cmder) error {
	slot, node, err := c.cmdSlotAndNode(c.state(), cmd)
	if err != nil {
		cmd.setErr(err)
		return err
	}
	var ask bool
	for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
		if ask {
			pipe := node.Client.Pipeline()
			pipe.Process(NewCmd("ASKING"))
			pipe.Process(cmd)
			_, err = pipe.Exec()
			pipe.Close()
			ask = false
		} else {
			err = node.Client.Process(cmd)
		}
		if err == nil {
			return nil
		}
		if c.opt.ReadOnly && internal.IsLoadingError(err) {
			node.loading = time.Now()
			continue
		}
		if internal.IsRetryableError(err) {
			node, err = c.nodes.Random()
			if err != nil {
				cmd.setErr(err)
				return err
			}
			continue
		}
		var moved bool
		var addr string
		moved, ask, addr = internal.IsMovedError(err)
		if moved || ask {
			state := c.state()
			if state != nil && slot >= 0 {
				master, _ := state.slotMasterNode(slot)
				if moved && (master == nil || master.Client.getAddr() != addr) {
					c.lazyReloadSlots()
				}
			}
			node, err = c.nodes.Get(addr)
			if err != nil {
				cmd.setErr(err)
				return err
			}
			continue
		}
		break
	}
	return cmd.Err()
}
func (c *ClusterClient) ForEachNode(fn func(client *Client) error) error {
	nodes, err := c.nodes.All()
	if err != nil {
		return err
	}
	var wg sync.WaitGroup
	errCh := make(chan error, 1)
	for _, node := range nodes {
		wg.Add(1)
		go func(node *clusterNode) {
			defer wg.Done()
			err := fn(node.Client)
			if err != nil {
				select {
				case errCh <- err:
				default:
				}
			}
		}(node)
	}
	wg.Wait()
	select {
	case err := <-errCh:
		return err
	default:
		return nil
	}
}