// doValidate: runs in a separate goroutine.
runs in a separate goroutine.
// doValidate : runs in a separate goroutine .
func (v *DataValidator) doValidate() {
	defer v.wg.Done()
	if err := v.waitSyncerRunning(); err != nil {
		v.errChan <- terror.Annotate(err, "failed to wait syncer running")
		return
	}
	location := v.syncer.checkpoint.FlushedGlobalPoint()
	if v.streamerController.IsClosed() {
		err := v.streamerController.Start(v.tctx, location)
		if err != nil {
			v.errChan <- terror.Annotate(err, "fail to start streamer controller")
			return
		}
	}
	v.L.Info("start continuous validation")
	v.startValidateWorkers()
	defer func() {
		for _, worker := range v.workers {
			worker.close()
		}
	}()
	currLoc := location.CloneWithFlavor(v.cfg.Flavor)
	for {
		e, err := v.streamerController.GetEvent(v.tctx)
		if err != nil {
			v.errChan <- terror.Annotate(err, "fail to get binlog from stream controller")
			return
		}
		switch ev := e.Event.(type) {
		case *replication.RotateEvent:
			currLoc.Position.Name = string(ev.NextLogName)
			currLoc.Position.Pos = uint32(ev.Position)
		case *replication.GTIDEvent, *replication.MariadbGTIDEvent:
			currLoc.Position.Pos = e.Header.LogPos
			gtidStr, _ := event.GetGTIDStr(e)
			if err = currLoc.Update(gtidStr); err != nil {
				v.errChan <- terror.Annotate(err, "failed to update gtid set")
				return
			}
		default:
			currLoc.Position.Pos = e.Header.LogPos
		}
		err = v.waitSyncerSynced(currLoc)
		if err != nil {
			v.errChan <- terror.Annotate(err, "failed to wait syncer")
			return
		}
		if ev, ok := e.Event.(*replication.RowsEvent); ok {
			if err = v.processRowsEvent(e.Header, ev); err != nil {
				v.L.Warn("failed to process event: ", zap.Reflect("error", err))
				v.errChan <- terror.Annotate(err, "failed to process event")
				return
			}
		}
	}
}
func ( v * DataValidator ) doValidate ( ) { defer v . wg . Done ( ) if err := v . waitSyncerRunning ( ) ; err != nil { v . errChan <- terror . Annotate ( err , " " ) return } location := v . syncer . checkpoint . FlushedGlobalPoint ( ) if v . streamerController . IsClosed ( ) { err := v . streamerController . Start ( v . tctx , location ) if err != nil { v . errChan <- terror . Annotate ( err , " " ) return } } v . L . Info ( " " ) v . startValidateWorkers ( ) defer func ( ) { for _ , worker := range v . workers { worker . close ( ) } } ( ) currLoc := location . CloneWithFlavor ( v . cfg . Flavor ) for { e , err := v . streamerController . GetEvent ( v . tctx ) if err != nil { v . errChan <- terror . Annotate ( err , " " ) return } switch ev := e . Event . ( type ) { case * replication . RotateEvent : currLoc . Position . Name = string ( ev . NextLogName ) currLoc . Position . Pos = uint32 ( ev . Position ) case * replication . GTIDEvent , * replication . MariadbGTIDEvent : currLoc . Position . Pos = e . Header . LogPos gtidStr , _ := event . GetGTIDStr ( e ) if err = currLoc . Update ( gtidStr ) ; err != nil { v . errChan <- terror . Annotate ( err , " " ) return } default : currLoc . Position . Pos = e . Header . LogPos } err = v . waitSyncerSynced ( currLoc ) if err != nil { v . errChan <- terror . Annotate ( err , " " ) return } if ev , ok := e . Event . ( * replication . RowsEvent ) ; ok { if err = v . processRowsEvent ( e . Header , ev ) ; err != nil { v . L . Warn ( " " , zap . Reflect ( " " , err ) ) v . errChan <- terror . Annotate ( err , " " ) return } } } }
runs in a separate goroutine.
runs in a separate goroutine .
// todo: if validator starts on task start, we need to make sure the location we got is the init location of syncer // todo: right now, there's change they have a gap // it's for test, some fields in streamerController is mocked, cannot call Start // wait until syncer synced current event