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

[GCS]Optimizing actor info query interface (#11067)

Browse Source 
* add part code

* add part code

* fix review comment

* fix review comment

* fix review comment

* fix crash bug

* fix ut bug

* fix bug

* fix ut bug

Co-authored-by: 灵洵 <fengbin.ffb@antfin.com>
This commit is contained in:
fangfengbin
2020-10-01 02:34:42 +08:00
committed by GitHub
parent 474af36591
commit 138d6cced9
5 changed files with 70 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/gcs/gcs_client/test/global_state_accessor_test.cc
-21
View File
@@ -229,27 +229,6 @@ TEST_F(GlobalStateAccessorTest, TestObjectTable) {
}
}
TEST_F(GlobalStateAccessorTest, TestActorTable) {
int actor_count = 1;
ASSERT_EQ(global_state_->GetAllActorInfo().size(), 0);
auto job_id = JobID::FromInt(1);
std::vector<ActorID> actor_ids;
actor_ids.reserve(actor_count);
for (int index = 0; index < actor_count; ++index) {
auto actor_table_data = Mocker::GenActorTableData(job_id);
actor_ids.emplace_back(ActorID::FromBinary(actor_table_data->actor_id()));
std::promise<bool> promise;
RAY_CHECK_OK(gcs_client_->Actors().AsyncRegister(
actor_table_data, [&promise](Status status) { promise.set_value(status.ok()); }));
WaitReady(promise.get_future(), timeout_ms_);
}
ASSERT_EQ(global_state_->GetAllActorInfo().size(), actor_count);
for (auto &actor_id : actor_ids) {
ASSERT_TRUE(global_state_->GetActorInfo(actor_id));
}
}
TEST_F(GlobalStateAccessorTest, TestWorkerTable) {
ASSERT_EQ(global_state_->GetAllWorkerInfo().size(), 0);
// Add worker info
src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc
+28 -28
View File
@@ -157,10 +157,24 @@ class ServiceBasedGcsClientTest : public ::testing::Test {
return WaitReady(promise.get_future(), timeout_ms_);
}
bool RegisterActor(const std::shared_ptr<rpc::ActorTableData> &actor_table_data) {
bool RegisterActor(const std::shared_ptr<rpc::ActorTableData> &actor_table_data,
bool is_detached = true) {
std::promise<bool> promise;
RAY_CHECK_OK(gcs_client_->Actors().AsyncRegister(
actor_table_data, [&promise](Status status) { promise.set_value(status.ok()); }));
rpc::TaskSpec message;
auto actor_id = ActorID::FromBinary(actor_table_data->actor_id());
message.set_job_id(actor_id.JobId().Binary());
message.set_type(TaskType::ACTOR_CREATION_TASK);
message.set_task_id(TaskID::ForActorCreationTask(actor_id).Binary());
message.set_caller_id(actor_id.Binary());
message.set_max_retries(0);
message.set_num_returns(1);
message.set_parent_task_id(TaskID::ForActorCreationTask(actor_id).Binary());
message.mutable_actor_creation_task_spec()->set_actor_id(actor_id.Binary());
message.mutable_actor_creation_task_spec()->set_is_detached(is_detached);
TaskSpecification task_spec(message);
RAY_CHECK_OK(gcs_client_->Actors().AsyncRegisterActor(
task_spec, [&promise](Status status) { promise.set_value(status.ok()); }));
return WaitReady(promise.get_future(), timeout_ms_);
}
@@ -580,17 +594,11 @@ TEST_F(ServiceBasedGcsClientTest, TestActorInfo) {
// Register an actor to GCS.
ASSERT_TRUE(RegisterActor(actor_table_data));
ASSERT_TRUE(GetActor(actor_id).state() == rpc::ActorTableData::ALIVE);
ASSERT_TRUE(GetActor(actor_id).state() == rpc::ActorTableData::DEPENDENCIES_UNREADY);
// Cancel subscription to an actor.
UnsubscribeActor(actor_id);
WaitForActorUnsubscribed(actor_id);
// Update dynamic states of actor in GCS.
actor_table_data->set_state(rpc::ActorTableData::DEAD);
ASSERT_TRUE(UpdateActor(actor_id, actor_table_data));
ASSERT_TRUE(GetActor(actor_id).state() == rpc::ActorTableData::DEAD);
WaitForExpectedCount(actor_update_count, 1);
}
TEST_F(ServiceBasedGcsClientTest, TestActorCheckpoint) {
@@ -633,8 +641,8 @@ TEST_F(ServiceBasedGcsClientTest, TestActorSubscribeAll) {
ASSERT_TRUE(SubscribeAllActors(on_subscribe));
// Register an actor to GCS.
ASSERT_TRUE(RegisterActor(actor_table_data1));
ASSERT_TRUE(RegisterActor(actor_table_data2));
ASSERT_FALSE(RegisterActor(actor_table_data1, false));
ASSERT_FALSE(RegisterActor(actor_table_data2, false));
WaitForExpectedCount(actor_update_count, 2);
}
@@ -980,13 +988,13 @@ TEST_F(ServiceBasedGcsClientTest, TestActorTableResubscribe) {
// Subscribe to updates for this actor.
ASSERT_TRUE(SubscribeActor(actor_id, actor_subscribe));
ASSERT_TRUE(RegisterActor(actor_table_data));
ASSERT_FALSE(RegisterActor(actor_table_data, false));
// We should receive a new ALIVE notification from the subscribe channel.
// We should receive a new DEAD notification from the subscribe channel.
WaitForExpectedCount(num_subscribe_all_notifications, 1);
WaitForExpectedCount(num_subscribe_one_notifications, 1);
CheckActorData(subscribe_all_notifications[0], rpc::ActorTableData::ALIVE);
CheckActorData(subscribe_one_notifications[0], rpc::ActorTableData::ALIVE);
CheckActorData(subscribe_all_notifications[0], rpc::ActorTableData::DEAD);
CheckActorData(subscribe_one_notifications[0], rpc::ActorTableData::DEAD);
// Restart GCS server.
RestartGcsServer();
@@ -995,20 +1003,12 @@ TEST_F(ServiceBasedGcsClientTest, TestActorTableResubscribe) {
// didn't restart, it will fetch data again from the GCS server. The GCS will destroy
// the actor because it finds that the actor is out of scope, so we'll receive another
// notification of DEAD state.
WaitForExpectedCount(num_subscribe_all_notifications, 2);
WaitForExpectedCount(num_subscribe_one_notifications, 2);
CheckActorData(subscribe_all_notifications[1], rpc::ActorTableData::DEAD);
CheckActorData(subscribe_one_notifications[1], rpc::ActorTableData::DEAD);
// Update the actor state to ALIVE.
actor_table_data->set_state(rpc::ActorTableData::ALIVE);
ASSERT_TRUE(UpdateActor(actor_id, actor_table_data));
// We should receive a new ALIVE notification from the subscribe channel.
WaitForExpectedCount(num_subscribe_all_notifications, 3);
WaitForExpectedCount(num_subscribe_one_notifications, 3);
CheckActorData(subscribe_all_notifications[2], rpc::ActorTableData::ALIVE);
CheckActorData(subscribe_one_notifications[2], rpc::ActorTableData::ALIVE);
CheckActorData(subscribe_all_notifications[1], rpc::ActorTableData::DEAD);
CheckActorData(subscribe_one_notifications[1], rpc::ActorTableData::DEAD);
CheckActorData(subscribe_all_notifications[2], rpc::ActorTableData::DEAD);
CheckActorData(subscribe_one_notifications[2], rpc::ActorTableData::DEAD);
}
TEST_F(ServiceBasedGcsClientTest, TestObjectTableResubscribe) {
src/ray/gcs/gcs_server/gcs_actor_manager.cc
+39 -47
View File
@@ -146,22 +146,21 @@ void GcsActorManager::HandleGetActorInfo(const rpc::GetActorInfoRequest &request
RAY_LOG(DEBUG) << "Getting actor info"
<< ", job id = " << actor_id.JobId() << ", actor id = " << actor_id;
auto on_done = [actor_id, reply, send_reply_callback](
const Status &status,
const boost::optional<ActorTableData> &result) {
if (result) {
reply->mutable_actor_table_data()->CopyFrom(*result);
const auto &registered_actor_iter = registered_actors_.find(actor_id);
if (registered_actor_iter != registered_actors_.end()) {
reply->mutable_actor_table_data()->CopyFrom(
registered_actor_iter->second->GetActorTableData());
} else {
const auto &destroyed_actor_iter = destroyed_actors_.find(actor_id);
if (destroyed_actor_iter != destroyed_actors_.end()) {
reply->mutable_actor_table_data()->CopyFrom(
destroyed_actor_iter->second->GetActorTableData());
}
RAY_LOG(DEBUG) << "Finished getting actor info, job id = " << actor_id.JobId()
<< ", actor id = " << actor_id << ", status = " << status;
GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK());
};
// Look up the actor_id in the GCS.
Status status = gcs_table_storage_->ActorTable().Get(actor_id, on_done);
if (!status.ok()) {
on_done(status, boost::none);
}
RAY_LOG(DEBUG) << "Finished getting actor info, job id = " << actor_id.JobId()
<< ", actor id = " << actor_id;
GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK());
}
void GcsActorManager::HandleGetAllActorInfo(const rpc::GetAllActorInfoRequest &request,
@@ -169,19 +168,14 @@ void GcsActorManager::HandleGetAllActorInfo(const rpc::GetAllActorInfoRequest &r
rpc::SendReplyCallback send_reply_callback) {
RAY_LOG(DEBUG) << "Getting all actor info.";
auto on_done = [reply, send_reply_callback](
const std::unordered_map<ActorID, ActorTableData> &result) {
for (auto &it : result) {
reply->add_actor_table_data()->CopyFrom(it.second);
}
RAY_LOG(DEBUG) << "Finished getting all actor info.";
GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK());
};
Status status = gcs_table_storage_->ActorTable().GetAll(on_done);
if (!status.ok()) {
on_done(std::unordered_map<ActorID, ActorTableData>());
for (const auto &iter : registered_actors_) {
reply->add_actor_table_data()->CopyFrom(iter.second->GetActorTableData());
}
for (const auto &iter : destroyed_actors_) {
reply->add_actor_table_data()->CopyFrom(iter.second->GetActorTableData());
}
RAY_LOG(DEBUG) << "Finished getting all actor info.";
GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK());
}
void GcsActorManager::HandleGetNamedActorInfo(
@@ -190,37 +184,24 @@ void GcsActorManager::HandleGetNamedActorInfo(
const std::string &name = request.name();
RAY_LOG(DEBUG) << "Getting actor info, name = " << name;
auto on_done = [name, reply, send_reply_callback](
const Status &status,
const boost::optional<ActorTableData> &result) {
if (status.ok()) {
if (result) {
reply->mutable_actor_table_data()->CopyFrom(*result);
}
} else {
RAY_LOG(ERROR) << "Failed to get actor info: " << status.ToString()
<< ", name = " << name;
}
GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
};
// Try to look up the actor ID for the named actor.
ActorID actor_id = GetActorIDByName(name);
Status status = Status::OK();
if (actor_id.IsNil()) {
// The named actor was not found.
std::stringstream stream;
stream << "Actor with name '" << name << "' was not found.";
on_done(Status::NotFound(stream.str()), boost::none);
RAY_LOG(WARNING) << stream.str();
status = Status::NotFound(stream.str());
} else {
// Look up the actor_id in the GCS.
Status status = gcs_table_storage_->ActorTable().Get(actor_id, on_done);
if (!status.ok()) {
on_done(status, boost::none);
}
const auto &iter = registered_actors_.find(actor_id);
RAY_CHECK(iter != registered_actors_.end());
reply->mutable_actor_table_data()->CopyFrom(iter->second->GetActorTableData());
RAY_LOG(DEBUG) << "Finished getting actor info, job id = " << actor_id.JobId()
<< ", actor id = " << actor_id;
}
GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
}
void GcsActorManager::HandleRegisterActorInfo(
const rpc::RegisterActorInfoRequest &request, rpc::RegisterActorInfoReply *reply,
@@ -580,6 +561,7 @@ void GcsActorManager::DestroyActor(const ActorID &actor_id) {
auto it = registered_actors_.find(actor_id);
RAY_CHECK(it != registered_actors_.end())
<< "Tried to destroy actor that does not exist " << actor_id;
destroyed_actors_.emplace(it->first, it->second);
const auto actor = std::move(it->second);
registered_actors_.erase(it);
@@ -932,8 +914,8 @@ void GcsActorManager::LoadInitialData(const EmptyCallback &done) {
done](const std::unordered_map<ActorID, ActorTableData> &result) {
std::unordered_map<NodeID, std::vector<WorkerID>> node_to_workers;
for (auto &item : result) {
auto actor = std::make_shared<GcsActor>(item.second);
if (item.second.state() != ray::rpc::ActorTableData::DEAD) {
auto actor = std::make_shared<GcsActor>(item.second);
registered_actors_.emplace(item.first, actor);
if (!actor->GetName().empty()) {
@@ -970,6 +952,8 @@ void GcsActorManager::LoadInitialData(const EmptyCallback &done) {
RAY_CHECK(!actor->GetNodeID().IsNil());
node_to_workers[actor->GetNodeID()].emplace_back(actor->GetWorkerID());
}
} else {
destroyed_actors_.emplace(item.first, actor);
}
}
@@ -1012,6 +996,14 @@ void GcsActorManager::OnJobFinished(const JobID &job_id) {
RAY_CHECK_OK(
gcs_table_storage_->ActorTable().BatchDelete(non_detached_actors, nullptr));
for (auto iter = destroyed_actors_.begin(); iter != destroyed_actors_.end();) {
if (iter->first.JobId() == job_id && !iter->second->IsDetached()) {
destroyed_actors_.erase(iter++);
} else {
iter++;
}
}
// Get checkpoint id first from checkpoint id table and delete all checkpoints
// related to this job
RAY_CHECK_OK(gcs_table_storage_->ActorCheckpointIdTable().GetByJobId(
src/ray/gcs/gcs_server/gcs_actor_manager.h
+2
View File
@@ -366,6 +366,8 @@ class GcsActorManager : public rpc::ActorInfoHandler {
/// All registered actors (unresoved and pending actors are also included).
/// TODO(swang): Use unique_ptr instead of shared_ptr.
absl::flat_hash_map<ActorID, std::shared_ptr<GcsActor>> registered_actors_;
/// All destroyed actors.
absl::flat_hash_map<ActorID, std::shared_ptr<GcsActor>> destroyed_actors_;
/// Maps actor names to their actor ID for lookups by name.
absl::flat_hash_map<std::string, ActorID> named_actors_;
/// The actors which dependencies have not been resolved.
src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc
+1 -47
View File
@@ -457,23 +457,6 @@ TEST_F(GcsServerTest, TestActorInfo) {
JobID job_id = JobID::FromInt(1);
auto actor_table_data = Mocker::GenActorTableData(job_id);
// Register actor
rpc::RegisterActorInfoRequest register_actor_info_request;
register_actor_info_request.mutable_actor_table_data()->CopyFrom(*actor_table_data);
ASSERT_TRUE(RegisterActorInfo(register_actor_info_request));
boost::optional<rpc::ActorTableData> result =
GetActorInfo(actor_table_data->actor_id());
ASSERT_TRUE(result->state() == rpc::ActorTableData::ALIVE);
// Update actor state
rpc::UpdateActorInfoRequest update_actor_info_request;
actor_table_data->set_state(rpc::ActorTableData::DEAD);
update_actor_info_request.set_actor_id(actor_table_data->actor_id());
update_actor_info_request.mutable_actor_table_data()->CopyFrom(*actor_table_data);
ASSERT_TRUE(UpdateActorInfo(update_actor_info_request));
result = GetActorInfo(actor_table_data->actor_id());
ASSERT_TRUE(result->state() == rpc::ActorTableData::DEAD);
// Add actor checkpoint
ActorCheckpointID checkpoint_id = ActorCheckpointID::FromRandom();
rpc::ActorCheckpointData checkpoint;
@@ -520,16 +503,8 @@ TEST_F(GcsServerTest, TestJobGarbageCollection) {
add_job_request.mutable_data()->CopyFrom(*job_table_data);
ASSERT_TRUE(AddJob(add_job_request));
// Register actor for job
auto actor_table_data = Mocker::GenActorTableData(job_id);
rpc::RegisterActorInfoRequest register_actor_info_request;
register_actor_info_request.mutable_actor_table_data()->CopyFrom(*actor_table_data);
ASSERT_TRUE(RegisterActorInfo(register_actor_info_request));
boost::optional<rpc::ActorTableData> actor_result =
GetActorInfo(actor_table_data->actor_id());
ASSERT_TRUE(actor_result->state() == rpc::ActorTableData::ALIVE);
// Add actor checkpoint
auto actor_table_data = Mocker::GenActorTableData(job_id);
ActorCheckpointID checkpoint_id = ActorCheckpointID::FromRandom();
rpc::ActorCheckpointData checkpoint;
checkpoint.set_actor_id(actor_table_data->actor_id());
@@ -551,13 +526,6 @@ TEST_F(GcsServerTest, TestJobGarbageCollection) {
// Register detached actor for job
auto detached_actor_table_data = Mocker::GenActorTableData(job_id);
detached_actor_table_data->set_is_detached(true);
rpc::RegisterActorInfoRequest register_detached_actor_info_request;
register_detached_actor_info_request.mutable_actor_table_data()->CopyFrom(
*detached_actor_table_data);
ASSERT_TRUE(RegisterActorInfo(register_detached_actor_info_request));
boost::optional<rpc::ActorTableData> detached_actor_result =
GetActorInfo(detached_actor_table_data->actor_id());
ASSERT_TRUE(detached_actor_result->state() == rpc::ActorTableData::ALIVE);
// Add checkpoint for detached actor
ActorCheckpointID detached_checkpoint_id = ActorCheckpointID::FromRandom();
@@ -593,20 +561,6 @@ TEST_F(GcsServerTest, TestJobGarbageCollection) {
};
ASSERT_TRUE(WaitForCondition(condition_func, 10 * 1000));
condition_func = [this, &actor_table_data, &checkpoint_id]() -> bool {
return !GetActorCheckpoint(actor_table_data->actor_id(), checkpoint_id.Binary())
.has_value();
};
ASSERT_TRUE(WaitForCondition(condition_func, 10 * 1000));
condition_func = [this, &actor_table_data]() -> bool {
return !GetActorCheckpointID(actor_table_data->actor_id()).has_value();
};
ASSERT_TRUE(WaitForCondition(condition_func, 10 * 1000));
detached_actor_result = GetActorInfo(detached_actor_table_data->actor_id());
ASSERT_TRUE(detached_actor_result->state() == rpc::ActorTableData::ALIVE);
detached_checkpoint_result = GetActorCheckpoint(detached_actor_table_data->actor_id(),
detached_checkpoint_id.Binary());
ASSERT_TRUE(detached_checkpoint_result->actor_id() ==