wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-19 11:27:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

[GCS]Eviction of dead nodes cached in GCS (#11323)

Browse Source
This commit is contained in:
fangfengbin
2020-10-12 15:54:32 +08:00
committed by GitHub
parent 31117b5e96
commit d1579819e9
5 changed files with 72 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/common/ray_config_def.h
+2
View File
@@ -234,6 +234,8 @@ RAY_CONFIG(uint32_t, gcs_create_actor_retry_interval_ms, 200)
RAY_CONFIG(uint32_t, gcs_create_placement_group_retry_interval_ms, 200)
/// Maximum number of destroyed actors in GCS server memory cache.
RAY_CONFIG(uint32_t, maximum_gcs_destroyed_actor_cached_count, 10000)
/// Maximum number of dead nodes in GCS server memory cache.
RAY_CONFIG(uint32_t, maximum_gcs_dead_node_cached_count, 1000)
/// Maximum number of times to retry putting an object when the plasma store is full.
/// Can be set to -1 to enable unlimited retries.
src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc
+37 -1
View File
@@ -29,7 +29,8 @@ class ServiceBasedGcsClientTest : public ::testing::Test {
ServiceBasedGcsClientTest() {
RayConfig::instance().initialize(
{{"ping_gcs_rpc_server_max_retries", std::to_string(60)},
{"maximum_gcs_destroyed_actor_cached_count", std::to_string(10)}});
{"maximum_gcs_destroyed_actor_cached_count", std::to_string(10)},
{"maximum_gcs_dead_node_cached_count", std::to_string(10)}});
TestSetupUtil::StartUpRedisServers(std::vector<int>());
}
@@ -571,6 +572,18 @@ class ServiceBasedGcsClientTest : public ::testing::Test {
ASSERT_TRUE(actor.state() == expected_state);
}
absl::flat_hash_set<NodeID> RegisterNodeAndMarkDead(int node_count) {
absl::flat_hash_set<NodeID> node_ids;
for (int index = 0; index < node_count; ++index) {
auto node_info = Mocker::GenNodeInfo();
auto node_id = NodeID::FromBinary(node_info->node_id());
EXPECT_TRUE(RegisterNode(*node_info));
EXPECT_TRUE(UnregisterNode(node_id));
node_ids.insert(node_id);
}
return node_ids;
}
// GCS server.
gcs::GcsServerConfig config_;
std::unique_ptr<gcs::GcsServer> gcs_server_;
@@ -1328,6 +1341,29 @@ TEST_F(ServiceBasedGcsClientTest, TestRandomEvictDestroyedActors) {
EXPECT_TRUE(WaitForCondition(condition, timeout_ms_.count()));
}
TEST_F(ServiceBasedGcsClientTest, TestEvictExpiredDeadNodes) {
// Simulate the scenario of node dead.
int node_count = RayConfig::instance().maximum_gcs_dead_node_cached_count();
RegisterNodeAndMarkDead(node_count);
// Restart GCS.
RestartGcsServer();
const auto &node_ids = RegisterNodeAndMarkDead(node_count);
// Get all nodes.
auto condition = [this]() {
return GetNodeInfoList().size() ==
RayConfig::instance().maximum_gcs_dead_node_cached_count();
};
EXPECT_TRUE(WaitForCondition(condition, timeout_ms_.count()));
auto nodes = GetNodeInfoList();
for (const auto &node : nodes) {
EXPECT_TRUE(node_ids.contains(NodeID::FromBinary(node.node_id())));
}
}
// TODO(sang): Add tests after adding asyncAdd
} // namespace ray
src/ray/gcs/gcs_server/gcs_node_manager.cc
+22 -5
View File
@@ -168,7 +168,8 @@ GcsNodeManager::GcsNodeManager(boost::asio::io_service &main_io_service,
main_io_service_.post([this, node_id] {
if (auto node = RemoveNode(node_id, /* is_intended = */ false)) {
node->set_state(rpc::GcsNodeInfo::DEAD);
RAY_CHECK(dead_nodes_.emplace(node_id, node).second);
node->set_timestamp(current_sys_time_ms());
AddDeadNodeToCache(node);
auto on_done = [this, node_id, node](const Status &status) {
auto on_done = [this, node_id, node](const Status &status) {
RAY_CHECK_OK(gcs_pub_sub_->Publish(
@@ -213,7 +214,8 @@ void GcsNodeManager::HandleUnregisterNode(const rpc::UnregisterNodeRequest &requ
RAY_LOG(INFO) << "Unregistering node info, node id = " << node_id;
if (auto node = RemoveNode(node_id, /* is_intended = */ true)) {
node->set_state(rpc::GcsNodeInfo::DEAD);
RAY_CHECK(dead_nodes_.emplace(node_id, node).second);
node->set_timestamp(current_sys_time_ms());
AddDeadNodeToCache(node);
auto on_done = [this, node_id, node, reply,
send_reply_callback](const Status &status) {
@@ -347,7 +349,7 @@ void GcsNodeManager::HandleDeleteResources(const rpc::DeleteResourcesRequest &re
void GcsNodeManager::HandleSetInternalConfig(const rpc::SetInternalConfigRequest &request,
rpc::SetInternalConfigReply *reply,
rpc::SendReplyCallback send_reply_callback) {
auto on_done = [reply, send_reply_callback, request](const Status status) {
auto on_done = [reply, send_reply_callback, request](const Status &status) {
RAY_LOG(DEBUG) << "Set internal config: " << request.config().DebugString();
GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
};
@@ -359,7 +361,7 @@ void GcsNodeManager::HandleGetInternalConfig(const rpc::GetInternalConfigRequest
rpc::GetInternalConfigReply *reply,
rpc::SendReplyCallback send_reply_callback) {
auto get_system_config = [reply, send_reply_callback](
ray::Status status,
const ray::Status &status,
const boost::optional<rpc::StoredConfig> &config) {
if (config.has_value()) {
reply->mutable_config()->CopyFrom(config.get());
@@ -377,7 +379,7 @@ void GcsNodeManager::HandleGetAllAvailableResources(
for (const auto &iter : GetClusterRealtimeResources()) {
rpc::AvailableResources resource;
resource.set_node_id(iter.first.Binary());
for (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);
@@ -464,8 +466,13 @@ void GcsNodeManager::LoadInitialData(const EmptyCallback &done) {
AddNode(std::make_shared<rpc::GcsNodeInfo>(item.second));
} else if (item.second.state() == rpc::GcsNodeInfo::DEAD) {
dead_nodes_.emplace(item.first, std::make_shared<rpc::GcsNodeInfo>(item.second));
sorted_dead_node_list_.emplace_back(item.first, item.second.timestamp());
}
}
sorted_dead_node_list_.sort([](const std::pair<NodeID, int64_t> &left,
const std::pair<NodeID, int64_t> &right) {
return left.second < right.second;
});
auto get_node_resource_callback =
[this, done](const std::unordered_map<NodeID, ResourceMap> &result) {
@@ -513,5 +520,15 @@ void GcsNodeManager::UpdatePlacementGroupLoad(
});
}
void GcsNodeManager::AddDeadNodeToCache(std::shared_ptr<rpc::GcsNodeInfo> node) {
if (dead_nodes_.size() >= RayConfig::instance().maximum_gcs_dead_node_cached_count()) {
dead_nodes_.erase(sorted_dead_node_list_.begin()->first);
sorted_dead_node_list_.erase(sorted_dead_node_list_.begin());
}
auto node_id = NodeID::FromBinary(node->node_id());
dead_nodes_.emplace(node_id, node);
sorted_dead_node_list_.emplace_back(node_id, node->timestamp());
}
} // namespace gcs
} // namespace ray
src/ray/gcs/gcs_server/gcs_node_manager.h
+9
View File
@@ -246,6 +246,12 @@ class GcsNodeManager : public rpc::NodeInfoHandler {
};
private:
/// Add the dead node to the cache. If the cache is full, the earliest dead node is
/// evicted.
///
/// \param node The node which is dead.
void AddDeadNodeToCache(std::shared_ptr<rpc::GcsNodeInfo> node);
/// The main event loop for node failure detector.
boost::asio::io_service &main_io_service_;
/// Detector to detect the failure of node.
@@ -256,6 +262,9 @@ class GcsNodeManager : public rpc::NodeInfoHandler {
absl::flat_hash_map<NodeID, std::shared_ptr<rpc::GcsNodeInfo>> alive_nodes_;
/// Dead nodes.
absl::flat_hash_map<NodeID, std::shared_ptr<rpc::GcsNodeInfo>> dead_nodes_;
/// The nodes are sorted according to the timestamp, and the oldest is at the head of
/// the list.
std::list<std::pair<NodeID, int64_t>> sorted_dead_node_list_;
/// Cluster resources.
absl::flat_hash_map<NodeID, rpc::ResourceMap> cluster_resources_;
/// Listeners which monitors the addition of nodes.
src/ray/protobuf/gcs.proto
+2
View File
@@ -259,6 +259,8 @@ message GcsNodeInfo {
// The port at which the node will expose metrics to.
int32 metrics_export_port = 9;
// Timestamp that the node is dead.
int64 timestamp = 10;
}
// Represents the demand for a particular resource shape.