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

Optimize node register&worker failure log (#9833)

Browse Source
This commit is contained in:
fangfengbin
2020-08-10 11:41:45 +08:00
committed by GitHub
parent a2bfdcbf24
commit 26b36a1982
3 changed files with 49 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/gcs/gcs_server/gcs_node_manager.cc
+4 -6
View File
@@ -172,12 +172,14 @@ void GcsNodeManager::HandleRegisterNode(const rpc::RegisterNodeRequest &request,
rpc::RegisterNodeReply *reply,
rpc::SendReplyCallback send_reply_callback) {
ClientID node_id = ClientID::FromBinary(request.node_info().node_id());
RAY_LOG(INFO) << "Registering node info, node id = " << node_id;
RAY_LOG(INFO) << "Registering node info, node id = " << node_id
<< ", address = " << request.node_info().node_manager_address();
AddNode(std::make_shared<rpc::GcsNodeInfo>(request.node_info()));
auto on_done = [this, node_id, request, reply,
send_reply_callback](const Status &status) {
RAY_CHECK_OK(status);
RAY_LOG(INFO) << "Finished registering node info, node id = " << node_id;
RAY_LOG(INFO) << "Finished registering node info, node id = " << node_id
<< ", address = " << request.node_info().node_manager_address();
RAY_CHECK_OK(gcs_pub_sub_->Publish(NODE_CHANNEL, node_id.Hex(),
request.node_info().SerializeAsString(), nullptr));
GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
@@ -214,7 +216,6 @@ void GcsNodeManager::HandleUnregisterNode(const rpc::UnregisterNodeRequest &requ
void GcsNodeManager::HandleGetAllNodeInfo(const rpc::GetAllNodeInfoRequest &request,
rpc::GetAllNodeInfoReply *reply,
rpc::SendReplyCallback send_reply_callback) {
RAY_LOG(DEBUG) << "Getting all nodes info.";
for (const auto &entry : alive_nodes_) {
reply->add_node_info_list()->CopyFrom(*entry.second);
}
@@ -222,7 +223,6 @@ void GcsNodeManager::HandleGetAllNodeInfo(const rpc::GetAllNodeInfoRequest &requ
reply->add_node_info_list()->CopyFrom(*entry.second);
}
GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK());
RAY_LOG(DEBUG) << "Finished getting all node info.";
}
void GcsNodeManager::HandleReportHeartbeat(const rpc::ReportHeartbeatRequest &request,
@@ -245,7 +245,6 @@ void GcsNodeManager::HandleGetResources(const rpc::GetResourcesRequest &request,
rpc::GetResourcesReply *reply,
rpc::SendReplyCallback send_reply_callback) {
ClientID node_id = ClientID::FromBinary(request.node_id());
RAY_LOG(DEBUG) << "Getting node resources, node id = " << node_id;
auto iter = cluster_resources_.find(node_id);
if (iter != cluster_resources_.end()) {
for (auto &resource : iter->second.items()) {
@@ -253,7 +252,6 @@ void GcsNodeManager::HandleGetResources(const rpc::GetResourcesRequest &request,
}
}
GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK());
RAY_LOG(DEBUG) << "Finished getting node resources, node id = " << node_id;
}
void GcsNodeManager::HandleUpdateResources(const rpc::UpdateResourcesRequest &request,
src/ray/gcs/gcs_server/gcs_worker_manager.cc
+42 -30
View File
@@ -21,44 +21,56 @@ void GcsWorkerManager::HandleReportWorkerFailure(
const rpc::ReportWorkerFailureRequest &request, rpc::ReportWorkerFailureReply *reply,
rpc::SendReplyCallback send_reply_callback) {
const rpc::Address worker_address = request.worker_failure().worker_address();
RAY_LOG(DEBUG) << "Reporting worker failure, " << worker_address.DebugString();
const auto worker_id = WorkerID::FromBinary(worker_address.worker_id());
const auto node_id = ClientID::FromBinary(worker_address.raylet_id());
std::stringstream log_stream;
log_stream << "Reporting worker failure, worker id = " << worker_id
<< ", node id = " << node_id
<< ", address = " << worker_address.ip_address();
if (request.worker_failure().intentional_disconnect()) {
RAY_LOG(INFO) << log_stream.str();
} else {
RAY_LOG(WARNING) << log_stream.str();
}
auto worker_failure_data = std::make_shared<WorkerTableData>();
worker_failure_data->CopyFrom(request.worker_failure());
worker_failure_data->set_is_alive(false);
const auto worker_id = WorkerID::FromBinary(worker_address.worker_id());
// Before handle ReportWorkerFailureRequest, you should check if the worker is exists.
auto on_get_done =
[this, worker_address, worker_id, worker_failure_data, reply, send_reply_callback](
const Status &status, const boost::optional<WorkerTableData> &result) {
if (result) {
auto on_put_done = [this, worker_address, worker_id, worker_failure_data, reply,
send_reply_callback](const Status &status) {
if (!status.ok()) {
RAY_LOG(ERROR) << "Failed to report worker failure, "
<< worker_address.DebugString();
} else {
RAY_CHECK_OK(gcs_pub_sub_->Publish(WORKER_CHANNEL, worker_id.Binary(),
worker_failure_data->SerializeAsString(),
nullptr));
}
GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
};
// The worker exists in worker table, you can update the info of this worker.
Status report_status = gcs_table_storage_->WorkerTable().Put(
worker_id, *worker_failure_data, on_put_done);
if (!report_status.ok()) {
on_put_done(report_status);
}
auto on_get_done = [this, worker_address, worker_id, node_id, worker_failure_data,
reply, send_reply_callback](
const Status &status,
const boost::optional<WorkerTableData> &result) {
if (result) {
auto on_put_done = [this, worker_address, worker_id, node_id, worker_failure_data,
reply, send_reply_callback](const Status &status) {
if (!status.ok()) {
RAY_LOG(ERROR) << "Failed to report worker failure, worker id = " << worker_id
<< ", node id = " << node_id
<< ", address = " << worker_address.ip_address();
} else {
// The worker doesn't exists in worker table.
RAY_LOG(WARNING) << "Failed to report worker failure, the worker doesn't "
"exist, "
<< worker_address.DebugString();
GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
RAY_CHECK_OK(gcs_pub_sub_->Publish(WORKER_CHANNEL, worker_id.Binary(),
worker_failure_data->SerializeAsString(),
nullptr));
}
GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
};
// The worker exists in worker table, you can update the info of this worker.
Status report_status = gcs_table_storage_->WorkerTable().Put(
worker_id, *worker_failure_data, on_put_done);
if (!report_status.ok()) {
on_put_done(report_status);
}
} else {
// The worker doesn't exists in worker table.
RAY_LOG(WARNING) << "Failed to report worker failure, the worker doesn't "
"exist, worker id = "
<< worker_id << ", node id = " << node_id
<< ", address = " << worker_address.ip_address();
GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
}
};
Status status = gcs_table_storage_->WorkerTable().Get(worker_id, on_get_done);
if (!status.ok()) {
on_get_done(status, boost::none);
src/ray/raylet/node_manager.cc
+3 -3
View File
@@ -1382,9 +1382,9 @@ void NodeManager::ProcessDisconnectClientMessage(
local_queues_.RemoveDriverTaskId(TaskID::ComputeDriverTaskId(driver_id));
worker_pool_.DisconnectDriver(worker);
RAY_LOG(DEBUG) << "Driver (pid=" << worker->GetProcess().GetId()
<< ") is disconnected. "
<< "job_id: " << worker->GetAssignedJobId();
RAY_LOG(INFO) << "Driver (pid=" << worker->GetProcess().GetId()
<< ") is disconnected. "
<< "job_id: " << worker->GetAssignedJobId();
}
client->Close();