validator(dm): fix deadlock of validator (#7269)

close #7241

dm/syncer/data_validator.go

@@ -725,10 +725,10 @@ func (v *DataValidator) Stop() {
 
 func (v *DataValidator) stopInner() {
 	v.Lock()
-	defer v.Unlock()
 	v.L.Info("stopping")
 	if v.Stage() != pb.Stage_Running {
 		v.L.Warn("not started")
+		v.Unlock()
 		return
 	}
 
@@ -737,8 +737,13 @@ func (v *DataValidator) stopInner() {
 	v.fromDB.Close()
 	v.toDB.Close()
 
+	// release the lock so that the error routine can process errors
+	// wait until all errors are recorded
+	v.Unlock()
 	v.wg.Wait()
 	close(v.errChan) // close error chan after all possible sender goroutines stopped
+	v.Lock()         // lock and modify the stage
+	defer v.Unlock()
 
 	v.setStage(pb.Stage_Stopped)
 	v.L.Info("stopped")

dm/syncer/data_validator_test.go

@@ -198,6 +198,42 @@ func TestValidatorErrorProcessRoutine(t *testing.T) {
 	require.Len(t, validator.result.Errors, 1)
 }
 
+func TestValidatorDeadLock(t *testing.T) {
+	require.NoError(t, failpoint.Enable("github.com/pingcap/tiflow/dm/syncer/ValidatorMockUpstreamTZ", `return()`))
+	defer func() {
+		require.NoError(t, failpoint.Disable("github.com/pingcap/tiflow/dm/syncer/ValidatorMockUpstreamTZ"))
+	}()
+	cfg := genSubtaskConfig(t)
+	syncerObj := NewSyncer(cfg, nil, nil)
+	_, _, err := conn.InitMockDBFull()
+	require.NoError(t, err)
+	defer func() {
+		conn.DefaultDBProvider = &conn.DefaultDBProviderImpl{}
+	}()
+
+	validator := NewContinuousDataValidator(cfg, syncerObj, false)
+	validator.persistHelper.schemaInitialized.Store(true)
+	validator.Start(pb.Stage_Running)
+	require.Equal(t, pb.Stage_Running, validator.Stage())
+	validator.wg.Add(1)
+	go func() {
+		defer func() {
+			// ignore panic when try to insert error to a closed channel,
+			// which will happen after the validator is successfully stopped.
+			// The panic is expected.
+			validator.wg.Done()
+			// nolint:errcheck
+			recover()
+		}()
+		for i := 0; i < 100; i++ {
+			validator.sendError(context.Canceled) // prevent from stopping the validator
+		}
+	}()
+	// stuck if the validator doesn't unlock before waiting wg
+	validator.Stop()
+	require.Equal(t, pb.Stage_Stopped, validator.Stage())
+}
+
 type mockedCheckPointForValidator struct {
 	CheckPoint
 	cnt     int