From d35776e92d622b6b80344c29c09f1acfe4a01e58 Mon Sep 17 00:00:00 2001 From: Tao Wang Date: Thu, 2 Jul 2020 13:56:49 +0800 Subject: [PATCH] [Core] Do not heartbeat resources if no update (#9182) --- src/ray/common/ray_config_def.h | 4 + src/ray/common/task/scheduling_resources.cc | 8 +- src/ray/common/task/scheduling_resources.h | 19 ++- .../test/service_based_gcs_client_test.cc | 4 + src/ray/gcs/gcs_server/gcs_node_manager.cc | 8 +- src/ray/gcs/gcs_server/gcs_node_manager.h | 2 + src/ray/raylet/node_manager.cc | 125 ++++++++++++++---- src/ray/raylet/node_manager.h | 5 + 8 files changed, 142 insertions(+), 33 deletions(-) diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h index 5adc543b1..0721f481f 100644 --- a/src/ray/common/ray_config_def.h +++ b/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. diff --git a/src/ray/common/task/scheduling_resources.cc b/src/ray/common/task/scheduling_resources.cc index 8836a5494..4d0f4b2ef 100644 --- a/src/ray/common/task/scheduling_resources.cc +++ b/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, diff --git a/src/ray/common/task/scheduling_resources.h b/src/ray/common/task/scheduling_resources.h index 9d39a8963..7f4a0bdcc 100644 --- a/src/ray/common/task/scheduling_resources.h +++ b/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. diff --git a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc index d5d9188a4..a625848fd 100644 --- a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc +++ b/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(); 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(); 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); diff --git a/src/ray/gcs/gcs_server/gcs_node_manager.cc b/src/ray/gcs/gcs_server/gcs_node_manager.cc index 52e6cfcfc..8709fc749 100644 --- a/src/ray/gcs/gcs_server/gcs_node_manager.cc +++ b/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. diff --git a/src/ray/gcs/gcs_server/gcs_node_manager.h b/src/ray/gcs/gcs_server/gcs_node_manager.h index 34823079c..5c1ee55ea 100644 --- a/src/ray/gcs/gcs_server/gcs_node_manager.h +++ b/src/ray/gcs/gcs_server/gcs_node_manager.h @@ -182,6 +182,8 @@ class GcsNodeManager : public rpc::NodeInfoHandler { std::function 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 diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 5f5b9747a..fa294c3f7 100644 --- a/src/ray/raylet/node_manager.cc +++ b/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(); 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; diff --git a/src/ray/raylet/node_manager.h b/src/ray/raylet/node_manager.h index 4a52eab36..b02e13a10 100644 --- a/src/ray/raylet/node_manager.h +++ b/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