// handleTaskOnce handles single copTask, successful results are send to channel.
// If error happened, returns error. If region split or meet lock, returns the remain tasks.
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)
}
1001635
func (m *Memberlist) invokeAckHandler(ack ackResp, timestamp time.Time) {
	m.ackLock.Lock()
	ah, ok := m.ackHandlers[ack.SeqNo]
	delete(m.ackHandlers, ack.SeqNo)
	m.ackLock.Unlock()
	if !ok {
		return
	}
	ah.timer.Stop()
	ah.ackFn(ack.Payload, timestamp)
}
711444
func (c *CheckpointWriter) UpdateTask(taskID string, status workflowpb.TaskState, err error) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	errorMessage := ""
	if err != nil {
		errorMessage = err.Error()
	}

	t := c.checkpoint.Tasks[taskID]
	t.State = status
	t.Error = errorMessage
	return c.saveLocked()
}
69135
func (c *ModelBuilderContext) EnsureTask(task Task) error {
	task = c.setLifecycleOverride(task)
	key := buildTaskKey(task)

	existing, found := c.Tasks[key]
	if found {
		if reflect.DeepEqual(task, existing) {
			glog.V(8).Infof("EnsureTask ignoring identical ")
			return nil
		} else {
			glog.Warningf("EnsureTask found task mismatch for %q", key)
			glog.Warningf("\tExisting: %v", existing)
			glog.Warningf("\tNew: %v", task)

			return fmt.Errorf("cannot add different task with same key %q", key)
		}
	}
	c.Tasks[key] = task
	return nil
}
493236
func (q *sessionQueue) AcceptTask(task *backupTask) (reserveStatus, bool) {
	q.queueCond.L.Lock()

	// Examine the current reserve status of the session queue.
	curStatus := q.reserveStatus()
	switch curStatus {

	// The session queue is exhausted, and cannot accept the task because it
	// is full. Reject the task such that it can be tried against a
	// different session.
	case reserveExhausted:
		q.queueCond.L.Unlock()
		return curStatus, false

	// The session queue is not exhausted. Compute the sweep and reward
	// outputs as a function of the session parameters. If the outputs are
	// dusty or uneconomical to backup, the task is rejected and will not be
	// tried again.
	//
	// TODO(conner): queue backups and retry with different session params.
	case reserveAvailable:
		err := task.bindSession(q.cfg.ClientSession)
		if err != nil {
			q.queueCond.L.Unlock()
			log.Debugf("SessionQueue %s rejected backup chanid=%s "+
				"commit-height=%d: %v", q.ID(), task.id.ChanID,
				task.id.CommitHeight, err)
			return curStatus, false
		}
	}

	// The sweep and reward outputs satisfy the session's policy, queue the
	// task for final signing and delivery.
	q.pendingQueue.PushBack(task)

	// Finally, compute the session's *new* reserve status. This will be
	// used by the client to determine if it can continue using this session
	// queue, or if it should negotiate a new one.
	newStatus := q.reserveStatus()
	q.queueCond.L.Unlock()

	q.queueCond.Signal()

	return newStatus, true
}
1352005
func GetTaskAncestry(taskName string, tasks []wfv1.DAGTask) []string {
	taskByName := make(map[string]wfv1.DAGTask)
	for _, task := range tasks {
		taskByName[task.Name] = task
	}

	visited := make(map[string]bool)
	var getAncestry func(s string)
	getAncestry = func(currTask string) {
		task := taskByName[currTask]
		for _, depTask := range task.Dependencies {
			getAncestry(depTask)
		}
		if currTask != taskName {
			visited[currTask] = true
		}
	}

	getAncestry(taskName)
	ancestry := make([]string, 0)
	for ancestor := range visited {
		ancestry = append(ancestry, ancestor)
	}
	return ancestry
}
1262687
func (a *Client) DeleteTask(params *DeleteTaskParams, authInfo runtime.ClientAuthInfoWriter) (*DeleteTaskNoContent, error) {
	// TODO: Validate the params before sending
	if params == nil {
		params = NewDeleteTaskParams()
	}

	result, err := a.transport.Submit(&runtime.ClientOperation{
		ID:                 "deleteTask",
		Method:             "DELETE",
		PathPattern:        "/tasks/{id}",
		ProducesMediaTypes: []string{"application/vnd.goswagger.examples.task-tracker.v1+json"},
		ConsumesMediaTypes: []string{"application/vnd.goswagger.examples.task-tracker.v1+json"},
		Schemes:            []string{"http", "https"},
		Params:             params,
		Reader:             &DeleteTaskReader{formats: a.formats},
		AuthInfo:           authInfo,
		Context:            params.Context,
		Client:             params.HTTPClient,
	})
	if err != nil {
		return nil, err
	}
	return result.(*DeleteTaskNoContent), nil

}
1289342
func SelectTaskFromTaskPb(pb *PlanPb, ctx *Context, sel *rel.SqlSelect) (Task, error) {
	switch {
	case pb.Source != nil:
		return SourceFromPB(pb, ctx)
	case pb.Where != nil:
		return WhereFromPB(pb), nil
	case pb.Having != nil:
		return HavingFromPB(pb), nil
	case pb.GroupBy != nil:
		return GroupByFromPB(pb), nil
	case pb.Order != nil:
		return OrderFromPB(pb), nil
	case pb.Projection != nil:
		return ProjectionFromPB(pb, sel), nil
	case pb.JoinMerge != nil:
		u.Warnf("JoinMerge not implemented: %T", pb)
	case pb.JoinKey != nil:
		u.Warnf("JoinKey not implemented: %T", pb)
	default:
		u.Warnf("not implemented: %#v", pb)
	}
	return nil, ErrNotImplemented
}
182541
func (h *AsyncHttpHandler) handle(f func() (string, error)) {
	go func() {
		logger.Info("Started job %v", h.id)

		ts := time.Now()
		url, err := f()
		logger.Info("Completed job %v in %v", h.id, time.Since(ts))

		if err != nil {
			h.CompletedWithError(err)
		} else if url != "" {
			h.CompletedWithLocation(url)
		} else {
			h.Completed()
		}
	}()
}
37447
func (task *CleanupTask) Stop() {
	task.once.Do(func() {
		if cleanup := task.Cleanup; cleanup != nil {
			cleanup()
		}
	})
}
366792
func (r *Application) AllTaskRunning() bool {
	if r.Instances == nil || *r.Instances == 0 {
		return true
	}
	if r.Tasks == nil {
		return false
	}
	if r.TasksRunning == *r.Instances {
		return true
	}
	return false
}
1087871
func (cs *Server) handleNonce(w http.ResponseWriter, r *http.Request) {
	if r.Method != "GET" {
		http.Error(w, "not a GET", http.StatusMethodNotAllowed)
		return
	}

	mac := hmac.New(sha256.New, cs.keyHMAC)
	message := fmt.Sprintf("%d", time.Now().Unix())
	mac.Write([]byte(message))
	messageHash := string(mac.Sum(nil))
	nonce := fmt.Sprintf("%s-%x", message, messageHash)
	if _, err := io.Copy(w, strings.NewReader(nonce)); err != nil {
		log.Printf("error sending nonce: %v", err)
		return
	}
}
266897
func (s *Server) RemoveTask(ctx context.Context, request *api.RemoveTaskRequest) (*api.RemoveTaskResponse, error) {
	if request.TaskID == "" {
		return nil, status.Errorf(codes.InvalidArgument, errInvalidArgument.Error())
	}

	err := s.store.Update(func(tx store.Tx) error {
		return store.DeleteTask(tx, request.TaskID)
	})
	if err != nil {
		if err == store.ErrNotExist {
			return nil, status.Errorf(codes.NotFound, "task %s not found", request.TaskID)
		}
		return nil, err
	}
	return &api.RemoveTaskResponse{}, nil
}
158676
func (c *ServiceClient) RemoveTask(task *TaskServices) {
	ops := operations{}

	for _, service := range task.Services {
		id := makeTaskServiceID(task.AllocID, task.Name, service, task.Canary)
		ops.deregServices = append(ops.deregServices, id)

		for _, check := range service.Checks {
			cid := makeCheckID(id, check)
			ops.deregChecks = append(ops.deregChecks, cid)

			if check.TriggersRestarts() {
				c.checkWatcher.Unwatch(cid)
			}
		}
	}

	// Remove the task from the alloc's registrations
	c.removeTaskRegistration(task.AllocID, task.Name)

	// Now add them to the deregistration fields; main Run loop will update
	c.commit(&ops)
}
1319529
func (b *MultiHandler) Handle(rec *Record) {
	wg := sync.WaitGroup{}
	wg.Add(len(b.handlers))
	for _, handler := range b.handlers {
		go func(handler Handler) {
			handler.Handle(rec)
			wg.Done()
		}(handler)
	}
	wg.Wait()
}
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
}
962912
func (s *globalTaskService) TaskWaitC(taskID int) <-chan int {
	c := make(chan int)

	go func() {
		defer close(c)

		s.RLock()
		t, ok := s.tasks[taskID]
		s.RUnlock()

		if !ok {
			return
		}

		// remove the task from the queue after a configured about of time
		defer s.taskRemoveAfter(t)

		// signal that the task is complete
		<-t.done
	}()

	return c
}