Run all the steps for the sample in sequence
960715
public function runSample()
    {
        $this->_setupOrderReference();
        $this->_confirmOrderReference();
        $amazonAuthorizationId = $this->_performAuthorization();
        $this->_waitUntilAuthorizationProcessingIsCompleted($amazonAuthorizationId);
        $this->_cancelOrder();
        $this->_getOrderReferenceDetails();
       
        print "Cancellation completed" . PHP_EOL;
    }
992162
def samples_to_batches(samples: Iterable, batch_size: int):
        """"""
        it = iter(samples)
        while True:
            with suppress(StopIteration):
                batch_in, batch_out = [], []
                for i in range(batch_size):
                    sample_in, sample_out = next(it)
                    batch_in.append(sample_in)
                    batch_out.append(sample_out)
            if not batch_in:
                raise StopIteration
            yield np.array(batch_in), np.array(batch_out)
348793
def run(self):
        """"""

        inside = 0
        for draws in range(1, self.data['samples']):
            # generate points and check whether they are inside the unit circle
            r1, r2 = (random(), random())
            if r1 ** 2 + r2 ** 2 < 1.0:
                inside += 1

            if draws % 1000 != 0:
                continue

            # debug
            yield self.emit('log', {'draws': draws, 'inside': inside})

            # calculate pi and its uncertainty given the current draws
            p = inside / draws
            pi = {
                'estimate': 4.0 * inside / draws,
                'uncertainty': 4.0 * math.sqrt(draws * p * (1.0 - p)) / draws,
            }

            # send status to frontend
            yield self.set_state(pi=pi)

        yield self.emit('log', {'action': 'done'})
1364832
def run_steps(node: Node, pipeline: RenderingPipeline, **args):
    """"""
    for step in pipeline.steps:
        result = step(node, pipeline=pipeline, **args)
        if isinstance(result, dict):
            args = {**args, **result}
141257
def run_pipeline_steps(steps, context):
    
    logger.debug("starting")
    assert isinstance(
        context, dict), "context must be a dictionary, even if empty {}."

    if steps is None:
        logger.debug("No steps found to execute.")
    else:
        step_count = 0

        for step in steps:
            step_instance = Step(step)
            step_instance.run_step(context)
            step_count += 1

        logger.debug(f"executed {step_count} steps")

    logger.debug("done")
277344
def run_step(context):
    
    logger.debug("started")

    context.clear()
    logger.info(f"Context wiped. New context size: {len(context)}")

    logger.debug("done")
1255997
protected void runAllStepsInLoop(List<GherkinConditionedLoopedStep> loopedSteps) throws TechnicalException {
        for (final GherkinConditionedLoopedStep loopedStep : loopedSteps) {
            final List<GherkinStepCondition> stepConditions = new ArrayList<>();
            final String[] expecteds = loopedStep.getExpected().split(";");
            final String[] actuals = loopedStep.getActual().split(";");
            if (actuals.length != expecteds.length) {
                throw new TechnicalException(Messages.getMessage(TechnicalException.TECHNICAL_EXPECTED_ACTUAL_SIZE_DIFFERENT));
            }
            for (int i = 0; i < expecteds.length; i++) {
                stepConditions.add(new GherkinStepCondition(loopedStep.getKey(), expecteds[i], actuals[i]));
            }
            boolean found = false;
            for (final Entry<String, Method> elem : Context.getCucumberMethods().entrySet()) {
                final Matcher matcher = Pattern.compile("value=(.*)\\)").matcher(elem.getKey());
                if (matcher.find()) {
                    final Matcher matcher2 = Pattern.compile(matcher.group(1)).matcher(loopedStep.getStep());
                    if (matcher2.find()) {
                        Object[] tab;
                        if (elem.getValue().isAnnotationPresent(Conditioned.class)) {
                            tab = new Object[matcher2.groupCount() + 1];
                            tab[matcher2.groupCount()] = stepConditions;
                        } else {
                            tab = new Object[matcher2.groupCount()];
                        }

                        for (int i = 0; i < matcher2.groupCount(); i++) {
                            final Parameter param = elem.getValue().getParameters()[i];
                            if (param.getType() == int.class) {
                                final int ii = Integer.parseInt(matcher2.group(i + 1));
                                tab[i] = ii;
                            } else if (param.getType() == boolean.class) {
                                tab[i] = Boolean.parseBoolean(matcher2.group(i + 1));
                            } else {
                                tab[i] = matcher2.group(i + 1);
                            }
                        }
                        try {
                            found = true;
                            elem.getValue().invoke(NoraUiInjector.getNoraUiInjectorSource().getInstance(elem.getValue().getDeclaringClass()), tab);
                            break;
                        } catch (final Exception e) {
                            throw new TechnicalException("\"" + loopedStep.getStep() + "\"", e.getCause());
                        }
                    }
                }
            }
            if (!found) {
                throw new TechnicalException(String.format(Messages.getMessage(TechnicalException.TECHNICAL_ERROR_STEP_UNDEFINED), loopedStep.getStep()));
            }
        }
    }
453164
def run(items, run_parallel):
    
    to_process = []
    extras = []
    for batch, cur_items in _group_by_batches(items).items():
        if _ready_for_het_analysis(cur_items):
            to_process.append((batch, cur_items))
        else:
            for data in cur_items:
                extras.append([data])
    processed = run_parallel("heterogeneity_estimate", ([xs, b, xs[0]["config"]] for b, xs in to_process))
    return _group_by_sample_and_batch(extras + processed)
806287
def loop(self):
        """"""
        while True:
            sleep(1)
            new_file_list = self.walk(self.file_path, {})
            if new_file_list != self.file_list:
                if self.debug:
                    self.diff_list(new_file_list, self.file_list)
                self.run_tests()
                self.file_list = new_file_list
656212
def run_step_group(pipeline_definition, step_group_name, context):
    """"""
    logger.debug(f"starting {step_group_name}")
    assert step_group_name

    steps = get_pipeline_steps(pipeline=pipeline_definition,
                               steps_group=step_group_name)

    run_pipeline_steps(steps=steps, context=context)

    logger.debug(f"done {step_group_name}")
1237833
def _run_train_step(self, data, mode='train'):
        
        epoch_size = ((len(data) // self.batch_size) - 1) // self.num_steps
        costs = 0.0
        iters = 0
        step = 0
        state = self._init_state.eval()
        op = self._train_op if mode == 'train' else tf.no_op()

        for step, (x, y) in enumerate(
            utilities.seq_data_iterator(
                data, self.batch_size, self.num_steps)):
            cost, state, _ = self.tf_session.run(
                [self.cost, self.final_state, op],
                {self.input_data: x,
                 self.input_labels: y,
                 self._init_state: state})

            costs += cost
            iters += self.num_steps

        if step % (epoch_size // 10) == 10:
            print("%.3f perplexity" % (step * 1.0 / epoch_size))

        return np.exp(costs / iters)
985751
public void run() {
    // Input step
    Database db = inputStep.getDatabase();
    hier = db.getHierarchy();

    // Algorithms - Data Mining Step
    algorithmStep.runAlgorithms(db);

    // TODO: this could be nicer
    hier.add(db, new SettingsResult(settings));

    // Evaluation
    evaluationStep.runEvaluators(hier, db);

    // Output / Visualization
    outputStep.runResultHandlers(hier, db);
  }
666500
def _excerpt_step(n_samples, n_excerpts=None, excerpt_size=None):
    """"""
    assert n_excerpts >= 2
    step = max((n_samples - excerpt_size) // (n_excerpts - 1),
               excerpt_size)
    return step
888600
def sample_all(generators, *args, **kwargs):
	''''''
	return [sample(gen, *args, **kwargs) for gen in  generators]
623850
def _run_step(step_obj, step_declaration, initialized_resources):
    """"""

    start_time = time.time()

    # Open any resources that need to be opened before we run this step
    for res_name in step_declaration.resources.opened:
        initialized_resources[res_name].open()

    # Create a dictionary of all of the resources that are required for this step
    used_resources = {local_name: initialized_resources[global_name] for local_name, global_name in step_declaration.resources.used.items()}

    # Allow steps with no resources to not need a resources keyword parameter
    if len(used_resources) > 0:
        out = step_obj.run(resources=used_resources)
    else:
        out = step_obj.run()

    # Close any resources that need to be closed before we run this step
    for res_name in step_declaration.resources.closed:
        initialized_resources[res_name].close()

    end_time = time.time()

    return (end_time - start_time, out)
1013592
def run_step(self):
        """"""
        formatted_replacements = self.context.get_formatted_iterable(
            self.replace_pairs)

        iter = StreamReplacePairsRewriterStep.iter_replace_strings(
            formatted_replacements)
        rewriter = StreamRewriter(iter)
        super().run_step(rewriter)
184220
def run(self):
        ''' '''
        # return value of the last executed statement
        self.last_res = None
        self.start_time = time.time()
        self.completed = defaultdict(int)
        #
        # prepare environments, namely variables that can be used by the step
        #
        # * step_name:  name of the step, can be used by step process to determine
        #               actions dynamically.
        env.sos_dict.set('step_name', self.step.step_name())
        env.sos_dict.set('__last_step__', self.step.last_step)
        self.log('start')
        env.sos_dict.set(
            'step_id',
            textMD5(
                f'{env.sos_dict["workflow_id"]} {env.sos_dict["step_name"]} {self.step.md5}'
            ))
        env.sos_dict.set('master_id', env.config['master_id'])
        # used by nested workflow
        env.sos_dict.set('__step_context__', self.step.context)

        env.sos_dict.set('_runtime', {})
        # * input:      input files, which should be __step_output__ if it is defined, or
        #               None otherwise.
        # * _input:     first batch of input, which should be input if no input statement is used
        # * output:     None at first, can be redefined by output statement
        # * _output:    None at first, can be redefined by output statement
        # * depends:    None at first, can be redefined by depends statement
        # * _depends:   None at first, can be redefined by depends statement
        #
        if '__step_output__' not in env.sos_dict or env.sos_dict[
                '__step_output__'].unspecified():
            env.sos_dict.set('step_input', sos_targets([]))
        else:
            env.sos_dict.set(
                'step_input',
                env.sos_dict['__step_output__']._remove_empty_groups())
        # input can be Undetermined from undetermined output from last step
        env.sos_dict.set('_input', copy.deepcopy(env.sos_dict['step_input']))

        if '__default_output__' in env.sos_dict:
            # if step is triggered by sos_step, it should not be considered as
            # output of the step. #981
            env.sos_dict.set(
                '__default_output__',
                sos_targets([
                    x for x in env.sos_dict['__default_output__']._targets
                    if not isinstance(x, sos_step)
                ]))
            env.sos_dict.set('step_output',
                             copy.deepcopy(env.sos_dict['__default_output__']))
            env.sos_dict.set('_output',
                             copy.deepcopy(env.sos_dict['__default_output__']))
        else:
            env.sos_dict.set('step_output', sos_targets([]))
            # output is said to be unspecified until output: is used
            env.sos_dict.set('_output', sos_targets(_undetermined=True))

        env.sos_dict.set('step_depends', sos_targets([]))
        env.sos_dict.set('_depends', sos_targets([]))
        # _index is needed for pre-input action's active option and for debug output of scripts
        env.sos_dict.set('_index', 0)

        if 'STEP' in env.config['SOS_DEBUG'] or 'ALL' in env.config['SOS_DEBUG']:
            env.log_to_file(
                'STEP',
                f'Executing step {env.sos_dict["step_name"]} with step_input {env.sos_dict["step_input"]} and step_output {env.sos_dict["step_output"]}'
            )

        if self.step.task_params:
            try:
                task_queue = get_value_of_param(
                    'queue',
                    self.step.task_params,
                    extra_dict=env.sos_dict.dict())
                if task_queue:
                    env.sos_dict['_runtime']['queue'] = task_queue[0]
            except Exception as e:
                raise ValueError(
                    f'Failed to determine value of parameter queue of {self.step.task_params}: {e}'
                )
            # check concurrent #1134
            try:
                task_concurrency = get_value_of_param(
                    'concurrent',
                    self.step.task_params,
                    extra_dict=env.sos_dict.dict())
                if task_concurrency:
                    env.sos_dict['_runtime']['concurrent'] = task_concurrency[0]
            except Exception as e:
                raise ValueError(
                    f'Failed to determine value of parameter queue of {self.step.task_params}: {e}'
                )
        if (env.config['default_queue'] in ('None', 'none', None) and
            'queue' not in env.sos_dict['_runtime']) or \
            ('queue' in env.sos_dict['_runtime'] and
            env.sos_dict['_runtime']['queue'] in ('none', 'None', None)):
            # remove task statement
            if len(self.step.statements
                  ) >= 1 and self.step.statements[-1][0] == '!':
                self.step.statements[-1][1] += '\n' + self.step.task
            else:
                self.step.statements.append(['!', self.step.task])
            self.step.task = None
        elif 'queue' not in env.sos_dict[
                '_runtime'] or not env.sos_dict['_runtime']['queue']:
            if env.config['default_queue']:
                env.sos_dict['_runtime']['queue'] = env.config['default_queue']
            else:
                env.sos_dict['_runtime']['queue'] = 'localhost'

        # look for input statement.
        input_statement_idx = [
            idx for idx, x in enumerate(self.step.statements)
            if x[0] == ':' and x[1] == 'input'
        ]
        if not input_statement_idx:
            input_statement_idx = None
        elif len(input_statement_idx) == 1:
            input_statement_idx = input_statement_idx[0]
        else:
            raise ValueError(
                f'More than one step input are specified in step {self.step.step_name()}'
            )

        # if shared is true, we have to disable concurrent because we
        # do not yet return anything from shared.
        self.concurrent_substep = 'shared' not in self.step.options and \
            ('concurrent' not in env.sos_dict['_runtime'] or env.sos_dict['_runtime']['concurrent'] is True)
        if input_statement_idx is not None:
            # execute before input stuff
            for statement in self.step.statements[:input_statement_idx]:
                if statement[0] == ':':
                    # wait for all dependent targets to be resolved to be resolved
                    key, value = statement[1:3]
                    if key != 'depends':
                        raise ValueError(
                            f'Step input should be specified before {key}')
                    while True:
                        try:
                            args, kwargs = SoS_eval(
                                f'__null_func__({value})',
                                extra_dict={
                                    '__null_func__': __null_func__,
                                    'output_from': __output_from__,
                                    'named_output': __named_output__,
                                    'traced': __traced__
                                })
                            dfiles = expand_depends_files(*args)
                            # dfiles can be Undetermined
                            runner = self.process_depends_args(dfiles, **kwargs)
                            try:
                                yreq = next(runner)
                                while True:
                                    yres = yield yreq
                                    yreq = runner.send(yres)
                            except StopIteration as e:
                                pass
                        except (UnknownTarget, RemovedTarget) as e:
                            runner = self.handle_unknown_target(e)
                            try:
                                yreq = next(runner)
                                while True:
                                    yres = yield yreq
                                    yreq = runner.send(yres)
                            except StopIteration as e:
                                pass
                            continue
                        except UnavailableLock:
                            raise
                        except Exception as e:
                            raise RuntimeError(
                                f'Failed to process step {key} ({value.strip()}): {e}'
                            )
                        break
                else:
                    try:
                        self.execute(statement[1])
                    except StopInputGroup as e:
                        # stop before substeps, because there is no output statement before it
                        # we do not have to worry about keep_output
                        if e.message:
                            env.logger.info(e.message)
                        return self.collect_result()
            # input statement
            stmt = self.step.statements[input_statement_idx][2]
            self.log('input statement', stmt)
            while True:
                # wait for all targets to be resovled
                try:
                    args, kwargs = SoS_eval(
                        f"__null_func__({stmt})",
                        extra_dict={
                            '__null_func__': __null_func__,
                            'output_from': __output_from__,
                            'named_output': __named_output__,
                            'traced': __traced__
                        })
                    # Files will be expanded differently with different running modes
                    input_files: sos_targets = expand_input_files(
                        *args, **{
                            k: v
                            for k, v in kwargs.items()
                            if k not in SOS_INPUT_OPTIONS
                        })
                    runner = self.process_input_args(
                        input_files, **{
                            k: v
                            for k, v in kwargs.items()
                            if k in SOS_INPUT_OPTIONS
                        })
                    try:
                        yreq = next(runner)
                        while True:
                            yres = yield yreq
                            yreq = runner.send(yres)
                    except StopIteration as e:
                        self._substeps = e.value
                    #
                    if 'concurrent' in kwargs and kwargs['concurrent'] is False:
                        self.concurrent_substep = False
                except (UnknownTarget, RemovedTarget) as e:
                    runner = self.handle_unknown_target(e)
                    try:
                        yreq = next(runner)
                        while True:
                            yres = yield yreq
                            yreq = runner.send(yres)
                    except StopIteration as e:
                        pass
                    continue
                except UnavailableLock:
                    raise
                except Exception as e:
                    raise ValueError(
                        f'Failed to process input statement {stmt}: {e}')
                break

            input_statement_idx += 1
        elif env.sos_dict['step_input'].groups:
            # if default has groups...
            # default case
            self._substeps = env.sos_dict['step_input'].groups
            # assuming everything starts from 0 is after input
            input_statement_idx = 0
        else:
            # default case
            self._substeps = [env.sos_dict['step_input']]
            # assuming everything starts from 0 is after input
            input_statement_idx = 0

        self.proc_results = []
        self.vars_to_be_shared = set()
        if 'shared' in self.step.options:
            self.vars_to_be_shared = parse_shared_vars(
                self.step.options['shared'])
        self.vars_to_be_shared = sorted([
            x[5:] if x.startswith('step_') else x
            for x in self.vars_to_be_shared
            if x not in ('step_', 'step_input', 'step_output', 'step_depends')
        ])
        self.shared_vars = [{} for x in self._substeps]
        # run steps after input statement, which will be run multiple times for each input
        # group.
        env.sos_dict.set('__num_groups__', len(self._substeps))

        # determine if a single index or the whole step should be skipped
        skip_index = False
        # signatures of each index, which can remain to be None if no output
        # is defined.
        self.output_groups = [sos_targets([]) for x in self._substeps]
        self.depends_groups = [sos_targets([]) for x in self._substeps]

        # used to prevent overlapping output from substeps
        self._all_outputs = set()
        self._subworkflow_results = []

        if any('sos_run' in x[1] for x in self.step.statements[input_statement_idx:]) and \
            'shared' not in self.step.options and not self.step.task and \
            self.step.statements[-1][0] == '!' and \
            (len(self.step.statements) == 1 or self.step.statements[-2][0] == ':') and \
            is_sos_run_the_only_last_stmt(self.step.statements[-1][1]):
            env.sos_dict.set('__concurrent_subworkflow__', True)

        if self.concurrent_substep:
            if len(self._substeps) <= 1 or env.config['run_mode'] == 'dryrun':
                self.concurrent_substep = False
            elif len([
                    x for x in self.step.statements[input_statement_idx:]
                    if x[0] != ':'
            ]) > 1:
                self.concurrent_substep = False
                env.logger.debug(
                    'Substeps are executed sequentially because of existence of directives between statements.'
                )
            elif any('sos_run' in x[1]
                     for x in self.step.statements[input_statement_idx:]):
                self.concurrent_substep = False
                env.logger.debug(
                    'Substeps are executed sequentially because of existence of multiple nested workflow.'
                )
            else:
                self.prepare_substep()

        try:
            self.completed['__substep_skipped__'] = 0
            self.completed['__substep_completed__'] = len(self._substeps)
            self._completed_concurrent_substeps = 0
            # pending signatures are signatures for steps with external tasks
            pending_signatures = [None for x in self._substeps]

            for idx, g in enumerate(self._substeps):
                # other variables
                #
                _vars = {}
                # now, let us expose target level variables as lists
                if len(g) > 1:
                    names = set.union(
                        *[set(x._dict.keys()) for x in g._targets])
                elif len(g) == 1:
                    names = set(g._targets[0]._dict.keys())
                else:
                    names = set()
                for name in names:
                    _vars[name] = [x.get(name) for x in g._targets]
                # then we expose all group level variables
                _vars.update(g._dict)
                _vars.update(env.sos_dict['step_input']._dict)
                env.sos_dict.update(_vars)

                env.sos_dict.set('_input', copy.deepcopy(g))
                # set vars to _input
                #env.sos_dict['_input'].set(**v)

                self.log('_input')
                env.sos_dict.set('_index', idx)

                # in interactive mode, because sos_dict are always shared
                # execution of a substep, especially when it calls a nested
                # workflow, would change step_name, __step_context__ etc, and
                # we will have to reset these variables to make sure the next
                # substep would execute normally. Batch mode is immune to this
                # problem because nested workflows are executed in their own
                # process/context etc
                if env.config['run_mode'] == 'interactive':
                    env.sos_dict.set('step_name', self.step.step_name())
                    env.sos_dict.set(
                        'step_id',
                        hash((env.sos_dict["workflow_id"],
                              env.sos_dict["step_name"], self.step.md5)))
                    # used by nested workflow
                    env.sos_dict.set('__step_context__', self.step.context)
                #
                pre_statement = []
                if not any(st[0] == ':' and st[1] == 'output' for st in self.step.statements[input_statement_idx:]) and \
                        '__default_output__' in env.sos_dict:
                    pre_statement = [[':', 'output', '_output']]

                # if there is no statement, no task, claim success
                post_statement = []
                if not any(
                        st[0] == '!'
                        for st in self.step.statements[input_statement_idx:]):
                    if self.step.task:
                        # if there is only task, we insert a fake statement so that it can be executed by the executor
                        post_statement = [['!', '']]
                    else:
                        # complete case: no step, no statement
                        send_message_to_controller([
                            'progress', 'substep_completed',
                            env.sos_dict['step_id']
                        ])

                all_statements = pre_statement + self.step.statements[
                    input_statement_idx:] + post_statement
                is_input_verified = True
                for statement_idx, statement in enumerate(all_statements):
                    is_last_runblock = statement_idx == len(all_statements) - 1

                    # if input is undertermined, we can only process output:
                    if not g.valid() and statement[0] != ':':
                        raise RuntimeError('Undetermined input encountered')
                    if statement[0] == ':':
                        key, value = statement[1:3]
                        # output, depends, and process can be processed multiple times
                        while True:
                            # loop for all unresolved targets to be resolved
                            try:
                                args, kwargs = SoS_eval(
                                    f'__null_func__({value})',
                                    extra_dict={
                                        '__null_func__': __null_func__,
                                        'output_from': __output_from__,
                                        'named_output': __named_output__,
                                        'traced': __traced__
                                    })
                                # dynamic output or dependent files
                                if key == 'output':
                                    # if output is defined, its default value needs to be cleared
                                    if idx == 0:
                                        env.sos_dict.set(
                                            'step_output', sos_targets())
                                    ofiles: sos_targets = expand_output_files(
                                        value, *args, **{
                                            k: v
                                            for k, v in kwargs.items()
                                            if k not in SOS_OUTPUT_OPTIONS
                                        })
                                    if g.valid() and ofiles.valid():
                                        if any(
                                                x in g._targets
                                                for x in ofiles
                                                if not isinstance(x, sos_step)):
                                            raise RuntimeError(
                                                f'Overlapping input and output files: {", ".join(repr(x) for x in ofiles if x in g)}'
                                            )

                                    # set variable _output and output
                                    self.process_output_args(
                                        ofiles, **{
                                            k: v
                                            for k, v in kwargs.items()
                                            if k in SOS_OUTPUT_OPTIONS
                                        })
                                    self.output_groups[idx] = env.sos_dict[
                                        '_output']
                                elif key == 'depends':
                                    try:
                                        dfiles = expand_depends_files(*args)
                                        # dfiles can be Undetermined
                                        runner = self.process_depends_args(
                                            dfiles, **kwargs)
                                        try:
                                            yreq = next(runner)
                                            while True:
                                                yres = yield yreq
                                                yreq = runner.send(yres)
                                        except StopIteration as e:
                                            pass
                                        self.depends_groups[idx] = env.sos_dict[
                                            '_depends']
                                        self.log('_depends')
                                    except Exception as e:
                                        #env.logger.info(e)
                                        raise
                                else:
                                    raise RuntimeError(
                                        f'Unrecognized directive {key}')
                                # everything is ok, break
                                break
                            except (UnknownTarget, RemovedTarget) as e:
                                runner = self.handle_unknown_target(e)
                                try:
                                    yreq = next(runner)
                                    while True:
                                        yres = yield yreq
                                        yreq = runner.send(yres)
                                except StopIteration as e:
                                    pass
                                continue
                            except UnavailableLock:
                                raise
                            except Exception as e:
                                # if input is Undertermined, it is possible that output cannot be processed
                                # due to that, and we just return
                                if not g.valid():
                                    env.logger.debug(e)
                                    return self.collect_result()
                                raise RuntimeError(
                                    f'Failed to process step {key} ({value.strip()}): {e}'
                                )
                    elif is_last_runblock:
                        try:
                            if self.concurrent_substep:
                                if 'STEP' in env.config['SOS_DEBUG'] or 'ALL' in env.config['SOS_DEBUG']:
                                    env.log_to_file(
                                        'STEP',
                                        f'Execute substep {env.sos_dict["step_name"]} {idx} concurrently with {self._completed_concurrent_substeps} completed'
                                    )

                                # the ignatures are supposed to be written by substep worker, however
                                # the substep worker might send tasks back to the step worker and
                                # we should write the signatures after the tasks are completed
                                if env.config['sig_mode'] != 'ignore' and not env.sos_dict['_output'].unspecified() \
                                    and self.step.task:
                                    pending_signatures[idx] = RuntimeInfo(
                                        statementMD5(
                                            [statement[1], self.step.task]),
                                        env.sos_dict['_input'],
                                        env.sos_dict['_output'],
                                        env.sos_dict['_depends'],
                                        env.sos_dict['__signature_vars__'],
                                        shared_vars=self.vars_to_be_shared)
                                #
                                # step_output: needed only when it is undetermined
                                # step_input: not needed
                                # _input, _output, _depends, _index: needed
                                # step_name: for debug scripts
                                # step_id, workflow_id: for reporting to controller
                                # '__signature_vars__' to be used for signature creation
                                #
                                # __step_context__ is not needed because substep
                                # executor does not support nested workflow

                                proc_vars = env.sos_dict['__signature_vars__'] \
                                    | {'_input', '_output', '_depends', '_index',
                                     'step_output', 'step_name',
                                      '_runtime', 'step_id', 'workflow_id', '__num_groups__',
                                      '__signature_vars__'}
                                self.proc_results.append({})
                                self.submit_substep(
                                    dict(
                                        stmt=statement[1],
                                        global_def=self.step.global_def,
                                        #1225: the step might contain large variables from global section, but
                                        # we do not have to sent them if they are not used in substeps.
                                        global_vars={
                                            x: y
                                            for x, y in
                                            self.step.global_vars.items()
                                            if x in
                                            env.sos_dict['__signature_vars__']
                                        },
                                        task=self.step.task,
                                        task_params=self.step.task_params,
                                        proc_vars=env.sos_dict
                                        .clone_selected_vars(proc_vars),
                                        shared_vars=self.vars_to_be_shared,
                                        config=env.config))
                                # we check if the previous task has been completed and process them
                                # because further steps might need to be done
                                try:
                                    runner = self.process_returned_substep_result(
                                        wait=False)
                                    yreq = next(runner)
                                    while True:
                                        yres = yield yreq
                                        yreq = runner.send(yres)
                                except StopIteration:
                                    pass
                            else:
                                if env.config[
                                        'sig_mode'] == 'ignore' or env.sos_dict[
                                            '_output'].unspecified():
                                    if 'STEP' in env.config['SOS_DEBUG'] or 'ALL' in env.config['SOS_DEBUG']:
                                        env.log_to_file(
                                            'STEP',
                                            f'Execute substep {env.sos_dict["step_name"]} without signature'
                                        )
                                    try:
                                        if is_input_verified:
                                            verify_input()
                                            is_input_verified = False
                                        if env.sos_dict.get(
                                                '__concurrent_subworkflow__',
                                                False):
                                            self._subworkflow_results.append(
                                                self.execute(
                                                    statement[1],
                                                    return_result=True))
                                        else:
                                            self.execute(statement[1])
                                    finally:
                                        if not self.step.task:
                                            # if no task, this step is __completed
                                            # complete case: local skip without task
                                            send_message_to_controller([
                                                'progress', 'substep_completed',
                                                env.sos_dict['step_id']
                                            ])
                                    if 'shared' in self.step.options:
                                        try:
                                            self.shared_vars[
                                                env.sos_dict['_index']].update({
                                                    x: env.sos_dict[x]
                                                    for x in
                                                    self.vars_to_be_shared
                                                    if x in env.sos_dict
                                                })
                                        except Exception as e:
                                            raise ValueError(
                                                f'Missing shared variable {e}.')
                                else:
                                    sig = RuntimeInfo(
                                        statementMD5(
                                            [statement[1], self.step.task]),
                                        env.sos_dict['_input'],
                                        env.sos_dict['_output'],
                                        env.sos_dict['_depends'],
                                        env.sos_dict['__signature_vars__'],
                                        shared_vars=self.vars_to_be_shared)
                                    if 'STEP' in env.config['SOS_DEBUG'] or 'ALL' in env.config['SOS_DEBUG']:
                                        env.log_to_file(
                                            'STEP',
                                            f'Execute substep {env.sos_dict["step_name"]} with signature {sig.sig_id}'
                                        )
                                    # if singaure match, we skip the substep even  if
                                    # there are tasks.
                                    matched = validate_step_sig(sig)
                                    skip_index = bool(matched)
                                    if matched:
                                        if env.sos_dict[
                                                'step_output'].undetermined():
                                            self.output_groups[env.sos_dict[
                                                '_index']] = matched["output"]
                                        if 'vars' in matched:
                                            self.shared_vars[
                                                env.sos_dict['_index']].update(
                                                    matched["vars"])
                                        # complete case: local skip without task
                                        send_message_to_controller([
                                            'progress', 'substep_ignored',
                                            env.sos_dict['step_id']
                                        ])
                                        # do not execute the rest of the statement
                                        break
                                    else:
                                        sig.lock()
                                        try:
                                            if is_input_verified:
                                                verify_input()
                                                is_input_verified = False
                                            if env.sos_dict.get(
                                                    '__concurrent_subworkflow__',
                                                    False):
                                                self._subworkflow_results.append(
                                                    self.execute(
                                                        statement[1],
                                                        return_result=True))
                                            else:
                                                self.execute(statement[1])
                                            if 'shared' in self.step.options:
                                                try:
                                                    self.shared_vars[
                                                        env.sos_dict[
                                                            '_index']].update({
                                                                x:
                                                                env.sos_dict[x]
                                                                for x in self.
                                                                vars_to_be_shared
                                                                if x in
                                                                env.sos_dict
                                                            })
                                                except Exception as e:
                                                    raise ValueError(
                                                        f'Missing shared variable {e}.'
                                                    )
                                        finally:
                                            # if this is the end of substep, save the signature
                                            # otherwise we need to wait for the completion
                                            # of the task.
                                            if not self.step.task:
                                                if env.sos_dict[
                                                        'step_output'].undetermined(
                                                        ):
                                                    output = reevaluate_output()
                                                    self.output_groups[
                                                        env.sos_dict[
                                                            '_index']] = output
                                                    sig.set_output(output)
                                                sig.write()
                                                # complete case : local execution without task
                                                send_message_to_controller([
                                                    'progress',
                                                    'substep_completed',
                                                    env.sos_dict['step_id']
                                                ])
                                            else:
                                                pending_signatures[idx] = sig
                                            sig.release()
                        except StopInputGroup as e:
                            if not e.keep_output:
                                clear_output()
                                self.output_groups[idx] = sos_targets([])
                            if e.message:
                                env.logger.info(e.message)
                            skip_index = True
                            break
                        except Exception as e:
                            clear_output()
                            raise
                    else:
                        # if it is not the last statement group (e.g. statements before :output)
                        # we execute locally without anything like signature
                        if is_input_verified:
                            verify_input()
                            is_input_verified = False
                        try:
                            self.execute(statement[1])
                        except StopInputGroup as e:
                            if not e.keep_output:
                                clear_output()
                                self.output_groups[idx] = sos_targets([])
                            if e.message:
                                env.logger.info(e.message)
                            skip_index = True
                            break
                        except Exception as e:
                            clear_output()
                            raise
                # if there is no statement , but there are tasks, we should
                # check signature here.
                if not any(x[0] == '!' for x in self.step.statements[input_statement_idx:]) and self.step.task and not self.concurrent_substep \
                    and env.config['sig_mode'] != 'ignore' and not env.sos_dict['_output'].unspecified():
                    sig = RuntimeInfo(
                        statementMD5([self.step.task]),
                        env.sos_dict['_input'],
                        env.sos_dict['_output'],
                        env.sos_dict['_depends'],
                        env.sos_dict['__signature_vars__'],
                        shared_vars=self.vars_to_be_shared)
                    if 'STEP' in env.config['SOS_DEBUG'] or 'ALL' in env.config['SOS_DEBUG']:
                        env.log_to_file(
                            'STEP',
                            f'Check task-only step {env.sos_dict["step_name"]} with signature {sig.sig_id}'
                        )
                    matched = validate_step_sig(sig)
                    skip_index = bool(matched)
                    if matched:
                        if env.sos_dict['step_output'].undetermined():
                            self.output_groups[
                                env.sos_dict['_index']] = matched["output"]
                        self.shared_vars[env.sos_dict['_index']].update(
                            matched["vars"])
                        # complete case: step with task ignored
                        send_message_to_controller([
                            'progress', 'substep_ignored',
                            env.sos_dict['step_id']
                        ])
                    pending_signatures[idx] = sig

                # if this index is skipped, go directly to the next one
                if skip_index:
                    self.completed['__substep_skipped__'] += 1
                    self.completed['__substep_completed__'] -= 1
                    skip_index = False
                    continue

                # if concurrent input group, tasks are handled in substep
                if self.concurrent_substep or not self.step.task:
                    continue

                if env.config[
                        'run_mode'] == 'dryrun' and env.sos_dict['_index'] != 0:
                    continue

                # check if the task is active
                if 'active' in env.sos_dict['_runtime']:
                    active = env.sos_dict['_runtime']['active']
                    if active is True:
                        pass
                    elif active is False:
                        continue
                    elif isinstance(active, int):
                        if active >= 0 and env.sos_dict['_index'] != active:
                            continue
                        if active < 0 and env.sos_dict[
                                '_index'] != active + env.sos_dict[
                                    '__num_groups__']:
                            continue
                    elif isinstance(active, Sequence):
                        allowed_index = list([
                            x if x >= 0 else env.sos_dict['__num_groups__'] + x
                            for x in active
                        ])
                        if env.sos_dict['_index'] not in allowed_index:
                            continue
                    elif isinstance(active, slice):
                        allowed_index = list(
                            range(env.sos_dict['__num_groups__']))[active]
                        if env.sos_dict['_index'] not in allowed_index:
                            continue
                    else:
                        raise RuntimeError(
                            f'Unacceptable value for option active: {active}')

                #
                self.log('task')
                try:
                    task_id, taskdef, task_vars = create_task(
                        self.step.global_def, self.step.global_vars,
                        self.step.task, self.step.task_params)
                    task = self.submit_task({
                        'index': env.sos_dict['_index'],
                        'task_id': task_id,
                        'task_def': taskdef,
                        'task_vars': task_vars
                    })
                    self.proc_results.append(task)
                except Exception as e:
                    # FIXME: cannot catch exception from subprocesses
                    if env.verbosity > 2:
                        sys.stderr.write(get_traceback())
                    raise RuntimeError(
                        f'Failed to execute process\n"{short_repr(self.step.task)}"\n{e}'
                    )
                #
                # if not concurrent, we have to wait for the completion of the task
                if 'concurrent' in env.sos_dict['_runtime'] and env.sos_dict[
                        '_runtime']['concurrent'] is False:
                    # in this case the steps must be executed not concurrently
                    runner = self.wait_for_results(all_submitted=False)
                    try:
                        yreq = next(runner)
                        while True:
                            yres = yield yreq
                            yreq = runner.send(yres)
                    except StopIteration:
                        pass
                #
                # endfor loop for each input group
                #
            if self._subworkflow_results:
                try:
                    runner = self.wait_for_subworkflows(
                        self._subworkflow_results)
                    yreq = next(runner)
                    while True:
                        yres = yield yreq
                        yreq = runner.send(yres)
                except StopIteration:
                    pass
                env.sos_dict.pop('__concurrent_subworkflow__')

            runner = self.wait_for_results(all_submitted=True)
            try:
                yreq = next(runner)
                while True:
                    yres = yield yreq
                    yreq = runner.send(yres)
            except StopIteration:
                pass

            for idx, res in enumerate(self.proc_results):
                if 'sig_skipped' in res:
                    self.completed['__substep_skipped__'] += 1
                    self.completed['__substep_completed__'] -= 1
                if 'output' in res:
                    self.output_groups[idx] = res["output"]

            # check results
            for proc_result in [
                    x for x in self.proc_results if x['ret_code'] == 0
            ]:
                if 'stdout' in proc_result and proc_result['stdout']:
                    sys.stdout.write(proc_result['stdout'])
                if 'stderr' in proc_result and proc_result['stderr']:
                    sys.stderr.write(proc_result['stderr'])

            # now that output is settled, we can write remaining signatures
            for idx, res in enumerate(self.proc_results):
                if pending_signatures[idx] is not None and res[
                        'ret_code'] == 0 and not 'sig_skipped' in res:
                    pending_signatures[idx].write()
                if res['ret_code'] != 0 and 'output' in res:
                    clear_output(output=res['output'])

            for proc_result in [
                    x for x in self.proc_results if x['ret_code'] != 0
            ]:
                if 'stdout' in proc_result and proc_result['stdout']:
                    sys.stdout.write(proc_result['stdout'])
                if 'stderr' in proc_result and proc_result['stderr']:
                    sys.stderr.write(proc_result['stderr'])
                if 'exception' in proc_result:
                    excp = proc_result['exception']
                    if isinstance(excp, StopInputGroup):
                        if excp.message:
                            env.logger.info(excp.message)
                        self.output_groups[proc_result['index']] = sos_targets(
                            [])
                    #elif isinstance(e, RemovedTarget):
                    #
                    # in theory, we should be able to handled removed target from here
                    # by rerunning the substep, but we it is too much work for this
                    # corner case. Let us simply rerunt he entire step.
                    else:
                        raise excp
                else:
                    raise RuntimeError(
                        f"Substep failed with return code {proc_result['ret_code']}"
                    )

            # if output is Undetermined, re-evalulate it
            # finalize output from output_groups because some output might be skipped
            # this is the final version of the output but we do maintain output
            # during the execution of step, for compatibility.
            env.sos_dict.set('step_output',
                             sos_targets([])._add_groups(self.output_groups))
            env.sos_dict.set('step_depends',
                             sos_targets([])._add_groups(self.depends_groups))

            # if there exists an option shared, the variable would be treated as
            # provides=sos_variable(), and then as step_output
            if 'shared' in self.step.options:
                self.shared_vars = evaluate_shared(self.shared_vars,
                                                   self.step.options['shared'])
                env.sos_dict.quick_update(self.shared_vars)
            self.log('output')
            self.verify_output()
            substeps = self.completed['__substep_completed__'] + \
                self.completed['__substep_skipped__']
            self.completed['__step_completed__'] = self.completed[
                '__substep_completed__'] / substeps
            self.completed['__step_skipped__'] = self.completed[
                '__substep_skipped__'] / substeps
            if self.completed['__step_completed__'].is_integer():
                self.completed['__step_completed__'] = int(
                    self.completed['__step_completed__'])
            if self.completed['__step_skipped__'].is_integer():
                self.completed['__step_skipped__'] = int(
                    self.completed['__step_skipped__'])

            def file_only(targets):
                if not isinstance(targets, sos_targets):
                    env.logger.warning(
                        f"Unexpected input or output target for reporting. Empty list returned: {targets}"
                    )
                    return []
                return [(str(x), x.size())
                        for x in targets._targets
                        if isinstance(x, file_target)]

            step_info = {
                'step_id': self.step.md5,
                'start_time': self.start_time,
                'stepname': self.step.step_name(True),
                'substeps': len(self._substeps),
                'input': file_only(env.sos_dict['step_input']),
                'output': file_only(env.sos_dict['step_output']),
                'completed': dict(self.completed),
                'end_time': time.time()
            }
            send_message_to_controller([
                'workflow_sig', 'step', env.sos_dict["workflow_id"],
                repr(step_info)
            ])
            return self.collect_result()
        finally:
            if self.concurrent_substep:
                close_socket(self.result_pull_socket, 'substep collector')