store/copr: set upper limit for extra concurrency

store/copr/coprocessor.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"math"
+	"runtime"
 	"strconv"
 	"strings"
 	"sync"
@@ -72,6 +73,8 @@ const (
 	smallTaskSigma         = 0.5
 )
 
+var smallTaskConcurrencyLimit = 20 * runtime.NumCPU()
+
 // CopClient is coprocessor client.
 type CopClient struct {
 	kv.RequestTypeSupportedChecker
@@ -592,8 +595,11 @@ func smallTaskConcurrency(tasks []*copTask) (int, int) {
 	}
 	// Calculate the extra concurrency for small tasks
 	// extra concurrency = tasks / (1 + sigma * sqrt(log(tasks ^ 2)))
-	extraConc := float64(res) / (1 + smallTaskSigma*math.Sqrt(2*math.Log(float64(res))))
-	return res, int(extraConc)
+	extraConc := int(float64(res) / (1 + smallTaskSigma*math.Sqrt(2*math.Log(float64(res)))))
+	if extraConc > smallTaskConcurrencyLimit {
+		extraConc = smallTaskConcurrencyLimit
+	}
+	return res, extraConc
 }
 
 type copIterator struct {

store/copr/coprocessor_test.go

@@ -774,3 +774,21 @@ func TestBuildCopTasksWithRowCountHint(t *testing.T) {
 	// task[3] ["t"-"z"]
 	require.Equal(t, tasks[3].RowCountHint, 10)
 }
+
+func TestSmallTaskConcurrency(t *testing.T) {
+	originSmallTaskLimit := smallTaskConcurrencyLimit
+	defer func() {
+		smallTaskConcurrencyLimit = originSmallTaskLimit
+	}()
+	smallTaskConcurrencyLimit = 10
+	smallTaskCount := 1000
+	tasks := make([]*copTask, 0, smallTaskCount)
+	for i := 0; i < smallTaskCount; i++ {
+		tasks = append(tasks, &copTask{
+			RowCountHint: 1,
+		})
+	}
+	count, conc := smallTaskConcurrency(tasks)
+	require.Equal(t, smallTaskConcurrencyLimit, conc)
+	require.Equal(t, smallTaskCount, count)
+}