Merge remote-tracking branch 'eolinker/master' into feature/Feature-fix_alert

# Conflicts:
#	.gitignore
#	app/console/main.go
#	common/redis-manager/factory.go
#	common/redis-manager/proxy.go
#	common/redis/sentinel_ring2.go

.gitignore

@@ -2,4 +2,4 @@
 /vendor/
 /out/
 /vendor
-/.vscode/
\ No newline at end of file
+/.vscode/

app/console/main.go

@@ -14,11 +14,12 @@ import (
 var (
 	UserPassword string
 	UserName     string
-	ConfFilepath = "./config/goku.conf"
+	ConfFilePath = "./config/goku.conf"
+
 )
 
 func main() {
-	flag.StringVar(&ConfFilepath, "c", "./config/goku.conf", "Please provide a valid configuration file path")
+	flag.StringVar(&ConfFilePath, "c", "./config/goku.conf", "Please provide a valid configuration file path")
 	flag.StringVar(&UserName, "u", "", "Please provide user name")
 	flag.StringVar(&UserPassword, "p", "", "Please provide user password")
 	isDebug := flag.Bool("debug", false, "")
@@ -28,7 +29,7 @@ func main() {
 		log.StartDebug()
 	}
 	// 初始化配置
-	if err := conf.ReadConfigure(ConfFilepath); err != nil {
+	if err := conf.ReadConfigure(ConfFilePath); err != nil {
 		log.Panic(err)
 		return
 	}

common/redis-manager/factory.go

@@ -2,7 +2,7 @@ package redis_manager
 
 import (
 	"fmt"
-	"github.com/eolinker/goku/common/redis"
+	"github.com/go-redis/redis"
 
 	"sort"
 )
@@ -29,21 +29,7 @@ func Create(config RedisConfig) Redis {
 				config:  config,
 			}
 		}
-	case RedisModeSentinel:
-		{
 
-			option := redis.SentinelRingOptions{
-				Addrs:    config.GetAddrs(),
-				Masters:  config.GetMasters(),
-				Password: config.GetPassword(),
-				DB:       config.GetDbIndex(),
-				PoolSize: _PoolSize,
-			}
-			return &redisProxy{
-				Cmdable: redis.NewSentinelRing(&option),
-				config:  config,
-			}
-		}
 	case RedisModeStand:
 		{
 

common/redis-manager/proxy.go

 package redis_manager
 
 import (
-	"github.com/eolinker/goku/common/redis"
+	"github.com/go-redis/redis"
 )
 
 type redisProxy struct {
@@ -25,18 +25,7 @@ func (p *redisProxy) Nodes() []string {
 			}(ch)
 
 		}
-	case RedisModeSentinel:
-		{
 
-			conn := p.Cmdable.(*redis.SentinelRing)
-			go func(ch chan string) {
-				conn.ForEachAddr(func(addr string) error {
-					ch <- addr
-					return nil
-				})
-				close(ch)
-			}(ch)
-		}
 	case RedisModeStand:
 		{
 
@@ -65,12 +54,7 @@ func (p *redisProxy) Foreach(fn func(client *redis.Client) error) error {
 			conn := p.Cmdable.(*redis.ClusterClient)
 			return conn.ForEachMaster(fn)
 		}
-	case RedisModeSentinel:
-		{
 
-			conn := p.Cmdable.(*redis.SentinelRing)
-			return conn.ForEachShard(fn)
-		}
 	case RedisModeStand:
 		{
 

common/redis-manager/redis.go

 package redis_manager
 
-import "github.com/eolinker/goku/common/redis"
+import "github.com/go-redis/redis"
 
 const (
 	RedisModeCluster  = "cluster"
-	RedisModeSentinel = "sentinel"
 	RedisModeStand    = "stand"
 )
 

common/redis/interface.go

-package redis
-
-type RedisClient interface {
-	Cmdable
-}

common/redis/internal/consistenthash/consistenthash.go

-/*
-Copyright 2013 Google Inc.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-     http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-// Package consistenthash provides an implementation of a ring hash.
-package consistenthash
-
-import (
-	"hash/crc32"
-	"sort"
-	"strconv"
-)
-
-type Hash func(data []byte) uint32
-
-type Map struct {
-	hash     Hash
-	replicas int
-	keys     []int // Sorted
-	hashMap  map[int]string
-}
-
-func New(replicas int, fn Hash) *Map {
-	m := &Map{
-		replicas: replicas,
-		hash:     fn,
-		hashMap:  make(map[int]string),
-	}
-	if m.hash == nil {
-		m.hash = crc32.ChecksumIEEE
-	}
-	return m
-}
-
-// Returns true if there are no items available.
-func (m *Map) IsEmpty() bool {
-	return len(m.keys) == 0
-}
-
-// Adds some keys to the hash.
-func (m *Map) Add(keys ...string) {
-	for _, key := range keys {
-		for i := 0; i < m.replicas; i++ {
-			hash := int(m.hash([]byte(strconv.Itoa(i) + key)))
-			m.keys = append(m.keys, hash)
-			m.hashMap[hash] = key
-		}
-	}
-	sort.Ints(m.keys)
-}
-
-// Gets the closest item in the hash to the provided key.
-func (m *Map) Get(key string) string {
-	if m.IsEmpty() {
-		return ""
-	}
-
-	hash := int(m.hash([]byte(key)))
-
-	// Binary search for appropriate replica.
-	idx := sort.Search(len(m.keys), func(i int) bool { return m.keys[i] >= hash })
-
-	// Means we have cycled back to the first replica.
-	if idx == len(m.keys) {
-		idx = 0
-	}
-
-	return m.hashMap[m.keys[idx]]
-}

common/redis/internal/error.go

-package internal
-
-import (
-	"io"
-	"net"
-	"strings"
-
-	"github.com/eolinker/goku/common/redis/internal/proto"
-)
-
-func IsRetryableError(err error, retryTimeout bool) bool {
-	if err == nil {
-		return false
-	}
-	if err == io.EOF {
-		return true
-	}
-	if netErr, ok := err.(net.Error); ok {
-		if netErr.Timeout() {
-			return retryTimeout
-		}
-		return true
-	}
-	s := err.Error()
-	if s == "ERR max number of clients reached" {
-		return true
-	}
-	if strings.HasPrefix(s, "LOADING ") {
-		return true
-	}
-	if strings.HasPrefix(s, "READONLY ") {
-		return true
-	}
-	if strings.HasPrefix(s, "CLUSTERDOWN ") {
-		return true
-	}
-	return false
-}
-
-func IsRedisError(err error) bool {
-	_, ok := err.(proto.RedisError)
-	return ok
-}
-
-func IsBadConn(err error, allowTimeout bool) bool {
-	if err == nil {
-		return false
-	}
-	if IsRedisError(err) {
-		// #790
-		return IsReadOnlyError(err)
-	}
-	if allowTimeout {
-		if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
-			return false
-		}
-	}
-	return true
-}
-
-func IsMovedError(err error) (moved bool, ask bool, addr string) {
-	if !IsRedisError(err) {
-		return
-	}
-
-	s := err.Error()
-	if strings.HasPrefix(s, "MOVED ") {
-		moved = true
-	} else if strings.HasPrefix(s, "ASK ") {
-		ask = true
-	} else {
-		return
-	}
-
-	ind := strings.LastIndex(s, " ")
-	if ind == -1 {
-		return false, false, ""
-	}
-	addr = s[ind+1:]
-	return
-}
-
-func IsLoadingError(err error) bool {
-	return strings.HasPrefix(err.Error(), "LOADING ")
-}
-
-func IsReadOnlyError(err error) bool {
-	return strings.HasPrefix(err.Error(), "READONLY ")
-}

common/redis/internal/hashtag/hashtag.go

-package hashtag
-
-import (
-	"math/rand"
-	"strings"
-)
-
-const slotNumber = 16384
-
-// CRC16 implementation according to CCITT standards.
-// Copyright 2001-2010 Georges Menie (www.menie.org)
-// Copyright 2013 The Go Authors. All rights reserved.
-// http://redis.io/topics/cluster-spec#appendix-a-crc16-reference-implementation-in-ansi-c
-var crc16tab = [256]uint16{
-	0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50a5, 0x60c6, 0x70e7,
-	0x8108, 0x9129, 0xa14a, 0xb16b, 0xc18c, 0xd1ad, 0xe1ce, 0xf1ef,
-	0x1231, 0x0210, 0x3273, 0x2252, 0x52b5, 0x4294, 0x72f7, 0x62d6,
-	0x9339, 0x8318, 0xb37b, 0xa35a, 0xd3bd, 0xc39c, 0xf3ff, 0xe3de,
-	0x2462, 0x3443, 0x0420, 0x1401, 0x64e6, 0x74c7, 0x44a4, 0x5485,
-	0xa56a, 0xb54b, 0x8528, 0x9509, 0xe5ee, 0xf5cf, 0xc5ac, 0xd58d,
-	0x3653, 0x2672, 0x1611, 0x0630, 0x76d7, 0x66f6, 0x5695, 0x46b4,
-	0xb75b, 0xa77a, 0x9719, 0x8738, 0xf7df, 0xe7fe, 0xd79d, 0xc7bc,
-	0x48c4, 0x58e5, 0x6886, 0x78a7, 0x0840, 0x1861, 0x2802, 0x3823,
-	0xc9cc, 0xd9ed, 0xe98e, 0xf9af, 0x8948, 0x9969, 0xa90a, 0xb92b,
-	0x5af5, 0x4ad4, 0x7ab7, 0x6a96, 0x1a71, 0x0a50, 0x3a33, 0x2a12,
-	0xdbfd, 0xcbdc, 0xfbbf, 0xeb9e, 0x9b79, 0x8b58, 0xbb3b, 0xab1a,
-	0x6ca6, 0x7c87, 0x4ce4, 0x5cc5, 0x2c22, 0x3c03, 0x0c60, 0x1c41,
-	0xedae, 0xfd8f, 0xcdec, 0xddcd, 0xad2a, 0xbd0b, 0x8d68, 0x9d49,
-	0x7e97, 0x6eb6, 0x5ed5, 0x4ef4, 0x3e13, 0x2e32, 0x1e51, 0x0e70,
-	0xff9f, 0xefbe, 0xdfdd, 0xcffc, 0xbf1b, 0xaf3a, 0x9f59, 0x8f78,
-	0x9188, 0x81a9, 0xb1ca, 0xa1eb, 0xd10c, 0xc12d, 0xf14e, 0xe16f,
-	0x1080, 0x00a1, 0x30c2, 0x20e3, 0x5004, 0x4025, 0x7046, 0x6067,
-	0x83b9, 0x9398, 0xa3fb, 0xb3da, 0xc33d, 0xd31c, 0xe37f, 0xf35e,
-	0x02b1, 0x1290, 0x22f3, 0x32d2, 0x4235, 0x5214, 0x6277, 0x7256,
-	0xb5ea, 0xa5cb, 0x95a8, 0x8589, 0xf56e, 0xe54f, 0xd52c, 0xc50d,
-	0x34e2, 0x24c3, 0x14a0, 0x0481, 0x7466, 0x6447, 0x5424, 0x4405,
-	0xa7db, 0xb7fa, 0x8799, 0x97b8, 0xe75f, 0xf77e, 0xc71d, 0xd73c,
-	0x26d3, 0x36f2, 0x0691, 0x16b0, 0x6657, 0x7676, 0x4615, 0x5634,
-	0xd94c, 0xc96d, 0xf90e, 0xe92f, 0x99c8, 0x89e9, 0xb98a, 0xa9ab,
-	0x5844, 0x4865, 0x7806, 0x6827, 0x18c0, 0x08e1, 0x3882, 0x28a3,
-	0xcb7d, 0xdb5c, 0xeb3f, 0xfb1e, 0x8bf9, 0x9bd8, 0xabbb, 0xbb9a,
-	0x4a75, 0x5a54, 0x6a37, 0x7a16, 0x0af1, 0x1ad0, 0x2ab3, 0x3a92,
-	0xfd2e, 0xed0f, 0xdd6c, 0xcd4d, 0xbdaa, 0xad8b, 0x9de8, 0x8dc9,
-	0x7c26, 0x6c07, 0x5c64, 0x4c45, 0x3ca2, 0x2c83, 0x1ce0, 0x0cc1,
-	0xef1f, 0xff3e, 0xcf5d, 0xdf7c, 0xaf9b, 0xbfba, 0x8fd9, 0x9ff8,
-	0x6e17, 0x7e36, 0x4e55, 0x5e74, 0x2e93, 0x3eb2, 0x0ed1, 0x1ef0,
-}
-
-func Key(key string) string {
-	if s := strings.IndexByte(key, '{'); s > -1 {
-		if e := strings.IndexByte(key[s+1:], '}'); e > 0 {
-			return key[s+1 : s+e+1]
-		}
-	}
-	return key
-}
-
-func RandomSlot() int {
-	return rand.Intn(slotNumber)
-}
-
-// hashSlot returns a consistent slot number between 0 and 16383
-// for any given string key.
-func Slot(key string) int {
-	if key == "" {
-		return RandomSlot()
-	}
-	key = Key(key)
-	return int(crc16sum(key)) % slotNumber
-}
-
-func crc16sum(key string) (crc uint16) {
-	for i := 0; i < len(key); i++ {
-		crc = (crc << 8) ^ crc16tab[(byte(crc>>8)^key[i])&0x00ff]
-	}
-	return
-}

common/redis/internal/hashtag/hashtag_test.go

-package hashtag
-
-import (
-	"math/rand"
-	"testing"
-)
-
-func TestGinkgoSuite(t *testing.T) {
-	RegisterFailHandler(Fail)
-	RunSpecs(t, "hashtag")
-}
-
-var _ = Describe("CRC16", func() {
-
-	// http://redis.io/topics/cluster-spec#keys-distribution-model
-	It("should calculate CRC16", func() {
-		tests := []struct {
-			s string
-			n uint16
-		}{
-			{"123456789", 0x31C3},
-			{string([]byte{83, 153, 134, 118, 229, 214, 244, 75, 140, 37, 215, 215}), 21847},
-		}
-
-		for _, test := range tests {
-			Expect(crc16sum(test.s)).To(Equal(test.n), "for %s", test.s)
-		}
-	})
-
-})
-
-var _ = Describe("HashSlot", func() {
-
-	It("should calculate hash slots", func() {
-		tests := []struct {
-			key  string
-			slot int
-		}{
-			{"123456789", 12739},
-			{"{}foo", 9500},
-			{"foo{}", 5542},
-			{"foo{}{bar}", 8363},
-			{"", 10503},
-			{"", 5176},
-			{string([]byte{83, 153, 134, 118, 229, 214, 244, 75, 140, 37, 215, 215}), 5463},
-		}
-		// Empty keys receive random slot.
-		rand.Seed(100)
-
-		for _, test := range tests {
-			Expect(Slot(test.key)).To(Equal(test.slot), "for %s", test.key)
-		}
-	})
-
-	It("should extract keys from tags", func() {
-		tests := []struct {
-			one, two string
-		}{
-			{"foo{bar}", "bar"},
-			{"{foo}bar", "foo"},
-			{"{user1000}.following", "{user1000}.followers"},
-			{"foo{{bar}}zap", "{bar"},
-			{"foo{bar}{zap}", "bar"},
-		}
-
-		for _, test := range tests {
-			Expect(Slot(test.one)).To(Equal(Slot(test.two)), "for %s <-> %s", test.one, test.two)
-		}
-	})
-
-})

common/redis/internal/internal.go

-package internal
-
-import (
-	"math/rand"
-	"time"
-)
-
-// Retry backoff with jitter sleep to prevent overloaded conditions during intervals
-// https://www.awsarchitectureblog.com/2015/03/backoff.html
-func RetryBackoff(retry int, minBackoff, maxBackoff time.Duration) time.Duration {
-	if retry < 0 {
-		retry = 0
-	}
-
-	backoff := minBackoff << uint(retry)
-	if backoff > maxBackoff || backoff < minBackoff {
-		backoff = maxBackoff
-	}
-
-	if backoff == 0 {
-		return 0
-	}
-	return time.Duration(rand.Int63n(int64(backoff)))
-}

common/redis/internal/internal_test.go

-package internal
-
-import (
-	"testing"
-	"time"
-)
-
-func TestRetryBackoff(t *testing.T) {
-	RegisterTestingT(t)
-
-	for i := -1; i <= 16; i++ {
-		backoff := RetryBackoff(i, time.Millisecond, 512*time.Millisecond)
-		Expect(backoff >= 0).To(BeTrue())
-		Expect(backoff <= 512*time.Millisecond).To(BeTrue())
-	}
-}

common/redis/internal/log.go

-package internal
-
-import (
-	"fmt"
-	"log"
-)
-
-var Logger *log.Logger
-
-func Logf(s string, args ...interface{}) {
-	if Logger == nil {
-		return
-	}
-	Logger.Output(2, fmt.Sprintf(s, args...))
-}

common/redis/internal/once.go

-/*
-Copyright 2014 The Camlistore Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-     http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package internal
-
-import (
-	"sync"
-	"sync/atomic"
-)
-
-// A Once will perform a successful action exactly once.
-//
-// Unlike a sync.Once, this Once's func returns an error
-// and is re-armed on failure.
-type Once struct {
-	m    sync.Mutex
-	done uint32
-}
-
-// Do calls the function f if and only if Do has not been invoked
-// without error for this instance of Once.  In other words, given
-// 	var once Once
-// if once.Do(f) is called multiple times, only the first call will
-// invoke f, even if f has a different value in each invocation unless
-// f returns an error.  A new instance of Once is required for each
-// function to execute.
-//
-// Do is intended for initialization that must be run exactly once.  Since f
-// is niladic, it may be necessary to use a function literal to capture the
-// arguments to a function to be invoked by Do:
-// 	err := config.once.Do(func() error { return config.init(filename) })
-func (o *Once) Do(f func() error) error {
-	if atomic.LoadUint32(&o.done) == 1 {
-		return nil
-	}
-	// Slow-path.
-	o.m.Lock()
-	defer o.m.Unlock()
-	var err error
-	if o.done == 0 {
-		err = f()
-		if err == nil {
-			atomic.StoreUint32(&o.done, 1)
-		}
-	}
-	return err
-}

common/redis/internal/pool/bench_test.go

-package pool_test
-
-import (
-	"fmt"
-	"testing"
-	"time"
-
-	"github.com/go-redis/redis/internal/pool"
-)
-
-type poolGetPutBenchmark struct {
-	poolSize int
-}
-
-func (bm poolGetPutBenchmark) String() string {
-	return fmt.Sprintf("pool=%d", bm.poolSize)
-}
-
-func BenchmarkPoolGetPut(b *testing.B) {
-	benchmarks := []poolGetPutBenchmark{
-		{1},
-		{2},
-		{8},
-		{32},
-		{64},
-		{128},
-	}
-	for _, bm := range benchmarks {
-		b.Run(bm.String(), func(b *testing.B) {
-			connPool := pool.NewConnPool(&pool.Options{
-				Dialer:             dummyDialer,
-				PoolSize:           bm.poolSize,
-				PoolTimeout:        time.Second,
-				IdleTimeout:        time.Hour,
-				IdleCheckFrequency: time.Hour,
-			})
-
-			b.ResetTimer()
-
-			b.RunParallel(func(pb *testing.PB) {
-				for pb.Next() {
-					cn, err := connPool.Get()
-					if err != nil {
-						b.Fatal(err)
-					}
-					connPool.Put(cn)
-				}
-			})
-		})
-	}
-}
-
-type poolGetRemoveBenchmark struct {
-	poolSize int
-}
-
-func (bm poolGetRemoveBenchmark) String() string {
-	return fmt.Sprintf("pool=%d", bm.poolSize)
-}
-
-func BenchmarkPoolGetRemove(b *testing.B) {
-	benchmarks := []poolGetRemoveBenchmark{
-		{1},
-		{2},
-		{8},
-		{32},
-		{64},
-		{128},
-	}
-	for _, bm := range benchmarks {
-		b.Run(bm.String(), func(b *testing.B) {
-			connPool := pool.NewConnPool(&pool.Options{
-				Dialer:             dummyDialer,
-				PoolSize:           bm.poolSize,
-				PoolTimeout:        time.Second,
-				IdleTimeout:        time.Hour,
-				IdleCheckFrequency: time.Hour,
-			})
-
-			b.ResetTimer()
-
-			b.RunParallel(func(pb *testing.PB) {
-				for pb.Next() {
-					cn, err := connPool.Get()
-					if err != nil {
-						b.Fatal(err)
-					}
-					connPool.Remove(cn)
-				}
-			})
-		})
-	}
-}

common/redis/internal/pool/conn.go

-package pool
-
-import (
-	"net"
-	"sync/atomic"
-	"time"
-
-	"github.com/eolinker/goku/common/redis/internal/proto"
-)
-
-var noDeadline = time.Time{}
-
-type Conn struct {
-	netConn net.Conn
-
-	rd       *proto.Reader
-	rdLocked bool
-	wr       *proto.Writer
-
-	Inited    bool
-	pooled    bool
-	createdAt time.Time
-	usedAt    atomic.Value
-}
-
-func NewConn(netConn net.Conn) *Conn {
-	cn := &Conn{
-		netConn:   netConn,
-		createdAt: time.Now(),
-	}
-	cn.rd = proto.NewReader(netConn)
-	cn.wr = proto.NewWriter(netConn)
-	cn.SetUsedAt(time.Now())
-	return cn
-}
-
-func (cn *Conn) UsedAt() time.Time {
-	return cn.usedAt.Load().(time.Time)
-}
-
-func (cn *Conn) SetUsedAt(tm time.Time) {
-	cn.usedAt.Store(tm)
-}
-
-func (cn *Conn) SetNetConn(netConn net.Conn) {
-	cn.netConn = netConn
-	cn.rd.Reset(netConn)
-	cn.wr.Reset(netConn)
-}
-
-func (cn *Conn) setReadTimeout(timeout time.Duration) error {
-	now := time.Now()
-	cn.SetUsedAt(now)
-	if timeout > 0 {
-		return cn.netConn.SetReadDeadline(now.Add(timeout))
-	}
-	return cn.netConn.SetReadDeadline(noDeadline)
-}
-
-func (cn *Conn) setWriteTimeout(timeout time.Duration) error {
-	now := time.Now()
-	cn.SetUsedAt(now)
-	if timeout > 0 {
-		return cn.netConn.SetWriteDeadline(now.Add(timeout))
-	}
-	return cn.netConn.SetWriteDeadline(noDeadline)
-}
-
-func (cn *Conn) Write(b []byte) (int, error) {
-	return cn.netConn.Write(b)
-}
-
-func (cn *Conn) RemoteAddr() net.Addr {
-	return cn.netConn.RemoteAddr()
-}
-
-func (cn *Conn) WithReader(timeout time.Duration, fn func(rd *proto.Reader) error) error {
-	_ = cn.setReadTimeout(timeout)
-	return fn(cn.rd)
-}
-
-func (cn *Conn) WithWriter(timeout time.Duration, fn func(wr *proto.Writer) error) error {
-	_ = cn.setWriteTimeout(timeout)
-
-	firstErr := fn(cn.wr)
-	err := cn.wr.Flush()
-	if err != nil && firstErr == nil {
-		firstErr = err
-	}
-	return firstErr
-}
-
-func (cn *Conn) Close() error {
-	return cn.netConn.Close()
-}

common/redis/internal/pool/export_test.go

-package pool
-
-import "time"
-
-func (cn *Conn) SetCreatedAt(tm time.Time) {
-	cn.createdAt = tm
-}

common/redis/internal/pool/main_test.go

-package pool_test
-
-import (
-	"net"
-	"sync"
-	"testing"
-)
-
-func TestGinkgoSuite(t *testing.T) {
-	RegisterFailHandler(Fail)
-	RunSpecs(t, "pool")
-}
-
-func perform(n int, cbs ...func(int)) {
-	var wg sync.WaitGroup
-	for _, cb := range cbs {
-		for i := 0; i < n; i++ {
-			wg.Add(1)
-			go func(cb func(int), i int) {
-				defer GinkgoRecover()
-				defer wg.Done()
-
-				cb(i)
-			}(cb, i)
-		}
-	}
-	wg.Wait()
-}
-
-func dummyDialer() (net.Conn, error) {
-	return &net.TCPConn{}, nil
-}

common/redis/internal/pool/pool.go

-package pool
-
-import (
-	"errors"
-	"net"
-	"sync"
-	"sync/atomic"
-	"time"
-
-	"github.com/eolinker/goku/common/redis/internal"
-)
-
-var ErrClosed = errors.New("redis: client is closed")
-var ErrPoolTimeout = errors.New("redis: connection pool timeout")
-
-var timers = sync.Pool{
-	New: func() interface{} {
-		t := time.NewTimer(time.Hour)
-		t.Stop()
-		return t
-	},
-}
-
-// Stats contains pool state information and accumulated stats.
-type Stats struct {
-	Hits     uint32 // number of times free connection was found in the pool
-	Misses   uint32 // number of times free connection was NOT found in the pool
-	Timeouts uint32 // number of times a wait timeout occurred
-
-	TotalConns uint32 // number of total connections in the pool
-	IdleConns  uint32 // number of idle connections in the pool
-	StaleConns uint32 // number of stale connections removed from the pool
-}
-
-type Pooler interface {
-	NewConn() (*Conn, error)
-	CloseConn(*Conn) error
-
-	Get() (*Conn, error)
-	Put(*Conn)
-	Remove(*Conn)
-
-	Len() int
-	IdleLen() int
-	Stats() *Stats
-
-	Close() error
-}
-
-type Options struct {
-	Dialer  func() (net.Conn, error)
-	OnClose func(*Conn) error
-
-	PoolSize           int
-	MinIdleConns       int
-	MaxConnAge         time.Duration
-	PoolTimeout        time.Duration
-	IdleTimeout        time.Duration
-	IdleCheckFrequency time.Duration
-}
-
-type ConnPool struct {
-	opt *Options
-
-	dialErrorsNum uint32 // atomic
-
-	lastDialErrorMu sync.RWMutex
-	lastDialError   error
-
-	queue chan struct{}
-
-	connsMu      sync.Mutex
-	conns        []*Conn
-	idleConns    []*Conn
-	poolSize     int
-	idleConnsLen int
-
-	stats Stats
-
-	_closed uint32 // atomic
-}
-
-var _ Pooler = (*ConnPool)(nil)
-
-func NewConnPool(opt *Options) *ConnPool {
-	p := &ConnPool{
-		opt: opt,
-
-		queue:     make(chan struct{}, opt.PoolSize),
-		conns:     make([]*Conn, 0, opt.PoolSize),
-		idleConns: make([]*Conn, 0, opt.PoolSize),
-	}
-
-	for i := 0; i < opt.MinIdleConns; i++ {
-		p.checkMinIdleConns()
-	}
-
-	if opt.IdleTimeout > 0 && opt.IdleCheckFrequency > 0 {
-		go p.reaper(opt.IdleCheckFrequency)
-	}
-
-	return p
-}
-
-func (p *ConnPool) checkMinIdleConns() {
-	if p.opt.MinIdleConns == 0 {
-		return
-	}
-	if p.poolSize < p.opt.PoolSize && p.idleConnsLen < p.opt.MinIdleConns {
-		p.poolSize++
-		p.idleConnsLen++
-		go p.addIdleConn()
-	}
-}
-
-func (p *ConnPool) addIdleConn() {
-	cn, err := p.newConn(true)
-	if err != nil {
-		return
-	}
-
-	p.connsMu.Lock()
-	p.conns = append(p.conns, cn)
-	p.idleConns = append(p.idleConns, cn)
-	p.connsMu.Unlock()
-}
-
-func (p *ConnPool) NewConn() (*Conn, error) {
-	return p._NewConn(false)
-}
-
-func (p *ConnPool) _NewConn(pooled bool) (*Conn, error) {
-	cn, err := p.newConn(pooled)
-	if err != nil {
-		return nil, err
-	}
-
-	p.connsMu.Lock()
-	p.conns = append(p.conns, cn)
-	if pooled {
-		if p.poolSize < p.opt.PoolSize {
-			p.poolSize++
-		} else {
-			cn.pooled = false
-		}
-	}
-	p.connsMu.Unlock()
-	return cn, nil
-}
-
-func (p *ConnPool) newConn(pooled bool) (*Conn, error) {
-	if p.closed() {
-		return nil, ErrClosed
-	}
-
-	if atomic.LoadUint32(&p.dialErrorsNum) >= uint32(p.opt.PoolSize) {
-		return nil, p.getLastDialError()
-	}
-
-	netConn, err := p.opt.Dialer()
-	if err != nil {
-		p.setLastDialError(err)
-		if atomic.AddUint32(&p.dialErrorsNum, 1) == uint32(p.opt.PoolSize) {
-			go p.tryDial()
-		}
-		return nil, err
-	}
-
-	cn := NewConn(netConn)
-	cn.pooled = pooled
-	return cn, nil
-}
-
-func (p *ConnPool) tryDial() {
-	for {
-		if p.closed() {
-			return
-		}
-
-		conn, err := p.opt.Dialer()
-		if err != nil {
-			p.setLastDialError(err)
-			time.Sleep(time.Second)
-			continue
-		}
-
-		atomic.StoreUint32(&p.dialErrorsNum, 0)
-		_ = conn.Close()
-		return
-	}
-}
-
-func (p *ConnPool) setLastDialError(err error) {
-	p.lastDialErrorMu.Lock()
-	p.lastDialError = err
-	p.lastDialErrorMu.Unlock()
-}
-
-func (p *ConnPool) getLastDialError() error {
-	p.lastDialErrorMu.RLock()
-	err := p.lastDialError
-	p.lastDialErrorMu.RUnlock()
-	return err
-}
-
-// Get returns existed connection from the pool or creates a new one.
-func (p *ConnPool) Get() (*Conn, error) {
-	if p.closed() {
-		return nil, ErrClosed
-	}
-
-	err := p.waitTurn()
-	if err != nil {
-		return nil, err
-	}
-
-	for {
-		p.connsMu.Lock()
-		cn := p.popIdle()
-		p.connsMu.Unlock()
-
-		if cn == nil {
-			break
-		}
-
-		if p.isStaleConn(cn) {
-			_ = p.CloseConn(cn)
-			continue
-		}
-
-		atomic.AddUint32(&p.stats.Hits, 1)
-		return cn, nil
-	}
-
-	atomic.AddUint32(&p.stats.Misses, 1)
-
-	newcn, err := p._NewConn(true)
-	if err != nil {
-		p.freeTurn()
-		return nil, err
-	}
-
-	return newcn, nil
-}
-
-func (p *ConnPool) getTurn() {
-	p.queue <- struct{}{}
-}
-
-func (p *ConnPool) waitTurn() error {
-	select {
-	case p.queue <- struct{}{}:
-		return nil
-	default:
-		timer := timers.Get().(*time.Timer)
-		timer.Reset(p.opt.PoolTimeout)
-
-		select {
-		case p.queue <- struct{}{}:
-			if !timer.Stop() {
-				<-timer.C
-			}
-			timers.Put(timer)
-			return nil
-		case <-timer.C:
-			timers.Put(timer)
-			atomic.AddUint32(&p.stats.Timeouts, 1)
-			return ErrPoolTimeout
-		}
-	}
-}
-
-func (p *ConnPool) freeTurn() {
-	<-p.queue
-}
-
-func (p *ConnPool) popIdle() *Conn {
-	if len(p.idleConns) == 0 {
-		return nil
-	}
-
-	idx := len(p.idleConns) - 1
-	cn := p.idleConns[idx]
-	p.idleConns = p.idleConns[:idx]
-	p.idleConnsLen--
-	p.checkMinIdleConns()
-	return cn
-}
-
-func (p *ConnPool) Put(cn *Conn) {
-	if !cn.pooled {
-		p.Remove(cn)
-		return
-	}
-
-	p.connsMu.Lock()
-	p.idleConns = append(p.idleConns, cn)
-	p.idleConnsLen++
-	p.connsMu.Unlock()
-	p.freeTurn()
-}
-
-func (p *ConnPool) Remove(cn *Conn) {
-	p.removeConn(cn)
-	p.freeTurn()
-	_ = p.closeConn(cn)
-}
-
-func (p *ConnPool) CloseConn(cn *Conn) error {
-	p.removeConn(cn)
-	return p.closeConn(cn)
-}
-
-func (p *ConnPool) removeConn(cn *Conn) {
-	p.connsMu.Lock()
-	for i, c := range p.conns {
-		if c == cn {
-			p.conns = append(p.conns[:i], p.conns[i+1:]...)
-			if cn.pooled {
-				p.poolSize--
-				p.checkMinIdleConns()
-			}
-			break
-		}
-	}
-	p.connsMu.Unlock()
-}
-
-func (p *ConnPool) closeConn(cn *Conn) error {
-	if p.opt.OnClose != nil {
-		_ = p.opt.OnClose(cn)
-	}
-	return cn.Close()
-}
-
-// Len returns total number of connections.
-func (p *ConnPool) Len() int {
-	p.connsMu.Lock()
-	n := len(p.conns)
-	p.connsMu.Unlock()
-	return n
-}
-
-// IdleLen returns number of idle connections.
-func (p *ConnPool) IdleLen() int {
-	p.connsMu.Lock()
-	n := p.idleConnsLen
-	p.connsMu.Unlock()
-	return n
-}
-
-func (p *ConnPool) Stats() *Stats {
-	idleLen := p.IdleLen()
-	return &Stats{
-		Hits:     atomic.LoadUint32(&p.stats.Hits),
-		Misses:   atomic.LoadUint32(&p.stats.Misses),
-		Timeouts: atomic.LoadUint32(&p.stats.Timeouts),
-
-		TotalConns: uint32(p.Len()),
-		IdleConns:  uint32(idleLen),
-		StaleConns: atomic.LoadUint32(&p.stats.StaleConns),
-	}
-}
-
-func (p *ConnPool) closed() bool {
-	return atomic.LoadUint32(&p._closed) == 1
-}
-
-func (p *ConnPool) Filter(fn func(*Conn) bool) error {
-	var firstErr error
-	p.connsMu.Lock()
-	for _, cn := range p.conns {
-		if fn(cn) {
-			if err := p.closeConn(cn); err != nil && firstErr == nil {
-				firstErr = err
-			}
-		}
-	}
-	p.connsMu.Unlock()
-	return firstErr
-}
-
-func (p *ConnPool) Close() error {
-	if !atomic.CompareAndSwapUint32(&p._closed, 0, 1) {
-		return ErrClosed
-	}
-
-	var firstErr error
-	p.connsMu.Lock()
-	for _, cn := range p.conns {
-		if err := p.closeConn(cn); err != nil && firstErr == nil {
-			firstErr = err
-		}
-	}
-	p.conns = nil
-	p.poolSize = 0
-	p.idleConns = nil
-	p.idleConnsLen = 0
-	p.connsMu.Unlock()
-
-	return firstErr
-}
-
-func (p *ConnPool) reapStaleConn() *Conn {
-	if len(p.idleConns) == 0 {
-		return nil
-	}
-
-	cn := p.idleConns[0]
-	if !p.isStaleConn(cn) {
-		return nil
-	}
-
-	p.idleConns = append(p.idleConns[:0], p.idleConns[1:]...)
-	p.idleConnsLen--
-
-	return cn
-}
-
-func (p *ConnPool) ReapStaleConns() (int, error) {
-	var n int
-	for {
-		p.getTurn()
-
-		p.connsMu.Lock()
-		cn := p.reapStaleConn()
-		p.connsMu.Unlock()
-
-		if cn != nil {
-			p.removeConn(cn)
-		}
-
-		p.freeTurn()
-
-		if cn != nil {
-			p.closeConn(cn)
-			n++
-		} else {
-			break
-		}
-	}
-	return n, nil
-}
-
-func (p *ConnPool) reaper(frequency time.Duration) {
-	ticker := time.NewTicker(frequency)
-	defer ticker.Stop()
-
-	for range ticker.C {
-		if p.closed() {
-			break
-		}
-		n, err := p.ReapStaleConns()
-		if err != nil {
-			internal.Logf("ReapStaleConns failed: %s", err)
-			continue
-		}
-		atomic.AddUint32(&p.stats.StaleConns, uint32(n))
-	}
-}
-
-func (p *ConnPool) isStaleConn(cn *Conn) bool {
-	if p.opt.IdleTimeout == 0 && p.opt.MaxConnAge == 0 {
-		return false
-	}
-
-	now := time.Now()
-	if p.opt.IdleTimeout > 0 && now.Sub(cn.UsedAt()) >= p.opt.IdleTimeout {
-		return true
-	}
-	if p.opt.MaxConnAge > 0 && now.Sub(cn.createdAt) >= p.opt.MaxConnAge {
-		return true
-	}
-
-	return false
-}

common/redis/internal/pool/pool_single.go

-package pool
-
-type SingleConnPool struct {
-	cn *Conn
-}
-
-var _ Pooler = (*SingleConnPool)(nil)
-
-func NewSingleConnPool(cn *Conn) *SingleConnPool {
-	return &SingleConnPool{
-		cn: cn,
-	}
-}
-
-func (p *SingleConnPool) NewConn() (*Conn, error) {
-	panic("not implemented")
-}
-
-func (p *SingleConnPool) CloseConn(*Conn) error {
-	panic("not implemented")
-}
-
-func (p *SingleConnPool) Get() (*Conn, error) {
-	return p.cn, nil
-}
-
-func (p *SingleConnPool) Put(cn *Conn) {
-	if p.cn != cn {
-		panic("p.cn != cn")
-	}
-}
-
-func (p *SingleConnPool) Remove(cn *Conn) {
-	if p.cn != cn {
-		panic("p.cn != cn")
-	}
-}
-
-func (p *SingleConnPool) Len() int {
-	return 1
-}
-
-func (p *SingleConnPool) IdleLen() int {
-	return 0
-}
-
-func (p *SingleConnPool) Stats() *Stats {
-	return nil
-}
-
-func (p *SingleConnPool) Close() error {
-	return nil
-}

common/redis/internal/pool/pool_sticky.go

-package pool
-
-import "sync"
-
-type StickyConnPool struct {
-	pool     *ConnPool
-	reusable bool
-
-	cn     *Conn
-	closed bool
-	mu     sync.Mutex
-}
-
-var _ Pooler = (*StickyConnPool)(nil)
-
-func NewStickyConnPool(pool *ConnPool, reusable bool) *StickyConnPool {
-	return &StickyConnPool{
-		pool:     pool,
-		reusable: reusable,
-	}
-}
-
-func (p *StickyConnPool) NewConn() (*Conn, error) {
-	panic("not implemented")
-}
-
-func (p *StickyConnPool) CloseConn(*Conn) error {
-	panic("not implemented")
-}
-
-func (p *StickyConnPool) Get() (*Conn, error) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
-	if p.closed {
-		return nil, ErrClosed
-	}
-	if p.cn != nil {
-		return p.cn, nil
-	}
-
-	cn, err := p.pool.Get()
-	if err != nil {
-		return nil, err
-	}
-
-	p.cn = cn
-	return cn, nil
-}
-
-func (p *StickyConnPool) putUpstream() {
-	p.pool.Put(p.cn)
-	p.cn = nil
-}
-
-func (p *StickyConnPool) Put(cn *Conn) {}
-
-func (p *StickyConnPool) removeUpstream() {
-	p.pool.Remove(p.cn)
-	p.cn = nil
-}
-
-func (p *StickyConnPool) Remove(cn *Conn) {
-	p.removeUpstream()
-}
-
-func (p *StickyConnPool) Len() int {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
-	if p.cn == nil {
-		return 0
-	}
-	return 1
-}
-
-func (p *StickyConnPool) IdleLen() int {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
-	if p.cn == nil {
-		return 1
-	}
-	return 0
-}
-
-func (p *StickyConnPool) Stats() *Stats {
-	return nil
-}
-
-func (p *StickyConnPool) Close() error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
-	if p.closed {
-		return ErrClosed
-	}
-	p.closed = true
-
-	if p.cn != nil {
-		if p.reusable {
-			p.putUpstream()
-		} else {
-			p.removeUpstream()
-		}
-	}
-
-	return nil
-}