merge master for cse region client

internal/retry/backoff.go → config/retry/backoff.go

@@ -143,7 +143,13 @@ func (b *Backoffer) BackoffWithCfgAndMaxSleep(cfg *Config, maxSleepMs int, err e
 	if b.noop {
 		return err
 	}
-	if b.maxSleep > 0 && (b.totalSleep-b.excludedSleep) >= b.maxSleep {
+	maxBackoffTimeExceeded := (b.totalSleep - b.excludedSleep) >= b.maxSleep
+	maxExcludedTimeExceeded := false
+	if maxLimit, ok := isSleepExcluded[cfg.name]; ok {
+		maxExcludedTimeExceeded = b.excludedSleep >= maxLimit && b.excludedSleep >= b.maxSleep
+	}
+	maxTimeExceeded := maxBackoffTimeExceeded || maxExcludedTimeExceeded
+	if b.maxSleep > 0 && maxTimeExceeded {
 		longestSleepCfg, longestSleepTime := b.longestSleepCfg()
 		errMsg := fmt.Sprintf("%s backoffer.maxSleep %dms is exceeded, errors:", cfg.String(), b.maxSleep)
 		for i, err := range b.errors {
@@ -163,7 +169,8 @@ func (b *Backoffer) BackoffWithCfgAndMaxSleep(cfg *Config, maxSleepMs int, err e
 			backoffDetail.WriteString(":")
 			backoffDetail.WriteString(strconv.Itoa(times))
 		}
-		errMsg += fmt.Sprintf("\ntotal-backoff-times: %v, backoff-detail: %v", totalTimes, backoffDetail.String())
+		errMsg += fmt.Sprintf("\ntotal-backoff-times: %v, backoff-detail: %v, maxBackoffTimeExceeded: %v, maxExcludedTimeExceeded: %v",
+			totalTimes, backoffDetail.String(), maxBackoffTimeExceeded, maxExcludedTimeExceeded)
 		returnedErr := err
 		if longestSleepCfg != nil {
 			errMsg += fmt.Sprintf("\nlongest sleep type: %s, time: %dms", longestSleepCfg.String(), longestSleepTime)

internal/retry/backoff_test.go → config/retry/backoff_test.go

@@ -71,7 +71,7 @@ func TestBackoffErrorType(t *testing.T) {
 	err = b.Backoff(BoIsWitness, errors.New("peer is witness"))
 	assert.Nil(t, err)
 	// wait it exceed max sleep
-	for i := 0; i < 10; i++ {
+	for i := 0; i < 15; i++ {
 		err = b.Backoff(BoTxnNotFound, errors.New("txn not found"))
 		if err != nil {
 			// Next backoff should return error of backoff that sleeps for longest time.
@@ -98,3 +98,15 @@ func TestBackoffDeepCopy(t *testing.T) {
 		assert.ErrorIs(t, err, BoMaxDataNotReady.err)
 	}
 }
+
+func TestBackoffWithMaxExcludedExceed(t *testing.T) {
+	setBackoffExcluded(BoTiKVServerBusy.name, 1)
+	b := NewBackofferWithVars(context.TODO(), 1, nil)
+	err := b.Backoff(BoTiKVServerBusy, errors.New("server is busy"))
+	assert.Nil(t, err)
+
+	// As the total excluded sleep is greater than the max limited value, error should be returned.
+	err = b.Backoff(BoTiKVServerBusy, errors.New("server is busy"))
+	assert.NotNil(t, err)
+	assert.Greater(t, b.excludedSleep, b.maxSleep)
+}

config/retry/config.go

@@ -0,0 +1,207 @@
+// Copyright 2021 TiKV Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// NOTE: The code in this file is based on code from the
+// TiDB project, licensed under the Apache License v 2.0
+//
+// https://github.com/pingcap/tidb/tree/cc5e161ac06827589c4966674597c137cc9e809c/store/tikv/retry/config.go
+//
+
+// Copyright 2021 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package retry
+
+import (
+	"context"
+	"math"
+	"math/rand"
+	"strings"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	tikverr "github.com/tikv/client-go/v2/error"
+	"github.com/tikv/client-go/v2/internal/logutil"
+	"github.com/tikv/client-go/v2/kv"
+	"github.com/tikv/client-go/v2/metrics"
+	"go.uber.org/zap"
+)
+
+// Config is the configuration of the Backoff function.
+type Config struct {
+	name   string
+	metric *prometheus.Observer
+	fnCfg  *BackoffFnCfg
+	err    error
+}
+
+// backoffFn is the backoff function which compute the sleep time and do sleep.
+type backoffFn func(ctx context.Context, maxSleepMs int) int
+
+func (c *Config) createBackoffFn(vars *kv.Variables) backoffFn {
+	if strings.EqualFold(c.name, txnLockFastName) {
+		return newBackoffFn(vars.BackoffLockFast, c.fnCfg.cap, c.fnCfg.jitter)
+	}
+	return newBackoffFn(c.fnCfg.base, c.fnCfg.cap, c.fnCfg.jitter)
+}
+
+// BackoffFnCfg is the configuration for the backoff func which implements exponential backoff with
+// optional jitters.
+// See http://www.awsarchitectureblog.com/2015/03/backoff.html
+type BackoffFnCfg struct {
+	base   int
+	cap    int
+	jitter int
+}
+
+// NewBackoffFnCfg creates the config for BackoffFn.
+func NewBackoffFnCfg(base, cap, jitter int) *BackoffFnCfg {
+	return &BackoffFnCfg{
+		base,
+		cap,
+		jitter,
+	}
+}
+
+// NewConfig creates a new Config for the Backoff operation.
+func NewConfig(name string, metric *prometheus.Observer, backoffFnCfg *BackoffFnCfg, err error) *Config {
+	return &Config{
+		name:   name,
+		metric: metric,
+		fnCfg:  backoffFnCfg,
+		err:    err,
+	}
+}
+
+func (c *Config) String() string {
+	return c.name
+}
+
+// SetErrors sets a more detailed error instead of the default bo config.
+func (c *Config) SetErrors(err error) {
+	c.err = err
+}
+
+func (c *Config) SetBackoffFnCfg(fnCfg *BackoffFnCfg) {
+	c.fnCfg = fnCfg
+}
+
+const txnLockFastName = "txnLockFast"
+
+// Backoff Config variables.
+var (
+	// TODO: distinguish tikv and tiflash in metrics
+	BoTiKVRPC    = NewConfig("tikvRPC", &metrics.BackoffHistogramRPC, NewBackoffFnCfg(100, 2000, EqualJitter), tikverr.ErrTiKVServerTimeout)
+	BoTiFlashRPC = NewConfig("tiflashRPC", &metrics.BackoffHistogramRPC, NewBackoffFnCfg(100, 2000, EqualJitter), tikverr.ErrTiFlashServerTimeout)
+	BoTxnLock    = NewConfig("txnLock", &metrics.BackoffHistogramLock, NewBackoffFnCfg(100, 3000, EqualJitter), tikverr.ErrResolveLockTimeout)
+	BoPDRPC      = NewConfig("pdRPC", &metrics.BackoffHistogramPD, NewBackoffFnCfg(500, 3000, EqualJitter), tikverr.NewErrPDServerTimeout(""))
+	// change base time to 2ms, because it may recover soon.
+	BoRegionMiss               = NewConfig("regionMiss", &metrics.BackoffHistogramRegionMiss, NewBackoffFnCfg(2, 500, NoJitter), tikverr.ErrRegionUnavailable)
+	BoRegionScheduling         = NewConfig("regionScheduling", &metrics.BackoffHistogramRegionScheduling, NewBackoffFnCfg(2, 500, NoJitter), tikverr.ErrRegionUnavailable)
+	BoTiKVServerBusy           = NewConfig("tikvServerBusy", &metrics.BackoffHistogramServerBusy, NewBackoffFnCfg(2000, 10000, EqualJitter), tikverr.ErrTiKVServerBusy)
+	BoTiKVDiskFull             = NewConfig("tikvDiskFull", &metrics.BackoffHistogramTiKVDiskFull, NewBackoffFnCfg(500, 5000, NoJitter), tikverr.ErrTiKVDiskFull)
+	BoRegionRecoveryInProgress = NewConfig("regionRecoveryInProgress", &metrics.BackoffHistogramRegionRecoveryInProgress, NewBackoffFnCfg(100, 10000, EqualJitter), tikverr.ErrRegionRecoveryInProgress)
+	BoTiFlashServerBusy        = NewConfig("tiflashServerBusy", &metrics.BackoffHistogramServerBusy, NewBackoffFnCfg(2000, 10000, EqualJitter), tikverr.ErrTiFlashServerBusy)
+	BoTxnNotFound              = NewConfig("txnNotFound", &metrics.BackoffHistogramEmpty, NewBackoffFnCfg(2, 500, NoJitter), tikverr.ErrResolveLockTimeout)
+	BoStaleCmd                 = NewConfig("staleCommand", &metrics.BackoffHistogramStaleCmd, NewBackoffFnCfg(2, 1000, NoJitter), tikverr.ErrTiKVStaleCommand)
+	BoMaxTsNotSynced           = NewConfig("maxTsNotSynced", &metrics.BackoffHistogramEmpty, NewBackoffFnCfg(2, 500, NoJitter), tikverr.ErrTiKVMaxTimestampNotSynced)
+	BoMaxDataNotReady          = NewConfig("dataNotReady", &metrics.BackoffHistogramDataNotReady, NewBackoffFnCfg(2, 2000, NoJitter), tikverr.ErrRegionDataNotReady)
+	BoMaxRegionNotInitialized  = NewConfig("regionNotInitialized", &metrics.BackoffHistogramEmpty, NewBackoffFnCfg(2, 1000, NoJitter), tikverr.ErrRegionNotInitialized)
+	BoIsWitness                = NewConfig("isWitness", &metrics.BackoffHistogramIsWitness, NewBackoffFnCfg(1000, 10000, EqualJitter), tikverr.ErrIsWitness)
+	// TxnLockFast's `base` load from vars.BackoffLockFast when create BackoffFn.
+	BoTxnLockFast = NewConfig(txnLockFastName, &metrics.BackoffHistogramLockFast, NewBackoffFnCfg(2, 3000, EqualJitter), tikverr.ErrResolveLockTimeout)
+)
+
+var isSleepExcluded = map[string]int{
+	BoTiKVServerBusy.name: 600000, // The max excluded limit is 10min.
+	// add BoTiFlashServerBusy if appropriate
+}
+
+// setBackoffExcluded is used for test only.
+func setBackoffExcluded(name string, maxVal int) {
+	if _, ok := isSleepExcluded[name]; ok {
+		isSleepExcluded[name] = maxVal
+	}
+}
+
+const (
+	// NoJitter makes the backoff sequence strict exponential.
+	NoJitter = 1 + iota
+	// FullJitter applies random factors to strict exponential.
+	FullJitter
+	// EqualJitter is also randomized, but prevents very short sleeps.
+	EqualJitter
+	// DecorrJitter increases the maximum jitter based on the last random value.
+	DecorrJitter
+)
+
+// newBackoffFn creates a backoff func which implements exponential backoff with
+// optional jitters.
+// See http://www.awsarchitectureblog.com/2015/03/backoff.html
+func newBackoffFn(base, cap, jitter int) backoffFn {
+	if base < 2 {
+		// Top prevent panic in 'rand.Intn'.
+		base = 2
+	}
+	attempts := 0
+	lastSleep := base
+	return func(ctx context.Context, maxSleepMs int) int {
+		var sleep int
+		switch jitter {
+		case NoJitter:
+			sleep = expo(base, cap, attempts)
+		case FullJitter:
+			v := expo(base, cap, attempts)
+			sleep = rand.Intn(v)
+		case EqualJitter:
+			v := expo(base, cap, attempts)
+			sleep = v/2 + rand.Intn(v/2)
+		case DecorrJitter:
+			sleep = int(math.Min(float64(cap), float64(base+rand.Intn(lastSleep*3-base))))
+		}
+		logutil.BgLogger().Debug("backoff",
+			zap.Int("base", base),
+			zap.Int("sleep", sleep),
+			zap.Int("attempts", attempts))
+
+		realSleep := sleep
+		// when set maxSleepMs >= 0 in `tikv.BackoffWithMaxSleep` will force sleep maxSleepMs milliseconds.
+		if maxSleepMs >= 0 && realSleep > maxSleepMs {
+			realSleep = maxSleepMs
+		}
+		select {
+		case <-time.After(time.Duration(realSleep) * time.Millisecond):
+			attempts++
+			lastSleep = sleep
+			return realSleep
+		case <-ctx.Done():
+			return 0
+		}
+	}
+}
+
+func expo(base, cap, n int) int {
+	return int(math.Min(float64(cap), float64(base)*math.Pow(2.0, float64(n))))
+}

examples/gcworker/go.mod

@@ -1,7 +1,54 @@
 module gcworker
 
-go 1.20
+go 1.21
 
 require github.com/tikv/client-go/v2 v2.0.0
 
+require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cloudfoundry/gosigar v1.3.6 // indirect
+	github.com/coreos/go-semver v0.3.1 // indirect
+	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
+	github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548 // indirect
+	github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/google/btree v1.1.2 // indirect
+	github.com/google/uuid v1.3.1 // indirect
+	github.com/grpc-ecosystem/go-grpc-middleware v1.1.0 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
+	github.com/opentracing/opentracing-go v1.2.0 // indirect
+	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
+	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
+	github.com/pingcap/kvproto v0.0.0-20230904082117-ecdbf1f8c130 // indirect
+	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/prometheus/client_golang v1.15.1 // indirect
+	github.com/prometheus/client_model v0.3.0 // indirect
+	github.com/prometheus/common v0.42.0 // indirect
+	github.com/prometheus/procfs v0.9.0 // indirect
+	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
+	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
+	github.com/tikv/pd/client v0.0.0-20230724080549-de985b8e0afc // indirect
+	github.com/twmb/murmur3 v1.1.3 // indirect
+	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
+	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
+	go.etcd.io/etcd/client/v3 v3.5.10 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.26.0 // indirect
+	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
+	golang.org/x/net v0.17.0 // indirect
+	golang.org/x/sync v0.3.0 // indirect
+	golang.org/x/sys v0.13.0 // indirect
+	golang.org/x/text v0.13.0 // indirect
+	google.golang.org/genproto v0.0.0-20231016165738-49dd2c1f3d0b // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20231016165738-49dd2c1f3d0b // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b // indirect
+	google.golang.org/grpc v1.59.0 // indirect
+	google.golang.org/protobuf v1.31.0 // indirect
+	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
+)
+
 replace github.com/tikv/client-go/v2 => ../../

examples/rawkv/go.mod

@@ -1,7 +1,54 @@
 module rawkv
 
-go 1.20
+go 1.21
 
 require github.com/tikv/client-go/v2 v2.0.0
 
+require (
+	github.com/benbjohnson/clock v1.3.0 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cloudfoundry/gosigar v1.3.6 // indirect
+	github.com/coreos/go-semver v0.3.1 // indirect
+	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
+	github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548 // indirect
+	github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/google/btree v1.1.2 // indirect
+	github.com/google/uuid v1.3.1 // indirect
+	github.com/grpc-ecosystem/go-grpc-middleware v1.1.0 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
+	github.com/opentracing/opentracing-go v1.2.0 // indirect
+	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
+	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
+	github.com/pingcap/kvproto v0.0.0-20230904082117-ecdbf1f8c130 // indirect
+	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/prometheus/client_golang v1.15.1 // indirect
+	github.com/prometheus/client_model v0.3.0 // indirect
+	github.com/prometheus/common v0.42.0 // indirect
+	github.com/prometheus/procfs v0.9.0 // indirect
+	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
+	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
+	github.com/tikv/pd/client v0.0.0-20230724080549-de985b8e0afc // indirect
+	github.com/twmb/murmur3 v1.1.3 // indirect
+	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
+	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
+	go.etcd.io/etcd/client/v3 v3.5.10 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.26.0 // indirect
+	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
+	golang.org/x/net v0.17.0 // indirect
+	golang.org/x/sync v0.3.0 // indirect
+	golang.org/x/sys v0.13.0 // indirect
+	golang.org/x/text v0.13.0 // indirect
+	google.golang.org/genproto v0.0.0-20231016165738-49dd2c1f3d0b // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b // indirect
+	google.golang.org/grpc v1.59.0 // indirect
+	google.golang.org/protobuf v1.31.0 // indirect
+	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
+)
+
 replace github.com/tikv/client-go/v2 => ../../