refactor: rename CoprRequestRateLimit to CoprRequestLimiter for consistency

Signed-off-by: Zhigao TONG <tongzhigao@pingcap.com>

Author: solotzg

pkg/distsql/distsql_test.go

@@ -129,37 +129,37 @@ func TestSelectAppliesQueryCopStoreLimiter(t *testing.T) {
 
 	request := buildRequest(kv.TiKV)
 	response, err := Select(checkRequest(func(req *kv.Request) {
-		require.Nil(t, req.CoprRequestRateLimit)
+		require.Nil(t, req.CoprRequestLimiter)
 		require.Same(t, dctx.QueryCopStoreLimiter, req.QueryCopStoreLimiter)
 	}), dctx, request, colTypes)
 	require.NoError(t, err)
 	require.NoError(t, response.Close())
 
 	request = buildRequest(kv.TiFlash)
 	response, err = Select(checkRequest(func(req *kv.Request) {
-		require.Nil(t, req.CoprRequestRateLimit)
+		require.Nil(t, req.CoprRequestLimiter)
 		require.Nil(t, req.QueryCopStoreLimiter)
 	}), dctx, request, colTypes)
 	require.NoError(t, err)
 	require.NoError(t, response.Close())
 
 	request = buildRequest(kv.TiKV)
-	explicitRateLimit := kv.NewCoprRequestRateLimit(7)
-	request.CoprRequestRateLimit = explicitRateLimit
+	explicitRateLimit := kv.NewCoprRequestLimiter(7)
+	request.CoprRequestLimiter = explicitRateLimit
 	response, err = Select(checkRequest(func(req *kv.Request) {
-		require.Same(t, explicitRateLimit, req.CoprRequestRateLimit)
+		require.Same(t, explicitRateLimit, req.CoprRequestLimiter)
 		require.Same(t, dctx.QueryCopStoreLimiter, req.QueryCopStoreLimiter)
-		require.False(t, req.CoprRequestRateLimit.Acquire(make(chan struct{})))
-		req.CoprRequestRateLimit.Release()
+		require.False(t, req.CoprRequestLimiter.Acquire(make(chan struct{})))
+		req.CoprRequestLimiter.Release()
 	}), dctx, request, colTypes)
 	require.NoError(t, err)
 	require.NoError(t, response.Close())
 
 	dctx.QueryCopStoreLimiter = nil
 	request = buildRequest(kv.TiKV)
-	request.CoprRequestRateLimit = explicitRateLimit
+	request.CoprRequestLimiter = explicitRateLimit
 	response, err = Select(checkRequest(func(req *kv.Request) {
-		require.Same(t, explicitRateLimit, req.CoprRequestRateLimit)
+		require.Same(t, explicitRateLimit, req.CoprRequestLimiter)
 		require.Nil(t, req.QueryCopStoreLimiter)
 	}), dctx, request, colTypes)
 	require.NoError(t, err)

pkg/distsql/request_builder.go

@@ -390,9 +390,9 @@ func (builder *RequestBuilder) SetConcurrency(concurrency int) *RequestBuilder {
 	return builder
 }
 
-// SetCoprRequestRateLimit sets a shared in-flight cop request limiter for this request.
-func (builder *RequestBuilder) SetCoprRequestRateLimit(rateLimit kv.CoprRequestLimiter) *RequestBuilder {
-	builder.Request.CoprRequestRateLimit = rateLimit
+// SetCoprRequestLimiter sets a shared in-flight cop request limiter for this request.
+func (builder *RequestBuilder) SetCoprRequestLimiter(rateLimit kv.CoprRequestLimiter) *RequestBuilder {
+	builder.Request.CoprRequestLimiter = rateLimit
 	return builder
 }
 

pkg/executor/distsql.go

@@ -935,7 +935,7 @@ func (e *IndexLookUpExecutor) startIndexWorker(ctx context.Context, initBatchSiz
 			return
 		}
 
-		sharedCoprRequestRateLimit := getMergeSortSharedCoprRequestRateLimit(needMerge, e.dctx.DistSQLConcurrency)
+		sharedCoprRequestLimiter := getMergeSortSharedCoprRequestLimiter(needMerge, e.dctx.DistSQLConcurrency)
 		mergeSortIndexScanConcurrency := getMergeSortIndexScanConcurrency(needMerge, len(kvRanges), e.dctx.DistSQLConcurrency)
 		results := make([]distsql.SelectResult, 0, len(kvRanges))
 		for idx := range kvRanges {
@@ -960,7 +960,7 @@ func (e *IndexLookUpExecutor) startIndexWorker(ctx context.Context, initBatchSiz
 				len(kvRanges),
 				worker.batchSize,
 				mergeSortIndexScanConcurrency,
-				sharedCoprRequestRateLimit,
+				sharedCoprRequestLimiter,
 			)
 			if err != nil {
 				for _, r := range results {
@@ -1021,17 +1021,14 @@ func getSelectResultInFlightCost(result distsql.SelectResult) int {
 	return inFlightCost
 }
 
-func getMergeSortSharedCoprRequestRateLimit(needMerge bool, distSQLConcurrency int) kv.CoprRequestLimiter {
+func getMergeSortSharedCoprRequestLimiter(needMerge bool, distSQLConcurrency int) kv.CoprRequestLimiter {
 	if !needMerge {
 		return nil
 	}
 	// Use a shared limiter to bound aggregate in-flight cop requests across
 	// all partitions in merge-sort mode.
-	capacity := distSQLConcurrency
-	if capacity < 1 {
-		capacity = 1
-	}
-	return kv.NewCoprRequestRateLimit(2 * capacity)
+	capacity := max(distSQLConcurrency, 1)
+	return kv.NewCoprRequestLimiter(2 * capacity)
 }
 
 func getMergeSortIndexScanConcurrency(needMerge bool, kvRangesCount int, distSQLConcurrency int) int {
@@ -1066,7 +1063,7 @@ func (e *IndexLookUpExecutor) buildIndexSelectResultForRange(
 	totalRanges int,
 	batchSize int,
 	indexScanConcurrency int,
-	sharedCoprRequestRateLimit kv.CoprRequestLimiter,
+	sharedCoprRequestLimiter kv.CoprRequestLimiter,
 ) (distsql.SelectResult, error) {
 	if tblScanIdxForRewritePartitionID >= 0 {
 		// We should set the TblScan's TableID to the partition physical ID to make sure
@@ -1092,7 +1089,7 @@ func (e *IndexLookUpExecutor) buildIndexSelectResultForRange(
 		SetClosestReplicaReadAdjuster(newClosestReadAdjuster(e.dctx, &builder.Request, e.idxNetDataSize/float64(totalRanges))).
 		SetMemTracker(tracker).
 		SetConnIDAndConnAlias(e.dctx.ConnectionID, e.dctx.SessionAlias).
-		SetCoprRequestRateLimit(sharedCoprRequestRateLimit)
+		SetCoprRequestLimiter(sharedCoprRequestLimiter)
 
 	if e.indexLookUpPushDown {
 		// Paging and Cop-cache is not supported in index lookup push down.

pkg/kv/kv.go

@@ -446,8 +446,8 @@ type coprRequestWaiter struct {
 	admitted bool
 }
 
-// NewCoprRequestRateLimit creates a cop request limiter with capacity n.
-func NewCoprRequestRateLimit(n int) CoprRequestLimiter {
+// NewCoprRequestLimiter creates a cop request limiter with capacity n.
+func NewCoprRequestLimiter(n int) CoprRequestLimiter {
 	if n <= 0 {
 		return nil
 	}
@@ -594,7 +594,7 @@ func (l *QueryCopStoreLimiter) GetStoreLimiter(storeID uint64) CoprRequestLimite
 	if limiter, ok := l.stores.Load(storeID); ok {
 		return limiter.(CoprRequestLimiter)
 	}
-	newLimiter := NewCoprRequestRateLimit(l.limit)
+	newLimiter := NewCoprRequestLimiter(l.limit)
 	limiter, _ := l.stores.LoadOrStore(storeID, newLimiter)
 	return limiter.(CoprRequestLimiter)
 }
@@ -789,10 +789,10 @@ type Request struct {
 	// ResponseIterator.Next is called. If concurrency is greater than 1, the request will be
 	// sent to multiple storage units concurrently.
 	Concurrency int
-	// CoprRequestRateLimit, if not nil, is used as the shared in-flight request
+	// CoprRequestLimiter, if not nil, is used as the shared in-flight request
 	// limiter for all cop iterators created from this request. The token lifecycle
 	// is tied to request send/response receive instead of result consumption.
-	CoprRequestRateLimit CoprRequestLimiter
+	CoprRequestLimiter CoprRequestLimiter
 	// QueryCopStoreLimiter, if not nil, limits in-flight cop request attempts
 	// per TiKV store within this request's statement.
 	QueryCopStoreLimiter *QueryCopStoreLimiter

pkg/kv/kv_test.go

@@ -39,7 +39,7 @@ func genRandHex(length int) []byte {
 }
 
 func TestCoprRequestLimiterWaitsUntilRelease(t *testing.T) {
-	limiter := NewCoprRequestRateLimit(1)
+	limiter := NewCoprRequestLimiter(1)
 	done := make(chan struct{})
 	require.False(t, limiter.Acquire(done))
 
@@ -70,7 +70,7 @@ func TestCoprRequestLimiterWaitsUntilRelease(t *testing.T) {
 }
 
 func TestCoprRequestLimiterAcquireCanBeCanceled(t *testing.T) {
-	limiter := NewCoprRequestRateLimit(1)
+	limiter := NewCoprRequestLimiter(1)
 	require.False(t, limiter.Acquire(make(chan struct{})))
 
 	done := make(chan struct{})
@@ -94,15 +94,15 @@ func TestCoprRequestLimiterAcquireCanBeCanceled(t *testing.T) {
 }
 
 func TestCoprRequestLimiterRedundantReleasePanics(t *testing.T) {
-	limiter := NewCoprRequestRateLimit(1)
+	limiter := NewCoprRequestLimiter(1)
 	require.Panics(t, func() {
 		limiter.Release()
 	})
 }
 
 func TestCoprRequestLimiterConcurrentAcquireRelease(t *testing.T) {
 	const capacity = int64(3)
-	limiter := NewCoprRequestRateLimit(int(capacity))
+	limiter := NewCoprRequestLimiter(int(capacity))
 	done := make(chan struct{})
 	var active atomic.Int64
 	var maxActive atomic.Int64

pkg/store/copr/copr_test/coprocessor_test.go

@@ -139,7 +139,7 @@ func TestBuildCopIteratorWithRowCountHint(t *testing.T) {
 	require.Equal(t, rateLimit.GetCapacity(), 4)
 }
 
-func TestBuildCopIteratorWithSharedRequestRateLimit(t *testing.T) {
+func TestBuildCopIteratorWithSharedRequestLimiter(t *testing.T) {
 	store, err := mockstore.NewMockStore()
 	require.NoError(t, err)
 	defer require.NoError(t, store.Close())
@@ -161,18 +161,18 @@ func TestBuildCopIteratorWithSharedRequestRateLimit(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			shared := kv.NewCoprRequestRateLimit(7)
+			shared := kv.NewCoprRequestLimiter(7)
 			req := &kv.Request{
-				Tp:                   kv.ReqTypeDAG,
-				KeyRanges:            kv.NewNonPartitionedKeyRanges(ranges),
-				Concurrency:          15,
-				KeepOrder:            tc.keepOrder,
-				CoprRequestRateLimit: shared,
+				Tp:                 kv.ReqTypeDAG,
+				KeyRanges:          kv.NewNonPartitionedKeyRanges(ranges),
+				Concurrency:        15,
+				KeepOrder:          tc.keepOrder,
+				CoprRequestLimiter: shared,
 			}
 			it, errRes := copClient.BuildCopIterator(ctx, req, vars, opt)
 			require.Nil(t, errRes)
-			require.Same(t, shared, it.GetRequestRateLimit())
-			require.Equal(t, 7, it.GetRequestRateLimit().Capacity())
+			require.Same(t, shared, it.GetRequestLimiter())
+			require.Equal(t, 7, it.GetRequestLimiter().Capacity())
 		})
 	}
 }

pkg/store/copr/coprocessor.go

@@ -1348,9 +1348,9 @@ func (it *copIterator) GetSendRate() *util.RateLimit {
 	return it.sendRate
 }
 
-// GetRequestRateLimit returns the shared request rate-limit object.
-func (it *copIterator) GetRequestRateLimit() kv.CoprRequestLimiter {
-	return it.req.CoprRequestRateLimit
+// GetRequestLimiter returns the shared request rate-limit object.
+func (it *copIterator) GetRequestLimiter() kv.CoprRequestLimiter {
+	return it.req.CoprRequestLimiter
 }
 
 // GetTasks returns the built tasks.
@@ -1814,7 +1814,7 @@ func (worker *copIteratorWorker) handleTaskOnce(bo *Backoffer, task *copTask) (*
 	if exit {
 		return nil, nil
 	}
-	releaseRequestRateLimit, exit := acquireCoprRequestLimiter(worker.req.CoprRequestRateLimit, worker.finishCh)
+	releaseCoprRequestLimiter, exit := acquireCoprRequestLimiter(worker.req.CoprRequestLimiter, worker.finishCh)
 	if exit {
 		if releaseQueryCopStoreLimiter != nil {
 			releaseQueryCopStoreLimiter()
@@ -1826,8 +1826,8 @@ func (worker *copIteratorWorker) handleTaskOnce(bo *Backoffer, task *copTask) (*
 	// remaining panic-safe.
 	resp, rpcCtx, storeAddr, err := func() (*tikvrpc.Response, *tikv.RPCContext, string, error) {
 		defer func() {
-			if releaseRequestRateLimit != nil {
-				releaseRequestRateLimit()
+			if releaseCoprRequestLimiter != nil {
+				releaseCoprRequestLimiter()
 			}
 			if releaseQueryCopStoreLimiter != nil {
 				releaseQueryCopStoreLimiter()