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

close #4811

dm/dm/worker/subtask.go

@@ -478,7 +478,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

@@ -520,3 +520,34 @@ func (t *testSubTask) TestSubtaskFastQuit(c *C) {
 	}
 	c.Assert(st.Stage(), Equals, pb.Stage_Stopped)
 }
+
+func (t *testSubTask) TestSubtaskRace(c *C) {
+	// 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: "testSubtaskScene",
+		Mode: config.ModeFull,
+	}
+	st := NewSubTask(cfg, nil, "worker")
+	c.Assert(st.Stage(), DeepEquals, pb.Stage_New)
+	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
+}