get which task should tuple be sent to
491237
public List<Integer> grouper(List<Object> values) {
        if (GrouperType.global.equals(groupType)) {
            // send to task which taskId is 0
            return JStormUtils.mk_list(outTasks.get(0));
        } else if (GrouperType.fields.equals(groupType)) {
            // field grouping
            return fieldsGrouper.grouper(values);
        } else if (GrouperType.all.equals(groupType)) {
            // send to every task
            return outTasks;
        } else if (GrouperType.shuffle.equals(groupType)) {
            // random, but the random is different from none
            return shuffer.grouper(values);
        } else if (GrouperType.none.equals(groupType)) {
            int rnd = Math.abs(random.nextInt() % outTasks.size());
            return JStormUtils.mk_list(outTasks.get(rnd));
        } else if (GrouperType.custom_obj.equals(groupType)) {
            return customGrouper.grouper(values);
        } else if (GrouperType.custom_serialized.equals(groupType)) {
            return customGrouper.grouper(values);
        } else {
            LOG.warn("Unsupported group type");
        }

        return new ArrayList<>();
    }
1031537
@Service
    public List<Task> getListOfPendingTask(String taskType, Integer start, Integer count) {
        List<Task> tasks = executionService.getPendingTasksForTaskType(taskType);
        int total = start + count;
        total = (tasks.size() > total) ? total : tasks.size();
        if (start > tasks.size()) start = tasks.size();
        return tasks.subList(start, total);
    }
910192
def _task_batcher(tasks, batch_size=None):
    
    from itertools import izip_longest

    if not batch_size:
        batch_size = DEFAULT_TASK_BATCH_SIZE

    # Ensure the batch size is under the task api limit.
    batch_size = min(batch_size, 100)

    args = [iter(tasks)] * batch_size
    return ([task for task in group if task] for group in izip_longest(*args))
1314533
private Tuple getNextTuple()
    {
    if( nextTuple_ == null)
      {
      // Still traversing values for current start variable?
      int tupleSize = getTupleSize();
      if( !(values_ != null && values_.hasNext()))
        {
        // No, still traversing subtuples for current start variable?
        values_ = null;
        subTuple_ = null;
        if( !(subTuples_ != null && subTuples_.hasNext()))
          {
          // No, advance to next start variable.
          subTuples_ = null;
          varStart_++;
          if( tupleSize > 1)
            {
            for( ;
                 varStart_ < varEnd_
                   &&
                   (subTuples_ = new TupleIterator( tupleSize - 1, varDefs_.subList( varStart_ + 1, varDefs_.size()), randSeq_))
                   .hasNext() == false;

                 varStart_++);
            }
          }
        }

      // More tuples remaining?
      if( varStart_ < varEnd_)
        {
        VarDef startVarDef = varDefs_.get( varStart_);
        if( values_ == null)
          {
          // Resume traversing values for current start variable.
          values_ = RandSeq.reorderIf( randSeq_, startVarDef.getValidValues());
          if( !values_.hasNext())
            {
            throw new IllegalStateException( "Can't complete tuples -- no valid values defined for var=" + startVarDef);
            }
          if( tupleSize > 1)
            {
            // Advance to next subtuple for current start variable.
            subTuple_ = subTuples_.next();
            }
          }

        nextTuple_ = new Tuple();
        nextTuple_.add( new VarBindingDef( startVarDef, values_.next()));
        if( subTuple_ != null)
          {
          nextTuple_.addAll( subTuple_);
          }
        }
      }

    return nextTuple_;
    }
825800
def has_task_of_type(self, typ):
    """"""
    for task_type in self.task_types():
      if issubclass(task_type, typ):
        return True
    return False
302955
def catch_result(task_func):
    

    @functools.wraps(task_func, assigned=available_attrs(task_func))
    def dec(*args, **kwargs):
        # inicialize
        orig_stdout = sys.stdout
        sys.stdout = content = StringIO()
        task_response = task_func(*args, **kwargs)
        # catch
        sys.stdout = orig_stdout
        content.seek(0)
        # propagate to the response
        task_response['stdout'] = content.read()
        return task_response
    return dec
398985
def _taskdict(task):
    '''
    
    '''
    if task is None:
        task = asyncio.current_task()

    assert task
    taskvars = getattr(task, '_syn_taskvars', None)

    if taskvars is None:
        taskvars = varinit(task)

    return taskvars
1236819
def get_task(self, name):
        """"""
        if name in self.tasks.keys():
            task = self.tasks[name]
        else:
            task = Task(name)
            self.tasks[name] = task
        return task
902064
def help(self, context):
        
        context.info('%s\n%s [global options] [task] [task options]...\n' % (self.name, sys.argv[0]))

        def print_parameter(prn, p):
            if p.description:
                suffix = '    - {0.description}'.format(p)
            else:
                suffix = ''
            if p.constraint is bool:
                prn('    {0.arg_name}'.format(p) + suffix)
            else:
                prn('    {0.arg_name} [{0.value}]'.format(p) + suffix)

        context.info('Global options:')
        for parameter in self.context_parameters.values():
            print_parameter(context.info, parameter)
        context.info()

        context.info('Commands:')
        name_template = '  {0.name:<%d}' % min(max(map(len, self.available_tasks)), 20)
        for task in self.available_tasks.values():
            printer = context.debug if task.hidden else context.info
            if task.description:
                printer((name_template + ' - {0.description}').format(task))
            else:
                printer(name_template.format(task))
            for parameter in task.parameters.values():
                print_parameter(printer, parameter)
955908
def get_valid_task_types():
    
    return frozendict({
        'scriptworker': verify_scriptworker_task,
        'balrog': verify_balrog_task,
        'beetmover': verify_beetmover_task,
        'bouncer': verify_bouncer_task,
        'build': verify_build_task,
        'l10n': verify_build_task,
        'repackage': verify_build_task,
        'action': verify_parent_task,
        'decision': verify_parent_task,
        'docker-image': verify_docker_image_task,
        'pushapk': verify_pushapk_task,
        'pushsnap': verify_pushsnap_task,
        'shipit': verify_shipit_task,
        'signing': verify_signing_task,
        'partials': verify_partials_task,
    })
491016
def get(self):
        ''''''
        if self.bucket.get() < 1:
            return None
        now = time.time()
        self.mutex.acquire()
        try:
            task = self.priority_queue.get_nowait()
            self.bucket.desc()
        except Queue.Empty:
            self.mutex.release()
            return None
        task.exetime = now + self.processing_timeout
        self.processing.put(task)
        self.mutex.release()
        return task.taskid
1226351
def _apply_task(task: Task, args: Tuple, kwargs: Dict[str, Any]) -> Any:
        """"""
        if args is None:
            args = ()
        if kwargs is None:
            kwargs = {}

        start = monotonic()
        try:
            return task.apply(*args, **kwargs)
        finally:
            delta = monotonic() - start
            logger.info("%s finished in %i seconds." % (task.name, delta))
971630
def _get_avaliable_tasks(self):
        """"""
        base_task = posixpath.join(self._queue_path, self.TASK_PREFIX)
        tasks = self._client.kv.find(prefix=base_task)
        return sorted(tasks.items())
629388
def task_family(cls):
        
        if not cls.get_task_namespace():
            return cls.__name__
        else:
            return "{}.{}".format(cls.get_task_namespace(), cls.__name__)
222825
def task_config(self, task: Task) -> Any:
        ''''''

        return self.get(task.__class__.__name__)
1305809
function defineProcessBatchTask() {
    const taskDef = TaskDef.defineTask(processBatch.name, processBatch, batchAndCancellableSettings);
    taskDef.defineSubTask(executeAllProcessOneTasks.name, executeAllProcessOneTasks, batchAndCancellableSettings);
    taskDef.defineSubTask(executeAllProcessAllTasks.name, executeAllProcessAllTasks, batchAndCancellableSettings);
    taskDef.defineSubTask(discardUnusableRecords.name, discardUnusableRecords, batchAndUnusableRecordsSettings);
    taskDef.defineSubTask(preFinaliseBatch.name, preFinaliseBatch, batchSettings);
    return taskDef;
  }
242300
private List<Integer> admitSpoutTuple(String streamId, List<Object> tuple,
                                        Object messageId, Integer emitDirectTaskId) {
    // No need to send tuples if it is already terminated
    if (getPhysicalPlanHelper().isTerminatedComponent()) {
      return null;
    }

    // Start construct the data tuple
    HeronTuples.HeronDataTuple.Builder bldr = initTupleBuilder(streamId, tuple, emitDirectTaskId);

    if (messageId != null) {
      RootTupleInfo tupleInfo = new RootTupleInfo(streamId, messageId);
      if (ackEnabled) {
        // This message is rooted
        HeronTuples.RootId.Builder rtbldr = establishRootId(tupleInfo);
        bldr.addRoots(rtbldr);
      } else {
        immediateAcks.offer(tupleInfo);
      }
    }

    sendTuple(bldr, streamId, tuple);

    // TODO:- remove this after changing the API
    return null;
  }