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

add actor table data creation method to pb_util.h (#6746)

Browse Source
This commit is contained in:
micafan
2020-01-08 22:39:17 +08:00
committed by Zhijun Fu
parent 70c7d47c09
commit 0a5d0109a4
4 changed files with 29 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/core_worker/actor_manager.cc
+2 -1
View File
@@ -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()) {
src/ray/gcs/pb_util.h
+27
View File
@@ -3,6 +3,7 @@
#include <memory>
#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<ray::rpc::JobTableData> CreateJobTableData(
return job_info_ptr;
}
/// Helper function to produce actor table data.
inline std::shared_ptr<ray::rpc::ActorTableData> 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<ray::rpc::ActorTableData>();
// 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
src/ray/gcs/redis_accessor.cc
-25
View File
@@ -8,31 +8,6 @@ namespace ray {
namespace gcs {
std::shared_ptr<gcs::ActorTableData> 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<ray::gcs::ActorTableData>();
// 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()) {}
src/ray/gcs/redis_accessor.h
-5
View File
@@ -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<gcs::ActorTableData> 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.