[autoscaler] Flag flip for resource_demand_scheduler should take into account queue (#11615)

python/ray/autoscaler/_private/constants.py

@@ -32,6 +32,11 @@ AUTOSCALER_UPDATE_INTERVAL_S = env_integer("AUTOSCALER_UPDATE_INTERVAL_S", 5)
 AUTOSCALER_HEARTBEAT_TIMEOUT_S = env_integer("AUTOSCALER_HEARTBEAT_TIMEOUT_S",
                                              30)
 
+# The maximum allowed resource demand vector size to guarantee the resource
+# demand scheduler bin packing algorithm takes a reasonable amount of time
+# to run.
+AUTOSCALER_MAX_RESOURCE_DEMAND_VECTOR_SIZE = 1000
+
 # Max number of retries to AWS (default is 5, time increases exponentially)
 BOTO_MAX_RETRIES = env_integer("BOTO_MAX_RETRIES", 12)
 # Max number of retries to create an EC2 node (retry different subnet)

python/ray/monitor.py

@@ -12,6 +12,8 @@ from ray.autoscaler._private.autoscaler import StandardAutoscaler
 from ray.autoscaler._private.commands import teardown_cluster
 from ray.autoscaler._private.constants import AUTOSCALER_UPDATE_INTERVAL_S
 from ray.autoscaler._private.load_metrics import LoadMetrics
+from ray.autoscaler._private.constants import \
+    AUTOSCALER_MAX_RESOURCE_DEMAND_VECTOR_SIZE
 import ray.gcs_utils
 import ray.utils
 import ray.ray_constants as ray_constants
@@ -56,9 +58,17 @@ def parse_resource_demands(resource_load_by_shape):
                 backlog_queue = waiting_bundles
             for _ in range(resource_demand_pb.backlog_size):
                 backlog_queue.append(request_shape)
+            if len(waiting_bundles+infeasible_bundles) > \
+                    AUTOSCALER_MAX_RESOURCE_DEMAND_VECTOR_SIZE:
+                break
     except Exception:
         logger.exception("Failed to parse resource demands.")
-    return waiting_bundles, infeasible_bundles
+
+    # Bound the total number of bundles to 2xMAX_RESOURCE_DEMAND_VECTOR_SIZE.
+    # This guarantees the resource demand scheduler bin packing algorithm takes
+    # a reasonable amount of time to run.
+    return waiting_bundles[:AUTOSCALER_MAX_RESOURCE_DEMAND_VECTOR_SIZE], \
+        infeasible_bundles[:AUTOSCALER_MAX_RESOURCE_DEMAND_VECTOR_SIZE]
 
 
 class Monitor:

python/ray/tests/test_resource_demand_scheduler.py

@@ -23,6 +23,8 @@ from ray.autoscaler.tags import TAG_RAY_USER_NODE_TYPE, TAG_RAY_NODE_KIND, \
                                 NODE_KIND_WORKER, TAG_RAY_NODE_STATUS, \
                                 STATUS_UP_TO_DATE, STATUS_UNINITIALIZED
 from ray.test_utils import same_elements
+from ray.autoscaler._private.constants import \
+    AUTOSCALER_MAX_RESOURCE_DEMAND_VECTOR_SIZE
 
 from time import sleep
 
@@ -365,6 +367,111 @@ def test_calculate_node_resources():
     assert to_launch == {"p2.8xlarge": 1}
 
 
+def test_backlog_queue_impact_on_binpacking_time():
+    new_types = copy.deepcopy(TYPES_A)
+    new_types["p2.8xlarge"]["max_workers"] = 1000
+    new_types["m4.16xlarge"]["max_workers"] = 1000
+
+    def test_backlog_queue_impact_on_binpacking_time_aux(
+            num_available_nodes, time_to_assert, demand_request_shape):
+        provider = MockProvider()
+        scheduler = ResourceDemandScheduler(
+            provider, new_types, max_workers=10000)
+
+        provider.create_node({}, {
+            TAG_RAY_USER_NODE_TYPE: "m4.16xlarge",
+            TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE
+        }, num_available_nodes)
+        # <num_available_nodes> m4.16xlarge instances.
+        cpu_ips = provider.non_terminated_node_ips({})
+        provider.create_node({}, {
+            TAG_RAY_USER_NODE_TYPE: "p2.8xlarge",
+            TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE
+        }, num_available_nodes)
+        # <num_available_nodes>  m4.16xlarge and <num_available_nodes>
+        # p2.8xlarge instances.
+        all_nodes = provider.non_terminated_nodes({})
+        all_ips = provider.non_terminated_node_ips({})
+        gpu_ips = [ip for ip in all_ips if ip not in cpu_ips]
+        usage_by_ip = {}
+        # 2x<num_available_nodes> free nodes (<num_available_nodes> m4.16xlarge
+        # and <num_available_nodes> p2.8xlarge instances).
+        for i in range(num_available_nodes):
+            usage_by_ip[cpu_ips[i]] = {"CPU": 64}
+            usage_by_ip[gpu_ips[i]] = {"GPU": 8, "CPU": 32}
+        demands = demand_request_shape * \
+            AUTOSCALER_MAX_RESOURCE_DEMAND_VECTOR_SIZE
+        t1 = time.time()
+        to_launch = scheduler.get_nodes_to_launch(all_nodes, {}, demands,
+                                                  usage_by_ip, [])
+        t2 = time.time()
+        assert t2 - t1 < time_to_assert
+        print("The time took to launch", to_launch,
+              "with number of available nodes set to", num_available_nodes,
+              "is:", t2 - t1)
+        return to_launch
+
+    # The assertions below use 10s but the actual time took when this test was
+    # measured on 2.3 GHz 8-Core Intel (I9-9880H) Core i9 is commented inline.
+
+    # Check the time it takes when there are 0 nodes available and the demand
+    # is requires adding another ~100 nodes.
+    to_launch = test_backlog_queue_impact_on_binpacking_time_aux(
+        num_available_nodes=0,
+        time_to_assert=10,  # real time 0.2s.
+        demand_request_shape=[{
+            "GPU": 1
+        }, {
+            "CPU": 1
+        }])
+    # If not for the max launch concurrency the next assert should be:
+    # {'m4.large': 4, 'm4.4xlarge': 2, 'm4.16xlarge': 15, 'p2.8xlarge': 125}.
+    assert to_launch == {
+        "m4.large": 4,
+        "m4.4xlarge": 2,
+        "m4.16xlarge": 5,
+        "p2.8xlarge": 5
+    }
+
+    # Check the time it takes when there are 100 nodes available and the demand
+    # requires another 75 nodes.
+    to_launch = test_backlog_queue_impact_on_binpacking_time_aux(
+        num_available_nodes=50,
+        time_to_assert=10,  # real time 0.075s.
+        demand_request_shape=[{
+            "GPU": 1
+        }, {
+            "CPU": 2
+        }])
+    # If not for the max launch concurrency the next assert should be:
+    # {'p2.8xlarge': 75}.
+    assert to_launch == {"p2.8xlarge": 50}
+
+    # Check the time it takes when there are 250 nodes available and can
+    # cover the demand.
+    to_launch = test_backlog_queue_impact_on_binpacking_time_aux(
+        num_available_nodes=125,
+        time_to_assert=10,  # real time 0.06s.
+        demand_request_shape=[{
+            "GPU": 1
+        }, {
+            "CPU": 1
+        }])
+    assert to_launch == {}
+
+    # Check the time it takes when there are 1000 nodes available and the
+    # demand requires another 1000 nodes.
+    to_launch = test_backlog_queue_impact_on_binpacking_time_aux(
+        num_available_nodes=500,
+        time_to_assert=10,  # real time 1.32s.
+        demand_request_shape=[{
+            "GPU": 8
+        }, {
+            "CPU": 64
+        }])
+    assert to_launch == {"m4.16xlarge": 500, "p2.8xlarge": 500}
+
+
 class TestPlacementGroupScaling:
     def test_strategies(self):
         provider = MockProvider()