// failedTask notifies the cache that a loading failed.
196142
func failedTask(id string, err error) task {
	return func(c *cache) error {
		// Check for discarded Cacheable first.
		if c.buckets[id] == nil {
			return nil
		}
		// Notify all waiters.
		for _, waiter := range c.buckets[id].waiters {
			waiter <- func() (Cacheable, error) {
				return nil, errors.Annotate(err, ErrLoading, errorMessages, id)
			}
		}
		delete(c.buckets, id)
		return nil
	}
}
1164511
func (c *Client) EnableTask(id string) *EnableTaskResult {
	resp, err := c.do("PUT", fmt.Sprintf("/tasks/%v/enable", id), ContentTypeJSON)
	if err != nil {
		return &EnableTaskResult{Err: err}
	}

	switch resp.Meta.Type {
	case rbody.ScheduledTaskEnabledType:
		return &EnableTaskResult{resp.Body.(*rbody.ScheduledTaskEnabled), nil}
	case rbody.ErrorType:
		return &EnableTaskResult{Err: resp.Body.(*rbody.Error)}
	default:
		return &EnableTaskResult{Err: ErrAPIResponseMetaType}
	}
}
1003239
func (i *FuncInterceptor) AfterTaskFailed(activity *swf.PollForActivityTaskOutput, err error) {
	if i.AfterTaskFailedFn != nil {
		i.AfterTaskFailedFn(activity, err)
	}
}
378661
func (ow *outerWorker) buildTask(ctx context.Context) (*lookUpJoinTask, error) {
	ow.executor.newFirstChunk()

	task := &lookUpJoinTask{
		doneCh:            make(chan error, 1),
		outerResult:       ow.executor.newFirstChunk(),
		encodedLookUpKeys: chunk.NewChunkWithCapacity([]*types.FieldType{types.NewFieldType(mysql.TypeBlob)}, ow.ctx.GetSessionVars().MaxChunkSize),
		lookupMap:         mvmap.NewMVMap(),
	}
	task.memTracker = memory.NewTracker(stringutil.MemoizeStr(func() string { return fmt.Sprintf("lookup join task %p", task) }), -1)
	task.memTracker.AttachTo(ow.parentMemTracker)

	ow.increaseBatchSize()
	if ow.lookup.isOuterJoin { // if is outerJoin, push the requiredRows down
		requiredRows := int(atomic.LoadInt64(&ow.lookup.requiredRows))
		task.outerResult.SetRequiredRows(requiredRows, ow.maxBatchSize)
	} else {
		task.outerResult.SetRequiredRows(ow.batchSize, ow.maxBatchSize)
	}

	task.memTracker.Consume(task.outerResult.MemoryUsage())
	for !task.outerResult.IsFull() {
		err := ow.executor.Next(ctx, chunk.NewRecordBatch(ow.executorChk))
		if err != nil {
			return task, err
		}
		if ow.executorChk.NumRows() == 0 {
			break
		}

		oldMemUsage := task.outerResult.MemoryUsage()
		task.outerResult.Append(ow.executorChk, 0, ow.executorChk.NumRows())
		newMemUsage := task.outerResult.MemoryUsage()
		task.memTracker.Consume(newMemUsage - oldMemUsage)
	}
	if task.outerResult.NumRows() == 0 {
		return nil, nil
	}

	if ow.filter != nil {
		outerMatch := make([]bool, 0, task.outerResult.NumRows())
		var err error
		task.outerMatch, err = expression.VectorizedFilter(ow.ctx, ow.filter, chunk.NewIterator4Chunk(task.outerResult), outerMatch)
		if err != nil {
			return task, err
		}
		task.memTracker.Consume(int64(cap(task.outerMatch)))
	}
	return task, nil
}
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
}
520391
func (worker *Worker) taskRetry(signature *tasks.Signature) error {
	// Update task state to RETRY
	if err := worker.server.GetBackend().SetStateRetry(signature); err != nil {
		return fmt.Errorf("Set state to 'retry' for task %s returned error: %s", signature.UUID, err)
	}

	// Decrement the retry counter, when it reaches 0, we won't retry again
	signature.RetryCount--

	// Increase retry timeout
	signature.RetryTimeout = retry.FibonacciNext(signature.RetryTimeout)

	// Delay task by signature.RetryTimeout seconds
	eta := time.Now().UTC().Add(time.Second * time.Duration(signature.RetryTimeout))
	signature.ETA = &eta

	log.WARNING.Printf("Task %s failed. Going to retry in %d seconds.", signature.UUID, signature.RetryTimeout)

	// Send the task back to the queue
	_, err := worker.server.SendTask(signature)
	return err
}
1180778
func (c *Cluster) GetTask(input string) (types.Task, error) {
	var task *swarmapi.Task
	if err := c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
		t, err := getTask(ctx, state.controlClient, input)
		if err != nil {
			return err
		}
		task = t
		return nil
	}); err != nil {
		return types.Task{}, err
	}
	return convert.TaskFromGRPC(*task)
}
1345644
func (c *Client) GetTask(id int) (Task, error) {
	task := Task{}
	stringID := strconv.Itoa(id)
	r := c.NewRequest("GET", "/tasks/"+stringID)
	resp, err := c.DoRequest(r)

	if err != nil {
		log.Printf("Error requesting task %v", err)
		return task, err
	}
	defer resp.Body.Close()

	resBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		log.Printf("Error reading task request %v", resBody)
		return task, err
	}
	err = json.Unmarshal(resBody, &task)
	if err != nil {
		log.Printf("Error unmarshaling task %v", err)
	}
	return task, err
}
177759
func loading(c *cache, id string) {
	cacheable, err := c.load(id)
	if err != nil {
		c.taskc <- failedTask(id, err)
	} else {
		c.taskc <- successTask(id, cacheable)
	}
}
536803
func (a *Archivist) ArchiveTask(c context.Context, task *logdog.ArchiveTask) error {
	c = log.SetFields(c, log.Fields{
		"project": task.Project,
		"id":      task.Id,
	})
	log.Debugf(c, "Received archival task.")

	err := a.archiveTaskImpl(c, task)

	failure := isFailure(err)
	log.Fields{
		log.ErrorKey: err,
		"failure":    failure,
	}.Infof(c, "Finished archive task.")

	// Add a result metric.
	tsCount.Add(c, 1, !failure)

	return err
}
852647
func (cat *catalog) marshalTask(task proto.Message) ([]byte, error) {
	if cat.GetTaskManager(task) == nil {
		return nil, fmt.Errorf("unrecognized task definition type %T", task)
	}
	// Enumerate all fields of the wrapper until we find a matching type.
	taskType := reflect.TypeOf(task)
	wrapper := messages.TaskDefWrapper{}
	v := reflect.ValueOf(&wrapper).Elem()
	for i := 0; i < v.NumField(); i++ {
		field := v.Field(i)
		if field.Type() == taskType {
			field.Set(reflect.ValueOf(task))
			return proto.Marshal(&wrapper)
		}
	}
	// This can happen only if TaskDefWrapper wasn't updated when a new task type
	// was added. This is a developer's mistake, not a config mistake.
	return nil, fmt.Errorf("could not find a field of type %T in TaskDefWrapper", task)
}
1360537
func (c *Client) UpdateTask(ctx context.Context, id int64, tu TaskUpdate, opt *Filter) (Task, error) {
	task := new(Task)
	err := c.request(ctx, "PUT", fmt.Sprintf("tasks/%d", id), tu, nil, opt, task)
	return *task, err
}
841962
func RollbackTaskOnError(err *error, task string, action Action) {
	chain := NewActionChain()
	chain.PushTask(task, action)
	chain.RollbackOnError(err)
}
731461
func (s *StartImportTaskOutput) SetTask(v *ImportTask) *StartImportTaskOutput {
	s.Task = v
	return s
}
13619
func (worker *copIteratorWorker) handleTaskOnce(bo *Backoffer, task *copTask, ch chan<- *copResponse) ([]*copTask, error) {
	failpoint.Inject("handleTaskOnceError", func(val failpoint.Value) {
		if val.(bool) {
			failpoint.Return(nil, errors.New("mock handleTaskOnce error"))
		}
	})

	sender := NewRegionRequestSender(worker.store.regionCache, worker.store.client)
	req := &tikvrpc.Request{
		Type: task.cmdType,
		Cop: &coprocessor.Request{
			Tp:     worker.req.Tp,
			Data:   worker.req.Data,
			Ranges: task.ranges.toPBRanges(),
		},
		Context: kvrpcpb.Context{
			IsolationLevel: pbIsolationLevel(worker.req.IsolationLevel),
			Priority:       kvPriorityToCommandPri(worker.req.Priority),
			NotFillCache:   worker.req.NotFillCache,
			HandleTime:     true,
			ScanDetail:     true,
		},
	}
	startTime := time.Now()
	resp, rpcCtx, err := sender.SendReqCtx(bo, req, task.region, ReadTimeoutMedium)
	if err != nil {
		return nil, errors.Trace(err)
	}
	// Set task.storeAddr field so its task.String() method have the store address information.
	task.storeAddr = sender.storeAddr
	costTime := time.Since(startTime)
	if costTime > minLogCopTaskTime {
		worker.logTimeCopTask(costTime, task, bo, resp)
	}
	metrics.TiKVCoprocessorHistogram.Observe(costTime.Seconds())

	if task.cmdType == tikvrpc.CmdCopStream {
		return worker.handleCopStreamResult(bo, rpcCtx, resp.CopStream, task, ch)
	}

	// Handles the response for non-streaming copTask.
	return worker.handleCopResponse(bo, rpcCtx, &copResponse{pbResp: resp.Cop}, task, ch, nil)
}
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
}
384672
func getTask(res *http.Response, e error) (*Task, error) {
	if e != nil {
		return nil, e
	}
	var task Task
	err := json.NewDecoder(res.Body).Decode(&task)
	if err != nil {
		return nil, err
	}
	if task.State == "ERROR" {
		// Critical: return task as well, so that it can be examined
		// for error details.
		return &task, TaskError{task.ID, getFailedStep(&task)}
	}
	return &task, nil
}