// WaitTask blocks until the task identified by the given taskID is completed on
// Algolia engine.
761231
func (i *Index) WaitTask(taskID int) error {
	return waitWithRetry(func() (bool, error) {
		res, err := i.GetStatus(taskID)
		if err != nil {
			return true, err
		}
		if res.Status == "published" {
			return true, nil
		}
		return false, nil
	})
}
587613
func (c *EC2) WaitUntilConversionTaskCompleted(input *DescribeConversionTasksInput) error {
	return c.WaitUntilConversionTaskCompletedWithContext(aws.BackgroundContext(), input)
}
339356
func (_m *ContainerRuntimeExecutor) Wait(containerID string) error {
	ret := _m.Called(containerID)

	var r0 error
	if rf, ok := ret.Get(0).(func(string) error); ok {
		r0 = rf(containerID)
	} else {
		r0 = ret.Error(0)
	}

	return r0
}
182322
func (c *Chain) WaitReady() error {
	if err := c.isRunning(); err != nil {
		return err
	}

	select {
	case c.submitC <- nil:
	case <-c.doneC:
		return errors.Errorf("chain is stopped")
	}

	return nil
}
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}
}
999516
func (t *Task) Delay(c *Client, d time.Duration) error {
	return c.Delay(t, d)
}
1327045
func (*NewGoroutineDispatcher) DispatchTask(task *swf.PollForActivityTaskOutput, handler func(*swf.PollForActivityTaskOutput)) {
	go handler(task)
}
1105563
func (mq *BoltDbMQ) delayTask(job *models.Task) (*models.Task, error) {
	readyAt := time.Now().Add(time.Duration(job.Delay) * time.Second)
	err := mq.db.Update(func(tx *bolt.Tx) error {
		b := tx.Bucket(delayQueueName)
		id, _ := b.NextSequence()
		buf, err := json.Marshal(job)
		if err != nil {
			return err
		}

		key := msgKey(id)
		err = b.Put(key, buf)
		if err != nil {
			return err
		}

		pb := make([]byte, 4)
		binary.BigEndian.PutUint32(pb[:], uint32(*job.Priority))
		reservation := resKey(key, readyAt)
		return b.Put(reservation, pb)
	})
	return job, err
}
785651
func (s *Scheduler) ExecuteTaskGroup(ctx context.Context,
	fc *flow.Flow,
	taskGroupStatus *pb.FlowExecutionStatus_TaskGroup,
	wg *sync.WaitGroup,
	taskGroup *plan.TaskGroup,
	bid float64, relatedFiles []resource.FileResource) {

	defer wg.Done()

	tasks := taskGroup.Tasks
	lastTask := tasks[len(tasks)-1]
	if tasks[0].Step.IsOnDriverSide {
		// these should be only one task on the driver side
		if err := taskGroupStatus.Track(func(exeStatus *pb.FlowExecutionStatus_TaskGroup_Execution) error {
			return s.localExecute(ctx, fc, exeStatus, lastTask, wg)
		}); err != nil {
			log.Fatalf("Failed to execute on driver side: %v", err)
		}
		return
	}
	if !needsInputFromDriver(tasks[0]) {
		// wait until inputs are registed
		s.shardLocator.waitForInputDatasetShardLocations(tasks[0])
	}
	if isInputOnDisk(tasks[0]) && !isRestartableTasks(tasks) {
		// for non-restartable taskGroup, wait until on disk inputs are completed
		for _, stepGroup := range taskGroup.ParentStepGroup.Parents {
			stepGroup.WaitForAllTasksToComplete()
		}
	}

	// fmt.Printf("inputs of %s is %s\n", tasks[0].Name(), s.allInputLocations(tasks[0]))

	pickedServerChan := make(chan market.Supply, 1)
	s.Market.AddDemand(market.Requirement(taskGroup), bid, pickedServerChan)

	// get assigned executor location
	supply := <-pickedServerChan
	allocation := supply.Object.(*pb.Allocation)
	defer s.Market.ReturnSupply(supply)

	if needsInputFromDriver(tasks[0]) {
		// tell the driver to write to me
		for _, shard := range tasks[0].InputShards {
			// println("registering", shard.Name(), "at", allocation.Location.URL())
			s.setShardLocation(shard, pb.DataLocation{
				Name:     shard.Name(),
				Location: allocation.Location,
				OnDisk:   shard.Dataset.GetIsOnDiskIO(),
			})
		}
	}

	for _, shard := range lastTask.OutputShards {
		// println("registering", shard.Name(), "at", allocation.Location.URL(), "onDisk", shard.Dataset.GetIsOnDiskIO())
		s.setShardLocation(shard, pb.DataLocation{
			Name:     shard.Name(),
			Location: allocation.Location,
			OnDisk:   shard.Dataset.GetIsOnDiskIO(),
		})
	}

	// send driver code only when using go mapper reducer
	var hasGoCode bool
	for _, t := range tasks {
		hasGoCode = hasGoCode || t.Step.IsGoCode
	}
	if hasGoCode {
		relatedFiles = append(relatedFiles, resource.FileResource{os.Args[0], "."})
	}

	if len(relatedFiles) > 0 {
		err := withClient(allocation.Location.URL(), func(client pb.GleamAgentClient) error {
			for _, relatedFile := range relatedFiles {
				err := sendRelatedFile(ctx, client, fc.HashCode, relatedFile)
				if err != nil {
					taskGroup.MarkStop(err)
					return err
				}
			}
			return nil
		})
		if err != nil {
			log.Fatalf("Failed to send related files: %v", err)
		}
	}

	fn := func() error {
		err := taskGroupStatus.Track(func(exeStatus *pb.FlowExecutionStatus_TaskGroup_Execution) error {
			return s.remoteExecuteOnLocation(ctx, fc, taskGroupStatus, exeStatus, taskGroup, allocation, wg)
		})
		if err != nil {
			log.Printf("Failed to remoteExecuteOnLocation %v: %v", allocation, err)
		}
		taskGroup.MarkStop(err)
		return err
	}

	util.ExecuteWithCleanup(
		ctx,
		func() error {
			if isRestartableTasks(tasks) {
				return util.TimeDelayedRetry(fn, time.Minute, 3*time.Minute)
			} else {
				return fn()
			}
		},
		func() {
			var w sync.WaitGroup
			for _, shard := range lastTask.OutputShards {
				w.Add(1)
				// println("deleting", shard.Name(), "from", allocation.Location.URL())
				go func(shard *flow.DatasetShard) {
					defer w.Done()
					if err := sendDeleteRequest(allocation.Location.URL(), &pb.DeleteDatasetShardRequest{
						Name: shard.Name(),
					}); err != nil {
						println("Purging dataset error:", err.Error())
					}
				}(shard)
			}
			w.Wait()
		},
	)

}
22622
func (ar *allocRunner) runTasks() {
	for _, task := range ar.tasks {
		go task.Run()
	}

	for _, task := range ar.tasks {
		<-task.WaitCh()
	}
}
316868
func (s *SnapshotCreateService) WaitForCompletion(waitForCompletion bool) *SnapshotCreateService {
	s.waitForCompletion = &waitForCompletion
	return s
}
1339891
func GetTask(tx *bolt.Tx, id string) (*api.Task, error) {
	var t api.Task

	if err := withTaskBucket(tx, id, func(bkt *bolt.Bucket) error {
		p := bkt.Get(bucketKeyData)
		if p == nil {
			return errTaskUnknown
		}

		return proto.Unmarshal(p, &t)
	}); err != nil {
		return nil, err
	}

	return &t, nil
}
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
}
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
}
662659
func (f *asyncFlusher) waitAll() (int, error) {
	close(f.chunks)
	f.wg.Wait()
	if len(f.errs) == 0 {
		return f.rowsSent, nil
	}
	return f.rowsSent, f.errs
}
542411
func TaskWaitAllC(ctx types.Context, taskIDs ...int) <-chan int {
	return getTaskService(ctx).TaskWaitAllC(taskIDs...)
}
536056
func AddTask(t *automationpb.TaskContainer, taskName string, parameters map[string]string) {
	t.ParallelTasks = append(t.ParallelTasks, NewTask(taskName, parameters))
}