Add support for cluster mode client and pool unit test

Author: Runze Cui

cluster.go

@@ -128,15 +128,6 @@ func (opt *ClusterOptions) init() {
 		opt.MaxRetryBackoff = 512 * time.Millisecond
 	}
 
-	if opt.ConnReqQueueSize <= 0 {
-		// If ConnReqQueueSize is less than or equal to 0, set it to 1,000,000
-		opt.ConnReqQueueSize = 1000000
-	}
-	if opt.ConnReqQueueSize <= 100*opt.PoolSize {
-		// If ConnReqQueueSize is less than or equal to 100 times PoolSize, set it to 100 times PoolSize
-		opt.ConnReqQueueSize = 100 * opt.PoolSize
-	}
-
 	if opt.NewClient == nil {
 		opt.NewClient = NewClient
 	}
@@ -167,6 +158,7 @@ func (opt *ClusterOptions) clientOptions() *Options {
 		PoolTimeout:        opt.PoolTimeout,
 		IdleTimeout:        opt.IdleTimeout,
 		IdleCheckFrequency: disableIdleCheck,
+		ConnReqQueueSize:   opt.ConnReqQueueSize,
 
 		TLSConfig: opt.TLSConfig,
 		// If ClusterSlots is populated, then we probably have an artificial

internal/pool/pool_dynamic.go

@@ -10,6 +10,10 @@ import (
 )
 
 const (
+	// connReqsQueueSize: buffered chan size of connection opener, this value should be larger than the maximum typical
+	// value used for poolSize, otherwise it might block ALL calls in Pool until pending connection request is satisfied
+	connReqsQueueSize = 1000000
+
 	minIdleCheckFrequency = time.Second
 
 	// tryDialFrequency: frequency used to try dial underlying resource to check if it is recovered after a consecutive
@@ -62,10 +66,19 @@ type DynamicConnPool struct {
 var _ DynamicPooler = (*DynamicConnPool)(nil)
 
 func NewDynamicConnPool(opt *Options) *DynamicConnPool {
+	connReqQueueSize := opt.ConnReqQueueSize
+	if connReqQueueSize <= 0 {
+		connReqQueueSize = connReqsQueueSize
+	}
+	// the buffer size should be at least 100x of the pool size to handle spiking
+	if connReqQueueSize <= 100*opt.PoolSize {
+		connReqQueueSize = 100 * opt.PoolSize
+	}
+
 	p := &DynamicConnPool{
 		opt:      opt,
 		closedCh: make(chan struct{}),
-		connReqs: make(chan ctxConnChan, opt.ConnReqQueueSize),
+		connReqs: make(chan ctxConnChan, connReqQueueSize),
 	}
 
 	p.connsMu.Lock()
@@ -198,10 +211,12 @@ func (p *DynamicConnPool) IdleLen() int {
 	return p.idleConnsLen
 }
 
+// QueueLen returns number of connections in the queue
 func (p *DynamicConnPool) QueueLen() int {
 	return len(p.connReqs)
 }
 
+// QueueCap returns the capacity of the queue
 func (p *DynamicConnPool) QueueCap() int {
 	return cap(p.connReqs)
 }

internal/pool/pool_dynamic_test.go

@@ -13,14 +13,15 @@ import (
 	"gitlab.myteksi.net/dbops/Redis/v8/internal/pool"
 )
 
-func assertPoolStats(pooler pool.DynamicPooler, hits, misses, timeouts, totals, idles, stales uint32) {
+func assertPoolStats(pooler pool.DynamicPooler, hits, misses, timeouts, totals, idles, stales, capacity uint32) {
 	Expect(pooler.Stats()).To(Equal(&pool.Stats{
 		Hits:       hits,
 		Misses:     misses,
 		Timeouts:   timeouts,
 		TotalConns: totals,
 		IdleConns:  idles,
 		StaleConns: stales,
+		QueueCap:   capacity,
 	}))
 }
 
@@ -63,21 +64,60 @@ var _ = Describe("ConnPool", func() {
 			PoolTimeout:        time.Hour,
 			IdleTimeout:        time.Millisecond,
 			IdleCheckFrequency: time.Millisecond * 500,
+			ConnReqQueueSize:   1000000,
 		})
 		wg.Wait()
 
 		// MinIdleConns should optimistically increase poolSize though actual conn creation is pending
-		assertPoolStats(connPool, 0, 0, 0, 10, 10, 0)
+		assertPoolStats(connPool, 0, 0, 0, 10, 10, 0, 1000000)
 
 		// no error since no idle conn queued yet during pool close and stats remains
 		Expect(connPool.Close()).NotTo(HaveOccurred())
-		assertPoolStats(connPool, 0, 0, 0, 10, 10, 0)
+		assertPoolStats(connPool, 0, 0, 0, 10, 10, 0, 1000000)
 
 		close(closedChan) // release the channel to make all dail success
 
 		// wait for 5ms and all conn put back idle list should fail and decrease pool size
 		time.Sleep(time.Millisecond * 5)
-		assertPoolStats(connPool, 0, 0, 0, 0, 0, 0)
+		assertPoolStats(connPool, 0, 0, 0, 0, 0, 0, 1000000)
+	})
+
+	It("should create idle connections then safe close and queue size is bigger than 100x of pool size", func() {
+		const minIdleConns = 10
+
+		var (
+			wg         sync.WaitGroup
+			closedChan = make(chan struct{})
+		)
+		wg.Add(minIdleConns)
+		connPool = pool.NewDynamicConnPool(&pool.Options{
+			Dialer: func(ctx context.Context) (net.Conn, error) {
+				wg.Done()
+				<-closedChan
+				return &net.TCPConn{}, nil
+			},
+			PoolSize:           10,
+			MinIdleConns:       minIdleConns,
+			MaxIdleConns:       10,
+			PoolTimeout:        time.Hour,
+			IdleTimeout:        time.Millisecond,
+			IdleCheckFrequency: time.Millisecond * 500,
+			ConnReqQueueSize:   100,
+		})
+		wg.Wait()
+
+		// MinIdleConns should optimistically increase poolSize though actual conn creation is pending
+		assertPoolStats(connPool, 0, 0, 0, 10, 10, 0, 1000)
+
+		// no error since no idle conn queued yet during pool close and stats remains
+		Expect(connPool.Close()).NotTo(HaveOccurred())
+		assertPoolStats(connPool, 0, 0, 0, 10, 10, 0, 1000)
+
+		close(closedChan) // release the channel to make all dail success
+
+		// wait for 5ms and all conn put back idle list should fail and decrease pool size
+		time.Sleep(time.Millisecond * 5)
+		assertPoolStats(connPool, 0, 0, 0, 0, 0, 0, 1000)
 	})
 
 	It("should unblock client when conn is removed", func() {
@@ -642,7 +682,7 @@ var _ = Describe("dynamic update", func() {
 			getConnsExpectNoErr(connPool, minIdleConns-1) // use all connections
 
 			// all get request should hit and use up all idle connections
-			assertPoolStats(connPool, minIdleConns-1, 0, 0, minIdleConns-1, 0, 0)
+			assertPoolStats(connPool, minIdleConns-1, 0, 0, minIdleConns-1, 0, 0, 1000000)
 
 			done := make(chan struct{})
 			go func() {

options.go

@@ -115,7 +115,7 @@ type Options struct {
 	// Buffered chan size of connection opener, this value should be larger than the maximum typical
 	// value used for poolSize, otherwise it might block ALL calls in Pool until pending connection request is satisfied
 	// Default is 1,000,000 and minimum is 100x of the pool size
-	// Expect to see memory usage increase when queue size is increasing
+	// Expect memory usage increase when queue size is increasing, no dynamic reloading allowed
 	ConnReqQueueSize int
 
 	// Enables read only queries on slave nodes.
@@ -202,15 +202,6 @@ func (opt *Options) init() {
 	case 0:
 		opt.MaxRetryBackoff = 512 * time.Millisecond
 	}
-
-	if opt.ConnReqQueueSize <= 0 {
-		// If ConnReqQueueSize is less than or equal to 0, set it to 1,000,000
-		opt.ConnReqQueueSize = 1000000
-	}
-	if opt.ConnReqQueueSize <= 100*opt.PoolSize {
-		// If ConnReqQueueSize is less than or equal to 100 times PoolSize, set it to 100 times PoolSize
-		opt.ConnReqQueueSize = 100 * opt.PoolSize
-	}
 }
 
 func (opt *Options) clone() *Options {

redis_dynamic_test.go

@@ -164,6 +164,10 @@ var limiterError = fmt.Errorf("limiter error")
 type errorLimiter struct {
 }
 
+func (l *errorLimiter) IsCBOpen() bool {
+	return true
+}
+
 func (l *errorLimiter) Execute(f func() error) error {
 	return nil
 }
@@ -181,6 +185,10 @@ type normalLimiter struct {
 	errors []error
 }
 
+func (l *normalLimiter) IsCBOpen() bool {
+	return false
+}
+
 func (l *normalLimiter) Execute(f func() error) error {
 	l.count++
 	return f()