add go mysql redis sample

Signed-off-by: vishal979 <shantapowertech2@gmail.com>

go-mysql-redis/vendor/gopkg.in/redis.v5/ring.go

@@ -0,0 +1,420 @@
+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
+}