Repository: modin-project/modin
Pull request: FIX-#3203: Fix hanging of Modin XGBoost
Signed-off-by: Alexey Prutskov <alexey.prutskov@intel.com>

<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/CONTRIBUTING.html
if you have questions about contributing.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

- [x] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/contributing.html)
- [x] passes `flake8 modin`
- [x] passes `black --check modin`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #3203  <!-- issue must be created for each patch -->
- [ ] tests added and passing
- [x] module layout described at `docs/developer/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->


Problem statement:
Modin XGBoost hangs
### Describe the problem
<!-- Describe the problem clearly here. -->
ModinXGBoostActors claims all cluster resources. Pending internal tasks can't start and warning message is raised:

```bash
2021-06-25 08:37:46,124 WARNING worker.py:1114 -- The actor or task with ID 46766c5b0167bafaffffffffffffffffffffffff01000000 cannot be scheduled right now. It requires {CPU: 1.000000} for placement, but this node only has remaining {0.000000/112.000000 CPU, 227.353494 GiB/227.353494 GiB memory, 101.428594 GiB/101.428594 GiB object_store_memory, 0.000000/112.000000 CPU_group_eb4a2862d9a2cf9c4c2617c2763cd46f, 28.000000/28.000000 CPU_group_2_eb4a2862d9a2cf9c4c2617c2763cd46f, 28.000000/28.000000 CPU_group_3_eb4a2862d9a2cf9c4c2617c2763cd46f, 0.980000/1.000000 node:10.125.130.30, 28.000000/28.000000 CPU_group_0_eb4a2862d9a2cf9c4c2617c2763cd46f, 28.000000/28.000000 CPU_group_1_eb4a2862d9a2cf9c4c2617c2763cd46f}
. In total there are 112 pending tasks and 0 pending actors on this node. This is likely due to all cluster resources being claimed by actors. To resolve the issue, consider creating fewer actors or increase the resources available to this Ray cluster. You can ignore this message if this Ray cluster is expected to auto-scale or if you specified a runtime_env for this task or actor because it takes time to install.
```

After that, execution hangs.

Hints from issue discussion:

Commit message: FIX-#3203: Remove superfluous changes

Signed-off-by: Alexey Prutskov <alexey.prutskov@intel.com>

Patch to review:
diff --git a/modin/experimental/xgboost/xgboost_ray.py b/modin/experimental/xgboost/xgboost_ray.py
index 77b7276e9fc..e7537b0b525 100644
--- a/modin/experimental/xgboost/xgboost_ray.py
+++ b/modin/experimental/xgboost/xgboost_ray.py
@@ -27,7 +27,6 @@
 import xgboost as xgb
 import ray
 from ray.services import get_node_ip_address
-from ray.util.placement_group import placement_group, remove_placement_group
 import pandas
 
 from modin.distributed.dataframe.pandas import from_partitions
@@ -249,11 +248,10 @@ def _get_min_cpus_per_node():
     int
         Min number of CPUs per node.
     """
-    # TODO: max_node_cpus -> min_node_cpus
-    max_node_cpus = min(
+    min_node_cpus = min(
         node.get("Resources", {}).get("CPU", 0.0) for node in ray.nodes()
     )
-    return max_node_cpus if max_node_cpus > 0.0 else _get_cluster_cpus()
+    return min_node_cpus if min_node_cpus > 0.0 else _get_cluster_cpus()
 
 
 def _get_cpus_per_actor(num_actors):
@@ -303,38 +301,14 @@ def _get_num_actors(num_actors):
         num_actors_per_node = max(1, int(min_cpus_per_node // 8))
         return num_actors_per_node * len(ray.nodes())
     elif isinstance(num_actors, int):
+        assert (
+            num_actors % len(ray.nodes()) == 0
+        ), "`num_actors` must be a multiple to number of nodes in Ray cluster."
         return num_actors
     else:
-        RuntimeError("`num_actors` must be int or None")
-
-
-def _create_placement_group(num_cpus_per_actor, num_actors):
-    """
-    Create Ray placement group to grab resources.
-
-    Parameters
-    ----------
-    num_cpus_per_actor : int
-        Number of CPUs per actor.
-    num_actors : int
-        Number of actors.
-
-    Returns
-    -------
-    ray.util.PlacementGroup
-        Placement group with grabbed resources.
-    """
-    cpu_bundle = {"CPU": num_cpus_per_actor}
-    bundles = [cpu_bundle for _ in range(num_actors)]
-
-    pg = placement_group(bundles, strategy="SPREAD")
-
-    ready, _ = ray.wait([pg.ready()], timeout=100)
-
-    if ready is None:
-        raise TimeoutError("Placement group creation timeout.")
-
-    return pg
+        raise RuntimeError(
+            "`num_actors` must be int, 'default_train', 'default_predict' or 'None'"
+        )
 
 
 def create_actors(num_actors):
@@ -352,14 +326,22 @@ def create_actors(num_actors):
         Pair of actors list and placement group of actors.
     """
     num_cpus_per_actor = _get_cpus_per_actor(num_actors)
-    pg = _create_placement_group(num_cpus_per_actor, num_actors)
+    nodes_ips = [
+        key for key, _ in ray.cluster_resources().items() if key.startswith("node:")
+    ]
+    num_actors_per_node = num_actors // len(nodes_ips)
+    actors_ips = [ip for ip in nodes_ips for _ in range(num_actors_per_node)]
+
     actors = [
-        ModinXGBoostActor.options(
-            num_cpus=num_cpus_per_actor, placement_group=pg
-        ).remote(i, nthread=num_cpus_per_actor)
-        for i in range(num_actors)
+        (
+            node_ip.split("node:")[-1],
+            ModinXGBoostActor.options(resources={node_ip: 0.01}).remote(
+                i, nthread=num_cpus_per_actor
+            ),
+        )
+        for i, node_ip in enumerate(actors_ips)
     ]
-    return actors, pg
+    return actors
 
 
 def _split_data_across_actors(
@@ -394,7 +376,7 @@ def _split_data_across_actors(
             data_for_aligning=X_parts_by_actors,
         )
 
-    for rank, actor in enumerate(actors):
+    for rank, (_, actor) in enumerate(actors):
         X_parts = X_parts_by_actors[rank] if is_predict else X_parts_by_actors[rank][0]
         if y_parts is None:
             set_func(actor, *X_parts)
@@ -445,10 +427,7 @@ def _assign_row_partitions_to_actors(
 
             # Group actors which are one the same ip
             actor_ips = defaultdict(list)
-
-            for rank, ip in enumerate(
-                ray.get([actor.get_ip.remote() for actor in actors])
-            ):
+            for rank, (ip, _) in enumerate(actors):
                 actor_ips[ip].append(rank)
 
             # Get distribution of parts between nodes ({ip:[(part, position),..],..})
@@ -623,7 +602,7 @@ def _train(
     if num_actors > len(X_row_parts):
         num_actors = len(X_row_parts)
 
-    actors, pg = create_actors(num_actors)
+    actors = create_actors(num_actors)
 
     add_as_eval_method = None
     if evals:
@@ -660,12 +639,12 @@ def _train(
 
         # Train
         fut = [
-            actor.train.remote(rabit_args, params, *args, **kwargs) for actor in actors
+            actor.train.remote(rabit_args, params, *args, **kwargs)
+            for _, actor in actors
         ]
         # All results should be the same because of Rabit tracking. So we just
         # return the first one.
         result = ray.get(fut[0])
-        remove_placement_group(pg)
         LOGGER.info(f"Training time: {time.time() - s} s")
         return result
 
@@ -712,7 +691,7 @@ def _predict(
         num_actors = len(X_row_parts)
 
     # Create remote actors
-    actors, pg = create_actors(num_actors)
+    actors = create_actors(num_actors)
 
     # Split data across workers
     _split_data_across_actors(
@@ -729,12 +708,9 @@ def _predict(
 
     predictions = [
         tuple(actor.predict.options(num_returns=2).remote(booster, **kwargs))
-        for actor in actors
+        for _, actor in actors
     ]
 
-    ray.wait([part for _, part in predictions], num_returns=len(predictions))
-    remove_placement_group(pg)
-
     result = from_partitions(predictions, 0)
     LOGGER.info(f"Prediction time: {time.time() - s} s")
 


Code review comments:
diff_hunk: @@ -352,14 +326,22 @@ def create_actors(num_actors):
         Pair of actors list and placement group of actors.
     """
     num_cpus_per_actor = _get_cpus_per_actor(num_actors)
-    pg = _create_placement_group(num_cpus_per_actor, num_actors)
+    nodes_ips = [
+        key for key, _ in ray.cluster_resources().items() if key.startswith("node:")
path: modin/experimental/xgboost/xgboost_ray.py
text: @user1:
```suggestion
        key for key in ray.cluster_resources().keys() if key.startswith("node:")
```

@author:
fixed

diff_hunk: @@ -352,14 +326,22 @@ def create_actors(num_actors):
         Pair of actors list and placement group of actors.
     """
     num_cpus_per_actor = _get_cpus_per_actor(num_actors)
-    pg = _create_placement_group(num_cpus_per_actor, num_actors)
+    nodes_ips = [
path: modin/experimental/xgboost/xgboost_ray.py
text: @user1:
```suggestion
    node_ips = [
```

@author:
fixed