2015-01-24 15:12:48 +03:00
|
|
|
package redis
|
|
|
|
|
|
|
|
import (
|
|
|
|
"math/rand"
|
|
|
|
"strings"
|
|
|
|
"sync"
|
|
|
|
"sync/atomic"
|
|
|
|
"time"
|
|
|
|
)
|
|
|
|
|
|
|
|
type ClusterClient struct {
|
|
|
|
commandable
|
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
addrs []string
|
2015-04-08 12:28:21 +03:00
|
|
|
slots [][]string
|
2015-05-01 09:33:47 +03:00
|
|
|
slotsMx sync.RWMutex // Protects slots and addrs.
|
2015-03-30 17:10:53 +03:00
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
clients map[string]*Client
|
2015-05-01 09:33:47 +03:00
|
|
|
closed bool
|
|
|
|
clientsMx sync.RWMutex // Protects clients and closed.
|
2015-03-30 17:10:53 +03:00
|
|
|
|
|
|
|
opt *ClusterOptions
|
2015-01-24 15:12:48 +03:00
|
|
|
|
|
|
|
_reload uint32
|
|
|
|
}
|
|
|
|
|
|
|
|
// NewClusterClient initializes a new cluster-aware client using given options.
|
|
|
|
// A list of seed addresses must be provided.
|
2015-04-04 16:46:57 +03:00
|
|
|
func NewClusterClient(opt *ClusterOptions) *ClusterClient {
|
2015-01-24 15:12:48 +03:00
|
|
|
client := &ClusterClient{
|
2015-04-08 12:28:50 +03:00
|
|
|
addrs: opt.Addrs,
|
2015-04-13 16:33:44 +03:00
|
|
|
slots: make([][]string, hashSlots),
|
2015-04-04 16:46:57 +03:00
|
|
|
clients: make(map[string]*Client),
|
2015-01-24 15:12:48 +03:00
|
|
|
opt: opt,
|
|
|
|
_reload: 1,
|
|
|
|
}
|
|
|
|
client.commandable.process = client.process
|
2015-04-13 16:33:44 +03:00
|
|
|
client.reloadIfDue()
|
2015-05-01 11:01:01 +03:00
|
|
|
go client.reaper()
|
2015-04-04 16:46:57 +03:00
|
|
|
return client
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
|
|
|
|
2015-05-01 09:33:47 +03:00
|
|
|
// Close closes the cluster client, releasing any open resources.
|
|
|
|
//
|
|
|
|
// It is rare to Close a Client, as the Client is meant to be
|
|
|
|
// long-lived and shared between many goroutines.
|
2015-01-24 15:12:48 +03:00
|
|
|
func (c *ClusterClient) Close() error {
|
2015-05-01 11:01:01 +03:00
|
|
|
defer c.clientsMx.Unlock()
|
2015-05-01 09:33:47 +03:00
|
|
|
c.clientsMx.Lock()
|
|
|
|
|
|
|
|
if c.closed {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
c.closed = true
|
2015-04-28 17:00:10 +03:00
|
|
|
c.resetClients()
|
2015-05-01 09:33:47 +03:00
|
|
|
c.setSlots(nil)
|
2015-04-04 16:46:57 +03:00
|
|
|
return nil
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
// getClient returns a Client for a given address.
|
2015-03-18 13:41:24 +03:00
|
|
|
func (c *ClusterClient) getClient(addr string) (*Client, error) {
|
|
|
|
if addr == "" {
|
|
|
|
return c.randomClient()
|
|
|
|
}
|
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
c.clientsMx.RLock()
|
|
|
|
client, ok := c.clients[addr]
|
|
|
|
if ok {
|
|
|
|
c.clientsMx.RUnlock()
|
2015-03-18 13:41:24 +03:00
|
|
|
return client, nil
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
2015-04-04 16:46:57 +03:00
|
|
|
c.clientsMx.RUnlock()
|
2015-01-24 15:12:48 +03:00
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
c.clientsMx.Lock()
|
2015-05-01 09:33:47 +03:00
|
|
|
if c.closed {
|
2015-05-01 11:01:01 +03:00
|
|
|
c.clientsMx.Unlock()
|
2015-05-01 09:33:47 +03:00
|
|
|
return nil, errClosed
|
|
|
|
}
|
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
client, ok = c.clients[addr]
|
2015-01-24 15:12:48 +03:00
|
|
|
if !ok {
|
|
|
|
opt := c.opt.clientOptions()
|
|
|
|
opt.Addr = addr
|
2015-05-02 16:19:22 +03:00
|
|
|
client = NewClient(opt)
|
2015-04-04 16:46:57 +03:00
|
|
|
c.clients[addr] = client
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
2015-04-04 16:46:57 +03:00
|
|
|
c.clientsMx.Unlock()
|
|
|
|
|
2015-03-18 13:41:24 +03:00
|
|
|
return client, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (c *ClusterClient) slotAddrs(slot int) []string {
|
|
|
|
c.slotsMx.RLock()
|
|
|
|
addrs := c.slots[slot]
|
|
|
|
c.slotsMx.RUnlock()
|
|
|
|
return addrs
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
|
|
|
|
2015-05-01 09:33:47 +03:00
|
|
|
// randomClient returns a Client for the first pingable node.
|
2015-04-04 16:46:57 +03:00
|
|
|
func (c *ClusterClient) randomClient() (client *Client, err error) {
|
|
|
|
for i := 0; i < 10; i++ {
|
|
|
|
n := rand.Intn(len(c.addrs))
|
2015-03-18 13:41:24 +03:00
|
|
|
client, err = c.getClient(c.addrs[n])
|
|
|
|
if err != nil {
|
|
|
|
continue
|
|
|
|
}
|
2015-04-04 16:46:57 +03:00
|
|
|
err = client.Ping().Err()
|
|
|
|
if err == nil {
|
|
|
|
return client, nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2015-01-24 15:12:48 +03:00
|
|
|
func (c *ClusterClient) process(cmd Cmder) {
|
|
|
|
var ask bool
|
|
|
|
|
|
|
|
c.reloadIfDue()
|
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
slot := hashSlot(cmd.clusterKey())
|
2015-01-24 15:12:48 +03:00
|
|
|
|
2015-03-18 13:41:24 +03:00
|
|
|
var addr string
|
|
|
|
if addrs := c.slotAddrs(slot); len(addrs) > 0 {
|
|
|
|
addr = addrs[0] // First address is master.
|
|
|
|
}
|
|
|
|
|
|
|
|
client, err := c.getClient(addr)
|
|
|
|
if err != nil {
|
|
|
|
cmd.setErr(err)
|
|
|
|
return
|
2015-04-04 16:46:57 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
for attempt := 0; attempt <= c.opt.getMaxRedirects(); attempt++ {
|
2015-01-24 15:12:48 +03:00
|
|
|
if ask {
|
2015-04-04 16:46:57 +03:00
|
|
|
pipe := client.Pipeline()
|
2015-01-24 15:12:48 +03:00
|
|
|
pipe.Process(NewCmd("ASKING"))
|
|
|
|
pipe.Process(cmd)
|
|
|
|
_, _ = pipe.Exec()
|
|
|
|
ask = false
|
|
|
|
} else {
|
2015-04-04 16:46:57 +03:00
|
|
|
client.Process(cmd)
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// If there is no (real) error, we are done!
|
|
|
|
err := cmd.Err()
|
2015-04-04 16:46:57 +03:00
|
|
|
if err == nil || err == Nil || err == TxFailedErr {
|
2015-01-24 15:12:48 +03:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2015-04-07 12:30:06 +03:00
|
|
|
// On network errors try random node.
|
2015-04-04 16:46:57 +03:00
|
|
|
if isNetworkError(err) {
|
2015-04-07 12:30:06 +03:00
|
|
|
client, err = c.randomClient()
|
|
|
|
if err != nil {
|
|
|
|
return
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
|
|
|
cmd.reset()
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2015-03-18 13:41:24 +03:00
|
|
|
var moved bool
|
|
|
|
var addr string
|
|
|
|
moved, ask, addr = isMovedError(err)
|
|
|
|
if moved || ask {
|
|
|
|
if moved {
|
|
|
|
c.scheduleReload()
|
|
|
|
}
|
|
|
|
client, err = c.getClient(addr)
|
|
|
|
if err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
cmd.reset()
|
|
|
|
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
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
// Closes all clients and returns last error if there are any.
|
|
|
|
func (c *ClusterClient) resetClients() (err error) {
|
|
|
|
for addr, client := range c.clients {
|
|
|
|
if e := client.Close(); e != nil {
|
|
|
|
err = e
|
|
|
|
}
|
|
|
|
delete(c.clients, addr)
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
2015-04-04 16:46:57 +03:00
|
|
|
return err
|
|
|
|
}
|
2015-01-24 15:12:48 +03:00
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
func (c *ClusterClient) setSlots(slots []ClusterSlotInfo) {
|
2015-03-30 17:53:28 +03:00
|
|
|
c.slotsMx.Lock()
|
2015-01-24 15:12:48 +03:00
|
|
|
|
2015-04-07 12:30:06 +03:00
|
|
|
seen := make(map[string]struct{})
|
2015-04-08 12:28:21 +03:00
|
|
|
for _, addr := range c.addrs {
|
|
|
|
seen[addr] = struct{}{}
|
|
|
|
}
|
|
|
|
|
2015-04-13 16:33:44 +03:00
|
|
|
for i := 0; i < hashSlots; i++ {
|
|
|
|
c.slots[i] = c.slots[i][:0]
|
|
|
|
}
|
2015-04-04 16:46:57 +03:00
|
|
|
for _, info := range slots {
|
2015-04-07 12:30:06 +03:00
|
|
|
for slot := info.Start; slot <= info.End; slot++ {
|
2015-04-08 12:28:21 +03:00
|
|
|
c.slots[slot] = info.Addrs
|
2015-04-07 12:30:06 +03:00
|
|
|
}
|
|
|
|
|
2015-04-08 12:28:21 +03:00
|
|
|
for _, addr := range info.Addrs {
|
|
|
|
if _, ok := seen[addr]; !ok {
|
|
|
|
c.addrs = append(c.addrs, addr)
|
|
|
|
seen[addr] = struct{}{}
|
|
|
|
}
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
c.slotsMx.Unlock()
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
// Closes all connections and reloads slot cache, if due.
|
|
|
|
func (c *ClusterClient) reloadIfDue() (err error) {
|
|
|
|
if !atomic.CompareAndSwapUint32(&c._reload, 1, 0) {
|
|
|
|
return
|
|
|
|
}
|
2015-01-24 15:12:48 +03:00
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
client, err := c.randomClient()
|
|
|
|
if err != nil {
|
|
|
|
return err
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
slots, err := client.ClusterSlots().Result()
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
c.setSlots(slots)
|
2015-01-24 15:12:48 +03:00
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
return nil
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
|
|
|
|
2015-04-04 16:46:57 +03:00
|
|
|
// Schedules slots reload on next request.
|
|
|
|
func (c *ClusterClient) scheduleReload() {
|
|
|
|
atomic.StoreUint32(&c._reload, 1)
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
|
|
|
|
2015-04-05 15:41:16 +03:00
|
|
|
// reaper closes idle connections to the cluster.
|
2015-05-01 11:01:01 +03:00
|
|
|
func (c *ClusterClient) reaper() {
|
2015-05-01 13:24:24 +03:00
|
|
|
ticker := time.NewTicker(time.Minute)
|
2015-05-01 11:01:01 +03:00
|
|
|
defer ticker.Stop()
|
2015-04-05 15:41:16 +03:00
|
|
|
for _ = range ticker.C {
|
2015-05-01 09:33:47 +03:00
|
|
|
c.clientsMx.RLock()
|
|
|
|
|
|
|
|
if c.closed {
|
|
|
|
c.clientsMx.RUnlock()
|
2015-05-01 11:01:01 +03:00
|
|
|
break
|
2015-05-01 09:33:47 +03:00
|
|
|
}
|
|
|
|
|
2015-04-08 12:28:21 +03:00
|
|
|
for _, client := range c.clients {
|
2015-04-05 15:41:16 +03:00
|
|
|
pool := client.connPool
|
2015-05-01 09:33:47 +03:00
|
|
|
// pool.First removes idle connections from the pool and
|
|
|
|
// returns first non-idle connection. So just put returned
|
|
|
|
// connection back.
|
2015-04-05 15:41:16 +03:00
|
|
|
if cn := pool.First(); cn != nil {
|
|
|
|
pool.Put(cn)
|
|
|
|
}
|
|
|
|
}
|
2015-05-01 09:33:47 +03:00
|
|
|
|
|
|
|
c.clientsMx.RUnlock()
|
2015-04-05 15:41:16 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-01-24 15:12:48 +03:00
|
|
|
//------------------------------------------------------------------------------
|
|
|
|
|
|
|
|
type ClusterOptions struct {
|
|
|
|
// A seed-list of host:port addresses of known cluster nodes
|
|
|
|
Addrs []string
|
|
|
|
|
|
|
|
// An optional password
|
|
|
|
Password string
|
|
|
|
|
|
|
|
// The maximum number of MOVED/ASK redirects to follow, before
|
|
|
|
// giving up. Default: 16
|
|
|
|
MaxRedirects int
|
|
|
|
|
2015-04-17 14:44:56 +03:00
|
|
|
// Following options are copied from `redis.Options`.
|
|
|
|
PoolSize int
|
|
|
|
DialTimeout, ReadTimeout, WriteTimeout, PoolTimeout, IdleTimeout time.Duration
|
2015-01-24 15:12:48 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
func (opt *ClusterOptions) getMaxRedirects() int {
|
2015-04-17 14:44:56 +03:00
|
|
|
if opt.MaxRedirects == 0 {
|
2015-01-24 15:12:48 +03:00
|
|
|
return 16
|
|
|
|
}
|
|
|
|
return opt.MaxRedirects
|
|
|
|
}
|
|
|
|
|
|
|
|
func (opt *ClusterOptions) clientOptions() *Options {
|
|
|
|
return &Options{
|
|
|
|
DB: 0,
|
|
|
|
Password: opt.Password,
|
|
|
|
|
2015-04-17 14:44:56 +03:00
|
|
|
DialTimeout: opt.DialTimeout,
|
2015-01-24 15:12:48 +03:00
|
|
|
ReadTimeout: opt.ReadTimeout,
|
|
|
|
WriteTimeout: opt.WriteTimeout,
|
|
|
|
|
2015-04-17 14:44:56 +03:00
|
|
|
PoolSize: opt.PoolSize,
|
|
|
|
PoolTimeout: opt.PoolTimeout,
|
2015-01-24 15:12:48 +03:00
|
|
|
IdleTimeout: opt.IdleTimeout,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
//------------------------------------------------------------------------------
|
|
|
|
|
|
|
|
const hashSlots = 16384
|
|
|
|
|
2015-03-23 11:23:33 +03:00
|
|
|
// hashSlot returns a consistent slot number between 0 and 16383
|
|
|
|
// for any given string key.
|
|
|
|
func hashSlot(key string) int {
|
2015-01-24 15:12:48 +03:00
|
|
|
if s := strings.IndexByte(key, '{'); s > -1 {
|
|
|
|
if e := strings.IndexByte(key[s+1:], '}'); e > 0 {
|
|
|
|
key = key[s+1 : s+e+1]
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if key == "" {
|
|
|
|
return rand.Intn(hashSlots)
|
|
|
|
}
|
|
|
|
return int(crc16sum(key)) % hashSlots
|
|
|
|
}
|