utils: close redis connection immediately if client is not recyclable (#1419,#1345)

Author: spinlock

pkg/utils/redis/client.go

@@ -26,6 +26,10 @@ type Client struct {
 
 	LastUse time.Time
 	Timeout time.Duration
+
+	Pipeline struct {
+		Send, Recv uint64
+	}
 }
 
 func NewClientNoAuth(addr string, timeout time.Duration) (*Client, error) {
@@ -51,6 +55,18 @@ func (c *Client) Close() error {
 	return c.conn.Close()
 }
 
+func (c *Client) isRecyclable() bool {
+	switch {
+	case c.conn.Err() != nil:
+		return false
+	case c.Pipeline.Send != c.Pipeline.Recv:
+		return false
+	case c.Timeout != 0 && c.Timeout <= time.Since(c.LastUse):
+		return false
+	}
+	return true
+}
+
 func (c *Client) Do(cmd string, args ...interface{}) (interface{}, error) {
 	r, err := c.conn.Do(cmd, args...)
 	if err != nil {
@@ -70,6 +86,7 @@ func (c *Client) Send(cmd string, args ...interface{}) error {
 		c.Close()
 		return errors.Trace(err)
 	}
+	c.Pipeline.Send++
 	return nil
 }
 
@@ -87,6 +104,8 @@ func (c *Client) Receive() (interface{}, error) {
 		c.Close()
 		return nil, errors.Trace(err)
 	}
+	c.Pipeline.Recv++
+
 	c.LastUse = time.Now()
 
 	if err, ok := r.(redigo.Error); ok {
@@ -327,18 +346,6 @@ func NewPool(auth string, timeout time.Duration) *Pool {
 	return p
 }
 
-func (p *Pool) isRecyclable(c *Client) bool {
-	switch {
-	case c.conn.Err() != nil:
-		return false
-	case p.timeout == 0:
-		return true
-	case p.timeout >= time.Since(c.LastUse):
-		return true
-	}
-	return false
-}
-
 func (p *Pool) Close() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -368,10 +375,10 @@ func (p *Pool) Cleanup() error {
 	for addr, list := range p.pool {
 		for i := list.Len(); i != 0; i-- {
 			c := list.Remove(list.Front()).(*Client)
-			if p.isRecyclable(c) {
-				list.PushBack(c)
-			} else {
+			if !c.isRecyclable() {
 				c.Close()
+			} else {
+				list.PushBack(c)
 			}
 		}
 		if list.Len() == 0 {
@@ -398,10 +405,10 @@ func (p *Pool) getClientFromCache(addr string) (*Client, error) {
 	if list := p.pool[addr]; list != nil {
 		for i := list.Len(); i != 0; i-- {
 			c := list.Remove(list.Front()).(*Client)
-			if p.isRecyclable(c) {
-				return c, nil
-			} else {
+			if !c.isRecyclable() {
 				c.Close()
+			} else {
+				return c, nil
 			}
 		}
 	}
@@ -411,7 +418,7 @@ func (p *Pool) getClientFromCache(addr string) (*Client, error) {
 func (p *Pool) PutClient(c *Client) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	if p.closed || !p.isRecyclable(c) {
+	if !c.isRecyclable() || p.closed {
 		c.Close()
 	} else {
 		cache := p.pool[c.Addr]

pkg/utils/redis/sentinel.go

@@ -108,6 +108,9 @@ func (s *Sentinel) dispatch(ctx context.Context, sentinel string, timeout time.D
 
 func (s *Sentinel) subscribeCommand(client *Client, sentinel string,
 	onSubscribed func()) error {
+	defer func() {
+		client.Close()
+	}()
 	var channels = []interface{}{"+switch-master"}
 	go func() {
 		client.Send("SUBSCRIBE", channels...)
@@ -219,6 +222,11 @@ func (s *Sentinel) Subscribe(sentinels []string, timeout time.Duration, onMajori
 }
 
 func (s *Sentinel) existsCommand(client *Client, names []string) (map[string]bool, error) {
+	defer func() {
+		if !client.isRecyclable() {
+			client.Close()
+		}
+	}()
 	go func() {
 		for _, name := range names {
 			client.Send("SENTINEL", "get-master-addr-by-name", name)
@@ -239,6 +247,11 @@ func (s *Sentinel) existsCommand(client *Client, names []string) (map[string]boo
 }
 
 func (s *Sentinel) slavesCommand(client *Client, names []string) (map[string][]map[string]string, error) {
+	defer func() {
+		if !client.isRecyclable() {
+			client.Close()
+		}
+	}()
 	exists, err := s.existsCommand(client, names)
 	if err != nil {
 		return nil, err
@@ -279,6 +292,11 @@ func (s *Sentinel) slavesCommand(client *Client, names []string) (map[string][]m
 }
 
 func (s *Sentinel) mastersCommand(client *Client) (map[int]map[string]string, error) {
+	defer func() {
+		if !client.isRecyclable() {
+			client.Close()
+		}
+	}()
 	values, err := redigo.Values(client.Do("SENTINEL", "masters"))
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -417,6 +435,11 @@ type MonitorConfig struct {
 }
 
 func (s *Sentinel) monitorGroupsCommand(client *Client, sentniel string, config *MonitorConfig, groups map[int]*net.TCPAddr) error {
+	defer func() {
+		if !client.isRecyclable() {
+			client.Close()
+		}
+	}()
 	var names []string
 	for gid := range groups {
 		names = append(names, s.NodeName(gid))
@@ -562,6 +585,11 @@ func (s *Sentinel) MonitorGroups(sentinels []string, timeout time.Duration, conf
 }
 
 func (s *Sentinel) removeCommand(client *Client, names []string) error {
+	defer func() {
+		if !client.isRecyclable() {
+			client.Close()
+		}
+	}()
 	exists, err := s.existsCommand(client, names)
 	if err != nil {
 		return err
@@ -708,6 +736,11 @@ type SentinelGroup struct {
 }
 
 func (s *Sentinel) MastersAndSlavesClient(client *Client) (map[string]*SentinelGroup, error) {
+	defer func() {
+		if !client.isRecyclable() {
+			client.Close()
+		}
+	}()
 	masters, err := s.mastersCommand(client)
 	if err != nil {
 		return nil, err