docs: def _execute_sub_tasks(task_id, params, sig_content, verbosity, runmode,
                         sigmode, monitor_interval, resource_monitor_interval,
                         master_runtime):
      '''If this is a master task, execute as individual tasks'''
      m = ProcessMonitor(
          task_id,
          monitor_interval=monitor_interval,
          resource_monitor_interval=resource_monitor_interval,
          max_walltime=params.sos_dict['_runtime'].get('max_walltime', None),
          max_mem=params.sos_dict['_runtime'].get('max_mem', None),
          max_procs=params.sos_dict['_runtime'].get('max_procs', None),
          sos_dict=params.sos_dict)
      m.start()
  
      env.logger.info(f'{task_id} ``started``')
  
      master_out = os.path.join(
          os.path.expanduser('~'), '.sos', 'tasks', task_id + '.out')
      master_err = os.path.join(
          os.path.expanduser('~'), '.sos', 'tasks', task_id + '.err')
      # if this is a master task, calling each sub task
      with open(master_out, 'wb') as out, open(master_err, 'wb') as err:
  
          def copy_out_and_err(result):
              tid = result['task']
              out.write(
                  f'{tid}: {"completed" if result["ret_code"] == 0 else "failed"}\n'
                  .encode())
              if 'output' in result:
                  out.write(f'output: {result["output"]}\n'.encode())
              sub_out = os.path.join(
                  os.path.expanduser('~'), '.sos', 'tasks', tid + '.out')
              if os.path.isfile(sub_out):
                  with open(sub_out, 'rb') as sout:
                      out.write(sout.read())
                  try:
                      os.remove(sub_out)
                  except Exception as e:
                      env.logger.warning(f'Failed to remove {sub_out}: {e}')
  
              sub_err = os.path.join(
                  os.path.expanduser('~'), '.sos', 'tasks', tid + '.err')
              if 'exception' in result:
                  err.write(str(result['exception']).encode())
              err.write(
                  f'{tid}: {"completed" if result["ret_code"] == 0 else "failed"}\n'
                  .encode())
              if os.path.isfile(sub_err):
                  with open(sub_err, 'rb') as serr:
                      err.write(serr.read())
                  try:
                      os.remove(sub_err)
                  except Exception as e:
                      env.logger.warning(f'Failed to remove {sub_err}: {e}')
  
              # remove other files as well
              try:
                  remove_task_files(tid, ['.out', '.err'])
              except Exception as e:
                  env.logger.debug(f'Failed to remove files {tid}: {e}')
  
          if params.num_workers > 1:
              from multiprocessing.pool import Pool
              p = Pool(params.num_workers)
              results = []
              for tid, tdef in params.task_stack:
                  if hasattr(params, 'common_dict'):
                      tdef.sos_dict.update(params.common_dict)
                  results.append(
                      p.apply_async(
                          _execute_task, ((tid, tdef, {
                              tid: sig_content.get(tid, {})
                          }), verbosity, runmode, sigmode, None, None, {
                              x: master_runtime.get(x, {})
                              for x in ('_runtime', tid)
                          }),
                          callback=copy_out_and_err))
              for idx, r in enumerate(results):
                  results[idx] = r.get()
              p.close()
              p.join()
              # we wait for all results to be ready to return or raise
              # but we only raise exception for one of the subtasks
              # for res in results:
              #     if 'exception' in res:
              #         failed = [x.get("task", "")
              #                   for x in results if "exception" in x]
              #         env.logger.error(
              #             f'{task_id} ``failed`` due to failure of subtask{"s" if len(failed) > 1 else ""} {", ".join(failed)}')
              #         return {'ret_code': 1, 'exception': res['exception'], 'task': task_id}
          else:
              results = []
              for tid, tdef in params.task_stack:
                  if hasattr(params, 'common_dict'):
                      tdef.sos_dict.update(params.common_dict)
                  # no monitor process for subtasks
                  res = _execute_task(
                      (tid, tdef, {
                          tid: sig_content.get(tid, {})
                      }),
                      verbosity=verbosity,
                      runmode=runmode,
                      sigmode=sigmode,
                      monitor_interval=None,
                      resource_monitor_interval=None,
                      master_runtime={
                          x: master_runtime.get(x, {}) for x in ('_runtime', tid)
                      })
                  try:
                      copy_out_and_err(res)
                  except Exception as e:
                      env.logger.warning(
                          f'Failed to copy result of subtask {tid}: {e}')
                  results.append(res)
              # for res in results:
              #     if 'exception' in res:
              #         failed = [x.get("task", "")
              #                   for x in results if "exception" in x]
              #         env.logger.error(
              #             f'{task_id} ``failed`` due to failure of subtask{"s" if len(failed) > 1 else ""} {", ".join(failed)}')
              #         return {'ret_code': 1, 'exception': res['exception'], 'task': task_id}
      #
      # now we collect result
      all_res = {
          'ret_code': 0,
          'output': None,
          'subtasks': {},
          'shared': {},
          'skipped': 0,
          'signature': {}
      }
      for tid, x in zip(params.task_stack, results):
          all_res['subtasks'][tid[0]] = x
  
          if 'exception' in x:
              all_res['exception'] = x['exception']
              all_res['ret_code'] += 1
              continue
          all_res['ret_code'] += x['ret_code']
          if all_res['output'] is None:
              all_res['output'] = copy.deepcopy(x['output'])
          else:
              try:
                  all_res['output'].extend(x['output'], keep_groups=True)
              except Exception as e:
                  env.logger.warning(
                      f"Failed to extend output {all_res['output']} with {x['output']}"
                  )
          all_res['shared'].update(x['shared'])
          # does not care if one or all subtasks are executed or skipped.
          all_res['skipped'] += x.get('skipped', 0)
          if 'signature' in x:
              all_res['signature'].update(x['signature'])
  
      if all_res['ret_code'] != 0:
          if all_res['ret_code'] == len(results):
              env.logger.info(f'All {len(results)} tasks in {task_id} ``failed``')
          else:
              env.logger.info(
                  f'{all_res["ret_code"]} of {len(results)} tasks in {task_id} ``failed``'
              )
          # if some failed, some skipped, not skipped
          if 'skipped' in all_res:
              all_res.pop('skipped')
      elif all_res['skipped']:
          if all_res['skipped'] == len(results):
              env.logger.info(
                  f'All {len(results)} tasks in {task_id} ``ignored`` or skipped')
          else:
              # if only partial skip, we still save signature and result etc
              env.logger.info(
                  f'{all_res["skipped"]} of {len(results)} tasks in {task_id} ``ignored`` or skipped'
              )
              all_res.pop('skipped')
      else:
          env.logger.info(f'All {len(results)} tasks in {task_id} ``completed``')
      return all_res