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

[GCS]Reduce get operations of worker table (#11599)

Browse Source 
* [GCS]Reduce get operations of worker table

* fix ut bug

* fix ut bug

* fix review comment

Co-authored-by: 灵洵 <fengbin.ffb@antfin.com>
This commit is contained in:
fangfengbin
2020-11-11 10:11:25 +08:00
committed by GitHub
parent 8afd2acdc1
commit 433e4f32da
3 changed files with 23 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc
+2 -2
View File
@@ -971,14 +971,14 @@ TEST_F(ServiceBasedGcsClientTest, TestWorkerInfo) {
auto worker_data = Mocker::GenWorkerTableData();
worker_data->mutable_worker_address()->set_worker_id(WorkerID::FromRandom().Binary());
ASSERT_TRUE(ReportWorkerFailure(worker_data));
WaitForExpectedCount(worker_failure_count, 0);
WaitForExpectedCount(worker_failure_count, 1);
// Add a worker to GCS.
ASSERT_TRUE(AddWorker(worker_data));
// Report a worker failure to GCS when this worker is actually exist.
ASSERT_TRUE(ReportWorkerFailure(worker_data));
WaitForExpectedCount(worker_failure_count, 1);
WaitForExpectedCount(worker_failure_count, 2);
}
TEST_F(ServiceBasedGcsClientTest, TestErrorInfo) {
src/ray/gcs/gcs_server/gcs_worker_manager.cc
+19 -35
View File
@@ -40,45 +40,29 @@ void GcsWorkerManager::HandleReportWorkerFailure(
worker_failure_data->CopyFrom(request.worker_failure());
worker_failure_data->set_is_alive(false);
// Before handle ReportWorkerFailureRequest, you should check if the worker is exists.
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 {
stats::UnintentionalWorkerFailures.Record(1);
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_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 id = "
<< worker_id << ", node id = " << node_id
<< ", address = " << worker_address.ip_address();
GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
stats::UnintentionalWorkerFailures.Record(1);
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);
};
Status status = gcs_table_storage_->WorkerTable().Get(worker_id, on_get_done);
// As soon as the worker starts, it will register with GCS. It ensures that GCS receives
// the worker registration information first and then the worker failure message, so we
// delete the get operation. Related issues:
// https://github.com/ray-project/ray/pull/11599
Status status =
gcs_table_storage_->WorkerTable().Put(worker_id, *worker_failure_data, on_done);
if (!status.ok()) {
on_get_done(status, boost::none);
on_done(status);
}
}
src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc
+2 -2
View File
@@ -708,7 +708,7 @@ TEST_F(GcsServerTest, TestWorkerInfo) {
report_worker_failure_request.mutable_worker_failure()->CopyFrom(*worker_failure_data);
ASSERT_TRUE(ReportWorkerFailure(report_worker_failure_request));
std::vector<rpc::WorkerTableData> worker_table_data = GetAllWorkerInfo();
ASSERT_TRUE(worker_table_data.size() == 0);
ASSERT_TRUE(worker_table_data.size() == 1);
// Add worker info
auto worker_data = Mocker::GenWorkerTableData();
@@ -716,7 +716,7 @@ TEST_F(GcsServerTest, TestWorkerInfo) {
rpc::AddWorkerInfoRequest add_worker_request;
add_worker_request.mutable_worker_data()->CopyFrom(*worker_data);
ASSERT_TRUE(AddWorkerInfo(add_worker_request));
ASSERT_TRUE(GetAllWorkerInfo().size() == 1);
ASSERT_TRUE(GetAllWorkerInfo().size() == 2);
// Get worker info
boost::optional<rpc::WorkerTableData> result =