Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: add backoff and jitter to retry #643

Open
wants to merge 6 commits into
base: main
Choose a base branch
from
+1,796 −884
Open

feat: add backoff and jitter to retry #643

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
5337da7
feat: add backoff and jitter to retry
proost Oct 3, 2024
9e884c3
refactor: check before retry wait
proost Oct 5, 2024
78ad959
style: rollback old code style
proost Oct 5, 2024
ea85114
refactor: do not return error
proost Oct 6, 2024
d172784
refactor: strictly exponential backoff
proost Oct 6, 2024
e32373d
refactor: backward compatibility
proost Oct 6, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
166 changes: 143 additions & 23 deletions client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,10 +14,11 @@ type singleClient struct {
stop uint32
cmd Builder
retry bool
retryHandler retryHandler
DisableCache bool
}

func newSingleClient(opt *ClientOption, prev conn, connFn connFn) (*singleClient, error) {
func newSingleClient(opt *ClientOption, prev conn, connFn connFn, retryer retryHandler) (*singleClient, error) {
if len(opt.InitAddress) == 0 {
return nil, ErrNoAddr
}
Expand All @@ -31,26 +32,37 @@ func newSingleClient(opt *ClientOption, prev conn, connFn connFn) (*singleClient
if err := conn.Dial(); err != nil {
return nil, err
}
return newSingleClientWithConn(conn, cmds.NewBuilder(cmds.NoSlot), !opt.DisableRetry, opt.DisableCache), nil
return newSingleClientWithConn(conn, cmds.NewBuilder(cmds.NoSlot), !opt.DisableRetry, opt.DisableCache, retryer), nil
}

func newSingleClientWithConn(conn conn, builder Builder, retry, disableCache bool) *singleClient {
return &singleClient{cmd: builder, conn: conn, retry: retry, DisableCache: disableCache}
func newSingleClientWithConn(conn conn, builder Builder, retry, disableCache bool, retryer retryHandler) *singleClient {
return &singleClient{cmd: builder, conn: conn, retry: retry, retryHandler: retryer, DisableCache: disableCache}
}

func (c *singleClient) B() Builder {
return c.cmd
}

func (c *singleClient) Do(ctx context.Context, cmd Completed) (resp RedisResult) {
attempts := 1
retry:
resp = c.conn.Do(ctx, cmd)
if c.retry && cmd.IsReadOnly() && c.isRetryable(resp.NonRedisError(), ctx) {
shouldRetry, err := c.retryHandler.WaitUntilNextRetry(
ctx,
func() bool { return c.retry && cmd.IsReadOnly() && c.isRetryable(resp.NonRedisError(), ctx) },
attempts,
resp.Error(),
)
if shouldRetry {
attempts++
goto retry
}
if resp.NonRedisError() == nil { // not recycle cmds if error, since cmds may be used later in pipe. consider recycle them by pipe
cmds.PutCompleted(cmd)
}
if err != nil {
return newErrResult(err)
}
return resp
}

Expand All @@ -75,81 +87,139 @@ func (c *singleClient) DoMulti(ctx context.Context, multi ...Completed) (resps [
if len(multi) == 0 {
return nil
}
var (
attempts = 1
shouldRetry bool
errAbortWaiting error
)
retry:
resps = c.conn.DoMulti(ctx, multi...).s
if c.retry && allReadOnly(multi) {
for _, resp := range resps {
if c.isRetryable(resp.NonRedisError(), ctx) {
shouldRetry, errAbortWaiting = c.retryHandler.WaitUntilNextRetry(
ctx, func() bool { return c.isRetryable(resp.NonRedisError(), ctx) }, attempts, resp.Error(),
)
if shouldRetry {
attempts++
goto retry
}
if errAbortWaiting != nil {
break
}
}
}
for i, cmd := range multi {
if resps[i].NonRedisError() == nil {
cmds.PutCompleted(cmd)
}
}
if errAbortWaiting != nil {
return fillErrs(len(multi), errAbortWaiting)
}
return resps
}

func (c *singleClient) DoMultiCache(ctx context.Context, multi ...CacheableTTL) (resps []RedisResult) {
if len(multi) == 0 {
return nil
}
var (
attempts = 1
shouldRetry bool
errAbortWaiting error
)
retry:
resps = c.conn.DoMultiCache(ctx, multi...).s
if c.retry {
for _, resp := range resps {
if c.isRetryable(resp.NonRedisError(), ctx) {
shouldRetry, errAbortWaiting = c.retryHandler.WaitUntilNextRetry(
ctx, func() bool { return c.isRetryable(resp.NonRedisError(), ctx) }, attempts, resp.Error(),
)
if shouldRetry {
attempts++
goto retry
}
if errAbortWaiting != nil {
break
}
}
}
for i, cmd := range multi {
if err := resps[i].NonRedisError(); err == nil || err == ErrDoCacheAborted {
cmds.PutCacheable(cmd.Cmd)
}
}
if errAbortWaiting != nil {
return fillErrs(len(multi), errAbortWaiting)
}
return resps
}

func (c *singleClient) DoCache(ctx context.Context, cmd Cacheable, ttl time.Duration) (resp RedisResult) {
var (
attempts = 1
shouldRetry bool
errAbortWaiting error
)
retry:
resp = c.conn.DoCache(ctx, cmd, ttl)
if c.retry && c.isRetryable(resp.NonRedisError(), ctx) {
shouldRetry, errAbortWaiting = c.retryHandler.WaitUntilNextRetry(
ctx, func() bool { return c.retry && c.isRetryable(resp.NonRedisError(), ctx) }, attempts, resp.Error(),
)
if shouldRetry {
attempts++
goto retry
}
if err := resp.NonRedisError(); err == nil || err == ErrDoCacheAborted {
cmds.PutCacheable(cmd)
}
if errAbortWaiting != nil {
return newErrResult(errAbortWaiting)
}
return resp
}

func (c *singleClient) Receive(ctx context.Context, subscribe Completed, fn func(msg PubSubMessage)) (err error) {
var (
attempts = 1
shouldRetry bool
errAbortWaiting error
)
retry:
err = c.conn.Receive(ctx, subscribe, fn)
if c.retry {
if _, ok := err.(*RedisError); !ok && c.isRetryable(err, ctx) {
goto retry
}
shouldRetry, errAbortWaiting = c.retryHandler.WaitUntilNextRetry(
ctx,
func() bool {
_, ok := err.(*RedisError)
return c.retry && !ok && c.isRetryable(err, ctx)
},
attempts,
err,
)
if shouldRetry {
attempts++
goto retry
}
if err == nil {
cmds.PutCompleted(subscribe)
}
if errAbortWaiting != nil {
return errAbortWaiting
}
return err
}

func (c *singleClient) Dedicated(fn func(DedicatedClient) error) (err error) {
wire := c.conn.Acquire()
dsc := &dedicatedSingleClient{cmd: c.cmd, conn: c.conn, wire: wire, retry: c.retry}
dsc := &dedicatedSingleClient{cmd: c.cmd, conn: c.conn, wire: wire, retry: c.retry, retryHandler: c.retryHandler}
err = fn(dsc)
dsc.release()
return err
}

func (c *singleClient) Dedicate() (DedicatedClient, func()) {
wire := c.conn.Acquire()
dsc := &dedicatedSingleClient{cmd: c.cmd, conn: c.conn, wire: wire, retry: c.retry}
dsc := &dedicatedSingleClient{cmd: c.cmd, conn: c.conn, wire: wire, retry: c.retry, retryHandler: c.retryHandler}
return dsc, dsc.release
}

Expand All @@ -168,32 +238,53 @@ type dedicatedSingleClient struct {
mark uint32
cmd Builder

retry bool
retry bool
retryHandler retryHandler
}

func (c *dedicatedSingleClient) B() Builder {
return c.cmd
}

func (c *dedicatedSingleClient) Do(ctx context.Context, cmd Completed) (resp RedisResult) {
var (
attempts = 1
shouldRetry bool
errAbortWaiting error
)
retry:
if err := c.check(); err != nil {
return newErrResult(err)
}
resp = c.wire.Do(ctx, cmd)
if c.retry && cmd.IsReadOnly() && isRetryable(resp.NonRedisError(), c.wire, ctx) {
shouldRetry, errAbortWaiting = c.retryHandler.WaitUntilNextRetry(
ctx,
func() bool { return c.retry && cmd.IsReadOnly() && isRetryable(resp.NonRedisError(), c.wire, ctx) },
attempts,
resp.Error(),
)
if shouldRetry {
attempts++
goto retry
}
if resp.NonRedisError() == nil {
cmds.PutCompleted(cmd)
}
if errAbortWaiting != nil {
return newErrResult(errAbortWaiting)
}
return resp
}

func (c *dedicatedSingleClient) DoMulti(ctx context.Context, multi ...Completed) (resp []RedisResult) {
if len(multi) == 0 {
return nil
}
var (
attempts = 1
shouldRetry bool
errAbortWaiting error
)
retryable := c.retry
if retryable {
retryable = allReadOnly(multi)
Expand All @@ -203,31 +294,60 @@ retry:
return fillErrs(len(multi), err)
}
resp = c.wire.DoMulti(ctx, multi...).s
if retryable && anyRetryable(resp, c.wire, ctx) {
goto retry
}
for i, cmd := range multi {
shouldRetry, errAbortWaiting = c.retryHandler.WaitUntilNextRetry(
ctx,
func() bool { return retryable && isRetryable(resp[i].NonRedisError(), c.wire, ctx) },
attempts,
resp[i].Error(),
)
if shouldRetry {
attempts++
goto retry
}
if errAbortWaiting != nil {
break
}
if resp[i].NonRedisError() == nil {
cmds.PutCompleted(cmd)
}
}
if errAbortWaiting != nil {
return fillErrs(len(multi), errAbortWaiting)
}
return resp
}

func (c *dedicatedSingleClient) Receive(ctx context.Context, subscribe Completed, fn func(msg PubSubMessage)) (err error) {
var (
attempts = 1
shouldRetry bool
errAbortWaiting error
)
retry:
if err := c.check(); err != nil {
return err
}
err = c.wire.Receive(ctx, subscribe, fn)
if c.retry {
if _, ok := err.(*RedisError); !ok && isRetryable(err, c.wire, ctx) {
goto retry
}
shouldRetry, errAbortWaiting = c.retryHandler.WaitUntilNextRetry(
ctx,
func() bool {
_, ok := err.(*RedisError)
return c.retry && !ok && isRetryable(err, c.wire, ctx)
},
attempts,
err,
)
if shouldRetry {
attempts++
goto retry
}
if err == nil {
cmds.PutCompleted(subscribe)
}
if errAbortWaiting != nil {
return errAbortWaiting
}
return err
}

Expand Down
Loading