docs: def _execute_task(task_id,
                    verbosity=None,
                    runmode='run',
                    sigmode=None,
                    monitor_interval=5,
                    resource_monitor_interval=60,
                    master_runtime={}):
      '''A function that execute specified task within a local dictionary
      (from SoS env.sos_dict). This function should be self-contained in that
      it can be handled by a task manager, be executed locally in a separate
      process or remotely on a different machine.'''
      # start a monitoring file, which would be killed after the job
      # is done (killed etc)
      if isinstance(task_id, str):
          params, master_runtime = TaskFile(task_id).get_params_and_runtime()
          sig_content = TaskFile(task_id).signature
          subtask = False
      else:
          # subtask
          subtask = True
          task_id, params, sig_content = task_id
          if 'TASK' in env.config['SOS_DEBUG'] or 'ALL' in env.config['SOS_DEBUG']:
              env.log_to_file('TASK', f'Executing subtask {task_id}')
  
      # update local runtime with master runtime
      if '_runtime' in master_runtime:
          params.sos_dict['_runtime'].update(master_runtime['_runtime'])
      if task_id in master_runtime:
          params.sos_dict.update(master_runtime[task_id])
  
      if hasattr(params, 'task_stack'):
          return _execute_sub_tasks(task_id, params, sig_content, verbosity,
                                    runmode, sigmode, monitor_interval,
                                    resource_monitor_interval, master_runtime)
  
      global_def, task, sos_dict = params.global_def, params.task, params.sos_dict
  
      # task output
      env.sos_dict.set(
          '__std_out__',
          os.path.join(
              os.path.expanduser('~'), '.sos', 'tasks', task_id + '.out'))
      env.sos_dict.set(
          '__std_err__',
          os.path.join(
              os.path.expanduser('~'), '.sos', 'tasks', task_id + '.err'))
      env.logfile = os.path.join(
          os.path.expanduser('~'), '.sos', 'tasks', task_id + '.err')
      # clear the content of existing .out and .err file if exists, but do not create one if it does not exist
      if os.path.exists(env.sos_dict['__std_out__']):
          open(env.sos_dict['__std_out__'], 'w').close()
      if os.path.exists(env.sos_dict['__std_err__']):
          open(env.sos_dict['__std_err__'], 'w').close()
  
      if verbosity is not None:
          env.verbosity = verbosity
  
      prepare_env(global_def[0], global_def[1])
  
      if '_runtime' not in sos_dict:
          sos_dict['_runtime'] = {}
  
      # pulse thread
      if monitor_interval is not None:
          m = ProcessMonitor(
              task_id,
              monitor_interval=monitor_interval,
              resource_monitor_interval=resource_monitor_interval,
              max_walltime=sos_dict['_runtime'].get('max_walltime', None),
              max_mem=sos_dict['_runtime'].get('max_mem', None),
              max_procs=sos_dict['_runtime'].get('max_procs', None),
              sos_dict=sos_dict)
  
          m.start()
      env.config['run_mode'] = runmode
      if runmode == 'dryrun':
          env.config['sig_mode'] = 'ignore'
      elif sigmode is not None:
          env.config['sig_mode'] = sigmode
      #
      (env.logger.debug if subtask else env.logger.info)(f'{task_id} ``started``')
  
      env.sos_dict.quick_update(sos_dict)
  
      for key in [
              'step_input', '_input', 'step_output', '_output', 'step_depends',
              '_depends'
      ]:
          if key in sos_dict and isinstance(sos_dict[key], sos_targets):
              # resolve remote() target
              env.sos_dict.set(
                  key,
                  sos_dict[key].remove_targets(type=sos_step).resolve_remote())
  
      # when no output is specified, we just treat the task as having no output (determined)
      env.sos_dict['_output']._undetermined = False
      sig = None if env.config['sig_mode'] == 'ignore' else InMemorySignature(
          env.sos_dict['_input'],
          env.sos_dict['_output'],
          env.sos_dict['_depends'],
          env.sos_dict['__signature_vars__'],
          shared_vars=parse_shared_vars(env.sos_dict['_runtime'].get(
              'shared', None)))
  
      if sig and _validate_task_signature(sig, sig_content.get(task_id, {}),
                                          task_id, subtask):
          #env.logger.info(f'{task_id} ``skipped``')
          return collect_task_result(
              task_id, sos_dict, skipped=True, signature=sig)
  
      # if we are to really execute the task, touch the task file so that sos status shows correct
      # execution duration.
      if not subtask:
          sos_dict['start_time'] = time.time()
  
      try:
          # go to 'workdir'
          if '_runtime' in sos_dict and 'workdir' in sos_dict['_runtime']:
              if not os.path.isdir(
                      os.path.expanduser(sos_dict['_runtime']['workdir'])):
                  try:
                      os.makedirs(
                          os.path.expanduser(sos_dict['_runtime']['workdir']))
                      os.chdir(
                          os.path.expanduser(sos_dict['_runtime']['workdir']))
                  except Exception as e:
                      # sometimes it is not possible to go to a "workdir" because of
                      # file system differences, but this should be ok if a work_dir
                      # has been specified.
                      env.logger.debug(
                          f'Failed to create workdir {sos_dict["_runtime"]["workdir"]}: {e}'
                      )
              else:
                  os.chdir(os.path.expanduser(sos_dict['_runtime']['workdir']))
          #
          orig_dir = os.getcwd()
  
          # we will need to check existence of targets because the task might
          # be executed on a remote host where the targets are not available.
          for target in (sos_dict['_input'] if isinstance(sos_dict['_input'], list) else []) + \
                  (sos_dict['_depends'] if isinstance(sos_dict['_depends'], list) else []):
              # if the file does not exist (although the signature exists)
              # request generation of files
              if isinstance(target, str):
                  if not file_target(target).target_exists('target'):
                      # remove the signature and regenerate the file
                      raise RuntimeError(f'{target} not found')
              # the sos_step target should not be checked in tasks because tasks are
              # independently executable units.
              elif not isinstance(
                      target, sos_step) and not target.target_exists('target'):
                  raise RuntimeError(f'{target} not found')
  
          # create directory. This usually has been done at the step level but the task can be executed
          # on a remote host where the directory does not yet exist.
          ofiles = env.sos_dict['_output']
          if ofiles.valid():
              for ofile in ofiles:
                  parent_dir = ofile.parent
                  if not parent_dir.is_dir():
                      parent_dir.mkdir(parents=True, exist_ok=True)
  
                      # go to user specified workdir
          if '_runtime' in sos_dict and 'workdir' in sos_dict['_runtime']:
              if not os.path.isdir(
                      os.path.expanduser(sos_dict['_runtime']['workdir'])):
                  try:
                      os.makedirs(
                          os.path.expanduser(sos_dict['_runtime']['workdir']))
                  except Exception as e:
                      raise RuntimeError(
                          f'Failed to create workdir {sos_dict["_runtime"]["workdir"]}: {e}'
                      )
              os.chdir(os.path.expanduser(sos_dict['_runtime']['workdir']))
          # set environ ...
          # we join PATH because the task might be executed on a different machine
          if '_runtime' in sos_dict:
              if 'env' in sos_dict['_runtime']:
                  for key, value in sos_dict['_runtime']['env'].items():
                      if 'PATH' in key and key in os.environ:
                          new_path = OrderedDict()
                          for p in value.split(os.pathsep):
                              new_path[p] = 1
                          for p in value.split(os.environ[key]):
                              new_path[p] = 1
                          os.environ[key] = os.pathsep.join(new_path.keys())
                      else:
                          os.environ[key] = value
              if 'prepend_path' in sos_dict['_runtime']:
                  if isinstance(sos_dict['_runtime']['prepend_path'], str):
                      os.environ['PATH'] = sos_dict['_runtime']['prepend_path'] + \
                          os.pathsep + os.environ['PATH']
                  elif isinstance(env.sos_dict['_runtime']['prepend_path'],
                                  Sequence):
                      os.environ['PATH'] = os.pathsep.join(
                          sos_dict['_runtime']
                          ['prepend_path']) + os.pathsep + os.environ['PATH']
                  else:
                      raise ValueError(
                          f'Unacceptable input for option prepend_path: {sos_dict["_runtime"]["prepend_path"]}'
                      )
  
          # step process
          SoS_exec(task)
  
          (env.logger.debug
           if subtask else env.logger.info)(f'{task_id} ``completed``')
  
      except StopInputGroup as e:
          # task ignored with stop_if exception
          if not e.keep_output:
              clear_output()
              env.sos_dict['_output'] = sos_targets([])
          if e.message:
              env.logger.info(e.message)
          return {
              'ret_code': 0,
              'task': task_id,
              'input': sos_targets([]),
              'output': env.sos_dict['_output'],
              'depends': sos_targets([]),
              'shared': {}
          }
      except KeyboardInterrupt:
          env.logger.error(f'{task_id} ``interrupted``')
          raise
      except subprocess.CalledProcessError as e:
          return {
              'ret_code': e.returncode,
              'task': task_id,
              'shared': {},
              'exception': RuntimeError(e.stderr)
          }
      except ProcessKilled:
          env.logger.error(f'{task_id} ``interrupted``')
          raise
      except Exception as e:
          msg = get_traceback_msg(e)
          # env.logger.error(f'{task_id} ``failed``: {msg}')
          with open(
                  os.path.join(
                      os.path.expanduser('~'), '.sos', 'tasks', task_id + '.err'),
                  'a') as err:
              err.write(msg + '\n')
          return {
              'ret_code': 1,
              'exception': RuntimeError(msg),
              'task': task_id,
              'shared': {}
          }
      finally:
          os.chdir(orig_dir)
  
      return collect_task_result(task_id, sos_dict, signature=sig)