Execute a single step and return results. The result for batch mode is the
        input, output etc returned as alias, and for interactive mode is the return value
        of the last expression.
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')
17119
public function batchExecute()
    {
        $results = $this->execute(new InvokeSpec($this->units, false))->getResults();

        // Make right order in sequence of results. It's required operation, because JSON-RPC2
        // specification define: "The Response objects being returned from a batch call MAY be returned
        // in any order within the Array. The Client SHOULD match contexts between the set of Request objects and the
        // resulting set of Response objects based on the id member within each Object."

        $callMap = [];
        foreach ($this->units as $index => $unit) {
            /** @var Invoke $unit */
            $callMap[$unit->getRawId()] = $index;
        }

        if (count($results) !== count($this->units)) {
            throw new InvalidResponseException();
        }

        $resultSequence = [];
        foreach ($results as $result) {
            if ($result instanceof Result) {
                /** @var Result $result */
                $resultSequence[ $callMap[$result->getId()] ] = $result->getResult();
            } elseif ($result instanceof Error) {
                /** @var Error $result */
                $resultSequence[ $callMap[$result->getId()] ] = $this->serverErrorMode === self::ERRMODE_EXCEPTION ? $result->getBaseException() : null;
            }
        }
        ksort($resultSequence);
        $this->units = [];

        return $resultSequence;
    }
463896
def perform(self):
        
        last_value = None
        last_step = None

        while self.items.qsize():

            item = self.items.get()

            if item.flag == self.do:
                last_value = item.item(*item.args, **item.kwargs)
                last_step = item.message
            elif item.flag == self.expect:
                message = item.message
                local = {'value': last_value, 'expectation': item.item}
                expression = 'value {operator} expectation'.format(operator=item.operator)
                result = eval(expression, local)

                # Format the error message
                format_vars = {
                    'actual': last_value,
                    'expected': item.item,
                    'step': last_step,
                    'operator': item.operator
                }

                for var, val in format_vars.iteritems():
                    message = message.replace('{' + str(var) + '}', str(val))

                assert result, message

        return last_value
84584
private void executeBatch(BatchStatement batchState) {
        if (batchState.size() > 0) {
            m_logger.debug("Executing synchronous batch with {} statements", batchState.size());
            m_dbservice.getSession().execute(batchState);
        }
    }
251628
def do(self, command):
        
        echo = self.echo and command or ''
        if not self.logged_in:
            return echo + '\n' + self._get_prompt()

        response = self.commands.eval(command)
        if response is None:
            return echo + '\n' + self._get_prompt()
        return echo + response
366423
def execute_step(self, step):
        
        inputs = self.get_inputs(step.ins)
        outputs = defaultdict(list)
        for output in step.run(inputs):
            for k, v in output.items():
                outputs[k].extend(v)
        for output in step.finalise():
            for k, v in output.items():
                outputs[k].extend(v)
        self.unresolved_steps.remove(step)
        return outputs
445852
def invoke(self, results):
        
        args = [results.get(d) for d in self.deps]
        return self.component(*args)
28829
def perform_command(self):
        
        # if no actual arguments, print help
        if len(self.actual_arguments) < 1:
            return self.print_help(short=True)

        # check if we have a recognized tool switch
        for cls, switches in self.TOOLS:
            if self.has_option(switches):
                arguments = [a for a in sys.argv if a not in switches]
                return cls(invoke=(self.invoke + u" %s" % switches[0])).run(arguments=arguments)

        # check if we have -h, --help, or --version
        if u"-h" in self.actual_arguments:
            return self.print_help(short=True)
        if u"--help" in self.actual_arguments:
            return self.print_help(short=False)
        if u"--version" in self.actual_arguments:
            return self.print_name_version()

        # default to run ExecuteTaskCLI
        return ExecuteTaskCLI(invoke=self.invoke).run(arguments=sys.argv)
1109596
def batch(self, methods):
        
        methods = list(methods) # methods can be iterator
        if not methods:
            #accepts also empty list (or generator) - it allows call batch directly with query result (eventually empty)
            return
        queries, callbacks = list(zip(*[m(batch=True) for m in methods]))
        # perform all the operations in one batch
        responses = self.execute("", "POST", requests=queries)
        # perform the callbacks with the response data (updating the existing
        # objets, etc)

        batched_errors = []
        for callback, response in zip(callbacks, responses):
            if "success" in response:
                callback(response["success"])
            else:
                batched_errors.append(response["error"])

        if batched_errors:
            raise core.ParseBatchError(batched_errors)
433107
def send_command_return(self, obj, command, *arguments):
        
        return self._perform_command('{}/{}'.format(self.session_url, obj.ref), command, OperReturnType.line_output,
                                     *arguments).json()
619628
def execute
      return unless @start && @limit
      @next_to_insert = @start
      while @next_to_insert <= @limit
        stride = @throttler.stride
        affected_rows = @connection.update(copy(bottom, top(stride)))

        if @throttler && affected_rows > 0
          @throttler.run
        end

        @printer.notify(bottom, @limit)
        @next_to_insert = top(stride) + 1
        break if @start == @limit
      end
      @printer.end
    end
716239
def batch(self, queries):
        
        with self.connection() as cur:
            for sql, vars in queries:

                if self.debug:
                    print(cur.mogrify(query, vars))
                cur.execute(sql, vars)
996926
def Execute(self, *params, **kw):
        """"""
        fp = self.__expandparamstodict(params, kw)
        return self._get_subfolder('execute/', GPExecutionResult, fp)
956369
public Result[] batch(List<? extends Row> actions) throws IOException {
    try {
      Object[] resultsOrErrors = new Object[actions.size()];
      batchCallback(actions, resultsOrErrors, null);
      // At this point we are guaranteed that the array only contains results,
      // if it had any errors, batch would've thrown an exception
      Result[] results = new Result[resultsOrErrors.length];
      System.arraycopy(resultsOrErrors, 0, results, 0, results.length);
      return results;
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      LOG.error("Encountered exception in batch(List<>).", e);
      throw new IOException("Batch error", e);
    }
  }
823253
def do_execute(self):
        
        result = None
        evl = self.input.payload
        pltclassifier.plot_prc(
            evl,
            class_index=self.resolve_option("class_index"),
            title=self.resolve_option("title"),
            key_loc=self.resolve_option("key_loc"),
            outfile=self.resolve_option("outfile"),
            wait=bool(self.resolve_option("wait")))
        return result
554769
def _merge_result(self, command, res, **kwargs):
        
        if command in self.result_callbacks:
            return self.result_callbacks[command](res, **kwargs)

        # Default way to handle result
        return first_key(res)
30774
function Result(cmd, err, stdout, stderr) {
  this.cmd = cmd;
  this.err = err;
  this.stdout = strip(stdout);
  this.stderr = strip(stderr);
}