diff --git a/.github/workflows/release-please.yml b/.github/workflows/release-please.yml index 3eac0a1..a6b84ce 100644 --- a/.github/workflows/release-please.yml +++ b/.github/workflows/release-please.yml @@ -53,3 +53,49 @@ jobs: run: | uv build uv publish --token ${{ secrets.PYPI_TOKEN }} + + # release-please only bumps the version and changelog on master. Without this + # dev keeps whatever version it had -- it sat at 0.10.0 while 0.11.7 was on + # PyPI, so `abses.__version__` read 0.10.0 from a dev checkout. + backmerge-dev: + needs: release-please + if: ${{ needs.release-please.outputs.release_created }} + runs-on: ubuntu-latest + permissions: + contents: read + pull-requests: write + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Open a PR merging the release back into dev + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + TAG: ${{ needs.release-please.outputs.tag_name }} + run: | + set -euo pipefail + git fetch origin master dev + + if [ "$(git rev-list --count origin/dev..origin/master)" -eq 0 ]; then + echo "dev already contains master; nothing to back-merge." + exit 0 + fi + + existing=$(gh pr list --base dev --head master --state open \ + --json number --jq '.[0].number // empty') + if [ -n "$existing" ]; then + echo "Back-merge PR already open: #$existing" + exit 0 + fi + + gh pr create --base dev --head master \ + --title "chore(release): merge ${TAG} back into dev" \ + --body "Automated back-merge after ${TAG}. + + release-please commits the version bump and changelog on \`master\` only, + so \`dev\` needs this to stay in step. Merging is safe to do without review + when the diff is just \`pyproject.toml\` and \`CHANGELOG.md\`. + + Note: PRs opened by \`GITHUB_TOKEN\` do not trigger workflows, so this one + starts with no checks. Close and reopen it if you want CI to run." diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index a36f9ed..f296fed 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -45,6 +45,11 @@ jobs: os: [ubuntu-latest, macos-latest, windows-latest] python-version: ["3.11", "3.12", "3.13"] + # A hard interpreter crash (see #171) is silent without this: the job just + # reports an exit code and no Python traceback at all. + env: + PYTHONFAULTHANDLER: "1" + steps: - uses: actions/checkout@v4 @@ -68,6 +73,12 @@ jobs: run: | uv sync --dev + # Separates "importing abses crashes" from "the test run crashes", which + # the combined `make test-all` step cannot distinguish -- see #171. + - name: Import smoke test + run: | + uv run python -X faulthandler -c "import abses; print(abses.__version__)" + - name: Run Make test-all (unit + notebooks + tox) run: | uv run make test-all diff --git a/abses/core/experiment.py b/abses/core/experiment.py index 9ab16b4..7c58ef8 100644 --- a/abses/core/experiment.py +++ b/abses/core/experiment.py @@ -40,6 +40,7 @@ from hydra.core.hydra_config import HydraConf, HydraConfig from joblib import Parallel, delayed from omegaconf import DictConfig, OmegaConf +from omegaconf.errors import OmegaConfBaseException from tqdm.auto import tqdm from abses.core.job_manager import ExperimentManager @@ -100,6 +101,48 @@ def relative_path_from_to(from_path: Path, to_path: Path) -> Path: ) +# Hydra ships exactly one built-in launcher, `BasicLauncher`, and it is serial. +# Anything outside this prefix is a third-party launcher plugin, which we treat +# as parallel unless its own config says otherwise: the launchers that exist in +# practice (joblib, submitit, ray) all are, and assuming parallel only costs us +# a layer of nesting we skip. +_SERIAL_LAUNCHER_PREFIX = "hydra._internal.core_plugins." + + +def launcher_is_parallel(launcher: Optional[DictConfig]) -> bool: + """Whether a Hydra launcher config actually runs jobs concurrently. + + Hydra always populates `hydra.launcher`, for plain runs as well as for + multirun, defaulting to `BasicLauncher` -- which executes jobs in a plain + `for` loop. "A launcher is configured" therefore says nothing about + concurrency. + + Args: + launcher: The `hydra.launcher` config node, if there is one. + + Returns: + True when the launcher is a plugin that runs jobs in parallel, so the + caller should not add a second layer of processes of its own. + """ + if not launcher: + return False + target = str(launcher.get("_target_", "")) + if not target or target.startswith(_SERIAL_LAUNCHER_PREFIX): + return False + # A plugin can still be told to run serially. `n_jobs` is joblib's knob and + # joblib is the only launcher that spells it that way; `n_jobs: 1` selects + # joblib's sequential backend, which runs every job in the calling process + # exactly like BasicLauncher does. Other values -- including the -1 default + # and an absent key -- leave the launcher concurrent. + try: + n_jobs = OmegaConf.select(launcher, "n_jobs", default=None) + except OmegaConfBaseException: + # An unreadable value tells us nothing; keep the optimistic default + # rather than aborting the run from inside a predicate. + return True + return n_jobs != 1 + + def run_single( model_cls: Type[MainModelProtocol], cfg: DictConfig, @@ -228,9 +271,13 @@ def cfg(self, cfg: DictConfig | str | Path): self._cfg = cfg def _is_hydra_parallel(self) -> bool: - """检查是否在 Hydra 并行环境中""" + """Whether Hydra is already running jobs in parallel for us. + + When it is, `batch_run` runs its repeats sequentially rather than + nesting a second layer of processes inside each Hydra job. + """ if self.is_hydra_job(): - return self.hydra_config.launcher is not None + return launcher_is_parallel(self.hydra_config.launcher) return False @classmethod @@ -390,6 +437,26 @@ def _get_seed(self, run_id: int, job_id: Optional[int] = None) -> Optional[int]: r = random.Random(self._base_seed + job_id * 1000 + run_id) return r.randrange(2**32) + def _record_result( + self, result: Tuple[Tuple[int, int], Optional[int], pd.DataFrame] + ) -> None: + """Register what one `run_single` call produced. + + Both the sequential and the parallel branch of `_batch_run_repeats` + record through here, so a repeat is stored the same way whichever + branch ran it. + + Args: + result: The `(key, seed, datasets)` triple `run_single` returns. + """ + key, seed, datasets = result + self._manager.update_result( + key=key, + datasets=datasets, + seed=seed, + overrides=self.overrides, + ) + def _get_logging_mode(self) -> str: """Get logging mode from experiment configuration. @@ -501,19 +568,26 @@ def _batch_run_repeats( # Use print instead of logger to avoid writing to model run log files print(f"Repeat {run_id}: Logging to {log_path}") - run_single( - model_cls=self.model_cls, - cfg=cfg, - key=(self.job_id, run_id), - outpath=self.outpath, - seed=self._get_seed(run_id), - hooks=self._manager.hooks, - **self._extra_kwargs, + self._record_result( + run_single( + model_cls=self.model_cls, + cfg=cfg, + key=(self.job_id, run_id), + outpath=self.outpath, + seed=self._get_seed(run_id), + hooks=self._manager.hooks, + **self._extra_kwargs, + ) ) else: if number_process is None: - cpu_count = os.cpu_count() - number_process = max(1, cpu_count or 1 // 2) + # `or 1` covers os.cpu_count() returning None on exotic + # platforms. This used to read `cpu_count or 1 // 2`, where + # `1 // 2` binds first and evaluates to 0, so the default has + # always been every core rather than half of them; spelling it + # out keeps that behaviour instead of silently halving it. + cpu_count = os.cpu_count() or 1 + number_process = max(1, cpu_count) number_process = min(number_process, repeats) results = Parallel( @@ -537,13 +611,8 @@ def _batch_run_repeats( ) ) # 在主进程中批量更新结果 - for key, seed, dataset in results: - self._manager.update_result( - key=key, - datasets=dataset, - seed=seed, - overrides=self.overrides, - ) + for result in results: + self._record_result(result) def batch_run( self, diff --git a/tests/conftest.py b/tests/conftest.py index 5d5fab8..d45cd39 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -176,3 +176,18 @@ def mock_points_gdf(): "geometry": [Point(0, 0), Point(1, 1), Point(2, 2)], } return gpd.GeoDataFrame(data, crs="epsg:4326") + + +@pytest.fixture(name="reset_experiment_manager") +def reset_experiment_manager(): + """Reset the ExperimentManager singleton between tests. + + The manager is a singleton pinned to the first model class it sees, so a + test that experiments on a different model class has to clear it first. + """ + from abses.core.job_manager import ExperimentManager + + original = getattr(ExperimentManager, "_instance", None) + ExperimentManager._instance = None + yield + ExperimentManager._instance = original diff --git a/tests/core/test_experiment.py b/tests/core/test_experiment.py index 9f139a7..be7dda6 100644 --- a/tests/core/test_experiment.py +++ b/tests/core/test_experiment.py @@ -10,11 +10,20 @@ 5. 实验结果的收集 """ +import os +from contextlib import contextmanager +from copy import deepcopy +from pathlib import Path + import pytest +from hydra import compose, initialize +from hydra.core.global_hydra import GlobalHydra +from hydra.core.hydra_config import HydraConfig +from omegaconf import OmegaConf from abses import MainModel -from abses.core.experiment import Experiment -from tests.helper import RandomAddingMod +from abses.core.experiment import Experiment, launcher_is_parallel +from tests.helper import PidReportingMod, RandomAddingMod class TestExperimentBasic: @@ -68,18 +77,7 @@ def test_parameter_override(self, test_config): class TestExperimentRandom: """测试实验的随机性控制""" - @pytest.fixture(autouse=True) - def setup_class(self): - """在每个测试类运行前重置实验管理器""" - from abses.core.job_manager import ExperimentManager - - # 保存当前的实例 - self._original_instance = getattr(ExperimentManager, "_instance") - # 清空实例 - setattr(ExperimentManager, "_instance", None) - yield - # 测试结束后恢复原来的实例 - setattr(ExperimentManager, "_instance", self._original_instance) + pytestmark = pytest.mark.usefixtures("reset_experiment_manager") def test_seed_control(self, test_config): """测试随机种子控制""" @@ -100,3 +98,149 @@ def test_seed_control(self, test_config): assert results1.equals(results2) # 验证不同种子产生不同结果 assert not results1.equals(results3) + + +class TestLauncherIsParallel: + """`launcher_is_parallel` is True only when the launcher really is concurrent. + + Regression for #169: the check used to be `launcher is not None`, but Hydra + configures a serial BasicLauncher for plain runs and multirun alike, so it + was always true and `Experiment.batch_run(parallels=...)` never reached its + parallel branch. + """ + + def test_basic_launcher_is_not_parallel(self): + """BasicLauncher runs jobs in a `for` loop, which is not parallel.""" + launcher = OmegaConf.create( + {"_target_": "hydra._internal.core_plugins.basic_launcher.BasicLauncher"} + ) + assert launcher_is_parallel(launcher) is False + + @pytest.mark.parametrize( + "target", + [ + "hydra_plugins.hydra_joblib_launcher.joblib_launcher.JoblibLauncher", + "hydra_plugins.hydra_submitit_launcher.submitit_launcher.LocalSubmititLauncher", + "hydra_plugins.hydra_submitit_launcher.submitit_launcher.SlurmSubmititLauncher", + "hydra_plugins.hydra_ray_launcher.ray_launcher.RayLauncher", + ], + ) + def test_plugin_launchers_are_parallel(self, target): + """Real launcher plugins do run jobs concurrently, so abses must yield.""" + launcher = OmegaConf.create({"_target_": target}) + assert launcher_is_parallel(launcher) is True + + def test_joblib_with_one_job_is_not_parallel(self): + """`n_jobs: 1` makes joblib run every job in the calling process. + + joblib's `Parallel(n_jobs=1)` uses the sequential backend, so this + launcher is as serial as BasicLauncher despite being a plugin. + """ + launcher = OmegaConf.create( + { + "_target_": "hydra_plugins.hydra_joblib_launcher" + ".joblib_launcher.JoblibLauncher", + "n_jobs": 1, + } + ) + assert launcher_is_parallel(launcher) is False + + def test_unreadable_n_jobs_falls_back_to_parallel(self): + """An `n_jobs` we cannot read must not crash the run. + + This is a predicate on the way to `batch_run`; raising here would abort + the whole experiment. Assuming parallel only costs a layer of nesting. + """ + launcher = OmegaConf.create( + { + "_target_": "hydra_plugins.hydra_joblib_launcher" + ".joblib_launcher.JoblibLauncher", + "n_jobs": "${undefined_key}", + } + ) + assert launcher_is_parallel(launcher) is True + + +@contextmanager +def _inside_hydra_job(launcher_target: str, output_dir: Path): + """Make the process look like a running Hydra job with the given launcher.""" + GlobalHydra.instance().clear() + with initialize(version_base=None, config_path="../config"): + cfg = compose(config_name="test_config.yaml", return_hydra_config=True) + OmegaConf.set_struct(cfg, False) + cfg.hydra.launcher = OmegaConf.create({"_target_": launcher_target}) + cfg.hydra.job.id = 0 + cfg.hydra.runtime.output_dir = str(output_dir) + HydraConfig.instance().set_config(cfg) + try: + yield + finally: + HydraConfig.instance().cfg = None + + +class TestNumProcessInsideHydra: + """`parallels` must still take effect inside a Hydra job (#169). + + Hydra's default BasicLauncher is serial, so abses has to do the + parallelising itself rather than yielding to the launcher. + """ + + pytestmark = pytest.mark.usefixtures("reset_experiment_manager") + + @pytest.fixture(name="pid_config") + def pid_reporting_config(self, test_config, tmp_path): + """A config whose only final report is the process each repeat ran in.""" + cfg = deepcopy(test_config) + cfg.reports.final = {"worker_pid": "worker_pid"} + cfg.outpath = str(tmp_path) + return cfg + + def test_sequential_path_still_records_results(self, pid_config): + """Running the repeats in-process must not lose their results. + + Only the parallel branch used to feed `run_single`'s return value back + to the manager, so any sequential run reported an empty summary. + """ + exp = Experiment.new(PidReportingMod, pid_config) + exp.batch_run(repeats=3, parallels=1, display_progress=False) + + summary = exp.summary() + assert len(summary) == 3, f"expected 3 recorded runs, got {len(summary)}" + assert set(summary["worker_pid"]) == {os.getpid()} + + def test_repeats_runs_in_worker_processes(self, pid_config, tmp_path): + """Repeats run in worker processes rather than in the parent.""" + exp = Experiment.new(PidReportingMod, pid_config) + with _inside_hydra_job( + "hydra._internal.core_plugins.basic_launcher.BasicLauncher", + tmp_path, + ): + exp.batch_run(repeats=4, parallels=4, display_progress=False) + + summary = exp.summary() + assert len(summary) == 4, f"expected 4 recorded runs, got {len(summary)}" + + # Whether the repeats land on one worker or four is joblib's dispatch + # timing, not a promise this code makes: these runs are short enough + # that one worker can take all four before the others have started. + # What distinguishes the two branches is the parent process, and loky + # never executes in it. + pids = set(summary["worker_pid"]) + assert os.getpid() not in pids, ( + f"repeats ran in the parent, so the parallel branch was skipped: {pids}" + ) + + def test_yields_to_a_real_parallel_launcher(self, pid_config, tmp_path): + """A launcher plugin already parallelises, so abses must not nest.""" + exp = Experiment.new(PidReportingMod, pid_config) + with _inside_hydra_job( + "hydra_plugins.hydra_joblib_launcher.joblib_launcher.JoblibLauncher", + tmp_path, + ): + exp.batch_run(repeats=3, parallels=4, display_progress=False) + + summary = exp.summary() + assert len(summary) == 3, f"expected 3 recorded runs, got {len(summary)}" + assert set(summary["worker_pid"]) == {os.getpid()}, ( + "abses nested its own workers inside an already-parallel launcher" + ) diff --git a/tests/helper.py b/tests/helper.py index e4bd251..dd8be7e 100644 --- a/tests/helper.py +++ b/tests/helper.py @@ -5,6 +5,8 @@ # GitHub : https://github.com/SongshGeo # Website: https://cv.songshgeo.com/ +import os + from abses import Actor, MainModel @@ -30,3 +32,17 @@ def create_actors_with_metric(model: MainModel, n: int): for i, actor in enumerate(actors): actor.test = float(i) return actors + + +class PidReportingMod(MainModel): + """Reports the OS process it ran in, so parallelism is observable. + + `run_single` executes in a worker process and only its reported vars travel + back, so a final reporter on this attribute is how a test can see which + process each repeat actually ran in. + """ + + @property + def worker_pid(self) -> int: + """PID of the process running this model.""" + return os.getpid() diff --git a/uv.lock b/uv.lock index fd5c1e0..3197164 100644 --- a/uv.lock +++ b/uv.lock @@ -8,7 +8,7 @@ resolution-markers = [ [[package]] name = "abses" -version = "0.10.0" +version = "0.11.7" source = { editable = "." } dependencies = [ { name = "fiona" },