// StartSubTask creates a subtask and run it.
// TODO(ehco) rename this func.
StartSubTask creates a subtask and run it.
TODO(ehco) rename this func.
// StartSubTask creates a subtask and run it . // TODO ( ehco ) rename this func .
func (w *SourceWorker) StartSubTask(cfg *config.SubTaskConfig, expectStage, validatorStage pb.Stage, needLock bool) error {
	if needLock {
		w.Lock()
		defer w.Unlock()
	}
	err := copyConfigFromSource(cfg, w.cfg, w.relayEnabled.Load())
	if err != nil {
		return err
	}
	st := NewSubTask(cfg, w.etcdClient, w.name)
	w.subTaskHolder.recordSubTask(st)
	if w.closed.Load() {
		st.fail(terror.ErrWorkerAlreadyClosed.Generate())
		return nil
	}
	cfg2, err := cfg.DecryptPassword()
	if err != nil {
		st.fail(errors.Annotate(err, "start sub task"))
		return nil
	}
	st.cfg = cfg2
	st.cfg.WorkerName = w.name
	if w.relayEnabled.Load() && w.relayPurger.Purging() {
		st.fail(terror.ErrWorkerRelayIsPurging.Generate(cfg.Name))
		return nil
	}
	w.l.Info("subtask created", zap.Stringer("config", cfg2))
	st.Run(expectStage, validatorStage, w.getRelayWithoutLock())
	return nil
}
func ( w * SourceWorker ) StartSubTask ( cfg * config . SubTaskConfig , expectStage , validatorStage pb . Stage , needLock bool ) error { if needLock { w . Lock ( ) defer w . Unlock ( ) } err := copyConfigFromSource ( cfg , w . cfg , w . relayEnabled . Load ( ) ) if err != nil { return err } st := NewSubTask ( cfg , w . etcdClient , w . name ) w . subTaskHolder . recordSubTask ( st ) if w . closed . Load ( ) { st . fail ( terror . ErrWorkerAlreadyClosed . Generate ( ) ) return nil } cfg2 , err := cfg . DecryptPassword ( ) if err != nil { st . fail ( errors . Annotate ( err , " " ) ) return nil } st . cfg = cfg2 st . cfg . WorkerName = w . name if w . relayEnabled . Load ( ) && w . relayPurger . Purging ( ) { st . fail ( terror . ErrWorkerRelayIsPurging . Generate ( cfg . Name ) ) return nil } w . l . Info ( " " , zap . Stringer ( " " , cfg2 ) ) st . Run ( expectStage , validatorStage , w . getRelayWithoutLock ( ) ) return nil }
StartSubTask creates a subtask and run it.
StartSubTask creates a subtask and run it .
// copy some config item from dm-worker's source config // directly put cfg into subTaskHolder // the uniqueness of subtask should be assured by etcd // inject worker name to this subtask config // TODO: retry until purged finished