gorush/vendor/gopkg.in/redis.v5/ring.go

421 lines
8.6 KiB
Go
Raw Normal View History

package redis
import (
"errors"
"fmt"
"math/rand"
"strconv"
"sync"
"sync/atomic"
"time"
"gopkg.in/redis.v5/internal"
"gopkg.in/redis.v5/internal/consistenthash"
"gopkg.in/redis.v5/internal/hashtag"
"gopkg.in/redis.v5/internal/pool"
)
var errRingShardsDown = errors.New("redis: all ring shards are down")
// RingOptions are used to configure a ring client and should be
// passed to NewRing.
type RingOptions struct {
// Map of name => host:port addresses of ring shards.
Addrs map[string]string
// Frequency of PING commands sent to check shards availability.
// Shard is considered down after 3 subsequent failed checks.
HeartbeatFrequency time.Duration
// Following options are copied from Options struct.
DB int
Password string
MaxRetries int
DialTimeout time.Duration
ReadTimeout time.Duration
WriteTimeout time.Duration
PoolSize int
PoolTimeout time.Duration
IdleTimeout time.Duration
IdleCheckFrequency time.Duration
}
func (opt *RingOptions) init() {
if opt.HeartbeatFrequency == 0 {
opt.HeartbeatFrequency = 500 * time.Millisecond
}
}
func (opt *RingOptions) clientOptions() *Options {
return &Options{
DB: opt.DB,
Password: opt.Password,
DialTimeout: opt.DialTimeout,
ReadTimeout: opt.ReadTimeout,
WriteTimeout: opt.WriteTimeout,
PoolSize: opt.PoolSize,
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()
}
// Vote votes to set shard state and returns true if state was changed.
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()
}
// Ring is a Redis client that uses constistent hashing to distribute
// keys across multiple Redis servers (shards). It's safe for
// concurrent use by multiple goroutines.
//
// Ring monitors the state of each shard and removes dead shards from
// the ring. When shard comes online it is added back to the ring. This
// gives you maximum availability and partition tolerance, but no
// consistency between different shards or even clients. Each client
// uses shards that are available to the client and does not do any
// coordination when shard state is changed.
//
// Ring should be used when you need multiple Redis servers for caching
// and can tolerate losing data when one of the servers dies.
// Otherwise you should use Redis Cluster.
type Ring struct {
cmdable
opt *RingOptions
nreplicas int
mu sync.RWMutex
hash *consistenthash.Map
shards map[string]*ringShard
cmdsInfoOnce *sync.Once
cmdsInfo map[string]*CommandInfo
closed bool
}
func NewRing(opt *RingOptions) *Ring {
const nreplicas = 100
opt.init()
ring := &Ring{
opt: opt,
nreplicas: nreplicas,
hash: consistenthash.New(nreplicas, nil),
shards: make(map[string]*ringShard),
cmdsInfoOnce: new(sync.Once),
}
ring.cmdable.process = ring.Process
for name, addr := range opt.Addrs {
clopt := opt.clientOptions()
clopt.Addr = addr
ring.addClient(name, NewClient(clopt))
}
go ring.heartbeat()
return ring
}
// PoolStats returns accumulated connection pool stats.
func (c *Ring) PoolStats() *PoolStats {
var acc PoolStats
for _, shard := range c.shards {
s := shard.Client.connPool.Stats()
acc.Requests += s.Requests
acc.Hits += s.Hits
acc.Timeouts += s.Timeouts
acc.TotalConns += s.TotalConns
acc.FreeConns += s.FreeConns
}
return &acc
}
// ForEachShard concurrently calls the fn on each live shard in the ring.
// It returns the first error if any.
func (c *Ring) ForEachShard(fn func(client *Client) error) error {
var wg sync.WaitGroup
errCh := make(chan error, 1)
for _, shard := range c.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) cmdInfo(name string) *CommandInfo {
c.cmdsInfoOnce.Do(func() {
for _, shard := range c.shards {
cmdsInfo, err := shard.Client.Command().Result()
if err == nil {
c.cmdsInfo = cmdsInfo
return
}
}
c.cmdsInfoOnce = &sync.Once{}
})
if c.cmdsInfo == nil {
return nil
}
return c.cmdsInfo[name]
}
func (c *Ring) addClient(name string, cl *Client) {
c.mu.Lock()
c.hash.Add(name)
c.shards[name] = &ringShard{Client: cl}
c.mu.Unlock()
}
func (c *Ring) shardByKey(key string) (*ringShard, error) {
key = hashtag.Key(key)
c.mu.RLock()
if c.closed {
c.mu.RUnlock()
return nil, pool.ErrClosed
}
name := c.hash.Get(key)
if name == "" {
c.mu.RUnlock()
return nil, errRingShardsDown
}
shard := c.shards[name]
c.mu.RUnlock()
return shard, nil
}
func (c *Ring) randomShard() (*ringShard, error) {
return c.shardByKey(strconv.Itoa(rand.Int()))
}
func (c *Ring) shardByName(name string) (*ringShard, error) {
if name == "" {
return c.randomShard()
}
c.mu.RLock()
shard := c.shards[name]
c.mu.RUnlock()
return shard, nil
}
func (c *Ring) cmdShard(cmd Cmder) (*ringShard, error) {
cmdInfo := c.cmdInfo(cmd.name())
firstKey := cmd.arg(cmdFirstKeyPos(cmd, cmdInfo))
return c.shardByKey(firstKey)
}
func (c *Ring) Process(cmd Cmder) error {
shard, err := c.cmdShard(cmd)
if err != nil {
cmd.setErr(err)
return err
}
return shard.Client.Process(cmd)
}
// rebalance removes dead shards from the Ring.
func (c *Ring) rebalance() {
hash := consistenthash.New(c.nreplicas, nil)
for name, shard := range c.shards {
if shard.IsUp() {
hash.Add(name)
}
}
c.mu.Lock()
c.hash = hash
c.mu.Unlock()
}
// heartbeat monitors state of each shard in the ring.
func (c *Ring) heartbeat() {
ticker := time.NewTicker(c.opt.HeartbeatFrequency)
defer ticker.Stop()
for _ = range ticker.C {
var rebalance bool
c.mu.RLock()
if c.closed {
c.mu.RUnlock()
break
}
for _, shard := range c.shards {
err := shard.Client.Ping().Err()
if shard.Vote(err == nil || err == pool.ErrPoolTimeout) {
internal.Logf("ring shard state changed: %s", shard)
rebalance = true
}
}
c.mu.RUnlock()
if rebalance {
c.rebalance()
}
}
}
// Close closes the ring client, releasing any open resources.
//
// It is rare to Close a Ring, as the Ring is meant to be long-lived
// and shared between many goroutines.
func (c *Ring) 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
return firstErr
}
func (c *Ring) Pipeline() *Pipeline {
pipe := Pipeline{
exec: c.pipelineExec,
}
pipe.cmdable.process = pipe.Process
pipe.statefulCmdable.process = pipe.Process
return &pipe
}
func (c *Ring) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
return c.Pipeline().pipelined(fn)
}
func (c *Ring) pipelineExec(cmds []Cmder) (firstErr error) {
cmdsMap := make(map[string][]Cmder)
for _, cmd := range cmds {
cmdInfo := c.cmdInfo(cmd.name())
name := cmd.arg(cmdFirstKeyPos(cmd, cmdInfo))
if name != "" {
name = c.hash.Get(hashtag.Key(name))
}
cmdsMap[name] = append(cmdsMap[name], cmd)
}
for i := 0; i <= c.opt.MaxRetries; i++ {
var failedCmdsMap map[string][]Cmder
for name, cmds := range cmdsMap {
shard, err := c.shardByName(name)
if err != nil {
setCmdsErr(cmds, err)
if firstErr == nil {
firstErr = err
}
continue
}
cn, _, err := shard.Client.conn()
if err != nil {
setCmdsErr(cmds, err)
if firstErr == nil {
firstErr = err
}
continue
}
canRetry, err := shard.Client.pipelineProcessCmds(cn, cmds)
shard.Client.putConn(cn, err, false)
if err == nil {
continue
}
if firstErr == nil {
firstErr = err
}
if canRetry && internal.IsRetryableError(err) {
if failedCmdsMap == nil {
failedCmdsMap = make(map[string][]Cmder)
}
failedCmdsMap[name] = cmds
}
}
if len(failedCmdsMap) == 0 {
break
}
cmdsMap = failedCmdsMap
}
return firstErr
}