docs: def run(self):
          '''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. '''
          # 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')