// DoWait adds a task into the pool within a given duration if not accepted by then it returns with  an error
484714
func (w *WorkPool) DoWait(c interface{}, wo Work, t time.Duration) error {
	dow := doWork{c, wo}

	w.measureHealth()

	atomic.AddInt64(&w.pendingWork, 1)

	select {
	case w.tasks <- dow:
		atomic.AddInt64(&w.pendingWork, -1)
		return nil
	case <-time.After(t):
		atomic.AddInt64(&w.pendingWork, -1)
		return ErrWorkRequestDenied
	}

}
662267
func (f *PlatformFilter) SetTask(t *api.Task) bool {
	placement := t.Spec.Placement
	if placement != nil {
		// copy the platform information
		f.supportedPlatforms = placement.Platforms
		if len(placement.Platforms) > 0 {
			return true
		}
	}
	return false
}
382315
func (pool *LoadBalancedPool) DoWithTimeout(req *http.Request,
	timeout time.Duration) (*http.Response, error) {
	return pool.DoWithParams(req, DoParams{Timeout: timeout})
}
462734
func NewWaitingTask(msg string) *WaitingTask {
	ch := make(chan *Update, 1)
	ch <- &Update{
		S:  fmt.Sprintf("%s: ...", msg),
		At: time.Now(),
	}

	return &WaitingTask{ch: ch}
}
262332
func (r *Request) ProtoAtLeast(major, minor int) bool {
	return r.ProtoMajor > major ||
		r.ProtoMajor == major && r.ProtoMinor >= minor
}
539422
func ConnectWait(t time.Duration) Option {
	return func(o *Options) error {
		o.ConnectTimeout = t
		return nil
	}
}
888702
func (c *Client) Delay(t *Task, delay time.Duration) error {

	client, err := c.pool.Get()

	if err != nil {
		return err
	}
	defer client.Close()

	b, err := t.marshal()
	if err != nil {
		return fmt.Errorf("Could not marshal task: %s", err)
	}

	ar := disque.AddRequest{
		Job: disque.Job{
			Queue: qname(t.Name),
			Data:  b,
		},
		Timeout:   c.enqueueTimeout,
		Replicate: c.replicationFactor,
		Delay:     delay,
		TTL:       t.ttl,
		Retry:     t.retry,
	}

	t.jobId, err = client.Add(ar)

	return err
}
538930
func worker_task(name string, i int) {
	workername := fmt.Sprintf("Worker-%s-%d", name, i)

	worker, _ := zmq.NewSocket(zmq.REQ)
	defer worker.Close()
	worker.SetIdentity(workername)
	worker.Connect("ipc://" + name + "-localbe.ipc")

	//  Tell broker we're ready for work
	worker.SendMessage(WORKER_READY)

	//  Process messages as they arrive
	for {
		msg, err := worker.RecvMessage(0)
		if err != nil {
			fmt.Println("worker_task interrupted", name)
			break //  Interrupted
		}

		i := len(msg) - 1
		fmt.Printf("%s: %s\n", workername, msg[i])
		worker.SendMessage(msg[:i], "OK from "+workername)
	}
}
553644
func (p Pool) ParallelDo(tasks ...func() error) error {
	jobs := make(chan func() error, len(tasks))
	errs := make(chan error, len(tasks))
	wg := &sync.WaitGroup{}

	wg.Add(p.Count)
	for i := 0; i < p.Count; i++ {
		p.spawnWorker(jobs, errs, wg)
	}

	for _, task := range tasks {
		jobs <- task
	}

	close(jobs)

	wg.Wait()

	close(errs)

	var combinedErrors []error
	for e := range errs {
		combinedErrors = append(combinedErrors, e)
	}

	if len(combinedErrors) > 0 {
		return bosherr.NewMultiError(combinedErrors...)
	}

	return nil
}
552401
func (e *CElement) PrevWait() *CElement {
	for {
		e.mtx.RLock()
		prev := e.prev
		prevWg := e.prevWg
		removed := e.removed
		e.mtx.RUnlock()

		if prev != nil || removed {
			return prev
		}

		prevWg.Wait()
	}
}
760856
func (c oauthClient) Tasks() ([]wl.Task, error) {
	lists, err := c.Lists()
	if err != nil {
		return nil, err
	}

	listCount := len(lists)
	c.logger.Debug(
		"tasks",
		map[string]interface{}{"listCount": listCount},
	)

	tasksChan := make(chan []wl.Task, listCount)
	idErrChan := make(chan idErr, listCount)
	for _, l := range lists {
		go func(list wl.List) {
			c.logger.Debug(
				"tasks - getting tasks for list",
				map[string]interface{}{"listID": list.ID},
			)
			tasks, err := c.TasksForListID(list.ID)
			idErrChan <- idErr{idType: "list", id: list.ID, err: err}
			tasksChan <- tasks
		}(l)
	}

	e := multiIDErr{}
	for i := 0; i < listCount; i++ {
		idErr := <-idErrChan
		if idErr.err != nil {
			c.logger.Debug(
				"tasks - error received getting tasks for list",
				map[string]interface{}{"listID": idErr.id, "err": err},
			)
			e.addError(idErr)
		}
	}

	totalTasks := []wl.Task{}
	for i := 0; i < listCount; i++ {
		tasks := <-tasksChan
		totalTasks = append(totalTasks, tasks...)
	}

	if len(e.errors()) > 0 {
		return totalTasks, e
	}

	return totalTasks, nil
}
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)
}
1162947
func TaskCreate(tasktype TaskType) *Task {
	Ctask := DmTaskCreate(int(tasktype))
	if Ctask == nil {
		return nil
	}
	task := &Task{unmanaged: Ctask}
	runtime.SetFinalizer(task, (*Task).destroy)
	return task
}
662026
func (w *Worker) TaskQueue(tasks ...Task) (taskIds []string, err error) {
	outTasks := make([]map[string]interface{}, 0, len(tasks))

	for _, task := range tasks {
		thisTask := map[string]interface{}{
			"code_name": task.CodeName,
			"payload":   task.Payload,
			"priority":  task.Priority,
			"cluster":   task.Cluster,
			"label":     task.Label,
		}
		if task.Timeout != nil {
			thisTask["timeout"] = (*task.Timeout).Seconds()
		}
		if task.Delay != nil {
			thisTask["delay"] = int64((*task.Delay).Seconds())
		}

		outTasks = append(outTasks, thisTask)
	}

	in := map[string][]map[string]interface{}{"tasks": outTasks}
	out := struct {
		Tasks []struct {
			Id string `json:"id"`
		} `json:"tasks"`
		Msg string `json:"msg"`
	}{}

	err = w.tasks().Req("POST", &in, &out)
	if err != nil {
		return
	}

	taskIds = make([]string, 0, len(out.Tasks))
	for _, task := range out.Tasks {
		taskIds = append(taskIds, task.Id)
	}

	return
}
217709
func TaskArchival(c context.Context, state *coordinator.LogStreamState, delay time.Duration) error {
	// Now task the archival.
	state.Updated = clock.Now(c).UTC()
	state.ArchivalKey = []byte{'1'} // Use a fake key just to signal that we've tasked the archival.
	if err := ds.Put(c, state); err != nil {
		log.Fields{
			log.ErrorKey: err,
		}.Errorf(c, "Failed to Put() LogStream.")
		return grpcutil.Internal
	}

	project := string(coordinator.ProjectFromNamespace(state.Parent.Namespace()))
	id := string(state.ID())
	t, err := tqTask(&logdog.ArchiveTask{Project: project, Id: id})
	if err != nil {
		log.WithError(err).Errorf(c, "could not create archival task")
		return grpcutil.Internal
	}
	t.Delay = delay
	if err := taskqueue.Add(c, ArchiveQueueName, t); err != nil {
		log.WithError(err).Errorf(c, "could not task archival")
		return grpcutil.Internal
	}
	return nil

}
542411
func TaskWaitAllC(ctx types.Context, taskIDs ...int) <-chan int {
	return getTaskService(ctx).TaskWaitAllC(taskIDs...)
}
927387
func finishCopTask(ctx sessionctx.Context, task task) task {
	t, ok := task.(*copTask)
	if !ok {
		return task
	}
	// FIXME: When it is a double reading. The cost should be more expensive. The right cost should add the
	// `NetWorkStartCost` * (totalCount / perCountIndexRead)
	t.finishIndexPlan()
	if t.tablePlan != nil {
		t.cst += t.count() * netWorkFactor
	}
	newTask := &rootTask{
		cst: t.cst,
	}
	if t.indexPlan != nil && t.tablePlan != nil {
		p := PhysicalIndexLookUpReader{tablePlan: t.tablePlan, indexPlan: t.indexPlan}.Init(ctx)
		p.stats = t.tablePlan.statsInfo()
		if t.doubleReadNeedProj {
			schema := p.IndexPlans[0].(*PhysicalIndexScan).dataSourceSchema
			proj := PhysicalProjection{Exprs: expression.Column2Exprs(schema.Columns)}.Init(ctx, p.stats, nil)
			proj.SetSchema(schema)
			proj.SetChildren(p)
			newTask.p = proj
		} else {
			newTask.p = p
		}
	} else if t.indexPlan != nil {
		p := PhysicalIndexReader{indexPlan: t.indexPlan}.Init(ctx)
		p.stats = t.indexPlan.statsInfo()
		newTask.p = p
	} else {
		p := PhysicalTableReader{tablePlan: t.tablePlan}.Init(ctx)
		p.stats = t.tablePlan.statsInfo()
		newTask.p = p
	}
	return newTask
}