Release3.2 fix 1345 (#1419)

* redis: don't recycle redis connection if error happens

* redis: close client immediately if error occurs

Author: spinlock

pkg/topom/topom_slots.go

@@ -396,7 +396,7 @@ func (s *Topom) newSlotActionExecutor(sid int) (func(db int) (remains int, nextd
 			if err != nil {
 				return 0, -1, err
 			}
-			defer s.action.redisp.PutClient(c)
+			defer s.action.redisp.PutClient(c, err)
 
 			if err := c.Select(db); err != nil {
 				return 0, -1, err

pkg/topom/topom_stats.go

@@ -79,7 +79,7 @@ func (s *Topom) RefreshRedisStats(timeout time.Duration) (*sync2.Future, error)
 			if err != nil {
 				return nil, err
 			}
-			defer s.ha.redisp.PutClient(c)
+			defer s.ha.redisp.PutClient(c, err)
 			m, err := c.Info()
 			if err != nil {
 				return nil, err

pkg/utils/redis/client.go

@@ -54,6 +54,7 @@ func (c *Client) Close() error {
 func (c *Client) Do(cmd string, args ...interface{}) (interface{}, error) {
 	r, err := c.conn.Do(cmd, args...)
 	if err != nil {
+		c.Close()
 		return nil, errors.Trace(err)
 	}
 	c.LastUse = time.Now()
@@ -64,9 +65,26 @@ func (c *Client) Do(cmd string, args ...interface{}) (interface{}, error) {
 	return r, nil
 }
 
+func (c *Client) Send(cmd string, args ...interface{}) error {
+	if err := c.conn.Send(cmd, args...); err != nil {
+		c.Close()
+		return errors.Trace(err)
+	}
+	return nil
+}
+
+func (c *Client) Flush() error {
+	if err := c.conn.Flush(); err != nil {
+		c.Close()
+		return errors.Trace(err)
+	}
+	return nil
+}
+
 func (c *Client) Receive() (interface{}, error) {
 	r, err := c.conn.Receive()
 	if err != nil {
+		c.Close()
 		return nil, errors.Trace(err)
 	}
 	c.LastUse = time.Now()
@@ -170,11 +188,11 @@ func (c *Client) SetMaster(master string) error {
 	if err != nil {
 		return errors.Trace(err)
 	}
-	c.conn.Send("MULTI")
-	c.conn.Send("CONFIG", "SET", "masterauth", c.Auth)
-	c.conn.Send("SLAVEOF", host, port)
-	c.conn.Send("CONFIG", "REWRITE")
-	c.conn.Send("CLIENT", "KILL", "TYPE", "normal")
+	c.Send("MULTI")
+	c.Send("CONFIG", "SET", "masterauth", c.Auth)
+	c.Send("SLAVEOF", host, port)
+	c.Send("CONFIG", "REWRITE")
+	c.Send("CLIENT", "KILL", "TYPE", "normal")
 	values, err := redigo.Values(c.Do("EXEC"))
 	if err != nil {
 		return errors.Trace(err)
@@ -385,10 +403,10 @@ func (p *Pool) getClientFromCache(addr string) (*Client, error) {
 	return nil, nil
 }
 
-func (p *Pool) PutClient(c *Client) {
+func (p *Pool) PutClient(c *Client, err error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	if p.closed || !p.isRecyclable(c) {
+	if err != nil || p.closed || !p.isRecyclable(c) {
 		c.Close()
 	} else {
 		cache := p.pool[c.Addr]
@@ -400,22 +418,30 @@ func (p *Pool) PutClient(c *Client) {
 	}
 }
 
-func (p *Pool) Info(addr string) (map[string]string, error) {
+func (p *Pool) Info(addr string) (_ map[string]string, err error) {
 	c, err := p.GetClient(addr)
 	if err != nil {
 		return nil, err
 	}
-	defer p.PutClient(c)
-	return c.Info()
+	defer p.PutClient(c, err)
+	m, err := c.Info()
+	if err != nil {
+		return nil, err
+	}
+	return m, nil
 }
 
-func (p *Pool) InfoFull(addr string) (map[string]string, error) {
+func (p *Pool) InfoFull(addr string) (_ map[string]string, err error) {
 	c, err := p.GetClient(addr)
 	if err != nil {
 		return nil, err
 	}
-	defer p.PutClient(c)
-	return c.InfoFull()
+	defer p.PutClient(c, err)
+	m, err := c.InfoFull()
+	if err != nil {
+		return nil, err
+	}
+	return m, nil
 }
 
 type InfoCache struct {

pkg/utils/redis/sentinel.go

@@ -110,8 +110,8 @@ func (s *Sentinel) subscribeCommand(client *Client, sentinel string,
 	onSubscribed func()) error {
 	var channels = []interface{}{"+switch-master"}
 	go func() {
-		client.conn.Send("SUBSCRIBE", channels...)
-		client.conn.Flush()
+		client.Send("SUBSCRIBE", channels...)
+		client.Flush()
 	}()
 	for _, sub := range channels {
 		values, err := redigo.Values(client.Receive())
@@ -221,10 +221,10 @@ func (s *Sentinel) Subscribe(sentinels []string, timeout time.Duration, onMajori
 func (s *Sentinel) existsCommand(client *Client, names []string) (map[string]bool, error) {
 	go func() {
 		for _, name := range names {
-			client.conn.Send("SENTINEL", "get-master-addr-by-name", name)
+			client.Send("SENTINEL", "get-master-addr-by-name", name)
 		}
 		if len(names) != 0 {
-			client.conn.Flush()
+			client.Flush()
 		}
 	}()
 	exists := make(map[string]bool, len(names))
@@ -250,10 +250,10 @@ func (s *Sentinel) slavesCommand(client *Client, names []string) (map[string][]m
 				continue
 			}
 			pending++
-			client.conn.Send("SENTINEL", "slaves", name)
+			client.Send("SENTINEL", "slaves", name)
 		}
 		if pending != 0 {
-			client.conn.Flush()
+			client.Flush()
 		}
 	}()
 	results := make(map[string][]map[string]string, len(names))
@@ -427,10 +427,10 @@ func (s *Sentinel) monitorGroupsCommand(client *Client, sentniel string, config
 	go func() {
 		for gid, tcpAddr := range groups {
 			var ip, port = tcpAddr.IP.String(), tcpAddr.Port
-			client.conn.Send("SENTINEL", "monitor", s.NodeName(gid), ip, port, config.Quorum)
+			client.Send("SENTINEL", "monitor", s.NodeName(gid), ip, port, config.Quorum)
 		}
 		if len(groups) != 0 {
-			client.conn.Flush()
+			client.Flush()
 		}
 	}()
 	for range groups {
@@ -460,10 +460,10 @@ func (s *Sentinel) monitorGroupsCommand(client *Client, sentniel string, config
 			if config.ClientReconfigScript != "" {
 				args = append(args, "client-reconfig-script", config.ClientReconfigScript)
 			}
-			client.conn.Send("SENTINEL", args...)
+			client.Send("SENTINEL", args...)
 		}
 		if len(groups) != 0 {
-			client.conn.Flush()
+			client.Flush()
 		}
 	}()
 	for range groups {
@@ -573,10 +573,10 @@ func (s *Sentinel) removeCommand(client *Client, names []string) error {
 				continue
 			}
 			pending++
-			client.conn.Send("SENTINEL", "remove", name)
+			client.Send("SENTINEL", "remove", name)
 		}
 		if pending != 0 {
-			client.conn.Flush()
+			client.Flush()
 		}
 	}()
 	for _, name := range names {