From 3fbd8be851fa1fb22714f3eb5af8f50f958339fb Mon Sep 17 00:00:00 2001 From: Tao Wang Date: Thu, 12 Nov 2020 16:01:19 +0800 Subject: [PATCH] [Placement Group]Do not really subtract resources, just count (#11894) * [Placement Group]Do not really subtract resources, just count * add todo --- src/ray/gcs/gcs_server/gcs_node_manager.cc | 7 +++-- src/ray/gcs/gcs_server/gcs_node_manager.h | 5 ++-- .../gcs_placement_group_scheduler.cc | 26 ++++++++++++------- .../gcs_server/test/gcs_node_manager_test.cc | 8 +++--- 4 files changed, 25 insertions(+), 21 deletions(-) diff --git a/src/ray/gcs/gcs_server/gcs_node_manager.cc b/src/ray/gcs/gcs_server/gcs_node_manager.cc index 0d937e074..954f16bd6 100644 --- a/src/ray/gcs/gcs_server/gcs_node_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_node_manager.cc @@ -337,7 +337,7 @@ void GcsNodeManager::HandleGetAllAvailableResources( for (const auto &iter : GetClusterRealtimeResources()) { rpc::AvailableResources resource; resource.set_node_id(iter.first.Binary()); - for (const auto &res : iter.second->GetResourceAmountMap()) { + for (const auto &res : iter.second.GetResourceAmountMap()) { (*resource.mutable_resources_available())[res.first] = res.second.ToDouble(); } reply->add_resources_list()->CopyFrom(resource); @@ -534,13 +534,12 @@ void GcsNodeManager::UpdateNodeRealtimeResources( const NodeID &node_id, const rpc::HeartbeatTableData &heartbeat) { if (!RayConfig::instance().light_heartbeat_enabled() || heartbeat.resources_available_changed()) { - auto resources_available = MapFromProtobuf(heartbeat.resources_available()); cluster_realtime_resources_[node_id] = - std::make_shared(resources_available); + ResourceSet(MapFromProtobuf(heartbeat.resources_available())); } } -const absl::flat_hash_map> +const absl::flat_hash_map &GcsNodeManager::GetClusterRealtimeResources() const { return cluster_realtime_resources_; } diff --git a/src/ray/gcs/gcs_server/gcs_node_manager.h b/src/ray/gcs/gcs_server/gcs_node_manager.h index 00f0d11bc..f3887ead7 100644 --- a/src/ray/gcs/gcs_server/gcs_node_manager.h +++ b/src/ray/gcs/gcs_server/gcs_node_manager.h @@ -166,8 +166,7 @@ class GcsNodeManager : public rpc::NodeInfoHandler { const rpc::HeartbeatTableData &heartbeat); /// Get cluster realtime resources. - const absl::flat_hash_map> - &GetClusterRealtimeResources() const; + const absl::flat_hash_map &GetClusterRealtimeResources() const; /// Update the placement group load information so that it will be reported through /// heartbeat. @@ -286,7 +285,7 @@ class GcsNodeManager : public rpc::NodeInfoHandler { /// Storage for GCS tables. std::shared_ptr gcs_table_storage_; /// Cluster realtime resources. - absl::flat_hash_map> cluster_realtime_resources_; + absl::flat_hash_map cluster_realtime_resources_; /// Placement group load information that is used for autoscaler. absl::optional> placement_group_load_; diff --git a/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc b/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc index f162a8894..58e12d213 100644 --- a/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc +++ b/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc @@ -48,7 +48,7 @@ ScheduleMap GcsStrictPackStrategy::Schedule( const auto &alive_nodes = context->node_manager_.GetClusterRealtimeResources(); std::vector> candidate_nodes; for (auto &node : alive_nodes) { - if (required_resources.IsSubset(*node.second)) { + if (required_resources.IsSubset(node.second)) { candidate_nodes.emplace_back((*context->node_to_bundles_)[node.first], node.first); } } @@ -77,12 +77,15 @@ ScheduleMap GcsPackStrategy::Schedule( // First fill up a node. If the node resource is insufficient, select a new node. // TODO(ffbin): We will speed this up in next PR. Currently it is a double for loop. ScheduleMap schedule_map; - const auto &alive_nodes = context->node_manager_.GetClusterRealtimeResources(); + // TODO(WangTao): This copy might take too much space once cluster grows very large. + // Would find better solution. + absl::flat_hash_map alive_nodes( + context->node_manager_.GetClusterRealtimeResources()); for (const auto &bundle : bundles) { const auto &required_resources = bundle->GetRequiredResources(); for (auto &node : alive_nodes) { - if (required_resources.IsSubset(*node.second)) { - node.second->SubtractResourcesStrict(required_resources); + if (required_resources.IsSubset(node.second)) { + node.second.SubtractResourcesStrict(required_resources); schedule_map[bundle->BundleId()] = node.first; break; } @@ -102,7 +105,10 @@ ScheduleMap GcsSpreadStrategy::Schedule( // bundles will be deployed to the previous nodes. So we start with the next node of the // last selected node. ScheduleMap schedule_map; - const auto &candidate_nodes = context->node_manager_.GetClusterRealtimeResources(); + // TODO(WangTao): This copy might take too much space once cluster grows very large. + // Would find better solution. + absl::flat_hash_map candidate_nodes( + context->node_manager_.GetClusterRealtimeResources()); if (candidate_nodes.empty()) { return schedule_map; } @@ -115,8 +121,8 @@ ScheduleMap GcsSpreadStrategy::Schedule( // meets the resource requirements. `iter_begin` is the next node of the last selected // node. for (; iter != candidate_nodes.end(); ++iter) { - if (required_resources.IsSubset(*iter->second)) { - iter->second->SubtractResourcesStrict(required_resources); + if (required_resources.IsSubset(iter->second)) { + iter->second.SubtractResourcesStrict(required_resources); schedule_map[bundle->BundleId()] = iter->first; break; } @@ -132,8 +138,8 @@ ScheduleMap GcsSpreadStrategy::Schedule( if (iter_begin != candidate_nodes.begin()) { // Traverse all the nodes from `candidate_nodes.begin()` to `iter_begin`. for (iter = candidate_nodes.begin(); iter != iter_begin; ++iter) { - if (required_resources.IsSubset(*iter->second)) { - iter->second->SubtractResourcesStrict(required_resources); + if (required_resources.IsSubset(iter->second)) { + iter->second.SubtractResourcesStrict(required_resources); schedule_map[bundle->BundleId()] = iter->first; break; } @@ -175,7 +181,7 @@ ScheduleMap GcsStrictSpreadStrategy::Schedule( const auto &required_resources = bundle->GetRequiredResources(); auto iter = candidate_nodes.begin(); for (; iter != candidate_nodes.end(); ++iter) { - if (required_resources.IsSubset(*iter->second)) { + if (required_resources.IsSubset(iter->second)) { schedule_map[bundle->BundleId()] = iter->first; candidate_nodes.erase(iter); break; diff --git a/src/ray/gcs/gcs_server/test/gcs_node_manager_test.cc b/src/ray/gcs/gcs_server/test/gcs_node_manager_test.cc index 7b54b410e..a9dfaa9a0 100644 --- a/src/ray/gcs/gcs_server/test/gcs_node_manager_test.cc +++ b/src/ray/gcs/gcs_server/test/gcs_node_manager_test.cc @@ -100,14 +100,14 @@ TEST_F(GcsNodeManagerTest, TestGetClusterRealtimeResources) { ResourceSet required_resources; required_resources.AddOrUpdateResource(cpu_resource, 9); - ASSERT_TRUE(required_resources.IsSubset(*node_resources[node_id])); + ASSERT_TRUE(required_resources.IsSubset(node_resources[node_id])); required_resources.AddOrUpdateResource(cpu_resource, 10); - ASSERT_TRUE(required_resources.IsSubset(*node_resources[node_id])); + ASSERT_TRUE(required_resources.IsSubset(node_resources[node_id])); required_resources.AddOrUpdateResource(cpu_resource, 10.1); - ASSERT_FALSE(required_resources.IsSubset(*node_resources[node_id])); + ASSERT_FALSE(required_resources.IsSubset(node_resources[node_id])); required_resources.DeleteResource(cpu_resource); required_resources.AddOrUpdateResource("GPU", 9); - ASSERT_FALSE(required_resources.IsSubset(*node_resources[node_id])); + ASSERT_FALSE(required_resources.IsSubset(node_resources[node_id])); } } // namespace ray