redis/cluster.go

1243 lines
24 KiB
Go
Raw Normal View History

2015-01-24 15:12:48 +03:00
package redis
import (
2016-12-13 18:28:39 +03:00
"fmt"
2015-01-24 15:12:48 +03:00
"math/rand"
"net"
2015-01-24 15:12:48 +03:00
"sync"
"sync/atomic"
"time"
2015-12-30 16:53:45 +03:00
2017-02-18 17:42:34 +03:00
"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"
2015-01-24 15:12:48 +03:00
)
2017-08-31 15:22:47 +03:00
var errClusterNoNodes = fmt.Errorf("redis: cluster has no nodes")
var errNilClusterState = fmt.Errorf("redis: cannot load cluster slots")
2016-10-09 14:12:32 +03:00
// ClusterOptions are used to configure a cluster client and should be
// passed to NewClusterClient.
type ClusterOptions struct {
// A seed list of host:port addresses of cluster nodes.
Addrs []string
// The maximum number of retries before giving up. Command is retried
// on network errors and MOVED/ASK redirects.
// Default is 16.
MaxRedirects int
2017-07-09 13:10:07 +03:00
// Enables read-only commands on slave nodes.
ReadOnly bool
2017-07-09 13:10:07 +03:00
// Allows routing read-only commands to the closest master or slave node.
RouteByLatency bool
// Following options are copied from Options struct.
2017-05-25 14:16:39 +03:00
OnConnect func(*Conn) error
2017-07-09 13:10:07 +03:00
MaxRetries int
MinRetryBackoff time.Duration
MaxRetryBackoff time.Duration
Password string
DialTimeout time.Duration
ReadTimeout time.Duration
WriteTimeout time.Duration
// PoolSize applies per cluster node and not for the whole cluster.
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
}
2017-07-09 13:10:07 +03:00
2017-08-31 15:22:47 +03:00
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
}
2017-07-09 13:10:07 +03:00
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{
2017-05-25 14:16:39 +03:00
OnConnect: opt.OnConnect,
2017-07-09 13:10:07 +03:00
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,
}
}
//------------------------------------------------------------------------------
type clusterNode struct {
Client *Client
2016-06-17 15:09:38 +03:00
Latency time.Duration
2017-07-09 13:10:07 +03:00
loading time.Time
generation uint32
2016-10-09 11:18:57 +03:00
}
func newClusterNode(clOpt *ClusterOptions, addr string) *clusterNode {
opt := clOpt.clientOptions()
opt.Addr = addr
node := clusterNode{
Client: NewClient(opt),
}
if clOpt.RouteByLatency {
2016-12-16 17:26:48 +03:00
node.updateLatency()
}
return &node
}
2016-12-16 17:26:48 +03:00
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
}
2016-10-09 11:18:57 +03:00
func (n *clusterNode) Loading() bool {
return !n.loading.IsZero() && time.Since(n.loading) < time.Minute
}
2017-07-09 13:10:07 +03:00
func (n *clusterNode) Generation() uint32 {
return n.generation
}
func (n *clusterNode) SetGeneration(gen uint32) {
if gen < n.generation {
panic("gen < n.generation")
}
n.generation = gen
}
//------------------------------------------------------------------------------
2015-01-24 15:12:48 +03:00
type clusterNodes struct {
2016-03-17 19:00:47 +03:00
opt *ClusterOptions
mu sync.RWMutex
addrs []string
nodes map[string]*clusterNode
closed bool
2017-07-09 13:10:07 +03:00
generation uint32
2015-01-24 15:12:48 +03:00
}
func newClusterNodes(opt *ClusterOptions) *clusterNodes {
return &clusterNodes{
opt: opt,
nodes: make(map[string]*clusterNode),
2015-01-24 15:12:48 +03:00
}
}
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
2016-03-17 19:00:47 +03:00
}
2017-09-11 08:58:56 +03:00
func (c *clusterNodes) Addrs() ([]string, error) {
2017-08-31 15:22:47 +03:00
c.mu.RLock()
2017-09-11 08:58:56 +03:00
closed := c.closed
addrs := c.addrs
c.mu.RUnlock()
2017-08-31 15:22:47 +03:00
2017-09-11 08:58:56 +03:00
if closed {
return nil, pool.ErrClosed
2017-08-31 15:22:47 +03:00
}
2017-09-11 08:58:56 +03:00
if len(addrs) == 0 {
return nil, errClusterNoNodes
2017-08-31 15:22:47 +03:00
}
2017-09-11 08:58:56 +03:00
return addrs, nil
2017-08-31 15:22:47 +03:00
}
2017-07-09 13:10:07 +03:00
func (c *clusterNodes) NextGeneration() uint32 {
c.generation++
return c.generation
}
// GC removes unused nodes.
func (c *clusterNodes) GC(generation uint32) error {
var collected []*clusterNode
c.mu.Lock()
for i := 0; i < len(c.addrs); {
addr := c.addrs[i]
node := c.nodes[addr]
if node.Generation() >= generation {
i++
continue
}
c.addrs = append(c.addrs[:i], c.addrs[i+1:]...)
delete(c.nodes, addr)
collected = append(collected, node)
}
c.mu.Unlock()
var firstErr error
for _, node := range collected {
if err := node.Client.Close(); err != nil && firstErr == nil {
firstErr = err
}
}
return firstErr
}
func (c *clusterNodes) All() ([]*clusterNode, error) {
2016-10-09 14:12:32 +03:00
c.mu.RLock()
defer c.mu.RUnlock()
if c.closed {
return nil, pool.ErrClosed
}
2015-12-16 17:11:52 +03:00
nodes := make([]*clusterNode, 0, len(c.nodes))
for _, node := range c.nodes {
nodes = append(nodes, node)
2016-01-19 19:36:40 +03:00
}
return nodes, nil
2016-01-19 19:36:40 +03:00
}
2017-07-09 13:10:07 +03:00
func (c *clusterNodes) GetOrCreate(addr string) (*clusterNode, error) {
var node *clusterNode
var ok bool
2015-01-24 15:12:48 +03:00
c.mu.RLock()
if !c.closed {
node, ok = c.nodes[addr]
}
c.mu.RUnlock()
if ok {
return node, nil
2016-03-17 19:00:47 +03:00
}
c.mu.Lock()
defer c.mu.Unlock()
if c.closed {
return nil, pool.ErrClosed
}
node, ok = c.nodes[addr]
if ok {
return node, nil
2015-03-18 13:41:24 +03:00
}
c.addrs = append(c.addrs, addr)
node = newClusterNode(c.opt, addr)
c.nodes[addr] = node
return node, nil
}
2015-01-24 15:12:48 +03:00
func (c *clusterNodes) Random() (*clusterNode, error) {
2017-09-11 08:58:56 +03:00
addrs, err := c.Addrs()
if err != nil {
return nil, err
}
var nodeErr error
for i := 0; i <= c.opt.MaxRedirects; i++ {
n := rand.Intn(len(addrs))
2017-07-09 13:10:07 +03:00
node, err := c.GetOrCreate(addrs[n])
if err != nil {
return nil, err
}
nodeErr = node.Client.ClusterInfo().Err()
if nodeErr == nil {
return node, nil
}
}
return nil, nodeErr
2015-01-24 15:12:48 +03:00
}
//------------------------------------------------------------------------------
type clusterState struct {
2017-07-09 13:10:07 +03:00
nodes *clusterNodes
masters []*clusterNode
slaves []*clusterNode
slots [][]*clusterNode
2017-07-09 13:10:07 +03:00
generation uint32
}
func newClusterState(nodes *clusterNodes, slots []ClusterSlot, origin string) (*clusterState, error) {
c := clusterState{
2017-07-09 13:10:07 +03:00
nodes: nodes,
generation: nodes.NextGeneration(),
slots: make([][]*clusterNode, hashtag.SlotNumber),
}
isLoopbackOrigin := isLoopbackAddr(origin)
for _, slot := range slots {
var nodes []*clusterNode
2017-07-09 13:10:07 +03:00
for i, slotNode := range slot.Nodes {
addr := slotNode.Addr
if !isLoopbackOrigin && isLoopbackAddr(addr) {
addr = origin
}
2017-07-09 13:10:07 +03:00
node, err := c.nodes.GetOrCreate(addr)
if err != nil {
return nil, err
}
2017-07-09 13:10:07 +03:00
node.SetGeneration(c.generation)
nodes = append(nodes, node)
2017-07-09 13:10:07 +03:00
if i == 0 {
c.masters = appendNode(c.masters, node)
} else {
c.slaves = appendNode(c.slaves, 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:
2016-10-09 11:18:57 +03:00
if slave := nodes[1]; !slave.Loading() {
return slave, nil
}
return nodes[0], nil
default:
2016-10-09 11:18:57 +03:00
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) {
2016-12-16 17:26:48 +03:00
const threshold = time.Millisecond
nodes := c.slotNodes(slot)
if len(nodes) == 0 {
return c.nodes.Random()
}
var node *clusterNode
for _, n := range nodes {
2016-12-16 17:26:48 +03:00
if n.Loading() {
continue
}
if node == nil || node.Latency-n.Latency > threshold {
node = n
2015-04-04 16:46:57 +03:00
}
}
return node, nil
2015-04-04 16:46:57 +03:00
}
func (c *clusterState) slotNodes(slot int) []*clusterNode {
2017-07-09 10:07:20 +03:00
if slot >= 0 && slot < len(c.slots) {
return c.slots[slot]
}
return nil
}
//------------------------------------------------------------------------------
// ClusterClient is a Redis Cluster client representing a pool of zero
// or more underlying connections. It's safe for concurrent use by
// multiple goroutines.
type ClusterClient struct {
cmdable
opt *ClusterOptions
nodes *clusterNodes
_state atomic.Value
2017-07-01 12:51:06 +03:00
cmdsInfoOnce internal.Once
2017-06-17 12:34:39 +03:00
cmdsInfo map[string]*CommandInfo
2017-07-09 13:10:07 +03:00
// Reports whether slots reloading is in progress.
reloading uint32
}
// NewClusterClient returns a Redis Cluster client as described in
// http://redis.io/topics/cluster-spec.
func NewClusterClient(opt *ClusterOptions) *ClusterClient {
opt.init()
c := &ClusterClient{
opt: opt,
nodes: newClusterNodes(opt),
}
2017-05-25 13:38:04 +03:00
c.setProcessor(c.Process)
// Add initial nodes.
for _, addr := range opt.Addrs {
2017-07-09 13:10:07 +03:00
_, _ = c.nodes.GetOrCreate(addr)
}
2017-03-04 14:04:27 +03:00
// Preload cluster slots.
for i := 0; i < 10; i++ {
2017-07-09 13:10:07 +03:00
state, err := c.reloadState()
2017-03-04 14:04:27 +03:00
if err == nil {
c._state.Store(state)
break
}
}
if opt.IdleCheckFrequency > 0 {
go c.reaper(opt.IdleCheckFrequency)
}
return c
}
2017-03-20 13:15:21 +03:00
// Options returns read-only Options that were used to create the client.
func (c *ClusterClient) Options() *ClusterOptions {
return c.opt
}
2017-08-31 15:22:47 +03:00
func (c *ClusterClient) retryBackoff(attempt int) time.Duration {
return internal.RetryBackoff(attempt, c.opt.MinRetryBackoff, c.opt.MaxRetryBackoff)
}
func (c *ClusterClient) state() (*clusterState, error) {
v := c._state.Load()
2017-03-04 14:04:27 +03:00
if v != nil {
2017-08-31 15:22:47 +03:00
return v.(*clusterState), nil
}
2017-09-11 08:58:56 +03:00
_, err := c.nodes.Addrs()
if err != nil {
2017-08-31 15:22:47 +03:00
return nil, err
}
2017-08-31 15:22:47 +03:00
2017-07-09 13:10:07 +03:00
c.lazyReloadState()
2017-08-31 15:22:47 +03:00
return nil, errNilClusterState
}
2017-06-17 12:34:39 +03:00
func (c *ClusterClient) cmdInfo(name string) *CommandInfo {
err := c.cmdsInfoOnce.Do(func() error {
node, err := c.nodes.Random()
if err != nil {
return err
}
cmdsInfo, err := node.Client.Command().Result()
if err != nil {
return err
}
c.cmdsInfo = cmdsInfo
return nil
})
if err != nil {
return nil
}
2017-08-31 15:22:47 +03:00
info := c.cmdsInfo[name]
if info == nil {
internal.Logf("info for cmd=%s not found", name)
}
return info
2017-06-17 12:34:39 +03:00
}
2017-08-31 15:22:47 +03:00
func (c *ClusterClient) cmdSlot(cmd Cmder) int {
cmdInfo := c.cmdInfo(cmd.Name())
firstKey := cmd.arg(cmdFirstKeyPos(cmd, cmdInfo))
return hashtag.Slot(firstKey)
}
2017-08-31 15:22:47 +03:00
func (c *ClusterClient) cmdSlotAndNode(state *clusterState, cmd Cmder) (int, *clusterNode, error) {
2017-06-17 12:34:39 +03:00
cmdInfo := c.cmdInfo(cmd.Name())
firstKey := cmd.arg(cmdFirstKeyPos(cmd, cmdInfo))
slot := hashtag.Slot(firstKey)
2016-12-09 16:52:36 +03:00
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
2015-04-04 16:46:57 +03:00
}
node, err := state.slotMasterNode(slot)
return slot, node, err
}
func (c *ClusterClient) Watch(fn func(*Tx) error, keys ...string) error {
2017-08-31 15:22:47 +03:00
if len(keys) == 0 {
return fmt.Errorf("redis: keys don't hash to the same slot")
}
2017-03-04 14:04:27 +03:00
2017-08-31 15:22:47 +03:00
state, err := c.state()
if err != nil {
return err
}
slot := hashtag.Slot(keys[0])
for _, key := range keys[1:] {
if hashtag.Slot(key) != slot {
return fmt.Errorf("redis: Watch requires all keys to be in the same slot")
}
}
2017-08-31 15:22:47 +03:00
node, err := state.slotMasterNode(slot)
if err != nil {
return err
}
2017-08-31 15:22:47 +03:00
for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
if attempt > 0 {
time.Sleep(c.retryBackoff(attempt))
}
err = node.Client.Watch(fn, keys...)
if err == nil {
break
}
moved, ask, addr := internal.IsMovedError(err)
if moved || ask {
c.lazyReloadState()
node, err = c.nodes.GetOrCreate(addr)
if err != nil {
return err
}
continue
}
return err
}
return err
}
// Close closes the cluster client, releasing any open resources.
//
// It is rare to Close a ClusterClient, as the ClusterClient is meant
// to be long-lived and shared between many goroutines.
func (c *ClusterClient) Close() error {
return c.nodes.Close()
}
2016-06-17 15:09:38 +03:00
func (c *ClusterClient) Process(cmd Cmder) error {
2017-08-31 15:22:47 +03:00
state, err := c.state()
if err != nil {
cmd.setErr(err)
return err
}
_, node, err := c.cmdSlotAndNode(state, cmd)
2016-06-17 15:09:38 +03:00
if err != nil {
cmd.setErr(err)
return err
}
var ask bool
2016-06-05 14:10:30 +03:00
for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
2017-07-09 13:10:07 +03:00
if attempt > 0 {
2017-08-31 15:22:47 +03:00
time.Sleep(c.retryBackoff(attempt))
2017-07-09 13:10:07 +03:00
}
2015-01-24 15:12:48 +03:00
if ask {
pipe := node.Client.Pipeline()
2015-01-24 15:12:48 +03:00
pipe.Process(NewCmd("ASKING"))
pipe.Process(cmd)
2016-10-09 11:18:57 +03:00
_, err = pipe.Exec()
pipe.Close()
2015-01-24 15:12:48 +03:00
ask = false
} else {
2016-10-09 11:18:57 +03:00
err = node.Client.Process(cmd)
2015-01-24 15:12:48 +03:00
}
2017-07-09 13:10:07 +03:00
// If there is no error - we are done.
2016-03-19 17:33:14 +03:00
if err == nil {
2017-08-31 15:22:47 +03:00
break
2015-01-24 15:12:48 +03:00
}
2016-10-09 11:18:57 +03:00
// If slave is loading - read from master.
if c.opt.ReadOnly && internal.IsLoadingError(err) {
2017-07-09 13:10:07 +03:00
// TODO: race
2016-10-09 11:18:57 +03:00
node.loading = time.Now()
continue
}
2017-08-31 15:22:47 +03:00
if internal.IsRetryableError(err) {
var nodeErr error
node, nodeErr = c.nodes.Random()
if nodeErr != nil {
break
}
2015-01-24 15:12:48 +03:00
continue
}
2015-03-18 13:41:24 +03:00
var moved bool
var addr string
moved, ask, addr = internal.IsMovedError(err)
2015-03-18 13:41:24 +03:00
if moved || ask {
2017-08-31 15:22:47 +03:00
c.lazyReloadState()
2017-08-31 15:22:47 +03:00
var nodeErr error
node, nodeErr = c.nodes.GetOrCreate(addr)
if nodeErr != nil {
break
2016-06-17 15:09:38 +03:00
}
2015-03-18 13:41:24 +03:00
continue
2015-01-24 15:12:48 +03:00
}
2015-03-18 13:41:24 +03:00
break
2015-01-24 15:12:48 +03:00
}
2016-06-17 15:09:38 +03:00
return cmd.Err()
}
2017-07-09 13:10:07 +03:00
// ForEachMaster concurrently calls the fn on each master node in the cluster.
// It returns the first error if any.
2017-07-09 13:10:07 +03:00
func (c *ClusterClient) ForEachMaster(fn func(client *Client) error) error {
2017-08-31 15:22:47 +03:00
state, err := c.state()
if err != nil {
return err
}
var wg sync.WaitGroup
errCh := make(chan error, 1)
2017-07-09 13:10:07 +03:00
for _, master := range state.masters {
wg.Add(1)
go func(node *clusterNode) {
defer wg.Done()
err := fn(node.Client)
if err != nil {
select {
case errCh <- err:
default:
}
}
2017-07-09 13:10:07 +03:00
}(master)
}
wg.Wait()
select {
case err := <-errCh:
return err
default:
return nil
}
}
2017-07-09 13:10:07 +03:00
// ForEachSlave concurrently calls the fn on each slave node in the cluster.
2016-06-17 15:09:38 +03:00
// It returns the first error if any.
2017-07-09 13:10:07 +03:00
func (c *ClusterClient) ForEachSlave(fn func(client *Client) error) error {
2017-08-31 15:22:47 +03:00
state, err := c.state()
if err != nil {
return err
}
2016-06-17 15:09:38 +03:00
var wg sync.WaitGroup
errCh := make(chan error, 1)
2017-07-09 13:10:07 +03:00
for _, slave := range state.slaves {
2016-06-17 15:09:38 +03:00
wg.Add(1)
go func(node *clusterNode) {
defer wg.Done()
2016-06-17 15:09:38 +03:00
err := fn(node.Client)
if err != nil {
select {
case errCh <- err:
default:
2016-06-17 15:09:38 +03:00
}
}
2017-07-09 13:10:07 +03:00
}(slave)
2016-06-17 15:09:38 +03:00
}
wg.Wait()
select {
case err := <-errCh:
return err
default:
return nil
}
2015-01-24 15:12:48 +03:00
}
2017-07-09 13:10:07 +03:00
// ForEachNode concurrently calls the fn on each known node in the cluster.
// It returns the first error if any.
func (c *ClusterClient) ForEachNode(fn func(client *Client) error) error {
2017-08-31 15:22:47 +03:00
state, err := c.state()
if err != nil {
return err
2017-07-09 13:10:07 +03:00
}
var wg sync.WaitGroup
errCh := make(chan error, 1)
worker := func(node *clusterNode) {
defer wg.Done()
err := fn(node.Client)
if err != nil {
select {
case errCh <- err:
default:
}
}
}
for _, node := range state.masters {
wg.Add(1)
go worker(node)
}
for _, node := range state.slaves {
wg.Add(1)
go worker(node)
}
wg.Wait()
select {
case err := <-errCh:
return err
default:
return nil
}
}
// PoolStats returns accumulated connection pool stats.
func (c *ClusterClient) PoolStats() *PoolStats {
2017-03-04 14:04:27 +03:00
var acc PoolStats
2017-08-31 15:22:47 +03:00
state, _ := c.state()
2017-07-09 13:10:07 +03:00
if state == nil {
2017-03-04 14:04:27 +03:00
return &acc
}
2017-07-09 13:10:07 +03:00
for _, node := range state.masters {
s := node.Client.connPool.Stats()
acc.Hits += s.Hits
acc.Misses += s.Misses
2017-07-09 13:10:07 +03:00
acc.Timeouts += s.Timeouts
2017-07-09 13:10:07 +03:00
acc.TotalConns += s.TotalConns
acc.FreeConns += s.FreeConns
acc.StaleConns += s.StaleConns
2017-07-09 13:10:07 +03:00
}
for _, node := range state.slaves {
s := node.Client.connPool.Stats()
acc.Hits += s.Hits
acc.Misses += s.Misses
acc.Timeouts += s.Timeouts
acc.TotalConns += s.TotalConns
acc.FreeConns += s.FreeConns
acc.StaleConns += s.StaleConns
2015-01-24 15:12:48 +03:00
}
2017-07-09 13:10:07 +03:00
return &acc
}
2015-01-24 15:12:48 +03:00
2017-07-09 13:10:07 +03:00
func (c *ClusterClient) lazyReloadState() {
2016-06-17 15:09:38 +03:00
if !atomic.CompareAndSwapUint32(&c.reloading, 0, 1) {
return
}
2017-03-04 14:04:27 +03:00
go func() {
2017-07-09 13:10:07 +03:00
defer atomic.StoreUint32(&c.reloading, 0)
for {
2017-08-31 15:22:47 +03:00
state, err := c.reloadState()
2017-03-04 14:04:27 +03:00
if err == pool.ErrClosed {
2017-07-09 13:10:07 +03:00
return
2017-03-04 14:04:27 +03:00
}
2017-07-09 13:10:07 +03:00
if err != nil {
time.Sleep(time.Millisecond)
continue
2017-03-04 14:04:27 +03:00
}
2017-07-09 13:10:07 +03:00
c._state.Store(state)
2017-08-31 15:22:47 +03:00
time.Sleep(5 * time.Second)
c.nodes.GC(state.generation)
2017-07-09 13:10:07 +03:00
break
2017-03-04 14:04:27 +03:00
}
}()
2015-01-24 15:12:48 +03:00
}
2017-07-09 13:10:07 +03:00
// Not thread-safe.
func (c *ClusterClient) reloadState() (*clusterState, error) {
2017-03-04 14:04:27 +03:00
node, err := c.nodes.Random()
if err != nil {
return nil, err
}
2017-03-04 14:04:27 +03:00
slots, err := node.Client.ClusterSlots().Result()
if err != nil {
return nil, err
2015-05-01 10:42:58 +03:00
}
2017-03-04 14:04:27 +03:00
return newClusterState(c.nodes, slots, node.Client.opt.Addr)
2015-01-24 15:12:48 +03:00
}
// reaper closes idle connections to the cluster.
func (c *ClusterClient) reaper(idleCheckFrequency time.Duration) {
ticker := time.NewTicker(idleCheckFrequency)
defer ticker.Stop()
2017-04-02 17:10:47 +03:00
for range ticker.C {
nodes, err := c.nodes.All()
if err != nil {
break
}
for _, node := range nodes {
_, err := node.Client.connPool.(*pool.ConnPool).ReapStaleConns()
2016-03-17 19:00:47 +03:00
if err != nil {
2016-04-09 14:52:01 +03:00
internal.Logf("ReapStaleConns failed: %s", err)
}
}
}
}
2017-05-02 18:00:53 +03:00
func (c *ClusterClient) Pipeline() Pipeliner {
pipe := Pipeline{
exec: c.pipelineExec,
}
2017-05-25 13:38:04 +03:00
pipe.setProcessor(pipe.Process)
return &pipe
}
2017-05-02 18:00:53 +03:00
func (c *ClusterClient) Pipelined(fn func(Pipeliner) error) ([]Cmder, error) {
return c.Pipeline().pipelined(fn)
}
func (c *ClusterClient) pipelineExec(cmds []Cmder) error {
2016-12-13 18:28:39 +03:00
cmdsMap, err := c.mapCmdsByNode(cmds)
if err != nil {
2017-08-31 15:22:47 +03:00
setCmdsErr(cmds, err)
2016-12-13 18:28:39 +03:00
return err
}
2017-08-31 15:22:47 +03:00
for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
if attempt > 0 {
time.Sleep(c.retryBackoff(attempt))
}
failedCmds := make(map[*clusterNode][]Cmder)
for node, cmds := range cmdsMap {
2017-07-09 10:07:20 +03:00
cn, _, err := node.Client.getConn()
if err != nil {
2016-06-17 15:09:38 +03:00
setCmdsErr(cmds, err)
continue
}
2017-08-31 15:22:47 +03:00
err = c.pipelineProcessCmds(node, cn, cmds, failedCmds)
if err == nil || internal.IsRedisError(err) {
_ = node.Client.connPool.Put(cn)
} else {
_ = node.Client.connPool.Remove(cn)
}
}
2016-12-13 18:28:39 +03:00
if len(failedCmds) == 0 {
break
}
cmdsMap = failedCmds
}
2017-08-31 15:22:47 +03:00
return firstCmdsErr(cmds)
}
2016-12-13 18:28:39 +03:00
func (c *ClusterClient) mapCmdsByNode(cmds []Cmder) (map[*clusterNode][]Cmder, error) {
2017-08-31 15:22:47 +03:00
state, err := c.state()
if err != nil {
setCmdsErr(cmds, err)
return nil, err
}
2016-12-13 18:28:39 +03:00
cmdsMap := make(map[*clusterNode][]Cmder)
for _, cmd := range cmds {
2017-08-31 15:22:47 +03:00
slot := c.cmdSlot(cmd)
node, err := state.slotMasterNode(slot)
2016-12-13 18:28:39 +03:00
if err != nil {
return nil, err
}
cmdsMap[node] = append(cmdsMap[node], cmd)
}
return cmdsMap, nil
}
func (c *ClusterClient) pipelineProcessCmds(
2017-08-31 15:22:47 +03:00
node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
2016-12-13 18:28:39 +03:00
) error {
cn.SetWriteTimeout(c.opt.WriteTimeout)
if err := writeCmd(cn, cmds...); err != nil {
setCmdsErr(cmds, err)
2017-08-31 15:22:47 +03:00
failedCmds[node] = cmds
2016-12-13 18:28:39 +03:00
return err
}
2016-12-13 18:28:39 +03:00
// Set read timeout for all commands.
cn.SetReadTimeout(c.opt.ReadTimeout)
return c.pipelineReadCmds(cn, cmds, failedCmds)
}
func (c *ClusterClient) pipelineReadCmds(
cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
) error {
for _, cmd := range cmds {
err := cmd.readReply(cn)
if err == nil {
continue
}
2017-08-31 15:22:47 +03:00
if c.checkMovedErr(cmd, err, failedCmds) {
continue
}
2016-12-13 18:28:39 +03:00
2017-08-31 15:22:47 +03:00
if internal.IsRedisError(err) {
continue
2016-12-13 18:28:39 +03:00
}
2017-08-31 15:22:47 +03:00
return err
}
2017-08-31 15:22:47 +03:00
return nil
2016-12-13 18:28:39 +03:00
}
2017-08-31 15:22:47 +03:00
func (c *ClusterClient) checkMovedErr(
cmd Cmder, err error, failedCmds map[*clusterNode][]Cmder,
) bool {
moved, ask, addr := internal.IsMovedError(err)
2016-12-13 18:28:39 +03:00
if moved {
2017-07-09 13:10:07 +03:00
c.lazyReloadState()
2017-07-09 13:10:07 +03:00
node, err := c.nodes.GetOrCreate(addr)
2016-12-13 18:28:39 +03:00
if err != nil {
2017-08-31 15:22:47 +03:00
return false
2016-12-13 18:28:39 +03:00
}
failedCmds[node] = append(failedCmds[node], cmd)
2017-08-31 15:22:47 +03:00
return true
2016-12-13 18:28:39 +03:00
}
2017-08-31 15:22:47 +03:00
2016-12-13 18:28:39 +03:00
if ask {
2017-07-09 13:10:07 +03:00
node, err := c.nodes.GetOrCreate(addr)
2016-12-13 18:28:39 +03:00
if err != nil {
2017-08-31 15:22:47 +03:00
return false
2016-12-13 18:28:39 +03:00
}
failedCmds[node] = append(failedCmds[node], NewCmd("ASKING"), cmd)
2017-08-31 15:22:47 +03:00
return true
2016-12-13 18:28:39 +03:00
}
2017-08-31 15:22:47 +03:00
return false
2016-12-13 18:28:39 +03:00
}
2016-12-16 15:19:53 +03:00
// TxPipeline acts like Pipeline, but wraps queued commands with MULTI/EXEC.
2017-05-02 18:00:53 +03:00
func (c *ClusterClient) TxPipeline() Pipeliner {
2016-12-13 18:28:39 +03:00
pipe := Pipeline{
exec: c.txPipelineExec,
}
2017-05-25 13:38:04 +03:00
pipe.setProcessor(pipe.Process)
2016-12-13 18:28:39 +03:00
return &pipe
}
2017-05-02 18:00:53 +03:00
func (c *ClusterClient) TxPipelined(fn func(Pipeliner) error) ([]Cmder, error) {
return c.TxPipeline().pipelined(fn)
2016-12-13 18:28:39 +03:00
}
func (c *ClusterClient) txPipelineExec(cmds []Cmder) error {
2017-08-31 15:22:47 +03:00
state, err := c.state()
2016-12-13 18:28:39 +03:00
if err != nil {
return err
}
2017-08-31 15:22:47 +03:00
cmdsMap := c.mapCmdsBySlot(cmds)
2016-12-13 18:28:39 +03:00
for slot, cmds := range cmdsMap {
2017-03-04 14:04:27 +03:00
node, err := state.slotMasterNode(slot)
2016-12-13 18:28:39 +03:00
if err != nil {
setCmdsErr(cmds, err)
continue
}
2016-12-13 18:28:39 +03:00
cmdsMap := map[*clusterNode][]Cmder{node: cmds}
2017-08-31 15:22:47 +03:00
for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
if attempt > 0 {
time.Sleep(c.retryBackoff(attempt))
}
2016-12-13 18:28:39 +03:00
failedCmds := make(map[*clusterNode][]Cmder)
for node, cmds := range cmdsMap {
2017-07-09 10:07:20 +03:00
cn, _, err := node.Client.getConn()
2016-12-13 18:28:39 +03:00
if err != nil {
setCmdsErr(cmds, err)
continue
}
err = c.txPipelineProcessCmds(node, cn, cmds, failedCmds)
2017-08-31 15:22:47 +03:00
if err == nil || internal.IsRedisError(err) {
_ = node.Client.connPool.Put(cn)
} else {
_ = node.Client.connPool.Remove(cn)
}
}
2016-12-13 18:28:39 +03:00
if len(failedCmds) == 0 {
break
}
cmdsMap = failedCmds
}
}
2017-08-31 15:22:47 +03:00
return firstCmdsErr(cmds)
2016-12-13 18:28:39 +03:00
}
2017-08-31 15:22:47 +03:00
func (c *ClusterClient) mapCmdsBySlot(cmds []Cmder) map[int][]Cmder {
2016-12-13 18:28:39 +03:00
cmdsMap := make(map[int][]Cmder)
for _, cmd := range cmds {
2017-08-31 15:22:47 +03:00
slot := c.cmdSlot(cmd)
2016-12-13 18:28:39 +03:00
cmdsMap[slot] = append(cmdsMap[slot], cmd)
}
2017-08-31 15:22:47 +03:00
return cmdsMap
2016-12-13 18:28:39 +03:00
}
2016-12-13 18:28:39 +03:00
func (c *ClusterClient) txPipelineProcessCmds(
node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
) error {
cn.SetWriteTimeout(c.opt.WriteTimeout)
if err := txPipelineWriteMulti(cn, cmds); err != nil {
setCmdsErr(cmds, err)
failedCmds[node] = cmds
return err
}
2016-12-13 18:28:39 +03:00
// Set read timeout for all commands.
cn.SetReadTimeout(c.opt.ReadTimeout)
2016-12-13 18:28:39 +03:00
if err := c.txPipelineReadQueued(cn, cmds, failedCmds); err != nil {
2017-08-31 15:22:47 +03:00
setCmdsErr(cmds, err)
2016-12-13 18:28:39 +03:00
return err
}
2017-08-31 15:22:47 +03:00
return pipelineReadCmds(cn, cmds)
2016-12-13 18:28:39 +03:00
}
func (c *ClusterClient) txPipelineReadQueued(
cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
) error {
// Parse queued replies.
var statusCmd StatusCmd
2017-08-31 15:22:47 +03:00
if err := statusCmd.readReply(cn); err != nil {
return err
2016-12-13 18:28:39 +03:00
}
for _, cmd := range cmds {
err := statusCmd.readReply(cn)
if err == nil {
continue
}
2017-08-31 15:22:47 +03:00
if c.checkMovedErr(cmd, err, failedCmds) || internal.IsRedisError(err) {
continue
2016-12-13 18:28:39 +03:00
}
2017-08-31 15:22:47 +03:00
return err
}
2016-12-13 18:28:39 +03:00
// Parse number of replies.
line, err := cn.Rd.ReadLine()
if err != nil {
if err == Nil {
err = TxFailedErr
}
return err
}
switch line[0] {
case proto.ErrorReply:
2017-08-31 15:22:47 +03:00
err := proto.ParseErrorReply(line)
for _, cmd := range cmds {
if !c.checkMovedErr(cmd, err, failedCmds) {
break
}
}
return err
2016-12-13 18:28:39 +03:00
case proto.ArrayReply:
// ok
default:
err := fmt.Errorf("redis: expected '*', but got line %q", line)
return err
}
2017-08-31 15:22:47 +03:00
return nil
}
2017-07-09 10:07:20 +03:00
func (c *ClusterClient) pubSub(channels []string) *PubSub {
opt := c.opt.clientOptions()
var node *clusterNode
return &PubSub{
opt: opt,
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
}
2017-08-31 15:22:47 +03:00
state, err := c.state()
if err != nil {
return nil, err
}
masterNode, err := state.slotMasterNode(slot)
2017-07-09 10:07:20 +03:00
if err != nil {
return nil, err
}
node = masterNode
}
return node.Client.newConn()
},
closeConn: func(cn *pool.Conn) error {
return node.Client.connPool.CloseConn(cn)
},
}
}
// 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)
if len(channels) > 0 {
_ = pubsub.Subscribe(channels...)
}
return pubsub
}
// 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)
if len(channels) > 0 {
_ = pubsub.PSubscribe(channels...)
}
return pubsub
}
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()
}
2017-07-09 13:10:07 +03:00
func appendNode(nodes []*clusterNode, node *clusterNode) []*clusterNode {
for _, n := range nodes {
if n == node {
return nodes
}
}
return append(nodes, node)
}