Commit 36fd4ef8 authored by Maiterth, Matthias's avatar Maiterth, Matthias
Browse files

adjusted dataloaders and smoke test to new job_dict. (job dict has named arguments only)

parent 2d60432c

raps/dataloaders/adastraMI250.py

+11 −2
Original line number Diff line number Diff line
@@ -177,8 +177,17 @@ def load_data_from_df(jobs_df: pd.DataFrame, **kwargs): 
        trace_end_time = start_time



        if wall_time > 0:

            job_info = job_dict(nodes_required, name, account, cpu_trace, gpu_trace, [],[],

                                end_state, scheduled_nodes, job_id, priority,

            job_info = job_dict(nodes_required=nodes_required,

                                name=name,

                                account=account,

                                cpu_trace=cpu_trace,

                                gpu_trace=gpu_trace,

                                ntx_trace=[],

                                nrx_trace=[],

                                end_state=end_state,

                                scheduled_nodes=scheduled_nodes,

                                id=job_id,

                                priority=priority,

                                submit_time=submit_time,

                                time_limit=time_limit,

                                start_time=start_time,

raps/dataloaders/frontier.py

+16 −7
Original line number Diff line number Diff line
@@ -280,9 +280,18 @@ def load_data_from_df(jobs_df: pd.DataFrame, jobprofile_df: pd.DataFrame, **kwar 
            continue  # SKIP!



        if gpu_trace.size > 0 and (jid == job_id or jid == '*'):  # and time_submit >= 0:

            job_info = job_dict(nodes_required, name, account, cpu_trace, gpu_trace, [], [],

                                end_state, scheduled_nodes,

                                job_id, priority,  # partition missing

            job_info = job_dict(

                nodes_required=nodes_required,

                name=name,

                account=account,

                cpu_trace=cpu_trace,

                gpu_trace=gpu_trace,

                nrx_trace=[],

                ntx_trace=[],

                end_state=end_state,

                scheduled_nodes=scheduled_nodes,

                id=job_id,

                priority=priority,  # partition missing

                submit_time=submit_time, time_limit=time_limit,

                start_time=start_time, end_time=end_time,

                wall_time=wall_time, trace_time=trace_time,

raps/dataloaders/fugaku.py

+1 −3
Original line number Diff line number Diff line
@@ -153,9 +153,7 @@ def load_data_from_df(df, **kwargs): 
            trace_time=trace_time,

            trace_start_time=trace_start_time,

            trace_end_time=trace_end_time,

            trace_missing_values=trace_missing_values

        )



            trace_missing_values=trace_missing_values)

        job_list.append(job_info)



    return job_list, telemetry_start, telemetry_end

raps/dataloaders/lassen.py

+12 −13
Original line number Diff line number Diff line
@@ -167,18 +167,18 @@ def load_data_from_df(allocation_df, node_df, step_df, **kwargs): 
            print('scheduled_nodes:', nodes_required, scheduled_nodes)



        if wall_time >= 0:

            job_info = job_dict(nodes_required,

                                name,

                                account,

                                cpu_trace,

                                gpu_trace,

                                net_tx,

                                net_rx,

                                end_state,

                                scheduled_nodes,

                                job_id,

                                priority,

                                partition,

            job_info = job_dict(nodes_required=nodes_required,

                                name=name,

                                account=account,

                                cpu_trace=cpu_trace,

                                gpu_trace=gpu_trace,

                                ntx_trace=net_tx,

                                nrx_trace=net_rx,

                                end_state=end_state,

                                scheduled_nodes=scheduled_nodes,

                                id=job_id,

                                priority=priority,

                                partition=partition,

                                submit_time=submit_time,

                                time_limit=time_limit,

                                start_time=start_time,
@@ -188,7 +188,6 @@ def load_data_from_df(allocation_df, node_df, step_df, **kwargs): 
                                trace_start_time=trace_start_time,

                                trace_end_time=trace_end_time,

                                trace_missing_values=trace_missing_values)



            job_list.append(job_info)



    return job_list, telemetry_start_time, telemetry_end_time

raps/schedulers/scheduleflow.py

+6 −0
Original line number Diff line number Diff line
@@ -37,6 +37,12 @@ class Scheduler: 
        # self.sf_action_list = []  # list as returned from sf_scheduler.stop_job





    def gif(self):

        logs = self._sf_runtime.get_stats()

        #vis_hanlder = _intScheduleFlow.VizualizationEngine(self.sf_scheduler.

        self._sf_runtime._Runtime__generate_gif()





    def sort_jobs(self, queue, accounts=None):

        """

        Optionally, pre-sort jobs.