vector: Merge upstream

Makefile

@@ -627,7 +627,7 @@ bazel_coverage_test: failpoint-enable bazel_ci_simple_prepare
 bazel_build:
 	mkdir -p bin
 	bazel $(BAZEL_GLOBAL_CONFIG) build $(BAZEL_CMD_CONFIG) \
-		//... --//build:with_nogo_flag=$(NOGO_FLAG)
+		//... --//build:with_nogo_flag=$(NOGO_FLAG) --subcommands
 	bazel $(BAZEL_GLOBAL_CONFIG) build $(BAZEL_CMD_CONFIG) \
 		//cmd/importer:importer //cmd/tidb-server:tidb-server //cmd/tidb-server:tidb-server-check --//build:with_nogo_flag=$(NOGO_FLAG)
 	cp bazel-out/k8-fastbuild/bin/cmd/tidb-server/tidb-server_/tidb-server ./bin

OWNERS

@@ -43,7 +43,6 @@ approvers:
   - guo-shaoge
   - hanfei1991
   - hawkingrei
-  - hi-rustin
   - hicqu
   - holys
   - hongyunyan
@@ -80,6 +79,7 @@ approvers:
   - qw4990
   - rebelice
   - Reminiscent
+  - Rustin170506
   - sdojjy
   - shenli
   - siddontang

WORKSPACE

@@ -12,9 +12,9 @@ http_archive(
 
 http_archive(
     name = "bazel_features",
-    sha256 = "d7787da289a7fb497352211ad200ec9f698822a9e0757a4976fd9f713ff372b3",
-    strip_prefix = "bazel_features-1.9.1",
-    url = "https://github.com/bazel-contrib/bazel_features/releases/download/v1.9.1/bazel_features-v1.9.1.tar.gz",
+    sha256 = "ba1282c1aa1d1fffdcf994ab32131d7c7551a9bc960fbf05f42d55a1b930cbfb",
+    strip_prefix = "bazel_features-1.15.0",
+    url = "https://github.com/bazel-contrib/bazel_features/releases/download/v1.15.0/bazel_features-v1.15.0.tar.gz",
 )
 
 load("@bazel_features//:deps.bzl", "bazel_features_deps")
@@ -48,12 +48,11 @@ http_archive(
 
 http_archive(
     name = "bazel_gazelle",
-    sha256 = "d76bf7a60fd8b050444090dfa2837a4eaf9829e1165618ee35dceca5cbdf58d5",
+    sha256 = "8ad77552825b078a10ad960bec6ef77d2ff8ec70faef2fd038db713f410f5d87",
     urls = [
-        "http://bazel-cache.pingcap.net:8080/bazelbuild/bazel-gazelle/releases/download/v0.37.0/bazel-gazelle-v0.37.0.tar.gz",
-        "https://mirror.bazel.build/github.com/bazelbuild/bazel-gazelle/releases/download/v0.37.0/bazel-gazelle-v0.37.0.tar.gz",
-        "https://github.com/bazelbuild/bazel-gazelle/releases/download/v0.37.0/bazel-gazelle-v0.37.0.tar.gz",
-        "http://ats.apps.svc/bazelbuild/bazel-gazelle/releases/download/v0.37.0/bazel-gazelle-v0.37.0.tar.gz",
+        "http://bazel-cache.pingcap.net:8080/bazelbuild/bazel-gazelle/releases/download/v0.38.0/bazel-gazelle-v0.38.0.tar.gz",
+        "https://github.com/bazelbuild/bazel-gazelle/releases/download/v0.38.0/bazel-gazelle-v0.38.0.tar.gz",
+        "http://ats.apps.svc/bazelbuild/bazel-gazelle/releases/download/v0.38.0/bazel-gazelle-v0.38.0.tar.gz",
     ],
 )
 
@@ -86,7 +85,7 @@ go_download_sdk(
         "https://mirrors.aliyun.com/golang/{}",
         "https://dl.google.com/go/{}",
     ],
-    version = "1.21.12",
+    version = "1.21.13",
 )
 
 go_register_toolchains(

br/cmd/br/BUILD.bazel

@@ -37,9 +37,11 @@ go_library(
         "//pkg/util",
         "//pkg/util/gctuner",
         "//pkg/util/logutil",
+        "//pkg/util/mathutil",
         "//pkg/util/memory",
         "//pkg/util/metricsutil",
         "//pkg/util/redact",
+        "//pkg/util/size",
         "@com_github_gogo_protobuf//proto",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_kvproto//pkg/brpb",
@@ -63,5 +65,8 @@ go_test(
     srcs = ["main_test.go"],
     embed = [":br_lib"],
     flaky = True,
-    deps = ["@org_uber_go_goleak//:goleak"],
+    deps = [
+        "@com_github_stretchr_testify//require",
+        "@org_uber_go_goleak//:goleak",
+    ],
 )

br/cmd/br/cmd.go

@@ -5,8 +5,10 @@ package main
 import (
 	"context"
 	"fmt"
+	"math"
 	"os"
 	"path/filepath"
+	"runtime/debug"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -21,9 +23,12 @@ import (
 	"github.com/pingcap/tidb/pkg/config"
 	tidbutils "github.com/pingcap/tidb/pkg/util"
 	"github.com/pingcap/tidb/pkg/util/logutil"
+	"github.com/pingcap/tidb/pkg/util/mathutil"
 	"github.com/pingcap/tidb/pkg/util/memory"
 	"github.com/pingcap/tidb/pkg/util/redact"
+	"github.com/pingcap/tidb/pkg/util/size"
 	"github.com/spf13/cobra"
+	"go.uber.org/zap"
 )
 
 var (
@@ -107,6 +112,25 @@ func AddFlags(cmd *cobra.Command) {
 	_ = cmd.PersistentFlags().MarkHidden(FlagRedactLog)
 }
 
+const quarterGiB uint64 = 256 * size.MB
+const halfGiB uint64 = 512 * size.MB
+const fourGiB uint64 = 4 * size.GB
+
+func calculateMemoryLimit(memleft uint64) uint64 {
+	// memreserved = f(memleft) = 512MB * memleft / (memleft + 4GB)
+	//  * f(0) = 0
+	//  * f(4GB) = 256MB
+	//  * f(+inf) -> 512MB
+	memreserved := halfGiB / (1 + fourGiB/(memleft|1))
+	// 0     memused          memtotal-memreserved  memtotal
+	// +--------+--------------------+----------------+
+	//          ^            br mem upper limit
+	//          +--------------------^
+	//             GOMEMLIMIT range
+	memlimit := memleft - memreserved
+	return memlimit
+}
+
 // Init initializes BR cli.
 func Init(cmd *cobra.Command) (err error) {
 	initOnce.Do(func() {
@@ -162,6 +186,34 @@ func Init(cmd *cobra.Command) (err error) {
 		}
 		log.ReplaceGlobals(lg, p)
 		memory.InitMemoryHook()
+		if debug.SetMemoryLimit(-1) == math.MaxInt64 {
+			memtotal, e := memory.MemTotal()
+			if e != nil {
+				err = e
+				return
+			}
+			memused, e := memory.MemUsed()
+			if e != nil {
+				err = e
+				return
+			}
+			if memused >= memtotal {
+				log.Warn("failed to obtain memory size, skip setting memory limit",
+					zap.Uint64("memused", memused), zap.Uint64("memtotal", memtotal))
+			} else {
+				memleft := memtotal - memused
+				memlimit := calculateMemoryLimit(memleft)
+				// BR command needs 256 MiB at least, if the left memory is less than 256 MiB,
+				// the memory limit cannot limit anyway and then finally OOM.
+				memlimit = mathutil.Max(memlimit, quarterGiB)
+				log.Info("calculate the rest memory",
+					zap.Uint64("memtotal", memtotal), zap.Uint64("memused", memused), zap.Uint64("memlimit", memlimit))
+				// No need to set memory limit because the left memory is sufficient.
+				if memlimit < uint64(math.MaxInt64) {
+					debug.SetMemoryLimit(int64(memlimit))
+				}
+			}
+		}
 
 		redactLog, e := cmd.Flags().GetBool(FlagRedactLog)
 		if e != nil {

br/cmd/br/main_test.go

@@ -21,6 +21,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
 )
 
@@ -75,3 +76,18 @@ func TestRunMain(*testing.T) {
 
 	<-waitCh
 }
+
+func TestCalculateMemoryLimit(t *testing.T) {
+	// f(0 Byte) = 0 Byte
+	require.Equal(t, uint64(0), calculateMemoryLimit(0))
+	// f(100 KB) = 87.5 KB
+	require.Equal(t, uint64(89600), calculateMemoryLimit(100*1024))
+	// f(100 MB) = 87.5 MB
+	require.Equal(t, uint64(91763188), calculateMemoryLimit(100*1024*1024))
+	// f(3.99 GB) = 3.74 GB
+	require.Equal(t, uint64(4026531839), calculateMemoryLimit(4*1024*1024*1024-1))
+	// f(4 GB) = 3.5 GB
+	require.Equal(t, uint64(3758096384), calculateMemoryLimit(4*1024*1024*1024))
+	// f(32 GB) = 31.5 GB
+	require.Equal(t, uint64(33822867456), calculateMemoryLimit(32*1024*1024*1024))
+}

br/pkg/backup/BUILD.bazel

@@ -63,11 +63,12 @@ go_test(
         "client_test.go",
         "main_test.go",
         "schema_test.go",
+        "store_test.go",
     ],
     embed = [":backup"],
     flaky = True,
     race = "on",
-    shard_count = 12,
+    shard_count = 14,
     deps = [
         "//br/pkg/conn",
         "//br/pkg/gluetidb/mock",
@@ -92,6 +93,7 @@ go_test(
         "@com_github_tikv_client_go_v2//testutils",
         "@com_github_tikv_pd_client//:client",
         "@io_opencensus_go//stats/view",
+        "@org_golang_google_grpc//:grpc",
         "@org_uber_go_goleak//:goleak",
     ],
 )

br/pkg/backup/client.go

@@ -1233,9 +1233,9 @@ func (bc *Client) OnBackupResponse(
 				return txnlock.NewLock(lockErr), nil
 			}
 		}
-		res := errContext.HandleIgnorableError(errPb, storeID)
+		res := utils.HandleBackupError(errPb, storeID, errContext)
 		switch res.Strategy {
-		case utils.GiveUpStrategy:
+		case utils.StrategyGiveUp:
 			errMsg := res.Reason
 			if len(errMsg) <= 0 {
 				errMsg = errPb.Msg

br/pkg/backup/store.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"io"
 	"os"
+	"sync"
 	"time"
 
 	"github.com/pingcap/errors"
@@ -57,12 +58,73 @@ func (r ResponseAndStore) GetStoreID() uint64 {
 	return r.StoreID
 }
 
+// timeoutRecv cancel the context if `Refresh()` is not called within the specified time `timeout`.
+type timeoutRecv struct {
+	wg        sync.WaitGroup
+	parentCtx context.Context
+	cancel    context.CancelCauseFunc
+
+	refresh chan struct{}
+}
+
+// Refresh the timeout ticker
+func (trecv *timeoutRecv) Refresh() {
+	select {
+	case <-trecv.parentCtx.Done():
+	case trecv.refresh <- struct{}{}:
+	}
+}
+
+// Stop the timeout ticker
+func (trecv *timeoutRecv) Stop() {
+	close(trecv.refresh)
+	trecv.wg.Wait()
+}
+
+var TimeoutOneResponse = time.Hour
+
+func (trecv *timeoutRecv) loop(timeout time.Duration) {
+	defer trecv.wg.Done()
+	ticker := time.NewTicker(timeout)
+	defer ticker.Stop()
+	for {
+		ticker.Reset(timeout)
+		select {
+		case <-trecv.parentCtx.Done():
+			return
+		case _, ok := <-trecv.refresh:
+			if !ok {
+				return
+			}
+		case <-ticker.C:
+			log.Warn("receive a backup response timeout")
+			trecv.cancel(errors.Errorf("receive a backup response timeout"))
+		}
+	}
+}
+
+func StartTimeoutRecv(ctx context.Context, timeout time.Duration) (context.Context, *timeoutRecv) {
+	cctx, cancel := context.WithCancelCause(ctx)
+	trecv := &timeoutRecv{
+		parentCtx: ctx,
+		cancel:    cancel,
+		refresh:   make(chan struct{}),
+	}
+	trecv.wg.Add(1)
+	go trecv.loop(timeout)
+	return cctx, trecv
+}
+
 func doSendBackup(
-	ctx context.Context,
+	pctx context.Context,
 	client backuppb.BackupClient,
 	req backuppb.BackupRequest,
 	respFn func(*backuppb.BackupResponse) error,
 ) error {
+	// Backup might be stuck on GRPC `waitonHeader`, so start a timeout ticker to
+	// terminate the backup if it does not receive any new response for a long time.
+	ctx, timerecv := StartTimeoutRecv(pctx, TimeoutOneResponse)
+	defer timerecv.Stop()
 	failpoint.Inject("hint-backup-start", func(v failpoint.Value) {
 		logutil.CL(ctx).Info("failpoint hint-backup-start injected, " +
 			"process will notify the shell.")
@@ -107,6 +169,7 @@ func doSendBackup(
 
 	for {
 		resp, err := bCli.Recv()
+		timerecv.Refresh()
 		if err != nil {
 			if errors.Cause(err) == io.EOF { // nolint:errorlint
 				logutil.CL(ctx).Debug("backup streaming finish",