[Core] Do not heartbeat resources if no update (#9182)

src/ray/common/ray_config_def.h

@@ -36,6 +36,10 @@ RAY_CONFIG(int64_t, handler_warning_timeout_ms, 100)
 
 /// The duration between heartbeats sent by the raylets.
 RAY_CONFIG(int64_t, raylet_heartbeat_timeout_milliseconds, 100)
+/// Whether to send heartbeat lightly. When it is enalbed, only changed part,
+/// like should_global_gc or changed resources, will be included in the heartbeat,
+/// and gcs only broadcast the changed heartbeat.
+RAY_CONFIG(bool, light_heartbeat_enabled, false)
 /// If a component has not sent a heartbeat in the last num_heartbeats_timeout
 /// heartbeat intervals, the raylet monitor process will report
 /// it as dead to the db_client table.

src/ray/common/task/scheduling_resources.cc

@@ -743,14 +743,18 @@ const ResourceSet &SchedulingResources::GetTotalResources() const {
   return resources_total_;
 }
 
-void SchedulingResources::SetLoadResources(ResourceSet &&newset) {
-  resources_load_ = newset;
+void SchedulingResources::SetTotalResources(ResourceSet &&newset) {
+  resources_total_ = newset;
 }
 
 const ResourceSet &SchedulingResources::GetLoadResources() const {
   return resources_load_;
 }
 
+void SchedulingResources::SetLoadResources(ResourceSet &&newset) {
+  resources_load_ = newset;
+}
+
 // Return specified resources back to SchedulingResources.
 void SchedulingResources::Release(const ResourceSet &resources) {
   return resources_available_.AddResourcesCapacityConstrained(resources,

src/ray/common/task/scheduling_resources.h

@@ -477,19 +477,28 @@ class SchedulingResources {
   /// \return Void.
   void SetAvailableResources(ResourceSet &&newset);
 
+  /// \brief Request the total resources capacity.
+  ///
+  /// \return Immutable set of resources with currently total capacity.
   const ResourceSet &GetTotalResources() const;
 
+  /// \brief Overwrite total resource capacity with the specified resource set.
+  ///
+  /// \param newset: The set of resources that replaces total resource capacity.
+  /// \return Void.
+  void SetTotalResources(ResourceSet &&newset);
+
+  /// \brief Request the resource load information.
+  ///
+  /// \return Immutable set of resources describing the load information.
+  const ResourceSet &GetLoadResources() const;
+
   /// \brief Overwrite information about resource load with new resource load set.
   ///
   /// \param newset: The set of resources that replaces resource load information.
   /// \return Void.
   void SetLoadResources(ResourceSet &&newset);
 
-  /// \brief Request the resource load information.
-  ///
-  /// \return Immutable set of resources describing the load information.
-  const ResourceSet &GetLoadResources() const;
-
   /// \brief Release the amount of resources specified.
   ///
   /// \param resources: the amount of resources to be released.

src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc

@@ -699,6 +699,8 @@ TEST_F(ServiceBasedGcsClientTest, TestNodeHeartbeat) {
   ClientID node_id = ClientID::FromBinary(node_info->node_id());
   auto heartbeat = std::make_shared<rpc::HeartbeatTableData>();
   heartbeat->set_client_id(node_id.Binary());
+  // Set this flag because GCS won't publish unchanged heartbeat.
+  heartbeat->set_should_global_gc(true);
   ASSERT_TRUE(ReportHeartbeat(heartbeat));
   WaitPendingDone(heartbeat_batch_count, 1);
 }
@@ -1001,6 +1003,8 @@ TEST_F(ServiceBasedGcsClientTest, TestNodeTableResubscribe) {
   ASSERT_TRUE(UpdateResources(node_id, key));
   auto heartbeat = std::make_shared<rpc::HeartbeatTableData>();
   heartbeat->set_client_id(node_info->node_id());
+  // Set this flag because GCS won't publish unchanged heartbeat.
+  heartbeat->set_should_global_gc(true);
   ASSERT_TRUE(ReportHeartbeat(heartbeat));
   WaitPendingDone(batch_heartbeat_count, 1);
 

src/ray/gcs/gcs_server/gcs_node_manager.cc

@@ -28,6 +28,7 @@ GcsNodeManager::NodeFailureDetector::NodeFailureDetector(
     : gcs_table_storage_(std::move(gcs_table_storage)),
       on_node_death_callback_(std::move(on_node_death_callback)),
       num_heartbeats_timeout_(RayConfig::instance().num_heartbeats_timeout()),
+      light_heartbeat_enabled_(RayConfig::instance().light_heartbeat_enabled()),
       detect_timer_(io_service),
       gcs_pub_sub_(std::move(gcs_pub_sub)) {
   Tick();
@@ -48,7 +49,12 @@ void GcsNodeManager::NodeFailureDetector::HandleHeartbeat(
   }
 
   iter->second = num_heartbeats_timeout_;
-  heartbeat_buffer_[node_id] = heartbeat_data;
+  if (!light_heartbeat_enabled_ || heartbeat_data.should_global_gc() ||
+      heartbeat_data.resources_available_label_size() > 0 ||
+      heartbeat_data.resources_total_label_size() > 0 ||
+      heartbeat_data.resource_load_label_size() > 0) {
+    heartbeat_buffer_[node_id] = heartbeat_data;
+  }
 }
 
 /// A periodic timer that checks for timed out clients.

src/ray/gcs/gcs_server/gcs_node_manager.h

@@ -182,6 +182,8 @@ class GcsNodeManager : public rpc::NodeInfoHandler {
     std::function<void(const ClientID &)> on_node_death_callback_;
     /// The number of heartbeats that can be missed before a node is removed.
     int64_t num_heartbeats_timeout_;
+    // Only the changed part will be included in heartbeat if this is true.
+    const bool light_heartbeat_enabled_;
     /// A timer that ticks every heartbeat_timeout_ms_ milliseconds.
     boost::asio::deadline_timer detect_timer_;
     /// For each Raylet that we receive a heartbeat from, the number of ticks

src/ray/raylet/node_manager.cc

@@ -137,6 +137,7 @@ NodeManager::NodeManager(boost::asio::io_service &io_service,
       object_pinning_enabled_(config.object_pinning_enabled),
       temp_dir_(config.temp_dir),
       object_manager_profile_timer_(io_service),
+      light_heartbeat_enabled_(RayConfig::instance().light_heartbeat_enabled()),
       initial_config_(config),
       local_available_resources_(config.resource_config),
       worker_pool_(
@@ -341,22 +342,70 @@ void NodeManager::Heartbeat() {
   auto heartbeat_data = std::make_shared<HeartbeatTableData>();
   SchedulingResources &local_resources = cluster_resource_map_[self_node_id_];
   heartbeat_data->set_client_id(self_node_id_.Binary());
+
   // TODO(atumanov): modify the heartbeat table protocol to use the ResourceSet directly.
   // TODO(atumanov): implement a ResourceSet const_iterator.
-  for (const auto &resource_pair :
-       local_resources.GetAvailableResources().GetResourceMap()) {
-    heartbeat_data->add_resources_available_label(resource_pair.first);
-    heartbeat_data->add_resources_available_capacity(resource_pair.second);
-  }
-  for (const auto &resource_pair : local_resources.GetTotalResources().GetResourceMap()) {
-    heartbeat_data->add_resources_total_label(resource_pair.first);
-    heartbeat_data->add_resources_total_capacity(resource_pair.second);
-  }
+  // If light heartbeat enabled, we only set filed that represent resources changed.
+  if (light_heartbeat_enabled_) {
+    if (!last_heartbeat_resources_.GetAvailableResources().IsEqual(
+            local_resources.GetAvailableResources())) {
+      for (const auto &resource_pair :
+           local_resources.GetAvailableResources().GetResourceMap()) {
+        heartbeat_data->add_resources_available_label(resource_pair.first);
+        heartbeat_data->add_resources_available_capacity(resource_pair.second);
+      }
+      last_heartbeat_resources_.SetAvailableResources(
+          ResourceSet(local_resources.GetAvailableResources()));
+    }
 
-  local_resources.SetLoadResources(local_queues_.GetResourceLoad());
-  for (const auto &resource_pair : local_resources.GetLoadResources().GetResourceMap()) {
-    heartbeat_data->add_resource_load_label(resource_pair.first);
-    heartbeat_data->add_resource_load_capacity(resource_pair.second);
+    if (!last_heartbeat_resources_.GetTotalResources().IsEqual(
+            local_resources.GetTotalResources())) {
+      for (const auto &resource_pair :
+           local_resources.GetTotalResources().GetResourceMap()) {
+        heartbeat_data->add_resources_total_label(resource_pair.first);
+        heartbeat_data->add_resources_total_capacity(resource_pair.second);
+      }
+      last_heartbeat_resources_.SetTotalResources(
+          ResourceSet(local_resources.GetTotalResources()));
+    }
+
+    local_resources.SetLoadResources(local_queues_.GetResourceLoad());
+    if (!last_heartbeat_resources_.GetLoadResources().IsEqual(
+            local_resources.GetLoadResources())) {
+      for (const auto &resource_pair :
+           local_resources.GetLoadResources().GetResourceMap()) {
+        heartbeat_data->add_resource_load_label(resource_pair.first);
+        heartbeat_data->add_resource_load_capacity(resource_pair.second);
+      }
+      last_heartbeat_resources_.SetLoadResources(
+          ResourceSet(local_resources.GetLoadResources()));
+    }
+  } else {
+    // If light heartbeat disabled, we send whole resources information every time.
+    for (const auto &resource_pair :
+         local_resources.GetAvailableResources().GetResourceMap()) {
+      heartbeat_data->add_resources_available_label(resource_pair.first);
+      heartbeat_data->add_resources_available_capacity(resource_pair.second);
+    }
+    last_heartbeat_resources_.SetAvailableResources(
+        ResourceSet(local_resources.GetAvailableResources()));
+
+    for (const auto &resource_pair :
+         local_resources.GetTotalResources().GetResourceMap()) {
+      heartbeat_data->add_resources_total_label(resource_pair.first);
+      heartbeat_data->add_resources_total_capacity(resource_pair.second);
+    }
+    last_heartbeat_resources_.SetTotalResources(
+        ResourceSet(local_resources.GetTotalResources()));
+
+    local_resources.SetLoadResources(local_queues_.GetResourceLoad());
+    for (const auto &resource_pair :
+         local_resources.GetLoadResources().GetResourceMap()) {
+      heartbeat_data->add_resource_load_label(resource_pair.first);
+      heartbeat_data->add_resource_load_capacity(resource_pair.second);
+    }
+    last_heartbeat_resources_.SetLoadResources(
+        ResourceSet(local_resources.GetLoadResources()));
   }
 
   // Set the global gc bit on the outgoing heartbeat message.
@@ -748,21 +797,47 @@ void NodeManager::HeartbeatAdded(const ClientID &client_id,
 
   SchedulingResources &remote_resources = it->second;
 
-  ResourceSet remote_total(VectorFromProtobuf(heartbeat_data.resources_total_label()),
-                           VectorFromProtobuf(heartbeat_data.resources_total_capacity()));
-  ResourceSet remote_available(
-      VectorFromProtobuf(heartbeat_data.resources_available_label()),
-      VectorFromProtobuf(heartbeat_data.resources_available_capacity()));
-  ResourceSet remote_load(VectorFromProtobuf(heartbeat_data.resource_load_label()),
-                          VectorFromProtobuf(heartbeat_data.resource_load_capacity()));
-  // TODO(atumanov): assert that the load is a non-empty ResourceSet.
-  remote_resources.SetAvailableResources(std::move(remote_available));
-  // Extract the load information and save it locally.
-  remote_resources.SetLoadResources(std::move(remote_load));
+  // If light heartbeat enabled, we update remote resources only when related resources
+  // map in heartbeat is not empty.
+  if (light_heartbeat_enabled_) {
+    if (heartbeat_data.resources_total_label_size() > 0) {
+      ResourceSet remote_total(
+          VectorFromProtobuf(heartbeat_data.resources_total_label()),
+          VectorFromProtobuf(heartbeat_data.resources_total_capacity()));
+      remote_resources.SetTotalResources(std::move(remote_total));
+    }
+    if (heartbeat_data.resources_available_label_size() > 0) {
+      ResourceSet remote_available(
+          VectorFromProtobuf(heartbeat_data.resources_available_label()),
+          VectorFromProtobuf(heartbeat_data.resources_available_capacity()));
+      remote_resources.SetAvailableResources(std::move(remote_available));
+    }
+    if (heartbeat_data.resource_load_label_size() > 0) {
+      ResourceSet remote_load(
+          VectorFromProtobuf(heartbeat_data.resource_load_label()),
+          VectorFromProtobuf(heartbeat_data.resource_load_capacity()));
+      // Extract the load information and save it locally.
+      remote_resources.SetLoadResources(std::move(remote_load));
+    }
+  } else {
+    // If light heartbeat disabled, we update remote resources every time.
+    ResourceSet remote_total(
+        VectorFromProtobuf(heartbeat_data.resources_total_label()),
+        VectorFromProtobuf(heartbeat_data.resources_total_capacity()));
+    remote_resources.SetTotalResources(std::move(remote_total));
+    ResourceSet remote_available(
+        VectorFromProtobuf(heartbeat_data.resources_available_label()),
+        VectorFromProtobuf(heartbeat_data.resources_available_capacity()));
+    remote_resources.SetAvailableResources(std::move(remote_available));
+    ResourceSet remote_load(VectorFromProtobuf(heartbeat_data.resource_load_label()),
+                            VectorFromProtobuf(heartbeat_data.resource_load_capacity()));
+    // Extract the load information and save it locally.
+    remote_resources.SetLoadResources(std::move(remote_load));
+  }
 
   if (new_scheduler_enabled_ && client_id != self_node_id_) {
     new_resource_scheduler_->AddOrUpdateNode(
-        client_id.Binary(), remote_total.GetResourceMap(),
+        client_id.Binary(), remote_resources.GetTotalResources().GetResourceMap(),
         remote_resources.GetAvailableResources().GetResourceMap());
     NewSchedulerSchedulePendingTasks();
     return;

src/ray/raylet/node_manager.h

@@ -676,6 +676,11 @@ class NodeManager : public rpc::NodeManagerServiceHandler {
   /// The time that the last heartbeat was sent at. Used to make sure we are
   /// keeping up with heartbeats.
   uint64_t last_heartbeat_at_ms_;
+  /// Only the changed part will be included in heartbeat if this is true.
+  const bool light_heartbeat_enabled_;
+  /// Cache which stores resources in last heartbeat used to check if they are changed.
+  /// Used by light heartbeat.
+  SchedulingResources last_heartbeat_resources_;
   /// The time that the last debug string was logged to the console.
   uint64_t last_debug_dump_at_ms_;
   /// The time that we last sent a FreeObjects request to other nodes for