diff --git a/src/ray/core_worker/actor_manager.cc b/src/ray/core_worker/actor_manager.cc index c50619c46..ca41aabc1 100644 --- a/src/ray/core_worker/actor_manager.cc +++ b/src/ray/core_worker/actor_manager.cc @@ -1,4 +1,5 @@ #include "ray/core_worker/actor_manager.h" +#include "ray/gcs/pb_util.h" #include "ray/gcs/redis_accessor.h" namespace ray { @@ -6,7 +7,7 @@ namespace ray { void ActorManager::PublishTerminatedActor(const TaskSpecification &actor_creation_task) { auto actor_id = actor_creation_task.ActorCreationId(); auto data = gcs::CreateActorTableData(actor_creation_task, rpc::Address(), - gcs::ActorTableData::DEAD, 0); + rpc::ActorTableData::DEAD, 0); auto update_callback = [actor_id](Status status) { if (!status.ok()) { diff --git a/src/ray/gcs/pb_util.h b/src/ray/gcs/pb_util.h index ddc5179fe..9fe979f79 100644 --- a/src/ray/gcs/pb_util.h +++ b/src/ray/gcs/pb_util.h @@ -3,6 +3,7 @@ #include #include "ray/common/id.h" +#include "ray/common/task/task_spec.h" #include "ray/protobuf/gcs.pb.h" namespace ray { @@ -29,6 +30,32 @@ inline std::shared_ptr CreateJobTableData( return job_info_ptr; } +/// Helper function to produce actor table data. +inline std::shared_ptr CreateActorTableData( + const TaskSpecification &task_spec, const ray::rpc::Address &address, + ray::rpc::ActorTableData::ActorState state, uint64_t remaining_reconstructions) { + RAY_CHECK(task_spec.IsActorCreationTask()); + auto actor_id = task_spec.ActorCreationId(); + auto actor_info_ptr = std::make_shared(); + // Set all of the static fields for the actor. These fields will not change + // even if the actor fails or is reconstructed. + actor_info_ptr->set_actor_id(actor_id.Binary()); + actor_info_ptr->set_parent_id(task_spec.CallerId().Binary()); + actor_info_ptr->set_actor_creation_dummy_object_id( + task_spec.ActorDummyObject().Binary()); + actor_info_ptr->set_job_id(task_spec.JobId().Binary()); + actor_info_ptr->set_max_reconstructions(task_spec.MaxActorReconstructions()); + actor_info_ptr->set_is_detached(task_spec.IsDetachedActor()); + // Set the fields that change when the actor is restarted. + actor_info_ptr->set_remaining_reconstructions(remaining_reconstructions); + actor_info_ptr->set_is_direct_call(task_spec.IsDirectCall()); + actor_info_ptr->mutable_address()->CopyFrom(address); + actor_info_ptr->mutable_owner_address()->CopyFrom( + task_spec.GetMessage().caller_address()); + actor_info_ptr->set_state(state); + return actor_info_ptr; +} + } // namespace gcs } // namespace ray diff --git a/src/ray/gcs/redis_accessor.cc b/src/ray/gcs/redis_accessor.cc index ebd3d1ebc..97dd00cbf 100644 --- a/src/ray/gcs/redis_accessor.cc +++ b/src/ray/gcs/redis_accessor.cc @@ -8,31 +8,6 @@ namespace ray { namespace gcs { -std::shared_ptr CreateActorTableData( - const TaskSpecification &task_spec, const rpc::Address &address, - gcs::ActorTableData::ActorState state, uint64_t remaining_reconstructions) { - RAY_CHECK(task_spec.IsActorCreationTask()); - auto actor_id = task_spec.ActorCreationId(); - auto actor_info_ptr = std::make_shared(); - // Set all of the static fields for the actor. These fields will not change - // even if the actor fails or is reconstructed. - actor_info_ptr->set_actor_id(actor_id.Binary()); - actor_info_ptr->set_parent_id(task_spec.CallerId().Binary()); - actor_info_ptr->set_actor_creation_dummy_object_id( - task_spec.ActorDummyObject().Binary()); - actor_info_ptr->set_job_id(task_spec.JobId().Binary()); - actor_info_ptr->set_max_reconstructions(task_spec.MaxActorReconstructions()); - actor_info_ptr->set_is_detached(task_spec.IsDetachedActor()); - // Set the fields that change when the actor is restarted. - actor_info_ptr->set_remaining_reconstructions(remaining_reconstructions); - actor_info_ptr->set_is_direct_call(task_spec.IsDirectCall()); - actor_info_ptr->mutable_address()->CopyFrom(address); - actor_info_ptr->mutable_owner_address()->CopyFrom( - task_spec.GetMessage().caller_address()); - actor_info_ptr->set_state(state); - return actor_info_ptr; -} - RedisActorInfoAccessor::RedisActorInfoAccessor(RedisGcsClient *client_impl) : client_impl_(client_impl), actor_sub_executor_(client_impl_->actor_table()) {} diff --git a/src/ray/gcs/redis_accessor.h b/src/ray/gcs/redis_accessor.h index f7a6529c2..ce28f4710 100644 --- a/src/ray/gcs/redis_accessor.h +++ b/src/ray/gcs/redis_accessor.h @@ -2,7 +2,6 @@ #define RAY_GCS_REDIS_ACCESSOR_H #include "ray/common/id.h" -#include "ray/common/task/task_spec.h" #include "ray/gcs/accessor.h" #include "ray/gcs/callback.h" #include "ray/gcs/subscription_executor.h" @@ -14,10 +13,6 @@ namespace gcs { class RedisGcsClient; -std::shared_ptr CreateActorTableData( - const TaskSpecification &task_spec, const rpc::Address &address, - gcs::ActorTableData::ActorState state, uint64_t remaining_reconstructions); - /// \class RedisActorInfoAccessor /// `RedisActorInfoAccessor` is an implementation of `ActorInfoAccessor` /// that uses Redis as the backend storage.