// ----------------------------------------------------------------------------
// redis tasks
// ----------------------------------------------------------------------------
328959
func doPing(id string, signal chan int, client redis.AsyncClient, cnt int) {
	var fr redis.FutureBool
	for i := 0; i < cnt; i++ {
		fr, _ = client.Ping()
	}
	fr.Get()
	signal <- 1
}
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
}
207386
func (c *Client) Cleanup(removeall bool) (Task, error) {
	task := Task{}
	r := c.NewRequest("POST", "/cleanup")
	var requestBody struct {
		Config struct {
			RemoveAll bool `json:"remove_all"`
		} `json:"config"`
	}
	requestBody.Config.RemoveAll = removeall
	b, err := json.Marshal(&requestBody)
	if err != nil {
		return task, err
	}
	r.body = bytes.NewBuffer(b)
	r.header["Content-Type"] = "application/json"
	resp, err := c.DoRequest(r)
	if err != nil {
		return task, err
	}

	defer resp.Body.Close()

	resBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		log.Printf("Error reading task request %v", resBody)
		return task, err
	}
	err = json.Unmarshal(resBody, &task)
	if err != nil {
		log.Printf("Error unmarshaling task %v", err)
	}
	return task, err
}
1087313
func (c *Client) resubscribe() {
	// Swap so that if we reconnect before all tasks are done,
	// we don't duplicate things.
	subs := c.subscribed
	c.subscribed = map[string][]*Listener{}

	for _, events := range subs {
		// We only need to subscribe on one event, so that the Redis
		// connection gets registered. We don't care about the others.
		c.tasks <- task{Listener: events[0], Action: subscribeAction, Force: true}
	}
}
276104
func (w *Worker) TaskCancel(taskId string) (err error) {
	_, err = w.tasks(taskId, "cancel").Request("POST", nil)
	return err
}
1220036
func (d *Dispatcher) Tasks(r *api.TasksRequest, stream api.Dispatcher_TasksServer) error {
	d.rpcRW.RLock()
	defer d.rpcRW.RUnlock()

	dctx, err := d.isRunningLocked()
	if err != nil {
		return err
	}

	nodeInfo, err := ca.RemoteNode(stream.Context())
	if err != nil {
		return err
	}
	nodeID := nodeInfo.NodeID

	fields := logrus.Fields{
		"node.id":      nodeID,
		"node.session": r.SessionID,
		"method":       "(*Dispatcher).Tasks",
	}
	if nodeInfo.ForwardedBy != nil {
		fields["forwarder.id"] = nodeInfo.ForwardedBy.NodeID
	}
	log.G(stream.Context()).WithFields(fields).Debug("")

	if _, err = d.nodes.GetWithSession(nodeID, r.SessionID); err != nil {
		return err
	}

	tasksMap := make(map[string]*api.Task)
	nodeTasks, cancel, err := store.ViewAndWatch(
		d.store,
		func(readTx store.ReadTx) error {
			tasks, err := store.FindTasks(readTx, store.ByNodeID(nodeID))
			if err != nil {
				return err
			}
			for _, t := range tasks {
				tasksMap[t.ID] = t
			}
			return nil
		},
		api.EventCreateTask{Task: &api.Task{NodeID: nodeID},
			Checks: []api.TaskCheckFunc{api.TaskCheckNodeID}},
		api.EventUpdateTask{Task: &api.Task{NodeID: nodeID},
			Checks: []api.TaskCheckFunc{api.TaskCheckNodeID}},
		api.EventDeleteTask{Task: &api.Task{NodeID: nodeID},
			Checks: []api.TaskCheckFunc{api.TaskCheckNodeID}},
	)
	if err != nil {
		return err
	}
	defer cancel()

	for {
		if _, err := d.nodes.GetWithSession(nodeID, r.SessionID); err != nil {
			return err
		}

		var tasks []*api.Task
		for _, t := range tasksMap {
			// dispatcher only sends tasks that have been assigned to a node
			if t != nil && t.Status.State >= api.TaskStateAssigned {
				tasks = append(tasks, t)
			}
		}

		if err := stream.Send(&api.TasksMessage{Tasks: tasks}); err != nil {
			return err
		}

		// bursty events should be processed in batches and sent out snapshot
		var (
			modificationCnt int
			batchingTimer   *time.Timer
			batchingTimeout <-chan time.Time
		)

	batchingLoop:
		for modificationCnt < modificationBatchLimit {
			select {
			case event := <-nodeTasks:
				switch v := event.(type) {
				case api.EventCreateTask:
					tasksMap[v.Task.ID] = v.Task
					modificationCnt++
				case api.EventUpdateTask:
					if oldTask, exists := tasksMap[v.Task.ID]; exists {
						// States ASSIGNED and below are set by the orchestrator/scheduler,
						// not the agent, so tasks in these states need to be sent to the
						// agent even if nothing else has changed.
						if equality.TasksEqualStable(oldTask, v.Task) && v.Task.Status.State > api.TaskStateAssigned {
							// this update should not trigger action at agent
							tasksMap[v.Task.ID] = v.Task
							continue
						}
					}
					tasksMap[v.Task.ID] = v.Task
					modificationCnt++
				case api.EventDeleteTask:
					delete(tasksMap, v.Task.ID)
					modificationCnt++
				}
				if batchingTimer != nil {
					batchingTimer.Reset(batchingWaitTime)
				} else {
					batchingTimer = time.NewTimer(batchingWaitTime)
					batchingTimeout = batchingTimer.C
				}
			case <-batchingTimeout:
				break batchingLoop
			case <-stream.Context().Done():
				return stream.Context().Err()
			case <-dctx.Done():
				return dctx.Err()
			}
		}

		if batchingTimer != nil {
			batchingTimer.Stop()
		}
	}
}
589452
func (c *RedisCacher) Flush() error {
	if c.occupyMode {
		return c.c.FlushDb().Err()
	}

	keys, err := c.c.HKeys(c.hsetName).Result()
	if err != nil {
		return err
	}
	if err = c.c.Del(keys...).Err(); err != nil {
		return err
	}
	return c.c.Del(c.hsetName).Err()
}
520391
func (worker *Worker) taskRetry(signature *tasks.Signature) error {
	// Update task state to RETRY
	if err := worker.server.GetBackend().SetStateRetry(signature); err != nil {
		return fmt.Errorf("Set state to 'retry' for task %s returned error: %s", signature.UUID, err)
	}

	// Decrement the retry counter, when it reaches 0, we won't retry again
	signature.RetryCount--

	// Increase retry timeout
	signature.RetryTimeout = retry.FibonacciNext(signature.RetryTimeout)

	// Delay task by signature.RetryTimeout seconds
	eta := time.Now().UTC().Add(time.Second * time.Duration(signature.RetryTimeout))
	signature.ETA = &eta

	log.WARNING.Printf("Task %s failed. Going to retry in %d seconds.", signature.UUID, signature.RetryTimeout)

	// Send the task back to the queue
	_, err := worker.server.SendTask(signature)
	return err
}
1294163
func Rename(oldName string, newName string) error {
	lock.Lock()
	defer lock.Unlock()

	oldName = applyNamespace(oldName)
	newName = applyNamespace(newName)

	if griftList[oldName] == nil {
		return fmt.Errorf("No task named %s defined!", oldName)
	}
	griftList[newName] = griftList[oldName]
	delete(griftList, oldName)
	return nil
}
176477
func (r *Redis) Gather(acc telegraf.Accumulator) error {
	if !r.initialized {
		err := r.init(acc)
		if err != nil {
			return err
		}
	}

	var wg sync.WaitGroup

	for _, client := range r.clients {
		wg.Add(1)
		go func(client Client) {
			defer wg.Done()
			acc.AddError(r.gatherServer(client, acc))
		}(client)
	}

	wg.Wait()
	return nil
}
539151
func (p *workerPool) work() {
	for task := range p.tasksChan {
		task.Run(&p.wg)
	}
}
344596
func (r *RedisCache) Flush() (err error) {
	if r.occupyMode {
		_, err = r.do("FLUSHDB")
		return
	}

	keys, err := redigo.MultiBulk(r.do("HKEYS", r.key(GC_HASH_KEY)))
	if err != nil {
		return
	}

	fmt.Println(keys)

	conn := r.pool.Get()
	defer conn.Close()

	_, err = conn.Do("DEL", keys...)

	return
}
906922
func (r *Redis) Sort(key string) *SortCommand {
	return &SortCommand{redis: r, key: key}
}
877603
func (r *marathonClient) AllTasks(opts *AllTasksOpts) (*Tasks, error) {
	path, err := addOptions(marathonAPITasks, opts)
	if err != nil {
		return nil, err
	}

	tasks := new(Tasks)
	if err := r.apiGet(path, nil, tasks); err != nil {
		return nil, err
	}

	return tasks, nil
}
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
}
828748
func (tg *TaskGroup) Canonicalize(job *Job) {
	// Ensure that an empty and nil map are treated the same to avoid scheduling
	// problems since we use reflect DeepEquals.
	if len(tg.Meta) == 0 {
		tg.Meta = nil
	}

	// Set the default restart policy.
	if tg.RestartPolicy == nil {
		tg.RestartPolicy = NewRestartPolicy(job.Type)
	}

	if tg.ReschedulePolicy == nil {
		tg.ReschedulePolicy = NewReschedulePolicy(job.Type)
	}

	// Canonicalize Migrate for service jobs
	if job.Type == JobTypeService && tg.Migrate == nil {
		tg.Migrate = DefaultMigrateStrategy()
	}

	// Set a default ephemeral disk object if the user has not requested for one
	if tg.EphemeralDisk == nil {
		tg.EphemeralDisk = DefaultEphemeralDisk()
	}

	for _, task := range tg.Tasks {
		task.Canonicalize(job, tg)
	}
}
598039
func Do(ctx context.Context, tasks ...Task) error {
	n := len(tasks)

	done := make(chan error, n)
	for i, task := range tasks {
		go func(task Task) {
			err := task()
			if err != nil {
				done <- errors.WithMessage(err, fmt.Sprintf("task %d", i+1))
				return
			}
			done <- nil
		}(task)
	}

	for i := 0; i < n; i++ {
		select {
		case err := <-done:
			if err != nil {
				return errors.WithStack(err)
			}
		case <-ctx.Done():
			return werrors.ErrCancelled
		}
	}
	return nil
}