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

ray memory should collect statistics from all nodes (#7721)

Browse Source
This commit is contained in:
Eric Liang
2020-03-25 16:31:31 -07:00
committed by GitHub
parent 46404d8a0b
commit 23b6fdcda1
3 changed files with 66 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/raylet/node_manager.cc
+35 -16
View File
@@ -3380,12 +3380,11 @@ void NodeManager::HandleGetNodeStats(const rpc::GetNodeStatsRequest &node_stats_
}
}
std::string FormatMemoryInfo(
std::vector<std::shared_ptr<rpc::GetNodeStatsReply>> node_stats) {
std::string FormatMemoryInfo(std::vector<rpc::GetNodeStatsReply> node_stats) {
// First pass to compute object sizes.
absl::flat_hash_map<ObjectID, int64_t> object_sizes;
for (const auto &reply : node_stats) {
for (const auto &worker_stats : reply->workers_stats()) {
for (const auto &worker_stats : reply.workers_stats()) {
for (const auto &object_ref : worker_stats.core_worker_stats().object_refs()) {
auto obj_id = ObjectID::FromBinary(object_ref.object_id());
if (object_ref.object_size() > 0) {
@@ -3408,7 +3407,7 @@ std::string FormatMemoryInfo(
// Second pass builds the summary string for each node.
for (const auto &reply : node_stats) {
for (const auto &worker_stats : reply->workers_stats()) {
for (const auto &worker_stats : reply.workers_stats()) {
bool pid_printed = false;
for (const auto &object_ref : worker_stats.core_worker_stats().object_refs()) {
if (!object_ref.pinned_in_memory() && object_ref.local_ref_count() == 0 &&
@@ -3459,23 +3458,43 @@ std::string FormatMemoryInfo(
void NodeManager::HandleFormatGlobalMemoryInfo(
const rpc::FormatGlobalMemoryInfoRequest &request,
rpc::FormatGlobalMemoryInfoReply *reply, rpc::SendReplyCallback send_reply_callback) {
std::vector<std::shared_ptr<rpc::GetNodeStatsReply>> replies;
auto replies = std::make_shared<std::vector<rpc::GetNodeStatsReply>>();
auto local_request = std::make_shared<rpc::GetNodeStatsRequest>();
auto local_reply = std::make_shared<rpc::GetNodeStatsReply>();
local_request->set_include_memory_info(true);
// TODO(ekl): for (const auto &entry : remote_node_manager_clients_) {}
// to handle remote nodes
unsigned int num_nodes = remote_node_manager_clients_.size() + 1;
rpc::GetNodeStatsRequest stats_req;
stats_req.set_include_memory_info(true);
HandleGetNodeStats(*local_request, local_reply.get(),
[local_request, local_reply, replies, reply, send_reply_callback](
Status status, std::function<void()> success,
std::function<void()> failure) mutable {
replies.push_back(local_reply);
reply->set_memory_summary(FormatMemoryInfo(replies));
send_reply_callback(Status::OK(), nullptr, nullptr);
});
auto store_reply = [replies, reply, num_nodes,
send_reply_callback](const rpc::GetNodeStatsReply &local_reply) {
replies->push_back(local_reply);
if (replies->size() >= num_nodes) {
reply->set_memory_summary(FormatMemoryInfo(*replies));
send_reply_callback(Status::OK(), nullptr, nullptr);
}
};
// Fetch from remote nodes.
for (const auto &entry : remote_node_manager_clients_) {
entry.second->GetNodeStats(
stats_req, [replies, store_reply](const ray::Status &status,
const rpc::GetNodeStatsReply &r) {
if (!status.ok()) {
RAY_LOG(ERROR) << "Failed to get remote node stats: " << status.ToString();
}
store_reply(r);
});
}
// Fetch from the local node.
HandleGetNodeStats(
stats_req, local_reply.get(),
[local_reply, store_reply](Status status, std::function<void()> success,
std::function<void()> failure) mutable {
store_reply(*local_reply);
});
}
void NodeManager::HandleGlobalGC(const rpc::GlobalGCRequest &request,