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

feat: add backoff and jitter to retry #643

Merged
merged 11 commits into from
Oct 13, 2024
Merged

feat: add backoff and jitter to retry #643

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 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
1725eb5
refactor: remove useless return error
proost Oct 7, 2024
15415a1
refactor: remove useless code
proost Oct 7, 2024
217acd7
refactor: rollback code
proost Oct 7, 2024
2df785e
refactor: use retry error channel and redirects
proost Oct 11, 2024
f04b433
refactor: divide method
proost Oct 12, 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
88 changes: 71 additions & 17 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,22 +32,29 @@ 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) {
goto retry
shouldRetry := c.retryHandler.WaitOrSkipRetry(
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)
Expand Down Expand Up @@ -75,12 +83,19 @@ func (c *singleClient) DoMulti(ctx context.Context, multi ...Completed) (resps [
if len(multi) == 0 {
return nil
}
attempts := 1
retry:
resps = c.conn.DoMulti(ctx, multi...).s
if c.retry && allReadOnly(multi) {
for _, resp := range resps {
if c.isRetryable(resp.NonRedisError(), ctx) {
goto retry
shouldRetry := c.retryHandler.WaitOrSkipRetry(
ctx, attempts, resp.Error(),
)
if shouldRetry {
attempts++
goto retry
}
}
}
}
Expand All @@ -96,12 +111,19 @@ func (c *singleClient) DoMultiCache(ctx context.Context, multi ...CacheableTTL)
if len(multi) == 0 {
return nil
}
attempts := 1
retry:
resps = c.conn.DoMultiCache(ctx, multi...).s
if c.retry {
for _, resp := range resps {
if c.isRetryable(resp.NonRedisError(), ctx) {
goto retry
shouldRetry := c.retryHandler.WaitOrSkipRetry(
ctx, attempts, resp.Error(),
)
if shouldRetry {
attempts++
goto retry
}
}
}
}
Expand All @@ -114,10 +136,15 @@ retry:
}

func (c *singleClient) DoCache(ctx context.Context, cmd Cacheable, ttl time.Duration) (resp RedisResult) {
attempts := 1
retry:
resp = c.conn.DoCache(ctx, cmd, ttl)
if c.retry && c.isRetryable(resp.NonRedisError(), ctx) {
goto retry
shouldRetry := c.retryHandler.WaitOrSkipRetry(ctx, attempts, resp.Error())
if shouldRetry {
attempts++
goto retry
}
}
if err := resp.NonRedisError(); err == nil || err == ErrDoCacheAborted {
cmds.PutCacheable(cmd)
Expand All @@ -126,11 +153,16 @@ retry:
}

func (c *singleClient) Receive(ctx context.Context, subscribe Completed, fn func(msg PubSubMessage)) (err error) {
attempts := 1
retry:
err = c.conn.Receive(ctx, subscribe, fn)
if c.retry {
if _, ok := err.(*RedisError); !ok && c.isRetryable(err, ctx) {
goto retry
shouldRetry := c.retryHandler.WaitOrSkipRetry(ctx, attempts, err)
if shouldRetry {
attempts++
goto retry
}
}
}
if err == nil {
Expand All @@ -141,15 +173,15 @@ retry:

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,21 +200,29 @@ 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) {
attempts := 1
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) {
goto retry
shouldRetry := c.retryHandler.WaitOrSkipRetry(
ctx, attempts, resp.Error(),
)
if shouldRetry {
attempts++
goto retry
}
}
if resp.NonRedisError() == nil {
cmds.PutCompleted(cmd)
Expand All @@ -194,6 +234,7 @@ func (c *dedicatedSingleClient) DoMulti(ctx context.Context, multi ...Completed)
if len(multi) == 0 {
return nil
}
attempts := 1
retryable := c.retry
if retryable {
retryable = allReadOnly(multi)
Expand All @@ -203,10 +244,16 @@ 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 {
if retryable && isRetryable(resp[i].NonRedisError(), c.wire, ctx) {
shouldRetry := c.retryHandler.WaitOrSkipRetry(
ctx, attempts, resp[i].Error(),
)
if shouldRetry {
attempts++
goto retry
}
}
if resp[i].NonRedisError() == nil {
cmds.PutCompleted(cmd)
}
Expand All @@ -215,14 +262,21 @@ retry:
}

func (c *dedicatedSingleClient) Receive(ctx context.Context, subscribe Completed, fn func(msg PubSubMessage)) (err error) {
attempts := 1
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 := c.retryHandler.WaitOrSkipRetry(
ctx, attempts, err,
)
if shouldRetry {
attempts++
goto retry
}
}
}
if err == nil {
Expand Down
Loading