fix(experiment): 🐛 Make num_process work under Hydra

.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."

.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

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,

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