import (
"context"
"crypto/tls"
- "errors"
"fmt"
"math"
"math/rand"
"net"
+ "runtime"
"sort"
"sync"
"sync/atomic"
"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")
// and Cluster.ReloadState to manually trigger state reloading.
ClusterSlots func() ([]ClusterSlot, error)
+ // Optional hook that is called when a new node is created.
+ OnNewNode func(*Client)
+
// Following options are copied from Options struct.
OnConnect func(*Conn) error
+ Password string
+
MaxRetries int
MinRetryBackoff time.Duration
MaxRetryBackoff time.Duration
- Password string
DialTimeout time.Duration
ReadTimeout time.Duration
// PoolSize applies per cluster node and not for the whole cluster.
PoolSize int
+ MinIdleConns int
+ MaxConnAge time.Duration
PoolTimeout time.Duration
IdleTimeout time.Duration
IdleCheckFrequency time.Duration
opt.MaxRedirects = 8
}
- if opt.RouteByLatency || opt.RouteRandomly {
+ if (opt.RouteByLatency || opt.RouteRandomly) && opt.ClusterSlots == nil {
opt.ReadOnly = true
}
+ if opt.PoolSize == 0 {
+ opt.PoolSize = 5 * runtime.NumCPU()
+ }
+
switch opt.ReadTimeout {
case -1:
opt.ReadTimeout = 0
ReadTimeout: opt.ReadTimeout,
WriteTimeout: opt.WriteTimeout,
- PoolSize: opt.PoolSize,
- PoolTimeout: opt.PoolTimeout,
- IdleTimeout: opt.IdleTimeout,
-
+ PoolSize: opt.PoolSize,
+ MinIdleConns: opt.MinIdleConns,
+ MaxConnAge: opt.MaxConnAge,
+ PoolTimeout: opt.PoolTimeout,
+ IdleTimeout: opt.IdleTimeout,
IdleCheckFrequency: disableIdleCheck,
TLSConfig: opt.TLSConfig,
go node.updateLatency()
}
+ if clOpt.OnNewNode != nil {
+ clOpt.OnNewNode(node.Client)
+ }
+
return &node
}
clusterAddrs []string
closed bool
- nodeCreateGroup singleflight.Group
-
_generation uint32 // atomic
}
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()
node, ok := c.allNodes[addr]
if ok {
- _ = v.(*clusterNode).Close()
return node, err
}
- node = v.(*clusterNode)
+
+ node = newClusterNode(c.opt, addr)
c.allAddrs = appendIfNotExists(c.allAddrs, addr)
- if err == nil {
- c.clusterAddrs = append(c.clusterAddrs, addr)
- }
+ c.clusterAddrs = append(c.clusterAddrs, addr)
c.allNodes[addr] = node
return node, err
createdAt: time.Now(),
}
- isLoopbackOrigin := isLoopbackAddr(origin)
+ originHost, _, _ := net.SplitHostPort(origin)
+ isLoopbackOrigin := isLoopback(originHost)
+
for _, slot := range slots {
var nodes []*clusterNode
for i, slotNode := range slot.Nodes {
addr := slotNode.Addr
- if !isLoopbackOrigin && useOriginAddr(origin, addr) {
- addr = origin
+ if !isLoopbackOrigin {
+ addr = replaceLoopbackHost(addr, originHost)
}
node, err := c.nodes.GetOrCreate(addr)
return &c, nil
}
+func replaceLoopbackHost(nodeAddr, originHost string) string {
+ nodeHost, nodePort, err := net.SplitHostPort(nodeAddr)
+ if err != nil {
+ return nodeAddr
+ }
+
+ nodeIP := net.ParseIP(nodeHost)
+ if nodeIP == nil {
+ return nodeAddr
+ }
+
+ if !nodeIP.IsLoopback() {
+ return nodeAddr
+ }
+
+ // Use origin host which is not loopback and node port.
+ return net.JoinHostPort(originHost, nodePort)
+}
+
+func isLoopback(host string) bool {
+ ip := net.ParseIP(host)
+ if ip == nil {
+ return true
+ }
+ return ip.IsLoopback()
+}
+
func (c *clusterState) slotMasterNode(slot int) (*clusterNode, error) {
nodes := c.slotNodes(slot)
if len(nodes) > 0 {
n := rand.Intn(len(nodes)-1) + 1
slave = nodes[n]
if !slave.Loading() {
- break
+ return slave, nil
}
}
- return slave, nil
+
+ // All slaves are loading - use master.
+ return nodes[0], nil
}
}
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
-
+ state atomic.Value
reloading uint32 // atomic
}
}
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)
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
- }
+ _, err := c.Reload()
+ if err != nil {
+ return
}
+ time.Sleep(100 * time.Millisecond)
}()
}
}
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")
+ return c.Reload()
}
func (c *clusterStateHolder) ReloadOrGet() (*clusterState, error) {
c.processTxPipeline = c.defaultProcessTxPipeline
c.init()
-
- _, _ = c.state.Reload()
- _, _ = c.cmdsInfoCache.Get()
-
if opt.IdleCheckFrequency > 0 {
go c.reaper(opt.IdleCheckFrequency)
}
return c
}
-// ReloadState reloads cluster state. It calls ClusterSlots func
+func (c *ClusterClient) init() {
+ c.cmdable.setProcessor(c.Process)
+}
+
+// ReloadState reloads cluster state. If available it calls ClusterSlots func
// to get cluster slots information.
func (c *ClusterClient) ReloadState() error {
_, err := c.state.Reload()
return err
}
-func (c *ClusterClient) init() {
- c.cmdable.setProcessor(c.Process)
-}
-
func (c *ClusterClient) Context() context.Context {
if c.ctx != nil {
return c.ctx
}
func (c *ClusterClient) cmdSlot(cmd Cmder) int {
+ args := cmd.Args()
+ if args[0] == "cluster" && args[1] == "getkeysinslot" {
+ return args[2].(int)
+ }
+
cmdInfo := c.cmdInfo(cmd.Name())
return cmdSlot(cmd, cmdFirstKeyPos(cmd, cmdInfo))
}
}
cmdInfo := c.cmdInfo(cmd.Name())
- slot := cmdSlot(cmd, cmdFirstKeyPos(cmd, cmdInfo))
+ slot := c.cmdSlot(cmd)
- if cmdInfo != nil && cmdInfo.ReadOnly && c.opt.ReadOnly {
+ if c.opt.ReadOnly && cmdInfo != nil && cmdInfo.ReadOnly {
if c.opt.RouteByLatency {
node, err := state.slotClosestNode(slot)
return slot, node, err
if err == nil {
break
}
-
- if internal.IsRetryableError(err, true) {
+ if err != Nil {
c.state.LazyReload()
- continue
}
moved, ask, addr := internal.IsMovedError(err)
if moved || ask {
- c.state.LazyReload()
node, err = c.nodes.GetOrCreate(addr)
if err != nil {
return err
continue
}
- if err == pool.ErrClosed {
+ if err == pool.ErrClosed || internal.IsReadOnlyError(err) {
node, err = c.slotMasterNode(slot)
if err != nil {
return err
continue
}
+ if internal.IsRetryableError(err, true) {
+ continue
+ }
+
return err
}
return c.nodes.Close()
}
+// Do creates a Cmd from the args and processes the cmd.
+func (c *ClusterClient) Do(args ...interface{}) *Cmd {
+ cmd := NewCmd(args...)
+ c.Process(cmd)
+ return cmd
+}
+
func (c *ClusterClient) WrapProcess(
fn func(oldProcess func(Cmder) error) func(Cmder) error,
) {
if err == nil {
break
}
+ if err != Nil {
+ c.state.LazyReload()
+ }
- // If slave is loading - read from master.
+ // If slave is loading - pick another node.
if c.opt.ReadOnly && internal.IsLoadingError(err) {
node.MarkAsLoading()
- continue
- }
-
- if internal.IsRetryableError(err, true) {
- c.state.LazyReload()
-
- // First retry the same node.
- if attempt == 0 {
- continue
- }
-
- // Second try random node.
- node, err = c.nodes.Random()
- if err != nil {
- break
- }
+ node = nil
continue
}
var addr string
moved, ask, addr = internal.IsMovedError(err)
if moved || ask {
- c.state.LazyReload()
-
node, err = c.nodes.GetOrCreate(addr)
if err != nil {
break
continue
}
- if err == pool.ErrClosed {
+ if err == pool.ErrClosed || internal.IsReadOnlyError(err) {
node = nil
continue
}
+ if internal.IsRetryableError(err, true) {
+ // First retry the same node.
+ if attempt == 0 {
+ continue
+ }
+
+ // Second try random node.
+ node, err = c.nodes.Random()
+ if err != nil {
+ break
+ }
+ continue
+ }
+
break
}
acc.Timeouts += s.Timeouts
acc.TotalConns += s.TotalConns
- acc.FreeConns += s.FreeConns
+ acc.IdleConns += s.IdleConns
acc.StaleConns += s.StaleConns
}
acc.Timeouts += s.Timeouts
acc.TotalConns += s.TotalConns
- acc.FreeConns += s.FreeConns
+ acc.IdleConns += s.IdleConns
acc.StaleConns += s.StaleConns
}
}
func (c *ClusterClient) defaultProcessPipeline(cmds []Cmder) error {
- cmdsMap, err := c.mapCmdsByNode(cmds)
+ cmdsMap := newCmdsMap()
+ err := c.mapCmdsByNode(cmds, cmdsMap)
if err != nil {
setCmdsErr(cmds, err)
return err
time.Sleep(c.retryBackoff(attempt))
}
- failedCmds := make(map[*clusterNode][]Cmder)
+ failedCmds := newCmdsMap()
+ var wg sync.WaitGroup
- for node, cmds := range cmdsMap {
- cn, err := node.Client.getConn()
- if err != nil {
- if err == pool.ErrClosed {
- c.remapCmds(cmds, failedCmds)
- } else {
- setCmdsErr(cmds, err)
+ for node, cmds := range cmdsMap.m {
+ wg.Add(1)
+ go func(node *clusterNode, cmds []Cmder) {
+ defer wg.Done()
+
+ cn, err := node.Client.getConn()
+ if err != nil {
+ if err == pool.ErrClosed {
+ c.mapCmdsByNode(cmds, failedCmds)
+ } else {
+ setCmdsErr(cmds, err)
+ }
+ return
}
- continue
- }
- err = c.pipelineProcessCmds(node, cn, cmds, failedCmds)
- if err == nil || internal.IsRedisError(err) {
- node.Client.connPool.Put(cn)
- } else {
- node.Client.connPool.Remove(cn)
- }
+ err = c.pipelineProcessCmds(node, cn, cmds, failedCmds)
+ node.Client.releaseConnStrict(cn, err)
+ }(node, cmds)
}
- if len(failedCmds) == 0 {
+ wg.Wait()
+ if len(failedCmds.m) == 0 {
break
}
cmdsMap = failedCmds
}
- return firstCmdsErr(cmds)
+ return cmdsFirstErr(cmds)
}
-func (c *ClusterClient) mapCmdsByNode(cmds []Cmder) (map[*clusterNode][]Cmder, error) {
+type cmdsMap struct {
+ mu sync.Mutex
+ m map[*clusterNode][]Cmder
+}
+
+func newCmdsMap() *cmdsMap {
+ return &cmdsMap{
+ m: make(map[*clusterNode][]Cmder),
+ }
+}
+
+func (c *ClusterClient) mapCmdsByNode(cmds []Cmder, cmdsMap *cmdsMap) error {
state, err := c.state.Get()
if err != nil {
setCmdsErr(cmds, err)
- return nil, err
+ return err
}
- cmdsMap := make(map[*clusterNode][]Cmder)
cmdsAreReadOnly := c.cmdsAreReadOnly(cmds)
for _, cmd := range cmds {
var node *clusterNode
node, err = state.slotMasterNode(slot)
}
if err != nil {
- return nil, err
+ return err
}
- cmdsMap[node] = append(cmdsMap[node], cmd)
+ cmdsMap.mu.Lock()
+ cmdsMap.m[node] = append(cmdsMap.m[node], cmd)
+ cmdsMap.mu.Unlock()
}
- return cmdsMap, nil
+ return nil
}
func (c *ClusterClient) cmdsAreReadOnly(cmds []Cmder) bool {
return true
}
-func (c *ClusterClient) remapCmds(cmds []Cmder, failedCmds map[*clusterNode][]Cmder) {
- remappedCmds, err := c.mapCmdsByNode(cmds)
- if err != nil {
- setCmdsErr(cmds, err)
- return
- }
-
- for node, cmds := range remappedCmds {
- failedCmds[node] = cmds
- }
-}
-
func (c *ClusterClient) pipelineProcessCmds(
- node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
+ node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds *cmdsMap,
) error {
- cn.SetWriteTimeout(c.opt.WriteTimeout)
-
- err := writeCmd(cn, cmds...)
+ err := cn.WithWriter(c.opt.WriteTimeout, func(wr *proto.Writer) error {
+ return writeCmd(wr, cmds...)
+ })
if err != nil {
setCmdsErr(cmds, err)
- failedCmds[node] = cmds
+ failedCmds.mu.Lock()
+ failedCmds.m[node] = cmds
+ failedCmds.mu.Unlock()
return err
}
- // Set read timeout for all commands.
- cn.SetReadTimeout(c.opt.ReadTimeout)
-
- return c.pipelineReadCmds(cn, cmds, failedCmds)
+ err = cn.WithReader(c.opt.ReadTimeout, func(rd *proto.Reader) error {
+ return c.pipelineReadCmds(node, rd, cmds, failedCmds)
+ })
+ return err
}
func (c *ClusterClient) pipelineReadCmds(
- cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
+ node *clusterNode, rd *proto.Reader, cmds []Cmder, failedCmds *cmdsMap,
) error {
+ var firstErr error
for _, cmd := range cmds {
- err := cmd.readReply(cn)
+ err := cmd.readReply(rd)
if err == nil {
continue
}
continue
}
- return err
+ failedCmds.mu.Lock()
+ failedCmds.m[node] = append(failedCmds.m[node], cmd)
+ failedCmds.mu.Unlock()
+ if firstErr == nil {
+ firstErr = err
+ }
}
- return nil
+ return firstErr
}
func (c *ClusterClient) checkMovedErr(
- cmd Cmder, err error, failedCmds map[*clusterNode][]Cmder,
+ cmd Cmder, err error, failedCmds *cmdsMap,
) bool {
moved, ask, addr := internal.IsMovedError(err)
return false
}
- failedCmds[node] = append(failedCmds[node], cmd)
+ failedCmds.mu.Lock()
+ failedCmds.m[node] = append(failedCmds.m[node], cmd)
+ failedCmds.mu.Unlock()
return true
}
return false
}
- failedCmds[node] = append(failedCmds[node], NewCmd("ASKING"), cmd)
+ failedCmds.mu.Lock()
+ failedCmds.m[node] = append(failedCmds.m[node], NewCmd("ASKING"), cmd)
+ failedCmds.mu.Unlock()
return true
}
time.Sleep(c.retryBackoff(attempt))
}
- failedCmds := make(map[*clusterNode][]Cmder)
+ failedCmds := newCmdsMap()
+ var wg sync.WaitGroup
for node, cmds := range cmdsMap {
- cn, err := node.Client.getConn()
- if err != nil {
- if err == pool.ErrClosed {
- c.remapCmds(cmds, failedCmds)
- } else {
- setCmdsErr(cmds, err)
+ wg.Add(1)
+ go func(node *clusterNode, cmds []Cmder) {
+ defer wg.Done()
+
+ cn, err := node.Client.getConn()
+ if err != nil {
+ if err == pool.ErrClosed {
+ c.mapCmdsByNode(cmds, failedCmds)
+ } else {
+ setCmdsErr(cmds, err)
+ }
+ return
}
- continue
- }
- err = c.txPipelineProcessCmds(node, cn, cmds, failedCmds)
- if err == nil || internal.IsRedisError(err) {
- node.Client.connPool.Put(cn)
- } else {
- node.Client.connPool.Remove(cn)
- }
+ err = c.txPipelineProcessCmds(node, cn, cmds, failedCmds)
+ node.Client.releaseConnStrict(cn, err)
+ }(node, cmds)
}
- if len(failedCmds) == 0 {
+ wg.Wait()
+ if len(failedCmds.m) == 0 {
break
}
- cmdsMap = failedCmds
+ cmdsMap = failedCmds.m
}
}
- return firstCmdsErr(cmds)
+ return cmdsFirstErr(cmds)
}
func (c *ClusterClient) mapCmdsBySlot(cmds []Cmder) map[int][]Cmder {
}
func (c *ClusterClient) txPipelineProcessCmds(
- node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
+ node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds *cmdsMap,
) error {
- cn.SetWriteTimeout(c.opt.WriteTimeout)
- if err := txPipelineWriteMulti(cn, cmds); err != nil {
- setCmdsErr(cmds, err)
- failedCmds[node] = cmds
- return err
- }
-
- // Set read timeout for all commands.
- cn.SetReadTimeout(c.opt.ReadTimeout)
-
- if err := c.txPipelineReadQueued(cn, cmds, failedCmds); err != nil {
+ err := cn.WithWriter(c.opt.WriteTimeout, func(wr *proto.Writer) error {
+ return txPipelineWriteMulti(wr, cmds)
+ })
+ if err != nil {
setCmdsErr(cmds, err)
+ failedCmds.mu.Lock()
+ failedCmds.m[node] = cmds
+ failedCmds.mu.Unlock()
return err
}
- return pipelineReadCmds(cn, cmds)
+ err = cn.WithReader(c.opt.ReadTimeout, func(rd *proto.Reader) error {
+ err := c.txPipelineReadQueued(rd, cmds, failedCmds)
+ if err != nil {
+ setCmdsErr(cmds, err)
+ return err
+ }
+ return pipelineReadCmds(rd, cmds)
+ })
+ return err
}
func (c *ClusterClient) txPipelineReadQueued(
- cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
+ rd *proto.Reader, cmds []Cmder, failedCmds *cmdsMap,
) error {
// Parse queued replies.
var statusCmd StatusCmd
- if err := statusCmd.readReply(cn); err != nil {
+ if err := statusCmd.readReply(rd); err != nil {
return err
}
for _, cmd := range cmds {
- err := statusCmd.readReply(cn)
+ err := statusCmd.readReply(rd)
if err == nil {
continue
}
}
// Parse number of replies.
- line, err := cn.Rd.ReadLine()
+ line, err := rd.ReadLine()
if err != nil {
if err == Nil {
err = TxFailedErr
return nil
}
-func (c *ClusterClient) pubSub(channels []string) *PubSub {
+func (c *ClusterClient) pubSub() *PubSub {
var node *clusterNode
pubsub := &PubSub{
opt: c.opt.clientOptions(),
newConn: func(channels []string) (*pool.Conn, error) {
- if node == nil {
- var slot int
- if len(channels) > 0 {
- slot = hashtag.Slot(channels[0])
- } else {
- slot = -1
- }
+ if node != nil {
+ panic("node != nil")
+ }
- masterNode, err := c.slotMasterNode(slot)
- if err != nil {
- return nil, err
- }
- node = masterNode
+ slot := hashtag.Slot(channels[0])
+
+ var err error
+ node, err = c.slotMasterNode(slot)
+ if err != nil {
+ return nil, err
}
- return node.Client.newConn()
+
+ cn, err := node.Client.newConn()
+ if err != nil {
+ return nil, err
+ }
+
+ return cn, nil
},
closeConn: func(cn *pool.Conn) error {
- return node.Client.connPool.CloseConn(cn)
+ err := node.Client.connPool.CloseConn(cn)
+ node = nil
+ return err
},
}
pubsub.init()
+
return pubsub
}
// Subscribe subscribes the client to the specified channels.
// Channels can be omitted to create empty subscription.
func (c *ClusterClient) Subscribe(channels ...string) *PubSub {
- pubsub := c.pubSub(channels)
+ pubsub := c.pubSub()
if len(channels) > 0 {
_ = pubsub.Subscribe(channels...)
}
// PSubscribe subscribes the client to the given patterns.
// Patterns can be omitted to create empty subscription.
func (c *ClusterClient) PSubscribe(channels ...string) *PubSub {
- pubsub := c.pubSub(channels)
+ pubsub := c.pubSub()
if len(channels) > 0 {
_ = pubsub.PSubscribe(channels...)
}
return pubsub
}
-func useOriginAddr(originAddr, nodeAddr string) bool {
- nodeHost, nodePort, err := net.SplitHostPort(nodeAddr)
- if err != nil {
- return false
- }
-
- nodeIP := net.ParseIP(nodeHost)
- if nodeIP == nil {
- return false
- }
-
- if !nodeIP.IsLoopback() {
- return false
- }
-
- _, originPort, err := net.SplitHostPort(originAddr)
- if err != nil {
- return false
- }
-
- return nodePort == originPort
-}
-
-func isLoopbackAddr(addr string) bool {
- host, _, err := net.SplitHostPort(addr)
- if err != nil {
- return false
- }
-
- ip := net.ParseIP(host)
- if ip == nil {
- return false
- }
-
- return ip.IsLoopback()
-}
-
func appendUniqueNode(nodes []*clusterNode, node *clusterNode) []*clusterNode {
for _, n := range nodes {
if n == node {
Code Review / barometer.git / blobdiff