From ab0568e9ccf6d2d41bff9bda5d18db3340d4bddf Mon Sep 17 00:00:00 2001 From: D3Hunter Date: Fri, 3 Jul 2026 16:32:49 +0800 Subject: [PATCH 1/3] dxf, server: make max concurrent task configurable --- .../framework/integrationtests/bench_test.go | 22 +++++----- pkg/dxf/framework/proto/BUILD.bazel | 2 +- pkg/dxf/framework/proto/task.go | 42 ++++++++++++++++-- pkg/dxf/framework/proto/task_test.go | 17 +++++++ pkg/dxf/framework/scheduler/interface.go | 2 +- .../framework/scheduler/scheduler_manager.go | 7 +-- .../scheduler/scheduler_manager_nokit_test.go | 6 +-- pkg/dxf/framework/scheduler/scheduler_test.go | 10 ++--- pkg/dxf/framework/storage/table_test.go | 12 ++--- pkg/dxf/framework/storage/task_table.go | 2 +- pkg/server/handler/tests/dxf_test.go | 44 +++++++++++++++++++ pkg/server/handler/tikvhandler/dxf.go | 34 ++++++++++++++ pkg/server/http_status.go | 1 + .../addindextest1/disttask_test.go | 2 +- 14 files changed, 162 insertions(+), 41 deletions(-) diff --git a/pkg/dxf/framework/integrationtests/bench_test.go b/pkg/dxf/framework/integrationtests/bench_test.go index dc87029dc7fe7..85b6d69292534 100644 --- a/pkg/dxf/framework/integrationtests/bench_test.go +++ b/pkg/dxf/framework/integrationtests/bench_test.go @@ -41,7 +41,7 @@ import ( ) var ( - maxConcurrentTask = flag.Int("max-concurrent-task", proto.MaxConcurrentTask, "max concurrent task") + maxConcurrentTask = flag.Int("max-concurrent-task", proto.GetMaxConcurrentTask(), "max concurrent task") waitDuration = flag.Duration("task-wait-duration", 2*time.Minute, "task wait duration") schedulerInterval = flag.Duration("scheduler-interval", scheduler.CheckTaskFinishedInterval, "scheduler interval") taskExecutorMgrInterval = flag.Duration("task-executor-mgr-interval", taskexecutor.TaskCheckInterval, "task executor mgr interval") @@ -65,43 +65,43 @@ func BenchmarkSchedulerOverhead(b *testing.B) { }() schIntervalBak := scheduler.CheckTaskFinishedInterval exeMgrIntervalBak := taskexecutor.TaskCheckInterval - bak := proto.MaxConcurrentTask + restoreMaxConcurrentTask := proto.SetMaxConcurrentTaskForTest(*maxConcurrentTask) b.Cleanup(func() { - proto.MaxConcurrentTask = bak + restoreMaxConcurrentTask() scheduler.CheckTaskFinishedInterval = schIntervalBak taskexecutor.TaskCheckInterval = exeMgrIntervalBak }) - proto.MaxConcurrentTask = *maxConcurrentTask scheduler.CheckTaskFinishedInterval = *schedulerInterval taskexecutor.TaskCheckInterval = *taskExecutorMgrInterval - b.Logf("max concurrent task: %d", proto.MaxConcurrentTask) + maxConcurrentTaskValue := proto.GetMaxConcurrentTask() + b.Logf("max concurrent task: %d", maxConcurrentTaskValue) b.Logf("taks wait duration: %s", *waitDuration) b.Logf("task meta size: %d", *taskMetaSize) b.Logf("scheduler interval: %s", scheduler.CheckTaskFinishedInterval) b.Logf("task executor mgr interval: %s", taskexecutor.TaskCheckInterval) prepareForBenchTest(b) - c := testutil.NewTestDXFContext(b, 1, 2*proto.MaxConcurrentTask, false) + c := testutil.NewTestDXFContext(b, 1, 2*maxConcurrentTaskValue, false) registerTaskTypeForBench(c) if *noTask { time.Sleep(*waitDuration) } else { - // in this test, we will start 4*proto.MaxConcurrentTask tasks, but only - // proto.MaxConcurrentTask will be scheduled at the same time, for other + // in this test, we will start 4*maxConcurrentTaskValue tasks, but only + // maxConcurrentTaskValue will be scheduled at the same time, for other // tasks will be in queue only to check the performance of querying them. - for i := range 4 * proto.MaxConcurrentTask { + for i := range 4 * maxConcurrentTaskValue { taskKey := fmt.Sprintf("task-%03d", i) taskMeta := make([]byte, *taskMetaSize) _, err := handle.SubmitTask(c.Ctx, taskKey, proto.TaskTypeExample, c.Store.GetKeyspace(), 1, "", 0, taskMeta) require.NoError(c.T, err) } // task has 2 steps, each step has 1 subtask，wait in serial to reduce WaitTask check overhead. - // only wait first proto.MaxConcurrentTask and exit + // only wait first maxConcurrentTaskValue and exit time.Sleep(2 * *waitDuration) - for i := range proto.MaxConcurrentTask { + for i := range maxConcurrentTaskValue { taskKey := fmt.Sprintf("task-%03d", i) testutil.WaitTaskDoneOrPaused(c.Ctx, c.T, taskKey) } diff --git a/pkg/dxf/framework/proto/BUILD.bazel b/pkg/dxf/framework/proto/BUILD.bazel index 857d7bd32eb0a..54dd9ff5cde07 100644 --- a/pkg/dxf/framework/proto/BUILD.bazel +++ b/pkg/dxf/framework/proto/BUILD.bazel @@ -26,6 +26,6 @@ go_test( ], embed = [":proto"], flaky = True, - shard_count = 9, + shard_count = 10, deps = ["@com_github_stretchr_testify//require"], ) diff --git a/pkg/dxf/framework/proto/task.go b/pkg/dxf/framework/proto/task.go index 90e3b9684d44e..b3f8789d67acb 100644 --- a/pkg/dxf/framework/proto/task.go +++ b/pkg/dxf/framework/proto/task.go @@ -17,6 +17,7 @@ package proto import ( "cmp" "fmt" + "sync/atomic" "time" ) @@ -83,9 +84,44 @@ const ( NormalPriority = 512 ) -// MaxConcurrentTask is the max concurrency of task. -// TODO: remove this limit later. -var MaxConcurrentTask = 16 +const ( + // DefaultMaxConcurrentTask is the default max concurrency of task. + DefaultMaxConcurrentTask = 16 + // MinMaxConcurrentTask is the minimum allowed max concurrency of task. + MinMaxConcurrentTask = 16 + // MaxMaxConcurrentTask is the maximum allowed max concurrency of task. + MaxMaxConcurrentTask = 1000 +) + +var maxConcurrentTask atomic.Int64 + +func init() { + maxConcurrentTask.Store(DefaultMaxConcurrentTask) +} + +// GetMaxConcurrentTask returns the max concurrency of task. +func GetMaxConcurrentTask() int { + return int(maxConcurrentTask.Load()) +} + +// SetMaxConcurrentTask updates the max concurrency of task. +func SetMaxConcurrentTask(value int) error { + if value < MinMaxConcurrentTask || value > MaxMaxConcurrentTask { + return fmt.Errorf("max_concurrent_task %d is out of range [%d, %d]", + value, MinMaxConcurrentTask, MaxMaxConcurrentTask) + } + maxConcurrentTask.Store(int64(value)) + return nil +} + +// SetMaxConcurrentTaskForTest updates the max concurrency of task and returns a restore function. +func SetMaxConcurrentTaskForTest(value int) func() { + old := GetMaxConcurrentTask() + maxConcurrentTask.Store(int64(value)) + return func() { + maxConcurrentTask.Store(int64(old)) + } +} // ExtraParams is the extra params of task. // Note: only store params that's not used for filter or sort in this struct. diff --git a/pkg/dxf/framework/proto/task_test.go b/pkg/dxf/framework/proto/task_test.go index 87e0b25162b36..935d6c8669f7b 100644 --- a/pkg/dxf/framework/proto/task_test.go +++ b/pkg/dxf/framework/proto/task_test.go @@ -74,6 +74,23 @@ func TestTaskIsDone(t *testing.T) { } } +func TestMaxConcurrentTask(t *testing.T) { + restore := SetMaxConcurrentTaskForTest(DefaultMaxConcurrentTask) + defer restore() + + require.Equal(t, DefaultMaxConcurrentTask, GetMaxConcurrentTask()) + require.Equal(t, 1000, MaxMaxConcurrentTask) + for _, value := range []int{MinMaxConcurrentTask - 1, MaxMaxConcurrentTask + 1} { + require.Error(t, SetMaxConcurrentTask(value)) + require.Equal(t, DefaultMaxConcurrentTask, GetMaxConcurrentTask()) + } + + require.NoError(t, SetMaxConcurrentTask(128)) + require.Equal(t, 128, GetMaxConcurrentTask()) + require.NoError(t, SetMaxConcurrentTask(MaxMaxConcurrentTask)) + require.Equal(t, MaxMaxConcurrentTask, GetMaxConcurrentTask()) +} + func TestTaskCompare(t *testing.T) { taskA := Task{TaskBase: TaskBase{ ID: 100, diff --git a/pkg/dxf/framework/scheduler/interface.go b/pkg/dxf/framework/scheduler/interface.go index 89d0035def1eb..6f4b09ec63238 100644 --- a/pkg/dxf/framework/scheduler/interface.go +++ b/pkg/dxf/framework/scheduler/interface.go @@ -27,7 +27,7 @@ import ( // TaskManager defines the interface to access task table. type TaskManager interface { - // GetTopUnfinishedTasks returns unfinished tasks, limited by MaxConcurrentTask*2, + // GetTopUnfinishedTasks returns unfinished tasks, limited by GetMaxConcurrentTask()*2, // to make sure low ranking tasks can be scheduled if resource is enough. // The returned tasks are sorted by task order, see proto.Task. GetTopUnfinishedTasks(ctx context.Context) ([]*proto.TaskBase, error) diff --git a/pkg/dxf/framework/scheduler/scheduler_manager.go b/pkg/dxf/framework/scheduler/scheduler_manager.go index a13be8c1bfcfa..61cc723d2efc2 100644 --- a/pkg/dxf/framework/scheduler/scheduler_manager.go +++ b/pkg/dxf/framework/scheduler/scheduler_manager.go @@ -155,7 +155,7 @@ func NewManager(ctx context.Context, store kv.Storage, taskMgr TaskManager, serv serverID: serverID, }), logger: logger, - finishCh: make(chan struct{}, proto.MaxConcurrentTask), + finishCh: make(chan struct{}, proto.MaxMaxConcurrentTask), nodeRes: nodeRes, } schedulerManager.mu.schedulerMap = make(map[int64]Scheduler) @@ -244,7 +244,8 @@ func (sm *Manager) getSchedulableTasks(ctx context.Context) ([]*proto.TaskBase, defer r.End() getTasksFn := sm.taskMgr.GetTopUnfinishedTasks taskCnt := sm.getSchedulerCount() - if taskCnt >= proto.MaxConcurrentTask { + maxConcurrentTask := proto.GetMaxConcurrentTask() + if taskCnt >= maxConcurrentTask { // when we have reached the limit of concurrent tasks, we only handle // tasks in states that don't need resources, e.g. reverting/cancelling/ // pausing/modifying. @@ -291,7 +292,7 @@ func (sm *Manager) startSchedulers(schedulableTasks []*proto.TaskBase) error { switch task.State { case proto.TaskStatePending, proto.TaskStateRunning, proto.TaskStateResuming: taskCnt := sm.getSchedulerCount() - if taskCnt >= proto.MaxConcurrentTask { + if taskCnt >= proto.GetMaxConcurrentTask() { continue } reservedExecID, ok = sm.slotMgr.canReserve(task) diff --git a/pkg/dxf/framework/scheduler/scheduler_manager_nokit_test.go b/pkg/dxf/framework/scheduler/scheduler_manager_nokit_test.go index 013064427bc1e..83e97f7f21af5 100644 --- a/pkg/dxf/framework/scheduler/scheduler_manager_nokit_test.go +++ b/pkg/dxf/framework/scheduler/scheduler_manager_nokit_test.go @@ -363,11 +363,7 @@ func TestStartSchedulerCrossKeyspaceRuntime(t *testing.T) { } func TestFastRespondNoNeedResourceTaskWhenSchedulersReachLimit(t *testing.T) { - bak := proto.MaxConcurrentTask - t.Cleanup(func() { - proto.MaxConcurrentTask = bak - }) - proto.MaxConcurrentTask = 1 + t.Cleanup(proto.SetMaxConcurrentTaskForTest(1)) ctrl := gomock.NewController(t) defer ctrl.Finish() diff --git a/pkg/dxf/framework/scheduler/scheduler_test.go b/pkg/dxf/framework/scheduler/scheduler_test.go index afafd48736f1f..7005e6143e764 100644 --- a/pkg/dxf/framework/scheduler/scheduler_test.go +++ b/pkg/dxf/framework/scheduler/scheduler_test.go @@ -166,10 +166,9 @@ func checkSchedule(t *testing.T, taskCnt int, isSucc, isCancel, isSubtaskCancel, testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/domain/MockDisableDistTask", "return(true)") // test scheduleTaskLoop // test parallelism control - var originalConcurrency int + restoreMaxConcurrentTask := func() {} if taskCnt == 1 { - originalConcurrency = proto.MaxConcurrentTask - proto.MaxConcurrentTask = 1 + restoreMaxConcurrentTask = proto.SetMaxConcurrentTaskForTest(1) } store := testkit.CreateMockStore(t) @@ -190,10 +189,7 @@ func checkSchedule(t *testing.T, taskCnt int, isSucc, isCancel, isSubtaskCancel, sch.Start() defer func() { sch.Stop() - // make data race happy - if taskCnt == 1 { - proto.MaxConcurrentTask = originalConcurrency - } + restoreMaxConcurrentTask() }() // 3s diff --git a/pkg/dxf/framework/storage/table_test.go b/pkg/dxf/framework/storage/table_test.go index 6ac1c9b0c840a..e2f8478da4393 100644 --- a/pkg/dxf/framework/storage/table_test.go +++ b/pkg/dxf/framework/storage/table_test.go @@ -485,11 +485,7 @@ func TestSwitchTaskStepInBatch(t *testing.T) { func TestGetTopUnfinishedTasks(t *testing.T) { _, gm, ctx := testutil.InitTableTest(t) - bak := proto.MaxConcurrentTask - t.Cleanup(func() { - proto.MaxConcurrentTask = bak - }) - proto.MaxConcurrentTask = 4 + t.Cleanup(proto.SetMaxConcurrentTaskForTest(4)) require.NoError(t, gm.InitMeta(ctx, ":4000", "")) taskStates := []proto.TaskState{ proto.TaskStateSucceed, @@ -554,18 +550,18 @@ func TestGetTopUnfinishedTasks(t *testing.T) { require.Len(t, tasks, 8) require.Equal(t, []string{"key/6", "key/5", "key/1", "key/2", "key/3", "key/4", "key/8", "key/9"}, getTaskKeys(tasks)) - proto.MaxConcurrentTask = 6 + proto.SetMaxConcurrentTaskForTest(6) tasks, err = gm.GetTopUnfinishedTasks(ctx) require.NoError(t, err) require.Len(t, tasks, 11) require.Equal(t, []string{"key/6", "key/5", "key/1", "key/2", "key/3", "key/4", "key/8", "key/9", "key/10", "key/11", "key/12"}, getTaskKeys(tasks)) - proto.MaxConcurrentTask = 3 + proto.SetMaxConcurrentTaskForTest(3) tasks, err = gm.GetTopNoNeedResourceTasks(ctx) require.NoError(t, err) require.Equal(t, []string{"key/5", "key/3", "key/4", "key/12"}, getTaskKeys(tasks)) - proto.MaxConcurrentTask = 1 + proto.SetMaxConcurrentTaskForTest(1) tasks, err = gm.GetTopNoNeedResourceTasks(ctx) require.NoError(t, err) require.Equal(t, []string{"key/5", "key/3"}, getTaskKeys(tasks)) diff --git a/pkg/dxf/framework/storage/task_table.go b/pkg/dxf/framework/storage/task_table.go index 2afece8ba9d6d..d2a5cb856b3be 100644 --- a/pkg/dxf/framework/storage/task_table.go +++ b/pkg/dxf/framework/storage/task_table.go @@ -364,7 +364,7 @@ func (mgr *TaskManager) getTopTasks(ctx context.Context, states ...proto.TaskSta for _, s := range states { args = append(args, s) } - args = append(args, proto.MaxConcurrentTask*2) + args = append(args, proto.GetMaxConcurrentTask()*2) rs, err := mgr.ExecuteSQLWithNewSession(ctx, sql, args...) if err != nil { return nil, err diff --git a/pkg/server/handler/tests/dxf_test.go b/pkg/server/handler/tests/dxf_test.go index 58d739354148f..d42d2eb56e560 100644 --- a/pkg/server/handler/tests/dxf_test.go +++ b/pkg/server/handler/tests/dxf_test.go @@ -146,6 +146,50 @@ func TestDXFAPI(t *testing.T) { require.EqualValues(t, 2, out.PerKeyspace["ks1"]) }) + t.Run("max concurrent task api", func(t *testing.T) { + restore := proto.SetMaxConcurrentTaskForTest(proto.DefaultMaxConcurrentTask) + defer restore() + + runAndCheckReqFn(t, http.StatusBadRequest, "This api only support GET and POST method", func() (*http.Response, error) { + req, err := http.NewRequest(http.MethodDelete, ts.StatusURL("/dxf/task/max_concurrent"), nil) + require.NoError(t, err) + return http.DefaultClient.Do(req) + }) + for _, c := range [][2]string{ + {"/dxf/task/max_concurrent", "invalid value "}, + {"/dxf/task/max_concurrent?value=aa", "invalid value "}, + {"/dxf/task/max_concurrent?value=15", "out of range"}, + {fmt.Sprintf("/dxf/task/max_concurrent?value=%d", proto.MaxMaxConcurrentTask+1), "out of range"}, + } { + path, errMsg := c[0], c[1] + runAndCheckReqFn(t, http.StatusBadRequest, errMsg, func() (*http.Response, error) { + return ts.PostStatus(path, "", bytes.NewBuffer([]byte(""))) + }) + } + + body := runAndCheckReqFn(t, http.StatusOK, "", func() (*http.Response, error) { + return ts.FetchStatus("/dxf/task/max_concurrent") + }) + out := struct { + MaxConcurrentTask int `json:"max_concurrent_task"` + }{} + require.NoError(t, json.Unmarshal(body, &out)) + require.Equal(t, proto.DefaultMaxConcurrentTask, out.MaxConcurrentTask) + + body = runAndCheckReqFn(t, http.StatusOK, "", func() (*http.Response, error) { + return ts.PostStatus("/dxf/task/max_concurrent?value=128", "", bytes.NewBuffer([]byte(""))) + }) + require.NoError(t, json.Unmarshal(body, &out)) + require.Equal(t, 128, out.MaxConcurrentTask) + require.Equal(t, 128, proto.GetMaxConcurrentTask()) + + body = runAndCheckReqFn(t, http.StatusOK, "", func() (*http.Response, error) { + return ts.FetchStatus("/dxf/task/max_concurrent") + }) + require.NoError(t, json.Unmarshal(body, &out)) + require.Equal(t, 128, out.MaxConcurrentTask) + }) + t.Run("task history api", func(t *testing.T) { seedHistoryTasks := func(t *testing.T) []int64 { t.Helper() diff --git a/pkg/server/handler/tikvhandler/dxf.go b/pkg/server/handler/tikvhandler/dxf.go index 966fda3081bd6..ab6f40ec878d3 100644 --- a/pkg/server/handler/tikvhandler/dxf.go +++ b/pkg/server/handler/tikvhandler/dxf.go @@ -344,6 +344,40 @@ func (h *DXFScheduleTuneHandler) ServeHTTP(w http.ResponseWriter, req *http.Requ } } +// DXFTaskMaxConcurrentHandler handles the in-memory DXF task concurrency limit. +type DXFTaskMaxConcurrentHandler struct{} + +// NewDXFTaskMaxConcurrentHandler creates a new DXFTaskMaxConcurrentHandler. +func NewDXFTaskMaxConcurrentHandler() *DXFTaskMaxConcurrentHandler { + return &DXFTaskMaxConcurrentHandler{} +} + +func (*DXFTaskMaxConcurrentHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) { + switch req.Method { + case http.MethodGet: + handler.WriteData(w, map[string]any{ + "max_concurrent_task": proto.GetMaxConcurrentTask(), + }) + case http.MethodPost: + valueStr := req.FormValue("value") + value, err := strconv.Atoi(valueStr) + if err != nil { + handler.WriteError(w, errors.Errorf("invalid value %s, error %v", valueStr, err)) + return + } + if err := proto.SetMaxConcurrentTask(value); err != nil { + handler.WriteError(w, err) + return + } + logutil.BgLogger().Info("set DXF max concurrent task", zap.Int("maxConcurrentTask", value)) + handler.WriteData(w, map[string]any{ + "max_concurrent_task": proto.GetMaxConcurrentTask(), + }) + default: + handler.WriteError(w, errors.Errorf("This api only support GET and POST method")) + } +} + // DXFTaskMaxRuntimeSlotsHandler handles changing max runtime slots of DXF task. type DXFTaskMaxRuntimeSlotsHandler struct{} diff --git a/pkg/server/http_status.go b/pkg/server/http_status.go index 44821a2e33c55..59468fa830481 100644 --- a/pkg/server/http_status.go +++ b/pkg/server/http_status.go @@ -255,6 +255,7 @@ func (s *Server) startHTTPServer() { router.Handle("/dxf/schedule/tune", tikvhandler.NewDXFScheduleTuneHandler(tikvHandlerTool.Store.(kv.Storage))).Name("DXF_Schedule_Tune") router.Handle("/dxf/task/active", tikvhandler.NewDXFActiveTaskHandler()).Name("DXF_Task_Active") router.Handle("/dxf/task/history", tikvhandler.NewDXFTaskHistoryHandler()).Name("DXF_Task_History") + router.Handle("/dxf/task/max_concurrent", tikvhandler.NewDXFTaskMaxConcurrentHandler()).Name("DXF_Task_Max_Concurrent") router.Handle("/dxf/import-into/history/job/{keyspace}/{job_id}", tikvhandler.NewDXFImportIntoHistoryJobInfoHandler()).Name("DXF_Import_Into_History_Job_Info") router.Handle("/dxf/task/{taskID}/max_runtime_slots", tikvhandler.NewDXFTaskMaxRuntimeSlotsHandler()).Name("DXF_Task_Max_Runtime_Slots") } diff --git a/tests/realtikvtest/addindextest1/disttask_test.go b/tests/realtikvtest/addindextest1/disttask_test.go index 69433b13f99d4..fa749d9874e00 100644 --- a/tests/realtikvtest/addindextest1/disttask_test.go +++ b/tests/realtikvtest/addindextest1/disttask_test.go @@ -590,7 +590,7 @@ func TestAddIndexScheduleAway(t *testing.T) { } func TestAddIndexDistCleanUpBlock(t *testing.T) { - proto.MaxConcurrentTask = 1 + t.Cleanup(proto.SetMaxConcurrentTaskForTest(1)) testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/util/cpu/mockNumCpu", `return(1)`) ch := make(chan struct{}) testfailpoint.EnableCall(t, "github.com/pingcap/tidb/pkg/dxf/framework/scheduler/doCleanupTask", func() { From e6d7326184ab4bf789a4bbf5d8c4bcd5442e1799 Mon Sep 17 00:00:00 2001 From: D3Hunter Date: Fri, 3 Jul 2026 20:17:23 +0800 Subject: [PATCH 2/3] dxf, server: refine max concurrent task API --- pkg/dxf/framework/proto/task.go | 23 +++++++---- pkg/dxf/framework/proto/task_test.go | 8 ++-- .../framework/scheduler/scheduler_manager.go | 7 +++- pkg/dxf/framework/storage/table_test.go | 9 +++-- pkg/server/handler/tests/dxf_test.go | 39 +++++++++++-------- pkg/server/handler/tikvhandler/dxf.go | 22 +++++++---- pkg/server/http_status.go | 3 +- 7 files changed, 69 insertions(+), 42 deletions(-) diff --git a/pkg/dxf/framework/proto/task.go b/pkg/dxf/framework/proto/task.go index b3f8789d67acb..982ed617f38cf 100644 --- a/pkg/dxf/framework/proto/task.go +++ b/pkg/dxf/framework/proto/task.go @@ -85,14 +85,21 @@ const ( ) const ( - // DefaultMaxConcurrentTask is the default max concurrency of task. - DefaultMaxConcurrentTask = 16 - // MinMaxConcurrentTask is the minimum allowed max concurrency of task. - MinMaxConcurrentTask = 16 - // MaxMaxConcurrentTask is the maximum allowed max concurrency of task. - MaxMaxConcurrentTask = 1000 + // maxConcurrentTaskLowerBound is the minimum allowed DXF task concurrency. + maxConcurrentTaskLowerBound = 16 + // MaxConcurrentTaskUpperBound is the current safety cap for DXF task concurrency. + // TODO: remove this cap after the DXF scheduler no longer runs all schedulers on the owner node. + MaxConcurrentTaskUpperBound = 1000 + // DefaultMaxConcurrentTask is the default DXF task concurrency. + DefaultMaxConcurrentTask = maxConcurrentTaskLowerBound ) +// maxConcurrentTask is an owner-local emergency tuning knob for DXF scheduling. +// It is intentionally kept in memory only: it is not persisted to TiKV, is reset +// on restart, and only affects the TiDB node that receives the update. Operators +// should change it through the DXF owner node when many small tasks are blocked +// by the default limit. Raising it increases scheduler overhead and memory usage +// on the owner, so the owner node may need a larger resource spec first. var maxConcurrentTask atomic.Int64 func init() { @@ -106,9 +113,9 @@ func GetMaxConcurrentTask() int { // SetMaxConcurrentTask updates the max concurrency of task. func SetMaxConcurrentTask(value int) error { - if value < MinMaxConcurrentTask || value > MaxMaxConcurrentTask { + if value < maxConcurrentTaskLowerBound || value > MaxConcurrentTaskUpperBound { return fmt.Errorf("max_concurrent_task %d is out of range [%d, %d]", - value, MinMaxConcurrentTask, MaxMaxConcurrentTask) + value, maxConcurrentTaskLowerBound, MaxConcurrentTaskUpperBound) } maxConcurrentTask.Store(int64(value)) return nil diff --git a/pkg/dxf/framework/proto/task_test.go b/pkg/dxf/framework/proto/task_test.go index 935d6c8669f7b..c99faa0352d43 100644 --- a/pkg/dxf/framework/proto/task_test.go +++ b/pkg/dxf/framework/proto/task_test.go @@ -79,16 +79,16 @@ func TestMaxConcurrentTask(t *testing.T) { defer restore() require.Equal(t, DefaultMaxConcurrentTask, GetMaxConcurrentTask()) - require.Equal(t, 1000, MaxMaxConcurrentTask) - for _, value := range []int{MinMaxConcurrentTask - 1, MaxMaxConcurrentTask + 1} { + require.Equal(t, 1000, MaxConcurrentTaskUpperBound) + for _, value := range []int{maxConcurrentTaskLowerBound - 1, MaxConcurrentTaskUpperBound + 1} { require.Error(t, SetMaxConcurrentTask(value)) require.Equal(t, DefaultMaxConcurrentTask, GetMaxConcurrentTask()) } require.NoError(t, SetMaxConcurrentTask(128)) require.Equal(t, 128, GetMaxConcurrentTask()) - require.NoError(t, SetMaxConcurrentTask(MaxMaxConcurrentTask)) - require.Equal(t, MaxMaxConcurrentTask, GetMaxConcurrentTask()) + require.NoError(t, SetMaxConcurrentTask(MaxConcurrentTaskUpperBound)) + require.Equal(t, MaxConcurrentTaskUpperBound, GetMaxConcurrentTask()) } func TestTaskCompare(t *testing.T) { diff --git a/pkg/dxf/framework/scheduler/scheduler_manager.go b/pkg/dxf/framework/scheduler/scheduler_manager.go index 61cc723d2efc2..3d68ab6799ce0 100644 --- a/pkg/dxf/framework/scheduler/scheduler_manager.go +++ b/pkg/dxf/framework/scheduler/scheduler_manager.go @@ -154,8 +154,11 @@ func NewManager(ctx context.Context, store kv.Storage, taskMgr TaskManager, serv slotMgr: slotMgr, serverID: serverID, }), - logger: logger, - finishCh: make(chan struct{}, proto.MaxMaxConcurrentTask), + logger: logger, + // finishCh must be able to buffer finish signals for the largest runtime + // value of maxConcurrentTask. Otherwise, raising the limit after startup + // can make non-blocking sends drop signals until the periodic cleanup loop runs. + finishCh: make(chan struct{}, proto.MaxConcurrentTaskUpperBound), nodeRes: nodeRes, } schedulerManager.mu.schedulerMap = make(map[int64]Scheduler) diff --git a/pkg/dxf/framework/storage/table_test.go b/pkg/dxf/framework/storage/table_test.go index e2f8478da4393..18c94371279b5 100644 --- a/pkg/dxf/framework/storage/table_test.go +++ b/pkg/dxf/framework/storage/table_test.go @@ -550,21 +550,24 @@ func TestGetTopUnfinishedTasks(t *testing.T) { require.Len(t, tasks, 8) require.Equal(t, []string{"key/6", "key/5", "key/1", "key/2", "key/3", "key/4", "key/8", "key/9"}, getTaskKeys(tasks)) - proto.SetMaxConcurrentTaskForTest(6) + restoreMaxConcurrentTask := proto.SetMaxConcurrentTaskForTest(6) tasks, err = gm.GetTopUnfinishedTasks(ctx) require.NoError(t, err) require.Len(t, tasks, 11) require.Equal(t, []string{"key/6", "key/5", "key/1", "key/2", "key/3", "key/4", "key/8", "key/9", "key/10", "key/11", "key/12"}, getTaskKeys(tasks)) + restoreMaxConcurrentTask() - proto.SetMaxConcurrentTaskForTest(3) + restoreMaxConcurrentTask = proto.SetMaxConcurrentTaskForTest(3) tasks, err = gm.GetTopNoNeedResourceTasks(ctx) require.NoError(t, err) require.Equal(t, []string{"key/5", "key/3", "key/4", "key/12"}, getTaskKeys(tasks)) + restoreMaxConcurrentTask() - proto.SetMaxConcurrentTaskForTest(1) + restoreMaxConcurrentTask = proto.SetMaxConcurrentTaskForTest(1) tasks, err = gm.GetTopNoNeedResourceTasks(ctx) require.NoError(t, err) require.Equal(t, []string{"key/5", "key/3"}, getTaskKeys(tasks)) + restoreMaxConcurrentTask() } func TestGetUsedSlotsOnNodesAndBusyNodes(t *testing.T) { diff --git a/pkg/server/handler/tests/dxf_test.go b/pkg/server/handler/tests/dxf_test.go index d42d2eb56e560..49be8ab8964e3 100644 --- a/pkg/server/handler/tests/dxf_test.go +++ b/pkg/server/handler/tests/dxf_test.go @@ -149,17 +149,28 @@ func TestDXFAPI(t *testing.T) { t.Run("max concurrent task api", func(t *testing.T) { restore := proto.SetMaxConcurrentTaskForTest(proto.DefaultMaxConcurrentTask) defer restore() + const maxConcurrentTaskPath = "/dxf/schedule/max_concurrent_task" + checkMaxConcurrentTaskOutput := func(body []byte, expected int) { + out := struct { + MaxConcurrentTask int `json:"max_concurrent_task"` + Persistence string `json:"persistence"` + }{} + require.NoError(t, json.Unmarshal(body, &out)) + require.Equal(t, expected, out.MaxConcurrentTask) + require.Equal(t, "memory_only", out.Persistence) + require.NotContains(t, string(body), "scope") + } runAndCheckReqFn(t, http.StatusBadRequest, "This api only support GET and POST method", func() (*http.Response, error) { - req, err := http.NewRequest(http.MethodDelete, ts.StatusURL("/dxf/task/max_concurrent"), nil) + req, err := http.NewRequest(http.MethodDelete, ts.StatusURL(maxConcurrentTaskPath), nil) require.NoError(t, err) return http.DefaultClient.Do(req) }) for _, c := range [][2]string{ - {"/dxf/task/max_concurrent", "invalid value "}, - {"/dxf/task/max_concurrent?value=aa", "invalid value "}, - {"/dxf/task/max_concurrent?value=15", "out of range"}, - {fmt.Sprintf("/dxf/task/max_concurrent?value=%d", proto.MaxMaxConcurrentTask+1), "out of range"}, + {maxConcurrentTaskPath, "invalid value "}, + {maxConcurrentTaskPath + "?value=aa", "invalid value "}, + {maxConcurrentTaskPath + "?value=15", "out of range"}, + {fmt.Sprintf("%s?value=%d", maxConcurrentTaskPath, proto.MaxConcurrentTaskUpperBound+1), "out of range"}, } { path, errMsg := c[0], c[1] runAndCheckReqFn(t, http.StatusBadRequest, errMsg, func() (*http.Response, error) { @@ -168,26 +179,20 @@ func TestDXFAPI(t *testing.T) { } body := runAndCheckReqFn(t, http.StatusOK, "", func() (*http.Response, error) { - return ts.FetchStatus("/dxf/task/max_concurrent") + return ts.FetchStatus(maxConcurrentTaskPath) }) - out := struct { - MaxConcurrentTask int `json:"max_concurrent_task"` - }{} - require.NoError(t, json.Unmarshal(body, &out)) - require.Equal(t, proto.DefaultMaxConcurrentTask, out.MaxConcurrentTask) + checkMaxConcurrentTaskOutput(body, proto.DefaultMaxConcurrentTask) body = runAndCheckReqFn(t, http.StatusOK, "", func() (*http.Response, error) { - return ts.PostStatus("/dxf/task/max_concurrent?value=128", "", bytes.NewBuffer([]byte(""))) + return ts.PostStatus(maxConcurrentTaskPath+"?value=128", "", bytes.NewBuffer([]byte(""))) }) - require.NoError(t, json.Unmarshal(body, &out)) - require.Equal(t, 128, out.MaxConcurrentTask) + checkMaxConcurrentTaskOutput(body, 128) require.Equal(t, 128, proto.GetMaxConcurrentTask()) body = runAndCheckReqFn(t, http.StatusOK, "", func() (*http.Response, error) { - return ts.FetchStatus("/dxf/task/max_concurrent") + return ts.FetchStatus(maxConcurrentTaskPath) }) - require.NoError(t, json.Unmarshal(body, &out)) - require.Equal(t, 128, out.MaxConcurrentTask) + checkMaxConcurrentTaskOutput(body, 128) }) t.Run("task history api", func(t *testing.T) { diff --git a/pkg/server/handler/tikvhandler/dxf.go b/pkg/server/handler/tikvhandler/dxf.go index ab6f40ec878d3..58020d28c6c46 100644 --- a/pkg/server/handler/tikvhandler/dxf.go +++ b/pkg/server/handler/tikvhandler/dxf.go @@ -352,12 +352,15 @@ func NewDXFTaskMaxConcurrentHandler() *DXFTaskMaxConcurrentHandler { return &DXFTaskMaxConcurrentHandler{} } +// ServeHTTP implements http.Handler interface. +// +// The configured value is local to the TiDB process that handles the request +// and is kept in memory only. Send the request to the current DXF owner when +// tuning scheduler concurrency. func (*DXFTaskMaxConcurrentHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) { switch req.Method { case http.MethodGet: - handler.WriteData(w, map[string]any{ - "max_concurrent_task": proto.GetMaxConcurrentTask(), - }) + writeMaxConcurrentTask(w) case http.MethodPost: valueStr := req.FormValue("value") value, err := strconv.Atoi(valueStr) @@ -369,15 +372,20 @@ func (*DXFTaskMaxConcurrentHandler) ServeHTTP(w http.ResponseWriter, req *http.R handler.WriteError(w, err) return } - logutil.BgLogger().Info("set DXF max concurrent task", zap.Int("maxConcurrentTask", value)) - handler.WriteData(w, map[string]any{ - "max_concurrent_task": proto.GetMaxConcurrentTask(), - }) + logutil.BgLogger().Info("set in-memory DXF max concurrent task", zap.Int("maxConcurrentTask", value)) + writeMaxConcurrentTask(w) default: handler.WriteError(w, errors.Errorf("This api only support GET and POST method")) } } +func writeMaxConcurrentTask(w http.ResponseWriter) { + handler.WriteData(w, map[string]any{ + "max_concurrent_task": proto.GetMaxConcurrentTask(), + "persistence": "memory_only", + }) +} + // DXFTaskMaxRuntimeSlotsHandler handles changing max runtime slots of DXF task. type DXFTaskMaxRuntimeSlotsHandler struct{} diff --git a/pkg/server/http_status.go b/pkg/server/http_status.go index 59468fa830481..e9574fae7a45e 100644 --- a/pkg/server/http_status.go +++ b/pkg/server/http_status.go @@ -255,7 +255,8 @@ func (s *Server) startHTTPServer() { router.Handle("/dxf/schedule/tune", tikvhandler.NewDXFScheduleTuneHandler(tikvHandlerTool.Store.(kv.Storage))).Name("DXF_Schedule_Tune") router.Handle("/dxf/task/active", tikvhandler.NewDXFActiveTaskHandler()).Name("DXF_Task_Active") router.Handle("/dxf/task/history", tikvhandler.NewDXFTaskHistoryHandler()).Name("DXF_Task_History") - router.Handle("/dxf/task/max_concurrent", tikvhandler.NewDXFTaskMaxConcurrentHandler()).Name("DXF_Task_Max_Concurrent") + // This API updates only the TiDB process that handles the request and is not persisted. + router.Handle("/dxf/schedule/max_concurrent_task", tikvhandler.NewDXFTaskMaxConcurrentHandler()).Name("DXF_Schedule_Max_Concurrent_Task") router.Handle("/dxf/import-into/history/job/{keyspace}/{job_id}", tikvhandler.NewDXFImportIntoHistoryJobInfoHandler()).Name("DXF_Import_Into_History_Job_Info") router.Handle("/dxf/task/{taskID}/max_runtime_slots", tikvhandler.NewDXFTaskMaxRuntimeSlotsHandler()).Name("DXF_Task_Max_Runtime_Slots") } From 931976e33e1686eb2293e0989105e5596b5c5445 Mon Sep 17 00:00:00 2001 From: D3Hunter Date: Wed, 8 Jul 2026 22:25:06 +0800 Subject: [PATCH 3/3] docs: add DXF max concurrent task HTTP API --- docs/tidb_http_api.md | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/docs/tidb_http_api.md b/docs/tidb_http_api.md index 53b4d2e32ab2f..b1e524e2685d4 100644 --- a/docs/tidb_http_api.md +++ b/docs/tidb_http_api.md @@ -782,3 +782,41 @@ curl -X POST "http://{TiDBIP}:10080/test/delete/indexkey/{db}/{table}/{index}?{i "table": "t" } ``` + +## APIs unique to TiDB-X SYSTEM keyspace + +These APIs are registered only on TiDB instances running in the TiDB-X SYSTEM keyspace. + +### Get or update the DXF max concurrent task limit + +This API gets or updates the process-local maximum number of DXF tasks that can be scheduled concurrently. The value is kept in memory only, is not persisted to TiKV, and is reset when the TiDB process restarts. When updating the value, send the request to the current DXF owner. + +Get the current value: + +```shell +curl http://{TiDBIP}:10080/dxf/schedule/max_concurrent_task +``` + +Example response: + +```json +{ + "max_concurrent_task": 16, + "persistence": "memory_only" +} +``` + +Update the value: + +```shell +curl -X POST "http://{TiDBIP}:10080/dxf/schedule/max_concurrent_task?value={number}" +``` + +`value` must be an integer in the range `[16, 1000]`. The response uses the same format as the `GET` request: + +```json +{ + "max_concurrent_task": 128, + "persistence": "memory_only" +} +```