Merge branch 'fugaku' into 'main' (81419466) · Commits · ExaDigiT / sim-raps

config/fugaku/power.json

0 → 100644

+18 −0

Original line number	Diff line number	Diff line
		{
		"POWER_GPU_IDLE": 0,
		"POWER_GPU_MAX": 0,
		"POWER_CPU_IDLE": 30,
		"POWER_CPU_MAX": 150,
		"POWER_MEM": 10,
		"POWER_NIC": 0,
		"POWER_NVME": 0,
		"POWER_SWITCH": 0,
		"POWER_CDU": 0,
		"POWER_UPDATE_FREQ": 10,
		"RECTIFIER_PEAK_THRESHOLD": 0,
		"SIVOC_LOSS_CONSTANT": 13,
		"SIVOC_EFFICIENCY": 0.98,
		"RECTIFIER_LOSS_CONSTANT": 17,
		"RECTIFIER_EFFICIENCY": 0.96,
		"POWER_COST": 0.094
		}

config/fugaku/scheduler.json

0 → 100644

+18 −0

Original line number	Diff line number	Diff line
		{
		"SEED": 42,
		"JOB_ARRIVAL_TIME": 10,
		"MTBF": 11,
		"MAX_TIME": 88200,
		"TRACE_QUANTA": 10,
		"MIN_WALL_TIME": 3600,
		"MAX_WALL_TIME": 43200,
		"UI_UPDATE_FREQ": 3600,
		"MAX_NODES_PER_JOB": 3000,
		"JOB_END_PROBS": {
		"COMPLETED": 0.63,
		"FAILED": 0.13,
		"CANCELLED": 0.12,
		"TIMEOUT": 0.11,
		"NODE_FAIL": 0.01
		}
		}

config/fugaku/system.json

0 → 100644

+20 −0

Original line number	Diff line number	Diff line
		{
		"NUM_CDUS": 24,
		"RACKS_PER_CDU": 18,
		"NODES_PER_RACK": 368,
		"RECTIFIERS_PER_RACK": 8,
		"CHASSIS_PER_RACK": 8,
		"NODES_PER_BLADE": 1,
		"SWITCHES_PER_CHASSIS": 2,
		"NICS_PER_NODE": 1,
		"RECTIFIERS_PER_CHASSIS": 1,
		"NODES_PER_RECTIFIER": 48,
		"MISSING_RACKS": [],
		"DOWN_NODES": [],
		"CPUS_PER_NODE": 1,
		"GPUS_PER_NODE": 0,
		"CPU_PEAK_FLOPS": 3.379E12,
		"GPU_PEAK_FLOPS": 0,
		"CPU_FP_RATIO": 0.82,
		"GPU_FP_RATIO": 0.0
		}

raps/dataloaders/fugaku.py

0 → 100644

+111 −0

Original line number	Diff line number	Diff line
		"""
		Download parquet files from https://zenodo.org/records/11467483

		Note that F-Data doesn't give a list of nodes used, so we set 'scheduled_nodes' to None
		which triggers the scheduler to schedule the nodes itself.

		Also, power in F-Data is only given at node-level. We can use node-level power by
		adding the --validate option.

		The --reschedule will compute submit times from Poisson distribution, instead of using
		the submit times given in F-Data.

		python main.py --system fugaku -f /path/to/21_04.parquet --reschedule --validate --reschedule

		"""
		import pandas as pd
		from tqdm import tqdm
		from ..job import job_dict
		from ..utils import next_arrival


		def load_data(path, **kwargs):
		"""
		Loads data from the given Parquet file path and returns job info.

		Parameters:
		path (str): Path to the Parquet file.

		Returns:
		list: List of job dictionaries.
		"""
		# Load the parquet file
		parquet_file = path[0] # Assuming path is a list containing the path to the parquet file
		df = pd.read_parquet(parquet_file)

		# Process the DataFrame and pass to load_data_from_df
		return load_data_from_df(df, **kwargs)


		def load_data_from_df(df, **kwargs):
		"""
		Processes DataFrame to extract relevant job information and computes the time offset
		based on the earliest submission time.

		Parameters:
		df (pd.DataFrame): DataFrame containing job information.

		Returns:
		list: List of job dictionaries.
		"""
		encrypt_bool = kwargs.get('encrypt')
		fastforward = kwargs.get('fastforward')
		reschedule = kwargs.get('reschedule')
		validate = kwargs.get('validate')
		jid = kwargs.get('jid', '*')

		if fastforward: print(f"fast-forwarding {fastforward} seconds")

		job_list = []

		# Convert 'adt' (submit time) to datetime and find the earliest submission time
		df['adt'] = pd.to_datetime(df['adt'], errors='coerce')
		earliest_submit_time = df['adt'].min()

		# Loop through the DataFrame rows to extract job information
		for _, row in tqdm(df.iterrows(), total=len(df), desc="Processing Jobs"):
		nodes_required = row['nnumr'] if 'nnumr' in df.columns else 0
		name = row['jnam'] if 'jnam' in df.columns else 'unknown'

		if validate:
		cpu_trace = row['avgpcon']
		gpu_trace = cpu_trace

		else:
		cpu_trace = row['perf1'] if 'perf1' in df.columns else 0 # Assuming some performance metric as cpu_trace
		gpu_trace = 0 # Set to 0 as GPU trace is not explicitly provided

		wall_time = row['duration'] if 'duration' in df.columns else 0
		end_state = row['exit state'] if 'exit state' in df.columns else 'unknown'
		#scheduled_nodes = row['nnuma'] if 'nnuma' in df.columns else 0
		scheduled_nodes = None
		submit_time = row['adt'] if 'adt' in df.columns else earliest_submit_time
		if reschedule: # Let the scheduler reschedule the jobs
		time_offset = next_arrival()
		else:
		time_offset = (submit_time - earliest_submit_time).total_seconds() # Compute time offset in seconds

		job_id = row['jid'] if 'jid' in df.columns else 'unknown'
		priority = row['pri'] if 'pri' in df.columns else 0

		# Create job dictionary
		job_info = job_dict(
		nodes_required=nodes_required,
		name=name,
		cpu_trace=cpu_trace,
		gpu_trace=gpu_trace,
		wall_time=wall_time,
		end_state=end_state,
		scheduled_nodes=scheduled_nodes,
		time_offset=time_offset,
		job_id=job_id,
		priority=priority
		)

		job_list.append(job_info)

		return job_list


		# Sample usage:
		# fugaku_jobs = load_data(['/path/to/21_04.parquet'])

raps/scheduler.py

+12 −2

Original line number	Diff line number	Diff line
		@@ -86,6 +86,15 @@ class TickData:
		pue: Optional[float]


		def get_utilization(trace, time_quanta_index):
		if isinstance(trace, (list, np.ndarray)):
		return trace[time_quanta_index]
		elif isinstance(trace, (int, float)):
		return float(trace)
		else:
		raise TypeError(f"Invalid type for utilization: {type(trace)}.")


		class Scheduler:
		"""Job scheduler and simulation manager."""
		def __init__(self, total_nodes, down_nodes, power_manager, flops_manager, \
		@@ -249,8 +258,9 @@ class Scheduler:
		job.running_time = self.current_time - job.start_time

		time_quanta_index = (self.current_time - job.start_time) // TRACE_QUANTA
		cpu_util = job.cpu_trace[time_quanta_index]
		gpu_util = job.gpu_trace[time_quanta_index]

		cpu_util = get_utilization(job.cpu_trace, time_quanta_index)
		gpu_util = get_utilization(job.gpu_trace, time_quanta_index)

		self.flops_manager.update_flop_state(job.scheduled_nodes, cpu_util, gpu_util)
		job.power = self.power_manager.update_power_state(job.scheduled_nodes,