disttask: fix potential block on DXF scheduler (pingcap#59945)

close pingcap#59944

Author: wjhuang2016

pkg/disttask/framework/scheduler/scheduler_manager.go

@@ -359,7 +359,10 @@ func (sm *Manager) startScheduler(basicTask *proto.TaskBase, allocateSlots bool,
 		}()
 		metrics.UpdateMetricsForRunTask(task)
 		scheduler.ScheduleTask()
-		sm.finishCh <- struct{}{}
+		select {
+		case sm.finishCh <- struct{}{}:
+		default:
+		}
 	})
 }
 
@@ -385,6 +388,7 @@ func (sm *Manager) cleanupTaskLoop() {
 //
 //	tasks with global sort should clean up tmp files stored on S3.
 func (sm *Manager) doCleanupTask() {
+	failpoint.InjectCall("doCleanupTask")
 	tasks, err := sm.taskMgr.GetTasksInStates(
 		sm.ctx,
 		proto.TaskStateFailed,

tests/realtikvtest/addindextest1/BUILD.bazel

@@ -11,6 +11,7 @@ go_test(
     deps = [
         "//pkg/config",
         "//pkg/ddl/ingest",
+        "//pkg/disttask/framework/proto",
         "//pkg/disttask/framework/storage",
         "//pkg/disttask/framework/taskexecutor",
         "//pkg/errno",

tests/realtikvtest/addindextest1/disttask_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/config"
 	"github.com/pingcap/tidb/pkg/ddl/ingest"
+	"github.com/pingcap/tidb/pkg/disttask/framework/proto"
 	"github.com/pingcap/tidb/pkg/disttask/framework/storage"
 	"github.com/pingcap/tidb/pkg/disttask/framework/taskexecutor"
 	"github.com/pingcap/tidb/pkg/errno"
@@ -462,3 +463,35 @@ func TestAddIndexScheduleAway(t *testing.T) {
 	tk.MustExec("alter table t add index idx(b);")
 	require.NotEqual(t, int64(0), jobID.Load())
 }
+
+func TestAddIndexDistCleanUpBlock(t *testing.T) {
+	proto.MaxConcurrentTask = 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/disttask/framework/scheduler/doCleanupTask", func() {
+		<-ch
+	})
+	store := realtikvtest.CreateMockStoreAndSetup(t)
+	if store.Name() != "TiKV" {
+		t.Skip("TiKV store only")
+	}
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("drop database if exists test;")
+	tk.MustExec("create database test;")
+	tk.MustExec("use test;")
+	tk.MustExec(`set global tidb_enable_dist_task=1;`)
+	var wg sync.WaitGroup
+	for i := 0; i < 4; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			tk := testkit.NewTestKit(t, store)
+			tk.MustExec(fmt.Sprintf("create table test.t%d (a int, b int);", i))
+			tk.MustExec(fmt.Sprintf("insert into test.t%d values (1, 1);", i))
+			tk.MustExec(fmt.Sprintf("alter table test.t%d add index idx(b);", i))
+		}()
+	}
+	wg.Wait()
+	close(ch)
+}