Merge pull request #270 from appliedAI-Initiative/fix/249-negative-njobs

Fix #249: allow negative n_jobs

Author: mdbenito

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## Unreleased
 
+- Fixed a bug whereby `compute_shapley_values` would only spawn one process when
+  using `n_jobs=-1` and Monte Carlo methods.
+  [PR #270](https://github.com/appliedAI-Initiative/pyDVL/pull/270)
 - Bugfix in `RayParallelBackend`: wrong semantics for `kwargs`.
   [PR #268](https://github.com/appliedAI-Initiative/pyDVL/pull/268)
 - Splitting of problem preparation and solution in Least-Core computation.

src/pydvl/utils/parallel/backend.py

@@ -20,9 +20,7 @@
 
 from ..config import ParallelConfig
 
-__all__ = [
-    "init_parallel_backend",
-]
+__all__ = ["init_parallel_backend", "effective_n_jobs", "available_cpus"]
 
 T = TypeVar("T")
 
@@ -234,3 +232,25 @@ def available_cpus() -> int:
     if system() != "Linux":
         return os.cpu_count() or 1
     return len(os.sched_getaffinity(0))
+
+
+def effective_n_jobs(n_jobs: int, config: ParallelConfig = ParallelConfig()) -> int:
+    """Returns the effective number of jobs.
+
+    This number may vary depending on the parallel backend and the resources
+    available.
+
+    :param n_jobs: the number of jobs requested. If -1, the number of available
+        CPUs is returned.
+    :param config: instance of :class:`~pydvl.utils.config.ParallelConfig` with
+        cluster address, number of cpus, etc.
+    :return: the effective number of jobs, guaranteed to be >= 1.
+    :raises RuntimeError: if the effective number of jobs returned by the backend
+        is < 1.
+    """
+    parallel_backend = init_parallel_backend(config)
+    if (eff_n_jobs := parallel_backend.effective_n_jobs(n_jobs)) < 1:
+        raise RuntimeError(
+            f"Invalid number of jobs {eff_n_jobs} obtained from parallel backend {config.backend}"
+        )
+    return eff_n_jobs

src/pydvl/value/least_core/montecarlo.py

@@ -1,13 +1,13 @@
 import logging
 import warnings
-from typing import Iterable, Optional, Tuple
+from typing import Iterable, Optional
 
 import numpy as np
-from numpy.typing import NDArray
 
 from pydvl.utils.config import ParallelConfig
 from pydvl.utils.numeric import random_powerset
 from pydvl.utils.parallel import MapReduceJob
+from pydvl.utils.parallel.backend import effective_n_jobs
 from pydvl.utils.progress import maybe_progress
 from pydvl.utils.utility import Utility
 from pydvl.value.least_core.common import LeastCoreProblem, lc_solve_problem
@@ -145,16 +145,13 @@ def mclc_prepare_problem(
         )
         n_iterations = 2**n
 
-    iterations_per_job = max(1, n_iterations // n_jobs)
+    iterations_per_job = max(1, n_iterations // effective_n_jobs(n_jobs, config))
 
     map_reduce_job: MapReduceJob["Utility", "LeastCoreProblem"] = MapReduceJob(
         inputs=u,
         map_func=_montecarlo_least_core,
         reduce_func=_reduce_func,
-        map_kwargs=dict(
-            n_iterations=iterations_per_job,
-            progress=progress,
-        ),
+        map_kwargs=dict(n_iterations=iterations_per_job, progress=progress),
         n_jobs=n_jobs,
         config=config,
     )

src/pydvl/value/shapley/gt.py

@@ -25,6 +25,7 @@
 
 from pydvl.utils import MapReduceJob, ParallelConfig, Utility, maybe_progress
 from pydvl.utils.numeric import random_subset_of_size
+from pydvl.utils.parallel.backend import effective_n_jobs
 from pydvl.utils.status import Status
 from pydvl.value import ValuationResult
 
@@ -237,7 +238,7 @@ def group_testing_shapley(
             f"ε={eps:.02f} guarantee at .95 probability"
         )
 
-    iterations_per_job = max(1, n_iterations // n_jobs)
+    iterations_per_job = max(1, n_iterations // effective_n_jobs(n_jobs, config))
 
     def reducer(
         results_it: Iterable[Tuple[NDArray, NDArray]]

src/pydvl/value/shapley/montecarlo.py

@@ -44,6 +44,7 @@
 from ...utils.config import ParallelConfig
 from ...utils.numeric import random_powerset, running_moments
 from ...utils.parallel import MapReduceJob, init_parallel_backend
+from ...utils.parallel.backend import effective_n_jobs
 from ...utils.progress import maybe_progress
 from ...utils.status import Status
 from ...utils.utility import Utility
@@ -224,7 +225,7 @@ def permutation_montecarlo_shapley(
     :param progress: Whether to display progress bars for each job.
     :return: Object with the data values.
     """
-    iterations_per_job = max(1, n_iterations // n_jobs)
+    iterations_per_job = max(1, n_iterations // effective_n_jobs(n_jobs, config))
 
     map_reduce_job: MapReduceJob[Utility, "NDArray"] = MapReduceJob(
         u,

tests/utils/test_parallel.py

@@ -7,7 +7,7 @@
 import pytest
 
 from pydvl.utils.parallel import MapReduceJob, init_parallel_backend
-from pydvl.utils.parallel.backend import available_cpus
+from pydvl.utils.parallel.backend import available_cpus, effective_n_jobs
 from pydvl.utils.parallel.map_reduce import _get_value
 
 
@@ -25,6 +25,15 @@ def test_effective_n_jobs(parallel_config, num_workers):
         else:
             assert parallel_backend.effective_n_jobs(-1) == num_workers
 
+    for n_jobs in [-1, 1, 2]:
+        assert parallel_backend.effective_n_jobs(n_jobs) == effective_n_jobs(
+            n_jobs, parallel_config
+        )
+        assert effective_n_jobs(n_jobs, parallel_config) > 0
+
+    with pytest.raises(ValueError):
+        parallel_backend.effective_n_jobs(0)
+
 
 @pytest.fixture()
 def map_reduce_job_and_parameters(parallel_config, n_jobs, request):