From 1b6d98ceeb6d137f8ef794a91660172380eba39b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=A0=95=EC=8B=9C=EC=9B=90?= Date: Thu, 7 May 2026 01:23:26 +0900 Subject: [PATCH] =?UTF-8?q?Deep=20CFR=20traversal=20=EC=9A=B4=EC=98=81=20?= =?UTF-8?q?=EB=A1=9C=EA=B7=B8=20=EB=B3=B4=EA=B0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ..._selfplay_full_depth_slot_playability.yaml | 1 + .../games/classic/deep_cfr/config.py | 7 +- .../games/classic/deep_cfr/run_logger.py | 29 +++++++++ .../games/classic/deep_cfr/trainer.py | 64 ++++++++++++++++--- .../games/classic/deep_cfr/workers.py | 20 ++++++ tests/games/classic/test_deep_cfr_trainer.py | 14 +++- 6 files changed, 121 insertions(+), 14 deletions(-) create mode 100644 src/coolrl_lost_cities/games/classic/deep_cfr/run_logger.py diff --git a/configs/deep_cfr/deep_cfr_selfplay_full_depth_slot_playability.yaml b/configs/deep_cfr/deep_cfr_selfplay_full_depth_slot_playability.yaml index 453a933..2e9b3e6 100644 --- a/configs/deep_cfr/deep_cfr_selfplay_full_depth_slot_playability.yaml +++ b/configs/deep_cfr/deep_cfr_selfplay_full_depth_slot_playability.yaml @@ -37,6 +37,7 @@ traversal: cutoff_rollouts: 0 cutoff_rollout_policy: random cutoff_rollout_max_steps: 300 + progress_every_traversals: 10 num_workers: 8 traversal_worker_chunk_size: 8 regret_matching_epsilon: 0.0001 diff --git a/src/coolrl_lost_cities/games/classic/deep_cfr/config.py b/src/coolrl_lost_cities/games/classic/deep_cfr/config.py index 29687e3..34aea0f 100644 --- a/src/coolrl_lost_cities/games/classic/deep_cfr/config.py +++ b/src/coolrl_lost_cities/games/classic/deep_cfr/config.py @@ -104,6 +104,7 @@ class TraversalConfig(StrictModel): num_workers: int | str = 0 worker_chunk_size: int = 4 traversal_worker_chunk_size: int | None = None + progress_every_traversals: int = 0 endpoint_depth_bucket_width: int = 100 endpoint_depth_bucket_max: int = 1000 @@ -141,8 +142,10 @@ class TraversalConfig(StrictModel): if token == "auto": guess = max(1, (os.cpu_count() or 2) // 2) return min(guess, batches) if batches is not None and batches > 0 else guess - return max(0, int(token)) - return max(0, int(self.num_workers)) + workers = max(0, int(token)) + else: + workers = max(0, int(self.num_workers)) + return min(workers, batches) if batches is not None and batches > 0 else workers def resolved_traversals_per_player(self) -> int: if self.traversals_per_player is not None: diff --git a/src/coolrl_lost_cities/games/classic/deep_cfr/run_logger.py b/src/coolrl_lost_cities/games/classic/deep_cfr/run_logger.py new file mode 100644 index 0000000..72e6d9b --- /dev/null +++ b/src/coolrl_lost_cities/games/classic/deep_cfr/run_logger.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +from datetime import datetime +from pathlib import Path +from typing import Protocol + + +class RunLogger(Protocol): + def info(self, message: str) -> None: + """Record a low-frequency training log message.""" + + +def log_timestamp() -> str: + return datetime.now().astimezone().isoformat(timespec="seconds") + + +class NullRunLogger: + def info(self, message: str) -> None: + pass + + +class FileRunLogger: + def __init__(self, path: str | Path): + self.path = Path(path) + + def info(self, message: str) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + with self.path.open("a", encoding="utf-8") as handle: + handle.write(f"{log_timestamp()} {message}\n") diff --git a/src/coolrl_lost_cities/games/classic/deep_cfr/trainer.py b/src/coolrl_lost_cities/games/classic/deep_cfr/trainer.py index b875534..517dd80 100644 --- a/src/coolrl_lost_cities/games/classic/deep_cfr/trainer.py +++ b/src/coolrl_lost_cities/games/classic/deep_cfr/trainer.py @@ -20,6 +20,7 @@ from coolrl_lost_cities.games.classic.deep_cfr.encoding import input_dim from coolrl_lost_cities.games.classic.deep_cfr.evaluate import evaluate_strategy_network from coolrl_lost_cities.games.classic.deep_cfr.memory import ReservoirMemory, TrainingSample from coolrl_lost_cities.games.classic.deep_cfr.networks import DeepCFRMLP +from coolrl_lost_cities.games.classic.deep_cfr.run_logger import FileRunLogger, RunLogger from coolrl_lost_cities.games.classic.deep_cfr.traverser import DeepCFRTraverser, TraversalStats from coolrl_lost_cities.games.classic.deep_cfr.workers import ( TraversalWorkerBatch, @@ -84,6 +85,7 @@ class DeepCFRTrainer: game_config: LostCitiesConfig | None = None, *, device: str = "cpu", + run_logger: RunLogger | None = None, ) -> None: self.config = config or DeepCFRConfig() self.game_config = game_config or self.config.rules.to_lost_cities_config( @@ -126,6 +128,7 @@ class DeepCFRTrainer: self.metrics_path = self.run_dir / "metrics.jsonl" self.progress_path = self.run_dir / "runtime_progress.json" self.log_path = self.run_dir / "train.log" + self.run_logger = run_logger or FileRunLogger(self.log_path) self.self_play_league_snapshots: list[list[dict]] = [] def checkpoint_payload(self, metrics: IterationMetrics | None = None) -> dict: @@ -228,12 +231,23 @@ class DeepCFRTrainer: ) for network in self.advantage_networks: network.eval() + progress_every = int(self.config.traversal.progress_every_traversals) + completed = 0 + progress_started = time.perf_counter() for traversal_index in range(self.config.traversal.resolved_traversals_per_player()): for player in range(2): seed = self.config.run.seed + iteration * 10_000 + traversal_index * 10 + player state = GameState.new_game(self.game_config, seed=seed) _, stats = traverser.traverse(state, player, iteration) total_stats.accumulate(stats) + completed += 1 + if progress_every > 0 and completed % progress_every == 0: + elapsed = time.perf_counter() - progress_started + self.run_logger.info( + f"Traversal progress iteration={iteration} completed={completed} " + f"elapsed_seconds={elapsed:.2f} total_nodes={total_stats.nodes} " + f"nodes_per_second={total_stats.nodes / max(elapsed, 1.0e-12):.1f}" + ) return total_stats def _run_traversals_parallel(self, iteration: int) -> TraversalStats: @@ -241,17 +255,48 @@ class DeepCFRTrainer: total_stats = TraversalStats() if not batches: return total_stats + requested_workers = self.config.traversal.resolved_num_workers() max_workers = self.config.traversal.resolved_num_workers(len(batches)) + self.run_logger.info( + f"Traversal multiprocessing enabled iteration={iteration} " + f"requested_workers={requested_workers} effective_workers={max_workers} " + f"batches={len(batches)} chunk_size={self.config.traversal.resolved_worker_chunk_size()}" + ) + if max_workers < requested_workers: + self.run_logger.info( + f"Traversal worker count capped iteration={iteration} " + f"requested_workers={requested_workers} effective_workers={max_workers} " + f"available_batches={len(batches)}" + ) + progress_every = int(self.config.traversal.progress_every_traversals) + next_progress_at = progress_every if progress_every > 0 else None + progress_nodes = 0 + progress_traversals = 0 + progress_started = time.perf_counter() with ProcessPoolExecutor( max_workers=max_workers, mp_context=mp.get_context("spawn"), ) as executor: futures = [executor.submit(run_traversal_worker_batch, batch) for batch in batches] - for future in as_completed(futures): + total_batches = len(futures) + for completed_batches, future in enumerate(as_completed(futures), start=1): result = future.result() total_stats.accumulate(result.stats) self.advantage_memory.extend(result.advantage_samples, self.rng) self.strategy_memory.extend(result.strategy_samples, self.rng) + progress_nodes += result.stats.nodes + progress_traversals += result.traversals + if next_progress_at is not None and progress_traversals >= next_progress_at: + elapsed = time.perf_counter() - progress_started + self.run_logger.info( + f"Traversal multiprocessing progress iteration={iteration} " + f"completed_batches={completed_batches}/{total_batches} " + f"completed_traversals={progress_traversals} elapsed_seconds={elapsed:.2f} " + f"total_nodes={progress_nodes} " + f"nodes_per_second={progress_nodes / max(elapsed, 1.0e-12):.1f}" + ) + while next_progress_at is not None and next_progress_at <= progress_traversals: + next_progress_at += progress_every return total_stats def _worker_batches(self, iteration: int) -> list[TraversalWorkerBatch]: @@ -378,10 +423,9 @@ class DeepCFRTrainer: ) if self.iteration == 0 and self.metrics_path.exists(): self.metrics_path.unlink() - with self.log_path.open("a", encoding="utf-8") as handle: - handle.write( - f"Deep CFR run start iteration={self.iteration} seed={self.config.run.seed}\n" - ) + self.run_logger.info( + f"Deep CFR run start iteration={self.iteration} seed={self.config.run.seed}" + ) def _append_metrics(self, metrics: IterationMetrics, iteration_seconds: float) -> None: data = metrics.to_dict() @@ -390,11 +434,11 @@ class DeepCFRTrainer: with self.metrics_path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(data, sort_keys=True) + "\n") self.progress_path.write_text(json.dumps(data, indent=2, sort_keys=True), encoding="utf-8") - with self.log_path.open("a", encoding="utf-8") as handle: - handle.write( - f"iteration={metrics.iteration} nodes={metrics.traversal_nodes} adv_loss={metrics.advantage_loss:.6f} " - f"strategy_loss={metrics.strategy_loss:.6f} seconds={iteration_seconds:.3f}\n" - ) + self.run_logger.info( + f"iteration={metrics.iteration} nodes={metrics.traversal_nodes} " + f"adv_loss={metrics.advantage_loss:.6f} " + f"strategy_loss={metrics.strategy_loss:.6f} seconds={iteration_seconds:.3f}" + ) def _evaluate(self, iteration: int) -> dict[str, float | int]: if ( diff --git a/src/coolrl_lost_cities/games/classic/deep_cfr/workers.py b/src/coolrl_lost_cities/games/classic/deep_cfr/workers.py index 0dce38b..6d031eb 100644 --- a/src/coolrl_lost_cities/games/classic/deep_cfr/workers.py +++ b/src/coolrl_lost_cities/games/classic/deep_cfr/workers.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os from dataclasses import dataclass from typing import Any @@ -12,6 +13,23 @@ from coolrl_lost_cities.games.classic.deep_cfr.networks import DeepCFRMLP from coolrl_lost_cities.games.classic.deep_cfr.traverser import DeepCFRTraverser, TraversalStats from coolrl_lost_cities.games.classic.game import GameState, LostCitiesConfig +_TORCH_THREADS_CONFIGURED = False + + +def _configure_worker_torch_threads() -> None: + global _TORCH_THREADS_CONFIGURED + if _TORCH_THREADS_CONFIGURED: + return + os.environ.setdefault("OMP_NUM_THREADS", "1") + os.environ.setdefault("MKL_NUM_THREADS", "1") + torch.set_num_threads(1) + if hasattr(torch, "set_num_interop_threads"): + try: + torch.set_num_interop_threads(1) + except RuntimeError: + pass + _TORCH_THREADS_CONFIGURED = True + @dataclass(frozen=True) class TraversalWorkerBatch: @@ -37,6 +55,8 @@ class TraversalWorkerResult: def run_traversal_worker_batch(batch: TraversalWorkerBatch) -> TraversalWorkerResult: + _configure_worker_torch_threads() + cfg = config_from_dict(batch.config) device = torch.device("cpu") networks = [ diff --git a/tests/games/classic/test_deep_cfr_trainer.py b/tests/games/classic/test_deep_cfr_trainer.py index 9c0c610..f8eddd2 100644 --- a/tests/games/classic/test_deep_cfr_trainer.py +++ b/tests/games/classic/test_deep_cfr_trainer.py @@ -1,5 +1,7 @@ from __future__ import annotations +import re + import numpy as np from coolrl_lost_cities.games.classic.deep_cfr.encoding import encode_info_state, input_dim from coolrl_lost_cities.games.classic.game import GameState, LostCitiesConfig @@ -46,6 +48,7 @@ def test_deep_cfr_loads_mapped_legacy_reproduction_config() -> None: assert config.traversal.max_depth is None assert config.traversal.resolved_max_nodes() == 1000 assert config.traversal.resolved_worker_chunk_size() == 8 + assert config.traversal.progress_every_traversals == 10 assert config.optimization.resolved_advantage_batch_size() == 1024 assert config.optimization.resolved_strategy_batch_size() == 1024 assert config.optimization.resolved_advantage_train_steps() == 256 @@ -333,6 +336,8 @@ def test_deep_cfr_trainer_saves_loads_and_evaluates_checkpoint(tmp_path) -> None assert (checkpoint_dir / "metrics.jsonl").exists() assert (checkpoint_dir / "runtime_progress.json").exists() assert (checkpoint_dir / "train.log").exists() + train_log = (checkpoint_dir / "train.log").read_text(encoding="utf-8") + assert re.search(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}", train_log) assert restored.iteration == 1 assert "eval_random_games" in metrics[0].eval_metrics assert "eval_random_play_action_rate" in metrics[0].eval_metrics @@ -353,11 +358,12 @@ def test_deep_cfr_trainer_multiprocessing_smoke_run(tmp_path) -> None: "run": {"iterations": 1, "seed": 43}, "network": {"hidden_size": 16}, "traversal": { - "traversals_per_iteration": 2, + "traversals_per_iteration": 1, "max_depth": 2, "max_nodes": 32, - "num_workers": 2, + "num_workers": 8, "worker_chunk_size": 1, + "progress_every_traversals": 1, }, "optimization": {"batch_size": 2}, "checkpoint": { @@ -370,9 +376,13 @@ def test_deep_cfr_trainer_multiprocessing_smoke_run(tmp_path) -> None: ) metrics = trainer.train() + train_log = (tmp_path / "mp" / "train.log").read_text(encoding="utf-8") assert metrics[0].traversal_nodes > 0 assert metrics[0].advantage_samples > 0 + assert "Traversal multiprocessing enabled" in train_log + assert "Traversal worker count capped" in train_log + assert "Traversal multiprocessing progress" in train_log def test_deep_cfr_traversal_benchmark_smoke() -> None: