Pr1 part1

analyze_arrivalscale_occupancy.py

@@ -60,7 +60,7 @@
     r"(?P<baseline_off_cost_1k>-?[\d,]+(?:\.\d+)?|n/a)\s*€/1k.*?"
     r"Jobs=[\d,]+\/[\d,]+\s+\((?P<completion_rate>-?[\d.]+)%\),\s*"
     r"AvgWait=(?P<avg_wait>-?[\d.]+)h,.*?"
-    r"Dropped=(?P<agent_dropped>-?[\d,]+),.*?"
+    r"(?:Dropped|Lost)=(?P<agent_dropped>-?[\d,]+),.*?"
     r"Agent Occupancy \(Nodes\)=\s*(?P<occupancy>-?[\d.]+)%,\s*"
     r"Baseline Occupancy \(Nodes\)=\s*(?P<baseline_occupancy>-?[\d.]+)%",
     re.MULTILINE,
@@ -78,11 +78,11 @@
 )
 
 DROPPED_AGENT_SUMMARY_RE = re.compile(
-    r"Total Dropped Jobs \(Agent\):\s*(?P<agent>[\d,]+)"
+    r"Total (?:Dropped|Lost) Jobs \(Agent\):\s*(?P<agent>[\d,]+)"
 )
 
 DROPPED_BASELINE_SUMMARY_RE = re.compile(
-    r"Total Dropped Jobs \(Baseline\):\s*(?P<baseline>[\d,]+)"
+    r"Total (?:Dropped|Lost) Jobs \(Baseline\):\s*(?P<baseline>[\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 (

analyze_lambda_occupancy.py

@@ -53,7 +53,7 @@
     r"(?P<baseline_off_cost_1k>-?[\d,]+(?:\.\d+)?|n/a)\s*€/1k.*?"
     r"Jobs=[\d,]+\/[\d,]+\s+\((?P<completion_rate>-?[\d.]+)%\),\s*"
     r"AvgWait=(?P<avg_wait>-?[\d.]+)h,.*?"
-    r"Dropped=(?P<agent_dropped>-?[\d,]+),.*?"
+    r"(?:Dropped|Lost)=(?P<agent_dropped>-?[\d,]+),.*?"
     r"Agent Occupancy \(Nodes\)=\s*(?P<occupancy>-?[\d.]+)%,\s*"
     r"Baseline Occupancy \(Nodes\)=\s*(?P<baseline_occupancy>-?[\d.]+)%",
     re.MULTILINE,
@@ -71,11 +71,11 @@
 )
 
 DROPPED_AGENT_SUMMARY_RE = re.compile(
-    r"Total Dropped Jobs \(Agent\):\s*(?P<agent>[\d,]+)"
+    r"Total (?:Dropped|Lost) Jobs \(Agent\):\s*(?P<agent>[\d,]+)"
 )
 
 DROPPED_BASELINE_SUMMARY_RE = re.compile(
-    r"Total Dropped Jobs \(Baseline\):\s*(?P<baseline>[\d,]+)"
+    r"Total (?:Dropped|Lost) Jobs \(Baseline\):\s*(?P<baseline>[\d,]+)"
 )
 
 

analyze_seed_occupancy.py

@@ -57,7 +57,7 @@
     r"(?P<baseline_off_cost_1k>-?[\d,]+(?:\.\d+)?|n/a)\s*€/1k.*?"
     r"Jobs=[\d,]+\/[\d,]+\s+\((?P<completion_rate>-?[\d.]+)%\),\s*"
     r"AvgWait=(?P<avg_wait>-?[\d.]+)h,.*?"
-    r"Dropped=(?P<agent_dropped>-?[\d,]+),.*?"
+    r"(?:Dropped|Lost)=(?P<agent_dropped>-?[\d,]+),.*?"
     r"Agent Occupancy \(Nodes\)=\s*(?P<occupancy>-?[\d.]+)%,\s*"
     r"Baseline Occupancy \(Nodes\)=\s*(?P<baseline_occupancy>-?[\d.]+)%",
     re.MULTILINE,
@@ -75,11 +75,11 @@
 )
 
 DROPPED_AGENT_SUMMARY_RE = re.compile(
-    r"Total Dropped Jobs \(Agent\):\s*(?P<agent>[\d,]+)"
+    r"Total (?:Dropped|Lost) Jobs \(Agent\):\s*(?P<agent>[\d,]+)"
 )
 
 DROPPED_BASELINE_SUMMARY_RE = re.compile(
-    r"Total Dropped Jobs \(Baseline\):\s*(?P<baseline>[\d,]+)"
+    r"Total (?:Dropped|Lost) Jobs \(Baseline\):\s*(?P<baseline>[\d,]+)"
 )
 
 

src/callbacks.py

@@ -1,4 +1,4 @@
-from src.config import EPISODE_HOURS, MAX_QUEUE_SIZE, MAX_NODES
+from src.config import COST_IDLE_MW, COST_USED_MW, CORES_PER_NODE, MAX_NODES
 from stable_baselines3.common.callbacks import BaseCallback
 
 
@@ -24,63 +24,64 @@ 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))
+        dones = self.locals.get("dones")
+        if dones is None or not bool(dones[0]) or not env.metrics.episode_costs:
+            return True
 
-            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/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)
+        episode_data = env.metrics.episode_costs[-1]
 
-            # 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)
-            self.logger.record("metrics/jobs_submitted", env.metrics.episode_jobs_submitted)
-            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/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/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/loss_rate", loss_rate)
-            self.logger.record("metrics/jobs_rejected_queue_full", env.metrics.episode_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"]))
 
-            # 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)
-            self.logger.record("metrics/bl_jobs_submitted", env.metrics.episode_baseline_jobs_submitted)
-            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)
+        # 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"]))
 
-            # 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)
+        # 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