// NewExecuteTask initializes task with fields needed to query server nodes.
326018
func NewExecuteTask(cluster *Cluster, statement *Statement) *ExecuteTask {
	return &ExecuteTask{
		baseTask: newTask(cluster),
		taskID:   statement.TaskId,
		scan:     statement.IsScan(),
	}
}
708249
func NewRemoveTask(cluster *Cluster, packageName string) *RemoveTask {
	return &RemoveTask{
		baseTask:    newTask(cluster),
		packageName: packageName,
	}
}
536056
func AddTask(t *automationpb.TaskContainer, taskName string, parameters map[string]string) {
	t.ParallelTasks = append(t.ParallelTasks, NewTask(taskName, parameters))
}
107799
func dbRspProcessingTask(c *asyncConnHdl, ctl workerCtl) (sig *interrupt_code, te *taskStatus) {

	var req asyncReqPtr
	select {
	case sig := <-ctl:
		// interrupted
		return &sig, &ok_status
	case req = <-c.pendingResps:
		// continue to process
	}

	// process response to asyncRequest
	reader := c.super.reader
	cmd := req.cmd

	resp, e3 := GetResponse(reader, cmd) // REVU - protocol modified to handle VIRTUALS
	if e3 != nil {
		// system error
		log.Println("<TEMP DEBUG> Request sent to faults chan on error in GetResponse: ", e3)
		req.stat = rcverr
		req.error = newSystemErrorWithCause("GetResponse os.Error", e3)
		c.faults <- req
		return nil, &taskStatus{rcverr, e3}
	}

	// if responsed processed was for cmd QUIT then signal the rest of the crew
	// REVU - ok, a bit hacky but it works.
	if cmd == &QUIT {
		c.feedback <- workerStatus{0, quit_processed, nil, nil}
		fakesig := pause
		c.isShutdown = true
		SetFutureResult(req.future, cmd, resp)
		return &fakesig, &ok_status
	}

	SetFutureResult(req.future, cmd, resp)
	return nil, &ok_status
}
980505
func WriteTask(
	ctx types.Context,
	config gofig.Config,
	w http.ResponseWriter,
	store types.Store,
	task *types.Task,
	okStatus int) error {

	if store.GetBool("async") {
		WriteJSON(w, http.StatusAccepted, task)
		return nil
	}

	exeTimeoutDur, err := time.ParseDuration(
		config.GetString(types.ConfigServerTasksExeTimeout))
	if err != nil {
		exeTimeoutDur = time.Duration(time.Second * 60)
	}
	exeTimeout := time.NewTimer(exeTimeoutDur)

	select {
	case <-services.TaskWaitC(ctx, task.ID):
		if task.Error != nil {
			return task.Error
		}
		WriteJSON(w, okStatus, task.Result)
	case <-exeTimeout.C:
		WriteJSON(w, http.StatusRequestTimeout, task)
	}

	return nil
}
71894
func (m *Manager) CreateTransientAsync(name string, query string, userCredentials *cli.UserCredentials) *tasks.Task {
	if name == "" {
		panic("name must be present")
	}
	if query == "" {
		panic("query must be present")
	}

	return m.client.CreateTransient(m.httpEndpoint, name, query, userCredentials)
}
1125634
func (ds *DataSource) tryToGetDualTask() (task, error) {
	for _, cond := range ds.pushedDownConds {
		if con, ok := cond.(*expression.Constant); ok && con.DeferredExpr == nil {
			result, _, err := expression.EvalBool(ds.ctx, []expression.Expression{cond}, chunk.Row{})
			if err != nil {
				return nil, err
			}
			if !result {
				dual := PhysicalTableDual{}.Init(ds.ctx, ds.stats)
				dual.SetSchema(ds.schema)
				return &rootTask{
					p: dual,
				}, nil
			}
		}
	}
	return nil, nil
}
420708
func (a *ExecStmt) buildExecutor(ctx sessionctx.Context) (Executor, error) {
	if _, ok := a.Plan.(*plannercore.Execute); !ok {
		// Do not sync transaction for Execute statement, because the real optimization work is done in
		// "ExecuteExec.Build".
		useMaxTS, err := IsPointGetWithPKOrUniqueKeyByAutoCommit(ctx, a.Plan)
		if err != nil {
			return nil, err
		}
		if useMaxTS {
			logutil.Logger(context.Background()).Debug("init txnStartTS with MaxUint64", zap.Uint64("conn", ctx.GetSessionVars().ConnectionID), zap.String("text", a.Text))
			err = ctx.InitTxnWithStartTS(math.MaxUint64)
		} else if ctx.GetSessionVars().SnapshotTS != 0 {
			if _, ok := a.Plan.(*plannercore.CheckTable); ok {
				err = ctx.InitTxnWithStartTS(ctx.GetSessionVars().SnapshotTS)
			}
		}
		if err != nil {
			return nil, err
		}

		stmtCtx := ctx.GetSessionVars().StmtCtx
		if stmtPri := stmtCtx.Priority; stmtPri == mysql.NoPriority {
			switch {
			case useMaxTS:
				stmtCtx.Priority = kv.PriorityHigh
			case a.Expensive:
				stmtCtx.Priority = kv.PriorityLow
			}
		}
	}
	if _, ok := a.Plan.(*plannercore.Analyze); ok && ctx.GetSessionVars().InRestrictedSQL {
		ctx.GetSessionVars().StmtCtx.Priority = kv.PriorityLow
	}

	b := newExecutorBuilder(ctx, a.InfoSchema)
	e := b.build(a.Plan)
	if b.err != nil {
		return nil, errors.Trace(b.err)
	}

	// ExecuteExec is not a real Executor, we only use it to build another Executor from a prepared statement.
	if executorExec, ok := e.(*ExecuteExec); ok {
		err := executorExec.Build()
		if err != nil {
			return nil, err
		}
		a.isPreparedStmt = true
		a.Plan = executorExec.plan
		e = executorExec.stmtExec
	}
	return e, nil
}
32724
func (b *Broker) nextDelayedTask(key string) (result []byte, err error) {
	conn := b.open()
	defer conn.Close()

	defer func() {
		// Return connection to normal state on error.
		// https://redis.io/commands/discard
		if err != nil {
			conn.Do("DISCARD")
		}
	}()

	var (
		items [][]byte
		reply interface{}
	)

	var pollPeriod = 500 // default poll period for delayed tasks
	if b.GetConfig().Redis != nil {
		configuredPollPeriod := b.GetConfig().Redis.DelayedTasksPollPeriod
		// the default period is 0, which bombards redis with requests, despite
		// our intention of doing the opposite
		if configuredPollPeriod > 0 {
			pollPeriod = configuredPollPeriod
		}
	}

	for {
		// Space out queries to ZSET so we don't bombard redis
		// server with relentless ZRANGEBYSCOREs
		time.Sleep(time.Duration(pollPeriod) * time.Millisecond)
		if _, err = conn.Do("WATCH", key); err != nil {
			return
		}

		now := time.Now().UTC().UnixNano()

		// https://redis.io/commands/zrangebyscore
		items, err = redis.ByteSlices(conn.Do(
			"ZRANGEBYSCORE",
			key,
			0,
			now,
			"LIMIT",
			0,
			1,
		))
		if err != nil {
			return
		}
		if len(items) != 1 {
			err = redis.ErrNil
			return
		}

		_ = conn.Send("MULTI")
		_ = conn.Send("ZREM", key, items[0])
		reply, err = conn.Do("EXEC")
		if err != nil {
			return
		}

		if reply != nil {
			result = items[0]
			break
		}
	}

	return
}
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()
}
223172
func (s *globalTaskService) TaskInspect(taskID int) *types.Task {
	s.RLock()
	defer s.RUnlock()
	if t, ok := s.tasks[taskID]; ok {
		return &t.Task
	}
	return nil
}
791387
func (c *ECS) WaitUntilTasksRunning(input *DescribeTasksInput) error {
	return c.WaitUntilTasksRunningWithContext(aws.BackgroundContext(), input)
}
893364
func NewDeleteTask(ctx *middleware.Context, handler DeleteTaskHandler) *DeleteTask {
	return &DeleteTask{Context: ctx, Handler: handler}
}
1131419
func (itc *internalTabletConn) BeginExecuteBatch(ctx context.Context, target *querypb.Target, queries []*querypb.BoundQuery, asTransaction bool, options *querypb.ExecuteOptions) ([]sqltypes.Result, int64, error) {
	transactionID, err := itc.Begin(ctx, target, options)
	if err != nil {
		return nil, 0, err
	}
	results, err := itc.ExecuteBatch(ctx, target, queries, asTransaction, transactionID, options)
	return results, transactionID, err
}
496953
func (q *query) BeginExecuteBatch(ctx context.Context, request *querypb.BeginExecuteBatchRequest) (response *querypb.BeginExecuteBatchResponse, err error) {
	defer q.server.HandlePanic(&err)
	ctx = callerid.NewContext(callinfo.GRPCCallInfo(ctx),
		request.EffectiveCallerId,
		request.ImmediateCallerId,
	)

	results, transactionID, err := q.server.BeginExecuteBatch(ctx, request.Target, request.Queries, request.AsTransaction, request.Options)
	if err != nil {
		// if we have a valid transactionID, return the error in-band
		if transactionID != 0 {
			return &querypb.BeginExecuteBatchResponse{
				Error:         vterrors.ToVTRPC(err),
				TransactionId: transactionID,
			}, nil
		}
		return nil, vterrors.ToGRPC(err)
	}
	return &querypb.BeginExecuteBatchResponse{
		Results:       sqltypes.ResultsToProto3(results),
		TransactionId: transactionID,
	}, nil
}
200030
func (conn *gRPCQueryClient) BeginExecuteBatch(ctx context.Context, target *querypb.Target, queries []*querypb.BoundQuery, asTransaction bool, options *querypb.ExecuteOptions) (results []sqltypes.Result, transactionID int64, err error) {
	conn.mu.RLock()
	defer conn.mu.RUnlock()
	if conn.cc == nil {
		return nil, 0, tabletconn.ConnClosed
	}

	req := &querypb.BeginExecuteBatchRequest{
		Target:            target,
		EffectiveCallerId: callerid.EffectiveCallerIDFromContext(ctx),
		ImmediateCallerId: callerid.ImmediateCallerIDFromContext(ctx),
		Queries:           queries,
		AsTransaction:     asTransaction,
		Options:           options,
	}

	reply, err := conn.c.BeginExecuteBatch(ctx, req)
	if err != nil {
		return nil, 0, tabletconn.ErrorFromGRPC(err)
	}
	if reply.Error != nil {
		return nil, reply.TransactionId, tabletconn.ErrorFromVTRPC(reply.Error)
	}
	return sqltypes.Proto3ToResults(reply.Results), reply.TransactionId, nil
}
1089217
func NewRetryTaskState(signature *Signature) *TaskState {
	return &TaskState{
		TaskUUID: signature.UUID,
		State:    StateRetry,
	}
}