Merge pull request #4 from wangkechun/wkc/next

MaxRetryCapacity

Author: wangkechun

retry.go

@@ -23,17 +23,19 @@ type retryableFuncResp struct {
 }
 
 type Config struct {
-	FastRetryTime time.Duration // 默认 2s，建议设置成当前 API 的 PCT99
-	MaxRetryRate  float64       // 最大重试百分比，0.05 代表 5%，默认 5%，超出返回 ErrRetryQuotaExceeded
-	RetryCnt      int           // 最多发送的请求次数，默认 3 次，最少两次，其中一次是快速重试
-	RetryWaitTime time.Duration // 重试间隔，默认是 FastRetryTime / 10
-	Clock         Clock         // 模拟时钟，mock 用
+	FastRetryTime    time.Duration // 默认 2s，建议设置成当前 API 的 PCT99
+	MaxRetryRate     float64       // 最大重试百分比，0.05 代表 5%，默认 5%，超出返回 ErrRetryQuotaExceeded
+	RetryCnt         int           // 最多发送的请求次数，默认 3 次，最少两次，其中一次是快速重试
+	RetryWaitTime    time.Duration // 重试间隔，默认是 FastRetryTime / 10
+	Clock            Clock         // 模拟时钟，mock 用
+	MaxRetryCapacity int           // 平时没有超时的时候，允许积攒的重启请求最大数目，默认 1000 个
 }
 
 type Retry struct {
 	cfg           *Config
 	score         atomic.Int64
 	oneRetryScore int64
+	maxScore      int64
 }
 
 func New(config Config) *Retry {
@@ -60,13 +62,19 @@ func New(config Config) *Retry {
 		panic("bad MaxRetryRate")
 	}
 	r.oneRetryScore = int64(1 / r.cfg.MaxRetryRate)
+	if r.cfg.MaxRetryCapacity == 0 {
+		r.cfg.MaxRetryCapacity = 1000
+	}
+	r.maxScore = int64(r.cfg.MaxRetryCapacity) * r.oneRetryScore
 	return r
 }
 
 func (r *Retry) BackupRetry(ctx context.Context, retryableFunc func() (resp interface{}, err error)) (resp interface{}, err error) {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	r.score.Add(1)
+	if r.score.Load() < r.maxScore {
+		r.score.Add(1)
+	}
 	doFunc := func() retryableFuncResp {
 		resp, err := retryableFunc()
 		return retryableFuncResp{resp: resp, err: err}

retry_test.go

@@ -3,6 +3,7 @@ package fast_retry
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log"
 	"math/rand"
 	"os"
@@ -188,6 +189,71 @@ func TestMaxRetryRate(t *testing.T) {
 	require.True(t, callbackCnt.Load() < int64(float64(successCnt.Load())*1.2))
 }
 
+func TestMaxRetryCapacity(t *testing.T) {
+	if testing.Short() {
+		t.Skip()
+	}
+	ctx := context.Background()
+	var callbackCnt atomic.Int64
+	var errorCnt atomic.Int64
+	var successCnt atomic.Int64
+	// 模拟 FastRetryTime 设置不当，看请求是否放大两次
+	maxRetryCapacity := 100
+	r := New(Config{MaxRetryRate: 0.1, FastRetryTime: time.Second / 15, MaxRetryCapacity: maxRetryCapacity})
+	slow := false
+	fn := func() (resp interface{}, err error) {
+		callbackCnt.Inc()
+		if slow {
+			time.Sleep(time.Second / 10)
+		}
+		return "ok", nil
+	}
+	{
+		g, ctx := errgroup.WithContext(ctx)
+		for i := 0; i < 100; i++ {
+			g.Go(func() error {
+				for j := 0; j < 100; j++ {
+					_, err := r.BackupRetry(ctx, fn)
+					if err != nil {
+						fmt.Println(err)
+						errorCnt.Inc()
+					} else {
+						successCnt.Inc()
+					}
+				}
+				return nil
+			})
+		}
+		_ = g.Wait()
+	}
+
+	slow = true
+	{
+		g, ctx := errgroup.WithContext(ctx)
+		for i := 0; i < 50; i++ {
+			g.Go(func() error {
+				for j := 0; j < 100; j++ {
+					_, err := r.BackupRetry(ctx, fn)
+					if err != nil {
+						fmt.Println(err)
+						errorCnt.Inc()
+					} else {
+						successCnt.Inc()
+					}
+				}
+				return nil
+			})
+		}
+		_ = g.Wait()
+	}
+	// 先 10000 个快速请求，然后 5000 个慢请求
+	// 5000 里面会有 10% 被重试，之前 10000 会积累一部分 quota
+	t.Logf("%v %v %v", callbackCnt.Load(), errorCnt.Load(), successCnt.Load())
+	require.Equal(t, errorCnt.Load(), int64(0))
+	require.True(t, callbackCnt.Load() < int64(float64(successCnt.Load())*1.2))
+	require.True(t, callbackCnt.Load() <= int64(10000+5000+5000*0.1+maxRetryCapacity*2))
+}
+
 func BenchmarkRetry(b *testing.B) {
 	ctx := context.Background()
 	r := New(Config{MaxRetryRate: 0.1, FastRetryTime: time.Second / 100})