// buildSampTask returns tow variables, the first bool is whether the task meets region error
// and need to rebuild.
4040
func (e *AnalyzeFastExec) buildSampTask() (needRebuild bool, err error) {
	// Do get regions row count.
	bo := tikv.NewBackoffer(context.Background(), 500)
	e.rowCount = 0
	needRebuildForRoutine := make([]bool, e.concurrency)
	errs := make([]error, e.concurrency)
	sampTasksForRoutine := make([][]*AnalyzeFastTask, e.concurrency)
	e.sampLocs = make(chan *tikv.KeyLocation, e.concurrency)
	e.wg.Add(e.concurrency)
	for i := 0; i < e.concurrency; i++ {
		go e.getSampRegionsRowCount(bo, &needRebuildForRoutine[i], &errs[i], &sampTasksForRoutine[i])
	}

	defer func() {
		close(e.sampLocs)
		e.wg.Wait()
		if err != nil {
			return
		}
		for i := 0; i < e.concurrency; i++ {
			if errs[i] != nil {
				err = errs[i]
			}
			needRebuild = needRebuild || needRebuildForRoutine[i]
			e.sampTasks = append(e.sampTasks, sampTasksForRoutine[i]...)
		}
	}()

	store, _ := e.ctx.GetStore().(tikv.Storage)
	e.cache = store.GetRegionCache()
	startKey, endKey := tablecodec.GetTableHandleKeyRange(e.physicalTableID)
	// extract all regions contain the table
	e.scanTasks = e.scanTasks[:0]
	e.sampTasks = e.sampTasks[:0]
	targetKey := startKey
	for {
		// Search for the region which contains the targetKey.
		loc, err := e.cache.LocateKey(bo, targetKey)
		if err != nil {
			return false, err
		}
		if bytes.Compare(endKey, loc.StartKey) < 0 {
			break
		}
		// Set the next search key.
		targetKey = loc.EndKey

		// If the KV pairs in the region all belonging to the table, add it to the sample task.
		if bytes.Compare(startKey, loc.StartKey) <= 0 && bytes.Compare(loc.EndKey, endKey) <= 0 {
			e.sampLocs <- loc
			continue
		}

		e.scanTasks = append(e.scanTasks, loc)
		if bytes.Compare(loc.StartKey, startKey) < 0 {
			loc.StartKey = startKey
		}
		if bytes.Compare(endKey, loc.EndKey) < 0 || len(loc.EndKey) == 0 {
			loc.EndKey = endKey
			break
		}
	}

	return false, nil
}
1165542
func (i *FuncInterceptor) BeforeTask(decision *swf.PollForDecisionTaskOutput) {
	if i.BeforeTaskFn != nil {
		i.BeforeTaskFn(decision)
	}
}
1246371
func (server *Server) RegisterTask(name string, taskFunc interface{}) error {
	if err := tasks.ValidateTask(taskFunc); err != nil {
		return err
	}
	server.registeredTasks[name] = taskFunc
	server.broker.SetRegisteredTaskNames(server.GetRegisteredTaskNames())
	return nil
}
281328
func (*CallingGoroutineDispatcher) DispatchTask(task *swf.PollForDecisionTaskOutput, handler func(*swf.PollForDecisionTaskOutput)) {
	handler(task)
}
276076
func (m TaskManager) LaunchTask(c context.Context, ctl task.Controller) error {
	cfg := ctl.Task().(*messages.UrlFetchTask)
	started := clock.Now(c)

	// Defaults.
	method := cfg.Method
	if method == "" {
		method = "GET"
	}
	timeout := cfg.TimeoutSec
	if timeout == 0 {
		timeout = 60
	}

	ctl.DebugLog("%s %s", method, cfg.Url)

	// There must be no errors here in reality, since cfg is validated already by
	// ValidateProtoMessage.
	u, err := url.Parse(cfg.Url)
	if err != nil {
		return err
	}

	type tuple struct {
		resp *http.Response
		body []byte // first 4Kb of the response, for debug log
		err  error
	}
	result := make(chan tuple)

	// Do the fetch asynchronously with datastore update.
	go func() {
		defer close(result)
		c, _ := clock.WithTimeout(c, time.Duration(timeout)*time.Second)
		client := &http.Client{Transport: urlfetch.Get(c)}
		resp, err := client.Do(&http.Request{
			Method: method,
			URL:    u,
		})
		if err != nil {
			result <- tuple{nil, nil, err}
			return
		}
		defer resp.Body.Close()
		// Ignore read errors here. HTTP status code is set, it's the main output
		// of the operation. Read 4K only since we use body only for debug message
		// that is limited in size.
		buf := bytes.Buffer{}
		io.CopyN(&buf, resp.Body, 4096)
		result <- tuple{resp, buf.Bytes(), nil}
	}()

	// Save the invocation log now (since URL fetch can take up to 8 minutes).
	// Ignore errors. As long as final Save is OK, we don't care about this one.
	// Do NOT set status to StatusRunning, because by doing so we take
	// responsibility to detect crashes below and we don't want to do it (let
	// the scheduler retry LaunchTask automatically instead).
	if err := ctl.Save(c); err != nil {
		logging.Warningf(c, "Failed to save invocation state: %s", err)
	}

	// Wait for completion.
	res := <-result

	duration := clock.Now(c).Sub(started)
	status := task.StatusSucceeded
	if res.err != nil || res.resp.StatusCode >= 400 {
		status = task.StatusFailed
	}

	ctl.DebugLog("Finished with overall status %s in %s", status, duration)
	if res.err != nil {
		ctl.DebugLog("URL fetch error: %s", res.err)
	} else {
		ctl.DebugLog(dumpResponse(res.resp, res.body))
	}
	ctl.State().Status = status
	return nil
}
1132687
func UpdateTask(tx Tx, t *api.Task) error {
	return tx.update(tableTask, t)
}
329290
func (d *driverPluginClient) StartTask(c *TaskConfig) (*TaskHandle, *DriverNetwork, error) {
	req := &proto.StartTaskRequest{
		Task: taskConfigToProto(c),
	}

	resp, err := d.client.StartTask(d.doneCtx, req)
	if err != nil {
		st := status.Convert(err)
		if len(st.Details()) > 0 {
			if rec, ok := st.Details()[0].(*sproto.RecoverableError); ok {
				return nil, nil, structs.NewRecoverableError(err, rec.Recoverable)
			}
		}
		return nil, nil, grpcutils.HandleGrpcErr(err, d.doneCtx)
	}

	var net *DriverNetwork
	if resp.NetworkOverride != nil {
		net = &DriverNetwork{
			PortMap:       map[string]int{},
			IP:            resp.NetworkOverride.Addr,
			AutoAdvertise: resp.NetworkOverride.AutoAdvertise,
		}
		for k, v := range resp.NetworkOverride.PortMap {
			net.PortMap[k] = int(v)
		}
	}

	return taskHandleFromProto(resp.Handle), net, nil
}
1209665
func (w *CeleryWorker) RunTask(message *TaskMessage) (*ResultMessage, error) {

	// get task
	task := w.GetTask(message.Task)
	if task == nil {
		return nil, fmt.Errorf("task %s is not registered", message.Task)
	}

	// convert to task interface
	taskInterface, ok := task.(CeleryTask)
	if ok {
		if err := taskInterface.ParseKwargs(message.Kwargs); err != nil {
			return nil, err
		}
		val, err := taskInterface.RunTask()
		if err != nil {
			return nil, err
		}
		return getResultMessage(val), err
	}

	// use reflection to execute function ptr
	taskFunc := reflect.ValueOf(task)
	return runTaskFunc(&taskFunc, message)
}
1252328
func (f *MaxReplicasFilter) SetTask(t *api.Task) bool {
	if t.Spec.Placement != nil && t.Spec.Placement.MaxReplicas > 0 {
		f.t = t
		return true
	}

	return false
}
552508
func (s *scheduler) CreateTask(sch schedule.Schedule, wfMap *wmap.WorkflowMap, startOnCreate bool, opts ...core.TaskOption) (core.Task, core.TaskErrors) {
	return s.createTask(sch, wfMap, startOnCreate, "user", opts...)
}
795408
func (t TaskService) CreateTask(ctx context.Context, tsk *platform.Task) error {
	u, err := newURL(t.Addr, tasksPath)
	if err != nil {
		return err
	}

	taskBytes, err := json.Marshal(tsk)
	if err != nil {
		return err
	}

	req, err := http.NewRequest("POST", u.String(), bytes.NewReader(taskBytes))
	if err != nil {
		return err
	}

	req.Header.Set("Content-Type", "application/json")
	SetToken(t.Token, req)

	hc := newClient(u.Scheme, t.InsecureSkipVerify)

	resp, err := hc.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if err := CheckError(resp); err != nil {
		return err
	}

	var tr taskResponse
	if err := json.NewDecoder(resp.Body).Decode(&tr); err != nil {
		return err
	}
	*tsk = tr.Task

	return nil
}
362605
func (s *StopTaskOutput) SetTask(v *Task) *StopTaskOutput {
	s.Task = v
	return s
}
536056
func AddTask(t *automationpb.TaskContainer, taskName string, parameters map[string]string) {
	t.ParallelTasks = append(t.ParallelTasks, NewTask(taskName, parameters))
}
16744
func (m TaskManager) LaunchTask(c context.Context, ctl task.Controller) error {
	cfg := ctl.Task().(*messages.NoopTask)

	sleepFor := time.Duration(cfg.SleepMs) * time.Millisecond
	if sleepFor == 0 {
		sleepFor = 20 * time.Second
	}
	ctl.DebugLog("Running noop task for %s", sleepFor)
	clock.Sleep(c, sleepFor)

	for i := int64(0); i < cfg.TriggersCount; i++ {
		ctl.EmitTrigger(c, &internal.Trigger{
			Id:      fmt.Sprintf("noop:%d:%d", ctl.InvocationID(), i),
			Payload: &internal.Trigger_Noop{Noop: &api.NoopTrigger{}},
		})
	}

	ctl.State().Status = task.StatusSucceeded
	return nil
}
922696
func successTask(id string, cacheable Cacheable) task {
	return func(c *cache) error {
		// Check for discarded Cacheable first.
		if c.buckets[id] == nil {
			return nil
		}
		// Set bucket values.
		b := c.buckets[id]
		b.cacheable = cacheable
		b.status = statusLoaded
		b.loaded = time.Now()
		b.lastUsed = b.loaded
		// Notify all waiters.
		for _, waiter := range c.buckets[id].waiters {
			waiter <- func() (Cacheable, error) {
				return cacheable, nil
			}
		}
		b.waiters = nil
		return nil
	}
}
1204416
func ValidateTask(task interface{}) error {
	v := reflect.ValueOf(task)
	t := v.Type()

	// Task must be a function
	if t.Kind() != reflect.Func {
		return ErrTaskMustBeFunc
	}

	// Task must return at least a single value
	if t.NumOut() < 1 {
		return ErrTaskReturnsNoValue
	}

	// Last return value must be error
	lastReturnType := t.Out(t.NumOut() - 1)
	errorInterface := reflect.TypeOf((*error)(nil)).Elem()
	if !lastReturnType.Implements(errorInterface) {
		return ErrLastReturnValueMustBeError
	}

	return nil
}
309820
func (s *SubmitContainerStateChangeInput) SetTask(v string) *SubmitContainerStateChangeInput {
	s.Task = &v
	return s
}