// RunTasks is a wrapper for running all tasks from the list
421106
func RunTasks(path string, tasksToExlude []string) (successTasks []map[string]interface{}, failedTasks []map[string]interface{}) {
	var wg sync.WaitGroup
	sourcePath = path
	sourceGoPath = strings.Replace(sourcePath, os.Getenv("GOPATH")+"/src/", "", 1)

	excludeTasks(&checkList, tasksToExlude)

	wg.Add(len(checkList))
	for _, task := range checkList {
		go func(task checkItem) {
			if ok := task.run(); ok {
				successTasks = append(successTasks, structs.Map(task))
			} else {
				failedTasks = append(failedTasks, structs.Map(task))
			}
			wg.Done()
		}(task)
	}

	wg.Wait()
	return successTasks, failedTasks
}
969639
func (s *AttachedScheduler) Tasks(ctx context.Context, app string) ([]*twelvefactor.Task, error) {
	if !s.ShowAttached {
		return s.Scheduler.Tasks(ctx, app)
	}

	type instancesResult struct {
		instances []*twelvefactor.Task
		err       error
	}

	ch := make(chan instancesResult, 1)
	go func() {
		attachedInstances, err := s.dockerScheduler.InstancesFromAttachedRuns(ctx, app)
		ch <- instancesResult{attachedInstances, err}
	}()

	instances, err := s.Scheduler.Tasks(ctx, app)
	if err != nil {
		return instances, err
	}

	result := <-ch
	if err := result.err; err != nil {
		return instances, err
	}

	return append(instances, result.instances...), nil
}
1346561
func (ar *allocRunner) Run() {
	// Close the wait channel on return
	defer close(ar.waitCh)

	// Start the task state update handler
	go ar.handleTaskStateUpdates()

	// Start the alloc update handler
	go ar.handleAllocUpdates()

	// If task update chan has been closed, that means we've been shutdown.
	select {
	case <-ar.taskStateUpdateHandlerCh:
		return
	default:
	}

	// Run the prestart hooks if non-terminal
	if ar.shouldRun() {
		if err := ar.prerun(); err != nil {
			ar.logger.Error("prerun failed", "error", err)
			goto POST
		}
	}

	// Run the runners (blocks until they exit)
	ar.runTasks()

POST:
	// Run the postrun hooks
	if err := ar.postrun(); err != nil {
		ar.logger.Error("postrun failed", "error", err)
	}

}
1109878
func (g *TaskGroup) AddTask(t *Task) *TaskGroup {
	g.Tasks = append(g.Tasks, t)
	return g
}
808309
func ExpandBatchPoolStartTask(list []interface{}) (*batch.StartTask, error) {
	if len(list) == 0 {
		return nil, fmt.Errorf("Error: batch pool start task should be defined")
	}

	startTaskValue := list[0].(map[string]interface{})

	startTaskCmdLine := startTaskValue["command_line"].(string)
	maxTaskRetryCount := int32(startTaskValue["max_task_retry_count"].(int))
	waitForSuccess := startTaskValue["wait_for_success"].(bool)

	userIdentityList := startTaskValue["user_identity"].([]interface{})
	if len(userIdentityList) == 0 {
		return nil, fmt.Errorf("Error: batch pool start task user identity should be defined")
	}

	userIdentityValue := userIdentityList[0].(map[string]interface{})
	userIdentity := batch.UserIdentity{}

	if autoUserValue, ok := userIdentityValue["auto_user"]; ok {
		autoUser := autoUserValue.([]interface{})
		if len(autoUser) != 0 {
			autoUserMap := autoUser[0].(map[string]interface{})
			userIdentity.AutoUser = &batch.AutoUserSpecification{
				ElevationLevel: batch.ElevationLevel(autoUserMap["elevation_level"].(string)),
				Scope:          batch.AutoUserScope(autoUserMap["scope"].(string)),
			}

		}
	} else if userNameValue, ok := userIdentityValue["username"]; ok {
		userName := userNameValue.(string)
		userIdentity.UserName = &userName
	} else {
		return nil, fmt.Errorf("Error: either auto_user or user_name should be speicfied for Batch pool start task")
	}

	resourceFileList := startTaskValue["resource_file"].([]interface{})
	resourceFiles := make([]batch.ResourceFile, 0)
	for _, resourceFileValueTemp := range resourceFileList {
		resourceFileValue := resourceFileValueTemp.(map[string]interface{})
		resourceFile := batch.ResourceFile{}
		if v, ok := resourceFileValue["auto_storage_container_name"]; ok {
			autoStorageContainerName := v.(string)
			if autoStorageContainerName != "" {
				resourceFile.AutoStorageContainerName = &autoStorageContainerName
			}
		}
		if v, ok := resourceFileValue["storage_container_url"]; ok {
			storageContainerURL := v.(string)
			if storageContainerURL != "" {
				resourceFile.StorageContainerURL = &storageContainerURL
			}
		}
		if v, ok := resourceFileValue["http_url"]; ok {
			httpURL := v.(string)
			if httpURL != "" {
				resourceFile.HTTPURL = &httpURL
			}
		}
		if v, ok := resourceFileValue["blob_prefix"]; ok {
			blobPrefix := v.(string)
			if blobPrefix != "" {
				resourceFile.BlobPrefix = &blobPrefix
			}
		}
		if v, ok := resourceFileValue["file_path"]; ok {
			filePath := v.(string)
			if filePath != "" {
				resourceFile.FilePath = &filePath
			}
		}
		if v, ok := resourceFileValue["file_mode"]; ok {
			fileMode := v.(string)
			if fileMode != "" {
				resourceFile.FileMode = &fileMode
			}
		}
		resourceFiles = append(resourceFiles, resourceFile)
	}

	startTask := &batch.StartTask{
		CommandLine:       &startTaskCmdLine,
		MaxTaskRetryCount: &maxTaskRetryCount,
		WaitForSuccess:    &waitForSuccess,
		UserIdentity:      &userIdentity,
		ResourceFiles:     &resourceFiles,
	}

	// populate environment settings, if defined
	if environment := startTaskValue["environment"]; environment != nil {
		envMap := environment.(map[string]interface{})
		envSettings := make([]batch.EnvironmentSetting, 0)

		for k, v := range envMap {
			theValue := v.(string)
			theKey := k
			envSetting := batch.EnvironmentSetting{
				Name:  &theKey,
				Value: &theValue,
			}

			envSettings = append(envSettings, envSetting)
		}

		startTask.EnvironmentSettings = &envSettings
	}

	return startTask, nil
}
212636
func (c oauthClient) DeleteAllTasks() error {
	tasks, err := c.Tasks()
	if err != nil {
		return err
	}

	taskCount := len(tasks)
	c.logger.Debug(
		"delete-all-tasks",
		map[string]interface{}{"taskCount": taskCount},
	)

	idErrChan := make(chan idErr, taskCount)
	for _, f := range tasks {
		go func(task wl.Task) {
			c.logger.Debug(
				"delete-all-tasks - deleting task",
				map[string]interface{}{"taskID": task.ID},
			)
			err := c.DeleteTask(task)
			idErrChan <- idErr{idType: "task", id: task.ID, err: err}
		}(f)
	}

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

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

	return nil
}
864175
func (c *Cluster) GetTasks(options apitypes.TaskListOptions) ([]types.Task, error) {
	var r *swarmapi.ListTasksResponse

	if err := c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
		filterTransform := func(filter filters.Args) error {
			if filter.Contains("service") {
				serviceFilters := filter.Get("service")
				for _, serviceFilter := range serviceFilters {
					service, err := getService(ctx, state.controlClient, serviceFilter, false)
					if err != nil {
						return err
					}
					filter.Del("service", serviceFilter)
					filter.Add("service", service.ID)
				}
			}
			if filter.Contains("node") {
				nodeFilters := filter.Get("node")
				for _, nodeFilter := range nodeFilters {
					node, err := getNode(ctx, state.controlClient, nodeFilter)
					if err != nil {
						return err
					}
					filter.Del("node", nodeFilter)
					filter.Add("node", node.ID)
				}
			}
			if !filter.Contains("runtime") {
				// default to only showing container tasks
				filter.Add("runtime", "container")
				filter.Add("runtime", "")
			}
			return nil
		}

		filters, err := newListTasksFilters(options.Filters, filterTransform)
		if err != nil {
			return err
		}

		r, err = state.controlClient.ListTasks(
			ctx,
			&swarmapi.ListTasksRequest{Filters: filters},
			grpc.MaxCallRecvMsgSize(defaultRecvSizeForListResponse),
		)
		return err
	}); err != nil {
		return nil, err
	}

	tasks := make([]types.Task, 0, len(r.Tasks))
	for _, task := range r.Tasks {
		t, err := convert.TaskFromGRPC(*task)
		if err != nil {
			return nil, err
		}
		tasks = append(tasks, t)
	}
	return tasks, nil
}
1163246
func WalkTasks(tx *bolt.Tx, fn func(task *api.Task) error) error {
	bkt := getTasksBucket(tx)
	if bkt == nil {
		return nil
	}

	return bkt.ForEach(func(k, v []byte) error {
		tbkt := bkt.Bucket(k)

		p := tbkt.Get(bucketKeyData)
		var t api.Task
		if err := proto.Unmarshal(p, &t); err != nil {
			return err
		}

		return fn(&t)
	})
}
1104399
func (s *StartTaskOutput) SetTasks(v []*Task) *StartTaskOutput {
	s.Tasks = v
	return s
}
365799
func (p *ParallelRunner) Run() error {
	// default value is 0. The task will not run in this case.
	var parallelNum int
	switch p.concurrencyLevel {
	case Sequential:
		parallelNum = 1
	case Parallel:
		parallelNum = len(p.tasks)
	default:
		log.Fatalf("BUG: Invalid concurrency level: %v", p.concurrencyLevel)
	}
	// sem is a channel used to control the level of concurrency.
	sem := make(chan bool, parallelNum)
	wg := sync.WaitGroup{}
	for i, task := range p.tasks {
		if isTaskSucceeded(task) {
			continue
		}

		sem <- true
		if p.enableApprovals && !isTaskRunning(task) {
			p.waitForApproval(i)
		}
		select {
		case <-p.ctx.Done():
			// Break this run and return early. Do not try to execute any subsequent tasks.
			log.Infof("Workflow is cancelled, remaining tasks will be aborted")
			return nil
		default:
			wg.Add(1)
			go func(t *workflowpb.Task) {
				defer wg.Done()
				p.setUIMessage(fmt.Sprintf("Launch task: %v.", t.Id))
				defer func() { <-sem }()
				p.executeTask(t)
			}(task)
		}
	}
	wg.Wait()

	if p.enableApprovals {
		p.clearPhaseActions()
	}
	// TODO(yipeiw): collect error message from tasks.Error instead,
	// s.t. if the task is retried, we can update the error
	return nil
}
1256326
func (s *Scheduler) tasks(app string) ([]*ecs.Task, error) {
	services, err := s.Services(app)
	if err != nil {
		return nil, err
	}

	var arns []*string

	// Find all of the tasks started by the ECS services.
	for process, serviceArn := range services {
		id, err := arn.ResourceID(serviceArn)
		if err != nil {
			return nil, err
		}

		var taskArns []*string
		if err := s.ecs.ListTasksPages(&ecs.ListTasksInput{
			Cluster:     aws.String(s.Cluster),
			ServiceName: aws.String(id),
		}, func(resp *ecs.ListTasksOutput, lastPage bool) bool {
			taskArns = append(taskArns, resp.TaskArns...)
			return true
		}); err != nil {
			return nil, fmt.Errorf("error listing tasks for %s: %v", process, err)
		}

		if len(taskArns) == 0 {
			continue
		}

		arns = append(arns, taskArns...)
	}

	// Find all of the tasks started by Run.
	if err := s.ecs.ListTasksPages(&ecs.ListTasksInput{
		Cluster:   aws.String(s.Cluster),
		StartedBy: aws.String(app),
	}, func(resp *ecs.ListTasksOutput, lastPage bool) bool {
		arns = append(arns, resp.TaskArns...)
		return true
	}); err != nil {
		return nil, fmt.Errorf("error listing tasks started by %s: %v", app, err)
	}

	var tasks []*ecs.Task
	for _, chunk := range chunkStrings(arns, MaxDescribeTasks) {
		resp, err := s.ecs.DescribeTasks(&ecs.DescribeTasksInput{
			Cluster: aws.String(s.Cluster),
			Tasks:   chunk,
		})
		if err != nil {
			return nil, fmt.Errorf("error describing %d tasks: %v", len(chunk), err)
		}

		tasks = append(tasks, resp.Tasks...)
	}

	return tasks, nil
}
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
}
1220830
func (s *Scheduler) RunAll() {
	for i := 0; i < s.size; i++ {
		s.jobs[i].run()
	}
}
680991
func (r *marathonClient) KillTask(taskID string, opts *KillTaskOpts) (*Task, error) {
	appName := taskID[0:strings.LastIndex(taskID, ".")]
	appName = strings.Replace(appName, "_", "/", -1)
	taskID = strings.Replace(taskID, "/", "_", -1)

	path := fmt.Sprintf("%s/%s/tasks/%s", marathonAPIApps, appName, taskID)
	path, err := addOptions(path, opts)
	if err != nil {
		return nil, err
	}

	wrappedTask := new(struct {
		Task Task `json:"task"`
	})

	if err := r.apiDelete(path, nil, wrappedTask); err != nil {
		return nil, err
	}

	return &wrappedTask.Task, nil
}
547225
func (cluster *Cluster) RunBatch(bat *Batch) ([]interface{}, error) {
    for i := range bat.batches {
	go doBatch(&bat.batches[i])
    }

    for i := range bat.batches {
	<-bat.batches[i].done
    }

    var replies []interface{}
    for _, i := range bat.index {
	if bat.batches[i].err != nil {
	    return nil, bat.batches[i].err
	}

	replies = append(replies, bat.batches[i].cmds[0].reply)
	bat.batches[i].cmds = bat.batches[i].cmds[1:]
    }

    return replies, nil
}
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
}
669195
func StartWorkers(ctx context.Context, rnr *Runner, tasks <-chan task.Request) {
	var wg sync.WaitGroup
	defer wg.Wait()
	var hcmgr htfnmgr

	for {
		select {
		case <-ctx.Done():
			return
		case task := <-tasks:
			p := hcmgr.getPipe(ctx, rnr, task.Config)
			if p == nil {
				wg.Add(1)
				go runTaskReq(rnr, &wg, task)
				continue
			}

			rnr.Start()
			select {
			case <-ctx.Done():
				return
			case p <- task:
				rnr.Complete()
			}
		}
	}
}