Updating dependencies.

vendor/github.com/go-redis/redis/.travis.yml

@@ -9,6 +9,7 @@ go:
   - 1.8.x
   - 1.9.x
   - 1.10.x
+  - 1.11.x
   - tip
 
 matrix:

vendor/github.com/go-redis/redis/CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## Unreleased
+
+- Cluster and Ring pipelines process commands for each node in its own goroutine.
+
 ## 6.14
 
 - Added Options.MinIdleConns.

vendor/github.com/go-redis/redis/cluster.go

@@ -82,7 +82,7 @@ func (opt *ClusterOptions) init() {
 		opt.MaxRedirects = 8
 	}
 
-	if opt.RouteByLatency || opt.RouteRandomly {
+	if (opt.RouteByLatency || opt.RouteRandomly) && opt.ClusterSlots == nil {
 		opt.ReadOnly = true
 	}
 
@@ -831,7 +831,7 @@ func (c *ClusterClient) cmdSlotAndNode(cmd Cmder) (int, *clusterNode, error) {
 	cmdInfo := c.cmdInfo(cmd.Name())
 	slot := cmdSlot(cmd, cmdFirstKeyPos(cmd, cmdInfo))
 
-	if cmdInfo != nil && cmdInfo.ReadOnly && c.opt.ReadOnly {
+	if c.opt.ReadOnly && cmdInfo != nil && cmdInfo.ReadOnly {
 		if c.opt.RouteByLatency {
 			node, err := state.slotClosestNode(slot)
 			return slot, node, err
@@ -1241,7 +1241,8 @@ func (c *ClusterClient) WrapProcessPipeline(
 }
 
 func (c *ClusterClient) defaultProcessPipeline(cmds []Cmder) error {
-	cmdsMap, err := c.mapCmdsByNode(cmds)
+	cmdsMap := newCmdsMap()
+	err := c.mapCmdsByNode(cmds, cmdsMap)
 	if err != nil {
 		setCmdsErr(cmds, err)
 		return err
@@ -1252,28 +1253,31 @@ func (c *ClusterClient) defaultProcessPipeline(cmds []Cmder) error {
 			time.Sleep(c.retryBackoff(attempt))
 		}
 
-		failedCmds := make(map[*clusterNode][]Cmder)
+		failedCmds := newCmdsMap()
+		var wg sync.WaitGroup
 
-		for node, cmds := range cmdsMap {
-			cn, err := node.Client.getConn()
-			if err != nil {
-				if err == pool.ErrClosed {
-					c.remapCmds(cmds, failedCmds)
-				} else {
-					setCmdsErr(cmds, err)
+		for node, cmds := range cmdsMap.m {
+			wg.Add(1)
+			go func(node *clusterNode, cmds []Cmder) {
+				defer wg.Done()
+
+				cn, err := node.Client.getConn()
+				if err != nil {
+					if err == pool.ErrClosed {
+						c.mapCmdsByNode(cmds, failedCmds)
+					} else {
+						setCmdsErr(cmds, err)
+					}
+					return
 				}
-				continue
-			}
 
-			err = c.pipelineProcessCmds(node, cn, cmds, failedCmds)
-			if err == nil || internal.IsRedisError(err) {
-				node.Client.connPool.Put(cn)
-			} else {
-				node.Client.connPool.Remove(cn)
-			}
+				err = c.pipelineProcessCmds(node, cn, cmds, failedCmds)
+				node.Client.releaseConnStrict(cn, err)
+			}(node, cmds)
 		}
 
-		if len(failedCmds) == 0 {
+		wg.Wait()
+		if len(failedCmds.m) == 0 {
 			break
 		}
 		cmdsMap = failedCmds
@@ -1282,14 +1286,24 @@ func (c *ClusterClient) defaultProcessPipeline(cmds []Cmder) error {
 	return cmdsFirstErr(cmds)
 }
 
-func (c *ClusterClient) mapCmdsByNode(cmds []Cmder) (map[*clusterNode][]Cmder, error) {
+type cmdsMap struct {
+	mu sync.Mutex
+	m  map[*clusterNode][]Cmder
+}
+
+func newCmdsMap() *cmdsMap {
+	return &cmdsMap{
+		m: make(map[*clusterNode][]Cmder),
+	}
+}
+
+func (c *ClusterClient) mapCmdsByNode(cmds []Cmder, cmdsMap *cmdsMap) error {
 	state, err := c.state.Get()
 	if err != nil {
 		setCmdsErr(cmds, err)
-		return nil, err
+		return err
 	}
 
-	cmdsMap := make(map[*clusterNode][]Cmder)
 	cmdsAreReadOnly := c.cmdsAreReadOnly(cmds)
 	for _, cmd := range cmds {
 		var node *clusterNode
@@ -1301,11 +1315,13 @@ func (c *ClusterClient) mapCmdsByNode(cmds []Cmder) (map[*clusterNode][]Cmder, e
 			node, err = state.slotMasterNode(slot)
 		}
 		if err != nil {
-			return nil, err
+			return err
 		}
-		cmdsMap[node] = append(cmdsMap[node], cmd)
+		cmdsMap.mu.Lock()
+		cmdsMap.m[node] = append(cmdsMap.m[node], cmd)
+		cmdsMap.mu.Unlock()
 	}
-	return cmdsMap, nil
+	return nil
 }
 
 func (c *ClusterClient) cmdsAreReadOnly(cmds []Cmder) bool {
@@ -1318,27 +1334,17 @@ func (c *ClusterClient) cmdsAreReadOnly(cmds []Cmder) bool {
 	return true
 }
 
-func (c *ClusterClient) remapCmds(cmds []Cmder, failedCmds map[*clusterNode][]Cmder) {
-	remappedCmds, err := c.mapCmdsByNode(cmds)
-	if err != nil {
-		setCmdsErr(cmds, err)
-		return
-	}
-
-	for node, cmds := range remappedCmds {
-		failedCmds[node] = cmds
-	}
-}
-
 func (c *ClusterClient) pipelineProcessCmds(
-	node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
+	node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds *cmdsMap,
 ) error {
 	err := cn.WithWriter(c.opt.WriteTimeout, func(wr *proto.Writer) error {
 		return writeCmd(wr, cmds...)
 	})
 	if err != nil {
 		setCmdsErr(cmds, err)
-		failedCmds[node] = cmds
+		failedCmds.mu.Lock()
+		failedCmds.m[node] = cmds
+		failedCmds.mu.Unlock()
 		return err
 	}
 
@@ -1349,7 +1355,7 @@ func (c *ClusterClient) pipelineProcessCmds(
 }
 
 func (c *ClusterClient) pipelineReadCmds(
-	rd *proto.Reader, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
+	rd *proto.Reader, cmds []Cmder, failedCmds *cmdsMap,
 ) error {
 	for _, cmd := range cmds {
 		err := cmd.readReply(rd)
@@ -1371,7 +1377,7 @@ func (c *ClusterClient) pipelineReadCmds(
 }
 
 func (c *ClusterClient) checkMovedErr(
-	cmd Cmder, err error, failedCmds map[*clusterNode][]Cmder,
+	cmd Cmder, err error, failedCmds *cmdsMap,
 ) bool {
 	moved, ask, addr := internal.IsMovedError(err)
 
@@ -1383,7 +1389,9 @@ func (c *ClusterClient) checkMovedErr(
 			return false
 		}
 
-		failedCmds[node] = append(failedCmds[node], cmd)
+		failedCmds.mu.Lock()
+		failedCmds.m[node] = append(failedCmds.m[node], cmd)
+		failedCmds.mu.Unlock()
 		return true
 	}
 
@@ -1393,7 +1401,9 @@ func (c *ClusterClient) checkMovedErr(
 			return false
 		}
 
-		failedCmds[node] = append(failedCmds[node], NewCmd("ASKING"), cmd)
+		failedCmds.mu.Lock()
+		failedCmds.m[node] = append(failedCmds.m[node], NewCmd("ASKING"), cmd)
+		failedCmds.mu.Unlock()
 		return true
 	}
 
@@ -1433,31 +1443,34 @@ func (c *ClusterClient) defaultProcessTxPipeline(cmds []Cmder) error {
 				time.Sleep(c.retryBackoff(attempt))
 			}
 
-			failedCmds := make(map[*clusterNode][]Cmder)
+			failedCmds := newCmdsMap()
+			var wg sync.WaitGroup
 
 			for node, cmds := range cmdsMap {
-				cn, err := node.Client.getConn()
-				if err != nil {
-					if err == pool.ErrClosed {
-						c.remapCmds(cmds, failedCmds)
-					} else {
-						setCmdsErr(cmds, err)
-					}
-					continue
-				}
+				wg.Add(1)
+				go func(node *clusterNode, cmds []Cmder) {
+					defer wg.Done()
 
-				err = c.txPipelineProcessCmds(node, cn, cmds, failedCmds)
-				if err == nil || internal.IsRedisError(err) {
-					node.Client.connPool.Put(cn)
-				} else {
-					node.Client.connPool.Remove(cn)
-				}
+					cn, err := node.Client.getConn()
+					if err != nil {
+						if err == pool.ErrClosed {
+							c.mapCmdsByNode(cmds, failedCmds)
+						} else {
+							setCmdsErr(cmds, err)
+						}
+						return
+					}
+
+					err = c.txPipelineProcessCmds(node, cn, cmds, failedCmds)
+					node.Client.releaseConnStrict(cn, err)
+				}(node, cmds)
 			}
 
-			if len(failedCmds) == 0 {
+			wg.Wait()
+			if len(failedCmds.m) == 0 {
 				break
 			}
-			cmdsMap = failedCmds
+			cmdsMap = failedCmds.m
 		}
 	}
 
@@ -1474,14 +1487,16 @@ func (c *ClusterClient) mapCmdsBySlot(cmds []Cmder) map[int][]Cmder {
 }
 
 func (c *ClusterClient) txPipelineProcessCmds(
-	node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
+	node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds *cmdsMap,
 ) error {
 	err := cn.WithWriter(c.opt.WriteTimeout, func(wr *proto.Writer) error {
 		return txPipelineWriteMulti(wr, cmds)
 	})
 	if err != nil {
 		setCmdsErr(cmds, err)
-		failedCmds[node] = cmds
+		failedCmds.mu.Lock()
+		failedCmds.m[node] = cmds
+		failedCmds.mu.Unlock()
 		return err
 	}
 
@@ -1497,7 +1512,7 @@ func (c *ClusterClient) txPipelineProcessCmds(
 }
 
 func (c *ClusterClient) txPipelineReadQueued(
-	rd *proto.Reader, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
+	rd *proto.Reader, cmds []Cmder, failedCmds *cmdsMap,
 ) error {
 	// Parse queued replies.
 	var statusCmd StatusCmd

vendor/github.com/go-redis/redis/commands.go

@@ -8,13 +8,6 @@ import (
 	"github.com/go-redis/redis/internal"
 )
 
-func readTimeout(timeout time.Duration) time.Duration {
-	if timeout == 0 {
-		return 0
-	}
-	return timeout + 10*time.Second
-}
-
 func usePrecise(dur time.Duration) bool {
 	return dur < time.Second || dur%time.Second != 0
 }
@@ -1397,6 +1390,9 @@ func (c *cmdable) XRead(a *XReadArgs) *XStreamSliceCmd {
 	}
 
 	cmd := NewXStreamSliceCmd(args...)
+	if a.Block >= 0 {
+		cmd.setReadTimeout(a.Block)
+	}
 	c.process(cmd)
 	return cmd
 }
@@ -1455,6 +1451,9 @@ func (c *cmdable) XReadGroup(a *XReadGroupArgs) *XStreamSliceCmd {
 	}
 
 	cmd := NewXStreamSliceCmd(args...)
+	if a.Block >= 0 {
+		cmd.setReadTimeout(a.Block)
+	}
 	c.process(cmd)
 	return cmd
 }

vendor/github.com/go-redis/redis/internal/error.go

@@ -9,6 +9,9 @@ import (
 )
 
 func IsRetryableError(err error, retryTimeout bool) bool {
+	if err == nil {
+		return false
+	}
 	if err == io.EOF {
 		return true
 	}

vendor/github.com/go-redis/redis/options.go

@@ -48,7 +48,7 @@ type Options struct {
 	// Default is 5 seconds.
 	DialTimeout time.Duration
 	// Timeout for socket reads. If reached, commands will fail
-	// with a timeout instead of blocking.
+	// with a timeout instead of blocking. Use value -1 for no timeout and 0 for default.
 	// Default is 3 seconds.
 	ReadTimeout time.Duration
 	// Timeout for socket writes. If reached, commands will fail

vendor/github.com/go-redis/redis/pubsub.go

@@ -1,6 +1,7 @@
 package redis
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -10,6 +11,8 @@ import (
 	"github.com/go-redis/redis/internal/proto"
 )
 
+var errPingTimeout = errors.New("redis: ping timeout")
+
 // PubSub implements Pub/Sub commands bas described in
 // http://redis.io/topics/pubsub. Message receiving is NOT safe
 // for concurrent use by multiple goroutines.
@@ -51,7 +54,6 @@ func (c *PubSub) _conn(newChannels []string) (*pool.Conn, error) {
 	if c.closed {
 		return nil, pool.ErrClosed
 	}
-
 	if c.cn != nil {
 		return c.cn, nil
 	}
@@ -439,7 +441,6 @@ func (c *PubSub) initChannel() {
 		timer.Stop()
 
 		healthy := true
-		var pingErr error
 		for {
 			timer.Reset(timeout)
 			select {
@@ -449,10 +450,13 @@ func (c *PubSub) initChannel() {
 					<-timer.C
 				}
 			case <-timer.C:
-				pingErr = c.Ping()
+				pingErr := c.Ping()
 				if healthy {
 					healthy = false
 				} else {
+					if pingErr == nil {
+						pingErr = errPingTimeout
+					}
 					c.mu.Lock()
 					c._reconnect(pingErr)
 					c.mu.Unlock()

vendor/github.com/go-redis/redis/redis.go

@@ -77,14 +77,20 @@ func (c *baseClient) getConn() (*pool.Conn, error) {
 	return cn, nil
 }
 
-func (c *baseClient) releaseConn(cn *pool.Conn, err error) bool {
+func (c *baseClient) releaseConn(cn *pool.Conn, err error) {
 	if internal.IsBadConn(err, false) {
 		c.connPool.Remove(cn)
-		return false
+	} else {
+		c.connPool.Put(cn)
 	}
+}
 
-	c.connPool.Put(cn)
-	return true
+func (c *baseClient) releaseConnStrict(cn *pool.Conn, err error) {
+	if err == nil || internal.IsRedisError(err) {
+		c.connPool.Put(cn)
+	} else {
+		c.connPool.Remove(cn)
+	}
 }
 
 func (c *baseClient) initConn(cn *pool.Conn) error {
@@ -188,7 +194,11 @@ func (c *baseClient) retryBackoff(attempt int) time.Duration {
 
 func (c *baseClient) cmdTimeout(cmd Cmder) time.Duration {
 	if timeout := cmd.readTimeout(); timeout != nil {
-		return readTimeout(*timeout)
+		t := *timeout
+		if t == 0 {
+			return 0
+		}
+		return t + 10*time.Second
 	}
 	return c.opt.ReadTimeout
 }
@@ -244,12 +254,7 @@ func (c *baseClient) generalProcessPipeline(cmds []Cmder, p pipelineProcessor) e
 		}
 
 		canRetry, err := p(cn, cmds)
-
-		if err == nil || internal.IsRedisError(err) {
-			c.connPool.Put(cn)
-			break
-		}
-		c.connPool.Remove(cn)
+		c.releaseConnStrict(cn, err)
 
 		if !canRetry || !internal.IsRetryableError(err, true) {
 			break
@@ -319,7 +324,7 @@ func txPipelineReadQueued(rd *proto.Reader, cmds []Cmder) error {
 		return err
 	}
 
-	for _ = range cmds {
+	for range cmds {
 		err = statusCmd.readReply(rd)
 		if err != nil && !internal.IsRedisError(err) {
 			return err

vendor/github.com/go-redis/redis/ring.go

@@ -342,6 +342,7 @@ type Ring struct {
 	shards        *ringShards
 	cmdsInfoCache *cmdsInfoCache
 
+	process         func(Cmder) error
 	processPipeline func([]Cmder) error
 }
 
@@ -354,6 +355,7 @@ func NewRing(opt *RingOptions) *Ring {
 	}
 	ring.cmdsInfoCache = newCmdsInfoCache(ring.cmdsInfo)
 
+	ring.process = ring.defaultProcess
 	ring.processPipeline = ring.defaultProcessPipeline
 	ring.cmdable.setProcessor(ring.Process)
 
@@ -526,19 +528,34 @@ func (c *Ring) Do(args ...interface{}) *Cmd {
 func (c *Ring) WrapProcess(
 	fn func(oldProcess func(cmd Cmder) error) func(cmd Cmder) error,
 ) {
-	c.ForEachShard(func(c *Client) error {
-		c.WrapProcess(fn)
-		return nil
-	})
+	c.process = fn(c.process)
 }
 
 func (c *Ring) Process(cmd Cmder) error {
-	shard, err := c.cmdShard(cmd)
-	if err != nil {
-		cmd.setErr(err)
-		return err
+	return c.process(cmd)
+}
+
+func (c *Ring) defaultProcess(cmd Cmder) error {
+	for attempt := 0; attempt <= c.opt.MaxRetries; attempt++ {
+		if attempt > 0 {
+			time.Sleep(c.retryBackoff(attempt))
+		}
+
+		shard, err := c.cmdShard(cmd)
+		if err != nil {
+			cmd.setErr(err)
+			return err
+		}
+
+		err = shard.Client.Process(cmd)
+		if err == nil {
+			return nil
+		}
+		if !internal.IsRetryableError(err, cmd.readTimeout() == nil) {
+			return err
+		}
 	}
-	return shard.Client.Process(cmd)
+	return cmd.Err()
 }
 
 func (c *Ring) Pipeline() Pipeliner {
@@ -575,36 +592,42 @@ func (c *Ring) defaultProcessPipeline(cmds []Cmder) error {
 			time.Sleep(c.retryBackoff(attempt))
 		}
 
+		var mu sync.Mutex
 		var failedCmdsMap map[string][]Cmder
+		var wg sync.WaitGroup
 
 		for hash, cmds := range cmdsMap {
-			shard, err := c.shards.GetByHash(hash)
-			if err != nil {
-				setCmdsErr(cmds, err)
-				continue
-			}
+			wg.Add(1)
+			go func(hash string, cmds []Cmder) {
+				defer wg.Done()
 
-			cn, err := shard.Client.getConn()
-			if err != nil {
-				setCmdsErr(cmds, err)
-				continue
-			}
-
-			canRetry, err := shard.Client.pipelineProcessCmds(cn, cmds)
-			if err == nil || internal.IsRedisError(err) {
-				shard.Client.connPool.Put(cn)
-				continue
-			}
-			shard.Client.connPool.Remove(cn)
-
-			if canRetry && internal.IsRetryableError(err, true) {
-				if failedCmdsMap == nil {
-					failedCmdsMap = make(map[string][]Cmder)
+				shard, err := c.shards.GetByHash(hash)
+				if err != nil {
+					setCmdsErr(cmds, err)
+					return
 				}
-				failedCmdsMap[hash] = cmds
-			}
+
+				cn, err := shard.Client.getConn()
+				if err != nil {
+					setCmdsErr(cmds, err)
+					return
+				}
+
+				canRetry, err := shard.Client.pipelineProcessCmds(cn, cmds)
+				shard.Client.releaseConnStrict(cn, err)
+
+				if canRetry && internal.IsRetryableError(err, true) {
+					mu.Lock()
+					if failedCmdsMap == nil {
+						failedCmdsMap = make(map[string][]Cmder)
+					}
+					failedCmdsMap[hash] = cmds
+					mu.Unlock()
+				}
+			}(hash, cmds)
 		}
 
+		wg.Wait()
 		if len(failedCmdsMap) == 0 {
 			break
 		}

vendor/github.com/go-redis/redis/sentinel.go

@@ -119,7 +119,7 @@ func NewSentinelClient(opt *Options) *SentinelClient {
 	return c
 }
 
-func (c *SentinelClient) PubSub() *PubSub {
+func (c *SentinelClient) pubSub() *PubSub {
 	pubsub := &PubSub{
 		opt: c.opt,
 
@@ -132,14 +132,34 @@ func (c *SentinelClient) PubSub() *PubSub {
 	return pubsub
 }
 
+// Subscribe subscribes the client to the specified channels.
+// Channels can be omitted to create empty subscription.
+func (c *SentinelClient) Subscribe(channels ...string) *PubSub {
+	pubsub := c.pubSub()
+	if len(channels) > 0 {
+		_ = pubsub.Subscribe(channels...)
+	}
+	return pubsub
+}
+
+// PSubscribe subscribes the client to the given patterns.
+// Patterns can be omitted to create empty subscription.
+func (c *SentinelClient) PSubscribe(channels ...string) *PubSub {
+	pubsub := c.pubSub()
+	if len(channels) > 0 {
+		_ = pubsub.PSubscribe(channels...)
+	}
+	return pubsub
+}
+
 func (c *SentinelClient) GetMasterAddrByName(name string) *StringSliceCmd {
-	cmd := NewStringSliceCmd("SENTINEL", "get-master-addr-by-name", name)
+	cmd := NewStringSliceCmd("sentinel", "get-master-addr-by-name", name)
 	c.Process(cmd)
 	return cmd
 }
 
 func (c *SentinelClient) Sentinels(name string) *SliceCmd {
-	cmd := NewSliceCmd("SENTINEL", "sentinels", name)
+	cmd := NewSliceCmd("sentinel", "sentinels", name)
 	c.Process(cmd)
 	return cmd
 }
@@ -158,77 +178,78 @@ type sentinelFailover struct {
 	sentinel    *SentinelClient
 }
 
-func (d *sentinelFailover) Close() error {
-	return d.resetSentinel()
+func (c *sentinelFailover) Close() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.sentinel != nil {
+		return c.closeSentinel()
+	}
+	return nil
 }
 
-func (d *sentinelFailover) Pool() *pool.ConnPool {
-	d.poolOnce.Do(func() {
-		d.opt.Dialer = d.dial
-		d.pool = newConnPool(d.opt)
+func (c *sentinelFailover) Pool() *pool.ConnPool {
+	c.poolOnce.Do(func() {
+		c.opt.Dialer = c.dial
+		c.pool = newConnPool(c.opt)
 	})
-	return d.pool
+	return c.pool
 }
 
-func (d *sentinelFailover) dial() (net.Conn, error) {
-	addr, err := d.MasterAddr()
+func (c *sentinelFailover) dial() (net.Conn, error) {
+	addr, err := c.MasterAddr()
 	if err != nil {
 		return nil, err
 	}
-	return net.DialTimeout("tcp", addr, d.opt.DialTimeout)
+	return net.DialTimeout("tcp", addr, c.opt.DialTimeout)
 }
 
-func (d *sentinelFailover) MasterAddr() (string, error) {
-	d.mu.Lock()
-	defer d.mu.Unlock()
-
-	addr, err := d.masterAddr()
+func (c *sentinelFailover) MasterAddr() (string, error) {
+	addr, err := c.masterAddr()
 	if err != nil {
 		return "", err
 	}
-	d._switchMaster(addr)
-
+	c.switchMaster(addr)
 	return addr, nil
 }
 
-func (d *sentinelFailover) masterAddr() (string, error) {
-	// Try last working sentinel.
-	if d.sentinel != nil {
-		addr, err := d.sentinel.GetMasterAddrByName(d.masterName).Result()
-		if err == nil {
-			addr := net.JoinHostPort(addr[0], addr[1])
-			return addr, nil
-		}
-
-		internal.Logf("sentinel: GetMasterAddrByName name=%q failed: %s",
-			d.masterName, err)
-		d._resetSentinel()
+func (c *sentinelFailover) masterAddr() (string, error) {
+	addr := c.getMasterAddr()
+	if addr != "" {
+		return addr, nil
 	}
 
-	for i, sentinelAddr := range d.sentinelAddrs {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	for i, sentinelAddr := range c.sentinelAddrs {
 		sentinel := NewSentinelClient(&Options{
 			Addr: sentinelAddr,
 
-			DialTimeout:  d.opt.DialTimeout,
-			ReadTimeout:  d.opt.ReadTimeout,
-			WriteTimeout: d.opt.WriteTimeout,
+			MaxRetries: c.opt.MaxRetries,
 
-			PoolSize:    d.opt.PoolSize,
-			PoolTimeout: d.opt.PoolTimeout,
-			IdleTimeout: d.opt.IdleTimeout,
+			DialTimeout:  c.opt.DialTimeout,
+			ReadTimeout:  c.opt.ReadTimeout,
+			WriteTimeout: c.opt.WriteTimeout,
+
+			PoolSize:           c.opt.PoolSize,
+			PoolTimeout:        c.opt.PoolTimeout,
+			IdleTimeout:        c.opt.IdleTimeout,
+			IdleCheckFrequency: c.opt.IdleCheckFrequency,
+
+			TLSConfig: c.opt.TLSConfig,
 		})
 
-		masterAddr, err := sentinel.GetMasterAddrByName(d.masterName).Result()
+		masterAddr, err := sentinel.GetMasterAddrByName(c.masterName).Result()
 		if err != nil {
 			internal.Logf("sentinel: GetMasterAddrByName master=%q failed: %s",
-				d.masterName, err)
-			sentinel.Close()
+				c.masterName, err)
+			_ = sentinel.Close()
 			continue
 		}
 
 		// Push working sentinel to the top.
-		d.sentinelAddrs[0], d.sentinelAddrs[i] = d.sentinelAddrs[i], d.sentinelAddrs[0]
-		d.setSentinel(sentinel)
+		c.sentinelAddrs[0], c.sentinelAddrs[i] = c.sentinelAddrs[i], c.sentinelAddrs[0]
+		c.setSentinel(sentinel)
 
 		addr := net.JoinHostPort(masterAddr[0], masterAddr[1])
 		return addr, nil
@@ -237,17 +258,41 @@ func (d *sentinelFailover) masterAddr() (string, error) {
 	return "", errors.New("redis: all sentinels are unreachable")
 }
 
-func (c *sentinelFailover) switchMaster(addr string) {
-	c.mu.Lock()
-	c._switchMaster(addr)
-	c.mu.Unlock()
+func (c *sentinelFailover) getMasterAddr() string {
+	c.mu.RLock()
+	sentinel := c.sentinel
+	c.mu.RUnlock()
+
+	if sentinel == nil {
+		return ""
+	}
+
+	addr, err := sentinel.GetMasterAddrByName(c.masterName).Result()
+	if err != nil {
+		internal.Logf("sentinel: GetMasterAddrByName name=%q failed: %s",
+			c.masterName, err)
+		c.mu.Lock()
+		if c.sentinel == sentinel {
+			c.closeSentinel()
+		}
+		c.mu.Unlock()
+		return ""
+	}
+
+	return net.JoinHostPort(addr[0], addr[1])
 }
 
-func (c *sentinelFailover) _switchMaster(addr string) {
-	if c._masterAddr == addr {
+func (c *sentinelFailover) switchMaster(addr string) {
+	c.mu.RLock()
+	masterAddr := c._masterAddr
+	c.mu.RUnlock()
+	if masterAddr == addr {
 		return
 	}
 
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
 	internal.Logf("sentinel: new master=%q addr=%q",
 		c.masterName, addr)
 	_ = c.Pool().Filter(func(cn *pool.Conn) bool {
@@ -256,32 +301,22 @@ func (c *sentinelFailover) _switchMaster(addr string) {
 	c._masterAddr = addr
 }
 
-func (d *sentinelFailover) setSentinel(sentinel *SentinelClient) {
-	d.discoverSentinels(sentinel)
-	d.sentinel = sentinel
-	go d.listen(sentinel)
+func (c *sentinelFailover) setSentinel(sentinel *SentinelClient) {
+	c.discoverSentinels(sentinel)
+	c.sentinel = sentinel
+	go c.listen(sentinel)
 }
 
-func (d *sentinelFailover) resetSentinel() error {
-	var err error
-	d.mu.Lock()
-	if d.sentinel != nil {
-		err = d._resetSentinel()
-	}
-	d.mu.Unlock()
+func (c *sentinelFailover) closeSentinel() error {
+	err := c.sentinel.Close()
+	c.sentinel = nil
 	return err
 }
 
-func (d *sentinelFailover) _resetSentinel() error {
-	err := d.sentinel.Close()
-	d.sentinel = nil
-	return err
-}
-
-func (d *sentinelFailover) discoverSentinels(sentinel *SentinelClient) {
-	sentinels, err := sentinel.Sentinels(d.masterName).Result()
+func (c *sentinelFailover) discoverSentinels(sentinel *SentinelClient) {
+	sentinels, err := sentinel.Sentinels(c.masterName).Result()
 	if err != nil {
-		internal.Logf("sentinel: Sentinels master=%q failed: %s", d.masterName, err)
+		internal.Logf("sentinel: Sentinels master=%q failed: %s", c.masterName, err)
 		return
 	}
 	for _, sentinel := range sentinels {
@@ -290,49 +325,36 @@ func (d *sentinelFailover) discoverSentinels(sentinel *SentinelClient) {
 			key := vals[i].(string)
 			if key == "name" {
 				sentinelAddr := vals[i+1].(string)
-				if !contains(d.sentinelAddrs, sentinelAddr) {
-					internal.Logf(
-						"sentinel: discovered new sentinel=%q for master=%q",
-						sentinelAddr, d.masterName,
-					)
-					d.sentinelAddrs = append(d.sentinelAddrs, sentinelAddr)
+				if !contains(c.sentinelAddrs, sentinelAddr) {
+					internal.Logf("sentinel: discovered new sentinel=%q for master=%q",
+						sentinelAddr, c.masterName)
+					c.sentinelAddrs = append(c.sentinelAddrs, sentinelAddr)
 				}
 			}
 		}
 	}
 }
 
-func (d *sentinelFailover) listen(sentinel *SentinelClient) {
-	pubsub := sentinel.PubSub()
+func (c *sentinelFailover) listen(sentinel *SentinelClient) {
+	pubsub := sentinel.Subscribe("+switch-master")
 	defer pubsub.Close()
 
-	err := pubsub.Subscribe("+switch-master")
-	if err != nil {
-		internal.Logf("sentinel: Subscribe failed: %s", err)
-		d.resetSentinel()
-		return
-	}
-
+	ch := pubsub.Channel()
 	for {
-		msg, err := pubsub.ReceiveMessage()
-		if err != nil {
-			if err == pool.ErrClosed {
-				d.resetSentinel()
-				return
-			}
-			internal.Logf("sentinel: ReceiveMessage failed: %s", err)
-			continue
+		msg, ok := <-ch
+		if !ok {
+			break
 		}
 
 		switch msg.Channel {
 		case "+switch-master":
 			parts := strings.Split(msg.Payload, " ")
-			if parts[0] != d.masterName {
+			if parts[0] != c.masterName {
 				internal.Logf("sentinel: ignore addr for master=%q", parts[0])
 				continue
 			}
 			addr := net.JoinHostPort(parts[3], parts[4])
-			d.switchMaster(addr)
+			c.switchMaster(addr)
 		}
 	}
 }