From 21d6410b6eb50e914169ec3c3e14acc7cf4ac718 Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Mon, 23 Mar 2026 12:02:45 +0100 Subject: [PATCH 1/9] fix: escalate lost-job penalties and report proportional savings Apply a direct lost-job penalty of -1.0 for the first dropped job in a step, plus -0.25 for each additional dropped job, capped at 1000 extra drops. This makes repeated job loss increasingly costly while avoiding an unbounded reward hit. Also add proportional power/cost totals, mean prices, and savings percentages to the training summary. --- src/environment.py | 2 +- src/reward_calculation.py | 7 ++++--- train.py | 15 ++++++++++++++- 3 files changed, 19 insertions(+), 5 deletions(-) diff --git a/src/environment.py b/src/environment.py index 034e3f3..cd9a342 100644 --- a/src/environment.py +++ b/src/environment.py @@ -505,7 +505,7 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, self.metrics.episode_rewards.append(step_reward) self.metrics.jobs_dropped += num_dropped_this_step self.metrics.episode_jobs_dropped += num_dropped_this_step - + # print stats self.env_print(f"[6] End of step stats...") self.env_print("job queue: ", ' '.join(['[{} {} {} {}]'.format(d, a, n, c) for d, a, n, c in job_queue_2d if d > 0])) diff --git a/src/reward_calculation.py b/src/reward_calculation.py index 2a36188..3925a8b 100644 --- a/src/reward_calculation.py +++ b/src/reward_calculation.py @@ -367,9 +367,10 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo idle_penalty_norm = self._penalty_idle_normalized(num_idle_nodes) idle_penalty_weighted = weights.idle_weight * idle_penalty_norm - # 5. penalty for lost jobs (aged out or rejected because queue/backlog was full) - drop_penalty = self._penalty_drop(num_dropped_this_step) - drop_penalty_weighted = weights.drop_weight * drop_penalty + # 6. penalty for lost jobs (aged out or rejected because queue/backlog was full) + drop_penalty_weighted = 0 + if num_dropped_this_step > 0: + drop_penalty_weighted = -1.0 - 0.25 * min(num_dropped_this_step - 1, 1000) # harsher penalty for losing many jobs, capped at -251.0 for 1000+ jobs reward = ( efficiency_reward_weighted diff --git a/train.py b/train.py index e6f30c8..05abbac 100644 --- a/train.py +++ b/train.py @@ -325,9 +325,22 @@ def main(): total_agent_power_mwh = sum(float(ep.get('agent_power_consumption_mwh', 0.0)) for ep in env.metrics.episode_costs) total_baseline_power_mwh = sum(float(ep.get('baseline_power_consumption_mwh', 0.0)) for ep in env.metrics.episode_costs) total_baseline_off_power_mwh = sum(float(ep.get('baseline_power_consumption_off_mwh', 0.0)) for ep in env.metrics.episode_costs) + total_agent_prop_power_mwh = sum(float(ep.get('agent_prop_power_mwh', 0.0)) for ep in env.metrics.episode_costs) + total_baseline_prop_power_mwh = sum(float(ep.get('baseline_prop_power_mwh', 0.0)) for ep in env.metrics.episode_costs) + total_baseline_off_prop_power_mwh = sum(float(ep.get('baseline_off_prop_power_mwh', 0.0)) for ep in env.metrics.episode_costs) + total_agent_prop_cost = sum(float(ep.get('agent_prop_cost', 0.0)) for ep in env.metrics.episode_costs) + total_baseline_prop_cost = sum(float(ep.get('baseline_prop_cost', 0.0)) for ep in env.metrics.episode_costs) + total_baseline_off_prop_cost = sum(float(ep.get('baseline_off_prop_cost', 0.0)) for ep in env.metrics.episode_costs) + total_savings_prop_cost_vs_baseline = total_baseline_prop_cost - total_agent_prop_cost + total_savings_prop_cost_vs_baseline_off = total_baseline_off_prop_cost - total_agent_prop_cost total_agent_mean_price = (total_agent_cost / total_agent_power_mwh) if total_agent_power_mwh > 0 else 0.0 total_baseline_mean_price = (total_baseline_cost / total_baseline_power_mwh) if total_baseline_power_mwh > 0 else 0.0 total_baseline_off_mean_price = (total_baseline_off_cost / total_baseline_off_power_mwh) if total_baseline_off_power_mwh > 0 else 0.0 + total_agent_prop_mean_price = (total_agent_prop_cost / total_agent_prop_power_mwh) if total_agent_prop_power_mwh > 0 else 0.0 + total_baseline_prop_mean_price = (total_baseline_prop_cost / total_baseline_prop_power_mwh) if total_baseline_prop_power_mwh > 0 else 0.0 + total_baseline_off_prop_mean_price = (total_baseline_off_prop_cost / total_baseline_off_prop_power_mwh) if total_baseline_off_prop_power_mwh > 0 else 0.0 + prop_savings_pct_vs_baseline = safe_ratio(total_savings_prop_cost_vs_baseline * 100.0, total_baseline_prop_cost) + prop_savings_pct_vs_baseline_off = safe_ratio(total_savings_prop_cost_vs_baseline_off * 100.0, total_baseline_off_prop_cost) total_agent_completion_rate = (total_jobs_completed / total_jobs_submitted * 100) if total_jobs_submitted > 0 else 0.0 total_baseline_completion_rate = (total_baseline_completed / total_baseline_submitted * 100) if total_baseline_submitted > 0 else 0.0 total_savings_vs_baseline = total_baseline_cost - total_agent_cost @@ -365,7 +378,7 @@ def main(): print(f" Baseline: {fmt_optional(total_baseline_cost_per_1000_completed, 2, thousands=True)} €/1k jobs") print(f" Baseline_off: {fmt_optional(total_baseline_off_cost_per_1000_completed, 2, thousands=True)} €/1k jobs") - print("\n=== AGENT LOST JOBS PER SAVED EURO ===") + print(f"\n=== AGENT LOST JOBS PER SAVED EURO ===") print(f" Total Lost Jobs (Agent): {total_jobs_dropped:,}") print(f" Total Lost Jobs (Baseline): {total_baseline_jobs_dropped:,}") print(f" Vs Baseline: {fmt_optional(total_dropped_jobs_per_saved_euro, 6)} jobs/€") From 025a6690bff30fdebe70d9245cf6d8bb473fac2e Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Fri, 27 Mar 2026 12:07:34 +0100 Subject: [PATCH 2/9] Test job age: Only start penalizing age, when wait time > 0. Ideally, this allows free shifting within 24 rollout window. --- src/reward_calculation.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/reward_calculation.py b/src/reward_calculation.py index 3925a8b..05adddb 100644 --- a/src/reward_calculation.py +++ b/src/reward_calculation.py @@ -54,9 +54,9 @@ class RewardCalculator: # Price scaling uses active used nodes as work proxy, matching efficiency semantics. PRICE_ADVANTAGE_GAIN = 4.0 # Asymmetric node scaling: high-price execution ramps faster than low-price reward. - PRICE_NODE_TAU_POS = 70.0 + PRICE_NODE_TAU_POS = 40.0 PRICE_NODE_TAU_NEG = 40.0 - NEGATIVE_PRICE_NODE_TAU = 14.0 # fast node saturation only for negative-price overdrive + NEGATIVE_PRICE_NODE_TAU = 30.0 # fast node saturation only for negative-price overdrive NEGATIVE_PRICE_TAU = 8.0 # Overdrive terms for negative prices: # - gain controls overdrive strength during negative-price windows @@ -213,10 +213,10 @@ def _penalty_job_age(num_off_nodes: int, job_queue_2d: np.ndarray) -> float: valid_mask = job_queue_2d[:, 0] > 0 # [valid_mask, 1] selects column 1 (age) only for rows where mask is True max_age = job_queue_2d[valid_mask, 1].max() if valid_mask.any() else 0 - if max_age > 0: - tau_hours = WEEK_HOURS / 2.0 - max_factor = 1.0 - np.exp(-WEEK_HOURS / tau_hours) - factor = 1.0 - np.exp(-max_age / tau_hours) + if max_age > 24: + tau_hours = WEEK_HOURS + max_factor = 1.0 - np.exp(-(2*WEEK_HOURS) / tau_hours) + factor = 1.0 - np.exp(-(max_age-24) / tau_hours) factor = min(factor / max_factor, 1.0) job_age_penalty = factor return job_age_penalty From d02c6ca46e6621c33137e809a501acb94d991474 Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Mon, 23 Mar 2026 12:02:45 +0100 Subject: [PATCH 3/9] feat: allow disabling lost-job reward penalties Add an ALLOW_DROP_PENALTY switch to RewardCalculator and gate the dropped-job reward penalty behind it. The flag defaults to true, so the current behavior is preserved: the first lost job costs -1.0, each additional same-step loss adds -0.25, and the extra-drop penalty is capped after 1000 drops. --- src/reward_calculation.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/reward_calculation.py b/src/reward_calculation.py index 05adddb..639417f 100644 --- a/src/reward_calculation.py +++ b/src/reward_calculation.py @@ -71,6 +71,9 @@ class RewardCalculator: # Drop penalty: tanh saturation curve. TAU=20: 1 drop≈-0.05, 10 drops≈-0.46, 50 drops≈-1.0. DROP_PENALTY_TAU = 20.0 + + ALLOW_DROP_PENALTY = True # whether to include penalties for dropped jobs in the reward calculation + def __init__(self, prices: Prices) -> None: """ Initialize reward calculator with normalization bounds. @@ -369,7 +372,7 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo # 6. penalty for lost jobs (aged out or rejected because queue/backlog was full) drop_penalty_weighted = 0 - if num_dropped_this_step > 0: + if self.ALLOW_DROP_PENALTY and num_dropped_this_step > 0: drop_penalty_weighted = -1.0 - 0.25 * min(num_dropped_this_step - 1, 1000) # harsher penalty for losing many jobs, capped at -251.0 for 1000+ jobs reward = ( From 42afd37dfcf0d7ffdb9880013dc28abf759eea6c Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Mon, 30 Mar 2026 12:12:34 +0200 Subject: [PATCH 4/9] Teach defer-until-cheap scheduling with backlog pressure and launch-time wait metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reward now explicitly teaches “defer into cheap, then serve hard in cheap”: the old blackout penalty no longer fights deferral, cheap hours penalize under-service when backlog exists, overdue backlog after the 24h grace gets punished, and drop_weight now actually scales the drop penalty. I also reset queue/backlog state cleanly between episodes, switched AvgWait to launch-time wait instead of completion-time wait, and added end-of-episode pending/overdue metrics --- src/callbacks.py | 15 +++- src/environment.py | 172 +++++++++++++++++++++++++++----------- src/evaluation_summary.py | 2 + src/job_management.py | 12 +++ src/metrics_tracker.py | 35 ++++++-- src/plotter.py | 52 +++++++----- src/reward_calculation.py | 167 +++++++++++++++++++++++++----------- src/workloadgen.py | 8 +- train.py | 18 ++-- 9 files changed, 346 insertions(+), 135 deletions(-) diff --git a/src/callbacks.py b/src/callbacks.py index 2675696..87c8062 100644 --- a/src/callbacks.py +++ b/src/callbacks.py @@ -44,8 +44,12 @@ def _on_step(self) -> bool: # Job metrics (agent) completion_rate = (env.metrics.episode_jobs_completed / env.metrics.episode_jobs_submitted * 100 if env.metrics.episode_jobs_submitted > 0 else 0.0) - avg_wait = (env.metrics.episode_total_job_wait_time / env.metrics.episode_jobs_completed if env.metrics.episode_jobs_completed > 0 else 0.0) + avg_wait = ( + env.metrics.episode_total_job_wait_time_launch / env.metrics.episode_jobs_launched + if env.metrics.episode_jobs_launched > 0 else 0.0 + ) self.logger.record("metrics/jobs_submitted", env.metrics.episode_jobs_submitted) + self.logger.record("metrics/jobs_launched", env.metrics.episode_jobs_launched) self.logger.record("metrics/jobs_completed", env.metrics.episode_jobs_completed) self.logger.record("metrics/completion_rate", completion_rate) self.logger.record("metrics/avg_wait_hours", avg_wait) @@ -55,6 +59,9 @@ def _on_step(self) -> bool: self.logger.record("metrics/nodes_off", MAX_NODES - env.metrics.episode_on_nodes[-1]) self.logger.record("metrics/max_queue_size", env.metrics.episode_max_queue_size_reached) self.logger.record("metrics/max_backlog_size", env.metrics.episode_max_backlog_size_reached) + self.logger.record("metrics/pending_jobs_end", env.metrics.episode_pending_jobs_end) + self.logger.record("metrics/pending_core_hours_end", env.metrics.episode_pending_core_hours_end) + self.logger.record("metrics/overdue_jobs_end", env.metrics.episode_overdue_jobs_end) self.logger.record("metrics/jobs_dropped", env.metrics.episode_jobs_dropped) self.logger.record("metrics/jobs_lost_total", env.metrics.episode_jobs_dropped) loss_rate = (env.metrics.episode_jobs_dropped / env.metrics.episode_jobs_submitted * 100 if env.metrics.episode_jobs_submitted > 0 else 0.0) @@ -63,8 +70,12 @@ def _on_step(self) -> bool: # Job metrics (baseline) baseline_completion_rate = (env.metrics.episode_baseline_jobs_completed / env.metrics.episode_baseline_jobs_submitted * 100 if env.metrics.episode_baseline_jobs_submitted > 0 else 0.0) - baseline_avg_wait = (env.metrics.episode_baseline_total_job_wait_time / env.metrics.episode_baseline_jobs_completed if env.metrics.episode_baseline_jobs_completed > 0 else 0.0) + baseline_avg_wait = ( + env.metrics.episode_baseline_total_job_wait_time_launch / env.metrics.episode_baseline_jobs_launched + if env.metrics.episode_baseline_jobs_launched > 0 else 0.0 + ) self.logger.record("metrics/bl_jobs_submitted", env.metrics.episode_baseline_jobs_submitted) + self.logger.record("metrics/bl_jobs_launched", env.metrics.episode_baseline_jobs_launched) self.logger.record("metrics/bl_jobs_completed", env.metrics.episode_baseline_jobs_completed) self.logger.record("metrics/bl_completion_rate", baseline_completion_rate) self.logger.record("metrics/bl_avg_wait_hours", baseline_avg_wait) diff --git a/src/environment.py b/src/environment.py index cd9a342..f2f6aef 100644 --- a/src/environment.py +++ b/src/environment.py @@ -150,6 +150,9 @@ def __init__(self, self.metrics.reset_timeline_metrics() self.metrics.reset_episode_metrics() self._reset_timeline_state(start_index=0) + # A fresh env still needs one real reset() call before rollout starts. After that, + # episode boundaries should only roll metrics, not wipe the live simulation state. + self._timeline_initialized = False # actions: - change number of available nodes: # action_type: 0: decrease, 1: maintain, 2: increase @@ -224,58 +227,108 @@ def _mark_queue_backlog_mutation(self) -> None: """Invalidate pending-job stats cache after queue/backlog content changes.""" self._queue_backlog_version += 1 + def _pending_work_summary(self, job_queue_2d: np.ndarray, backlog_queue: deque | None = None) -> dict[str, float | int]: + """ + Summarize currently pending work across the visible queue and the overflow backlog. + + The reward only needs a few dense signals: + - instantaneous runnable demand (nodes * cores) + - longer-horizon remaining work (core-hours) + - overdue mass after the intentional deferral grace period + """ + if backlog_queue is None: + backlog_queue = self.backlog_queue + + active_jobs_mask = job_queue_2d[:, 0] > 0 + queue_rows = job_queue_2d[active_jobs_mask].astype(np.float32, copy=False) + + if backlog_queue: + backlog_rows = np.asarray(list(backlog_queue), dtype=np.float32) + else: + backlog_rows = np.empty((0, 4), dtype=np.float32) + + if queue_rows.size and backlog_rows.size: + combined_rows = np.vstack((queue_rows, backlog_rows)) + elif queue_rows.size: + combined_rows = queue_rows + elif backlog_rows.size: + combined_rows = backlog_rows + else: + combined_rows = np.empty((0, 4), dtype=np.float32) + + if combined_rows.size == 0: + return { + "pending_job_count": 0, + "pending_core_demand": 0.0, + "pending_core_hours": 0.0, + "pending_avg_duration": 0.0, + "pending_max_nodes": 0, + "backlog_size": len(backlog_queue), + "oldest_age": 0.0, + "overdue_jobs": 0, + "overdue_age_core_hours": 0.0, + } + + durations = combined_rows[:, 0] + ages = combined_rows[:, 1] + nodes = combined_rows[:, 2] + cores = combined_rows[:, 3] + core_demand = nodes * cores + + grace = float(self.reward_calculator.DEFERRAL_GRACE_HOURS) + overdue_age = np.clip(ages - grace, a_min=0.0, a_max=None) + overdue_mask = overdue_age > 0.0 + + return { + "pending_job_count": int(combined_rows.shape[0]), + "pending_core_demand": float(np.sum(core_demand)), + "pending_core_hours": float(np.sum(durations * core_demand)), + "pending_avg_duration": float(np.mean(durations)), + "pending_max_nodes": int(np.max(nodes)), + "backlog_size": len(backlog_queue), + "oldest_age": float(np.max(ages)), + "overdue_jobs": int(np.count_nonzero(overdue_mask)), + "overdue_age_core_hours": float(np.sum(overdue_age * core_demand)), + } + def _update_pending_job_stats(self, job_queue_2d: np.ndarray) -> None: """Update summary statistics for all outstanding jobs (queue + backlog).""" # Fast path: skip recalculation if queue/backlog version is unchanged. if self._cached_queue_backlog_version == self._queue_backlog_version: return # Stats unchanged from last step - # Slow path: recalculate pending stats after queue/backlog mutations. - # Collect stats from the main queue - current_backlog_size = len(self.backlog_queue) - active_jobs_mask = job_queue_2d[:, 0] > 0 - queue_durations = job_queue_2d[active_jobs_mask, 0] - queue_nodes = job_queue_2d[active_jobs_mask, 2] - queue_cores = job_queue_2d[active_jobs_mask, 3] - queue_count = len(queue_durations) - - # Collect stats from the backlog - backlog_count = current_backlog_size - if backlog_count > 0: - backlog_durations = np.array([job[0] for job in self.backlog_queue], dtype=np.int32) - backlog_nodes = np.array([job[2] for job in self.backlog_queue], dtype=np.int32) - backlog_cores = np.array([job[3] for job in self.backlog_queue], dtype=np.int32) - else: - backlog_durations = np.array([], dtype=np.int32) - backlog_nodes = np.array([], dtype=np.int32) - backlog_cores = np.array([], dtype=np.int32) - - # Combine stats - total_count = queue_count + backlog_count - if total_count > 0: - all_durations = np.concatenate([queue_durations, backlog_durations]) - all_nodes = np.concatenate([queue_nodes, backlog_nodes]) - all_cores = np.concatenate([queue_cores, backlog_cores]) - - # Core-hours = sum of (duration * nodes * cores_per_node) - total_core_hours = np.sum(all_durations * all_nodes * all_cores) - avg_duration = np.mean(all_durations) - max_nodes = np.max(all_nodes) - else: - total_core_hours = 0.0 - avg_duration = 0.0 - max_nodes = 0 - + summary = self._pending_work_summary(job_queue_2d) # Update state - self.state['pending_job_count'][0] = total_count - self.state['pending_core_hours'][0] = total_core_hours - self.state['pending_avg_duration'][0] = avg_duration - self.state['pending_max_nodes'][0] = max_nodes - self.state['backlog_size'][0] = backlog_count + self.state['pending_job_count'][0] = int(summary["pending_job_count"]) + self.state['pending_core_hours'][0] = float(summary["pending_core_hours"]) + self.state['pending_avg_duration'][0] = float(summary["pending_avg_duration"]) + self.state['pending_max_nodes'][0] = int(summary["pending_max_nodes"]) + self.state['backlog_size'][0] = int(summary["backlog_size"]) # Cache the queue/backlog version for next step. self._cached_queue_backlog_version = self._queue_backlog_version + def _resolve_reset_start_index(self, options: dict[str, Any]) -> int: + """ + Choose the initial price position for a true timeline reset. + + Normal episode rollovers are continuous and should not call this helper. It is + only used when the whole simulation timeline is intentionally restarted. + """ + if "price_start_index" in options: + if self.prices is not None and self.prices.external_prices is not None: + n_prices = len(self.prices.external_prices) + return int(options["price_start_index"]) % n_prices + return int(options["price_start_index"]) + + if self.prices.external_prices is not None: + if self.evaluation_mode: + return (self.episode_idx * EPISODE_HOURS) % len(self.prices.external_prices) + return int(self.np_random.integers(0, len(self.prices.external_prices))) + + # Even the synthetic logic prices benefit from varying the initial phase during training. + return int(self.np_random.integers(0, self.prices.PREDICTION_WINDOW)) + def reset(self, seed: int | None = None, options: dict[str, Any] | None = None) -> tuple[dict[str, np.ndarray], dict]: if options is None: options = {} @@ -289,14 +342,23 @@ def reset(self, seed: int | None = None, options: dict[str, Any] | None = None) self.episode_idx += 1 self.metrics.reset_episode_metrics() - if "price_start_index" in options: - if self.prices is not None and self.prices.external_prices is not None: - n_prices = len(self.prices.external_prices) - start_index = int(options["price_start_index"]) % n_prices - else: - start_index = int(options["price_start_index"]) - self.prices.reset(start_index=start_index) - self.state["predicted_prices"] = self.prices.predicted_prices.copy() + + hard_reset = bool(options.get("hard_reset", False)) + if not self._timeline_initialized or hard_reset: + # Hard reset: restart prices, queues, nodes, backlog, and running jobs. + # This is only for the very first env reset or for explicit ablations/debug runs. + start_index = self._resolve_reset_start_index(options) + self._reset_timeline_state(start_index=start_index) + self._timeline_initialized = True + else: + # Soft reset: keep the ongoing simulation exactly as-is and just start a new + # reporting window. Any price_start_index request is ignored unless a hard reset + # is explicitly requested, because jumping the price stream mid-timeline would + # break continuity. + job_queue_2d = self.state['job_queue'].reshape(-1, 4) + self._mark_queue_backlog_mutation() + self._update_pending_job_stats(job_queue_2d) + self.state['predicted_prices'] = self.prices.predicted_prices.copy() return self.state, {} @@ -374,6 +436,9 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, self.env_print(f">>> adding {len(new_jobs)} new jobs to the queue: {' '.join(['[{}h {} {}x{}]'.format(d, a, n, c) for d, a, n, c in new_jobs])}") self.env_print("job_queue: ", ' '.join(['[{} {} {} {}]'.format(d, a, n, c) for d, a, n, c in job_queue_2d if d > 0])) + # Snapshot the pending queue the agent is deciding about *before* launching jobs. + decision_pending_summary = self._pending_work_summary(job_queue_2d) + action_type, action_magnitude, do_refill = action action_magnitude += 1 @@ -412,6 +477,8 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, if extra_launched > 0: self.env_print(f" {extra_launched} additional jobs launched from backlog") + remaining_pending_summary = self._pending_work_summary(job_queue_2d) + # Update summary statistics for all outstanding jobs (queue + backlog) if queue_backlog_mutated: self._mark_queue_backlog_mutation() @@ -441,6 +508,11 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, self.metrics.episode_used_cores.append(num_used_cores) self.metrics.episode_job_queue_sizes.append(num_unprocessed_jobs) self.metrics.episode_price_stats.append(current_price) + self.metrics.episode_pending_jobs_end = int(remaining_pending_summary["pending_job_count"]) + self.metrics.episode_pending_core_demand_end = float(remaining_pending_summary["pending_core_demand"]) + self.metrics.episode_pending_core_hours_end = float(remaining_pending_summary["pending_core_hours"]) + self.metrics.episode_overdue_jobs_end = int(remaining_pending_summary["overdue_jobs"]) + self.metrics.episode_overdue_age_core_hours_end = float(remaining_pending_summary["overdue_age_core_hours"]) # Track max queue size (queue only, without backlog) queue_only_size = np.sum(job_queue_2d[:, 0] > 0) @@ -482,6 +554,8 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, num_used_nodes, num_idle_nodes, current_price, average_future_price, num_off_nodes, job_queue_2d, num_unprocessed_jobs, self.weights, num_dropped_this_step, self.env_print, num_on_nodes, num_used_cores, + decision_pending_core_demand=float(decision_pending_summary["pending_core_demand"]), + remaining_overdue_age_core_hours=float(remaining_pending_summary["overdue_age_core_hours"]), ) self.metrics.episode_reward += step_reward diff --git a/src/evaluation_summary.py b/src/evaluation_summary.py index f9eb575..7fdb147 100644 --- a/src/evaluation_summary.py +++ b/src/evaluation_summary.py @@ -57,6 +57,8 @@ def build_episode_summary_line( f"Jobs={int(episode_data['jobs_completed'])}/{int(episode_data['jobs_submitted'])} " f"({float(episode_data['completion_rate']):.0f}%), " f"AvgWait={float(episode_data['avg_wait_time']):.1f}h, " + f"PendingEnd={int(episode_data.get('pending_jobs_end', 0))}, " + f"OverdueEnd={int(episode_data.get('overdue_jobs_end', 0))}, " f"EpisodeMaxQueue={int(episode_data['max_queue_size'])}, " f"Lost={int(episode_data.get('jobs_lost_total', episode_data['jobs_dropped']))}, " f"TimelineMaxQueue={timeline_max_queue}, " diff --git a/src/job_management.py b/src/job_management.py index 994de55..2f5e25a 100644 --- a/src/job_management.py +++ b/src/job_management.py @@ -269,6 +269,18 @@ def assign_jobs_to_available_nodes( "allocation": job_allocation, "wait_time": int(job_age), } + # Record scheduling delay when the job starts. + # Completion metrics are tracked separately when the job actually finishes. + if is_baseline: + metrics.baseline_jobs_launched += 1 + metrics.baseline_total_job_wait_time_launch += int(job_age) + metrics.episode_baseline_jobs_launched += 1 + metrics.episode_baseline_total_job_wait_time_launch += int(job_age) + else: + metrics.jobs_launched += 1 + metrics.total_job_wait_time_launch += int(job_age) + metrics.episode_jobs_launched += 1 + metrics.episode_total_job_wait_time_launch += int(job_age) next_job_id += 1 # Clear job from queue diff --git a/src/metrics_tracker.py b/src/metrics_tracker.py index ef2dc71..b2fb571 100644 --- a/src/metrics_tracker.py +++ b/src/metrics_tracker.py @@ -37,8 +37,10 @@ def reset_timeline_metrics(self) -> None: # Agent job metrics (cumulative across episodes) self.jobs_submitted: int = 0 + self.jobs_launched: int = 0 self.jobs_completed: int = 0 self.total_job_wait_time: int = 0 + self.total_job_wait_time_launch: int = 0 self.max_queue_size_reached: int = 0 self.max_backlog_size_reached: int = 0 self.jobs_dropped: int = 0 @@ -46,8 +48,10 @@ def reset_timeline_metrics(self) -> None: # Baseline job metrics (cumulative across episodes) self.baseline_jobs_submitted: int = 0 + self.baseline_jobs_launched: int = 0 self.baseline_jobs_completed: int = 0 self.baseline_total_job_wait_time: int = 0 + self.baseline_total_job_wait_time_launch: int = 0 self.baseline_max_queue_size_reached: int = 0 self.baseline_max_backlog_size_reached: int = 0 self.baseline_jobs_dropped: int = 0 @@ -78,8 +82,10 @@ def reset_episode_metrics(self) -> None: # Agent job metrics (episode) self.episode_jobs_submitted: int = 0 + self.episode_jobs_launched: int = 0 self.episode_jobs_completed: int = 0 self.episode_total_job_wait_time: int = 0 + self.episode_total_job_wait_time_launch: int = 0 self.episode_max_queue_size_reached: int = 0 self.episode_max_backlog_size_reached: int = 0 self.episode_jobs_dropped: int = 0 @@ -87,13 +93,24 @@ def reset_episode_metrics(self) -> None: # Baseline job metrics (episode) self.episode_baseline_jobs_submitted: int = 0 + self.episode_baseline_jobs_launched: int = 0 self.episode_baseline_jobs_completed: int = 0 self.episode_baseline_total_job_wait_time: int = 0 + self.episode_baseline_total_job_wait_time_launch: int = 0 self.episode_baseline_max_queue_size_reached: int = 0 self.episode_baseline_max_backlog_size_reached: int = 0 self.episode_baseline_jobs_dropped: int = 0 self.episode_baseline_jobs_rejected_queue_full: int = 0 + # End-of-episode pending-work snapshot. + # These are updated every step so record_episode_completion() can report what + # backlog remains when the episode terminates. + self.episode_pending_jobs_end: int = 0 + self.episode_pending_core_demand_end: float = 0.0 + self.episode_pending_core_hours_end: float = 0.0 + self.episode_overdue_jobs_end: int = 0 + self.episode_overdue_age_core_hours_end: float = 0.0 + # Time series data for plotting (episode) self.episode_on_nodes: list[int] = [] self.episode_used_nodes: list[int] = [] @@ -121,15 +138,16 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i Returns: Dictionary with episode data """ - # Calculate average wait times + # Scheduling wait is measured when work launches, not when it eventually finishes. + # That makes "defer into cheap hours" visible immediately instead of only after completion. avg_wait_time: float = ( - self.episode_total_job_wait_time / self.episode_jobs_completed - if self.episode_jobs_completed > 0 + self.episode_total_job_wait_time_launch / self.episode_jobs_launched + if self.episode_jobs_launched > 0 else 0.0 ) baseline_avg_wait_time: float = ( - self.episode_baseline_total_job_wait_time / self.episode_baseline_jobs_completed - if self.episode_baseline_jobs_completed > 0 + self.episode_baseline_total_job_wait_time_launch / self.episode_baseline_jobs_launched + if self.episode_baseline_jobs_launched > 0 else 0.0 ) @@ -209,13 +227,20 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i 'total_reward': self.episode_reward, # Agent job metrics 'jobs_submitted': self.episode_jobs_submitted, + 'jobs_launched': self.episode_jobs_launched, 'jobs_completed': self.episode_jobs_completed, 'avg_wait_time': avg_wait_time, 'completion_rate': completion_rate, 'max_queue_size': self.episode_max_queue_size_reached, 'max_backlog_size': self.episode_max_backlog_size_reached, + 'pending_jobs_end': self.episode_pending_jobs_end, + 'pending_core_demand_end': self.episode_pending_core_demand_end, + 'pending_core_hours_end': self.episode_pending_core_hours_end, + 'overdue_jobs_end': self.episode_overdue_jobs_end, + 'overdue_age_core_hours_end': self.episode_overdue_age_core_hours_end, # Baseline job metrics 'baseline_jobs_submitted': self.episode_baseline_jobs_submitted, + 'baseline_jobs_launched': self.episode_baseline_jobs_launched, 'baseline_jobs_completed': self.episode_baseline_jobs_completed, 'baseline_avg_wait_time': baseline_avg_wait_time, 'baseline_completion_rate': baseline_completion_rate, diff --git a/src/plotter.py b/src/plotter.py index db4980a..d4309df 100644 --- a/src/plotter.py +++ b/src/plotter.py @@ -23,6 +23,30 @@ def _as_series(x: np.ndarray | list[float] | None, n: int) -> np.ndarray | None: return out +def _episode_launch_wait(metrics, baseline: bool = False) -> float: + """ + Return average queueing delay measured at launch. + + Old mock/test objects may still only expose completion-based wait fields, so this + helper keeps a compatible fallback for plotting utilities. + """ + if baseline: + launched = getattr(metrics, "episode_baseline_jobs_launched", metrics.episode_baseline_jobs_completed) + total_wait = getattr( + metrics, + "episode_baseline_total_job_wait_time_launch", + metrics.episode_baseline_total_job_wait_time, + ) + else: + launched = getattr(metrics, "episode_jobs_launched", metrics.episode_jobs_completed) + total_wait = getattr( + metrics, + "episode_total_job_wait_time_launch", + metrics.episode_total_job_wait_time, + ) + return (total_wait / launched) if launched > 0 else 0.0 + + def _compute_cumulative_savings(episode_costs: list[dict[str, float | int]]) -> dict[str, np.ndarray] | None: """ episode_costs: list of dicts with keys: @@ -113,7 +137,7 @@ def plot_episode(env: ComputeClusterEnv, num_hours: int, max_nodes: int, save: b if env.plot_config.plot_idle_penalty: ax2.plot(hours, env.metrics.episode_idle_penalties, color='green', linestyle='--', label='Idle Penalties') if env.plot_config.plot_job_age_penalty: - ax2.plot(hours, env.metrics.episode_job_age_penalties, color='yellow', linestyle='--', label='Job Age Penalties') + ax2.plot(hours, env.metrics.episode_job_age_penalties, color='yellow', linestyle='--', label='Backlog Pressure') ax2.tick_params(axis='y') if env.plot_config.plot_idle_penalty or env.plot_config.plot_job_age_penalty: @@ -131,16 +155,8 @@ def plot_episode(env: ComputeClusterEnv, num_hours: int, max_nodes: int, save: b if env.metrics.episode_baseline_jobs_submitted > 0 else 0 ) - avg_wait = ( - env.metrics.episode_total_job_wait_time / env.metrics.episode_jobs_completed - if env.metrics.episode_jobs_completed > 0 - else 0 - ) - baseline_avg_wait = ( - env.metrics.episode_baseline_total_job_wait_time / env.metrics.episode_baseline_jobs_completed - if env.metrics.episode_baseline_jobs_completed > 0 - else 0 - ) + avg_wait = _episode_launch_wait(env.metrics, baseline=False) + baseline_avg_wait = _episode_launch_wait(env.metrics, baseline=True) baseline_savings_pct = ( ((env.metrics.episode_baseline_cost - env.metrics.episode_total_cost) / env.metrics.episode_baseline_cost) * 100 if env.metrics.episode_baseline_cost > 0 @@ -197,16 +213,8 @@ def plot_dashboard(env: ComputeClusterEnv, num_hours: int, max_nodes: int, save: if env.metrics.episode_baseline_jobs_submitted > 0 else 0.0 ) - avg_wait = ( - (env.metrics.episode_total_job_wait_time / env.metrics.episode_jobs_completed) - if env.metrics.episode_jobs_completed > 0 - else 0.0 - ) - baseline_avg_wait = ( - (env.metrics.episode_baseline_total_job_wait_time / env.metrics.episode_baseline_jobs_completed) - if env.metrics.episode_baseline_jobs_completed > 0 - else 0.0 - ) + avg_wait = _episode_launch_wait(env.metrics, baseline=False) + baseline_avg_wait = _episode_launch_wait(env.metrics, baseline=True) base_cost = float(env.metrics.episode_baseline_cost) base_cost_off = float(env.metrics.episode_baseline_cost_off) @@ -268,7 +276,7 @@ def add_panel(title: str, series: np.ndarray | list[float] | None, ylabel: str, if env.plot_config.plot_idle_penalty: add_panel("Idle penalty (%)", env.metrics.episode_idle_penalties, "score", None) if env.plot_config.plot_job_age_penalty: - add_panel("Job-age penalty (%)", env.metrics.episode_job_age_penalties, "score", None) + add_panel("Backlog pressure (%)", env.metrics.episode_job_age_penalties, "score", None) if env.plot_config.plot_total_reward: add_panel("Total reward", getattr(env.metrics, "episode_rewards", None), "reward", None) diff --git a/src/reward_calculation.py b/src/reward_calculation.py index 639417f..2598c0d 100644 --- a/src/reward_calculation.py +++ b/src/reward_calculation.py @@ -58,12 +58,6 @@ class RewardCalculator: PRICE_NODE_TAU_NEG = 40.0 NEGATIVE_PRICE_NODE_TAU = 30.0 # fast node saturation only for negative-price overdrive NEGATIVE_PRICE_TAU = 8.0 - # Overdrive terms for negative prices: - # - gain controls overdrive strength during negative-price windows - # - floor guarantees a minimum positive drive proportional to negative-price strength and used work - # Toggle behavior: - # - capped mode (default): overdrive is folded into tanh, so reward stays <= 1 - # - uncapped mode: overdrive is added after tanh and can exceed 1 up to NEGATIVE_PRICE_OVERDRIVE_MAX_REWARD NEGATIVE_PRICE_OVERDRIVE_GAIN = 2.5 NEGATIVE_PRICE_OVERDRIVE_FLOOR = 0.35 NEGATIVE_PRICE_OVERDRIVE_ALLOW_ABOVE_ONE = True @@ -72,7 +66,11 @@ class RewardCalculator: DROP_PENALTY_TAU = 20.0 - ALLOW_DROP_PENALTY = True # whether to include penalties for dropped jobs in the reward calculation + # The first 24h are treated as deliberate deferral room; after that, starvation should ramp up quickly. + DEFERRAL_GRACE_HOURS = 24 + CHEAP_SERVICE_GAIN = 0.75 + OVERDUE_BACKLOG_GAIN = 1.25 + OVERDUE_AGE_CORE_HOUR_TAU = 0.5 * MAX_NODES * CORES_PER_NODE def __init__(self, prices: Prices) -> None: """ @@ -135,6 +133,37 @@ def _price_context_average(self, average_future_price: float) -> float: return (history_avg + future_avg) / 2 return average_future_price + def _price_phase_strengths(self, current_price: float) -> tuple[float, float]: + """ + Map the current price into a cheap-vs-expensive phase inside the visible forecast window. + + The quantile band is used as the "decision zone": + - at/below q_low -> fully cheap + - at/above q_high -> fully expensive + - inside the band -> smooth linear interpolation + + This is intentionally more explicit than the old sigmoid-based score. For the + synthetic 12h logic benchmark, it makes the cheap/expensive phase separation + obvious to the reward, which is exactly what we want to teach first. + """ + prediction_window = np.asarray(self.prices.predicted_prices, dtype=np.float32) + future_reference = prediction_window[1:] if prediction_window.size > 1 else prediction_window + if future_reference.size < 2: + return 0.0, 0.0 + + q_low, q_high = np.quantile( + future_reference, + [self.PRICE_QUANTILE_LOW, self.PRICE_QUANTILE_HIGH], + ) + price_band = float(q_high - q_low) + if price_band <= 1e-6: + return 0.0, 0.0 + + normalized = (current_price - float(q_low)) / price_band + cheap_strength = float(np.clip(1.0 - normalized, 0.0, 1.0)) + expensive_strength = float(np.clip(normalized, 0.0, 1.0)) + return cheap_strength, expensive_strength + def _reward_price_legacy(self, current_price: float, average_future_price: float, num_processed_jobs: int) -> float: """Legacy linear reward: preserved for comparison/ablation.""" context_avg = self._price_context_average(average_future_price) @@ -206,32 +235,6 @@ def _penalty_idle_normalized(self, num_idle_nodes: int) -> float: normalized_penalty = -self._normalize(current_penalty, self._min_idle_penalty, self._max_idle_penalty) return float(np.clip(normalized_penalty, -1, 0)) - @staticmethod - def _penalty_job_age(num_off_nodes: int, job_queue_2d: np.ndarray) -> float: - """Calculate saturated penalty for jobs waiting in queue when nodes are off.""" - job_age_penalty = 0.0 - if num_off_nodes > 0: - # Vectorized max age calculation (much faster than Python loop) - # [:, 0] selects column 0 (duration) for all rows; > 0 creates boolean mask - valid_mask = job_queue_2d[:, 0] > 0 - # [valid_mask, 1] selects column 1 (age) only for rows where mask is True - max_age = job_queue_2d[valid_mask, 1].max() if valid_mask.any() else 0 - if max_age > 24: - tau_hours = WEEK_HOURS - max_factor = 1.0 - np.exp(-(2*WEEK_HOURS) / tau_hours) - factor = 1.0 - np.exp(-(max_age-24) / tau_hours) - factor = min(factor / max_factor, 1.0) - job_age_penalty = factor - return job_age_penalty - - def _penalty_job_age_normalized(self, num_off_nodes: int, job_queue_2d: np.ndarray) -> float: - """Calculate normalized job age penalty [-1, 0].""" - current_penalty = self._penalty_job_age(num_off_nodes, job_queue_2d) - # _penalty_job_age already returns [0, 1]; negate to get [-1, 0] - # normalized_penalty = self._normalize(current_penalty, 0, -1) - normalized_penalty = -current_penalty - return float(np.clip(normalized_penalty, -1, 0)) - def _reward_energy_efficiency_normalized(self, num_used_nodes: int, num_idle_nodes: int) -> float: '''Redefine meaning of "efficiency". Use purely as "energy efficiency", aka: How much of the energy (in MW) which is currently needed, gets used for work. NOTE: Original efficiency function was doing 3 things at once. 1. Handled Blackout logic, with (2.) penalty-ish reward delay for unprocessed jobs, while blackout. @@ -303,22 +306,79 @@ def _reward_price_utilization(self, current_price: float, average_future_price: def _blackout_term(self, num_used_nodes: int, num_idle_nodes: int, num_unprocessed_jobs: int) -> float: """ - Reward/penalty for full blackout (all nodes off). - If queue is empty, reward the blackout. If jobs are waiting, apply a smooth penalty in [-1, 0]. + Reward a full blackout only when there truly is no work to do. + + The old implementation punished a queue during blackout immediately, which + collided with the benchmark objective of deferring expensive-hour work. + Deferral pressure now lives in the backlog term below instead of here. """ - BLACKOUT_QUEUE_THRESHOLD = 10 # jobs waiting until penalty saturates to -1 - SATURATION_FACTOR = 2 on_nodes = num_used_nodes + num_idle_nodes if on_nodes != 0: - return 0.0 # only care about full blackout + return 0.0 + + return 1.0 if num_unprocessed_jobs <= 0 else 0.0 - if num_unprocessed_jobs <= 0: - return 1.0 # correct blackout + def _penalty_job_age( + self, + current_price: float, + decision_pending_core_demand: float, + remaining_overdue_age_core_hours: float, + total_used_cores: int, + ) -> float: + """ + Combined backlog pressure term used in the legacy "job age" reward slot. + + It deliberately does two things: + 1. Cheap-hour service pressure: + If cheap compute is available *and* backlog exists, the agent should keep + the cluster busy instead of trickling. + 2. Overdue backlog pressure: + Once jobs have outlived the deferral grace period, leaving them pending + becomes increasingly expensive regardless of price. + """ + cheap_strength, _ = self._price_phase_strengths(current_price) + + cheap_service_shortfall = 0.0 + if cheap_strength > 0.0 and decision_pending_core_demand > 0.0: + step_capacity_cores = float(MAX_NODES * CORES_PER_NODE) + target_service = min(float(decision_pending_core_demand), step_capacity_cores) + achieved_service = min(float(total_used_cores), target_service) + cheap_service_shortfall = cheap_strength * (1.0 - achieved_service / max(target_service, 1e-6)) + + overdue_pressure = 0.0 + if remaining_overdue_age_core_hours > 0.0: + overdue_pressure = 1.0 - np.exp( + -float(remaining_overdue_age_core_hours) / max(self.OVERDUE_AGE_CORE_HOUR_TAU, 1e-6) + ) + + combined_pressure = ( + self.CHEAP_SERVICE_GAIN * cheap_service_shortfall + + self.OVERDUE_BACKLOG_GAIN * overdue_pressure + ) + return float(np.clip(combined_pressure, 0.0, 1.0)) + + def _penalty_job_age_normalized( + self, + current_price: float, + decision_pending_core_demand: float, + remaining_overdue_age_core_hours: float, + total_used_cores: int, + ) -> float: + """Legacy reward slot for backlog pressure, normalized to [-1, 0].""" + current_penalty = self._penalty_job_age( + current_price, + decision_pending_core_demand, + remaining_overdue_age_core_hours, + total_used_cores, + ) + return float(np.clip(-current_penalty, -1.0, 0.0)) - ratio = num_unprocessed_jobs / max(BLACKOUT_QUEUE_THRESHOLD, 1) - penalty = np.exp(-ratio * SATURATION_FACTOR) - 1.0 - return float(np.clip(penalty, -1.0, 0.0)) + def _penalty_drop(self, num_dropped: int) -> float: + """Heavy penalty for jobs dropped this step.""" + if num_dropped <= 0: + return 0.0 + return float(-1.0 - 0.25 * min(num_dropped - 1, 1000)) def _penalty_drop(self, num_dropped: int) -> float: """Drop penalty: tanh saturation curve bounded in [-1, 0].""" @@ -328,7 +388,8 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo num_off_nodes: int, job_queue_2d: np.ndarray, num_unprocessed_jobs: int, weights: Weights, num_dropped_this_step: int, env_print: Callable[..., None], num_on_nodes: int, - total_used_cores: int) -> tuple[float, float, float, float, float, float]: + total_used_cores: int, decision_pending_core_demand: float = 0.0, + remaining_overdue_age_core_hours: float = 0.0) -> tuple[float, float, float, float, float, float]: """ Calculate total reward by aggregating weighted components. @@ -346,6 +407,8 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo env_print: Print function for logging num_on_nodes: Number of powered-on nodes total_used_cores: Total cores in use across all powered nodes + decision_pending_core_demand: Total pending node-core demand before scheduling this step + remaining_overdue_age_core_hours: Post-scheduling overdue-age mass of still-pending jobs Returns: Tuple of (total reward, total cost, eff_reward_norm, price_reward, idle_penalty_norm, job_age_penalty_norm) @@ -362,8 +425,15 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo # legacy: price_reward = self._reward_price_normalized_legacy(current_price, average_future_price, total_used_cores) price_reward_weighted = weights.price_weight * price_reward - # 3. penalize delayed jobs, more if they are older. but only if there are turned off nodes - job_age_penalty_norm = self._penalty_job_age_normalized(num_off_nodes, job_queue_2d) + # 3. Push pending work into cheap hours and punish starving backlog after the grace period. + # The method name is kept for compatibility with existing plots/logs, but the semantics + # now describe backlog pressure rather than a simple "oldest queue age" penalty. + job_age_penalty_norm = self._penalty_job_age_normalized( + current_price, + decision_pending_core_demand, + remaining_overdue_age_core_hours, + total_used_cores, + ) job_age_penalty_weighted = weights.job_age_weight * job_age_penalty_norm # 4. penalty for idling nodes @@ -371,9 +441,10 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo idle_penalty_weighted = weights.idle_weight * idle_penalty_norm # 6. penalty for lost jobs (aged out or rejected because queue/backlog was full) - drop_penalty_weighted = 0 + drop_penalty_weighted = 0.0 if self.ALLOW_DROP_PENALTY and num_dropped_this_step > 0: - drop_penalty_weighted = -1.0 - 0.25 * min(num_dropped_this_step - 1, 1000) # harsher penalty for losing many jobs, capped at -251.0 for 1000+ jobs + # drop_penalty_weighted = weights.drop_weight * self._penalty_drop(num_dropped_this_step) + drop_penalty_weighted = 0.3 * self._penalty_drop(num_dropped_this_step) reward = ( efficiency_reward_weighted diff --git a/src/workloadgen.py b/src/workloadgen.py index 57e4c70..364f86d 100644 --- a/src/workloadgen.py +++ b/src/workloadgen.py @@ -53,13 +53,13 @@ class WorkloadGenConfig: # Burst 1: many small-ish jobs at once burst_small_prob: float = 0.0 burst_small_jobs_min: int = 50 - burst_small_jobs_max: int = 750 + burst_small_jobs_max: int = 1500 burst_small_duration_min: int = 1 - burst_small_duration_max: int = 8 + burst_small_duration_max: int = 2 burst_small_nodes_min: int = 1 - burst_small_nodes_max: int = 2 + burst_small_nodes_max: int = 1 burst_small_cores_min: int = 1 - burst_small_cores_max: int = 16 + burst_small_cores_max: int = 4 # Burst 2: heavy jobs (high duration + high resource demand) burst_heavy_prob: float = 0.0 diff --git a/train.py b/train.py index 05abbac..a6a06e4 100644 --- a/train.py +++ b/train.py @@ -310,13 +310,17 @@ def main(): # Calculate job metrics across all episodes total_jobs_submitted = sum(ep['jobs_submitted'] for ep in env.metrics.episode_costs) + total_jobs_launched = sum(int(ep.get('jobs_launched', ep['jobs_completed'])) for ep in env.metrics.episode_costs) total_jobs_completed = sum(ep['jobs_completed'] for ep in env.metrics.episode_costs) total_baseline_submitted = sum(ep['baseline_jobs_submitted'] for ep in env.metrics.episode_costs) + total_baseline_launched = sum(int(ep.get('baseline_jobs_launched', ep['baseline_jobs_completed'])) for ep in env.metrics.episode_costs) total_baseline_completed = sum(ep['baseline_jobs_completed'] for ep in env.metrics.episode_costs) - avg_wait_time = sum(ep['avg_wait_time'] * ep['jobs_completed'] for ep in env.metrics.episode_costs) / total_jobs_completed if total_jobs_completed > 0 else 0 - avg_baseline_wait_time = sum(ep['baseline_avg_wait_time'] * ep['baseline_jobs_completed'] for ep in env.metrics.episode_costs) / total_baseline_completed if total_baseline_completed > 0 else 0 + avg_wait_time = sum(ep['avg_wait_time'] * int(ep.get('jobs_launched', ep['jobs_completed'])) for ep in env.metrics.episode_costs) / total_jobs_launched if total_jobs_launched > 0 else 0 + avg_baseline_wait_time = sum(ep['baseline_avg_wait_time'] * int(ep.get('baseline_jobs_launched', ep['baseline_jobs_completed'])) for ep in env.metrics.episode_costs) / total_baseline_launched if total_baseline_launched > 0 else 0 avg_max_queue = sum(ep['max_queue_size'] for ep in env.metrics.episode_costs) / len(env.metrics.episode_costs) avg_baseline_max_queue = sum(ep['baseline_max_queue_size'] for ep in env.metrics.episode_costs) / len(env.metrics.episode_costs) + avg_pending_jobs_end = sum(int(ep.get('pending_jobs_end', 0)) for ep in env.metrics.episode_costs) / len(env.metrics.episode_costs) + avg_overdue_jobs_end = sum(int(ep.get('overdue_jobs_end', 0)) for ep in env.metrics.episode_costs) / len(env.metrics.episode_costs) total_agent_cost = sum(float(ep['agent_cost']) for ep in env.metrics.episode_costs) total_baseline_cost = sum(float(ep['baseline_cost']) for ep in env.metrics.episode_costs) total_baseline_off_cost = sum(float(ep['baseline_cost_off']) for ep in env.metrics.episode_costs) @@ -358,15 +362,19 @@ def main(): ) if arrivals_per_hour_by_episode else 0.0 std_arrivals_per_hour = arrivals_variance ** 0.5 - print("\n=== JOB PROCESSING METRICS ===") - print("\nAgent:") + print(f"\n=== JOB PROCESSING METRICS ===") + print(f"\nAgent:") + print(f" Jobs Launched: {total_jobs_launched:,} / {total_jobs_submitted:,}") print(f" Jobs Completed: {total_jobs_completed:,} / {total_jobs_submitted:,} ({total_agent_completion_rate:.1f}%)") print(f" Average Wait Time: {avg_wait_time:.1f} hours") print(f" Average Max Queue Size: {avg_max_queue:.0f}") + print(f" Average Pending Jobs At Episode End: {avg_pending_jobs_end:.1f}") + print(f" Average Overdue Jobs At Episode End: {avg_overdue_jobs_end:.1f}") print(f" Total Cost: €{total_agent_cost:,.0f}") print(f" Job Arrivals/Hour (mean ± std): {mean_arrivals_per_hour:.2f} ± {std_arrivals_per_hour:.2f}") - print("\nBaseline:") + print(f"\nBaseline:") + print(f" Jobs Launched: {total_baseline_launched:,} / {total_baseline_submitted:,}") print(f" Jobs Completed: {total_baseline_completed:,} / {total_baseline_submitted:,} ({total_baseline_completion_rate:.1f}%)") print(f" Average Wait Time: {avg_baseline_wait_time:.1f} hours") print(f" Average Max Queue Size: {avg_baseline_max_queue:.0f}") From db4897d8c004cc19bb3bdf4b1862912bc99cb4c5 Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Wed, 8 Apr 2026 15:30:27 +0200 Subject: [PATCH 5/9] Trigger episode-end flush after sustained drop streak Add an optional recovery path that flushes outstanding work only when the agent has been dropping jobs for a configured number of consecutive steps. - add --flush-after-drop-streak CLI option - track consecutive dropped-job steps in the environment - flush queue, backlog, and running jobs at episode end once the streak threshold is reached - apply a single flush penalty to the terminal step reward - record flushed jobs separately in metrics and include them in total lost jobs - add regression coverage for both triggered and non-triggered flush cases --- src/callbacks.py | 88 ++++++++++++---------- src/environment.py | 150 ++++++++++++++++++++++++++++++++++++-- src/metrics_tracker.py | 89 ++++++++++++++++++++-- src/reward_calculation.py | 13 ++-- train.py | 55 +++++++++----- train_iter.py | 17 ++++- 6 files changed, 335 insertions(+), 77 deletions(-) diff --git a/src/callbacks.py b/src/callbacks.py index 87c8062..ef76154 100644 --- a/src/callbacks.py +++ b/src/callbacks.py @@ -1,4 +1,4 @@ -from src.config import EPISODE_HOURS, MAX_QUEUE_SIZE, MAX_NODES +from src.config import EPISODE_HOURS, MAX_QUEUE_SIZE, COST_IDLE_MW, COST_USED_MW, CORES_PER_NODE, MAX_NODES from stable_baselines3.common.callbacks import BaseCallback @@ -25,22 +25,12 @@ def _on_rollout_start(self) -> None: def _on_step(self) -> bool: env = self.training_env.envs[0].unwrapped if env.metrics.current_hour == EPISODE_HOURS-1: - self.logger.record("metrics/total_reward", env.metrics.episode_reward) - self.logger.record("metrics/reward_eff", sum(env.metrics.episode_eff_rewards) / 100) - self.logger.record("metrics/reward_price", sum(env.metrics.episode_price_rewards) / 100) - self.logger.record("metrics/penalty_idle", sum(env.metrics.episode_idle_penalties) / 100) - self.logger.record("metrics/penalty_job_age", sum(env.metrics.episode_job_age_penalties) / 100) - self.logger.record("metrics/penalty_drop", sum(env.metrics.episode_drop_penalties)) - self.logger.record("metrics/cost", env.metrics.episode_total_cost) self.logger.record("metrics/savings", env.metrics.episode_baseline_cost - env.metrics.episode_total_cost) - savings_off = env.metrics.episode_baseline_cost_off - env.metrics.episode_total_cost - self.logger.record("metrics/savings_off", savings_off) - savings_off_clean = savings_off if env.metrics.episode_jobs_dropped == 0 else 0.0 - self.logger.record("metrics/savings_off_clean", savings_off_clean) + self.logger.record("metrics/savings_off", env.metrics.episode_baseline_cost_off - env.metrics.episode_total_cost) #self.logger.record("metrics/queue_fill_pct", env.metrics.episode_max_queue_size_reached / MAX_QUEUE_SIZE * 100) - self.logger.record("metrics/bl_cost", env.metrics.episode_baseline_cost) - self.logger.record("metrics/bl_cost_off", env.metrics.episode_baseline_cost_off) + self.logger.record("metrics/baseline_cost", env.metrics.episode_baseline_cost) + self.logger.record("metrics/baseline_cost_off", env.metrics.episode_baseline_cost_off) # Job metrics (agent) completion_rate = (env.metrics.episode_jobs_completed / env.metrics.episode_jobs_submitted * 100 if env.metrics.episode_jobs_submitted > 0 else 0.0) @@ -53,18 +43,18 @@ def _on_step(self) -> bool: self.logger.record("metrics/jobs_completed", env.metrics.episode_jobs_completed) self.logger.record("metrics/completion_rate", completion_rate) self.logger.record("metrics/avg_wait_hours", avg_wait) - self.logger.record("metrics/nodes_on", env.metrics.episode_on_nodes[-1]) - self.logger.record("metrics/nodes_used", env.metrics.episode_used_nodes[-1]) - self.logger.record("metrics/nodes_idle", env.metrics.episode_on_nodes[-1] - env.metrics.episode_used_nodes[-1]) - self.logger.record("metrics/nodes_off", MAX_NODES - env.metrics.episode_on_nodes[-1]) + self.logger.record("metrics/on_nodes", env.metrics.episode_on_nodes[-1]) + self.logger.record("metrics/used_nodes", env.metrics.episode_used_nodes[-1]) self.logger.record("metrics/max_queue_size", env.metrics.episode_max_queue_size_reached) self.logger.record("metrics/max_backlog_size", env.metrics.episode_max_backlog_size_reached) self.logger.record("metrics/pending_jobs_end", env.metrics.episode_pending_jobs_end) self.logger.record("metrics/pending_core_hours_end", env.metrics.episode_pending_core_hours_end) self.logger.record("metrics/overdue_jobs_end", env.metrics.episode_overdue_jobs_end) + jobs_lost_total = env.metrics.episode_jobs_dropped + env.metrics.episode_jobs_flushed self.logger.record("metrics/jobs_dropped", env.metrics.episode_jobs_dropped) - self.logger.record("metrics/jobs_lost_total", env.metrics.episode_jobs_dropped) - loss_rate = (env.metrics.episode_jobs_dropped / env.metrics.episode_jobs_submitted * 100 if env.metrics.episode_jobs_submitted > 0 else 0.0) + self.logger.record("metrics/jobs_flushed", env.metrics.episode_jobs_flushed) + self.logger.record("metrics/jobs_lost_total", jobs_lost_total) + loss_rate = (jobs_lost_total / env.metrics.episode_jobs_submitted * 100 if env.metrics.episode_jobs_submitted > 0 else 0.0) self.logger.record("metrics/loss_rate", loss_rate) self.logger.record("metrics/jobs_rejected_queue_full", env.metrics.episode_jobs_rejected_queue_full) @@ -74,24 +64,48 @@ def _on_step(self) -> bool: env.metrics.episode_baseline_total_job_wait_time_launch / env.metrics.episode_baseline_jobs_launched if env.metrics.episode_baseline_jobs_launched > 0 else 0.0 ) - self.logger.record("metrics/bl_jobs_submitted", env.metrics.episode_baseline_jobs_submitted) - self.logger.record("metrics/bl_jobs_launched", env.metrics.episode_baseline_jobs_launched) - self.logger.record("metrics/bl_jobs_completed", env.metrics.episode_baseline_jobs_completed) - self.logger.record("metrics/bl_completion_rate", baseline_completion_rate) - self.logger.record("metrics/bl_avg_wait_hours", baseline_avg_wait) - self.logger.record("metrics/bl_max_queue_size", env.metrics.episode_baseline_max_queue_size_reached) - self.logger.record("metrics/bl_max_backlog_size", env.metrics.episode_baseline_max_backlog_size_reached) - self.logger.record("metrics/bl_jobs_dropped", env.metrics.episode_baseline_jobs_dropped) - self.logger.record("metrics/bl_jobs_lost_total", env.metrics.episode_baseline_jobs_dropped) - baseline_loss_rate = (env.metrics.episode_baseline_jobs_dropped / env.metrics.episode_baseline_jobs_submitted * 100 if env.metrics.episode_baseline_jobs_submitted > 0 else 0.0) - self.logger.record("metrics/bl_loss_rate", baseline_loss_rate) - self.logger.record("metrics/bl_jobs_rejected_queue_full", env.metrics.episode_baseline_jobs_rejected_queue_full) + self.logger.record("metrics/baseline_jobs_submitted", env.metrics.episode_baseline_jobs_submitted) + self.logger.record("metrics/baseline_jobs_launched", env.metrics.episode_baseline_jobs_launched) + self.logger.record("metrics/baseline_jobs_completed", env.metrics.episode_baseline_jobs_completed) + self.logger.record("metrics/baseline_completion_rate", baseline_completion_rate) + self.logger.record("metrics/baseline_avg_wait_hours", baseline_avg_wait) + self.logger.record("metrics/baseline_max_queue_size", env.metrics.episode_baseline_max_queue_size_reached) + self.logger.record("metrics/baseline_max_backlog_size", env.metrics.episode_baseline_max_backlog_size_reached) + baseline_jobs_lost_total = env.metrics.episode_baseline_jobs_dropped + env.metrics.episode_baseline_jobs_flushed + self.logger.record("metrics/baseline_jobs_dropped", env.metrics.episode_baseline_jobs_dropped) + self.logger.record("metrics/baseline_jobs_flushed", env.metrics.episode_baseline_jobs_flushed) + self.logger.record("metrics/baseline_jobs_lost_total", baseline_jobs_lost_total) + baseline_loss_rate = (baseline_jobs_lost_total / env.metrics.episode_baseline_jobs_submitted * 100 if env.metrics.episode_baseline_jobs_submitted > 0 else 0.0) + self.logger.record("metrics/baseline_loss_rate", baseline_loss_rate) + self.logger.record("metrics/baseline_jobs_rejected_queue_full", env.metrics.episode_baseline_jobs_rejected_queue_full) - # Power metrics - self.logger.record("metrics/power_mwh", env.metrics.episode_total_power_consumption_mwh) - self.logger.record("metrics/bl_power_mwh", env.metrics.episode_baseline_power_consumption_mwh) - self.logger.record("metrics/bl_off_power_mwh", env.metrics.episode_baseline_power_consumption_off_mwh) - self.logger.record("metrics/savings_power_vs_baseline_off", env.metrics.episode_baseline_power_consumption_off_mwh - env.metrics.episode_total_power_consumption_mwh) + # Proportional (per-core) power metrics + _delta = COST_USED_MW - COST_IDLE_MW + agent_prop_power = sum( + COST_IDLE_MW * on + _delta * (cores / CORES_PER_NODE) + for on, cores in zip(env.metrics.episode_on_nodes, env.metrics.episode_used_cores) + ) + baseline_prop_power = sum( + COST_IDLE_MW * MAX_NODES + _delta * (cores / CORES_PER_NODE) + for cores in env.metrics.episode_baseline_used_cores + ) + baseline_off_prop_power = sum( + COST_IDLE_MW * used + _delta * (cores / CORES_PER_NODE) + for used, cores in zip(env.metrics.episode_baseline_used_nodes, env.metrics.episode_baseline_used_cores) + ) + self.logger.record("metrics/prop_power_mwh", agent_prop_power) + self.logger.record("metrics/baseline_prop_power_mwh", baseline_prop_power) + self.logger.record("metrics/baseline_off_prop_power_mwh", baseline_off_prop_power) + self.logger.record("metrics/savings_prop_power_vs_baseline_off", baseline_off_prop_power - agent_prop_power) + agent_prop_cost = sum( + (COST_IDLE_MW * on + _delta * (cores / CORES_PER_NODE)) * price + for on, cores, price in zip(env.metrics.episode_on_nodes, env.metrics.episode_used_cores, env.metrics.episode_price_stats) + ) + baseline_off_prop_cost = sum( + (COST_IDLE_MW * used + _delta * (cores / CORES_PER_NODE)) * price + for used, cores, price in zip(env.metrics.episode_baseline_used_nodes, env.metrics.episode_baseline_used_cores, env.metrics.episode_price_stats) + ) + self.logger.record("metrics/savings_prop_cost_vs_baseline_off", baseline_off_prop_cost - agent_prop_cost) return True diff --git a/src/environment.py b/src/environment.py index f2f6aef..2f82f27 100644 --- a/src/environment.py +++ b/src/environment.py @@ -20,7 +20,7 @@ from src.config import ( MAX_NODES, MAX_QUEUE_SIZE, MAX_CHANGE, MAX_JOB_DURATION, CORES_PER_NODE, MAX_CORES_PER_JOB, MAX_JOB_AGE_OBS, - MAX_NODES_PER_JOB, EPISODE_HOURS, PENALTY_DROPPED_JOB + MAX_NODES_PER_JOB, EPISODE_HOURS ) from src.job_management import ( process_ongoing_jobs, add_new_jobs, @@ -75,7 +75,9 @@ def __init__(self, workload_gen: WorkloadGenerator | None = None, job_arrival_scale: float = 1.0, jobs_exact_replay: bool = False, - output_dir: str = "sessions") -> None: + output_dir: str = "sessions", + jobs_exact_replay_aggregate: bool = False, + flush_after_drop_streak: int = 0) -> None: super().__init__() self.weights = weights @@ -90,6 +92,10 @@ def __init__(self, self.evaluation_mode = evaluation_mode self.job_arrival_scale = float(job_arrival_scale) self.jobs_exact_replay = bool(jobs_exact_replay) + self.jobs_exact_replay_aggregate = bool(jobs_exact_replay_aggregate) + self.flush_after_drop_streak = max(0, int(flush_after_drop_streak)) + self.consecutive_drop_steps = 0 + self.flush_pending_for_episode = False self.next_plot_save = self.steps_per_iteration @@ -118,7 +124,10 @@ def __init__(self, print(f"Parsed aggregated jobs for {len(self.jobs_sampler.aggregated_jobs)} hours") if self.jobs_exact_replay: max_raw_jobs = max((len(v) for v in self.jobs_sampler.jobs.values()), default=0) - print("Jobs replay mode: exact timeline (raw jobs per hour)") + if self.jobs_exact_replay_aggregate: + print("Jobs replay mode: exact timeline (aggregated per step)") + else: + print("Jobs replay mode: exact timeline (raw jobs per hour)") print(f"Max raw jobs per hour: {max_raw_jobs}") else: self.jobs_sampler.precalculate_hourly_jobs(CORES_PER_NODE, MAX_NODES_PER_JOB) @@ -182,6 +191,8 @@ def __init__(self, def _reset_timeline_state(self, start_index: int) -> None: self.prices.reset(start_index=start_index) + self.consecutive_drop_steps = 0 + self.flush_pending_for_episode = False self.state = { # Initialize all nodes to be 'online but free' (0) @@ -308,6 +319,88 @@ def _update_pending_job_stats(self, job_queue_2d: np.ndarray) -> None: # Cache the queue/backlog version for next step. self._cached_queue_backlog_version = self._queue_backlog_version + @staticmethod + def _count_queued_jobs(job_queue_2d: np.ndarray) -> int: + """Count active jobs represented in a dense queue array.""" + return int(np.count_nonzero(job_queue_2d[:, 0] > 0)) + + def _flush_workload_side( + self, + job_queue_2d: np.ndarray, + nodes: np.ndarray, + cores_available: np.ndarray, + running_jobs: dict[int, dict[str, Any]], + backlog_queue: deque, + ) -> int: + """ + Drop all outstanding work for one side of the simulation and reset its cluster. + + This is intentionally stronger than a normal episode rollover: queued jobs, + backlog jobs, and still-running jobs are all considered lost so the next + episode can start from a clean slate. + """ + flushed_jobs = ( + self._count_queued_jobs(job_queue_2d) + + len(backlog_queue) + + len(running_jobs) + ) + + job_queue_2d.fill(0) + nodes.fill(0) + cores_available.fill(CORES_PER_NODE) + running_jobs.clear() + backlog_queue.clear() + return int(flushed_jobs) + + def _flush_episode_state(self) -> dict[str, int | float | bool]: + """Flush both agent and baseline states after the terminal step of an episode.""" + agent_job_queue_2d = self.state['job_queue'].reshape(-1, 4) + baseline_job_queue_2d = self.baseline_state['job_queue'].reshape(-1, 4) + + agent_jobs_flushed = self._flush_workload_side( + agent_job_queue_2d, + self.state['nodes'], + self.cores_available, + self.running_jobs, + self.backlog_queue, + ) + baseline_jobs_flushed = self._flush_workload_side( + baseline_job_queue_2d, + self.baseline_state['nodes'], + self.baseline_cores_available, + self.baseline_running_jobs, + self.baseline_backlog_queue, + ) + + self.next_empty_slot = 0 + self.baseline_next_empty_slot = 0 + self.metrics.current_running_jobs = 0 + + if agent_jobs_flushed > 0: + self.metrics.jobs_flushed += agent_jobs_flushed + self.metrics.episode_jobs_flushed += agent_jobs_flushed + if baseline_jobs_flushed > 0: + self.metrics.baseline_jobs_flushed += baseline_jobs_flushed + self.metrics.episode_baseline_jobs_flushed += baseline_jobs_flushed + + flush_penalty = self.reward_calculator.loss_penalty(agent_jobs_flushed) + + if agent_jobs_flushed > 0 or baseline_jobs_flushed > 0: + self._mark_queue_backlog_mutation() + self._update_pending_job_stats(agent_job_queue_2d) + self.consecutive_drop_steps = 0 + self.flush_pending_for_episode = False + + self.state['job_queue'] = agent_job_queue_2d.flatten() + self.baseline_state['job_queue'] = baseline_job_queue_2d.flatten() + + return { + "flush_applied": bool(agent_jobs_flushed > 0 or baseline_jobs_flushed > 0), + "agent_jobs_flushed": agent_jobs_flushed, + "baseline_jobs_flushed": baseline_jobs_flushed, + "flush_penalty": float(flush_penalty), + } + def _resolve_reset_start_index(self, options: dict[str, Any]) -> int: """ Choose the initial price position for a true timeline reset. @@ -575,10 +668,18 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, self.metrics.episode_price_rewards.append(price_reward * 100) self.metrics.episode_job_age_penalties.append(job_age_penalty_norm * 100) self.metrics.episode_idle_penalties.append(idle_penalty_norm * 100) - self.metrics.episode_drop_penalties.append(PENALTY_DROPPED_JOB * num_dropped_this_step) self.metrics.episode_rewards.append(step_reward) self.metrics.jobs_dropped += num_dropped_this_step self.metrics.episode_jobs_dropped += num_dropped_this_step + if num_dropped_this_step > 0: + self.consecutive_drop_steps += 1 + else: + self.consecutive_drop_steps = 0 + if ( + self.flush_after_drop_streak > 0 + and self.consecutive_drop_steps >= self.flush_after_drop_streak + ): + self.flush_pending_for_episode = True # print stats self.env_print(f"[6] End of step stats...") @@ -595,6 +696,13 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, truncated = False terminated = False + flush_applied = False + flush_penalty = 0.0 + agent_jobs_flushed = 0 + baseline_jobs_flushed = 0 + drop_streak_steps = self.consecutive_drop_steps + flush_armed_by_drop_streak = bool(self.flush_pending_for_episode) + episode_flush_triggered_by_drop_streak = False if self.metrics.current_hour == EPISODE_HOURS: if self.render_mode == 'human': plot_episode(self, EPISODE_HOURS, MAX_NODES, False, True, self.current_step) @@ -609,6 +717,30 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, truncated = True terminated = False + if flush_armed_by_drop_streak: + episode_flush_triggered_by_drop_streak = True + flush_result = self._flush_episode_state() + flush_applied = bool(flush_result["flush_applied"]) + flush_penalty = float(flush_result["flush_penalty"]) + agent_jobs_flushed = int(flush_result["agent_jobs_flushed"]) + baseline_jobs_flushed = int(flush_result["baseline_jobs_flushed"]) + if flush_penalty != 0.0: + step_reward += 10*flush_penalty + self.metrics.episode_reward += flush_penalty + if self.metrics.rewards: + self.metrics.rewards[-1] += flush_penalty + if self.metrics.episode_rewards: + self.metrics.episode_rewards[-1] += flush_penalty + if flush_applied: + self.env_print( + f"[flush] Drop streak {self.flush_after_drop_streak}+ reached " + f"({drop_streak_steps} steps). " + f"Dropped outstanding work at episode boundary: " + f"agent={agent_jobs_flushed}, baseline={baseline_jobs_flushed}, penalty={flush_penalty:.4f}" + ) + else: + self.flush_pending_for_episode = False + # Record episode costs for long-term analysis self.metrics.record_episode_completion(self.current_episode) @@ -627,7 +759,15 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, "num_unprocessed_jobs": num_unprocessed_jobs, "num_on_nodes": num_on_nodes, "episode_jobs_dropped": self.metrics.episode_jobs_dropped, - "episode_jobs_lost_total": self.metrics.episode_jobs_dropped, + "episode_jobs_flushed": self.metrics.episode_jobs_flushed, + "episode_jobs_lost_total": self.metrics.episode_jobs_dropped + self.metrics.episode_jobs_flushed, + "episode_flush_applied": flush_applied, + "episode_flush_triggered_by_drop_streak": episode_flush_triggered_by_drop_streak, + "drop_streak_steps": drop_streak_steps, + "drop_streak_flush_armed": flush_armed_by_drop_streak, + "step_flush_penalty": flush_penalty, + "step_jobs_flushed": agent_jobs_flushed, + "step_baseline_jobs_flushed": baseline_jobs_flushed, } return self.state, step_reward, terminated, truncated, info diff --git a/src/metrics_tracker.py b/src/metrics_tracker.py index b2fb571..1484066 100644 --- a/src/metrics_tracker.py +++ b/src/metrics_tracker.py @@ -1,5 +1,7 @@ """Metrics tracking and episode recording for the PowerSched environment.""" +from src.config import COST_IDLE_MW, COST_USED_MW, CORES_PER_NODE, MAX_NODES + class MetricsTracker: """Tracks metrics throughout training episodes.""" @@ -14,6 +16,11 @@ def _safe_ratio(numerator: float, denominator: float) -> float: """Safe division; returns NaN when denominator is not positive.""" return (numerator / denominator) if denominator > 0.0 else float("nan") + @staticmethod + def _jobs_lost_total(dropped: int, flushed: int) -> int: + """Total lost jobs, including regular drops and episode-end flushes.""" + return int(dropped) + int(flushed) + def __init__(self) -> None: """Initialize all metric counters.""" self.reset_timeline_metrics() @@ -44,6 +51,7 @@ def reset_timeline_metrics(self) -> None: self.max_queue_size_reached: int = 0 self.max_backlog_size_reached: int = 0 self.jobs_dropped: int = 0 + self.jobs_flushed: int = 0 self.jobs_rejected_queue_full: int = 0 # Baseline job metrics (cumulative across episodes) @@ -55,6 +63,7 @@ def reset_timeline_metrics(self) -> None: self.baseline_max_queue_size_reached: int = 0 self.baseline_max_backlog_size_reached: int = 0 self.baseline_jobs_dropped: int = 0 + self.baseline_jobs_flushed: int = 0 self.baseline_jobs_rejected_queue_full: int = 0 # Time series data for plotting (cumulative) @@ -89,6 +98,7 @@ def reset_episode_metrics(self) -> None: self.episode_max_queue_size_reached: int = 0 self.episode_max_backlog_size_reached: int = 0 self.episode_jobs_dropped: int = 0 + self.episode_jobs_flushed: int = 0 self.episode_jobs_rejected_queue_full: int = 0 # Baseline job metrics (episode) @@ -100,6 +110,7 @@ def reset_episode_metrics(self) -> None: self.episode_baseline_max_queue_size_reached: int = 0 self.episode_baseline_max_backlog_size_reached: int = 0 self.episode_baseline_jobs_dropped: int = 0 + self.episode_baseline_jobs_flushed: int = 0 self.episode_baseline_jobs_rejected_queue_full: int = 0 # End-of-episode pending-work snapshot. @@ -124,7 +135,6 @@ def reset_episode_metrics(self) -> None: self.episode_price_rewards: list[float] = [] self.episode_idle_penalties: list[float] = [] self.episode_job_age_penalties: list[float] = [] - self.episode_drop_penalties: list[float] = [] self.episode_rewards: list[float] = [] self.episode_running_jobs_counts: list[int] = [] @@ -168,13 +178,36 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i if self.episode_jobs_submitted else 0.0 ) + jobs_lost_total = self._jobs_lost_total(self.episode_jobs_dropped, self.episode_jobs_flushed) + flush_rate: float = ( + (self.episode_jobs_flushed / self.episode_jobs_submitted * 100) + if self.episode_jobs_submitted + else 0.0 + ) + loss_rate: float = ( + (jobs_lost_total / self.episode_jobs_submitted * 100) + if self.episode_jobs_submitted + else 0.0 + ) baseline_drop_rate: float = ( (self.episode_baseline_jobs_dropped / self.episode_baseline_jobs_submitted * 100) if self.episode_baseline_jobs_submitted else 0.0 ) - loss_rate = drop_rate - baseline_loss_rate = baseline_drop_rate + baseline_jobs_lost_total = self._jobs_lost_total( + self.episode_baseline_jobs_dropped, + self.episode_baseline_jobs_flushed, + ) + baseline_flush_rate: float = ( + (self.episode_baseline_jobs_flushed / self.episode_baseline_jobs_submitted * 100) + if self.episode_baseline_jobs_submitted + else 0.0 + ) + baseline_loss_rate: float = ( + (baseline_jobs_lost_total / self.episode_baseline_jobs_submitted * 100) + if self.episode_baseline_jobs_submitted + else 0.0 + ) agent_mean_price: float = self._effective_mean_price( self.episode_total_cost, self.episode_total_power_consumption_mwh ) @@ -197,6 +230,38 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i savings_vs_baseline: float = self.episode_baseline_cost - self.episode_total_cost savings_vs_baseline_off: float = self.episode_baseline_cost_off - self.episode_total_cost + # Proportional (per-core) power: idle_base for all on-nodes + compute delta scaled by core utilization. + # Formula per step: COST_IDLE_MW * num_on + (COST_USED_MW - COST_IDLE_MW) * (cores_used / CORES_PER_NODE) + _compute_delta_mw = COST_USED_MW - COST_IDLE_MW + agent_prop_power_mwh: float = sum( + COST_IDLE_MW * on + _compute_delta_mw * (cores / CORES_PER_NODE) + for on, cores in zip(self.episode_on_nodes, self.episode_used_cores) + ) + # Baseline always has all MAX_NODES on + baseline_prop_power_mwh: float = sum( + COST_IDLE_MW * MAX_NODES + _compute_delta_mw * (cores / CORES_PER_NODE) + for cores in self.episode_baseline_used_cores + ) + # Baseline_off: only used nodes are on (no idle nodes) + baseline_off_prop_power_mwh: float = sum( + COST_IDLE_MW * used + _compute_delta_mw * (cores / CORES_PER_NODE) + for used, cores in zip(self.episode_baseline_used_nodes, self.episode_baseline_used_cores) + ) + # Proportional cost: same as prop power but multiplied by price at each step + agent_prop_cost: float = sum( + (COST_IDLE_MW * on + _compute_delta_mw * (cores / CORES_PER_NODE)) * price + for on, cores, price in zip(self.episode_on_nodes, self.episode_used_cores, self.episode_price_stats) + ) + baseline_prop_cost: float = sum( + (COST_IDLE_MW * MAX_NODES + _compute_delta_mw * (cores / CORES_PER_NODE)) * price + for cores, price in zip(self.episode_baseline_used_cores, self.episode_price_stats) + ) + baseline_off_prop_cost: float = sum( + (COST_IDLE_MW * used + _compute_delta_mw * (cores / CORES_PER_NODE)) * price + for used, cores, price in zip(self.episode_baseline_used_nodes, self.episode_baseline_used_cores, self.episode_price_stats) + ) + savings_prop_cost_vs_baseline: float = baseline_prop_cost - agent_prop_cost + savings_prop_cost_vs_baseline_off: float = baseline_off_prop_cost - agent_prop_cost dropped_jobs_per_saved_euro: float = self._safe_ratio( float(self.episode_jobs_dropped), savings_vs_baseline ) if savings_vs_baseline > 0.0 else float("nan") @@ -212,6 +277,14 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i 'agent_power_consumption_mwh': self.episode_total_power_consumption_mwh, 'baseline_power_consumption_mwh': self.episode_baseline_power_consumption_mwh, 'baseline_power_consumption_off_mwh': self.episode_baseline_power_consumption_off_mwh, + 'agent_prop_power_mwh': agent_prop_power_mwh, + 'baseline_prop_power_mwh': baseline_prop_power_mwh, + 'baseline_off_prop_power_mwh': baseline_off_prop_power_mwh, + 'agent_prop_cost': agent_prop_cost, + 'baseline_prop_cost': baseline_prop_cost, + 'baseline_off_prop_cost': baseline_off_prop_cost, + 'savings_prop_cost_vs_baseline': savings_prop_cost_vs_baseline, + 'savings_prop_cost_vs_baseline_off': savings_prop_cost_vs_baseline_off, 'agent_mean_price': agent_mean_price, 'baseline_mean_price': baseline_mean_price, 'baseline_off_mean_price': baseline_off_mean_price, @@ -246,15 +319,19 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i 'baseline_completion_rate': baseline_completion_rate, 'baseline_max_queue_size': self.episode_baseline_max_queue_size_reached, 'baseline_max_backlog_size': self.episode_baseline_max_backlog_size_reached, - # Loss metrics: includes age expirations and queue-full rejections. + # Loss metrics: includes age expirations, queue-full rejections, and episode-end flushes. "jobs_dropped": self.episode_jobs_dropped, - "jobs_lost_total": self.episode_jobs_dropped, + "jobs_flushed": self.episode_jobs_flushed, + "jobs_lost_total": jobs_lost_total, "drop_rate": drop_rate, + "flush_rate": flush_rate, "loss_rate": loss_rate, "jobs_rejected_queue_full": self.episode_jobs_rejected_queue_full, "baseline_jobs_dropped": self.episode_baseline_jobs_dropped, - "baseline_jobs_lost_total": self.episode_baseline_jobs_dropped, + "baseline_jobs_flushed": self.episode_baseline_jobs_flushed, + "baseline_jobs_lost_total": baseline_jobs_lost_total, "baseline_drop_rate": baseline_drop_rate, + "baseline_flush_rate": baseline_flush_rate, "baseline_loss_rate": baseline_loss_rate, "baseline_jobs_rejected_queue_full": self.episode_baseline_jobs_rejected_queue_full, } diff --git a/src/reward_calculation.py b/src/reward_calculation.py index 2598c0d..823b483 100644 --- a/src/reward_calculation.py +++ b/src/reward_calculation.py @@ -380,9 +380,14 @@ def _penalty_drop(self, num_dropped: int) -> float: return 0.0 return float(-1.0 - 0.25 * min(num_dropped - 1, 1000)) - def _penalty_drop(self, num_dropped: int) -> float: + def _penalty_drop_alt(self, num_dropped: int) -> float: """Drop penalty: tanh saturation curve bounded in [-1, 0].""" return -float(np.tanh(num_dropped / self.DROP_PENALTY_TAU)) + def loss_penalty(self, num_lost: int) -> float: + """Weighted penalty for jobs lost this step, including episode-end flushes.""" + if not self.ALLOW_DROP_PENALTY or num_lost <= 0: + return 0.0 + return 0.3 * self._penalty_drop(num_lost) def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: float, average_future_price: float, num_off_nodes: int, job_queue_2d: np.ndarray, @@ -441,10 +446,8 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo idle_penalty_weighted = weights.idle_weight * idle_penalty_norm # 6. penalty for lost jobs (aged out or rejected because queue/backlog was full) - drop_penalty_weighted = 0.0 - if self.ALLOW_DROP_PENALTY and num_dropped_this_step > 0: - # drop_penalty_weighted = weights.drop_weight * self._penalty_drop(num_dropped_this_step) - drop_penalty_weighted = 0.3 * self._penalty_drop(num_dropped_this_step) + # Episode-end flushes reuse the same penalty path via loss_penalty(). + drop_penalty_weighted = self.loss_penalty(num_dropped_this_step) reward = ( efficiency_reward_weighted diff --git a/train.py b/train.py index a6a06e4..0ae82db 100644 --- a/train.py +++ b/train.py @@ -38,7 +38,7 @@ def fmt_optional(value: float | None, precision: int = 2, thousands: bool = Fals return f"{value:,.{precision}f}" if thousands else f"{value:.{precision}f}" -STEPS_PER_ITERATION = 100000 +STEPS_PER_ITERATION = 10000 def main(): @@ -52,6 +52,7 @@ def main(): 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('--job-arrival-scale', type=float, default=1.0, help='Scale sampled arrivals per step (1.0 = unchanged).') parser.add_argument('--jobs-exact-replay', action='store_true', help='For --jobs mode, replay raw jobs in timeline order (no template aggregation).') + parser.add_argument('--jobs-exact-replay-aggregate', action='store_true', help='With --jobs-exact-replay, aggregate each sampled raw time-bin before enqueueing.') parser.add_argument('--plot-rewards', action='store_true', help='Per step, plot rewards for all possible num_idle_nodes & num_used_nodes (default: False).') parser.add_argument('--plot-eff-reward', action=argparse.BooleanOptionalAction, default=False, help='Include efficiency reward in the plot (dashed line).') parser.add_argument('--plot-price-reward', action=argparse.BooleanOptionalAction, default=False, help='Include price reward in the plot (dashed line).') @@ -76,6 +77,7 @@ def main(): add_workloadgen_args(parser) parser.add_argument("--plot-dashboard", action="store_true", help="Generate dashboard plot (per-hour panels + cumulative savings).") parser.add_argument("--dashboard-hours", type=int, default=24*14, help="Hours to show in dashboard time-series panels (default: 336).") + parser.add_argument("--dashboard-interval", type=int, default=10000, help="Hours between dashboard plots (default: 10000).") parser.add_argument("--model", type=int, default=None, help="Load a specific model by timestep number (e.g. 5000000 loads 5000000.zip).") parser.add_argument("--net-arch", type=str, default="64,64", help="Hidden layer sizes for policy and value networks (comma-separated, e.g., '256,128' or '512,256,128')") parser.add_argument("--device", type=str, default="auto", help="Device for training: 'auto' (default, uses CUDA if available), 'cuda', 'cpu'") @@ -83,6 +85,12 @@ def main(): parser.add_argument("--seed-sweep", action="store_true", help="Treat this run as part of a --seeds sweep and isolate outputs under a seed-specific session subdirectory.") parser.add_argument("--print-policy", action="store_true", help="Print structure of the policy network.") parser.add_argument("--seed-path", default="", help="Path if models are saved by seed (forwarded to train.py) - only used by analyze_seed_occupancy.py, ignored otherwise.") + parser.add_argument( + "--flush-after-drop-streak", + type=int, + default=0, + help="At episode end, flush only if the current consecutive dropped-job streak reached this many steps (0 disables).", + ) args = parser.parse_args() try: @@ -91,6 +99,8 @@ def main(): parser.error(str(exc)) if args.jobs_exact_replay and not norm_path(args.jobs): parser.error("--jobs-exact-replay requires --jobs") + if args.jobs_exact_replay_aggregate and not args.jobs_exact_replay: + parser.error("--jobs-exact-replay-aggregate requires --jobs-exact-replay") if args.workload_gen and args.job_arrival_scale != 1.0: print( "Warning: --job-arrival-scale is not allowed with --workload-gen; " @@ -171,7 +181,11 @@ def main(): workload_gen=workload_gen, job_arrival_scale=args.job_arrival_scale, jobs_exact_replay=args.jobs_exact_replay, - output_dir=args.output_dir) + output_dir=args.output_dir, + jobs_exact_replay_aggregate=args.jobs_exact_replay_aggregate, + flush_after_drop_streak=args.flush_after_drop_streak) + env.session_dir = session_root + env.plots_dir = plots_dir env.reset(seed=args.seed) # Check if there are any saved models in models_dir @@ -210,7 +224,7 @@ def main(): ) model = PPO.load(latest_model_file, env=env, tensorboard_log=log_dir, n_steps=64, batch_size=64, device=args.device) else: - print("Starting a new model training...") + print(f"Starting a new model training...") # Parse network architecture from comma-separated string (e.g., "256,128" -> [256, 128]) net_arch_layers = [int(x) for x in args.net_arch.split(',')] policy_kwargs = dict( @@ -245,7 +259,7 @@ def main(): print("Train a model first, then run evaluation mode.") return - print("=== EVALUATION MODE ===") + print(f"=== EVALUATION MODE ===") print(f"Evaluation period: {args.eval_months} months ({args.eval_months * 2} episodes, Each episode = 2 weeks)") if evaluation_plots_dir is None: raise RuntimeError("Evaluation plots directory could not be determined for the selected model.") @@ -297,13 +311,13 @@ def main(): results = plot_cumulative_savings(env, env.metrics.episode_costs, session_dir, save=True, show=args.render == 'human') plot_episode_summary(env, env.metrics.episode_costs, session_dir, save=True, show=args.render == 'human', suffix=f"eval_{args.eval_months}m") if results: - print("\n=== CUMULATIVE SAVINGS ANALYSIS ===") - print("\nVs Baseline (with idle nodes):") + print(f"\n=== CUMULATIVE SAVINGS ANALYSIS ===") + print(f"\nVs Baseline (with idle nodes):") print(f" Total Savings: €{results['total_savings']:,.0f}") print(f" Average Monthly Reduction: {results['avg_monthly_savings_pct']:.1f}%") print(f" Annual Savings Rate: €{results['total_savings'] * 12 / args.eval_months:,.0f}/year") - print("\nVs Baseline_off (no idle nodes):") + print(f"\nVs Baseline_off (no idle nodes):") print(f" Total Savings: €{results['total_savings_off']:,.0f}") print(f" Average Monthly Reduction: {results['avg_monthly_savings_pct_off']:.1f}%") print(f" Annual Savings Rate: €{results['total_savings_off'] * 12 / args.eval_months:,.0f}/year") @@ -381,7 +395,7 @@ def main(): print(f" Baseline Total Cost: €{total_baseline_cost:,.0f}") print(f" Baseline_off Total Cost: €{total_baseline_off_cost:,.0f}") - print("\n=== COST PER 1,000 COMPLETED JOBS ===") + print(f"\n=== COST PER 1,000 COMPLETED JOBS ===") print(f" Agent: {fmt_optional(total_agent_cost_per_1000_completed, 2, thousands=True)} €/1k jobs") print(f" Baseline: {fmt_optional(total_baseline_cost_per_1000_completed, 2, thousands=True)} €/1k jobs") print(f" Baseline_off: {fmt_optional(total_baseline_off_cost_per_1000_completed, 2, thousands=True)} €/1k jobs") @@ -393,13 +407,13 @@ def main(): print(f" Vs Baseline_off: {fmt_optional(total_dropped_jobs_per_saved_euro_off, 6)} jobs/€") - print("\n=== POWER & PRICE METRICS (TOTAL OVER EVALUATION) ===") - print(f" Agent: Power={total_agent_power_mwh:,.1f} MWh, Mean Price={total_agent_mean_price:.2f} €/MWh") - print(f" Baseline: Power={total_baseline_power_mwh:,.1f} MWh, Mean Price={total_baseline_mean_price:.2f} €/MWh") - print(f" Baseline_off: Power={total_baseline_off_power_mwh:,.1f} MWh, Mean Price={total_baseline_off_mean_price:.2f} €/MWh") - print("\n=== COST SAVINGS (TOTAL OVER EVALUATION) ===") - print(f" Vs Baseline: €{total_savings_vs_baseline:,.0f}, {fmt_optional(safe_ratio(total_savings_vs_baseline * 100.0, total_baseline_cost), 1)}%") - print(f" Vs Baseline_off: €{total_savings_vs_baseline_off:,.0f}, {fmt_optional(safe_ratio(total_savings_vs_baseline_off * 100.0, total_baseline_off_cost), 1)}%") + print(f"\n=== POWER & PRICE METRICS (TOTAL OVER EVALUATION) ===") + print(f" Agent: Power={total_agent_prop_power_mwh:,.1f} MWh, Mean Price={total_agent_prop_mean_price:.2f} €/MWh") + print(f" Baseline: Power={total_baseline_prop_power_mwh:,.1f} MWh, Mean Price={total_baseline_prop_mean_price:.2f} €/MWh") + print(f" Baseline_off: Power={total_baseline_off_prop_power_mwh:,.1f} MWh, Mean Price={total_baseline_off_prop_mean_price:.2f} €/MWh") + print(f"\n=== PROPORTIONAL COST SAVINGS (TOTAL OVER EVALUATION) ===") + print(f" Vs Baseline: €{total_savings_prop_cost_vs_baseline:,.0f}, {fmt_optional(prop_savings_pct_vs_baseline, 1)}%") + print(f" Vs Baseline_off: €{total_savings_prop_cost_vs_baseline_off:,.0f}, {fmt_optional(prop_savings_pct_vs_baseline_off, 1)}%") except Exception as e: print(f"Could not generate cumulative savings plot: {e}") @@ -433,11 +447,7 @@ def main(): print(f"iterations limit ({args.iter_limit}) reached: {iters}.") break try: - model.learn(total_timesteps=STEPS_PER_ITERATION, reset_num_timesteps=False, tb_log_name=f"PPO", callback=ComputeClusterCallback()) - print(f"Iteration {iters} finished in {time.time()-t0:.2f}s") - model.save(f"{models_dir}/{STEPS_PER_ITERATION * iters}.zip") - - if args.plot_dashboard: + if args.plot_dashboard and (STEPS_PER_ITERATION * (iters - 1)) % args.dashboard_interval == 0: # Only plot after the first iteration to avoid empty data try: plot_dashboard( env, @@ -449,6 +459,11 @@ def main(): ) except Exception as e: print(f"Dashboard plot failed (non-fatal): {e}") + + model.learn(total_timesteps=STEPS_PER_ITERATION, reset_num_timesteps=False, tb_log_name=f"PPO", callback=ComputeClusterCallback()) + print(f"Iteration {iters} finished in {time.time()-t0:.2f}s") + model.save(f"{models_dir}/{STEPS_PER_ITERATION * iters}.zip") + except PlottingComplete: print("Plotting complete, terminating training...") diff --git a/train_iter.py b/train_iter.py index ca601aa..755520f 100644 --- a/train_iter.py +++ b/train_iter.py @@ -118,6 +118,7 @@ def build_command( seed_sweep=False, evaluate_savings=False, eval_months=0, + flush_after_drop_streak=0, workloadgen_args=None, output_dir=None, ): @@ -147,6 +148,8 @@ def build_command( command += ["--seed-sweep"] if evaluate_savings: command += ["--evaluate-savings", "--eval-months", str(eval_months)] + if flush_after_drop_streak > 0: + command += ["--flush-after-drop-streak", str(flush_after_drop_streak)] if workloadgen_args: command += workloadgen_args if output_dir is not None: @@ -366,8 +369,8 @@ def _reap(proc, label, fh, elapsed): def run_all_parallel(combinations, max_parallel, iter_limit_per_step, session, prices, job_durations, jobs, hourly_jobs, job_arrival_scale, jobs_exact_replay, plot_dashboard, dashboard_hours, - seeds, seed_sweep, evaluate_savings, eval_months, workloadgen_args, - no_tui=False, output_dir=None): + seeds, seed_sweep, evaluate_savings, eval_months, flush_after_drop_streak, workloadgen_args, + no_tui=False): multi_seed = len(seeds) > 1 current_env = os.environ.copy() log_dir = make_log_dir(session, output_dir or "sessions") @@ -383,8 +386,7 @@ def launch(combo, seed): iter_limit_per_step, session, prices, job_durations, jobs, hourly_jobs, job_arrival_scale, jobs_exact_replay, plot_dashboard, dashboard_hours, seed, seed_sweep, - evaluate_savings, eval_months, workloadgen_args, - output_dir=output_dir, + evaluate_savings, eval_months, flush_after_drop_streak, workloadgen_args, ) log_path = os.path.join(log_dir, label_to_filename(label)) log_fh = open(log_path, "w") @@ -443,6 +445,12 @@ def main(): parser.add_argument("--parallel", type=int, default=1, metavar="N", help="Number of training runs to execute in parallel (default: 1, sequential)") parser.add_argument("--evaluate-savings", action="store_true", help="Forward to train.py to evaluate savings compared to baseline.") parser.add_argument("--eval-months", type=int, default=6, help="Number of months to evaluate savings over (forwarded to train.py)") + parser.add_argument( + "--flush-after-drop-streak", + type=int, + default=0, + help="Forward to train.py: flush at episode end only after this many consecutive dropped-job steps (0 disables).", + ) parser.add_argument("--no-tui", action="store_true", help="Disable interactive TUI; print plain progress lines instead (auto-disabled when not a TTY)") add_workloadgen_args(parser) @@ -508,6 +516,7 @@ def main(): evaluate_savings=args.evaluate_savings, eval_months=args.eval_months, workloadgen_args=workloadgen_args, + flush_after_drop_streak=args.flush_after_drop_streak, no_tui=args.no_tui, output_dir=args.output_dir, ) From 19d38b8d3b617eae5daa5c91068dc0608ba2a66d Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Thu, 9 Apr 2026 12:03:36 +0200 Subject: [PATCH 6/9] Change that flush happens during the step where the drop streak is reached. --- src/environment.py | 82 +++++++++++++++++++-------------------- src/metrics_tracker.py | 4 +- src/reward_calculation.py | 4 +- 3 files changed, 45 insertions(+), 45 deletions(-) diff --git a/src/environment.py b/src/environment.py index 2f82f27..a11a820 100644 --- a/src/environment.py +++ b/src/environment.py @@ -95,7 +95,6 @@ def __init__(self, self.jobs_exact_replay_aggregate = bool(jobs_exact_replay_aggregate) self.flush_after_drop_streak = max(0, int(flush_after_drop_streak)) self.consecutive_drop_steps = 0 - self.flush_pending_for_episode = False self.next_plot_save = self.steps_per_iteration @@ -192,7 +191,6 @@ def __init__(self, def _reset_timeline_state(self, start_index: int) -> None: self.prices.reset(start_index=start_index) self.consecutive_drop_steps = 0 - self.flush_pending_for_episode = False self.state = { # Initialize all nodes to be 'online but free' (0) @@ -352,8 +350,8 @@ def _flush_workload_side( backlog_queue.clear() return int(flushed_jobs) - def _flush_episode_state(self) -> dict[str, int | float | bool]: - """Flush both agent and baseline states after the terminal step of an episode.""" + def _flush_workload_state(self) -> dict[str, int | float | bool]: + """Flush both agent and baseline states immediately after a trigger step.""" agent_job_queue_2d = self.state['job_queue'].reshape(-1, 4) baseline_job_queue_2d = self.baseline_state['job_queue'].reshape(-1, 4) @@ -389,7 +387,6 @@ def _flush_episode_state(self) -> dict[str, int | float | bool]: self._mark_queue_backlog_mutation() self._update_pending_job_stats(agent_job_queue_2d) self.consecutive_drop_steps = 0 - self.flush_pending_for_episode = False self.state['job_queue'] = agent_job_queue_2d.flatten() self.baseline_state['job_queue'] = baseline_job_queue_2d.flatten() @@ -675,11 +672,45 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, self.consecutive_drop_steps += 1 else: self.consecutive_drop_steps = 0 - if ( + if self.consecutive_drop_steps > self.metrics.episode_max_drop_streak: + self.metrics.episode_max_drop_streak = self.consecutive_drop_steps + flush_applied = False + flush_penalty = 0.0 + agent_jobs_flushed = 0 + baseline_jobs_flushed = 0 + drop_streak_steps = self.consecutive_drop_steps + flush_triggered_by_drop_streak = ( self.flush_after_drop_streak > 0 and self.consecutive_drop_steps >= self.flush_after_drop_streak - ): - self.flush_pending_for_episode = True + ) + if flush_triggered_by_drop_streak: + flush_result = self._flush_workload_state() + flush_applied = bool(flush_result["flush_applied"]) + flush_penalty = float(flush_result["flush_penalty"]) + agent_jobs_flushed = int(flush_result["agent_jobs_flushed"]) + baseline_jobs_flushed = int(flush_result["baseline_jobs_flushed"]) + if flush_penalty != 0.0: + step_reward += flush_penalty + self.metrics.episode_reward += flush_penalty + if self.metrics.rewards: + self.metrics.rewards[-1] += flush_penalty + if self.metrics.episode_rewards: + self.metrics.episode_rewards[-1] += flush_penalty + + post_flush_pending_summary = self._pending_work_summary(self.state['job_queue'].reshape(-1, 4)) + self.metrics.episode_pending_jobs_end = int(post_flush_pending_summary["pending_job_count"]) + self.metrics.episode_pending_core_demand_end = float(post_flush_pending_summary["pending_core_demand"]) + self.metrics.episode_pending_core_hours_end = float(post_flush_pending_summary["pending_core_hours"]) + self.metrics.episode_overdue_jobs_end = int(post_flush_pending_summary["overdue_jobs"]) + self.metrics.episode_overdue_age_core_hours_end = float(post_flush_pending_summary["overdue_age_core_hours"]) + + if flush_applied: + self.env_print( + f"[flush] Drop streak {self.flush_after_drop_streak}+ reached " + f"({drop_streak_steps} steps). " + f"Dropped outstanding work immediately: " + f"agent={agent_jobs_flushed}, baseline={baseline_jobs_flushed}, penalty={flush_penalty:.4f}" + ) # print stats self.env_print(f"[6] End of step stats...") @@ -696,13 +727,6 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, truncated = False terminated = False - flush_applied = False - flush_penalty = 0.0 - agent_jobs_flushed = 0 - baseline_jobs_flushed = 0 - drop_streak_steps = self.consecutive_drop_steps - flush_armed_by_drop_streak = bool(self.flush_pending_for_episode) - episode_flush_triggered_by_drop_streak = False if self.metrics.current_hour == EPISODE_HOURS: if self.render_mode == 'human': plot_episode(self, EPISODE_HOURS, MAX_NODES, False, True, self.current_step) @@ -717,30 +741,6 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, truncated = True terminated = False - if flush_armed_by_drop_streak: - episode_flush_triggered_by_drop_streak = True - flush_result = self._flush_episode_state() - flush_applied = bool(flush_result["flush_applied"]) - flush_penalty = float(flush_result["flush_penalty"]) - agent_jobs_flushed = int(flush_result["agent_jobs_flushed"]) - baseline_jobs_flushed = int(flush_result["baseline_jobs_flushed"]) - if flush_penalty != 0.0: - step_reward += 10*flush_penalty - self.metrics.episode_reward += flush_penalty - if self.metrics.rewards: - self.metrics.rewards[-1] += flush_penalty - if self.metrics.episode_rewards: - self.metrics.episode_rewards[-1] += flush_penalty - if flush_applied: - self.env_print( - f"[flush] Drop streak {self.flush_after_drop_streak}+ reached " - f"({drop_streak_steps} steps). " - f"Dropped outstanding work at episode boundary: " - f"agent={agent_jobs_flushed}, baseline={baseline_jobs_flushed}, penalty={flush_penalty:.4f}" - ) - else: - self.flush_pending_for_episode = False - # Record episode costs for long-term analysis self.metrics.record_episode_completion(self.current_episode) @@ -762,9 +762,9 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, "episode_jobs_flushed": self.metrics.episode_jobs_flushed, "episode_jobs_lost_total": self.metrics.episode_jobs_dropped + self.metrics.episode_jobs_flushed, "episode_flush_applied": flush_applied, - "episode_flush_triggered_by_drop_streak": episode_flush_triggered_by_drop_streak, + "episode_flush_triggered_by_drop_streak": flush_triggered_by_drop_streak, "drop_streak_steps": drop_streak_steps, - "drop_streak_flush_armed": flush_armed_by_drop_streak, + "drop_streak_flush_armed": flush_triggered_by_drop_streak, "step_flush_penalty": flush_penalty, "step_jobs_flushed": agent_jobs_flushed, "step_baseline_jobs_flushed": baseline_jobs_flushed, diff --git a/src/metrics_tracker.py b/src/metrics_tracker.py index 1484066..4da9c13 100644 --- a/src/metrics_tracker.py +++ b/src/metrics_tracker.py @@ -18,7 +18,7 @@ def _safe_ratio(numerator: float, denominator: float) -> float: @staticmethod def _jobs_lost_total(dropped: int, flushed: int) -> int: - """Total lost jobs, including regular drops and episode-end flushes.""" + """Total lost jobs, including regular drops and streak-triggered flushes.""" return int(dropped) + int(flushed) def __init__(self) -> None: @@ -319,7 +319,7 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i 'baseline_completion_rate': baseline_completion_rate, 'baseline_max_queue_size': self.episode_baseline_max_queue_size_reached, 'baseline_max_backlog_size': self.episode_baseline_max_backlog_size_reached, - # Loss metrics: includes age expirations, queue-full rejections, and episode-end flushes. + # Loss metrics: includes age expirations, queue-full rejections, and streak-triggered flushes. "jobs_dropped": self.episode_jobs_dropped, "jobs_flushed": self.episode_jobs_flushed, "jobs_lost_total": jobs_lost_total, diff --git a/src/reward_calculation.py b/src/reward_calculation.py index 823b483..0ad7abf 100644 --- a/src/reward_calculation.py +++ b/src/reward_calculation.py @@ -384,7 +384,7 @@ def _penalty_drop_alt(self, num_dropped: int) -> float: """Drop penalty: tanh saturation curve bounded in [-1, 0].""" return -float(np.tanh(num_dropped / self.DROP_PENALTY_TAU)) def loss_penalty(self, num_lost: int) -> float: - """Weighted penalty for jobs lost this step, including episode-end flushes.""" + """Weighted penalty for jobs lost this step, including streak-triggered flushes.""" if not self.ALLOW_DROP_PENALTY or num_lost <= 0: return 0.0 return 0.3 * self._penalty_drop(num_lost) @@ -446,7 +446,7 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo idle_penalty_weighted = weights.idle_weight * idle_penalty_norm # 6. penalty for lost jobs (aged out or rejected because queue/backlog was full) - # Episode-end flushes reuse the same penalty path via loss_penalty(). + # Streak-triggered flushes reuse the same penalty path via loss_penalty(). drop_penalty_weighted = self.loss_penalty(num_dropped_this_step) reward = ( From 8579f5984569e67e7305b4cb4c3a6e4ae9766599 Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Thu, 9 Apr 2026 15:18:34 +0200 Subject: [PATCH 7/9] `Terminate episode on drop-streak flush with fixed penalty` --- src/environment.py | 29 +++++++++++++++++++---------- train.py | 2 +- train_iter.py | 2 +- 3 files changed, 21 insertions(+), 12 deletions(-) diff --git a/src/environment.py b/src/environment.py index a11a820..a506869 100644 --- a/src/environment.py +++ b/src/environment.py @@ -46,6 +46,7 @@ class ComputeClusterEnv(gym.Env): """An environment for scheduling compute jobs based on electricity price predictions.""" metadata = {'render.modes': ['human', 'none']} + DROP_STREAK_TERMINATION_PENALTY = -200.0 def render(self, mode: str = 'human') -> None: self.render_mode = mode @@ -678,6 +679,7 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, flush_penalty = 0.0 agent_jobs_flushed = 0 baseline_jobs_flushed = 0 + terminal_penalty_applied = 0.0 drop_streak_steps = self.consecutive_drop_steps flush_triggered_by_drop_streak = ( self.flush_after_drop_streak > 0 @@ -689,13 +691,14 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, flush_penalty = float(flush_result["flush_penalty"]) agent_jobs_flushed = int(flush_result["agent_jobs_flushed"]) baseline_jobs_flushed = int(flush_result["baseline_jobs_flushed"]) - if flush_penalty != 0.0: - step_reward += flush_penalty - self.metrics.episode_reward += flush_penalty - if self.metrics.rewards: - self.metrics.rewards[-1] += flush_penalty - if self.metrics.episode_rewards: - self.metrics.episode_rewards[-1] += flush_penalty + previous_step_reward = step_reward + terminal_penalty_applied = float(self.DROP_STREAK_TERMINATION_PENALTY) + step_reward = terminal_penalty_applied + self.metrics.episode_reward += terminal_penalty_applied - previous_step_reward + if self.metrics.rewards: + self.metrics.rewards[-1] = terminal_penalty_applied + if self.metrics.episode_rewards: + self.metrics.episode_rewards[-1] = terminal_penalty_applied post_flush_pending_summary = self._pending_work_summary(self.state['job_queue'].reshape(-1, 4)) self.metrics.episode_pending_jobs_end = int(post_flush_pending_summary["pending_job_count"]) @@ -709,7 +712,8 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, f"[flush] Drop streak {self.flush_after_drop_streak}+ reached " f"({drop_streak_steps} steps). " f"Dropped outstanding work immediately: " - f"agent={agent_jobs_flushed}, baseline={baseline_jobs_flushed}, penalty={flush_penalty:.4f}" + f"agent={agent_jobs_flushed}, baseline={baseline_jobs_flushed}, " + f"loss_penalty={flush_penalty:.4f}, terminal_penalty={terminal_penalty_applied:.4f}" ) # print stats @@ -727,7 +731,10 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, truncated = False terminated = False - if self.metrics.current_hour == EPISODE_HOURS: + if flush_triggered_by_drop_streak: + terminated = True + truncated = False + elif self.metrics.current_hour == EPISODE_HOURS: if self.render_mode == 'human': plot_episode(self, EPISODE_HOURS, MAX_NODES, False, True, self.current_step) if self.plot_config.plot_once: @@ -741,7 +748,8 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, truncated = True terminated = False - # Record episode costs for long-term analysis + if terminated or truncated: + # Record episode costs before reset so callbacks/evaluation can read the finished episode. self.metrics.record_episode_completion(self.current_episode) # flatten job_queue again @@ -766,6 +774,7 @@ def step(self, action: np.ndarray) -> tuple[dict[str, np.ndarray], float, bool, "drop_streak_steps": drop_streak_steps, "drop_streak_flush_armed": flush_triggered_by_drop_streak, "step_flush_penalty": flush_penalty, + "step_terminal_penalty": terminal_penalty_applied, "step_jobs_flushed": agent_jobs_flushed, "step_baseline_jobs_flushed": baseline_jobs_flushed, } diff --git a/train.py b/train.py index 0ae82db..973285d 100644 --- a/train.py +++ b/train.py @@ -89,7 +89,7 @@ def main(): "--flush-after-drop-streak", type=int, default=0, - help="At episode end, flush only if the current consecutive dropped-job streak reached this many steps (0 disables).", + help="Immediately flush and terminate the episode after this many consecutive dropped-job steps (0 disables).", ) args = parser.parse_args() diff --git a/train_iter.py b/train_iter.py index 755520f..a758e82 100644 --- a/train_iter.py +++ b/train_iter.py @@ -449,7 +449,7 @@ def main(): "--flush-after-drop-streak", type=int, default=0, - help="Forward to train.py: flush at episode end only after this many consecutive dropped-job steps (0 disables).", + help="Forward to train.py: immediately flush and terminate the episode after this many consecutive dropped-job steps (0 disables).", ) parser.add_argument("--no-tui", action="store_true", help="Disable interactive TUI; print plain progress lines instead (auto-disabled when not a TTY)") add_workloadgen_args(parser) From 10374a2af662ca27e33e0b00b159f0dc0ab45950 Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Thu, 9 Apr 2026 16:18:23 +0200 Subject: [PATCH 8/9] `Separate intrinsic overdue-starvation reward from cheap-hour backlog pressure` --- analyze_arrivalscale_occupancy.py | 9 +-- analyze_lambda_occupancy.py | 6 +- analyze_seed_occupancy.py | 6 +- src/callbacks.py | 4 +- src/reward_calculation.py | 108 ++++++++++++++++++++++-------- train.py | 4 +- 6 files changed, 95 insertions(+), 42 deletions(-) diff --git a/analyze_arrivalscale_occupancy.py b/analyze_arrivalscale_occupancy.py index 2b79a1d..7684db1 100644 --- a/analyze_arrivalscale_occupancy.py +++ b/analyze_arrivalscale_occupancy.py @@ -60,7 +60,7 @@ r"(?P-?[\d,]+(?:\.\d+)?|n/a)\s*€/1k.*?" r"Jobs=[\d,]+\/[\d,]+\s+\((?P-?[\d.]+)%\),\s*" r"AvgWait=(?P-?[\d.]+)h,.*?" - r"Dropped=(?P-?[\d,]+),.*?" + r"(?:Dropped|Lost)=(?P-?[\d,]+),.*?" r"Agent Occupancy \(Nodes\)=\s*(?P-?[\d.]+)%,\s*" r"Baseline Occupancy \(Nodes\)=\s*(?P-?[\d.]+)%", re.MULTILINE, @@ -78,11 +78,11 @@ ) DROPPED_AGENT_SUMMARY_RE = re.compile( - r"Total Dropped Jobs \(Agent\):\s*(?P[\d,]+)" + r"Total (?:Dropped|Lost) Jobs \(Agent\):\s*(?P[\d,]+)" ) DROPPED_BASELINE_SUMMARY_RE = re.compile( - r"Total Dropped Jobs \(Baseline\):\s*(?P[\d,]+)" + r"Total (?:Dropped|Lost) Jobs \(Baseline\):\s*(?P[\d,]+)" ) @@ -200,7 +200,8 @@ def parse_episode_metrics( if not occupancy: raise RuntimeError( "Could not parse episode metrics from train.py output. " - "Expected lines like 'Episode X: ... Savings=€.../€..., Power=..., CostPer1kCompleted=..., Agent Occupancy (Nodes)=...%'." + "Expected lines like 'Episode X: ... Savings=€.../€..., Power=..., CostPer1kCompleted=..., " + "(Lost|Dropped)=..., Agent Occupancy (Nodes)=...%'." ) return ( diff --git a/analyze_lambda_occupancy.py b/analyze_lambda_occupancy.py index 5f9df89..61cac37 100644 --- a/analyze_lambda_occupancy.py +++ b/analyze_lambda_occupancy.py @@ -53,7 +53,7 @@ r"(?P-?[\d,]+(?:\.\d+)?|n/a)\s*€/1k.*?" r"Jobs=[\d,]+\/[\d,]+\s+\((?P-?[\d.]+)%\),\s*" r"AvgWait=(?P-?[\d.]+)h,.*?" - r"Dropped=(?P-?[\d,]+),.*?" + r"(?:Dropped|Lost)=(?P-?[\d,]+),.*?" r"Agent Occupancy \(Nodes\)=\s*(?P-?[\d.]+)%,\s*" r"Baseline Occupancy \(Nodes\)=\s*(?P-?[\d.]+)%", re.MULTILINE, @@ -71,11 +71,11 @@ ) DROPPED_AGENT_SUMMARY_RE = re.compile( - r"Total Dropped Jobs \(Agent\):\s*(?P[\d,]+)" + r"Total (?:Dropped|Lost) Jobs \(Agent\):\s*(?P[\d,]+)" ) DROPPED_BASELINE_SUMMARY_RE = re.compile( - r"Total Dropped Jobs \(Baseline\):\s*(?P[\d,]+)" + r"Total (?:Dropped|Lost) Jobs \(Baseline\):\s*(?P[\d,]+)" ) diff --git a/analyze_seed_occupancy.py b/analyze_seed_occupancy.py index 10c4073..cfc6b84 100644 --- a/analyze_seed_occupancy.py +++ b/analyze_seed_occupancy.py @@ -57,7 +57,7 @@ r"(?P-?[\d,]+(?:\.\d+)?|n/a)\s*€/1k.*?" r"Jobs=[\d,]+\/[\d,]+\s+\((?P-?[\d.]+)%\),\s*" r"AvgWait=(?P-?[\d.]+)h,.*?" - r"Dropped=(?P-?[\d,]+),.*?" + r"(?:Dropped|Lost)=(?P-?[\d,]+),.*?" r"Agent Occupancy \(Nodes\)=\s*(?P-?[\d.]+)%,\s*" r"Baseline Occupancy \(Nodes\)=\s*(?P-?[\d.]+)%", re.MULTILINE, @@ -75,11 +75,11 @@ ) DROPPED_AGENT_SUMMARY_RE = re.compile( - r"Total Dropped Jobs \(Agent\):\s*(?P[\d,]+)" + r"Total (?:Dropped|Lost) Jobs \(Agent\):\s*(?P[\d,]+)" ) DROPPED_BASELINE_SUMMARY_RE = re.compile( - r"Total Dropped Jobs \(Baseline\):\s*(?P[\d,]+)" + r"Total (?:Dropped|Lost) Jobs \(Baseline\):\s*(?P[\d,]+)" ) diff --git a/src/callbacks.py b/src/callbacks.py index ef76154..f87d4ff 100644 --- a/src/callbacks.py +++ b/src/callbacks.py @@ -24,7 +24,9 @@ def _on_rollout_start(self) -> None: def _on_step(self) -> bool: env = self.training_env.envs[0].unwrapped - if env.metrics.current_hour == EPISODE_HOURS-1: + dones = self.locals.get("dones") + episode_finished = bool(dones[0]) if dones is not None else (env.metrics.current_hour == EPISODE_HOURS - 1) + if episode_finished: self.logger.record("metrics/cost", env.metrics.episode_total_cost) self.logger.record("metrics/savings", env.metrics.episode_baseline_cost - env.metrics.episode_total_cost) self.logger.record("metrics/savings_off", env.metrics.episode_baseline_cost_off - env.metrics.episode_total_cost) diff --git a/src/reward_calculation.py b/src/reward_calculation.py index 0ad7abf..28d8857 100644 --- a/src/reward_calculation.py +++ b/src/reward_calculation.py @@ -56,6 +56,8 @@ class RewardCalculator: # Asymmetric node scaling: high-price execution ramps faster than low-price reward. PRICE_NODE_TAU_POS = 40.0 PRICE_NODE_TAU_NEG = 40.0 + PRICE_QUANTILE_LOW = 0.10 + PRICE_QUANTILE_HIGH = 0.90 NEGATIVE_PRICE_NODE_TAU = 30.0 # fast node saturation only for negative-price overdrive NEGATIVE_PRICE_TAU = 8.0 NEGATIVE_PRICE_OVERDRIVE_GAIN = 2.5 @@ -71,6 +73,14 @@ class RewardCalculator: CHEAP_SERVICE_GAIN = 0.75 OVERDUE_BACKLOG_GAIN = 1.25 OVERDUE_AGE_CORE_HOUR_TAU = 0.5 * MAX_NODES * CORES_PER_NODE + # Separate intrinsic service-continuity signal: + # once work is overdue, staying near full blackout should hurt more than doing + # at least some useful work. This is intentionally unweighted so it remains a + # structural part of the objective rather than a tunable backlog coefficient. + INTRINSIC_STARVATION_GAIN = 1.5 + INTRINSIC_STARVATION_OVERDUE_TAU = 0.25 * MAX_NODES * CORES_PER_NODE + INTRINSIC_STARVATION_SERVICE_CORE_TAU = 12.0 + ALLOW_DROP_PENALTY = True # whether to include penalties for dropped jobs in the reward calculation def __init__(self, prices: Prices) -> None: """ @@ -319,23 +329,30 @@ def _blackout_term(self, num_used_nodes: int, num_idle_nodes: int, num_unprocess return 1.0 if num_unprocessed_jobs <= 0 else 0.0 + def _overdue_pressure(self, remaining_overdue_age_core_hours: float, tau: float) -> float: + """Smoothly map post-grace overdue mass into [0, 1].""" + if remaining_overdue_age_core_hours <= 0.0: + return 0.0 + return float( + 1.0 - np.exp( + -float(remaining_overdue_age_core_hours) / max(float(tau), 1e-6) + ) + ) + def _penalty_job_age( self, current_price: float, decision_pending_core_demand: float, - remaining_overdue_age_core_hours: float, total_used_cores: int, ) -> float: """ - Combined backlog pressure term used in the legacy "job age" reward slot. - - It deliberately does two things: - 1. Cheap-hour service pressure: - If cheap compute is available *and* backlog exists, the agent should keep - the cluster busy instead of trickling. - 2. Overdue backlog pressure: - Once jobs have outlived the deferral grace period, leaving them pending - becomes increasingly expensive regardless of price. + Weighted cheap-hour service pressure used in the legacy "job age" slot. + + This term now only teaches *when* pending work should be served: + if the current hour is cheap and runnable demand exists, under-serving that + hour is penalized. Post-grace starvation is handled separately by an + intrinsic reward so the "do nothing" failure mode is structural rather than + weight-dependent. """ cheap_strength, _ = self._price_phase_strengths(current_price) @@ -346,30 +363,46 @@ def _penalty_job_age( achieved_service = min(float(total_used_cores), target_service) cheap_service_shortfall = cheap_strength * (1.0 - achieved_service / max(target_service, 1e-6)) - overdue_pressure = 0.0 - if remaining_overdue_age_core_hours > 0.0: - overdue_pressure = 1.0 - np.exp( - -float(remaining_overdue_age_core_hours) / max(self.OVERDUE_AGE_CORE_HOUR_TAU, 1e-6) - ) + return float(np.clip(self.CHEAP_SERVICE_GAIN * cheap_service_shortfall, 0.0, 1.0)) + + def _reward_intrinsic_starvation( + self, + remaining_overdue_age_core_hours: float, + total_used_cores: int, + ) -> float: + """ + Intrinsic post-grace service-continuity reward. + + Once pending work has moved past the 24h grace window, staying near zero + service becomes increasingly negative. Any useful work reduces this penalty + smoothly, which makes a full blackout with overdue backlog worse than at + least some utilization. + """ + overdue_pressure = self._overdue_pressure( + remaining_overdue_age_core_hours, + self.INTRINSIC_STARVATION_OVERDUE_TAU, + ) + if overdue_pressure <= 0.0: + return 0.0 - combined_pressure = ( - self.CHEAP_SERVICE_GAIN * cheap_service_shortfall - + self.OVERDUE_BACKLOG_GAIN * overdue_pressure + zero_service_pressure = float( + np.exp( + -max(float(total_used_cores), 0.0) + / max(self.INTRINSIC_STARVATION_SERVICE_CORE_TAU, 1e-6) + ) ) - return float(np.clip(combined_pressure, 0.0, 1.0)) + return float(-self.INTRINSIC_STARVATION_GAIN * overdue_pressure * zero_service_pressure) def _penalty_job_age_normalized( self, current_price: float, decision_pending_core_demand: float, - remaining_overdue_age_core_hours: float, total_used_cores: int, ) -> float: - """Legacy reward slot for backlog pressure, normalized to [-1, 0].""" + """Cheap-hour service pressure, normalized to [-1, 0].""" current_penalty = self._penalty_job_age( current_price, decision_pending_core_demand, - remaining_overdue_age_core_hours, total_used_cores, ) return float(np.clip(-current_penalty, -1.0, 0.0)) @@ -425,21 +458,33 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo efficiency_reward_norm += self._blackout_term(num_used_nodes, num_idle_nodes, num_unprocessed_jobs) efficiency_reward_weighted = weights.efficiency_weight * efficiency_reward_norm - # 2. Increase reward if current price is favorable and currently useful work is high. - price_reward = self._reward_price_utilization(current_price, average_future_price, total_used_cores) # legacy: price_reward = self._reward_price_normalized_legacy(current_price, average_future_price, total_used_cores) + # 2. Increase reward if current price is favorable and currently useful work is high. + price_reward = self._reward_price_utilization(current_price, average_future_price, total_used_cores) price_reward_weighted = weights.price_weight * price_reward - # 3. Push pending work into cheap hours and punish starving backlog after the grace period. - # The method name is kept for compatibility with existing plots/logs, but the semantics - # now describe backlog pressure rather than a simple "oldest queue age" penalty. + # 3. Push pending work into cheap hours. The method name is kept for + # compatibility with existing plots/logs, but the semantics now describe + # cheap-hour service pressure rather than a simple "oldest queue age" penalty. job_age_penalty_norm = self._penalty_job_age_normalized( current_price, decision_pending_core_demand, + total_used_cores, + ) + + intrinsic_starvation_reward = self._reward_intrinsic_starvation( + remaining_overdue_age_core_hours, + total_used_cores, + ) + + job_age_penalty_weighted = weights.job_age_weight * job_age_penalty_norm + intrinsic_starvation_reward + + # 4. Intrinsic anti-starvation signal: once work is overdue, staying near + # zero throughput should be structurally worse than doing at least some work. + intrinsic_starvation_reward = self._reward_intrinsic_starvation( remaining_overdue_age_core_hours, total_used_cores, ) - job_age_penalty_weighted = weights.job_age_weight * job_age_penalty_norm # 4. penalty for idling nodes idle_penalty_norm = self._penalty_idle_normalized(num_idle_nodes) @@ -457,7 +502,12 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo + drop_penalty_weighted ) - 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}") + env_print( + f" > $$$TOTAL: {reward:.4f} = " + f"{efficiency_reward_weighted:.4f} + {price_reward_weighted:.4f} + " + f"{idle_penalty_weighted:.4f} + {job_age_penalty_weighted:.4f} + " + f"{intrinsic_starvation_reward:.4f} + {drop_penalty_weighted:.4f}" + ) env_print(f" > step cost: €{total_cost:.4f}") return reward, total_cost, efficiency_reward_norm, price_reward, idle_penalty_norm, job_age_penalty_norm diff --git a/train.py b/train.py index 973285d..6ef5a0a 100644 --- a/train.py +++ b/train.py @@ -38,7 +38,7 @@ def fmt_optional(value: float | None, precision: int = 2, thousands: bool = Fals return f"{value:,.{precision}f}" if thousands else f"{value:.{precision}f}" -STEPS_PER_ITERATION = 10000 +STEPS_PER_ITERATION = 100000 def main(): @@ -447,7 +447,7 @@ def main(): print(f"iterations limit ({args.iter_limit}) reached: {iters}.") break try: - if args.plot_dashboard and (STEPS_PER_ITERATION * (iters - 1)) % args.dashboard_interval == 0: # Only plot after the first iteration to avoid empty data + if args.plot_dashboard and iters > 1 and (STEPS_PER_ITERATION * (iters - 1)) % args.dashboard_interval == 0: # Only plot after the first iteration to avoid empty data try: plot_dashboard( env, From d3e448afdb86ec907034c950f68f67a21266feca Mon Sep 17 00:00:00 2001 From: Enis Lorenz Date: Thu, 9 Apr 2026 11:45:00 +0200 Subject: [PATCH 9/9] Align episode logging with finalized metrics data Log training metrics from the recorded episode summary instead of reading live env counters at the terminal step, so TensorBoard output matches the finalized episode data. - switch callback logging to `metrics.episode_costs[-1]` on actual episode end - add `on_nodes_end` and `used_nodes_end` snapshot fields to episode data - track and export `max_drop_streak` for each episode - include `MaxDropStreak` in the printed episode summary --- src/callbacks.py | 136 +++++++++++++++----------------------- src/evaluation_summary.py | 1 + src/metrics_tracker.py | 4 ++ src/reward_calculation.py | 13 ++-- 4 files changed, 64 insertions(+), 90 deletions(-) diff --git a/src/callbacks.py b/src/callbacks.py index f87d4ff..ae87c7d 100644 --- a/src/callbacks.py +++ b/src/callbacks.py @@ -1,4 +1,4 @@ -from src.config import EPISODE_HOURS, MAX_QUEUE_SIZE, COST_IDLE_MW, COST_USED_MW, CORES_PER_NODE, MAX_NODES +from src.config import COST_IDLE_MW, COST_USED_MW, CORES_PER_NODE, MAX_NODES from stable_baselines3.common.callbacks import BaseCallback @@ -25,89 +25,63 @@ def _on_rollout_start(self) -> None: def _on_step(self) -> bool: env = self.training_env.envs[0].unwrapped dones = self.locals.get("dones") - episode_finished = bool(dones[0]) if dones is not None else (env.metrics.current_hour == EPISODE_HOURS - 1) - if episode_finished: - self.logger.record("metrics/cost", env.metrics.episode_total_cost) - self.logger.record("metrics/savings", env.metrics.episode_baseline_cost - env.metrics.episode_total_cost) - self.logger.record("metrics/savings_off", env.metrics.episode_baseline_cost_off - env.metrics.episode_total_cost) - #self.logger.record("metrics/queue_fill_pct", env.metrics.episode_max_queue_size_reached / MAX_QUEUE_SIZE * 100) - self.logger.record("metrics/baseline_cost", env.metrics.episode_baseline_cost) - self.logger.record("metrics/baseline_cost_off", env.metrics.episode_baseline_cost_off) + if dones is None or not bool(dones[0]) or not env.metrics.episode_costs: + return True - # Job metrics (agent) - completion_rate = (env.metrics.episode_jobs_completed / env.metrics.episode_jobs_submitted * 100 if env.metrics.episode_jobs_submitted > 0 else 0.0) - avg_wait = ( - env.metrics.episode_total_job_wait_time_launch / env.metrics.episode_jobs_launched - if env.metrics.episode_jobs_launched > 0 else 0.0 - ) - self.logger.record("metrics/jobs_submitted", env.metrics.episode_jobs_submitted) - self.logger.record("metrics/jobs_launched", env.metrics.episode_jobs_launched) - self.logger.record("metrics/jobs_completed", env.metrics.episode_jobs_completed) - self.logger.record("metrics/completion_rate", completion_rate) - self.logger.record("metrics/avg_wait_hours", avg_wait) - self.logger.record("metrics/on_nodes", env.metrics.episode_on_nodes[-1]) - self.logger.record("metrics/used_nodes", env.metrics.episode_used_nodes[-1]) - self.logger.record("metrics/max_queue_size", env.metrics.episode_max_queue_size_reached) - self.logger.record("metrics/max_backlog_size", env.metrics.episode_max_backlog_size_reached) - self.logger.record("metrics/pending_jobs_end", env.metrics.episode_pending_jobs_end) - self.logger.record("metrics/pending_core_hours_end", env.metrics.episode_pending_core_hours_end) - self.logger.record("metrics/overdue_jobs_end", env.metrics.episode_overdue_jobs_end) - jobs_lost_total = env.metrics.episode_jobs_dropped + env.metrics.episode_jobs_flushed - self.logger.record("metrics/jobs_dropped", env.metrics.episode_jobs_dropped) - self.logger.record("metrics/jobs_flushed", env.metrics.episode_jobs_flushed) - self.logger.record("metrics/jobs_lost_total", jobs_lost_total) - loss_rate = (jobs_lost_total / env.metrics.episode_jobs_submitted * 100 if env.metrics.episode_jobs_submitted > 0 else 0.0) - self.logger.record("metrics/loss_rate", loss_rate) - self.logger.record("metrics/jobs_rejected_queue_full", env.metrics.episode_jobs_rejected_queue_full) + episode_data = env.metrics.episode_costs[-1] - # Job metrics (baseline) - baseline_completion_rate = (env.metrics.episode_baseline_jobs_completed / env.metrics.episode_baseline_jobs_submitted * 100 if env.metrics.episode_baseline_jobs_submitted > 0 else 0.0) - baseline_avg_wait = ( - env.metrics.episode_baseline_total_job_wait_time_launch / env.metrics.episode_baseline_jobs_launched - if env.metrics.episode_baseline_jobs_launched > 0 else 0.0 - ) - self.logger.record("metrics/baseline_jobs_submitted", env.metrics.episode_baseline_jobs_submitted) - self.logger.record("metrics/baseline_jobs_launched", env.metrics.episode_baseline_jobs_launched) - self.logger.record("metrics/baseline_jobs_completed", env.metrics.episode_baseline_jobs_completed) - self.logger.record("metrics/baseline_completion_rate", baseline_completion_rate) - self.logger.record("metrics/baseline_avg_wait_hours", baseline_avg_wait) - self.logger.record("metrics/baseline_max_queue_size", env.metrics.episode_baseline_max_queue_size_reached) - self.logger.record("metrics/baseline_max_backlog_size", env.metrics.episode_baseline_max_backlog_size_reached) - baseline_jobs_lost_total = env.metrics.episode_baseline_jobs_dropped + env.metrics.episode_baseline_jobs_flushed - self.logger.record("metrics/baseline_jobs_dropped", env.metrics.episode_baseline_jobs_dropped) - self.logger.record("metrics/baseline_jobs_flushed", env.metrics.episode_baseline_jobs_flushed) - self.logger.record("metrics/baseline_jobs_lost_total", baseline_jobs_lost_total) - baseline_loss_rate = (baseline_jobs_lost_total / env.metrics.episode_baseline_jobs_submitted * 100 if env.metrics.episode_baseline_jobs_submitted > 0 else 0.0) - self.logger.record("metrics/baseline_loss_rate", baseline_loss_rate) - self.logger.record("metrics/baseline_jobs_rejected_queue_full", env.metrics.episode_baseline_jobs_rejected_queue_full) + self.logger.record("metrics/cost", float(episode_data["agent_cost"])) + self.logger.record("metrics/savings", float(episode_data["savings_vs_baseline"])) + self.logger.record("metrics/savings_off", float(episode_data["savings_vs_baseline_off"])) + self.logger.record("metrics/baseline_cost", float(episode_data["baseline_cost"])) + self.logger.record("metrics/baseline_cost_off", float(episode_data["baseline_cost_off"])) - # Proportional (per-core) power metrics - _delta = COST_USED_MW - COST_IDLE_MW - agent_prop_power = sum( - COST_IDLE_MW * on + _delta * (cores / CORES_PER_NODE) - for on, cores in zip(env.metrics.episode_on_nodes, env.metrics.episode_used_cores) - ) - baseline_prop_power = sum( - COST_IDLE_MW * MAX_NODES + _delta * (cores / CORES_PER_NODE) - for cores in env.metrics.episode_baseline_used_cores - ) - baseline_off_prop_power = sum( - COST_IDLE_MW * used + _delta * (cores / CORES_PER_NODE) - for used, cores in zip(env.metrics.episode_baseline_used_nodes, env.metrics.episode_baseline_used_cores) - ) - self.logger.record("metrics/prop_power_mwh", agent_prop_power) - self.logger.record("metrics/baseline_prop_power_mwh", baseline_prop_power) - self.logger.record("metrics/baseline_off_prop_power_mwh", baseline_off_prop_power) - self.logger.record("metrics/savings_prop_power_vs_baseline_off", baseline_off_prop_power - agent_prop_power) - agent_prop_cost = sum( - (COST_IDLE_MW * on + _delta * (cores / CORES_PER_NODE)) * price - for on, cores, price in zip(env.metrics.episode_on_nodes, env.metrics.episode_used_cores, env.metrics.episode_price_stats) - ) - baseline_off_prop_cost = sum( - (COST_IDLE_MW * used + _delta * (cores / CORES_PER_NODE)) * price - for used, cores, price in zip(env.metrics.episode_baseline_used_nodes, env.metrics.episode_baseline_used_cores, env.metrics.episode_price_stats) - ) - self.logger.record("metrics/savings_prop_cost_vs_baseline_off", baseline_off_prop_cost - agent_prop_cost) + # Job metrics (agent) + self.logger.record("metrics/jobs_submitted", int(episode_data["jobs_submitted"])) + self.logger.record("metrics/jobs_launched", int(episode_data["jobs_launched"])) + self.logger.record("metrics/jobs_completed", int(episode_data["jobs_completed"])) + self.logger.record("metrics/completion_rate", float(episode_data["completion_rate"])) + self.logger.record("metrics/avg_wait_hours", float(episode_data["avg_wait_time"])) + self.logger.record("metrics/on_nodes", int(episode_data.get("on_nodes_end", 0))) + self.logger.record("metrics/used_nodes", int(episode_data.get("used_nodes_end", 0))) + self.logger.record("metrics/max_queue_size", int(episode_data["max_queue_size"])) + self.logger.record("metrics/max_backlog_size", int(episode_data["max_backlog_size"])) + self.logger.record("metrics/max_drop_streak", int(episode_data.get("max_drop_streak", 0))) + self.logger.record("metrics/pending_jobs_end", int(episode_data.get("pending_jobs_end", 0))) + self.logger.record("metrics/pending_core_hours_end", float(episode_data.get("pending_core_hours_end", 0.0))) + self.logger.record("metrics/overdue_jobs_end", int(episode_data.get("overdue_jobs_end", 0))) + self.logger.record("metrics/jobs_dropped", int(episode_data["jobs_dropped"])) + self.logger.record("metrics/jobs_flushed", int(episode_data.get("jobs_flushed", 0))) + self.logger.record("metrics/jobs_lost_total", int(episode_data["jobs_lost_total"])) + self.logger.record("metrics/loss_rate", float(episode_data["loss_rate"])) + self.logger.record("metrics/jobs_rejected_queue_full", int(episode_data["jobs_rejected_queue_full"])) + + # Job metrics (baseline) + self.logger.record("metrics/baseline_jobs_submitted", int(episode_data["baseline_jobs_submitted"])) + self.logger.record("metrics/baseline_jobs_launched", int(episode_data["baseline_jobs_launched"])) + self.logger.record("metrics/baseline_jobs_completed", int(episode_data["baseline_jobs_completed"])) + self.logger.record("metrics/baseline_completion_rate", float(episode_data["baseline_completion_rate"])) + self.logger.record("metrics/baseline_avg_wait_hours", float(episode_data["baseline_avg_wait_time"])) + self.logger.record("metrics/baseline_max_queue_size", int(episode_data["baseline_max_queue_size"])) + self.logger.record("metrics/baseline_max_backlog_size", int(episode_data["baseline_max_backlog_size"])) + self.logger.record("metrics/baseline_jobs_dropped", int(episode_data["baseline_jobs_dropped"])) + self.logger.record("metrics/baseline_jobs_flushed", int(episode_data.get("baseline_jobs_flushed", 0))) + self.logger.record("metrics/baseline_jobs_lost_total", int(episode_data["baseline_jobs_lost_total"])) + self.logger.record("metrics/baseline_loss_rate", float(episode_data["baseline_loss_rate"])) + self.logger.record("metrics/baseline_jobs_rejected_queue_full", int(episode_data["baseline_jobs_rejected_queue_full"])) + + # Proportional (per-core) power metrics + self.logger.record("metrics/prop_power_mwh", float(episode_data["agent_prop_power_mwh"])) + self.logger.record("metrics/baseline_prop_power_mwh", float(episode_data["baseline_prop_power_mwh"])) + self.logger.record("metrics/baseline_off_prop_power_mwh", float(episode_data["baseline_off_prop_power_mwh"])) + self.logger.record( + "metrics/savings_prop_power_vs_baseline_off", + float(episode_data["baseline_off_prop_power_mwh"]) - float(episode_data["agent_prop_power_mwh"]), + ) + self.logger.record( + "metrics/savings_prop_cost_vs_baseline_off", + float(episode_data["savings_prop_cost_vs_baseline_off"]), + ) return True diff --git a/src/evaluation_summary.py b/src/evaluation_summary.py index 7fdb147..2572094 100644 --- a/src/evaluation_summary.py +++ b/src/evaluation_summary.py @@ -60,6 +60,7 @@ def build_episode_summary_line( f"PendingEnd={int(episode_data.get('pending_jobs_end', 0))}, " f"OverdueEnd={int(episode_data.get('overdue_jobs_end', 0))}, " f"EpisodeMaxQueue={int(episode_data['max_queue_size'])}, " + f"MaxDropStreak={int(episode_data.get('max_drop_streak', 0))}, " f"Lost={int(episode_data.get('jobs_lost_total', episode_data['jobs_dropped']))}, " f"TimelineMaxQueue={timeline_max_queue}, " f"Agent Occupancy (Cores)={agent_occupancy_cores_pct:.2f}%, " diff --git a/src/metrics_tracker.py b/src/metrics_tracker.py index 4da9c13..580bf8b 100644 --- a/src/metrics_tracker.py +++ b/src/metrics_tracker.py @@ -99,6 +99,7 @@ def reset_episode_metrics(self) -> None: self.episode_max_backlog_size_reached: int = 0 self.episode_jobs_dropped: int = 0 self.episode_jobs_flushed: int = 0 + self.episode_max_drop_streak: int = 0 self.episode_jobs_rejected_queue_full: int = 0 # Baseline job metrics (episode) @@ -304,8 +305,11 @@ def record_episode_completion(self, current_episode: int) -> dict[str, float | i 'jobs_completed': self.episode_jobs_completed, 'avg_wait_time': avg_wait_time, 'completion_rate': completion_rate, + 'on_nodes_end': int(self.episode_on_nodes[-1]) if self.episode_on_nodes else 0, + 'used_nodes_end': int(self.episode_used_nodes[-1]) if self.episode_used_nodes else 0, 'max_queue_size': self.episode_max_queue_size_reached, 'max_backlog_size': self.episode_max_backlog_size_reached, + 'max_drop_streak': self.episode_max_drop_streak, 'pending_jobs_end': self.episode_pending_jobs_end, 'pending_core_demand_end': self.episode_pending_core_demand_end, 'pending_core_hours_end': self.episode_pending_core_hours_end, diff --git a/src/reward_calculation.py b/src/reward_calculation.py index 28d8857..56b6891 100644 --- a/src/reward_calculation.py +++ b/src/reward_calculation.py @@ -472,21 +472,16 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo total_used_cores, ) - intrinsic_starvation_reward = self._reward_intrinsic_starvation( - remaining_overdue_age_core_hours, - total_used_cores, - ) - - job_age_penalty_weighted = weights.job_age_weight * job_age_penalty_norm + intrinsic_starvation_reward - # 4. Intrinsic anti-starvation signal: once work is overdue, staying near # zero throughput should be structurally worse than doing at least some work. intrinsic_starvation_reward = self._reward_intrinsic_starvation( remaining_overdue_age_core_hours, total_used_cores, ) + + job_age_penalty_weighted = weights.job_age_weight * job_age_penalty_norm + intrinsic_starvation_reward - # 4. penalty for idling nodes + # 5. penalty for idling nodes idle_penalty_norm = self._penalty_idle_normalized(num_idle_nodes) idle_penalty_weighted = weights.idle_weight * idle_penalty_norm @@ -506,7 +501,7 @@ def calculate(self, num_used_nodes: int, num_idle_nodes: int, current_price: flo f" > $$$TOTAL: {reward:.4f} = " f"{efficiency_reward_weighted:.4f} + {price_reward_weighted:.4f} + " f"{idle_penalty_weighted:.4f} + {job_age_penalty_weighted:.4f} + " - f"{intrinsic_starvation_reward:.4f} + {drop_penalty_weighted:.4f}" + f"{drop_penalty_weighted:.4f}" ) env_print(f" > step cost: €{total_cost:.4f}")