[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

@@ -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<ResourceSet>(resources_available);
+        ResourceSet(MapFromProtobuf(heartbeat.resources_available()));
   }
 }
 
-const absl::flat_hash_map<NodeID, std::shared_ptr<ResourceSet>>
+const absl::flat_hash_map<NodeID, ResourceSet>
     &GcsNodeManager::GetClusterRealtimeResources() const {
   return cluster_realtime_resources_;
 }

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<NodeID, std::shared_ptr<ResourceSet>>
-      &GetClusterRealtimeResources() const;
+  const absl::flat_hash_map<NodeID, ResourceSet> &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::GcsTableStorage> gcs_table_storage_;
   /// Cluster realtime resources.
-  absl::flat_hash_map<NodeID, std::shared_ptr<ResourceSet>> cluster_realtime_resources_;
+  absl::flat_hash_map<NodeID, ResourceSet> cluster_realtime_resources_;
   /// Placement group load information that is used for autoscaler.
   absl::optional<std::shared_ptr<rpc::PlacementGroupLoad>> placement_group_load_;
 

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<std::pair<int64_t, NodeID>> 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<NodeID, ResourceSet> 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<NodeID, ResourceSet> 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;

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