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
If this is a master task, execute as individual tasks