From 9f0f54266064e203b0bdcc9d3fa947cb4518ebc0 Mon Sep 17 00:00:00 2001 From: Tao Wang Date: Wed, 17 Jun 2020 09:43:08 +0800 Subject: [PATCH] Remove actor table info from storage when a driver exits (#8761) * delete contents of table related to specified job when the job is dead * check status * implement GetByJobId in gcs table storage * add test case * add test case * fix test cases * expose MGET and make match_pattern only related with SCAN * add test case for table storage * delete checkpoint * make MGetValues static * add most test case * add object test case * avoid accessing to storage when get matched object ids per job id * rename job info handler * use listener to sense job finished * clear actor state * add comments, remove actions in task handler * let raylet do object cleaning. only remove non-detached actors * only remove informations of non-detached actor * remove unused methods --- src/ray/gcs/gcs_server/gcs_actor_manager.cc | 47 +++++ src/ray/gcs/gcs_server/gcs_actor_manager.h | 6 + src/ray/gcs/gcs_server/gcs_server.cc | 31 +-- src/ray/gcs/gcs_server/gcs_server.h | 2 +- .../gcs/gcs_server/job_info_handler_impl.cc | 31 ++- .../gcs/gcs_server/job_info_handler_impl.h | 15 +- .../gcs_server/test/gcs_server_rpc_test.cc | 186 +++++++++++++++--- .../store_client/in_memory_store_client.cc | 6 +- .../gcs/store_client/redis_store_client.cc | 8 +- src/ray/rpc/gcs_server/gcs_rpc_server.h | 3 + 10 files changed, 279 insertions(+), 56 deletions(-) diff --git a/src/ray/gcs/gcs_server/gcs_actor_manager.cc b/src/ray/gcs/gcs_server/gcs_actor_manager.cc index 06096eee8..d42d5015a 100644 --- a/src/ray/gcs/gcs_server/gcs_actor_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_actor_manager.cc @@ -764,5 +764,52 @@ void GcsActorManager::LoadInitialData(const EmptyCallback &done) { RAY_CHECK_OK(gcs_table_storage_->ActorTable().GetAll(callback)); } +void GcsActorManager::OnJobFinished(const JobID &job_id) { + auto on_done = [this, + job_id](const std::unordered_map &result) { + if (!result.empty()) { + std::vector non_detached_actors; + std::unordered_set non_detached_actors_set; + for (auto &item : result) { + if (!item.second.is_detached()) { + non_detached_actors.push_back(item.first); + non_detached_actors_set.insert(item.first); + } + } + RAY_CHECK_OK( + gcs_table_storage_->ActorTable().BatchDelete(non_detached_actors, nullptr)); + + // Get checkpoint id first from checkpoint id table and delete all checkpoints + // related to this job + RAY_CHECK_OK(gcs_table_storage_->ActorCheckpointIdTable().GetByJobId( + job_id, [this, non_detached_actors_set]( + const std::unordered_map &result) { + if (!result.empty()) { + std::vector checkpoints; + std::vector checkpoint_ids; + for (auto &item : result) { + if (non_detached_actors_set.find(item.first) != + non_detached_actors_set.end()) { + checkpoints.push_back(item.first); + for (auto &id : item.second.checkpoint_ids()) { + checkpoint_ids.push_back(ActorCheckpointID::FromBinary(id)); + } + } + } + + RAY_CHECK_OK(gcs_table_storage_->ActorCheckpointIdTable().BatchDelete( + checkpoints, nullptr)); + RAY_CHECK_OK(gcs_table_storage_->ActorCheckpointTable().BatchDelete( + checkpoint_ids, nullptr)); + } + })); + } + }; + + // Only non-detached actors should be deleted. We get all actors of this job and to the + // filtering. + RAY_CHECK_OK(gcs_table_storage_->ActorTable().GetByJobId(job_id, on_done)); +} + } // namespace gcs } // namespace ray diff --git a/src/ray/gcs/gcs_server/gcs_actor_manager.h b/src/ray/gcs/gcs_server/gcs_actor_manager.h index 1698d26a0..58722a3c9 100644 --- a/src/ray/gcs/gcs_server/gcs_actor_manager.h +++ b/src/ray/gcs/gcs_server/gcs_actor_manager.h @@ -222,6 +222,12 @@ class GcsActorManager : public rpc::ActorInfoHandler { /// \param done Callback that will be called when load is complete. void LoadInitialData(const EmptyCallback &done); + /// Delete non-detached actor information from durable storage once the associated job + /// finishes. + /// + /// \param job_id The id of finished job. + void OnJobFinished(const JobID &job_id); + private: /// A data structure representing an actor's owner. struct Owner { diff --git a/src/ray/gcs/gcs_server/gcs_server.cc b/src/ray/gcs/gcs_server/gcs_server.cc index 4a22d4efa..39c219244 100644 --- a/src/ray/gcs/gcs_server/gcs_server.cc +++ b/src/ray/gcs/gcs_server/gcs_server.cc @@ -61,7 +61,17 @@ void GcsServer::Start() { InitGcsActorManager(); // Register rpc service. - job_info_handler_ = InitJobInfoHandler(); + gcs_object_manager_ = InitObjectManager(); + object_info_service_.reset( + new rpc::ObjectInfoGrpcService(main_service_, *gcs_object_manager_)); + rpc_server_.RegisterService(*object_info_service_); + + task_info_handler_ = InitTaskInfoHandler(); + task_info_service_.reset( + new rpc::TaskInfoGrpcService(main_service_, *task_info_handler_)); + rpc_server_.RegisterService(*task_info_service_); + + InitJobInfoHandler(); job_info_service_.reset(new rpc::JobInfoGrpcService(main_service_, *job_info_handler_)); rpc_server_.RegisterService(*job_info_service_); @@ -73,16 +83,6 @@ void GcsServer::Start() { new rpc::NodeInfoGrpcService(main_service_, *gcs_node_manager_)); rpc_server_.RegisterService(*node_info_service_); - gcs_object_manager_ = InitObjectManager(); - object_info_service_.reset( - new rpc::ObjectInfoGrpcService(main_service_, *gcs_object_manager_)); - rpc_server_.RegisterService(*object_info_service_); - - task_info_handler_ = InitTaskInfoHandler(); - task_info_service_.reset( - new rpc::TaskInfoGrpcService(main_service_, *task_info_handler_)); - rpc_server_.RegisterService(*task_info_service_); - stats_handler_ = InitStatsHandler(); stats_service_.reset(new rpc::StatsGrpcService(main_service_, *stats_handler_)); rpc_server_.RegisterService(*stats_service_); @@ -198,9 +198,12 @@ void GcsServer::InitGcsActorManager() { RAY_CHECK_OK(gcs_pub_sub_->SubscribeAll(WORKER_FAILURE_CHANNEL, on_subscribe, nullptr)); } -std::unique_ptr GcsServer::InitJobInfoHandler() { - return std::unique_ptr( - new rpc::DefaultJobInfoHandler(gcs_table_storage_, gcs_pub_sub_)); +void GcsServer::InitJobInfoHandler() { + job_info_handler_ = std::unique_ptr( + new rpc::GcsJobInfoHandler(gcs_table_storage_, gcs_pub_sub_)); + job_info_handler_->AddJobFinishedListener([this](std::shared_ptr job_id) { + gcs_actor_manager_->OnJobFinished(*job_id); + }); } std::unique_ptr GcsServer::InitObjectManager() { diff --git a/src/ray/gcs/gcs_server/gcs_server.h b/src/ray/gcs/gcs_server/gcs_server.h index 49f8dfcda..e4227f8ef 100644 --- a/src/ray/gcs/gcs_server/gcs_server.h +++ b/src/ray/gcs/gcs_server/gcs_server.h @@ -81,7 +81,7 @@ class GcsServer { virtual void InitGcsActorManager(); /// The job info handler - virtual std::unique_ptr InitJobInfoHandler(); + virtual void InitJobInfoHandler(); /// The object manager virtual std::unique_ptr InitObjectManager(); diff --git a/src/ray/gcs/gcs_server/job_info_handler_impl.cc b/src/ray/gcs/gcs_server/job_info_handler_impl.cc index e20d1d068..71db7c4d9 100644 --- a/src/ray/gcs/gcs_server/job_info_handler_impl.cc +++ b/src/ray/gcs/gcs_server/job_info_handler_impl.cc @@ -18,9 +18,9 @@ namespace ray { namespace rpc { -void DefaultJobInfoHandler::HandleAddJob(const rpc::AddJobRequest &request, - rpc::AddJobReply *reply, - rpc::SendReplyCallback send_reply_callback) { +void GcsJobInfoHandler::HandleAddJob(const rpc::AddJobRequest &request, + rpc::AddJobReply *reply, + rpc::SendReplyCallback send_reply_callback) { JobID job_id = JobID::FromBinary(request.data().job_id()); RAY_LOG(INFO) << "Adding job, job id = " << job_id << ", driver pid = " << request.data().driver_pid(); @@ -41,7 +41,7 @@ void DefaultJobInfoHandler::HandleAddJob(const rpc::AddJobRequest &request, } } -void DefaultJobInfoHandler::HandleMarkJobFinished( +void GcsJobInfoHandler::HandleMarkJobFinished( const rpc::MarkJobFinishedRequest &request, rpc::MarkJobFinishedReply *reply, rpc::SendReplyCallback send_reply_callback) { JobID job_id = JobID::FromBinary(request.job_id()); @@ -55,6 +55,7 @@ void DefaultJobInfoHandler::HandleMarkJobFinished( } else { RAY_CHECK_OK(gcs_pub_sub_->Publish(JOB_CHANNEL, job_id.Binary(), job_table_data->SerializeAsString(), nullptr)); + ClearJobInfos(job_id); RAY_LOG(INFO) << "Finished marking job state, job id = " << job_id; } GCS_RPC_SEND_REPLY(send_reply_callback, reply, status); @@ -66,9 +67,25 @@ void DefaultJobInfoHandler::HandleMarkJobFinished( } } -void DefaultJobInfoHandler::HandleGetAllJobInfo( - const rpc::GetAllJobInfoRequest &request, rpc::GetAllJobInfoReply *reply, - rpc::SendReplyCallback send_reply_callback) { +void GcsJobInfoHandler::ClearJobInfos(const JobID &job_id) { + // Notify all listeners. + for (auto &listener : job_finished_listeners_) { + listener(std::make_shared(job_id)); + } +} + +/// Add listener to monitor the add action of nodes. +/// +/// \param listener The handler which process the add of nodes. +void GcsJobInfoHandler::AddJobFinishedListener( + std::function)> listener) { + RAY_CHECK(listener); + job_finished_listeners_.emplace_back(std::move(listener)); +} + +void GcsJobInfoHandler::HandleGetAllJobInfo(const rpc::GetAllJobInfoRequest &request, + rpc::GetAllJobInfoReply *reply, + rpc::SendReplyCallback send_reply_callback) { RAY_LOG(INFO) << "Getting all job info."; auto on_done = [reply, send_reply_callback]( const std::unordered_map &result) { diff --git a/src/ray/gcs/gcs_server/job_info_handler_impl.h b/src/ray/gcs/gcs_server/job_info_handler_impl.h index 19d935c1d..0aee366c5 100644 --- a/src/ray/gcs/gcs_server/job_info_handler_impl.h +++ b/src/ray/gcs/gcs_server/job_info_handler_impl.h @@ -14,6 +14,7 @@ #pragma once +#include "gcs_object_manager.h" #include "gcs_table_storage.h" #include "ray/gcs/pubsub/gcs_pub_sub.h" #include "ray/gcs/redis_gcs_client.h" @@ -23,10 +24,10 @@ namespace ray { namespace rpc { /// This implementation class of `JobInfoHandler`. -class DefaultJobInfoHandler : public rpc::JobInfoHandler { +class GcsJobInfoHandler : public rpc::JobInfoHandler { public: - explicit DefaultJobInfoHandler(std::shared_ptr gcs_table_storage, - std::shared_ptr gcs_pub_sub) + explicit GcsJobInfoHandler(std::shared_ptr gcs_table_storage, + std::shared_ptr gcs_pub_sub) : gcs_table_storage_(std::move(gcs_table_storage)), gcs_pub_sub_(std::move(gcs_pub_sub)) {} @@ -40,9 +41,17 @@ class DefaultJobInfoHandler : public rpc::JobInfoHandler { void HandleGetAllJobInfo(const GetAllJobInfoRequest &request, GetAllJobInfoReply *reply, SendReplyCallback send_reply_callback) override; + void AddJobFinishedListener( + std::function)> listener) override; + private: std::shared_ptr gcs_table_storage_; std::shared_ptr gcs_pub_sub_; + + /// Listeners which monitors the finish of jobs. + std::vector)>> job_finished_listeners_; + + void ClearJobInfos(const JobID &job_id); }; } // namespace rpc 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 ae2bc05ed..2faf95b96 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 @@ -102,20 +102,24 @@ class GcsServerTest : public ::testing::Test { return WaitReady(promise.get_future(), timeout_ms_); } - rpc::ActorTableData GetActorInfo(const std::string &actor_id) { + boost::optional GetActorInfo(const std::string &actor_id) { rpc::GetActorInfoRequest request; request.set_actor_id(actor_id); - rpc::ActorTableData actor_table_data; + boost::optional actor_table_data_opt; std::promise promise; client_->GetActorInfo( - request, [&actor_table_data, &promise](const Status &status, - const rpc::GetActorInfoReply &reply) { + request, [&actor_table_data_opt, &promise](const Status &status, + const rpc::GetActorInfoReply &reply) { RAY_CHECK_OK(status); - actor_table_data.CopyFrom(reply.actor_table_data()); + if (reply.has_actor_table_data()) { + actor_table_data_opt = reply.actor_table_data(); + } else { + actor_table_data_opt = boost::none; + } promise.set_value(true); }); EXPECT_TRUE(WaitReady(promise.get_future(), timeout_ms_)); - return actor_table_data; + return actor_table_data_opt; } bool AddActorCheckpoint(const rpc::AddActorCheckpointRequest &request) { @@ -129,38 +133,47 @@ class GcsServerTest : public ::testing::Test { return WaitReady(promise.get_future(), timeout_ms_); } - rpc::ActorCheckpointData GetActorCheckpoint(const std::string &actor_id, - const std::string &checkpoint_id) { + boost::optional GetActorCheckpoint( + const std::string &actor_id, const std::string &checkpoint_id) { rpc::GetActorCheckpointRequest request; request.set_actor_id(actor_id); request.set_checkpoint_id(checkpoint_id); - rpc::ActorCheckpointData checkpoint_data; + boost::optional checkpoint_data_opt; std::promise promise; client_->GetActorCheckpoint( - request, [&checkpoint_data, &promise](const Status &status, - const rpc::GetActorCheckpointReply &reply) { + request, [&checkpoint_data_opt, &promise]( + const Status &status, const rpc::GetActorCheckpointReply &reply) { RAY_CHECK_OK(status); - checkpoint_data.CopyFrom(reply.checkpoint_data()); + if (reply.has_checkpoint_data()) { + checkpoint_data_opt = reply.checkpoint_data(); + } else { + checkpoint_data_opt = boost::none; + } promise.set_value(true); }); EXPECT_TRUE(WaitReady(promise.get_future(), timeout_ms_)); - return checkpoint_data; + return checkpoint_data_opt; } - rpc::ActorCheckpointIdData GetActorCheckpointID(const std::string &actor_id) { + boost::optional GetActorCheckpointID( + const std::string &actor_id) { rpc::GetActorCheckpointIDRequest request; request.set_actor_id(actor_id); - rpc::ActorCheckpointIdData checkpoint_id_data; + boost::optional checkpoint_id_data_opt; std::promise promise; client_->GetActorCheckpointID( - request, [&checkpoint_id_data, &promise]( + request, [&checkpoint_id_data_opt, &promise]( const Status &status, const rpc::GetActorCheckpointIDReply &reply) { RAY_CHECK_OK(status); - checkpoint_id_data.CopyFrom(reply.checkpoint_id_data()); + if (reply.has_checkpoint_id_data()) { + checkpoint_id_data_opt = reply.checkpoint_id_data(); + } else { + checkpoint_id_data_opt = boost::none; + } promise.set_value(true); }); EXPECT_TRUE(WaitReady(promise.get_future(), timeout_ms_)); - return checkpoint_id_data; + return checkpoint_id_data_opt; } bool RegisterNode(const rpc::RegisterNodeRequest &request) { @@ -310,7 +323,9 @@ class GcsServerTest : public ::testing::Test { client_->GetTask(request, [&task_data, &promise](const Status &status, const rpc::GetTaskReply &reply) { if (status.ok()) { - task_data.CopyFrom(reply.task_data()); + if (reply.has_task_data()) { + task_data.CopyFrom(reply.task_data()); + } } promise.set_value(true); }); @@ -409,8 +424,9 @@ TEST_F(GcsServerTest, TestActorInfo) { 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)); - rpc::ActorTableData result = GetActorInfo(actor_table_data->actor_id()); - ASSERT_TRUE(result.state() == + boost::optional result = + GetActorInfo(actor_table_data->actor_id()); + ASSERT_TRUE(result->state() == rpc::ActorTableData_ActorState::ActorTableData_ActorState_ALIVE); // Update actor state @@ -421,7 +437,7 @@ TEST_F(GcsServerTest, TestActorInfo) { 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() == + ASSERT_TRUE(result->state() == rpc::ActorTableData_ActorState::ActorTableData_ActorState_DEAD); // Add actor checkpoint @@ -434,14 +450,14 @@ TEST_F(GcsServerTest, TestActorInfo) { rpc::AddActorCheckpointRequest add_actor_checkpoint_request; add_actor_checkpoint_request.mutable_checkpoint_data()->CopyFrom(checkpoint); ASSERT_TRUE(AddActorCheckpoint(add_actor_checkpoint_request)); - rpc::ActorCheckpointData checkpoint_result = + boost::optional checkpoint_result = GetActorCheckpoint(actor_table_data->actor_id(), checkpoint_id.Binary()); - ASSERT_TRUE(checkpoint_result.actor_id() == actor_table_data->actor_id()); - ASSERT_TRUE(checkpoint_result.checkpoint_id() == checkpoint_id.Binary()); - rpc::ActorCheckpointIdData checkpoint_id_result = + ASSERT_TRUE(checkpoint_result->actor_id() == actor_table_data->actor_id()); + ASSERT_TRUE(checkpoint_result->checkpoint_id() == checkpoint_id.Binary()); + boost::optional checkpoint_id_result = GetActorCheckpointID(actor_table_data->actor_id()); - ASSERT_TRUE(checkpoint_id_result.actor_id() == actor_table_data->actor_id()); - ASSERT_TRUE(checkpoint_id_result.checkpoint_ids_size() == 1); + ASSERT_TRUE(checkpoint_id_result->actor_id() == actor_table_data->actor_id()); + ASSERT_TRUE(checkpoint_id_result->checkpoint_ids_size() == 1); } TEST_F(GcsServerTest, TestJobInfo) { @@ -460,6 +476,120 @@ TEST_F(GcsServerTest, TestJobInfo) { ASSERT_TRUE(MarkJobFinished(mark_job_finished_request)); } +TEST_F(GcsServerTest, TestJobGarbageCollection) { + // Create job_table_data + JobID job_id = JobID::FromInt(1); + auto job_table_data = Mocker::GenJobTableData(job_id); + + // Add job + rpc::AddJobRequest add_job_request; + 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_ActorState::ActorTableData_ActorState_ALIVE); + + // Add actor checkpoint + ActorCheckpointID checkpoint_id = ActorCheckpointID::FromRandom(); + rpc::ActorCheckpointData checkpoint; + checkpoint.set_actor_id(actor_table_data->actor_id()); + checkpoint.set_checkpoint_id(checkpoint_id.Binary()); + checkpoint.set_execution_dependency(checkpoint_id.Binary()); + + rpc::AddActorCheckpointRequest add_actor_checkpoint_request; + add_actor_checkpoint_request.mutable_checkpoint_data()->CopyFrom(checkpoint); + ASSERT_TRUE(AddActorCheckpoint(add_actor_checkpoint_request)); + boost::optional checkpoint_result = + GetActorCheckpoint(actor_table_data->actor_id(), checkpoint_id.Binary()); + ASSERT_TRUE(checkpoint_result->actor_id() == actor_table_data->actor_id()); + ASSERT_TRUE(checkpoint_result->checkpoint_id() == checkpoint_id.Binary()); + boost::optional checkpoint_id_result = + GetActorCheckpointID(actor_table_data->actor_id()); + ASSERT_TRUE(checkpoint_id_result->actor_id() == actor_table_data->actor_id()); + ASSERT_TRUE(checkpoint_id_result->checkpoint_ids_size() == 1); + + // 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_ActorState::ActorTableData_ActorState_ALIVE); + + // Add checkpoint for detached actor + ActorCheckpointID detached_checkpoint_id = ActorCheckpointID::FromRandom(); + rpc::ActorCheckpointData detached_checkpoint; + detached_checkpoint.set_actor_id(detached_actor_table_data->actor_id()); + detached_checkpoint.set_checkpoint_id(detached_checkpoint_id.Binary()); + detached_checkpoint.set_execution_dependency(detached_checkpoint_id.Binary()); + + rpc::AddActorCheckpointRequest add_detached_actor_checkpoint_request; + add_detached_actor_checkpoint_request.mutable_checkpoint_data()->CopyFrom( + detached_checkpoint); + ASSERT_TRUE(AddActorCheckpoint(add_detached_actor_checkpoint_request)); + boost::optional detached_checkpoint_result = + GetActorCheckpoint(detached_actor_table_data->actor_id(), + detached_checkpoint_id.Binary()); + ASSERT_TRUE(detached_checkpoint_result->actor_id() == + detached_actor_table_data->actor_id()); + ASSERT_TRUE(detached_checkpoint_result->checkpoint_id() == + detached_checkpoint_id.Binary()); + boost::optional detached_checkpoint_id_result = + GetActorCheckpointID(detached_actor_table_data->actor_id()); + ASSERT_TRUE(detached_checkpoint_id_result->actor_id() == + detached_actor_table_data->actor_id()); + ASSERT_TRUE(detached_checkpoint_id_result->checkpoint_ids_size() == 1); + + // Mark job finished + rpc::MarkJobFinishedRequest mark_job_finished_request; + mark_job_finished_request.set_job_id(job_table_data->job_id()); + ASSERT_TRUE(MarkJobFinished(mark_job_finished_request)); + + std::function condition_func = [this, &actor_table_data]() -> bool { + return !GetActorInfo(actor_table_data->actor_id()).has_value(); + }; + 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_ActorState::ActorTableData_ActorState_ALIVE); + + detached_checkpoint_result = GetActorCheckpoint(detached_actor_table_data->actor_id(), + detached_checkpoint_id.Binary()); + ASSERT_TRUE(detached_checkpoint_result->actor_id() == + detached_actor_table_data->actor_id()); + ASSERT_TRUE(detached_checkpoint_result->checkpoint_id() == + detached_checkpoint_id.Binary()); + + detached_checkpoint_id_result = + GetActorCheckpointID(detached_actor_table_data->actor_id()); + ASSERT_TRUE(detached_checkpoint_id_result->actor_id() == + detached_actor_table_data->actor_id()); + ASSERT_TRUE(detached_checkpoint_id_result->checkpoint_ids_size() == 1); +} + TEST_F(GcsServerTest, TestNodeInfo) { // Create gcs node info auto gcs_node_info = Mocker::GenNodeInfo(); diff --git a/src/ray/gcs/store_client/in_memory_store_client.cc b/src/ray/gcs/store_client/in_memory_store_client.cc index 86bb619bb..494b472a6 100644 --- a/src/ray/gcs/store_client/in_memory_store_client.cc +++ b/src/ray/gcs/store_client/in_memory_store_client.cc @@ -121,7 +121,11 @@ Status InMemoryStoreClient::AsyncDeleteByIndex(const std::string &table_name, } table->index_keys_.erase(iter); } - main_io_service_.post([callback]() { callback(Status::OK()); }); + main_io_service_.post([callback]() { + if (callback) { + callback(Status::OK()); + } + }); return Status::OK(); } diff --git a/src/ray/gcs/store_client/redis_store_client.cc b/src/ray/gcs/store_client/redis_store_client.cc index 2836e7b7b..ce3c18c62 100644 --- a/src/ray/gcs/store_client/redis_store_client.cc +++ b/src/ray/gcs/store_client/redis_store_client.cc @@ -169,7 +169,9 @@ Status RedisStoreClient::AsyncDeleteByIndex(const std::string &table_name, }; RAY_CHECK_OK(AsyncBatchDelete(table_name, keys, batch_delete_callback)); } else { - callback(status); + if (callback) { + callback(status); + } } }; @@ -203,7 +205,9 @@ Status RedisStoreClient::DeleteByKeys(const std::vector &keys, callback](const std::shared_ptr &reply) { ++(*finished_count); if (*finished_count == size) { - callback(Status::OK()); + if (callback) { + callback(Status::OK()); + } } }; RAY_CHECK_OK(item.first->RunArgvAsync(item.second, delete_callback)); diff --git a/src/ray/rpc/gcs_server/gcs_rpc_server.h b/src/ray/rpc/gcs_server/gcs_rpc_server.h index 5f572ca02..1807f1981 100644 --- a/src/ray/rpc/gcs_server/gcs_rpc_server.h +++ b/src/ray/rpc/gcs_server/gcs_rpc_server.h @@ -63,6 +63,9 @@ class JobInfoGcsServiceHandler { virtual void HandleGetAllJobInfo(const GetAllJobInfoRequest &request, GetAllJobInfoReply *reply, SendReplyCallback send_reply_callback) = 0; + + virtual void AddJobFinishedListener( + std::function)> listener) = 0; }; /// The `GrpcService` for `JobInfoGcsService`.