syncer(dm): fix the data race issue (#5881)

close #4811

dm/dm/worker/subtask.go

@@ -552,7 +552,13 @@ func (st *SubTask) markResultCanceled() bool {
 func (st *SubTask) Result() *pb.ProcessResult {
 	st.RLock()
 	defer st.RUnlock()
-	return st.result
+	if st.result == nil {
+		return nil
+	}
+	tempProcessResult, _ := st.result.Marshal()
+	newProcessResult := &pb.ProcessResult{}
+	_ = newProcessResult.Unmarshal(tempProcessResult)
+	return newProcessResult
 }
 
 // Close stops the sub task.

dm/dm/worker/subtask_test.go

@@ -572,3 +572,35 @@ func TestValidatorStatus(t *testing.T) {
 	require.True(t, terror.ErrValidatorNotFound.Equal(err))
 	// validator != nil: will be tested in IT
 }
+
+func TestSubtaskRace(t *testing.T) {
+	// to test data race of Marshal() and markResultCanceled()
+	tempErrors := []*pb.ProcessError{}
+	tempDetail := []byte{}
+	tempProcessResult := pb.ProcessResult{
+		IsCanceled: false,
+		Errors:     tempErrors,
+		Detail:     tempDetail,
+	}
+	cfg := &config.SubTaskConfig{
+		Name: "test-subtask-race",
+		ValidatorCfg: config.ValidatorConfig{
+			Mode: config.ValidationFast,
+		},
+	}
+	st := NewSubTaskWithStage(cfg, pb.Stage_Paused, nil, "worker")
+	st.result = &tempProcessResult
+	tempQueryStatusResponse := pb.QueryStatusResponse{}
+	tempQueryStatusResponse.SubTaskStatus = make([]*pb.SubTaskStatus, 1)
+	tempSubTaskStatus := pb.SubTaskStatus{}
+	tempSubTaskStatus.Result = st.Result()
+	tempQueryStatusResponse.SubTaskStatus[0] = &tempSubTaskStatus
+	st.result.IsCanceled = false
+	go func() {
+		for i := 0; i < 10; i++ {
+			_, _ = tempQueryStatusResponse.Marshal()
+		}
+	}()
+	st.markResultCanceled()
+	// this test is to test data race, so don't need assert here
+}