diff --git a/environment.py b/environment.py index d34e131..a8e73e3 100644 --- a/environment.py +++ b/environment.py @@ -5,13 +5,16 @@ 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 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 @@ -37,6 +40,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 @@ -75,6 +81,17 @@ def env_print(self, *args): 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) + 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, session, @@ -96,7 +113,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 @@ -129,18 +147,25 @@ def __init__(self, self.prices = Prices(self.external_prices) + #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}") @@ -154,6 +179,19 @@ 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 + self.jobs_rejected_queue_full = 0 + self.baseline_jobs_rejected_queue_full = 0 + + # 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() @@ -211,7 +249,29 @@ def __init__(self, }) 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() + + # 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) @@ -219,7 +279,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 = { @@ -270,6 +330,21 @@ 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 + def step(self, action): self.current_step += 1 @@ -278,7 +353,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)})) @@ -296,8 +371,8 @@ def step(self, action): new_jobs_cores = [] new_jobs_count = 0 - if self.external_jobs: - jobs = jobs_sampler.sample_one_hourly(wrap=True)['hourly_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: new_jobs_count += 1 @@ -306,7 +381,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 @@ -314,19 +401,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)})") @@ -341,7 +461,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 @@ -368,6 +488,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 + 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) @@ -411,8 +532,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() @@ -518,6 +639,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 @@ -525,33 +647,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 - - # Add to job allocation + nodes[node_idx] = max(nodes[node_idx], job_duration) 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": job_duration, + "allocation": job_allocation, } self.next_job_id += 1 @@ -571,11 +682,30 @@ def assign_jobs_to_available_nodes(self, job_queue_2d, nodes, cores_available, r self.total_job_wait_time += job_age num_processed_jobs += 1 + continue + + # Not enough resources -> job waits and ages (or gets dropped) + new_age = job_age + 1 + + if new_age > MAX_JOB_AGE: + # Clear job from queue + job_queue_2d[job_idx] = [0, 0, 0, 0] + + # 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 - 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) @@ -584,8 +714,9 @@ 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) + _, 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) @@ -637,15 +768,20 @@ def calculate_reward(self, num_used_nodes, num_idle_nodes, current_price, averag self.job_age_penalties.append(job_age_penalty_norm * 100) self.idle_penalties.append(idle_penalty_norm * 100) + # 6. penalty for dropped jobs (WIP - unnormalized, weighted) + drop_penalty = min(0, PENALTY_DROPPED_JOB * self.num_dropped_this_step) + drop_penalty_weighted = self.weights.drop_weight * drop_penalty + reward = ( efficiency_reward_weighted # + 0.0 * turned_off_reward + price_reward_weighted + job_age_penalty_weighted + idle_penalty_weighted + + 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 @@ -756,6 +892,10 @@ def record_episode_completion(self): 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, 'agent_cost': float(self.total_cost), @@ -777,7 +917,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) diff --git a/prices_deterministic.py b/prices_deterministic.py new file mode 100644 index 0000000..f0dd799 --- /dev/null +++ b/prices_deterministic.py @@ -0,0 +1,151 @@ +import numpy as np +import matplotlib.pyplot as plt +from collections import deque + +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 = deque(maxlen=self.HISTORY_WINDOW) + 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(start_index=0) + + # ---------- 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, 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 = deque(maxlen=self.HISTORY_WINDOW) + + if self.external_prices is not None: + 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, + ) + + # ---------- *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) + + 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 23eba45..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 @@ -100,7 +99,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 +118,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, diff --git a/test_determinism_workloadgen.py b/test_determinism_workloadgen.py new file mode 100644 index 0000000..36fc234 --- /dev/null +++ b/test_determinism_workloadgen.py @@ -0,0 +1,76 @@ +# 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": + # 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) + 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) + unif_cfg = WorkloadGenConfig(arrivals="uniform", max_new_jobs_per_hour=1000) + + for cfg in (flat_cfg, pois_cfg, unif_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..3f658d4 --- /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 48.0 < mean < 52.0, mean # tighter band, still reliable for 2000 samples + +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..bf77c8f 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 @@ -33,10 +48,15 @@ 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)") 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 @@ -44,7 +64,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}") @@ -57,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}" @@ -75,14 +96,31 @@ def main(): if not os.path.exists(plots_dir): os.makedirs(plots_dir) + # 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=prices, - external_durations=job_durations_file_path, - external_jobs=jobs_file_path, - external_hourly_jobs=hourly_jobs_file_path, + 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, @@ -95,7 +133,8 @@ def main(): 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 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) 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)]