From 8d74a04a42921b2f94c8751d36b544e91a3a5803 Mon Sep 17 00:00:00 2001 From: Ameer Haj Ali Date: Mon, 2 Nov 2020 22:41:22 +0200 Subject: [PATCH] [autoscaler] Flag flip for resource_demand_scheduler should take into account queue (#11615) --- python/ray/autoscaler/_private/constants.py | 5 + python/ray/monitor.py | 12 +- .../tests/test_resource_demand_scheduler.py | 107 ++++++++++++++++++ src/ray/common/ray_config_def.h | 2 +- 4 files changed, 124 insertions(+), 2 deletions(-) diff --git a/python/ray/autoscaler/_private/constants.py b/python/ray/autoscaler/_private/constants.py index 4aca9eb49..de438197e 100644 --- a/python/ray/autoscaler/_private/constants.py +++ b/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) diff --git a/python/ray/monitor.py b/python/ray/monitor.py index f65b509cc..984caadfe 100644 --- a/python/ray/monitor.py +++ b/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: diff --git a/python/ray/tests/test_resource_demand_scheduler.py b/python/ray/tests/test_resource_demand_scheduler.py index 8239e1280..7b0748f55 100644 --- a/python/ray/tests/test_resource_demand_scheduler.py +++ b/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) + # 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) + # m4.16xlarge and + # 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 free nodes ( m4.16xlarge + # and 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() diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h index bc1b9418b..a6ce69808 100644 --- a/src/ray/common/ray_config_def.h +++ b/src/ray/common/ray_config_def.h @@ -291,7 +291,7 @@ RAY_CONFIG(int64_t, max_resource_shapes_per_load_report, 100) /// If true, the worker's queue backlog size will be propagated to the heartbeat batch /// data. -RAY_CONFIG(bool, report_worker_backlog, false) +RAY_CONFIG(bool, report_worker_backlog, true) /// The timeout for synchronous GCS requests in seconds. RAY_CONFIG(int64_t, gcs_server_request_timeout_seconds, 5)