Merge pull request #239 from appliedAI-Initiative/fix-ray-initialization

Fix ray initialization

Author: mdbenito

CHANGELOG.md

@@ -27,6 +27,8 @@
   adds `extra_values` argument to `ValuationResult`,
   adds `compute_removal_score()` and `compute_random_removal_score()` helper functions
   [PR #237](https://github.com/appliedAI-Initiative/pyDVL/pull/237)
+- Fixes bug in ray initialization in `RayParallelBackend` class
+  [PR #239](https://github.com/appliedAI-Initiative/pyDVL/pull/239)
 
 ## 0.3.0 - 💥 Breaking changes
 

src/pydvl/utils/parallel/backend.py

@@ -136,9 +136,8 @@ def __init__(self, config: ParallelConfig):
         config_dict.pop("backend")
         config_dict["num_cpus"] = config_dict.pop("n_local_workers")
         self.config = config_dict
-        if self.config["address"] is None:
-            self.config["ignore_reinit_error"] = True
-        ray.init(**self.config)
+        if not ray.is_initialized():
+            ray.init(**self.config)
 
     def get(
         self,
@@ -200,7 +199,7 @@ def init_parallel_backend(
     >>> config = ParallelConfig(backend="ray")
     >>> parallel_backend = init_parallel_backend(config)
     >>> parallel_backend
-    <RayParallelBackend: {'address': None, 'logging_level': 30, 'num_cpus': None, 'ignore_reinit_error': True}>
+    <RayParallelBackend: {'address': None, 'logging_level': 30, 'num_cpus': None}>
 
     """
     try:

tests/utils/conftest.py

@@ -1,8 +1,25 @@
 import pytest
+import ray
+from ray.cluster_utils import Cluster
 
 from pydvl.utils.config import ParallelConfig
 
 
-@pytest.fixture(scope="session", params=["sequential", "ray"])
+@pytest.fixture(scope="session", params=["sequential", "ray-local", "ray-external"])
 def parallel_config(request):
-    return ParallelConfig(backend=request.param)
+    if request.param == "sequential":
+        yield ParallelConfig(backend=request.param)
+    elif request.param == "ray-local":
+        yield ParallelConfig(backend="ray")
+        ray.shutdown()
+    elif request.param == "ray-external":
+        # Starts a head-node for the cluster.
+        cluster = Cluster(
+            initialize_head=True,
+            head_node_args={
+                "num_cpus": 4,
+            },
+        )
+        yield ParallelConfig(backend="ray", address=cluster.address)
+        ray.shutdown()
+        cluster.shutdown()