From 273a71278698946c07fb93e329a6b4d17a1f426a Mon Sep 17 00:00:00 2001 From: Tao Wang Date: Wed, 28 Oct 2020 06:52:42 +0800 Subject: [PATCH] [GCS]Decouple node failure detector with resoure related operations (#11465) --- src/ray/gcs/gcs_server/gcs_node_manager.cc | 146 +++++++++--------- src/ray/gcs/gcs_server/gcs_node_manager.h | 24 ++- .../gcs_server/gcs_placement_group_manager.cc | 4 +- .../gcs_server/gcs_placement_group_manager.h | 6 +- 4 files changed, 92 insertions(+), 88 deletions(-) diff --git a/src/ray/gcs/gcs_server/gcs_node_manager.cc b/src/ray/gcs/gcs_server/gcs_node_manager.cc index 4e23930d9..f07bf4a78 100644 --- a/src/ray/gcs/gcs_server/gcs_node_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_node_manager.cc @@ -55,23 +55,11 @@ void GcsNodeManager::NodeFailureDetector::HandleHeartbeat( } iter->second = num_heartbeats_timeout_; - if (!light_heartbeat_enabled_ || heartbeat_data.should_global_gc() || - heartbeat_data.resources_total_size() > 0 || - heartbeat_data.resources_available_changed() || - heartbeat_data.resource_load_changed()) { - heartbeat_buffer_[node_id] = heartbeat_data; - } -} - -void GcsNodeManager::NodeFailureDetector::UpdatePlacementGroupLoad( - std::shared_ptr placement_group_load) { - placement_group_load_ = absl::make_optional(placement_group_load); } /// A periodic timer that checks for timed out clients. void GcsNodeManager::NodeFailureDetector::Tick() { DetectDeadNodes(); - SendBatchedHeartbeat(); ScheduleTick(); } @@ -83,7 +71,6 @@ void GcsNodeManager::NodeFailureDetector::DetectDeadNodes() { auto node_id = current->first; RAY_LOG(WARNING) << "Node timed out: " << node_id; heartbeats_.erase(current); - heartbeat_buffer_.erase(node_id); if (on_node_death_callback_) { on_node_death_callback_(node_id); } @@ -91,55 +78,6 @@ void GcsNodeManager::NodeFailureDetector::DetectDeadNodes() { } } -void GcsNodeManager::NodeFailureDetector::SendBatchedHeartbeat() { - if (!heartbeat_buffer_.empty()) { - auto batch = std::make_shared(); - std::unordered_map aggregate_load; - for (auto &heartbeat : heartbeat_buffer_) { - // Aggregate the load reported by each raylet. - auto load = heartbeat.second.resource_load_by_shape(); - for (const auto &demand : load.resource_demands()) { - auto scheduling_key = ResourceSet(MapFromProtobuf(demand.shape())); - auto &aggregate_demand = aggregate_load[scheduling_key]; - aggregate_demand.set_num_ready_requests_queued( - aggregate_demand.num_ready_requests_queued() + - demand.num_ready_requests_queued()); - aggregate_demand.set_num_infeasible_requests_queued( - aggregate_demand.num_infeasible_requests_queued() + - demand.num_infeasible_requests_queued()); - if (RayConfig::instance().report_worker_backlog()) { - aggregate_demand.set_backlog_size(aggregate_demand.backlog_size() + - demand.backlog_size()); - } - } - heartbeat.second.clear_resource_load_by_shape(); - - batch->add_batch()->Swap(&heartbeat.second); - } - - for (auto &demand : aggregate_load) { - auto demand_proto = batch->mutable_resource_load_by_shape()->add_resource_demands(); - demand_proto->Swap(&demand.second); - for (const auto &resource_pair : demand.first.GetResourceMap()) { - (*demand_proto->mutable_shape())[resource_pair.first] = resource_pair.second; - } - } - - // Update placement group load to heartbeat batch. - // This is updated only one per second. - if (placement_group_load_.has_value()) { - auto placement_group_load = placement_group_load_.value(); - auto placement_group_load_proto = batch->mutable_placement_group_load(); - placement_group_load_proto->Swap(placement_group_load.get()); - placement_group_load_.reset(); - } - - RAY_CHECK_OK(gcs_pub_sub_->Publish(HEARTBEAT_BATCH_CHANNEL, "", - batch->SerializeAsString(), nullptr)); - heartbeat_buffer_.clear(); - } -} - void GcsNodeManager::NodeFailureDetector::ScheduleTick() { auto heartbeat_period = boost::posix_time::milliseconds( RayConfig::instance().raylet_heartbeat_timeout_milliseconds()); @@ -184,8 +122,11 @@ GcsNodeManager::GcsNodeManager(boost::asio::io_service &main_io_service, }); })), node_failure_detector_service_(node_failure_detector_io_service), + heartbeat_timer_(main_io_service), gcs_pub_sub_(gcs_pub_sub), - gcs_table_storage_(gcs_table_storage) {} + gcs_table_storage_(gcs_table_storage) { + SendBatchedHeartbeat(); +} void GcsNodeManager::HandleRegisterNode(const rpc::RegisterNodeRequest &request, rpc::RegisterNodeReply *reply, @@ -255,6 +196,13 @@ void GcsNodeManager::HandleReportHeartbeat(const rpc::ReportHeartbeatRequest &re // Update node realtime resources. UpdateNodeRealtimeResources(node_id, *heartbeat_data); + if (!RayConfig::instance().light_heartbeat_enabled() || + heartbeat_data->should_global_gc() || heartbeat_data->resources_total_size() > 0 || + heartbeat_data->resources_available_changed() || + heartbeat_data->resource_load_changed()) { + heartbeat_buffer_[node_id] = *heartbeat_data; + } + // Note: To avoid heartbeats being delayed by main thread, make sure heartbeat is always // handled by its own IO service. node_failure_detector_service_.post([this, node_id, heartbeat_data] { @@ -430,6 +378,7 @@ std::shared_ptr GcsNodeManager::RemoveNode( cluster_resources_.erase(node_id); // Remove from cluster realtime resources. cluster_realtime_resources_.erase(node_id); + heartbeat_buffer_.erase(node_id); if (!is_intended) { // Broadcast a warning to all of the drivers indicating that the node // has been marked as dead. @@ -512,12 +461,8 @@ const absl::flat_hash_map> } void GcsNodeManager::UpdatePlacementGroupLoad( - std::shared_ptr placement_group_load) const { - // Node failure detector code should be running in a separate thread to avoid heartbeat - // lagging. - node_failure_detector_service_.post([this, placement_group_load] { - node_failure_detector_->UpdatePlacementGroupLoad(move(placement_group_load)); - }); + const std::shared_ptr placement_group_load) { + placement_group_load_ = absl::make_optional(placement_group_load); } void GcsNodeManager::AddDeadNodeToCache(std::shared_ptr node) { @@ -532,5 +477,68 @@ void GcsNodeManager::AddDeadNodeToCache(std::shared_ptr node) sorted_dead_node_list_.emplace_back(node_id, node->timestamp()); } +void GcsNodeManager::SendBatchedHeartbeat() { + if (!heartbeat_buffer_.empty()) { + auto batch = std::make_shared(); + std::unordered_map aggregate_load; + for (auto &heartbeat : heartbeat_buffer_) { + // Aggregate the load reported by each raylet. + auto load = heartbeat.second.resource_load_by_shape(); + for (const auto &demand : load.resource_demands()) { + auto scheduling_key = ResourceSet(MapFromProtobuf(demand.shape())); + auto &aggregate_demand = aggregate_load[scheduling_key]; + aggregate_demand.set_num_ready_requests_queued( + aggregate_demand.num_ready_requests_queued() + + demand.num_ready_requests_queued()); + aggregate_demand.set_num_infeasible_requests_queued( + aggregate_demand.num_infeasible_requests_queued() + + demand.num_infeasible_requests_queued()); + if (RayConfig::instance().report_worker_backlog()) { + aggregate_demand.set_backlog_size(aggregate_demand.backlog_size() + + demand.backlog_size()); + } + } + heartbeat.second.clear_resource_load_by_shape(); + + batch->add_batch()->Swap(&heartbeat.second); + } + + for (auto &demand : aggregate_load) { + auto demand_proto = batch->mutable_resource_load_by_shape()->add_resource_demands(); + demand_proto->Swap(&demand.second); + for (const auto &resource_pair : demand.first.GetResourceMap()) { + (*demand_proto->mutable_shape())[resource_pair.first] = resource_pair.second; + } + } + + // Update placement group load to heartbeat batch. + // This is updated only one per second. + if (placement_group_load_.has_value()) { + auto placement_group_load = placement_group_load_.value(); + auto placement_group_load_proto = batch->mutable_placement_group_load(); + placement_group_load_proto->Swap(placement_group_load.get()); + placement_group_load_.reset(); + } + + RAY_CHECK_OK(gcs_pub_sub_->Publish(HEARTBEAT_BATCH_CHANNEL, "", + batch->SerializeAsString(), nullptr)); + heartbeat_buffer_.clear(); + } + + auto heartbeat_period = boost::posix_time::milliseconds( + RayConfig::instance().raylet_heartbeat_timeout_milliseconds()); + heartbeat_timer_.expires_from_now(heartbeat_period); + heartbeat_timer_.async_wait([this](const boost::system::error_code &error) { + if (error == boost::asio::error::operation_aborted) { + // `operation_aborted` is set when `heartbeat_timer_` is canceled or destroyed. + // The Monitor lifetime may be short than the object who use it. (e.g. gcs_server) + return; + } + RAY_CHECK(!error) << "Sending batched heartbeat failed with error: " + << error.message(); + SendBatchedHeartbeat(); + }); +} + } // namespace gcs } // namespace ray diff --git a/src/ray/gcs/gcs_server/gcs_node_manager.h b/src/ray/gcs/gcs_server/gcs_node_manager.h index 65680ecab..f927ea5ac 100644 --- a/src/ray/gcs/gcs_server/gcs_node_manager.h +++ b/src/ray/gcs/gcs_server/gcs_node_manager.h @@ -162,7 +162,7 @@ class GcsNodeManager : public rpc::NodeInfoHandler { /// /// \param placement_group_load placement group load protobuf. void UpdatePlacementGroupLoad( - std::shared_ptr placement_group_load) const; + const std::shared_ptr placement_group_load); protected: class NodeFailureDetector { @@ -199,12 +199,6 @@ class GcsNodeManager : public rpc::NodeInfoHandler { void HandleHeartbeat(const NodeID &node_id, const rpc::HeartbeatTableData &heartbeat_data); - /// Public interface to update placement group load information. - /// - /// \param placement_group_load placement group load protobuf. - void UpdatePlacementGroupLoad( - std::shared_ptr placement_group_load); - protected: /// A periodic timer that fires on every heartbeat period. Raylets that have /// not sent a heartbeat within the last num_heartbeats_timeout ticks will be @@ -215,9 +209,6 @@ class GcsNodeManager : public rpc::NodeInfoHandler { /// If found any, mark it as dead. void DetectDeadNodes(); - /// Send any buffered heartbeats as a single publish. - void SendBatchedHeartbeat(); - /// Schedule another tick after a short time. void ScheduleTick(); @@ -235,14 +226,10 @@ class GcsNodeManager : public rpc::NodeInfoHandler { /// For each Raylet that we receive a heartbeat from, the number of ticks /// that may pass before the Raylet will be declared dead. absl::flat_hash_map heartbeats_; - /// A buffer containing heartbeats received from node managers in the last tick. - absl::flat_hash_map heartbeat_buffer_; /// A publisher for publishing gcs messages. std::shared_ptr gcs_pub_sub_; /// Is the detect started. bool is_started_ = false; - /// Placement group load information that is used for autoscaler. - absl::optional> placement_group_load_; }; private: @@ -252,12 +239,17 @@ class GcsNodeManager : public rpc::NodeInfoHandler { /// \param node The node which is dead. void AddDeadNodeToCache(std::shared_ptr node); + /// Send any buffered heartbeats as a single publish. + void SendBatchedHeartbeat(); + /// The main event loop for node failure detector. boost::asio::io_service &main_io_service_; /// Detector to detect the failure of node. std::unique_ptr node_failure_detector_; /// The event loop for node failure detector. boost::asio::io_service &node_failure_detector_service_; + /// A timer that ticks every heartbeat_timeout_ms_ milliseconds. + boost::asio::deadline_timer heartbeat_timer_; /// Alive nodes. absl::flat_hash_map> alive_nodes_; /// Dead nodes. @@ -267,6 +259,8 @@ class GcsNodeManager : public rpc::NodeInfoHandler { std::list> sorted_dead_node_list_; /// Cluster resources. absl::flat_hash_map cluster_resources_; + /// A buffer containing heartbeats received from node managers in the last tick. + absl::flat_hash_map heartbeat_buffer_; /// Listeners which monitors the addition of nodes. std::vector)>> node_added_listeners_; @@ -279,6 +273,8 @@ class GcsNodeManager : public rpc::NodeInfoHandler { std::shared_ptr gcs_table_storage_; /// Cluster realtime resources. absl::flat_hash_map> cluster_realtime_resources_; + /// Placement group load information that is used for autoscaler. + absl::optional> placement_group_load_; }; } // namespace gcs diff --git a/src/ray/gcs/gcs_server/gcs_placement_group_manager.cc b/src/ray/gcs/gcs_server/gcs_placement_group_manager.cc index a593bd3e5..326381538 100644 --- a/src/ray/gcs/gcs_server/gcs_placement_group_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_placement_group_manager.cc @@ -375,12 +375,12 @@ void GcsPlacementGroupManager::OnNodeDead(const NodeID &node_id) { SchedulePendingPlacementGroups(); } -void GcsPlacementGroupManager::Tick() const { +void GcsPlacementGroupManager::Tick() { UpdatePlacementGroupLoad(); execute_after(io_context_, [this] { Tick(); }, 1000 /* milliseconds */); } -void GcsPlacementGroupManager::UpdatePlacementGroupLoad() const { +void GcsPlacementGroupManager::UpdatePlacementGroupLoad() { std::shared_ptr placement_group_load = std::make_shared(); int total_cnt = 0; diff --git a/src/ray/gcs/gcs_server/gcs_placement_group_manager.h b/src/ray/gcs/gcs_server/gcs_placement_group_manager.h index a1bbf2d36..e2e81b36e 100644 --- a/src/ray/gcs/gcs_server/gcs_placement_group_manager.h +++ b/src/ray/gcs/gcs_server/gcs_placement_group_manager.h @@ -199,10 +199,10 @@ class GcsPlacementGroupManager : public rpc::PlacementGroupInfoHandler { } // Method that is invoked every second. - void Tick() const; + void Tick(); // Update placement group load information so that the autoscaler can use it. - void UpdatePlacementGroupLoad() const; + void UpdatePlacementGroupLoad(); /// The io loop that is used to delay execution of tasks (e.g., /// execute_after). @@ -236,7 +236,7 @@ class GcsPlacementGroupManager : public rpc::PlacementGroupInfoHandler { PlacementGroupID scheduling_in_progress_id_ = PlacementGroupID::Nil(); /// Reference of GcsNodeManager. - const GcsNodeManager &gcs_node_manager_; + GcsNodeManager &gcs_node_manager_; }; } // namespace gcs