forked from mirror/redis
Retry timeout and retryable error
This commit is contained in:
parent
16e62e05a1
commit
21a1f58caf
|
@ -810,7 +810,7 @@ func (c *ClusterClient) _process(ctx context.Context, cmd Cmder) error {
|
|||
continue
|
||||
}
|
||||
|
||||
if isRetryableError(lastErr, cmd.readTimeout() == nil) {
|
||||
if shouldRetry(lastErr, cmd.readTimeout() == nil) {
|
||||
// First retry the same node.
|
||||
if attempt == 0 {
|
||||
continue
|
||||
|
@ -1466,7 +1466,7 @@ func (c *ClusterClient) Watch(ctx context.Context, fn func(*Tx) error, keys ...s
|
|||
continue
|
||||
}
|
||||
|
||||
if isRetryableError(err, true) {
|
||||
if shouldRetry(err, true) {
|
||||
continue
|
||||
}
|
||||
|
||||
|
|
23
error.go
23
error.go
|
@ -24,15 +24,16 @@ type Error interface {
|
|||
|
||||
var _ Error = proto.RedisError("")
|
||||
|
||||
func isRetryableError(err error, retryTimeout bool) bool {
|
||||
func shouldRetry(err error, retryTimeout bool) bool {
|
||||
switch err {
|
||||
case io.EOF, io.ErrUnexpectedEOF:
|
||||
return true
|
||||
case nil, context.Canceled, context.DeadlineExceeded:
|
||||
return false
|
||||
case io.EOF:
|
||||
return true
|
||||
}
|
||||
if netErr, ok := err.(net.Error); ok {
|
||||
if netErr.Timeout() {
|
||||
|
||||
if v, ok := err.(timeoutError); ok {
|
||||
if v.Timeout() {
|
||||
return retryTimeout
|
||||
}
|
||||
return true
|
||||
|
@ -51,6 +52,7 @@ func isRetryableError(err error, retryTimeout bool) bool {
|
|||
if strings.HasPrefix(s, "CLUSTERDOWN ") {
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
|
@ -63,16 +65,19 @@ func isBadConn(err error, allowTimeout bool) bool {
|
|||
if err == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
if isRedisError(err) {
|
||||
// Close connections in read only state in case domain addr is used
|
||||
// and domain resolves to a different Redis Server. See #790.
|
||||
return isReadOnlyError(err)
|
||||
}
|
||||
|
||||
if allowTimeout {
|
||||
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
|
||||
return false
|
||||
return !netErr.Temporary()
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
|
@ -106,3 +111,9 @@ func isLoadingError(err error) bool {
|
|||
func isReadOnlyError(err error) bool {
|
||||
return strings.HasPrefix(err.Error(), "READONLY ")
|
||||
}
|
||||
|
||||
//------------------------------------------------------------------------------
|
||||
|
||||
type timeoutError interface {
|
||||
Timeout() bool
|
||||
}
|
||||
|
|
|
@ -332,7 +332,7 @@ func startSentinel(port, masterName, masterPort string) (*redisProcess, error) {
|
|||
type badConnError string
|
||||
|
||||
func (e badConnError) Error() string { return string(e) }
|
||||
func (e badConnError) Timeout() bool { return false }
|
||||
func (e badConnError) Timeout() bool { return true }
|
||||
func (e badConnError) Temporary() bool { return false }
|
||||
|
||||
type badConn struct {
|
||||
|
|
4
redis.go
4
redis.go
|
@ -345,7 +345,7 @@ func (c *baseClient) _process(ctx context.Context, cmd Cmder) error {
|
|||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
retry = isRetryableError(err, retryTimeout)
|
||||
retry = shouldRetry(err, retryTimeout)
|
||||
return err
|
||||
})
|
||||
if err == nil || !retry {
|
||||
|
@ -430,7 +430,7 @@ func (c *baseClient) _generalProcessPipeline(
|
|||
canRetry, err = p(ctx, cn, cmds)
|
||||
return err
|
||||
})
|
||||
if lastErr == nil || !canRetry || !isRetryableError(lastErr, true) {
|
||||
if lastErr == nil || !canRetry || !shouldRetry(lastErr, true) {
|
||||
return lastErr
|
||||
}
|
||||
}
|
||||
|
|
2
ring.go
2
ring.go
|
@ -597,7 +597,7 @@ func (c *Ring) _process(ctx context.Context, cmd Cmder) error {
|
|||
}
|
||||
|
||||
lastErr = shard.Client.Process(ctx, cmd)
|
||||
if lastErr == nil || !isRetryableError(lastErr, cmd.readTimeout() == nil) {
|
||||
if lastErr == nil || !shouldRetry(lastErr, cmd.readTimeout() == nil) {
|
||||
return lastErr
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue