Skip to content

Commit

Permalink
Add proper SingleConnPool implementation
Browse files Browse the repository at this point in the history
  • Loading branch information
vmihailenco committed Aug 3, 2019
1 parent d6a99e7 commit 4e9cea8
Show file tree
Hide file tree
Showing 7 changed files with 211 additions and 68 deletions.
22 changes: 11 additions & 11 deletions cluster.go
Original file line number Diff line number Diff line change
Expand Up @@ -787,15 +787,15 @@ func (c *ClusterClient) process(ctx context.Context, cmd Cmder) error {
}

// If slave is loading - pick another node.
if c.opt.ReadOnly && internal.IsLoadingError(err) {
if c.opt.ReadOnly && isLoadingError(err) {
node.MarkAsFailing()
node = nil
continue
}

var moved bool
var addr string
moved, ask, addr = internal.IsMovedError(err)
moved, ask, addr = isMovedError(err)
if moved || ask {
node, err = c.nodes.Get(addr)
if err != nil {
Expand All @@ -804,12 +804,12 @@ func (c *ClusterClient) process(ctx context.Context, cmd Cmder) error {
continue
}

if err == pool.ErrClosed || internal.IsReadOnlyError(err) {
if err == pool.ErrClosed || isReadOnlyError(err) {
node = nil
continue
}

if internal.IsRetryableError(err, true) {
if isRetryableError(err, true) {
// First retry the same node.
if attempt == 0 {
continue
Expand Down Expand Up @@ -1173,9 +1173,9 @@ func (c *ClusterClient) pipelineReadCmds(
continue
}

if c.opt.ReadOnly && internal.IsLoadingError(err) {
if c.opt.ReadOnly && isLoadingError(err) {
node.MarkAsFailing()
} else if internal.IsRedisError(err) {
} else if isRedisError(err) {
continue
}

Expand All @@ -1192,7 +1192,7 @@ func (c *ClusterClient) pipelineReadCmds(
func (c *ClusterClient) checkMovedErr(
cmd Cmder, err error, failedCmds *cmdsMap,
) bool {
moved, ask, addr := internal.IsMovedError(err)
moved, ask, addr := isMovedError(err)

if moved {
c.state.LazyReload()
Expand Down Expand Up @@ -1346,7 +1346,7 @@ func (c *ClusterClient) txPipelineReadQueued(
continue
}

if c.checkMovedErr(cmd, err, failedCmds) || internal.IsRedisError(err) {
if c.checkMovedErr(cmd, err, failedCmds) || isRedisError(err) {
continue
}

Expand Down Expand Up @@ -1418,7 +1418,7 @@ func (c *ClusterClient) WatchContext(ctx context.Context, fn func(*Tx) error, ke
c.state.LazyReload()
}

moved, ask, addr := internal.IsMovedError(err)
moved, ask, addr := isMovedError(err)
if moved || ask {
node, err = c.nodes.Get(addr)
if err != nil {
Expand All @@ -1427,15 +1427,15 @@ func (c *ClusterClient) WatchContext(ctx context.Context, fn func(*Tx) error, ke
continue
}

if err == pool.ErrClosed || internal.IsReadOnlyError(err) {
if err == pool.ErrClosed || isReadOnlyError(err) {
node, err = c.slotMasterNode(slot)
if err != nil {
return err
}
continue
}

if internal.IsRetryableError(err, true) {
if isRetryableError(err, true) {
continue
}

Expand Down
35 changes: 16 additions & 19 deletions internal/error.go → error.go
Original file line number Diff line number Diff line change
@@ -1,20 +1,18 @@
package internal
package redis

import (
"context"
"errors"
"io"
"net"
"strings"

"github.com/go-redis/redis/internal/pool"
"github.com/go-redis/redis/internal/proto"
)

var ErrSingleConnPoolClosed = errors.New("redis: SingleConnPool is closed")

func IsRetryableError(err error, retryTimeout bool) bool {
func isRetryableError(err error, retryTimeout bool) bool {
switch err {
case nil, context.Canceled, context.DeadlineExceeded:
case nil, context.Canceled, context.DeadlineExceeded, pool.ErrBadConn:
return false
case io.EOF:
return true
Expand All @@ -25,9 +23,6 @@ func IsRetryableError(err error, retryTimeout bool) bool {
}
return true
}
if err == ErrSingleConnPoolClosed {
return true
}

s := err.Error()
if s == "ERR max number of clients reached" {
Expand All @@ -45,18 +40,20 @@ func IsRetryableError(err error, retryTimeout bool) bool {
return false
}

func IsRedisError(err error) bool {
func isRedisError(err error) bool {
_, ok := err.(proto.RedisError)
return ok
}

func IsBadConn(err error, allowTimeout bool) bool {
if err == nil {
func isBadConn(err error, allowTimeout bool) bool {
switch err {
case nil:
return false
case pool.ErrBadConn:
return true
}
if IsRedisError(err) {
// #790
return IsReadOnlyError(err)
if isRedisError(err) {
return isReadOnlyError(err) // #790
}
if allowTimeout {
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
Expand All @@ -66,8 +63,8 @@ func IsBadConn(err error, allowTimeout bool) bool {
return true
}

func IsMovedError(err error) (moved bool, ask bool, addr string) {
if !IsRedisError(err) {
func isMovedError(err error) (moved bool, ask bool, addr string) {
if !isRedisError(err) {
return
}

Expand All @@ -89,10 +86,10 @@ func IsMovedError(err error) (moved bool, ask bool, addr string) {
return
}

func IsLoadingError(err error) bool {
func isLoadingError(err error) bool {
return strings.HasPrefix(err.Error(), "LOADING ")
}

func IsReadOnlyError(err error) bool {
func isReadOnlyError(err error) bool {
return strings.HasPrefix(err.Error(), "READONLY ")
}
21 changes: 21 additions & 0 deletions example_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,27 @@ func ExampleClient() {
// missing_key does not exist
}

func ExampleConn() {
conn := redisdb.Conn()

err := conn.ClientSetName("foobar").Err()
if err != nil {
panic(err)
}

// Open other connections.
for i := 0; i < 10; i++ {
go redisdb.Ping()
}

s, err := conn.ClientGetName().Result()
if err != nil {
panic(err)
}
fmt.Println(s)
// Output: foobar
}

func ExampleClient_Set() {
// Last argument is expiration. Zero means the key has no
// expiration time.
Expand Down
Loading

0 comments on commit 4e9cea8

Please sign in to comment.