diff --git a/src/ray/gcs/gcs_client/test/global_state_accessor_test.cc b/src/ray/gcs/gcs_client/test/global_state_accessor_test.cc index f14e2675e..2f7dc171a 100644 --- a/src/ray/gcs/gcs_client/test/global_state_accessor_test.cc +++ b/src/ray/gcs/gcs_client/test/global_state_accessor_test.cc @@ -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 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 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 diff --git a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc index 5206cd048..79b390a99 100644 --- a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc +++ b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc @@ -157,10 +157,24 @@ class ServiceBasedGcsClientTest : public ::testing::Test { return WaitReady(promise.get_future(), timeout_ms_); } - bool RegisterActor(const std::shared_ptr &actor_table_data) { + bool RegisterActor(const std::shared_ptr &actor_table_data, + bool is_detached = true) { std::promise 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) { diff --git a/src/ray/gcs/gcs_server/gcs_actor_manager.cc b/src/ray/gcs/gcs_server/gcs_actor_manager.cc index 5ef74d0c7..d40d5bb80 100644 --- a/src/ray/gcs/gcs_server/gcs_actor_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_actor_manager.cc @@ -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 &result) { - if (result) { - reply->mutable_actor_table_data()->CopyFrom(*result); + const auto ®istered_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 &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()); + 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 &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 &result) { std::unordered_map> node_to_workers; for (auto &item : result) { + auto actor = std::make_shared(item.second); if (item.second.state() != ray::rpc::ActorTableData::DEAD) { - auto actor = std::make_shared(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( diff --git a/src/ray/gcs/gcs_server/gcs_actor_manager.h b/src/ray/gcs/gcs_server/gcs_actor_manager.h index 00a2bb7ec..35174ae0b 100644 --- a/src/ray/gcs/gcs_server/gcs_actor_manager.h +++ b/src/ray/gcs/gcs_server/gcs_actor_manager.h @@ -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> registered_actors_; + /// All destroyed actors. + absl::flat_hash_map> destroyed_actors_; /// Maps actor names to their actor ID for lookups by name. absl::flat_hash_map named_actors_; /// The actors which dependencies have not been resolved. diff --git a/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc b/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc index 1faffd166..4ca1472ea 100644 --- a/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc +++ b/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc @@ -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 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 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 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() ==