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
def _execute_sub_tasks ( task_id , params , sig_content , verbosity , runmode , sigmode , monitor_interval , resource_monitor_interval , master_runtime ) : 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' ) 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}' ) 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 ( ) else : results = [ ] for tid , tdef in params . task_stack : if hasattr ( params , 'common_dict' ) : tdef . sos_dict . update ( params . common_dict ) 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 ) 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' ] ) 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 '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 : 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