From 337932050c05d976748b0cc42cc1822134fc4ffd Mon Sep 17 00:00:00 2001 From: enlorenz Date: Mon, 1 Dec 2025 17:38:39 +0100 Subject: [PATCH 01/11] Added logic to account for Dropped jobs due to age. Drop counter for step and baseline_step. Assign_jobs now delivers "launched" and "dropped" numbers. Added penalty in reward function (not normalized or weighted yet). Centralized "remove job from list". Changed from "terminated" to "truncated", to account for "time-limit" end. Fix: Untracked mergers now up to date Minor Fixes: CodeRabbit's suggestions --- environment.py | 139 +++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 112 insertions(+), 27 deletions(-) diff --git a/environment.py b/environment.py index d34e131..6aed4fa 100644 --- a/environment.py +++ b/environment.py @@ -37,6 +37,9 @@ EPISODE_HOURS = WEEK_HOURS * 2 +PENALTY_DROPPED_JOB = -5.0 # explicit penalty for each job dropped due to exceeding MAX_JOB_AGE + + # possible rewards: # - cost savings (due to disabled nodes) # - reduced conventional energy usage @@ -74,6 +77,24 @@ def env_print(self, *args): """Prints only if the render mode is 'human'.""" if self.render_mode == 'human': print(*args) + + + # Centralize Job removal - Sanity Check, that next_empty_slot always points to FIRST empty: + def _clear_job_slot(self,job_queue_2d, job_idx, next_empty_slot_ref=None): + job_queue_2d[job_idx] = [0, 0, 0, 0] + if next_empty_slot_ref is not None and job_idx < next_empty_slot_ref[0]: + next_empty_slot_ref[0] = job_idx + + # Validator for debugging: + def _validate_next_empty(self,job_queue_2d, next_empty): + n = len(job_queue_2d) + if next_empty < n: + assert job_queue_2d[next_empty][0] == 0, "next_empty_slot not empty" + # everything before must be non-empty + if next_empty > 0: + assert np.all(job_queue_2d[:next_empty, 0] != 0), "hole before next_empty_slot" + + def __init__(self, weights: Weights, @@ -154,6 +175,14 @@ def __init__(self, self.total_cost = 0 self.baseline_cost = 0 self.baseline_cost_off = 0 + # Job tracking metrics for agent + self.jobs_dropped = 0 + self.dropped_this_episode = 0 + + # Job tracking metrics for baseline + self.baseline_jobs_dropped = 0 + self.baseline_dropped_this_episode = 0 + self.reset_state() @@ -270,6 +299,27 @@ def reset_state(self): self.baseline_jobs_completed = 0 self.baseline_total_job_wait_time = 0 self.baseline_max_queue_size_reached = 0 + # Job tracking metrics for agent + self.jobs_dropped = 0 + + # Job tracking metrics for baseline + self.baseline_jobs_dropped = 0 + self.baseline_dropped_this_episode = 0 + + # Agent + self.jobs_rejected_queue_full = 0 # new jobs we couldn't even enqueue + + # Baseline + self.baseline_jobs_rejected_queue_full = 0 + + self.dropped_this_episode = 0 + self.baseline_dropped_this_episode = 0 + self.prev_excess_dropped = 0 + + # if self.external_jobs and not self.workload_gen: + # self.jobs_sampler = DurationSampler() + # self.jobs_sampler.parse_jobs(self.external_jobs, 60) + # self.jobs_sampler.precalculate_hourly_jobs(CORES_PER_NODE, MAX_NODES_PER_JOB) def step(self, action): self.current_step += 1 @@ -368,6 +418,9 @@ def step(self, action): baseline_cost, baseline_cost_off = self.baseline_step(current_price, new_jobs_count, new_jobs_durations, new_jobs_nodes, new_jobs_cores) self.baseline_cost += baseline_cost self.baseline_cost_off += baseline_cost_off + excess = max(0, self.dropped_this_episode - self.baseline_dropped_this_episode) + self.new_excess = max(0, excess - self.prev_excess_dropped) + self.prev_excess_dropped = excess # calculate reward step_reward, step_cost = self.calculate_reward(num_used_nodes, num_idle_nodes, current_price, average_future_price, num_off_nodes, num_launched_jobs, num_node_changes, job_queue_2d, num_unprocessed_jobs) @@ -411,8 +464,8 @@ def step(self, action): plot(self, EPISODE_HOURS, MAX_NODES, True, False, self.current_step) self.next_plot_save += self.steps_per_iteration print(self.next_plot_save) - - terminated = True + truncated = True # Changed distinction, so we can identify if ended due to time-limit + terminated = False # Added sanity check # Record episode costs for long-term analysis self.record_episode_completion() @@ -525,33 +578,22 @@ def assign_jobs_to_available_nodes(self, job_queue_2d, nodes, cores_available, r if job_duration <= 0: continue - # Use NumPy vectorized operations to find candidate nodes + # Candidates: node is on and has enough free cores mask = (nodes >= 0) & (cores_available >= job_cores_per_node) candidate_nodes = np.where(mask)[0] - # Check if we have enough nodes for this job if len(candidate_nodes) >= job_nodes: - # We found enough nodes, assign the job + # Assign job to first job_nodes candidates job_allocation = [] - - # Use the first 'job_nodes' candidates for i in range(job_nodes): - node_idx = candidate_nodes[i] - - # Update node resources - cores_available[node_idx] -= job_cores_per_node - - # If this node now has a longer job duration, update it - if nodes[node_idx] < job_duration: - nodes[node_idx] = job_duration + node_idx = int(candidate_nodes[i]) + cores_available[node_idx] -= int(job_cores_per_node) + nodes[node_idx] = max(int(nodes[node_idx]), int(job_duration)) + job_allocation.append((node_idx, int(job_cores_per_node))) - # Add to job allocation - job_allocation.append((node_idx, job_cores_per_node)) - - # Add job to running_jobs running_jobs[self.next_job_id] = { - 'duration': job_duration, - 'allocation': job_allocation + "duration": int(job_duration), + "allocation": job_allocation, } self.next_job_id += 1 @@ -565,15 +607,40 @@ def assign_jobs_to_available_nodes(self, job_queue_2d, nodes, cores_available, r # Track job completion and wait time if is_baseline: self.baseline_jobs_completed += 1 - self.baseline_total_job_wait_time += job_age + self.baseline_total_job_wait_time += int(job_age) else: self.jobs_completed += 1 - self.total_job_wait_time += job_age + self.total_job_wait_time += int(job_age) + + num_launched += 1 + continue + + # Not enough resources -> job waits and ages (or gets dropped) + new_age = int(job_age) + 1 + + if new_age > MAX_JOB_AGE: + # Clear job from queue + job_queue_2d[job_idx] = [0, 0, 0, 0] - num_processed_jobs += 1 + # Update next_empty_slot if we cleared a slot before it + if job_idx < next_empty_slot: + next_empty_slot = job_idx + num_dropped += 1 + + if is_baseline: + self.baseline_jobs_dropped += 1 + self.baseline_dropped_this_episode += 1 + else: + self.jobs_dropped += 1 + self.dropped_this_episode += 1 else: - # Not enough resources for this job, increment its age - job_queue_2d[job_idx][1] += 1 + job_queue_2d[job_idx][1] = new_age + + + # DEBUG CHECK for next_empty -> Add a Flag to be called + if False: + if getattr(self, "strict_checks", False) and (self.current_step % 100 == 0): + self._validate_next_empty(job_queue_2d, next_empty_slot) return num_processed_jobs, next_empty_slot @@ -584,6 +651,7 @@ def baseline_step(self, current_price, new_jobs_count, new_jobs_durations, new_j new_baseline_jobs, self.baseline_next_empty_slot = self.add_new_jobs(job_queue_2d, new_jobs_count, new_jobs_durations, new_jobs_nodes, new_jobs_cores, self.baseline_next_empty_slot) self.baseline_jobs_submitted += len(new_baseline_jobs) + self.baseline_jobs_rejected_queue_full += (new_jobs_count - len(new_baseline_jobs)) _, self.baseline_next_empty_slot = self.assign_jobs_to_available_nodes(job_queue_2d, self.baseline_state['nodes'], self.baseline_cores_available, self.baseline_running_jobs, self.baseline_next_empty_slot, is_baseline=True) @@ -636,6 +704,9 @@ def calculate_reward(self, num_used_nodes, num_idle_nodes, current_price, averag self.price_rewards.append(price_reward_norm * 100) self.job_age_penalties.append(job_age_penalty_norm * 100) self.idle_penalties.append(idle_penalty_norm * 100) + + # Penalty is always non-positive + drop_penalty = min(0, PENALTY_DROPPED_JOB * self.new_excess) reward = ( efficiency_reward_weighted @@ -643,6 +714,7 @@ def calculate_reward(self, num_used_nodes, num_idle_nodes, current_price, averag + price_reward_weighted + job_age_penalty_weighted + idle_penalty_weighted + + drop_penalty ) self.env_print(f" > $$$TOTAL: {reward:.4f} = {efficiency_reward_weighted:.4f} + {price_reward_weighted:.4f} + {idle_penalty_weighted:.4f} + {job_age_penalty_weighted:.4f}") @@ -755,6 +827,10 @@ def record_episode_completion(self): # Calculate completion rates completion_rate = (self.jobs_completed / self.jobs_submitted * 100) if self.jobs_submitted > 0 else 0 baseline_completion_rate = (self.baseline_jobs_completed / self.baseline_jobs_submitted * 100) if self.baseline_jobs_submitted > 0 else 0 + + drop_rate = (self.jobs_dropped / self.jobs_submitted * 100) if self.jobs_submitted else 0.0 + baseline_drop_rate = (self.baseline_jobs_dropped / self.baseline_jobs_submitted * 100) if self.baseline_jobs_submitted else 0.0 + episode_data = { 'episode': self.current_episode, @@ -777,7 +853,16 @@ def record_episode_completion(self): 'baseline_jobs_completed': self.baseline_jobs_completed, 'baseline_avg_wait_time': float(baseline_avg_wait_time), 'baseline_completion_rate': float(baseline_completion_rate), - 'baseline_max_queue_size': self.baseline_max_queue_size_reached + 'baseline_max_queue_size': self.baseline_max_queue_size_reached, + + #Drop metrics + "jobs_dropped": self.jobs_dropped, + "drop_rate": float(drop_rate), + "jobs_rejected_queue_full": self.jobs_rejected_queue_full, + + "baseline_jobs_dropped": self.baseline_jobs_dropped, + "baseline_drop_rate": float(baseline_drop_rate), + "baseline_jobs_rejected_queue_full": self.baseline_jobs_rejected_queue_full, } self.episode_costs.append(episode_data) From 3a28447a6e9bb2a6d66a2a0d395ff82a6c7941ba Mon Sep 17 00:00:00 2001 From: enlorenz Date: Mon, 1 Dec 2025 17:43:49 +0100 Subject: [PATCH 02/11] Seeding for consistent randomizer: Seeds RNG once during reset, same RNG is used in any function that uses any type of RNG. Conserves determinism for training. Adjusted "sampler_hourly" to allow for this logic, and the way it's read in -> Now it's a "self." function. --- environment.py | 11 ++++++++++- sampler_hourly.py | 17 ++++++++++------- 2 files changed, 20 insertions(+), 8 deletions(-) diff --git a/environment.py b/environment.py index 6aed4fa..700ea57 100644 --- a/environment.py +++ b/environment.py @@ -10,8 +10,12 @@ from plot import plot, plot_reward from sampler_duration import durations_sampler from sampler_jobs import jobs_sampler +from sampler_jobs import DurationSampler from sampler_hourly import hourly_sampler +# For deterministic RNG +from gymnasium.utils import seeding + init() # Initialize colorama WEEK_HOURS = 168 @@ -149,6 +153,10 @@ def __init__(self, self.episode_costs = [] self.prices = Prices(self.external_prices) + + #Initialize deterministic RNG, instead of global RNG + self.np_random = None + self._seed = None if self.external_durations: durations_sampler.init(self.external_durations) @@ -347,7 +355,8 @@ def step(self, action): new_jobs_count = 0 if self.external_jobs: - jobs = jobs_sampler.sample_one_hourly(wrap=True)['hourly_jobs'] + # jobs = jobs_sampler.sample_one_hourly(wrap=True)['hourly_jobs'] + jobs = self.jobs_sampler.sample_one_hourly(wrap=True)["hourly_jobs"] if len(jobs) > 0: for job in jobs: new_jobs_count += 1 diff --git a/sampler_hourly.py b/sampler_hourly.py index 23eba45..f32d4b0 100644 --- a/sampler_hourly.py +++ b/sampler_hourly.py @@ -100,7 +100,7 @@ def parse_jobs(self, filepath): zero_pct = (dist["job_count"].count(0) / len(dist["job_count"]) * 100) if dist["job_count"] else 0 print(f" Hour {hour:2d}: avg={avg_count:.1f} jobs/hour, {zero_pct:.0f}% zero-job samples, {len(dist['durations'])} total jobs") - def sample(self, hour_of_day): + def sample(self, hour_of_day: int, rng, max_jobs: int | None = None): """ Sample jobs for a given hour of day. @@ -119,17 +119,20 @@ def sample(self, hour_of_day): dist = self.hour_distributions[hour_of_day] # Sample number of jobs for this hour (can be 0) - num_jobs = random.choice(dist["job_count"]) + num_jobs = rng.choice(dist["job_count"]) - if num_jobs == 0: + if num_jobs <= 0: return [] - # Sample characteristics for each job + if max_jobs is not None: + num_jobs = min(num_jobs, int(max_jobs)) + if num_jobs <= 0: + return [] jobs = [] for _ in range(num_jobs): - duration = random.choice(dist["durations"]) - nodes = random.choice(dist["nodes"]) - cores = random.choice(dist["cores_per_node"]) + duration = rng.choice(dist["durations"]) + nodes = rng.choice(dist["nodes"]) + cores = rng.choice(dist["cores_per_node"]) jobs.append({ "duration": duration, From 05e17744c7ded2ebad20d6dc89b602f8e91d2870 Mon Sep 17 00:00:00 2001 From: enlorenz Date: Mon, 1 Dec 2025 17:47:34 +0100 Subject: [PATCH 03/11] Added an alternative "prices" script. Before, when iterating over prices, it would iterate too much. Agent and env became out of synch. "get_predicted_prices" now "Advance_and_get..." Reset in env, now also resets prices state. CodeRabbit Fixes: Corrected seeding. Removed double initialization of jobs sampler. Corrected num_processed jobs. Renamed to prices_deterministic. Cleaned random in sampler. Added dataclass replace in workloadgen. Supposed to provide O(1) instead of O(n). Extra Fixes: Renamed unclear script names. Removed unused functions in environment. And minor adjustments. --- environment.py | 32 +++---- prices_deterministic.py | 144 ++++++++++++++++++++++++++++++++ sampler_hourly.py | 1 - test_determinism_workloadgen.py | 72 ++++++++++++++++ test_sanity_workloadgen.py | 43 ++++++++++ train.py | 17 +++- workloadgen.py | 110 ++++++++++++++++++++++++ 7 files changed, 392 insertions(+), 27 deletions(-) create mode 100644 prices_deterministic.py create mode 100644 test_determinism_workloadgen.py create mode 100644 test_sanity_workloadgen.py create mode 100644 workloadgen.py diff --git a/environment.py b/environment.py index 700ea57..519938a 100644 --- a/environment.py +++ b/environment.py @@ -5,7 +5,8 @@ import numpy as np from colorama import init, Fore -from prices import Prices +#from prices import Prices +from prices_deterministic import Prices # Test re-worked prices script from weights import Weights from plot import plot, plot_reward from sampler_duration import durations_sampler @@ -82,13 +83,6 @@ def env_print(self, *args): if self.render_mode == 'human': print(*args) - - # Centralize Job removal - Sanity Check, that next_empty_slot always points to FIRST empty: - def _clear_job_slot(self,job_queue_2d, job_idx, next_empty_slot_ref=None): - job_queue_2d[job_idx] = [0, 0, 0, 0] - if next_empty_slot_ref is not None and job_idx < next_empty_slot_ref[0]: - next_empty_slot_ref[0] = job_idx - # Validator for debugging: def _validate_next_empty(self,job_queue_2d, next_empty): n = len(job_queue_2d) @@ -248,7 +242,11 @@ def __init__(self, }) def reset(self, seed = None, options = None): + super().reset(seed=seed) + self.np_random, self._seed = seeding.np_random(seed) + self.reset_state() + self.prices.reset() self.state = { # Initialize all nodes to be 'online but free' (0) @@ -256,7 +254,7 @@ def reset(self, seed = None, options = None): # Initialize job queue to be empty 'job_queue': np.zeros((MAX_QUEUE_SIZE * 4), dtype=np.int32), # Initialize predicted prices array - 'predicted_prices': self.prices.get_predicted_prices(), + 'predicted_prices': self.prices.predicted_prices.copy(), } self.baseline_state = { @@ -323,11 +321,6 @@ def reset_state(self): self.dropped_this_episode = 0 self.baseline_dropped_this_episode = 0 self.prev_excess_dropped = 0 - - # if self.external_jobs and not self.workload_gen: - # self.jobs_sampler = DurationSampler() - # self.jobs_sampler.parse_jobs(self.external_jobs, 60) - # self.jobs_sampler.precalculate_hourly_jobs(CORES_PER_NODE, MAX_NODES_PER_JOB) def step(self, action): self.current_step += 1 @@ -336,7 +329,7 @@ def step(self, action): self.current_episode += 1 self.env_print(Fore.GREEN + f"\n[[[ Starting episode: {self.current_episode}, step: {self.current_step}, hour: {self.current_hour}" + Fore.RESET) - self.state['predicted_prices'] = self.prices.get_predicted_prices() + self.state['predicted_prices'] = self.prices.advance_and_get_predicted_prices() current_price = self.state['predicted_prices'][0] self.env_print("predicted_prices: ", np.array2string(self.state['predicted_prices'], separator=" ", max_line_width=np.inf, formatter={'float_kind': lambda x: "{:05.2f}".format(x)})) @@ -580,6 +573,7 @@ def adjust_nodes(self, action_type, action_magnitude, nodes, cores_available): def assign_jobs_to_available_nodes(self, job_queue_2d, nodes, cores_available, running_jobs, next_empty_slot, is_baseline=False): num_processed_jobs = 0 + num_dropped = 0 for job_idx, job in enumerate(job_queue_2d): job_duration, job_age, job_nodes, job_cores_per_node = job @@ -621,7 +615,7 @@ def assign_jobs_to_available_nodes(self, job_queue_2d, nodes, cores_available, r self.jobs_completed += 1 self.total_job_wait_time += int(job_age) - num_launched += 1 + num_processed_jobs += 1 continue # Not enough resources -> job waits and ages (or gets dropped) @@ -645,12 +639,6 @@ def assign_jobs_to_available_nodes(self, job_queue_2d, nodes, cores_available, r else: job_queue_2d[job_idx][1] = new_age - - # DEBUG CHECK for next_empty -> Add a Flag to be called - if False: - if getattr(self, "strict_checks", False) and (self.current_step % 100 == 0): - self._validate_next_empty(job_queue_2d, next_empty_slot) - return num_processed_jobs, next_empty_slot def baseline_step(self, current_price, new_jobs_count, new_jobs_durations, new_jobs_nodes, new_jobs_cores): diff --git a/prices_deterministic.py b/prices_deterministic.py new file mode 100644 index 0000000..6d21c9e --- /dev/null +++ b/prices_deterministic.py @@ -0,0 +1,144 @@ +import numpy as np +import matplotlib.pyplot as plt + +class Prices: + ELECTRICITY_PRICE_BASE = 20 + PERCENTILE_MIN = 1 + PERCENTILE_MAX = 99 + PREDICTION_WINDOW = 24 + HISTORY_WINDOW = 24 + + def __init__(self, external_prices=None): + self.original_prices = external_prices + self.external_prices = None + + self.price_shift = 0 + self.price_index = 0 + self.price_history = [] + self.predicted_prices = None + + if self.original_prices is not None: + prices = np.asarray(self.original_prices, dtype=np.float32) + if prices.size == 0: + raise ValueError("external_prices must be a non-empty sequence") + + min_price = float(np.min(prices)) + if min_price < 1: + self.price_shift = 1 - min_price + prices = prices + self.price_shift + + self.external_prices = prices + self.MIN_PRICE = float(np.percentile(self.external_prices, self.PERCENTILE_MIN)) + self.MAX_PRICE = float(np.percentile(self.external_prices, self.PERCENTILE_MAX)) + else: + self.external_prices = None + # Keep your defaults for normalization bounds + self.MAX_PRICE = 24 + self.MIN_PRICE = 16 + + # IMPORTANT: initialize state in one place + self.reset() + + # ---------- core price model (pure) ---------- + def _synthetic_price_at(self, t: int) -> float: + # deterministic daily sinusoid + base = self.ELECTRICITY_PRICE_BASE + return float(max(1.0, base * (1 + 0.2 * np.sin((t % 24) / 24 * 2 * np.pi)))) + + def get_real_price(self, shifted_price): + return shifted_price - self.price_shift + + def reset(self): + """Reset internal timeline/state to episode start (determinism-critical).""" + self.price_index = self.PREDICTION_WINDOW + self.price_history = deque(maxlen=self.HISTORY_WINDOW) + + if self.external_prices is not None: + self.predicted_prices = np.array( + self.external_prices[:self.PREDICTION_WINDOW], + dtype=np.float32, + copy=True, + ) + else: + self.predicted_prices = np.array( + [self._synthetic_price_at(i) for i in range(self.PREDICTION_WINDOW)], + dtype=np.float32, + ) + + # ---------- *stateful* stepping (used by env.step) ---------- + def get_next_price(self): + if self.external_prices is not None: + new_price = float(self.external_prices[self.price_index % len(self.external_prices)]) + else: + new_price = self._synthetic_price_at(self.price_index) + + self.price_index += 1 + + self.price_history.append(new_price) + #if len(self.price_history) > self.HISTORY_WINDOW: + # self.price_history.pop(0) + # deque automatically removes oldest when maxlen exceeded | Keep the old line for now. Remove during next PR. + + return new_price + + def get_price_context(self): + history_avg = float(np.mean(self.price_history)) if self.price_history else None + future_avg = float(np.mean(self.predicted_prices)) + return history_avg, future_avg + + def advance_and_get_predicted_prices(self): # Changed name for readability + new_price = self.get_next_price() + self.predicted_prices = np.roll(self.predicted_prices, -1) + self.predicted_prices[-1] = new_price + return self.predicted_prices.copy() + + # ---------- NON-MUTATING utilities ---------- + def _generated_prices_for_stats(self, n: int): + # Do NOT touch price_index/history here. + return np.array([self._synthetic_price_at(i) for i in range(n)], dtype=np.float32) + + def get_price_stats(self, use_original=False): + if use_original and self.original_prices is not None: + prices = np.asarray(self.original_prices, dtype=np.float32) + elif self.external_prices is not None: + prices = np.asarray(self.external_prices, dtype=np.float32) + else: + prices = self._generated_prices_for_stats(24 * 7 * 52) + + return { + 'min': float(np.min(prices)), + 'max': float(np.max(prices)), + 'mean': float(np.mean(prices)), + 'median': float(np.median(prices)), + 'std': float(np.std(prices)), + f'{self.PERCENTILE_MIN}th_percentile': float(np.percentile(prices, self.PERCENTILE_MIN)), + f'{self.PERCENTILE_MAX}th_percentile': float(np.percentile(prices, self.PERCENTILE_MAX)), + 'price_shift': float(self.price_shift), + } + + def plot_price_histogram(self, num_bins=50, save_path=None, use_original=False): + if use_original and self.original_prices is not None: + prices = np.asarray(self.original_prices, dtype=np.float32) + price_type = "Original" + elif self.external_prices is not None: + prices = np.asarray(self.external_prices, dtype=np.float32) + price_type = "Shifted" if self.price_shift else "Original" + else: + prices = self._generated_prices_for_stats(24 * 7 * 52) + price_type = "Generated" + + plt.figure(figsize=(10, 6)) + plt.hist(prices, bins=num_bins, edgecolor='black') + plt.title(f'Distribution of Electricity Prices ({price_type})') + plt.xlabel('Price ($/MWh)') + plt.ylabel('Frequency') + plt.axvline(np.percentile(prices, self.PERCENTILE_MIN), color='r', linestyle='dashed', linewidth=2, label=f'{self.PERCENTILE_MIN}th Percentile') + plt.axvline(np.percentile(prices, self.PERCENTILE_MAX), color='g', linestyle='dashed', linewidth=2, label=f'{self.PERCENTILE_MAX}th Percentile') + plt.axvline(np.mean(prices), color='b', linestyle='dashed', linewidth=2, label='Mean Price') + plt.legend() + + if save_path: + plt.savefig(save_path) + else: + plt.show() + plt.close() diff --git a/sampler_hourly.py b/sampler_hourly.py index f32d4b0..99eee06 100644 --- a/sampler_hourly.py +++ b/sampler_hourly.py @@ -1,7 +1,6 @@ import re import datetime import math -import random from collections import defaultdict import numpy as np diff --git a/test_determinism_workloadgen.py b/test_determinism_workloadgen.py new file mode 100644 index 0000000..4e44c84 --- /dev/null +++ b/test_determinism_workloadgen.py @@ -0,0 +1,72 @@ +# determinism test for workload generator +import numpy as np +from workloadgen import WorkloadGenerator, WorkloadGenConfig + +def jobs_to_tuples(jobs): + return [(j.duration, j.nodes, j.cores_per_node) for j in jobs] + +def check_determinism(cfg, seed=12345, hours=50): + g1 = WorkloadGenerator(cfg) + g2 = WorkloadGenerator(cfg) + + r1 = np.random.default_rng(seed) + r2 = np.random.default_rng(seed) + + for h in range(hours): + a = jobs_to_tuples(g1.sample(h, r1)) + b = jobs_to_tuples(g2.sample(h, r2)) + assert a == b, f"Determinism failed at hour={h}: {a[:3]} vs {b[:3]}" + print("[OK] determinism") + +def check_bounds(cfg, seed=1, hours=50): + gen = WorkloadGenerator(cfg) + rng = np.random.default_rng(seed) + + for h in range(hours): + jobs = gen.sample(h, rng) + assert len(jobs) <= cfg.max_new_jobs_per_hour + if cfg.hard_cap_jobs is not None: + assert len(jobs) <= cfg.hard_cap_jobs + + for j in jobs: + assert cfg.min_duration <= j.duration <= cfg.max_duration + assert cfg.min_nodes <= j.nodes <= cfg.max_nodes + assert cfg.min_cores <= j.cores_per_node <= cfg.max_cores + + print("[OK] bounds & caps") + +def check_distribution_smoke(cfg, seed=7, hours=2000): + gen = WorkloadGenerator(cfg) + rng = np.random.default_rng(seed) + + counts = np.array([len(gen.sample(h, rng)) for h in range(hours)], dtype=np.int32) + mean = float(counts.mean()) + p0 = float((counts == 0).mean()) + + print(f"[INFO] arrivals={cfg.arrivals} mean={mean:.2f} p0={p0:.3f} max={counts.max()}") + + #if cfg.arrivals == "flat": + if cfg.arrivals == "uniform": + # Uniform integers 0..M => mean ~ M/2 (close-ish for long runs) + expected = cfg.max_new_jobs_per_hour / 2.0 + assert abs(mean - expected) / expected < 0.10, "flat mean looks off (smoke check)" + elif cfg.arrivals == "poisson": + # Only a loose sanity check: mean should not be wildly off lambda unless capped heavily + target = min(cfg.poisson_lambda, cfg.max_new_jobs_per_hour) + if target > 0: + assert abs(mean - target) / target < 0.20, "poisson mean looks off (smoke check)" + + print("[OK] distribution smoke") + +def main(): + # Example configs to test + flat_cfg = WorkloadGenConfig(arrivals="flat", max_new_jobs_per_hour=1000) + pois_cfg = WorkloadGenConfig(arrivals="poisson", poisson_lambda=200.0, max_new_jobs_per_hour=1000) + + for cfg in (flat_cfg, pois_cfg): + check_determinism(cfg) + check_bounds(cfg) + check_distribution_smoke(cfg) + +if __name__ == "__main__": + main() diff --git a/test_sanity_workloadgen.py b/test_sanity_workloadgen.py new file mode 100644 index 0000000..4861244 --- /dev/null +++ b/test_sanity_workloadgen.py @@ -0,0 +1,43 @@ +# sanity test for workload generator +import numpy as np +from workloadgen import WorkloadGenerator, WorkloadGenConfig + +def assert_job_valid(j, cfg): + assert cfg.min_duration <= j.duration <= cfg.max_duration + assert cfg.min_nodes <= j.nodes <= cfg.max_nodes + assert cfg.min_cores <= j.cores_per_node <= cfg.max_cores + +def test_determinism(): + cfg = WorkloadGenConfig(arrivals="poisson", poisson_lambda=100.0, max_new_jobs_per_hour=1500) + gen = WorkloadGenerator(cfg) + + rng1 = np.random.default_rng(123) + rng2 = np.random.default_rng(123) + + for hour in range(50): + a = gen.sample(hour, rng1) + b = gen.sample(hour, rng2) + assert [(j.duration, j.nodes, j.cores_per_node) for j in a] == [(j.duration, j.nodes, j.cores_per_node) for j in b] + +def test_constraints(): + cfg = WorkloadGenConfig(arrivals="flat", flat_jobs_per_hour=200, max_new_jobs_per_hour=200) + gen = WorkloadGenerator(cfg) + rng = np.random.default_rng(7) + for hour in range(200): + jobs = gen.sample(hour, rng) + for j in jobs: + assert_job_valid(j, cfg) + +def test_poisson_mean_sanity(): + cfg = WorkloadGenConfig(arrivals="poisson", poisson_lambda=50.0, max_new_jobs_per_hour=1500) + gen = WorkloadGenerator(cfg) + rng = np.random.default_rng(1) + counts = [len(gen.sample(h, rng)) for h in range(2000)] + mean = float(np.mean(counts)) + assert 45.0 < mean < 55.0, mean # loose band, just to catch obvious bugs + +if __name__ == "__main__": + test_determinism() + test_constraints() + test_poisson_mean_sanity() + print("[OK] workloadgen sanity checks passed") diff --git a/train.py b/train.py index 1e86761..3904f8d 100644 --- a/train.py +++ b/train.py @@ -37,6 +37,10 @@ def main(): parser.add_argument("--session", default="default", help="Session ID") parser.add_argument("--evaluate-savings", action='store_true', help="Load latest model and evaluate long-term savings (no training)") parser.add_argument("--eval-months", type=int, default=12, help="Months to evaluate for savings analysis (default: 12, only used with --evaluate-savings)") + parser.add_argument("--workload-gen", type=str, default="", choices=["", "flat", "poisson", "uniform"], help="Enable workload generator (default: disabled).",) + parser.add_argument("--wg-poisson-lambda", type=float, default=200.0, help="Poisson lambda for jobs/hour.") + parser.add_argument("--wg-max-jobs-hour", type=int, default=1500, help="Cap jobs/hour for generator.") + args = parser.parse_args() prices_file_path = args.prices @@ -74,15 +78,20 @@ def main(): if not os.path.exists(plots_dir): os.makedirs(plots_dir) + + # Train.py passes strings; the env treats "" as falsy in some places and truthy in others. + # To be safe: normalize "" -> None here. + def norm_path(x): + return None if (x is None or str(x).strip() == "") else x env = ComputeClusterEnv(weights=weights, session=args.session, render_mode=args.render, quick_plot=args.quick_plot, - external_prices=prices, - external_durations=job_durations_file_path, - external_jobs=jobs_file_path, - external_hourly_jobs=hourly_jobs_file_path, + external_prices=norm_path(prices), + external_durations=norm_path(job_durations_file_path), + external_jobs=norm_path(jobs_file_path), + external_hourly_jobs=norm_path(hourly_jobs_file_path), plot_rewards=args.plot_rewards, plots_dir=plots_dir, plot_once=args.plot_once, diff --git a/workloadgen.py b/workloadgen.py new file mode 100644 index 0000000..09bb065 --- /dev/null +++ b/workloadgen.py @@ -0,0 +1,110 @@ +# workloadgen.py +from __future__ import annotations + + +'''A deterministic, configurable workload generator that can produce realistic and pathological +job streams (arrivals + job shapes), without relying on historic scheduler logs.''' + + +'''Requirements: +Hard: +- Deterministic under env.reset(seed=...): same seed + same config => identical job stream. +- Controllable: one can dial job rate, duration mix, node/cores mix, correlation strength, and “stress modes”. +- Composable: easy to plug in multiple “components” (baseline traffic + bursts + maintenance window, etc.). +- Future-proof for "wrong time estimates": job specs must be easy to extend with estimated_duration (and later extra fields). + +Soft (nice to have): +- Realistic correlations: e.g. longer jobs tend to request more nodes, daily arrival patterns, etc. +- Replaying canned “scenarios” (regression tests) with fixed seeds. +''' + + +from dataclasses import dataclass, replace +from typing import List, Optional +import numpy as np + + +@dataclass(frozen=True) +class JobSpec: + duration: int + nodes: int + cores_per_node: int + + +@dataclass(frozen=True) +class WorkloadGenConfig: + # arrivals: "flat" or "poisson" + arrivals: str = "poisson" + max_new_jobs_per_hour: int = 1500 + poisson_lambda: float = 200.0 + flat_jobs_per_hour: int = 200 # target arrivals for flat mode + flat_jitter: int = 0 # +/- jitter; 0 => perfectly flat + + + # resource ranges (v1: just uniform ranges; later we add mixtures/correlations) + min_duration: int = 1 + max_duration: int = 170 + min_nodes: int = 1 + max_nodes: int = 16 + min_cores: int = 1 + max_cores: int = 96 + + # optional hard cap safety (useful if someone sets poisson_lambda insane) + hard_cap_jobs: Optional[int] = None + + +class WorkloadGenerator: + def __init__(self, cfg: WorkloadGenConfig): + arrivals = cfg.arrivals.lower().strip() + if arrivals not in ("flat", "poisson", "uniform"): + raise ValueError(f"arrivals must be 'flat', 'uniform' or 'poisson', got: {cfg.arrivals}") + self.cfg = replace(cfg, arrivals=arrivals) + + def _sample_job_count(self, rng: np.random.Generator) -> int: + """ + Arrival modes: + - flat: constant arrivals around a target, optional +/- jitter (0 => perfectly constant) + - poisson: Poisson(lambda) + - uniform: discrete-uniform in [0, max_new_jobs_per_hour] (very noisy hour-to-hour) + """ + mode = self.cfg.arrivals + + if mode == "flat": + target = int(getattr(self.cfg, "flat_jobs_per_hour", self.cfg.max_new_jobs_per_hour)) + jitter = int(getattr(self.cfg, "flat_jitter", 0)) + + if jitter <= 0: + k = target + else: + k = int(rng.integers(target - jitter, target + jitter + 1)) + + elif mode == "poisson": + k = int(rng.poisson(self.cfg.poisson_lambda)) + + elif mode == "uniform": + # This is the old "flat". + k = int(rng.integers(0, self.cfg.max_new_jobs_per_hour + 1)) + + else: + raise ValueError(f"Unknown arrivals mode: {mode}") + + # clamp + safety + k = min(k, int(self.cfg.max_new_jobs_per_hour)) + if self.cfg.hard_cap_jobs is not None: + k = min(k, int(self.cfg.hard_cap_jobs)) + if k < 0: + k = 0 + return k + + def sample(self, hour_idx: int, rng: np.random.Generator) -> List[JobSpec]: + # hour_idx currently unused, but we keep it to enable daily patterns later. + n = self._sample_job_count(rng) + + if n == 0: + return [] + + durations = rng.integers(self.cfg.min_duration, self.cfg.max_duration + 1, size=n, dtype=np.int32) + nodes = rng.integers(self.cfg.min_nodes, self.cfg.max_nodes + 1, size=n, dtype=np.int32) + cores = rng.integers(self.cfg.min_cores, self.cfg.max_cores + 1, size=n, dtype=np.int32) + + return [JobSpec(int(durations[i]), int(nodes[i]), int(cores[i])) for i in range(n)] From 4c9e465d99174d5693c783dac5ded12e4c5d75f3 Mon Sep 17 00:00:00 2001 From: enlorenz Date: Mon, 1 Dec 2025 17:52:29 +0100 Subject: [PATCH 04/11] Workload generator: Added new script, including sanity checks. Generates timely distributions of incoming jobs. Direct input into agent not yet finished, but pipelines mostly set. train.py has added parsers to call generator. Fixed: Missing deque import, correct "flat" distribution check, removed duplicate lines. ist Commit-Beschreibung #2: Workload generator: Added new script, including sanity checks. Generates timely distributions of incoming jobs. Direct input into agent not yet finished, but pipelines mostly set. train.py has added parsers to call generator. Fixed: Missing deque import, correct "flat" distribution check, removed duplicate lines. Additional Fixes: Remove unused import, add queue capacity also for job randomizer, norm_path check now also for prices. Removed non-essential type casts. Hotfix Triplet, which occured during rebase. --- environment.py | 104 +++++++++++++++++++++++--------- prices_deterministic.py | 3 +- test_determinism_workloadgen.py | 14 +++-- test_sanity_workloadgen.py | 2 +- train.py | 44 +++++++++++--- 5 files changed, 123 insertions(+), 44 deletions(-) diff --git a/environment.py b/environment.py index 519938a..8552e25 100644 --- a/environment.py +++ b/environment.py @@ -10,7 +10,6 @@ from weights import Weights from plot import plot, plot_reward from sampler_duration import durations_sampler -from sampler_jobs import jobs_sampler from sampler_jobs import DurationSampler from sampler_hourly import hourly_sampler @@ -115,7 +114,8 @@ def __init__(self, skip_plot_used_nodes, skip_plot_job_queue, steps_per_iteration, - evaluation_mode=False): + evaluation_mode=False, + workload_gen=None): super().__init__() self.weights = weights @@ -151,19 +151,22 @@ def __init__(self, #Initialize deterministic RNG, instead of global RNG self.np_random = None self._seed = None + self.workload_gen = workload_gen + if self.external_durations: durations_sampler.init(self.external_durations) - if self.external_jobs: + if self.external_jobs and not self.workload_gen: + self.jobs_sampler = DurationSampler() print(f"Loading jobs from {self.external_jobs}") - jobs_sampler.parse_jobs(self.external_jobs, 60) - print(f"Parsed jobs for {len(jobs_sampler.jobs)} hours") - print(f"Parsed aggregated jobs for {len(jobs_sampler.aggregated_jobs)} hours") - jobs_sampler.precalculate_hourly_jobs(CORES_PER_NODE, MAX_NODES_PER_JOB) - print(f"Max jobs per hour: {jobs_sampler.max_new_jobs_per_hour}") - print(f"Max job duration: {jobs_sampler.max_job_duration}") - print(f"Parsed hourly jobs for {len(jobs_sampler.hourly_jobs)} hours") + self.jobs_sampler.parse_jobs(self.external_jobs, 60) + print(f"Parsed jobs for {len(self.jobs_sampler.jobs)} hours") + print(f"Parsed aggregated jobs for {len(self.jobs_sampler.aggregated_jobs)} hours") + self.jobs_sampler.precalculate_hourly_jobs(CORES_PER_NODE, MAX_NODES_PER_JOB) + print(f"Max jobs per hour: {self.jobs_sampler.max_new_jobs_per_hour}") + print(f"Max job duration: {self.jobs_sampler.max_job_duration}") + print(f"Parsed hourly jobs for {len(self.jobs_sampler.hourly_jobs)} hours") if self.external_hourly_jobs: print(f"Loading hourly jobs from {self.external_hourly_jobs}") @@ -321,6 +324,7 @@ def reset_state(self): self.dropped_this_episode = 0 self.baseline_dropped_this_episode = 0 self.prev_excess_dropped = 0 + def step(self, action): self.current_step += 1 @@ -348,7 +352,6 @@ def step(self, action): new_jobs_count = 0 if self.external_jobs: - # jobs = jobs_sampler.sample_one_hourly(wrap=True)['hourly_jobs'] jobs = self.jobs_sampler.sample_one_hourly(wrap=True)["hourly_jobs"] if len(jobs) > 0: for job in jobs: @@ -358,7 +361,19 @@ def step(self, action): new_jobs_cores.append(job['cores_per_node']) elif self.external_hourly_jobs: hour_of_day = (self.current_hour - 1) % 24 - jobs = hourly_sampler.sample(hour_of_day) + + # How much room is left right now? + queue_used = int(np.count_nonzero(job_queue_2d[:, 0] > 0)) + queue_free = max(0, MAX_QUEUE_SIZE - queue_used) + + # Hard cap: do not generate more than we *could* enqueue anyway + max_to_generate = min(queue_free, MAX_NEW_JOBS_PER_HOUR) + + if max_to_generate == 0: + jobs = [] + else: + jobs = hourly_sampler.sample(hour_of_day, rng=self.np_random, max_jobs=max_to_generate) + if len(jobs) > 0: for job in jobs: new_jobs_count += 1 @@ -366,19 +381,52 @@ def step(self, action): new_jobs_nodes.append(job['nodes']) new_jobs_cores.append(job['cores_per_node']) else: - new_jobs_count = np.random.randint(0, MAX_NEW_JOBS_PER_HOUR + 1) - if self.external_durations: - new_jobs_durations = durations_sampler.sample(new_jobs_count) +#----------------------Use Workload Generator for Randomizer------------------------------------------------------------------------------------ + if self.workload_gen is not None: + # How much room is left right now? + queue_used = int(np.count_nonzero(job_queue_2d[:, 0] > 0)) + queue_free = max(0, MAX_QUEUE_SIZE - queue_used) + + # Hard cap: do not generate more than we *could* enqueue anyway + max_to_generate = min(queue_free, MAX_NEW_JOBS_PER_HOUR) + + if max_to_generate == 0: + jobs = [] + else: + jobs = self.workload_gen.sample(self.current_hour - 1, self.np_random) + # In case the generator wants to produce more than we can enqueue: + if len(jobs) > max_to_generate: + jobs = jobs[:max_to_generate] + new_jobs_count = len(jobs) + if new_jobs_count > 0: + for j in jobs: + new_jobs_durations.append(j.duration) + new_jobs_nodes.append(j.nodes) + new_jobs_cores.append(j.cores_per_node) +#----------------------Legacy Randomizer-------------------------------------------------------------------------------------------------------- else: - new_jobs_durations = np.random.randint(1, MAX_JOB_DURATION + 1, size=new_jobs_count) - # Generate random node and core requirements - for _ in range(new_jobs_count): - new_jobs_nodes.append(np.random.randint(MIN_NODES_PER_JOB, MAX_NODES_PER_JOB + 1)) - new_jobs_cores.append(np.random.randint(MIN_CORES_PER_JOB, CORES_PER_NODE + 1)) + # new_jobs_count = np.random.randint(0, MAX_NEW_JOBS_PER_HOUR + 1) # Keep legacy code for now + new_jobs_count = self.np_random.integers(0, MAX_NEW_JOBS_PER_HOUR + 1) # Introduce new, non-global RNG + if self.external_durations: + new_jobs_durations = durations_sampler.sample(new_jobs_count) + else: + # new_jobs_durations = np.random.randint(1, MAX_JOB_DURATION + 1, size=new_jobs_count) # Keep legacy code for now + new_jobs_durations = self.np_random.integers(1, MAX_JOB_DURATION + 1, size=new_jobs_count) # Introduce new, non-global RNG + # Generate random node and core requirements + for _ in range(new_jobs_count): + # new_jobs_nodes.append(np.random.randint(MIN_NODES_PER_JOB, MAX_NODES_PER_JOB + 1)) # Keep legacy code for now + # new_jobs_cores.append(np.random.randint(MIN_CORES_PER_JOB, CORES_PER_NODE + 1)) # Keep legacy code for now + new_jobs_nodes.append(self.np_random.integers(MIN_NODES_PER_JOB, MAX_NODES_PER_JOB + 1)) # Introduce new, non-global RNG + new_jobs_cores.append(self.np_random.integers(MIN_CORES_PER_JOB, CORES_PER_NODE + 1)) # Introduce new, non-global RNG +#---------------------------------------------------------------------------------------------------------------------------------------------- + + self.env_print(f"[2] Adding {new_jobs_count} new jobs to the queue...") new_jobs, self.next_empty_slot = self.add_new_jobs(job_queue_2d, new_jobs_count, new_jobs_durations, new_jobs_nodes, new_jobs_cores, self.next_empty_slot) self.jobs_submitted += len(new_jobs) + self.jobs_rejected_queue_full += (new_jobs_count - len(new_jobs)) + self.env_print("nodes: ", np.array2string(self.state['nodes'], separator=' ', max_line_width=np.inf)) self.env_print(f"cores_available: {np.array2string(self.cores_available, separator=' ', max_line_width=np.inf)} ({np.sum(self.cores_available)})") @@ -589,13 +637,13 @@ def assign_jobs_to_available_nodes(self, job_queue_2d, nodes, cores_available, r # Assign job to first job_nodes candidates job_allocation = [] for i in range(job_nodes): - node_idx = int(candidate_nodes[i]) - cores_available[node_idx] -= int(job_cores_per_node) - nodes[node_idx] = max(int(nodes[node_idx]), int(job_duration)) - job_allocation.append((node_idx, int(job_cores_per_node))) + node_idx = candidate_nodes[i] + cores_available[node_idx] -= job_cores_per_node + nodes[node_idx] = max(nodes[node_idx], job_duration) + job_allocation.append((node_idx, job_cores_per_node)) running_jobs[self.next_job_id] = { - "duration": int(job_duration), + "duration": job_duration, "allocation": job_allocation, } self.next_job_id += 1 @@ -610,16 +658,16 @@ def assign_jobs_to_available_nodes(self, job_queue_2d, nodes, cores_available, r # Track job completion and wait time if is_baseline: self.baseline_jobs_completed += 1 - self.baseline_total_job_wait_time += int(job_age) + self.baseline_total_job_wait_time += job_age else: self.jobs_completed += 1 - self.total_job_wait_time += int(job_age) + self.total_job_wait_time += job_age num_processed_jobs += 1 continue # Not enough resources -> job waits and ages (or gets dropped) - new_age = int(job_age) + 1 + new_age = job_age + 1 if new_age > MAX_JOB_AGE: # Clear job from queue diff --git a/prices_deterministic.py b/prices_deterministic.py index 6d21c9e..f08636f 100644 --- a/prices_deterministic.py +++ b/prices_deterministic.py @@ -1,5 +1,6 @@ import numpy as np import matplotlib.pyplot as plt +from collections import deque class Prices: ELECTRICITY_PRICE_BASE = 20 @@ -75,8 +76,6 @@ def get_next_price(self): self.price_index += 1 self.price_history.append(new_price) - #if len(self.price_history) > self.HISTORY_WINDOW: - # self.price_history.pop(0) # deque automatically removes oldest when maxlen exceeded | Keep the old line for now. Remove during next PR. return new_price diff --git a/test_determinism_workloadgen.py b/test_determinism_workloadgen.py index 4e44c84..36fc234 100644 --- a/test_determinism_workloadgen.py +++ b/test_determinism_workloadgen.py @@ -45,11 +45,14 @@ def check_distribution_smoke(cfg, seed=7, hours=2000): print(f"[INFO] arrivals={cfg.arrivals} mean={mean:.2f} p0={p0:.3f} max={counts.max()}") - #if cfg.arrivals == "flat": - if cfg.arrivals == "uniform": - # Uniform integers 0..M => mean ~ M/2 (close-ish for long runs) - expected = cfg.max_new_jobs_per_hour / 2.0 + if cfg.arrivals == "flat": + # Flat mode: mean should be close to flat_jobs_per_hour (with jitter variation) + expected = cfg.flat_jobs_per_hour assert abs(mean - expected) / expected < 0.10, "flat mean looks off (smoke check)" + elif cfg.arrivals == "uniform": + # Uniform integers 0..M => mean ~ M/2 + expected = cfg.max_new_jobs_per_hour / 2.0 + assert abs(mean - expected) / expected < 0.10, "uniform mean looks off (smoke check)" elif cfg.arrivals == "poisson": # Only a loose sanity check: mean should not be wildly off lambda unless capped heavily target = min(cfg.poisson_lambda, cfg.max_new_jobs_per_hour) @@ -62,8 +65,9 @@ def main(): # Example configs to test flat_cfg = WorkloadGenConfig(arrivals="flat", max_new_jobs_per_hour=1000) pois_cfg = WorkloadGenConfig(arrivals="poisson", poisson_lambda=200.0, max_new_jobs_per_hour=1000) + unif_cfg = WorkloadGenConfig(arrivals="uniform", max_new_jobs_per_hour=1000) - for cfg in (flat_cfg, pois_cfg): + for cfg in (flat_cfg, pois_cfg, unif_cfg): check_determinism(cfg) check_bounds(cfg) check_distribution_smoke(cfg) diff --git a/test_sanity_workloadgen.py b/test_sanity_workloadgen.py index 4861244..3f658d4 100644 --- a/test_sanity_workloadgen.py +++ b/test_sanity_workloadgen.py @@ -34,7 +34,7 @@ def test_poisson_mean_sanity(): rng = np.random.default_rng(1) counts = [len(gen.sample(h, rng)) for h in range(2000)] mean = float(np.mean(counts)) - assert 45.0 < mean < 55.0, mean # loose band, just to catch obvious bugs + assert 48.0 < mean < 52.0, mean # tighter band, still reliable for 2000 samples if __name__ == "__main__": test_determinism() diff --git a/train.py b/train.py index 3904f8d..fd435f7 100644 --- a/train.py +++ b/train.py @@ -7,6 +7,21 @@ import glob import argparse import pandas as pd +from workloadgen import WorkloadGenerator, WorkloadGenConfig + +# Import environment variables: +from environment import ( + MAX_JOB_DURATION, + MIN_NODES_PER_JOB, MAX_NODES_PER_JOB, + MIN_CORES_PER_JOB, + CORES_PER_NODE, +) + + +# Train.py passes strings; the env treats "" as falsy in some places and truthy in others. +# To be safe: normalize "" -> None here. +def norm_path(x): + return None if (x is None or str(x).strip() == "") else x STEPS_PER_ITERATION = 100000 @@ -48,7 +63,7 @@ def main(): jobs_file_path = args.jobs hourly_jobs_file_path = args.hourly_jobs - if prices_file_path: + if norm_path(prices_file_path): df = pd.read_csv(prices_file_path, parse_dates=['Date']) prices = df['Price'].values.tolist() print(f"Loaded {len(prices)} prices from CSV: {prices_file_path}") @@ -78,17 +93,29 @@ def main(): if not os.path.exists(plots_dir): os.makedirs(plots_dir) - - # Train.py passes strings; the env treats "" as falsy in some places and truthy in others. - # To be safe: normalize "" -> None here. - def norm_path(x): - return None if (x is None or str(x).strip() == "") else x + + # Load Workload Generator: + + workload_gen = None + if args.workload_gen: + cfg = WorkloadGenConfig( + arrivals=args.workload_gen, + poisson_lambda=args.wg_poisson_lambda, + max_new_jobs_per_hour=args.wg_max_jobs_hour, + min_duration=1, + max_duration=MAX_JOB_DURATION, + min_nodes=MIN_NODES_PER_JOB, + max_nodes=MAX_NODES_PER_JOB, + min_cores=MIN_CORES_PER_JOB, + max_cores=CORES_PER_NODE, + ) + workload_gen = WorkloadGenerator(cfg) env = ComputeClusterEnv(weights=weights, session=args.session, render_mode=args.render, quick_plot=args.quick_plot, - external_prices=norm_path(prices), + external_prices=prices, external_durations=norm_path(job_durations_file_path), external_jobs=norm_path(jobs_file_path), external_hourly_jobs=norm_path(hourly_jobs_file_path), @@ -104,7 +131,8 @@ def norm_path(x): skip_plot_used_nodes=args.skip_plot_used_nodes, skip_plot_job_queue=args.skip_plot_job_queue, steps_per_iteration=STEPS_PER_ITERATION, - evaluation_mode=args.evaluate_savings) + evaluation_mode=args.evaluate_savings, + workload_gen=workload_gen) env.reset() # Check if there are any saved models in models_dir From dd75be68445b273f2bcaaf7e36cf6406e2ab4393 Mon Sep 17 00:00:00 2001 From: Alexey Rybalchenko Date: Thu, 18 Dec 2025 09:37:22 +0100 Subject: [PATCH 05/11] Add weight for the drop penalty --- environment.py | 9 +++++---- train.py | 4 +++- train_iter.py | 51 +++++++++++++++++++++++++++++++++----------------- weights.py | 5 +++-- 4 files changed, 45 insertions(+), 24 deletions(-) diff --git a/environment.py b/environment.py index 8552e25..41a204a 100644 --- a/environment.py +++ b/environment.py @@ -749,9 +749,10 @@ def calculate_reward(self, num_used_nodes, num_idle_nodes, current_price, averag self.price_rewards.append(price_reward_norm * 100) self.job_age_penalties.append(job_age_penalty_norm * 100) self.idle_penalties.append(idle_penalty_norm * 100) - - # Penalty is always non-positive + + # 6. penalty for dropped jobs (WIP - unnormalized, weighted) drop_penalty = min(0, PENALTY_DROPPED_JOB * self.new_excess) + drop_penalty_weighted = self.weights.drop_weight * drop_penalty reward = ( efficiency_reward_weighted @@ -759,10 +760,10 @@ def calculate_reward(self, num_used_nodes, num_idle_nodes, current_price, averag + price_reward_weighted + job_age_penalty_weighted + idle_penalty_weighted - + drop_penalty + + drop_penalty_weighted ) - self.env_print(f" > $$$TOTAL: {reward:.4f} = {efficiency_reward_weighted:.4f} + {price_reward_weighted:.4f} + {idle_penalty_weighted:.4f} + {job_age_penalty_weighted:.4f}") + self.env_print(f" > $$$TOTAL: {reward:.4f} = {efficiency_reward_weighted:.4f} + {price_reward_weighted:.4f} + {idle_penalty_weighted:.4f} + {job_age_penalty_weighted:.4f} + {drop_penalty_weighted:.4f}") self.env_print(f" > step cost: €{total_cost:.4f}") return reward, total_cost diff --git a/train.py b/train.py index fd435f7..bf77c8f 100644 --- a/train.py +++ b/train.py @@ -48,6 +48,7 @@ def main(): parser.add_argument("--price-weight", type=float, default=0.2, help="Weight for price reward") parser.add_argument("--idle-weight", type=float, default=0.1, help="Weight for idle penalty") parser.add_argument("--job-age-weight", type=float, default=0.0, help="Weight for job age penalty") + parser.add_argument("--drop-weight", type=float, default=0.0, help="Weight for dropped jobs penalty (WIP - default 0.0)") parser.add_argument("--iter-limit", type=int, default=0, help=f"Max number of training iterations (1 iteration = {STEPS_PER_ITERATION} steps)") parser.add_argument("--session", default="default", help="Session ID") parser.add_argument("--evaluate-savings", action='store_true', help="Load latest model and evaluate long-term savings (no training)") @@ -76,7 +77,8 @@ def main(): efficiency_weight=args.efficiency_weight, price_weight=args.price_weight, idle_weight=args.idle_weight, - job_age_weight=args.job_age_weight + job_age_weight=args.job_age_weight, + drop_weight=args.drop_weight ) weights_prefix = f"e{weights.efficiency_weight}_p{weights.price_weight}_i{weights.idle_weight}_d{weights.job_age_weight}" diff --git a/train_iter.py b/train_iter.py index 74a0eaa..3f205c8 100644 --- a/train_iter.py +++ b/train_iter.py @@ -8,7 +8,7 @@ def generate_weight_combinations(step=0.1, fixed_weights=None): weights = np.linspace(0, 1, num=int(1/step) + 1, endpoint=True) combinations = [] - weight_names = ['efficiency', 'price', 'idle', 'job-age'] + weight_names = ['efficiency', 'price', 'idle', 'job-age', 'drop'] if fixed_weights: # Get the names of weights that aren't fixed @@ -19,7 +19,7 @@ def generate_weight_combinations(step=0.1, fixed_weights=None): # If all but one weight is fixed, there's only one possible value remaining = round(1 - fixed_sum, 2) if 0 <= remaining <= 1: - combo = [0, 0, 0, 0] # Initialize with four zeros + combo = [0, 0, 0, 0, 0] # Initialize with five zeros # Set fixed weights for weight_name, value in fixed_weights.items(): combo[weight_names.index(weight_name)] = value @@ -28,11 +28,11 @@ def generate_weight_combinations(step=0.1, fixed_weights=None): combinations.append(tuple(combo)) elif len(variable_weights) == 2: - # If two weights are fixed, vary the other two + # If three weights are fixed, vary the other two for w in weights: remaining = round(1 - fixed_sum - w, 2) if 0 <= remaining <= 1: - combo = [0, 0, 0, 0] # Initialize with four zeros + combo = [0, 0, 0, 0, 0] # Initialize with five zeros # Set fixed weights for weight_name, value in fixed_weights.items(): combo[weight_names.index(weight_name)] = value @@ -42,11 +42,11 @@ def generate_weight_combinations(step=0.1, fixed_weights=None): combinations.append(tuple(combo)) elif len(variable_weights) == 3: - # If one weight is fixed, vary the other three + # If two weights are fixed, vary the other three for w1, w2 in itertools.product(weights, repeat=2): remaining = round(1 - fixed_sum - w1 - w2, 2) if 0 <= remaining <= 1: - combo = [0, 0, 0, 0] # Initialize with four zeros + combo = [0, 0, 0, 0, 0] # Initialize with five zeros # Set fixed weights for weight_name, value in fixed_weights.items(): combo[weight_names.index(weight_name)] = value @@ -56,16 +56,32 @@ def generate_weight_combinations(step=0.1, fixed_weights=None): combo[weight_names.index(variable_weights[2])] = remaining combinations.append(tuple(combo)) + elif len(variable_weights) == 4: + # If one weight is fixed, vary the other four + for w1, w2, w3 in itertools.product(weights, repeat=3): + remaining = round(1 - fixed_sum - w1 - w2 - w3, 2) + if 0 <= remaining <= 1: + combo = [0, 0, 0, 0, 0] # Initialize with five zeros + # Set fixed weights + for weight_name, value in fixed_weights.items(): + combo[weight_names.index(weight_name)] = value + # Set variable weights + combo[weight_names.index(variable_weights[0])] = round(w1, 2) + combo[weight_names.index(variable_weights[1])] = round(w2, 2) + combo[weight_names.index(variable_weights[2])] = round(w3, 2) + combo[weight_names.index(variable_weights[3])] = remaining + combinations.append(tuple(combo)) + else: # If no weight is fixed, generate all combinations - for e, p, i in itertools.product(weights, repeat=3): - ja = round(1 - e - p - i, 2) # job-age weight - if 0 <= ja <= 1: - combinations.append((round(e, 2), round(p, 2), round(i, 2), round(ja, 2))) + for e, p, i, ja in itertools.product(weights, repeat=4): + d = round(1 - e - p - i - ja, 2) # drop weight + if 0 <= d <= 1: + combinations.append((round(e, 2), round(p, 2), round(i, 2), round(ja, 2), round(d, 2))) return combinations -def run(efficiency_weight, price_weight, idle_weight, job_age_weight, iter_limit_per_step, session, prices, job_durations, jobs, hourly_jobs): +def run(efficiency_weight, price_weight, idle_weight, job_age_weight, drop_weight, iter_limit_per_step, session, prices, job_durations, jobs, hourly_jobs): python_executable = sys.executable command = [ python_executable, "train.py", @@ -73,6 +89,7 @@ def run(efficiency_weight, price_weight, idle_weight, job_age_weight, iter_limit "--price-weight", f"{price_weight:.2f}", "--idle-weight", f"{idle_weight:.2f}", "--job-age-weight", f"{job_age_weight:.2f}", + "--drop-weight", f"{drop_weight:.2f}", "--iter-limit", f"{iter_limit_per_step}", "--prices", f"{prices}", "--job-durations", f"{job_durations}", @@ -112,7 +129,7 @@ def main(): parser.add_argument('--job-durations', type=str, nargs='?', const="", default="", help='Path to a file containing job duration samples (for use with duration_sampler)') parser.add_argument('--jobs', type=str, nargs='?', const="", default="", help='Path to a file containing jobs samples (for use with jobs_sampler)') parser.add_argument('--hourly-jobs', type=str, nargs='?', const="", default="", help='Path to Slurm log file for hourly statistical sampling (for use with hourly_sampler)') - parser.add_argument("--fix-weights", type=str, help="Comma-separated list of weights to fix (efficiency,price,idle,job-age)") + parser.add_argument("--fix-weights", type=str, help="Comma-separated list of weights to fix (efficiency,price,idle,job-age,drop)") parser.add_argument("--fix-values", type=str, help="Comma-separated list of values for fixed weights") parser.add_argument("--iter-limit-per-step", type=int, help="Max number of training iterations per step (1 iteration = {TIMESTEPS} steps)") parser.add_argument("--session", help="Session ID") @@ -132,13 +149,13 @@ def main(): print(f"Execution preview:") for combo in combinations: - efficiency_weight, price_weight, idle_weight, job_age_weight = combo - print(f" efficiency={efficiency_weight}, price={price_weight}, idle={idle_weight}, job_age={job_age_weight}") + efficiency_weight, price_weight, idle_weight, job_age_weight, drop_weight = combo + print(f" efficiency={efficiency_weight}, price={price_weight}, idle={idle_weight}, job_age={job_age_weight}, drop={drop_weight}") for combo in combinations: - efficiency_weight, price_weight, idle_weight, job_age_weight = combo - print(f"Running with weights: efficiency={efficiency_weight}, price={price_weight}, idle={idle_weight}, job_age={job_age_weight}") - run(efficiency_weight, price_weight, idle_weight, job_age_weight, args.iter_limit_per_step, args.session, args.prices, args.job_durations, args.jobs, args.hourly_jobs) + efficiency_weight, price_weight, idle_weight, job_age_weight, drop_weight = combo + print(f"Running with weights: efficiency={efficiency_weight}, price={price_weight}, idle={idle_weight}, job_age={job_age_weight}, drop={drop_weight}") + run(efficiency_weight, price_weight, idle_weight, job_age_weight, drop_weight, args.iter_limit_per_step, args.session, args.prices, args.job_durations, args.jobs, args.hourly_jobs) if __name__ == "__main__": main() diff --git a/weights.py b/weights.py index 90ded56..e7f414e 100644 --- a/weights.py +++ b/weights.py @@ -7,9 +7,10 @@ class Weights: price_weight: float idle_weight: float job_age_weight: float + drop_weight: float def __str__(self): - return f"Weights: efficiency={self.efficiency_weight}, price={self.price_weight}, idle={self.idle_weight}, job_age={self.job_age_weight}. sum={self.sum():.2f}" + return f"Weights: efficiency={self.efficiency_weight}, price={self.price_weight}, idle={self.idle_weight}, job_age={self.job_age_weight}, drop={self.drop_weight}. sum={self.sum():.2f}" def sum(self): - return round(np.sum([self.efficiency_weight, self.price_weight, self.idle_weight, self.job_age_weight]), 2) + return round(np.sum([self.efficiency_weight, self.price_weight, self.idle_weight, self.job_age_weight, self.drop_weight]), 2) From 15aac5d25af34af160d05890f777ab29c34d6bae Mon Sep 17 00:00:00 2001 From: Alexey Rybalchenko Date: Thu, 18 Dec 2025 10:01:34 +0100 Subject: [PATCH 06/11] Exclude baseline from reward components --- environment.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/environment.py b/environment.py index 41a204a..2b43435 100644 --- a/environment.py +++ b/environment.py @@ -441,7 +441,7 @@ def step(self, action): # assign jobs to available nodes self.env_print(f"[4] Assigning jobs to available nodes...") - num_launched_jobs, self.next_empty_slot = self.assign_jobs_to_available_nodes(job_queue_2d, self.state['nodes'], self.cores_available, self.running_jobs, self.next_empty_slot) + num_launched_jobs, self.next_empty_slot, num_dropped_this_step = self.assign_jobs_to_available_nodes(job_queue_2d, self.state['nodes'], self.cores_available, self.running_jobs, self.next_empty_slot) self.env_print(f" {num_launched_jobs} jobs launched") # Calculate node utilization stats @@ -468,9 +468,7 @@ def step(self, action): baseline_cost, baseline_cost_off = self.baseline_step(current_price, new_jobs_count, new_jobs_durations, new_jobs_nodes, new_jobs_cores) self.baseline_cost += baseline_cost self.baseline_cost_off += baseline_cost_off - excess = max(0, self.dropped_this_episode - self.baseline_dropped_this_episode) - self.new_excess = max(0, excess - self.prev_excess_dropped) - self.prev_excess_dropped = excess + self.num_dropped_this_step = num_dropped_this_step # calculate reward step_reward, step_cost = self.calculate_reward(num_used_nodes, num_idle_nodes, current_price, average_future_price, num_off_nodes, num_launched_jobs, num_node_changes, job_queue_2d, num_unprocessed_jobs) @@ -687,7 +685,7 @@ def assign_jobs_to_available_nodes(self, job_queue_2d, nodes, cores_available, r else: job_queue_2d[job_idx][1] = new_age - return num_processed_jobs, next_empty_slot + return num_processed_jobs, next_empty_slot, num_dropped def baseline_step(self, current_price, new_jobs_count, new_jobs_durations, new_jobs_nodes, new_jobs_cores): job_queue_2d = self.baseline_state['job_queue'].reshape(-1, 4) @@ -698,7 +696,7 @@ def baseline_step(self, current_price, new_jobs_count, new_jobs_durations, new_j self.baseline_jobs_submitted += len(new_baseline_jobs) self.baseline_jobs_rejected_queue_full += (new_jobs_count - len(new_baseline_jobs)) - _, self.baseline_next_empty_slot = self.assign_jobs_to_available_nodes(job_queue_2d, self.baseline_state['nodes'], self.baseline_cores_available, self.baseline_running_jobs, self.baseline_next_empty_slot, is_baseline=True) + _, self.baseline_next_empty_slot, _ = self.assign_jobs_to_available_nodes(job_queue_2d, self.baseline_state['nodes'], self.baseline_cores_available, self.baseline_running_jobs, self.baseline_next_empty_slot, is_baseline=True) num_used_nodes = np.sum(self.baseline_state['nodes'] > 0) num_on_nodes = np.sum(self.baseline_state['nodes'] > -1) @@ -751,7 +749,7 @@ def calculate_reward(self, num_used_nodes, num_idle_nodes, current_price, averag self.idle_penalties.append(idle_penalty_norm * 100) # 6. penalty for dropped jobs (WIP - unnormalized, weighted) - drop_penalty = min(0, PENALTY_DROPPED_JOB * self.new_excess) + drop_penalty = min(0, PENALTY_DROPPED_JOB * self.num_dropped_this_step) drop_penalty_weighted = self.weights.drop_weight * drop_penalty reward = ( From 64dbbf546de9bdc7c0f435965e16dd91a2245771 Mon Sep 17 00:00:00 2001 From: Alexey Rybalchenko Date: Thu, 18 Dec 2025 10:02:55 +0100 Subject: [PATCH 07/11] formatting --- environment.py | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/environment.py b/environment.py index 2b43435..1e7121d 100644 --- a/environment.py +++ b/environment.py @@ -13,7 +13,7 @@ from sampler_jobs import DurationSampler from sampler_hourly import hourly_sampler -# For deterministic RNG +# For deterministic RNG from gymnasium.utils import seeding init() # Initialize colorama @@ -81,7 +81,7 @@ def env_print(self, *args): """Prints only if the render mode is 'human'.""" if self.render_mode == 'human': print(*args) - + # Validator for debugging: def _validate_next_empty(self,job_queue_2d, next_empty): n = len(job_queue_2d) @@ -114,7 +114,7 @@ def __init__(self, skip_plot_used_nodes, skip_plot_job_queue, steps_per_iteration, - evaluation_mode=False, + evaluation_mode=False, workload_gen=None): super().__init__() @@ -147,7 +147,7 @@ def __init__(self, self.episode_costs = [] self.prices = Prices(self.external_prices) - + #Initialize deterministic RNG, instead of global RNG self.np_random = None self._seed = None @@ -246,8 +246,8 @@ def __init__(self, def reset(self, seed = None, options = None): super().reset(seed=seed) - self.np_random, self._seed = seeding.np_random(seed) - + self.np_random, self._seed = seeding.np_random(seed) + self.reset_state() self.prices.reset() @@ -314,17 +314,17 @@ def reset_state(self): # Job tracking metrics for baseline self.baseline_jobs_dropped = 0 self.baseline_dropped_this_episode = 0 - + # Agent self.jobs_rejected_queue_full = 0 # new jobs we couldn't even enqueue # Baseline self.baseline_jobs_rejected_queue_full = 0 - + self.dropped_this_episode = 0 self.baseline_dropped_this_episode = 0 self.prev_excess_dropped = 0 - + def step(self, action): self.current_step += 1 @@ -361,7 +361,7 @@ def step(self, action): new_jobs_cores.append(job['cores_per_node']) elif self.external_hourly_jobs: hour_of_day = (self.current_hour - 1) % 24 - + # How much room is left right now? queue_used = int(np.count_nonzero(job_queue_2d[:, 0] > 0)) queue_free = max(0, MAX_QUEUE_SIZE - queue_used) @@ -684,7 +684,7 @@ def assign_jobs_to_available_nodes(self, job_queue_2d, nodes, cores_available, r self.dropped_this_episode += 1 else: job_queue_2d[job_idx][1] = new_age - + return num_processed_jobs, next_empty_slot, num_dropped def baseline_step(self, current_price, new_jobs_count, new_jobs_durations, new_jobs_nodes, new_jobs_cores): @@ -871,7 +871,7 @@ def record_episode_completion(self): # Calculate completion rates completion_rate = (self.jobs_completed / self.jobs_submitted * 100) if self.jobs_submitted > 0 else 0 baseline_completion_rate = (self.baseline_jobs_completed / self.baseline_jobs_submitted * 100) if self.baseline_jobs_submitted > 0 else 0 - + drop_rate = (self.jobs_dropped / self.jobs_submitted * 100) if self.jobs_submitted else 0.0 baseline_drop_rate = (self.baseline_jobs_dropped / self.baseline_jobs_submitted * 100) if self.baseline_jobs_submitted else 0.0 @@ -898,7 +898,7 @@ def record_episode_completion(self): 'baseline_avg_wait_time': float(baseline_avg_wait_time), 'baseline_completion_rate': float(baseline_completion_rate), 'baseline_max_queue_size': self.baseline_max_queue_size_reached, - + #Drop metrics "jobs_dropped": self.jobs_dropped, "drop_rate": float(drop_rate), From fa3895ef813199fa145987e2f5c76db68639f2e6 Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Thu, 18 Dec 2025 15:01:38 +0100 Subject: [PATCH 08/11] prices_deterministic now does not reset price list after 2 weeks. Track episode/time counter in environment, pass it to prices function, and read the current price. --- environment.py | 23 ++++++++++++++++++++++- prices_deterministic.py | 29 +++++++++++++++++++---------- 2 files changed, 41 insertions(+), 11 deletions(-) diff --git a/environment.py b/environment.py index 1e7121d..a253b27 100644 --- a/environment.py +++ b/environment.py @@ -187,6 +187,9 @@ def __init__(self, # Job tracking metrics for baseline self.baseline_jobs_dropped = 0 self.baseline_dropped_this_episode = 0 + #Initialize to -1, so that first reset() sets it to 0 + self.episode_idx = -1 + self.reset_state() @@ -248,8 +251,26 @@ def reset(self, seed = None, options = None): super().reset(seed=seed) self.np_random, self._seed = seeding.np_random(seed) + # Track which episode this env instance is on + if not hasattr(self, "episode_idx"): + self.episode_idx = 0 + else: + self.episode_idx += 1 + + # Reset counters & metrics self.reset_state() - self.prices.reset() + + # Choose starting index in the external price series + if self.prices is not None and getattr(self.prices, "external_prices", None) is not None: + n_prices = len(self.prices.external_prices) + episode_span = EPISODE_HOURS # e.g. 14 * 24 = 336 hours per episode + + # Episode k starts at hour k * episode_span (wrapping around the year) + start_index = (self.episode_idx * episode_span) % n_prices + self.prices.reset(start_index=start_index) + else: + # Synthetic prices or no external prices + self.prices.reset(start_index=0) self.state = { # Initialize all nodes to be 'online but free' (0) diff --git a/prices_deterministic.py b/prices_deterministic.py index f08636f..15e3b08 100644 --- a/prices_deterministic.py +++ b/prices_deterministic.py @@ -38,7 +38,7 @@ def __init__(self, external_prices=None): self.MIN_PRICE = 16 # IMPORTANT: initialize state in one place - self.reset() + self.reset(start_index=0) # ---------- core price model (pure) ---------- def _synthetic_price_at(self, t: int) -> float: @@ -49,18 +49,27 @@ def _synthetic_price_at(self, t: int) -> float: def get_real_price(self, shifted_price): return shifted_price - self.price_shift - def reset(self): - """Reset internal timeline/state to episode start (determinism-critical).""" - self.price_index = self.PREDICTION_WINDOW - self.price_history = deque(maxlen=self.HISTORY_WINDOW) + def reset(self, start_index: int = 0): + """Reset internal timeline/state to episode start. + + start_index is the index in external_prices for the *first* element + of the 24h prediction window. + """ + self.price_history = [] if self.external_prices is not None: - self.predicted_prices = np.array( - self.external_prices[:self.PREDICTION_WINDOW], - dtype=np.float32, - copy=True, - ) + n = len(self.external_prices) + start_index = start_index % n + + # 24-hour prediction window starting at start_index, with wrap + idxs = (np.arange(self.PREDICTION_WINDOW, dtype=np.int64) + start_index) % n + self.predicted_prices = self.external_prices[idxs].astype(np.float32, copy=True) + + # next unseen price *after* the window + self.price_index = (start_index + self.PREDICTION_WINDOW) % n else: + # synthetic mode + self.price_index = self.PREDICTION_WINDOW self.predicted_prices = np.array( [self._synthetic_price_at(i) for i in range(self.PREDICTION_WINDOW)], dtype=np.float32, From 2bdbdc1adec0c94b1675fe86bcd6dff4fb135e7b Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Thu, 18 Dec 2025 15:29:42 +0100 Subject: [PATCH 09/11] Fix: Implement bounded price history using deque. And removed unecessary import and initialization in env --- environment.py | 2 -- prices_deterministic.py | 5 ++--- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/environment.py b/environment.py index a253b27..f0baa88 100644 --- a/environment.py +++ b/environment.py @@ -5,7 +5,6 @@ import numpy as np from colorama import init, Fore -#from prices import Prices from prices_deterministic import Prices # Test re-worked prices script from weights import Weights from plot import plot, plot_reward @@ -344,7 +343,6 @@ def reset_state(self): self.dropped_this_episode = 0 self.baseline_dropped_this_episode = 0 - self.prev_excess_dropped = 0 def step(self, action): diff --git a/prices_deterministic.py b/prices_deterministic.py index 15e3b08..f0dd799 100644 --- a/prices_deterministic.py +++ b/prices_deterministic.py @@ -15,7 +15,7 @@ def __init__(self, external_prices=None): self.price_shift = 0 self.price_index = 0 - self.price_history = [] + self.price_history = deque(maxlen=self.HISTORY_WINDOW) self.predicted_prices = None if self.original_prices is not None: @@ -55,7 +55,7 @@ def reset(self, start_index: int = 0): start_index is the index in external_prices for the *first* element of the 24h prediction window. """ - self.price_history = [] + self.price_history = deque(maxlen=self.HISTORY_WINDOW) if self.external_prices is not None: n = len(self.external_prices) @@ -85,7 +85,6 @@ def get_next_price(self): self.price_index += 1 self.price_history.append(new_price) - # deque automatically removes oldest when maxlen exceeded | Keep the old line for now. Remove during next PR. return new_price From 8dc77fde85a5a1a171f9056a6ffd57c952f68965 Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Thu, 18 Dec 2025 15:42:03 +0100 Subject: [PATCH 10/11] Removed redundant initialzation --- environment.py | 1 - 1 file changed, 1 deletion(-) diff --git a/environment.py b/environment.py index f0baa88..7db3eae 100644 --- a/environment.py +++ b/environment.py @@ -342,7 +342,6 @@ def reset_state(self): self.baseline_jobs_rejected_queue_full = 0 self.dropped_this_episode = 0 - self.baseline_dropped_this_episode = 0 def step(self, action): From d4e7234da496fe1fbdac6de8f5ffa08e7cb96426 Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Thu, 18 Dec 2025 16:06:10 +0100 Subject: [PATCH 11/11] Fixed jobs_sampler call in step function and "rejected" counters now in init --- environment.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/environment.py b/environment.py index 7db3eae..a8e73e3 100644 --- a/environment.py +++ b/environment.py @@ -182,6 +182,8 @@ def __init__(self, # Job tracking metrics for agent self.jobs_dropped = 0 self.dropped_this_episode = 0 + self.jobs_rejected_queue_full = 0 + self.baseline_jobs_rejected_queue_full = 0 # Job tracking metrics for baseline self.baseline_jobs_dropped = 0 @@ -369,7 +371,7 @@ def step(self, action): new_jobs_cores = [] new_jobs_count = 0 - if self.external_jobs: + if self.external_jobs and not self.workload_gen: jobs = self.jobs_sampler.sample_one_hourly(wrap=True)["hourly_jobs"] if len(jobs) > 0: for job in jobs: