[core] Add deprecation warning and remove tests for _max_cpu_fraction_per_node (ray-project#54380)

Per discussion in: ray-project#53610, this
argument was introduced for Ray AIR but is no longer used by any
libraries.

Out of an abundance of caution, adding a warning for a release before
removing it.

Deleting the integration tests eagerly since they burn CPU time in CI.

---------

Signed-off-by: Edward Oakes <ed.nmi.oakes@gmail.com>

Author: edoakes

python/ray/tests/test_placement_group_5.py

@@ -261,155 +261,6 @@ def add_nodes_to_cluster(cluster):
     )
 
 
-@pytest.mark.parametrize("multi_bundle", [True, False])
-@pytest.mark.parametrize("even_pack", [True, False])
-@pytest.mark.parametrize("scheduling_strategy", ["SPREAD", "STRICT_PACK", "PACK"])
-def test_placement_group_max_cpu_frac(
-    ray_start_cluster, multi_bundle, even_pack, scheduling_strategy
-):
-    cluster = ray_start_cluster
-    cluster.add_node(num_cpus=4)
-    cluster.wait_for_nodes()
-    ray.init(address=cluster.address)
-
-    if multi_bundle:
-        bundles = [{"CPU": 1}] * 3
-    else:
-        bundles = [{"CPU": 3}]
-
-    # Input validation - max_cpu_fraction_per_node must be between 0 and 1.
-    with pytest.raises(ValueError):
-        ray.util.placement_group(bundles, _max_cpu_fraction_per_node=-1)
-    with pytest.raises(ValueError):
-        ray.util.placement_group(bundles, _max_cpu_fraction_per_node=2)
-
-    pg = ray.util.placement_group(
-        bundles, strategy=scheduling_strategy, _max_cpu_fraction_per_node=0.5
-    )
-
-    # Placement group will never be scheduled since it would violate the max CPU
-    # fraction reservation.
-    with pytest.raises(ray.exceptions.GetTimeoutError):
-        ray.get(pg.ready(), timeout=5)
-
-    # Add new node with enough CPU cores to scheduled placement group bundle while
-    # adhering to the max CPU fraction constraint.
-    if even_pack:
-        num_cpus = 6
-    else:
-        num_cpus = 8
-    cluster.add_node(num_cpus=num_cpus)
-    cluster.wait_for_nodes()
-    # The placement group should be schedulable so this shouldn't raise.
-    ray.get(pg.ready(), timeout=5)
-
-
-def test_placement_group_max_cpu_frac_multiple_pgs(ray_start_cluster):
-    """
-    Make sure when there's more than 1 pg, they respect the fraction.
-    """
-    cluster = ray_start_cluster
-    cluster.add_node(num_cpus=8)
-    cluster.wait_for_nodes()
-    ray.init(address=cluster.address)
-
-    # This pg should be scheduable.
-    pg = ray.util.placement_group([{"CPU": 4}], _max_cpu_fraction_per_node=0.5)
-    ray.get(pg.ready())
-
-    # When we schedule another placement group, it shouldn't be scheduled.
-    pg2 = ray.util.placement_group([{"CPU": 4}], _max_cpu_fraction_per_node=0.5)
-    with pytest.raises(ray.exceptions.GetTimeoutError):
-        ray.get(pg2.ready(), timeout=5)
-
-    # When you add a new node, it is finally schedulable.
-    cluster.add_node(num_cpus=8)
-    ray.get(pg2.ready())
-
-
-def test_placement_group_max_cpu_frac_edge_cases(ray_start_cluster):
-    """
-    _max_cpu_fraction_per_node <= 0  ---> should raise error (always)
-    _max_cpu_fraction_per_node = 0.999 --->
-        should exclude 1 CPU (this is already the case)
-    _max_cpu_fraction_per_node = 0.001 --->
-        should exclude 3 CPUs (not currently the case, we'll exclude all 4 CPUs).
-
-    Related: https://github.com/ray-project/ray/issues/26635
-    """
-    cluster = ray_start_cluster
-    cluster.add_node(num_cpus=4)
-    cluster.wait_for_nodes()
-    ray.init(address=cluster.address)
-
-    """
-    0 or 1 is not allowed.
-    """
-    with pytest.raises(ValueError):
-        ray.util.placement_group([{"CPU": 1}], _max_cpu_fraction_per_node=0)
-
-    """
-    Make sure when _max_cpu_fraction_per_node = 0.999, 1 CPU is always excluded.
-    """
-    pg = ray.util.placement_group(
-        [{"CPU": 1} for _ in range(4)], _max_cpu_fraction_per_node=0.999
-    )
-    # Since 1 CPU is excluded, we cannot schedule this pg.
-    with pytest.raises(ray.exceptions.GetTimeoutError):
-        ray.get(pg.ready(), timeout=5)
-    ray.util.remove_placement_group(pg)
-
-    # Since 1 CPU is excluded, we can schedule 1 num_cpus actor after creating
-    # CPU: 1 * 3 bundle placement groups.
-    @ray.remote(num_cpus=1)
-    class A:
-        def ready(self):
-            pass
-
-    # Try actor creation -> pg creation.
-    a = A.remote()
-    ray.get(a.ready.remote())
-    pg = ray.util.placement_group(
-        [{"CPU": 1} for _ in range(3)], _max_cpu_fraction_per_node=0.999
-    )
-    ray.get(pg.ready())
-
-    ray.kill(a)
-    ray.util.remove_placement_group(pg)
-
-    # Make sure the opposite order also works. pg creation -> actor creation.
-    pg = ray.util.placement_group(
-        [{"CPU": 1} for _ in range(3)], _max_cpu_fraction_per_node=0.999
-    )
-    a = A.remote()
-    ray.get(a.ready.remote())
-    ray.get(pg.ready())
-
-    ray.kill(a)
-    ray.util.remove_placement_group(pg)
-
-    """
-    _max_cpu_fraction_per_node = 0.001 --->
-        should exclude 3 CPUs (not currently the case, we'll exclude all 4 CPUs).
-    """
-    # We can schedule up to 1 pg.
-    pg = ray.util.placement_group([{"CPU": 1}], _max_cpu_fraction_per_node=0.001)
-    ray.get(pg.ready())
-    # Cannot schedule any more PG.
-    pg2 = ray.util.placement_group([{"CPU": 1}], _max_cpu_fraction_per_node=0.001)
-    with pytest.raises(ray.exceptions.GetTimeoutError):
-        ray.get(pg2.ready(), timeout=5)
-
-    # Since 3 CPUs are excluded, we can schedule actors.
-    actors = [A.remote() for _ in range(3)]
-    ray.get([a.ready.remote() for a in actors])
-
-    # Once pg 1 is removed, pg 2 can be created since there's 1 CPU that can be
-    # used for this pg.
-    ray.util.remove_placement_group(pg)
-    ray.get(pg2.ready())
-
-
 def test_placement_group_parallel_submission(ray_start_cluster):
     NUM_PARALLEL_PGS = 5
     cluster = ray_start_cluster

python/ray/util/placement_group.py

@@ -195,7 +195,6 @@ def placement_group(
     Return:
         PlacementGroup: Placement group object.
     """
-
     worker = ray._private.worker.global_worker
     worker.check_connected()
 
@@ -365,6 +364,12 @@ def validate_placement_group(
 
     assert _max_cpu_fraction_per_node is not None
 
+    if _max_cpu_fraction_per_node != 1.0:
+        warnings.warn(
+            "The experimental '_max_cpu_fraction_per_node' option for placement groups "
+            "is deprecated and will be removed in a future version of Ray."
+        )
+
     if _max_cpu_fraction_per_node <= 0 or _max_cpu_fraction_per_node > 1:
         raise ValueError(
             "Invalid argument `_max_cpu_fraction_per_node`: "