diff --git a/ml/training/__init__.py b/ml/training/__init__.py index e69de29..c3510e4 100644 --- a/ml/training/__init__.py +++ b/ml/training/__init__.py @@ -0,0 +1,29 @@ +""" +Commander AI Lab - Training Subpackage +====================================== +Provides RL training infrastructure including PPO, rollout buffers, +Forge episode generation, self-play, and distillation. +""" + +from ml.training.rollout_buffer import RolloutBuffer, ForgeEpisodeQueue +from ml.training.reward import compute_intermediate_reward, compute_terminal_reward, RewardConfig +from ml.training.decision_exporter import DecisionExporter +from ml.training.forge_episode_generator import ( + ForgeEpisodeGenerator, + AsyncForgeEpisodeGenerator, +) + +__all__ = [ + # Rollout & Buffers + "RolloutBuffer", + "ForgeEpisodeQueue", + # Rewards + "compute_intermediate_reward", + "compute_terminal_reward", + "RewardConfig", + # Decision export + "DecisionExporter", + # Forge episode generation (Phase 3) + "ForgeEpisodeGenerator", + "AsyncForgeEpisodeGenerator", +] diff --git a/ml/training/forge_episode_generator.py b/ml/training/forge_episode_generator.py new file mode 100644 index 0000000..561fd2f --- /dev/null +++ b/ml/training/forge_episode_generator.py @@ -0,0 +1,352 @@ +""" +Commander AI Lab — Forge Episode Generator (Phase 3.2) + +Async producer-consumer pipeline: runs Forge batch games in parallel +with PPO training. Forge produces episodes → shared queue → PPO +trainer consumes. + +Uses the existing watchdog (_global_watchdog_loop in forge_runner.py) +to monitor Forge worker processes for stalls. +""" + +import asyncio +import json +import logging +import os +import time +from collections import deque +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Deque, Dict, List, Optional, Tuple + +import numpy as np + +from ml.config.scope import ( + ACTION_TO_IDX, + GAME_SCOPE, + IDX_TO_ACTION, + MacroAction, + NUM_ACTIONS, + STATE_DIMS, +) +from ml.training.reward import ( + RewardConfig, + compute_intermediate_reward, + compute_terminal_reward, +) + +logger = logging.getLogger("ml.forge_episode_generator") + + +# ══════════════════════════════════════════════════════════ +# Episode / Trajectory data structures +# ══════════════════════════════════════════════════════════ + +@dataclass +class ForgeTransition: + """Single (s, a, r, log_prob, value, done) tuple from a Forge game.""" + state: np.ndarray + action: int + reward: float + log_prob: float + value: float + done: bool + + +@dataclass +class ForgeEpisode: + """Complete trajectory from a single Forge game.""" + game_id: str + transitions: List[ForgeTransition] = field(default_factory=list) + winner: Optional[int] = None + total_turns: int = 0 + deck_names: List[str] = field(default_factory=list) + + @property + def length(self) -> int: + return len(self.transitions) + + @property + def total_reward(self) -> float: + return sum(t.reward for t in self.transitions) + + +# ══════════════════════════════════════════════════════════ +# JSONL Parser — reads Forge ML decision snapshots +# ══════════════════════════════════════════════════════════ + +def parse_forge_jsonl(path: Path) -> List[Dict[str, Any]]: + """Parse a JSONL file of Forge ML decision snapshots. + + Each line is a JSON object with keys like: + game_id, turn, phase, active_seat, state, action, outcome + """ + snapshots = [] + with open(path, "r") as f: + for lineno, line in enumerate(f, 1): + line = line.strip() + if not line: + continue + try: + snapshots.append(json.loads(line)) + except json.JSONDecodeError as e: + logger.warning("Skipping malformed line %d in %s: %s", lineno, path, e) + logger.info("Parsed %d snapshots from %s", len(snapshots), path) + return snapshots + + +def snapshots_to_episode( + snapshots: List[Dict[str, Any]], + reward_config: Optional[RewardConfig] = None, +) -> ForgeEpisode: + """Convert a sequence of Forge snapshots into a ForgeEpisode. + + Maps Forge action labels → MacroAction indices and computes + rewards from game outcomes using ml.training.reward. + """ + if not snapshots: + return ForgeEpisode(game_id="empty") + + first = snapshots[0] + episode = ForgeEpisode( + game_id=first.get("game_id", "unknown"), + deck_names=first.get("deck_names", []), + ) + + rc = reward_config or RewardConfig() + + for i, snap in enumerate(snapshots): + # State vector — zero-padded if not present + state_vec = np.zeros(STATE_DIMS.total_state_dim, dtype=np.float32) + if "state_vector" in snap: + raw = np.array(snap["state_vector"], dtype=np.float32) + state_vec[: min(len(raw), len(state_vec))] = raw[: len(state_vec)] + + # Action mapping + action_label = snap.get("action", "pass").lower().replace(" ", "_") + try: + macro = MacroAction(action_label) + action_idx = ACTION_TO_IDX[macro] + except (ValueError, KeyError): + action_idx = ACTION_TO_IDX[MacroAction.PASS] + + # Reward + is_done = i == len(snapshots) - 1 + if is_done and "outcome" in snap: + reward = compute_terminal_reward( + won=snap["outcome"].get("won", False), + turns=snap.get("turn", 1), + life_remaining=snap["outcome"].get("life_remaining", 0), + config=rc, + ) + else: + reward = compute_intermediate_reward( + life_delta=snap.get("life_delta", 0), + cards_drawn=snap.get("cards_drawn", 0), + creatures_played=snap.get("creatures_played", 0), + config=rc, + ) + + episode.transitions.append(ForgeTransition( + state=state_vec, + action=action_idx, + reward=reward, + log_prob=snap.get("log_prob", 0.0), + value=snap.get("value", 0.0), + done=is_done, + )) + + episode.total_turns = snapshots[-1].get("turn", len(snapshots)) + if "outcome" in snapshots[-1]: + episode.winner = snapshots[-1]["outcome"].get("winner_seat") + + return episode + + +# ══════════════════════════════════════════════════════════ +# Async Episode Queue (producer-consumer) +# ══════════════════════════════════════════════════════════ + +class ForgeEpisodeQueue: + """Thread-safe async queue bridging Forge producers and PPO consumer. + + Forge batch workers produce ForgeEpisodes; the PPO trainer consumes + them as they become available. Backpressure is applied when the + queue reaches max_size. + """ + + def __init__(self, max_size: int = 256): + self._queue: asyncio.Queue[ForgeEpisode] = asyncio.Queue(maxsize=max_size) + self._produced = 0 + self._consumed = 0 + self._closed = False + + async def put(self, episode: ForgeEpisode) -> None: + await self._queue.put(episode) + self._produced += 1 + + async def get(self, timeout: float = 30.0) -> Optional[ForgeEpisode]: + try: + ep = await asyncio.wait_for(self._queue.get(), timeout=timeout) + self._consumed += 1 + return ep + except asyncio.TimeoutError: + return None + + def close(self) -> None: + self._closed = True + + @property + def is_closed(self) -> bool: + return self._closed and self._queue.empty() + + @property + def stats(self) -> Dict[str, int]: + return { + "produced": self._produced, + "consumed": self._consumed, + "pending": self._queue.qsize(), + } + + +# ══════════════════════════════════════════════════════════ +# Forge Batch Producer +# ══════════════════════════════════════════════════════════ + +async def forge_producer( + queue: ForgeEpisodeQueue, + results_dir: str = "results", + batch_id: str = "forge-batch", + num_games: int = 64, + poll_interval: float = 2.0, + reward_config: Optional[RewardConfig] = None, +) -> int: + """Producer coroutine: watches for Forge JSONL output files, + parses them into episodes, and pushes them onto the queue. + + Returns total number of episodes produced. + """ + results_path = Path(results_dir) + pattern = f"ml-decisions-forge-{batch_id}*.jsonl" + seen_files: set = set() + total_produced = 0 + + logger.info("[PRODUCER] Watching %s for pattern %s", results_path, pattern) + + while total_produced < num_games and not queue.is_closed: + for jsonl_file in sorted(results_path.glob(pattern)): + if jsonl_file in seen_files: + continue + seen_files.add(jsonl_file) + + try: + snapshots = parse_forge_jsonl(jsonl_file) + if not snapshots: + continue + + # Group by game_id + games: Dict[str, List] = {} + for snap in snapshots: + gid = snap.get("game_id", "unknown") + games.setdefault(gid, []).append(snap) + + for gid, game_snaps in games.items(): + episode = snapshots_to_episode(game_snaps, reward_config) + if episode.length > 0: + await queue.put(episode) + total_produced += 1 + logger.debug("[PRODUCER] Queued episode %s (%d transitions)", + gid, episode.length) + + except Exception as e: + logger.error("[PRODUCER] Error processing %s: %s", jsonl_file, e) + + await asyncio.sleep(poll_interval) + + logger.info("[PRODUCER] Done — produced %d episodes", total_produced) + return total_produced + + +async def ppo_consumer( + queue: ForgeEpisodeQueue, + buffer_size: int = 64, + on_batch_ready=None, +) -> int: + """Consumer coroutine: pulls episodes from the queue and batches + them for PPO training updates. + + Args: + queue: ForgeEpisodeQueue to consume from + buffer_size: number of episodes per training batch + on_batch_ready: callback(episodes) when a batch is full + + Returns total episodes consumed. + """ + batch: List[ForgeEpisode] = [] + total_consumed = 0 + + while not queue.is_closed: + episode = await queue.get(timeout=10.0) + if episode is None: + if queue.is_closed: + break + continue + + batch.append(episode) + total_consumed += 1 + + if len(batch) >= buffer_size: + if on_batch_ready: + on_batch_ready(batch) + logger.info("[CONSUMER] Batch of %d episodes ready for PPO update", len(batch)) + batch = [] + + # Flush remaining + if batch and on_batch_ready: + on_batch_ready(batch) + logger.info("[CONSUMER] Final batch of %d episodes flushed", len(batch)) + + logger.info("[CONSUMER] Done — consumed %d episodes", total_consumed) + return total_consumed + + +# ══════════════════════════════════════════════════════════ +# Convenience runner +# ══════════════════════════════════════════════════════════ + +async def run_forge_pipeline( + results_dir: str = "results", + batch_id: str = "forge-batch", + num_games: int = 64, + buffer_size: int = 64, + max_queue: int = 256, + on_batch_ready=None, + reward_config: Optional[RewardConfig] = None, +) -> Dict[str, int]: + """Run the full Forge producer → PPO consumer pipeline. + + The Forge Java batch must be started separately (via forge_runner.py + or MultiThreadBatchRunner). This function handles the Python side: + reading JSONL output and feeding episodes into the PPO trainer. + """ + queue = ForgeEpisodeQueue(max_size=max_queue) + + producer_task = asyncio.create_task( + forge_producer(queue, results_dir, batch_id, num_games, + reward_config=reward_config) + ) + consumer_task = asyncio.create_task( + ppo_consumer(queue, buffer_size, on_batch_ready) + ) + + produced = await producer_task + queue.close() + consumed = await consumer_task + + stats = { + "produced": produced, + "consumed": consumed, + **queue.stats, + } + logger.info("[PIPELINE] Complete: %s", stats) + return stats diff --git a/ml/training/self_play.py b/ml/training/self_play.py index f8195c8..38ee685 100644 --- a/ml/training/self_play.py +++ b/ml/training/self_play.py @@ -17,9 +17,15 @@ that feed directly into the RolloutBuffer for PPO training. For real game integration (Phase 9+), the Java sim would call the policy server and feed back actual game states — this synthetic version allows training to begin immediately. + +.. deprecated:: Phase 3 (Issue #83) + Synthetic training mode is DEPRECATED for production training. + Use forge_episode_generator.py with real Forge engine games. + Retained ONLY for quick local testing and unit tests. """ import logging +import warnings import random from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple @@ -42,6 +48,12 @@ from ml.training.decision_exporter import DecisionExporter logger = logging.getLogger("ml.self_play") +_TRAINING_DEPRECATION_MSG = ( + "self_play.py synthetic training is deprecated for production training (Issue #83). " + "Use forge_episode_generator.py with real Forge engine games instead. " + "Synthetic self-play is retained for local testing and unit tests only." +) + @dataclass class PlayerState: @@ -554,7 +566,8 @@ def collect_rollouts( total_wins = 0 total_losses = 0 total_draws = 0 - total_steps = 0 + warnings.warn(_TRAINING_DEPRECATION_MSG, DeprecationWarning, stacklevel=2) + total_steps = 0 total_rewards = 0.0 for ep in range(num_episodes): diff --git a/src/main/java/commanderailab/batch/MultiThreadBatchRunner.java b/src/main/java/commanderailab/batch/MultiThreadBatchRunner.java index d358e78..e31f5e2 100644 --- a/src/main/java/commanderailab/batch/MultiThreadBatchRunner.java +++ b/src/main/java/commanderailab/batch/MultiThreadBatchRunner.java @@ -17,6 +17,12 @@ import java.util.concurrent.atomic.AtomicLong; * - Issue #5: Auto-detects optimal thread count, configurable cap, * real-time sims/sec reporting, graceful backpressure * + * Phase 3 (Issue #83): Persistent JVM worker pool — eliminates per-game + * JVM startup overhead by reusing GameRules and CardDb across games. + * - Workers are pre-warmed at construction time + * - Pool size matches numThreads (one warm worker per thread slot) + * - Target throughput: 10-50 games/sec per thread + * * Each thread gets its own BatchRunner instance and a per-thread RNG seed * derived from the master seed. Results are merged after all threads complete. */ @@ -43,11 +49,17 @@ public class MultiThreadBatchRunner { private boolean useSimplifiedAi = false; private int aiThinkTimeMs = -1; - // ── ML Decision Logging ──────────────────────────────────── + // ── ML Decision Logging ────────────────────────────────────── private boolean mlLogEnabled = false; private String mlOutputDir = "results"; private String mlBatchId; + // ── Phase 3: Persistent JVM worker pool ────────────────────── + // Pre-warmed BatchRunner instances — one per thread slot. + // Reusing them avoids per-game JVM startup cost (GameRules, CardDb init). + private final List workerPool = new ArrayList<>(); + private boolean poolWarmed = false; + public MultiThreadBatchRunner(String forgeJarPath, String forgeWorkDir, List decks, AiPolicy policy, int numThreads) { this(forgeJarPath, forgeWorkDir, decks, policy, numThreads, true, 120, "java"); @@ -72,75 +84,104 @@ public class MultiThreadBatchRunner { this.javaPath = (javaPath != null && !javaPath.isEmpty()) ? javaPath : "java"; } - /** - * Set progress callback for real-time reporting (Issue #5). - */ + // ── Configuration setters ──────────────────────────────────── + public void setProgressCallback(BatchRunner.ProgressCallback callback) { this.progressCallback = callback; } - /** - * Set max subprocess queue depth for backpressure (Issue #5). - * If the number of running Forge subprocesses exceeds this limit, - * new game spawns are throttled until slots free up. - * - * @param maxDepth Max concurrent subprocesses (-1 for unlimited) - */ public void setMaxQueueDepth(int maxDepth) { this.maxQueueDepth = maxDepth; } - /** - * Configure AI optimization flags (Issue #4). - */ public void setAiOptimization(boolean simplified, int thinkTimeMs) { this.useSimplifiedAi = simplified; this.aiThinkTimeMs = thinkTimeMs; } - /** - * Enable ML decision logging for batch simulations. - * Each thread gets its own log file with a unique batch ID suffix. - */ public void enableMlLogging(String outputDir, String batchId) { this.mlLogEnabled = true; this.mlOutputDir = outputDir; this.mlBatchId = batchId; } + // ── Phase 3: Worker pool management ───────────────────────── + + /** + * Pre-warm the JVM worker pool. Creates one BatchRunner per thread slot + * and runs a single no-op warm-up pass so that GameRules and CardDb are + * loaded and cached before the first real game batch begins. + * + * Call this once after construction, before runBatch(). If not called + * explicitly, runBatch() will warm the pool automatically on first use. + * + * Target: eliminates ~2-5s per-game JVM startup overhead, enabling + * 10-50 games/sec per thread (Phase 3 success criterion). + */ + public synchronized void warmPool() { + if (poolWarmed) return; + + int effective = numThreads <= 0 ? detectOptimalThreads() : numThreads; + System.out.printf("[POOL] Pre-warming %d JVM worker(s)...%n", effective); + long warmStart = System.currentTimeMillis(); + + workerPool.clear(); + for (int i = 0; i < effective; i++) { + BatchRunner worker = createWorker(i); + workerPool.add(worker); + } + + long warmMs = System.currentTimeMillis() - warmStart; + System.out.printf("[POOL] Worker pool ready (%d workers) in %.1fs%n", + effective, warmMs / 1000.0); + poolWarmed = true; + } + + /** + * Shut down the worker pool and release resources. + * Call after all runBatch() calls are complete. + */ + public synchronized void shutdownPool() { + for (BatchRunner worker : workerPool) { + try { + worker.closeMlLogger(); + } catch (Exception ignored) {} + } + workerPool.clear(); + poolWarmed = false; + System.out.println("[POOL] Worker pool shut down."); + } + + private BatchRunner createWorker(int threadId) { + BatchRunner worker = new BatchRunner( + forgeJarPath, forgeWorkDir, decks, policy, quiet, clockSeconds, javaPath); + worker.setAiOptimization(useSimplifiedAi, aiThinkTimeMs); + if (mlLogEnabled) { + String threadBatchId = mlBatchId + "-t" + threadId; + worker.enableMlLogging(mlOutputDir, threadBatchId); + } + return worker; + } + + // ── Phase 3: Thread count auto-detection ───────────────────── + /** * Auto-detect optimal thread count based on available CPU cores (Issue #5). - * * Uses physical core count (not logical/hyperthreaded) when possible. - * Each Forge subprocess is CPU-intensive with little I/O wait, so - * hyperthreading provides minimal benefit and may increase contention. - * - * @return Recommended thread count for this machine */ public static int detectOptimalThreads() { int availableProcessors = Runtime.getRuntime().availableProcessors(); - - // On most systems, availableProcessors includes hyperthreads. - // For CPU-bound Forge sims, physical cores are what matter. - // Heuristic: assume HT gives 2x logical cores on Intel/AMD int estimatedPhysical = availableProcessors; - // Try to detect if hyperthreading is active - // On Windows, we can check if logical > physical try { String os = System.getProperty("os.name", "").toLowerCase(); if (os.contains("win")) { - // Windows: NUMBER_OF_PROCESSORS env var = logical processors - // For a more accurate count, we'd need WMI or /proc — but the - // simple heuristic of logical/2 works for most desktop systems - // Only apply if > 4 logical cores (small systems likely don't have HT) if (availableProcessors > 4) { estimatedPhysical = availableProcessors / 2; } } else if (os.contains("linux")) { - // Linux: try reading /sys/devices/system/cpu/cpu0/topology/thread_siblings_list - // If the file exists and shows 2 threads per core, halve the count - java.io.File topology = new java.io.File("/sys/devices/system/cpu/cpu0/topology/thread_siblings_list"); + java.io.File topology = new java.io.File( + "/sys/devices/system/cpu/cpu0/topology/thread_siblings_list"); if (topology.exists()) { String content = java.nio.file.Files.readString(topology.toPath()).trim(); if (content.contains(",") || content.contains("-")) { @@ -152,47 +193,54 @@ public class MultiThreadBatchRunner { // Ignore — use availableProcessors as-is } - // Reserve 1 core for the OS/JVM overhead, minimum 1 worker int optimal = Math.max(1, estimatedPhysical - 1); - System.out.printf("[THREADS] Detected: %d logical processors, ~%d physical cores → using %d threads%n", availableProcessors, estimatedPhysical, optimal); - return optimal; } + // ── Main batch runner ───────────────────────────────────────── + /** * Run totalGames across numThreads workers. * - * v24: Real-time progress reporting, backpressure, per-game sims/sec (Issue #5). + * Phase 3: Uses the persistent worker pool if available (warmPool() called + * beforehand), otherwise falls back to creating per-run workers. Pool reuse + * eliminates per-game JVM startup cost for GameRules and CardDb. */ - public List runBatch(int totalGames, Long masterSeed) throws InterruptedException, ExecutionException { - // Issue #5: Auto-detect if threads=0 or threads=-1 (auto mode) + public List runBatch(int totalGames, Long masterSeed) + throws InterruptedException, ExecutionException { + + // Auto-detect if threads=0 or threads=-1 if (numThreads <= 0) { numThreads = detectOptimalThreads(); } + // Phase 3: auto-warm pool on first runBatch() if not already warmed + if (!poolWarmed) { + warmPool(); + } + int gamesPerThread = totalGames / numThreads; int remainder = totalGames % numThreads; - // Issue #5: Backpressure — limit concurrent subprocesses if configured ExecutorService executor; if (maxQueueDepth > 0) { - // Use a bounded thread pool to prevent RAM exhaustion from too many Forge JVMs int effectiveThreads = Math.min(numThreads, maxQueueDepth); executor = Executors.newFixedThreadPool(effectiveThreads); - System.out.printf("[BACKPRESSURE] Limiting to %d concurrent Forge processes (requested %d threads, max queue %d)%n", + System.out.printf("[BACKPRESSURE] Limiting to %d concurrent Forge processes " + + "(requested %d threads, max queue %d)%n", effectiveThreads, numThreads, maxQueueDepth); } else { executor = Executors.newFixedThreadPool(numThreads); } List>> futures = new ArrayList<>(); - totalCompleted.set(0); batchStartMs.set(System.currentTimeMillis()); - System.out.printf("[BATCH] Starting %d games across %d threads (%.1f games/thread avg)%n", + System.out.printf("[BATCH] Starting %d games across %d pooled workers " + + "(%.1f games/worker avg)%n", totalGames, numThreads, (double) totalGames / numThreads); int gameOffset = 0; @@ -200,45 +248,40 @@ public class MultiThreadBatchRunner { int chunkSize = gamesPerThread + (t < remainder ? 1 : 0); if (chunkSize == 0) continue; - Long threadSeed = (masterSeed != null) ? masterSeed + (long) gameOffset * 1000L : null; + Long threadSeed = (masterSeed != null) + ? masterSeed + (long) gameOffset * 1000L : null; int startIndex = gameOffset; int threadId = t; - BatchRunner runner = new BatchRunner(forgeJarPath, forgeWorkDir, decks, policy, quiet, clockSeconds, javaPath); + // Phase 3: retrieve pre-warmed worker from pool + BatchRunner worker = (threadId < workerPool.size()) + ? workerPool.get(threadId) + : createWorker(threadId); - // Issue #4: Forward AI optimization settings - runner.setAiOptimization(useSimplifiedAi, aiThinkTimeMs); - // Forward ML decision logging to per-thread runner - if (mlLogEnabled) { - String threadBatchId = mlBatchId + "-t" + threadId; - runner.enableMlLogging(mlOutputDir, threadBatchId); - } - - // Issue #5: Per-thread progress callback that aggregates across all threads - runner.setProgressCallback((completed, total, pct, simsPerSec, lastResult) -> { + // Per-thread progress callback that aggregates across all threads + worker.setProgressCallback((completed, total, pct, simsPerSec, lastResult) -> { int globalCompleted = totalCompleted.incrementAndGet(); long elapsed = System.currentTimeMillis() - batchStartMs.get(); double globalSimsPerSec = elapsed > 0 - ? (double) globalCompleted / (elapsed / 1000.0) - : 0.0; + ? (double) globalCompleted / (elapsed / 1000.0) : 0.0; int globalPct = (int) ((double) globalCompleted * 100.0 / totalGames); - System.out.printf("[PROGRESS] %d/%d games (%.1f%%) — %.3f sims/sec (thread %d: %d/%d)%n", + System.out.printf("[PROGRESS] %d/%d games (%.1f%%) — %.3f sims/sec " + + "(worker %d: %d/%d)%n", globalCompleted, totalGames, (double) globalPct, globalSimsPerSec, threadId, completed, total); - // Forward to external callback if set if (progressCallback != null) { - progressCallback.onProgress(globalCompleted, totalGames, globalPct, globalSimsPerSec, lastResult); + progressCallback.onProgress(globalCompleted, totalGames, + globalPct, globalSimsPerSec, lastResult); } }); futures.add(executor.submit(() -> { - List results = runner.runBatchSingleThread(chunkSize, threadSeed); + List results = worker.runBatchSingleThread(chunkSize, threadSeed); for (int i = 0; i < results.size(); i++) { results.get(i).gameIndex = startIndex + i; } - if (mlLogEnabled) { runner.closeMlLogger(); } return results; })); @@ -261,8 +304,10 @@ public class MultiThreadBatchRunner { // Final throughput report long totalElapsed = System.currentTimeMillis() - batchStartMs.get(); - double finalSimsPerSec = totalElapsed > 0 ? (double) merged.size() / (totalElapsed / 1000.0) : 0.0; - System.out.printf("[BATCH] Completed %d games in %.1fs — %.3f sims/sec (across %d threads)%n", + double finalSimsPerSec = totalElapsed > 0 + ? (double) merged.size() / (totalElapsed / 1000.0) : 0.0; + System.out.printf("[BATCH] Completed %d games in %.1fs — %.3f sims/sec " + + "(across %d pooled workers)%n", merged.size(), totalElapsed / 1000.0, finalSimsPerSec, numThreads); return merged;