wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-08-03 13:10:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

[GCS]Decouple node failure detector with resoure related operations (#11465)

Browse Source
This commit is contained in:
Tao Wang
2020-10-27 15:52:42 -07:00
committed by GitHub
parent 1c40950877
commit 273a712786
4 changed files with 92 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/gcs/gcs_server/gcs_node_manager.cc
+77 -69
View File
@@ -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<rpc::PlacementGroupLoad> 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<rpc::HeartbeatBatchTableData>();
std::unordered_map<ResourceSet, rpc::ResourceDemand> 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<rpc::GcsNodeInfo> 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<NodeID, std::shared_ptr<ResourceSet>>
}
void GcsNodeManager::UpdatePlacementGroupLoad(
std::shared_ptr<rpc::PlacementGroupLoad> 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<rpc::PlacementGroupLoad> placement_group_load) {
placement_group_load_ = absl::make_optional(placement_group_load);
}
void GcsNodeManager::AddDeadNodeToCache(std::shared_ptr<rpc::GcsNodeInfo> node) {
@@ -532,5 +477,68 @@ void GcsNodeManager::AddDeadNodeToCache(std::shared_ptr<rpc::GcsNodeInfo> node)
sorted_dead_node_list_.emplace_back(node_id, node->timestamp());
}
void GcsNodeManager::SendBatchedHeartbeat() {
if (!heartbeat_buffer_.empty()) {
auto batch = std::make_shared<rpc::HeartbeatBatchTableData>();
std::unordered_map<ResourceSet, rpc::ResourceDemand> 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
src/ray/gcs/gcs_server/gcs_node_manager.h
+10 -14
View File
@@ -162,7 +162,7 @@ class GcsNodeManager : public rpc::NodeInfoHandler {
///
/// \param placement_group_load placement group load protobuf.
void UpdatePlacementGroupLoad(
std::shared_ptr<rpc::PlacementGroupLoad> placement_group_load) const;
const std::shared_ptr<rpc::PlacementGroupLoad> 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<rpc::PlacementGroupLoad> 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<NodeID, int64_t> heartbeats_;
/// A buffer containing heartbeats received from node managers in the last tick.
absl::flat_hash_map<NodeID, rpc::HeartbeatTableData> heartbeat_buffer_;
/// A publisher for publishing gcs messages.
std::shared_ptr<gcs::GcsPubSub> gcs_pub_sub_;
/// Is the detect started.
bool is_started_ = false;
/// Placement group load information that is used for autoscaler.
absl::optional<std::shared_ptr<rpc::PlacementGroupLoad>> 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<rpc::GcsNodeInfo> 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<NodeFailureDetector> 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<NodeID, std::shared_ptr<rpc::GcsNodeInfo>> alive_nodes_;
/// Dead nodes.
@@ -267,6 +259,8 @@ class GcsNodeManager : public rpc::NodeInfoHandler {
std::list<std::pair<NodeID, int64_t>> sorted_dead_node_list_;
/// Cluster resources.
absl::flat_hash_map<NodeID, rpc::ResourceMap> cluster_resources_;
/// A buffer containing heartbeats received from node managers in the last tick.
absl::flat_hash_map<NodeID, rpc::HeartbeatTableData> heartbeat_buffer_;
/// Listeners which monitors the addition of nodes.
std::vector<std::function<void(std::shared_ptr<rpc::GcsNodeInfo>)>>
node_added_listeners_;
@@ -279,6 +273,8 @@ class GcsNodeManager : public rpc::NodeInfoHandler {
std::shared_ptr<gcs::GcsTableStorage> gcs_table_storage_;
/// Cluster realtime resources.
absl::flat_hash_map<NodeID, std::shared_ptr<ResourceSet>> cluster_realtime_resources_;
/// Placement group load information that is used for autoscaler.
absl::optional<std::shared_ptr<rpc::PlacementGroupLoad>> placement_group_load_;
};
} // namespace gcs
src/ray/gcs/gcs_server/gcs_placement_group_manager.cc
+2 -2
View File
@@ -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<rpc::PlacementGroupLoad> placement_group_load =
std::make_shared<rpc::PlacementGroupLoad>();
int total_cnt = 0;
src/ray/gcs/gcs_server/gcs_placement_group_manager.h
+3 -3
View File
@@ -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