diff --git a/BUILD.bazel b/BUILD.bazel index 710c1b9c3..054c57bb4 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -47,6 +47,16 @@ cc_proto_library( deps = ["object_manager_proto"], ) +proto_library( + name = "core_worker_proto", + srcs = ["src/ray/protobuf/core_worker.proto"], +) + +cc_proto_library( + name = "core_worker_cc_proto", + deps = ["core_worker_proto"], +) + # === End of protobuf definitions === # === Begin of rpc definitions === @@ -239,6 +249,7 @@ cc_library( ]), copts = COPTS, deps = [ + ":core_worker_cc_proto", ":ray_common", ":ray_util", ":raylet_lib", diff --git a/src/ray/common/id.cc b/src/ray/common/id.cc index 0735d4d00..00964094a 100644 --- a/src/ray/common/id.cc +++ b/src/ray/common/id.cc @@ -139,6 +139,22 @@ const JobID ComputeJobId(const WorkerID &driver_id) { return JobID(driver_id); } +const ActorHandleID ComputeNextActorHandleId(const ActorHandleID &actor_handle_id, + int64_t num_forks) { + // Compute hashes. + SHA256_CTX ctx; + sha256_init(&ctx); + sha256_update(&ctx, reinterpret_cast(actor_handle_id.Data()), + actor_handle_id.Size()); + sha256_update(&ctx, reinterpret_cast(&num_forks), sizeof(num_forks)); + + // Compute the final actor handle ID from the hash. + BYTE buff[DIGEST_SIZE]; + sha256_final(&ctx, buff); + RAY_CHECK(DIGEST_SIZE >= ActorHandleID::Size()); + return ActorHandleID::FromBinary(std::string(buff, buff + ActorHandleID::Size())); +} + #define ID_OSTREAM_OPERATOR(id_type) \ std::ostream &operator<<(std::ostream &os, const id_type &id) { \ if (id.IsNil()) { \ diff --git a/src/ray/common/id.h b/src/ray/common/id.h index 667e55769..539a0774c 100644 --- a/src/ray/common/id.h +++ b/src/ray/common/id.h @@ -159,6 +159,14 @@ std::ostream &operator<<(std::ostream &os, const ObjectID &id); const TaskID GenerateTaskId(const JobID &job_id, const TaskID &parent_task_id, int parent_task_counter); +/// Compute the next actor handle ID of a new actor handle during a fork operation. +/// +/// \param actor_handle_id The actor handle ID of original actor. +/// \param num_forks The count of forks of original actor. +/// \return The next actor handle ID generated from the given info. +const ActorHandleID ComputeNextActorHandleId(const ActorHandleID &actor_handle_id, + int64_t num_forks); + template BaseID::BaseID() { // Using const_cast to directly change data is dangerous. The cached diff --git a/src/ray/core_worker/common.h b/src/ray/core_worker/common.h index 9dbdfdc77..750712760 100644 --- a/src/ray/core_worker/common.h +++ b/src/ray/core_worker/common.h @@ -5,6 +5,7 @@ #include "ray/common/buffer.h" #include "ray/common/id.h" +#include "ray/protobuf/gcs.pb.h" #include "ray/raylet/raylet_client.h" #include "ray/raylet/task_spec.h" @@ -13,13 +14,10 @@ namespace ray { /// Type of this worker. enum class WorkerType { WORKER, DRIVER }; -/// Language of Ray tasks and workers. -enum class WorkerLanguage { PYTHON, JAVA }; - /// Information about a remote function. struct RayFunction { /// Language of the remote function. - const WorkerLanguage language; + const ray::rpc::Language language; /// Function descriptor of the remote function. const std::vector function_descriptor; }; @@ -109,6 +107,48 @@ enum class StoreProviderType { PLASMA }; enum class TaskTransportType { RAYLET }; +/// Translate from ray::rpc::Language to Language type (required by raylet client). +/// +/// \param[in] language Language for a task. +/// \return Translated task language. +inline ::Language ToRayletTaskLanguage(ray::rpc::Language language) { + switch (language) { + case ray::rpc::Language::JAVA: + return ::Language::JAVA; + break; + case ray::rpc::Language::PYTHON: + return ::Language::PYTHON; + break; + case ray::rpc::Language::CPP: + return ::Language::CPP; + break; + default: + RAY_LOG(FATAL) << "Invalid language specified: " << static_cast(language); + break; + } +} + +/// Translate from Language to ray::rpc::Language type (required by core worker). +/// +/// \param[in] language Language for a task. +/// \return Translated task language. +inline ray::rpc::Language ToRpcTaskLanguage(::Language language) { + switch (language) { + case Language::JAVA: + return ray::rpc::Language::JAVA; + break; + case Language::PYTHON: + return ray::rpc::Language::PYTHON; + break; + case Language::CPP: + return ray::rpc::Language::CPP; + break; + default: + RAY_LOG(FATAL) << "Invalid language specified: " << static_cast(language); + break; + } +} + } // namespace ray #endif // RAY_CORE_WORKER_COMMON_H diff --git a/src/ray/core_worker/core_worker.cc b/src/ray/core_worker/core_worker.cc index c8ef189c8..03b20329c 100644 --- a/src/ray/core_worker/core_worker.cc +++ b/src/ray/core_worker/core_worker.cc @@ -4,9 +4,8 @@ namespace ray { CoreWorker::CoreWorker(const enum WorkerType worker_type, - const enum WorkerLanguage language, - const std::string &store_socket, const std::string &raylet_socket, - const JobID &job_id) + const ray::rpc::Language language, const std::string &store_socket, + const std::string &raylet_socket, const JobID &job_id) : worker_type_(worker_type), language_(language), store_socket_(store_socket), @@ -15,7 +14,7 @@ CoreWorker::CoreWorker(const enum WorkerType worker_type, raylet_client_(raylet_socket_, ClientID::FromBinary(worker_context_.GetWorkerID().Binary()), (worker_type_ == ray::WorkerType::WORKER), - worker_context_.GetCurrentJobID(), ToTaskLanguage(language_)), + worker_context_.GetCurrentJobID(), ToRayletTaskLanguage(language_)), task_interface_(*this), object_interface_(*this), task_execution_interface_(*this) { @@ -31,18 +30,4 @@ CoreWorker::CoreWorker(const enum WorkerType worker_type, } } -::Language CoreWorker::ToTaskLanguage(WorkerLanguage language) { - switch (language) { - case ray::WorkerLanguage::JAVA: - return ::Language::JAVA; - break; - case ray::WorkerLanguage::PYTHON: - return ::Language::PYTHON; - break; - default: - RAY_LOG(FATAL) << "invalid language specified: " << static_cast(language); - break; - } -} - } // namespace ray diff --git a/src/ray/core_worker/core_worker.h b/src/ray/core_worker/core_worker.h index 3afc28ee5..7abaf46b9 100644 --- a/src/ray/core_worker/core_worker.h +++ b/src/ray/core_worker/core_worker.h @@ -22,7 +22,7 @@ class CoreWorker { /// \param[in] langauge Language of this worker. /// /// NOTE(zhijunfu): the constructor would throw if a failure happens. - CoreWorker(const WorkerType worker_type, const WorkerLanguage language, + CoreWorker(const WorkerType worker_type, const ray::rpc::Language language, const std::string &store_socket, const std::string &raylet_socket, const JobID &job_id = JobID::Nil()); @@ -30,7 +30,7 @@ class CoreWorker { enum WorkerType WorkerType() const { return worker_type_; } /// Language of this worker. - enum WorkerLanguage Language() const { return language_; } + ray::rpc::Language Language() const { return language_; } /// Return the `CoreWorkerTaskInterface` that contains the methods related to task /// submisson. @@ -45,17 +45,11 @@ class CoreWorker { CoreWorkerTaskExecutionInterface &Execution() { return task_execution_interface_; } private: - /// Translate from WorkLanguage to Language type (required by raylet client). - /// - /// \param[in] language Language for a task. - /// \return Translated task language. - ::Language ToTaskLanguage(WorkerLanguage language); - /// Type of this worker. const enum WorkerType worker_type_; /// Language of this worker. - const enum WorkerLanguage language_; + const ray::rpc::Language language_; /// Plasma store socket name. const std::string store_socket_; diff --git a/src/ray/core_worker/core_worker_test.cc b/src/ray/core_worker/core_worker_test.cc index 773e3f26d..46bc10dc6 100644 --- a/src/ray/core_worker/core_worker_test.cc +++ b/src/ray/core_worker/core_worker_test.cc @@ -124,7 +124,7 @@ class CoreWorkerTest : public ::testing::Test { void TearDown() {} void TestNormalTask(const std::unordered_map &resources) { - CoreWorker driver(WorkerType::DRIVER, WorkerLanguage::PYTHON, + CoreWorker driver(WorkerType::DRIVER, ray::rpc::Language::PYTHON, raylet_store_socket_names_[0], raylet_socket_names_[0], JobID::FromRandom()); @@ -134,7 +134,7 @@ class CoreWorkerTest : public ::testing::Test { auto buffer1 = std::make_shared(array1, sizeof(array1)); - RayFunction func{ray::WorkerLanguage::PYTHON, {}}; + RayFunction func{ray::rpc::Language::PYTHON, {}}; std::vector args; args.emplace_back(TaskArg::PassByValue(buffer1)); @@ -165,7 +165,7 @@ class CoreWorkerTest : public ::testing::Test { std::vector args; args.emplace_back(TaskArg::PassByReference(object_id)); - RayFunction func{ray::WorkerLanguage::PYTHON, {}}; + RayFunction func{ray::rpc::Language::PYTHON, {}}; TaskOptions options; std::vector return_ids; @@ -184,7 +184,7 @@ class CoreWorkerTest : public ::testing::Test { } void TestActorTask(const std::unordered_map &resources) { - CoreWorker driver(WorkerType::DRIVER, WorkerLanguage::PYTHON, + CoreWorker driver(WorkerType::DRIVER, ray::rpc::Language::PYTHON, raylet_store_socket_names_[0], raylet_socket_names_[0], JobID::FromRandom()); @@ -195,7 +195,7 @@ class CoreWorkerTest : public ::testing::Test { uint8_t array[] = {1, 2, 3}; auto buffer = std::make_shared(array, sizeof(array)); - RayFunction func{ray::WorkerLanguage::PYTHON, {}}; + RayFunction func{ray::rpc::Language::PYTHON, {}}; std::vector args; args.emplace_back(TaskArg::PassByValue(buffer)); @@ -223,7 +223,7 @@ class CoreWorkerTest : public ::testing::Test { TaskOptions options{1, resources}; std::vector return_ids; - RayFunction func{ray::WorkerLanguage::PYTHON, {}}; + RayFunction func{ray::rpc::Language::PYTHON, {}}; RAY_CHECK_OK(driver.Tasks().SubmitActorTask(*actor_handle, func, args, options, &return_ids)); RAY_CHECK(return_ids.size() == 1); @@ -302,8 +302,41 @@ TEST_F(ZeroNodeTest, TestWorkerContext) { ASSERT_EQ(context.GetNextPutIndex(), 3); } +TEST_F(ZeroNodeTest, TestActorHandle) { + ActorHandle handle1(ActorID::FromRandom(), ActorHandleID::FromRandom(), + ray::rpc::Language::JAVA, + {"org.ray.exampleClass", "exampleMethod", "exampleSignature"}); + + auto forkedHandle1 = handle1.Fork(); + ASSERT_EQ(1, handle1.NumForks()); + ASSERT_EQ(handle1.ActorID(), forkedHandle1.ActorID()); + ASSERT_NE(handle1.ActorHandleID(), forkedHandle1.ActorHandleID()); + ASSERT_EQ(handle1.ActorLanguage(), forkedHandle1.ActorLanguage()); + ASSERT_EQ(handle1.ActorCreationTaskFunctionDescriptor(), + forkedHandle1.ActorCreationTaskFunctionDescriptor()); + ASSERT_EQ(handle1.ActorCursor(), forkedHandle1.ActorCursor()); + ASSERT_EQ(0, forkedHandle1.TaskCounter()); + ASSERT_EQ(0, forkedHandle1.NumForks()); + auto forkedHandle2 = handle1.Fork(); + ASSERT_EQ(2, handle1.NumForks()); + ASSERT_EQ(0, forkedHandle2.TaskCounter()); + ASSERT_EQ(0, forkedHandle2.NumForks()); + + std::string buffer; + handle1.Serialize(&buffer); + auto handle2 = ActorHandle::Deserialize(buffer); + ASSERT_EQ(handle1.ActorID(), handle2.ActorID()); + ASSERT_EQ(handle1.ActorHandleID(), handle2.ActorHandleID()); + ASSERT_EQ(handle1.ActorLanguage(), handle2.ActorLanguage()); + ASSERT_EQ(handle1.ActorCreationTaskFunctionDescriptor(), + handle2.ActorCreationTaskFunctionDescriptor()); + ASSERT_EQ(handle1.ActorCursor(), handle2.ActorCursor()); + ASSERT_EQ(handle1.TaskCounter(), handle2.TaskCounter()); + ASSERT_EQ(handle1.NumForks(), handle2.NumForks()); +} + TEST_F(SingleNodeTest, TestObjectInterface) { - CoreWorker core_worker(WorkerType::DRIVER, WorkerLanguage::PYTHON, + CoreWorker core_worker(WorkerType::DRIVER, ray::rpc::Language::PYTHON, raylet_store_socket_names_[0], raylet_socket_names_[0], JobID::FromRandom()); @@ -367,11 +400,11 @@ TEST_F(SingleNodeTest, TestObjectInterface) { } TEST_F(TwoNodeTest, TestObjectInterfaceCrossNodes) { - CoreWorker worker1(WorkerType::DRIVER, WorkerLanguage::PYTHON, + CoreWorker worker1(WorkerType::DRIVER, ray::rpc::Language::PYTHON, raylet_store_socket_names_[0], raylet_socket_names_[0], JobID::FromRandom()); - CoreWorker worker2(WorkerType::DRIVER, WorkerLanguage::PYTHON, + CoreWorker worker2(WorkerType::DRIVER, ray::rpc::Language::PYTHON, raylet_store_socket_names_[1], raylet_socket_names_[1], JobID::FromRandom()); @@ -458,7 +491,7 @@ TEST_F(TwoNodeTest, TestActorTaskCrossNodes) { TEST_F(SingleNodeTest, TestCoreWorkerConstructorFailure) { try { - CoreWorker core_worker(WorkerType::DRIVER, WorkerLanguage::PYTHON, "", + CoreWorker core_worker(WorkerType::DRIVER, ray::rpc::Language::PYTHON, "", raylet_socket_names_[0], JobID::FromRandom()); } catch (const std::exception &e) { std::cout << "Caught exception when constructing core worker: " << e.what(); diff --git a/src/ray/core_worker/mock_worker.cc b/src/ray/core_worker/mock_worker.cc index e08cb4bcc..62917cb94 100644 --- a/src/ray/core_worker/mock_worker.cc +++ b/src/ray/core_worker/mock_worker.cc @@ -17,8 +17,8 @@ namespace ray { class MockWorker { public: MockWorker(const std::string &store_socket, const std::string &raylet_socket) - : worker_(WorkerType::WORKER, WorkerLanguage::PYTHON, store_socket, raylet_socket, - JobID::FromRandom()) {} + : worker_(WorkerType::WORKER, ray::rpc::Language::PYTHON, store_socket, + raylet_socket, JobID::FromRandom()) {} void Run() { auto executor_func = [this](const RayFunction &ray_function, diff --git a/src/ray/core_worker/task_execution.cc b/src/ray/core_worker/task_execution.cc index 435be36ff..86093cefa 100644 --- a/src/ray/core_worker/task_execution.cc +++ b/src/ray/core_worker/task_execution.cc @@ -29,9 +29,7 @@ Status CoreWorkerTaskExecutionInterface::Run(const TaskExecutor &executor) { const auto &spec = task.GetTaskSpecification(); core_worker_.worker_context_.SetCurrentTask(spec); - WorkerLanguage language = (spec.GetLanguage() == ::Language::JAVA) - ? WorkerLanguage::JAVA - : WorkerLanguage::PYTHON; + ray::rpc::Language language = ToRpcTaskLanguage(spec.GetLanguage()); RayFunction func{language, spec.FunctionDescriptor()}; std::vector> args; diff --git a/src/ray/core_worker/task_interface.cc b/src/ray/core_worker/task_interface.cc index bd84e5e0b..8f1e05b0d 100644 --- a/src/ray/core_worker/task_interface.cc +++ b/src/ray/core_worker/task_interface.cc @@ -6,6 +6,92 @@ namespace ray { +ActorHandle::ActorHandle( + const class ActorID &actor_id, const class ActorHandleID &actor_handle_id, + const ray::rpc::Language actor_language, + const std::vector &actor_creation_task_function_descriptor) { + inner_.set_actor_id(actor_id.Data(), actor_id.Size()); + inner_.set_actor_handle_id(actor_handle_id.Data(), actor_handle_id.Size()); + inner_.set_actor_language(static_cast(actor_language)); + *inner_.mutable_actor_creation_task_function_descriptor() = { + actor_creation_task_function_descriptor.begin(), + actor_creation_task_function_descriptor.end()}; + inner_.set_actor_cursor(actor_id.Data(), actor_id.Size()); +} + +ActorHandle::ActorHandle(const ActorHandle &other) + : inner_(other.inner_), new_actor_handles_(other.new_actor_handles_) {} + +ray::ActorID ActorHandle::ActorID() const { + return ActorID::FromBinary(inner_.actor_id()); +}; + +ray::ActorHandleID ActorHandle::ActorHandleID() const { + return ActorHandleID::FromBinary(inner_.actor_handle_id()); +}; + +ray::rpc::Language ActorHandle::ActorLanguage() const { + return (ray::rpc::Language)inner_.actor_language(); +}; + +std::vector ActorHandle::ActorCreationTaskFunctionDescriptor() const { + return ray::rpc::VectorFromProtobuf(inner_.actor_creation_task_function_descriptor()); +}; + +ObjectID ActorHandle::ActorCursor() const { + return ObjectID::FromBinary(inner_.actor_cursor()); +}; + +int64_t ActorHandle::TaskCounter() const { return inner_.task_counter(); }; + +int64_t ActorHandle::NumForks() const { return inner_.num_forks(); }; + +ActorHandle ActorHandle::Fork() { + ActorHandle new_handle; + std::unique_lock guard(mutex_); + new_handle.inner_ = inner_; + inner_.set_num_forks(inner_.num_forks() + 1); + const auto next_actor_handle_id = ComputeNextActorHandleId( + ActorHandleID::FromBinary(inner_.actor_handle_id()), inner_.num_forks()); + new_handle.inner_.set_actor_handle_id(next_actor_handle_id.Data(), + next_actor_handle_id.Size()); + new_actor_handles_.push_back(next_actor_handle_id); + guard.unlock(); + + new_handle.inner_.set_task_counter(0); + new_handle.inner_.set_num_forks(0); + return new_handle; +} + +void ActorHandle::Serialize(std::string *output) { + std::unique_lock guard(mutex_); + inner_.SerializeToString(output); +} + +ActorHandle ActorHandle::Deserialize(const std::string &data) { + ActorHandle ret; + ret.inner_.ParseFromString(data); + return ret; +} + +ActorHandle::ActorHandle() {} + +void ActorHandle::SetActorCursor(const ObjectID &actor_cursor) { + inner_.set_actor_cursor(actor_cursor.Binary()); +}; + +int64_t ActorHandle::IncreaseTaskCounter() { + int64_t old = inner_.task_counter(); + inner_.set_task_counter(old + 1); + return old; +} + +std::vector ActorHandle::NewActorHandles() const { + return new_actor_handles_; +} + +void ActorHandle::ClearNewActorHandles() { new_actor_handles_.clear(); } + CoreWorkerTaskInterface::CoreWorkerTaskInterface(CoreWorker &core_worker) : core_worker_(core_worker) { task_submitters_.emplace( @@ -30,7 +116,7 @@ Status CoreWorkerTaskInterface::SubmitTask(const RayFunction &function, } auto task_arguments = BuildTaskArguments(args); - auto language = core_worker_.ToTaskLanguage(function.language); + auto language = ToRayletTaskLanguage(function.language); ray::raylet::TaskSpecification spec(context.GetCurrentJobID(), context.GetCurrentTaskID(), next_task_index, @@ -54,14 +140,14 @@ Status CoreWorkerTaskInterface::CreateActor( std::vector return_ids; return_ids.push_back(ObjectID::ForTaskReturn(task_id, 1)); ActorID actor_creation_id = ActorID::FromBinary(return_ids[0].Binary()); - *actor_handle = std::unique_ptr( - new ActorHandle(actor_creation_id, ActorHandleID::Nil())); + new ActorHandle(actor_creation_id, ActorHandleID::Nil(), function.language, + function.function_descriptor)); (*actor_handle)->IncreaseTaskCounter(); (*actor_handle)->SetActorCursor(return_ids[0]); auto task_arguments = BuildTaskArguments(args); - auto language = core_worker_.ToTaskLanguage(function.language); + auto language = ToRayletTaskLanguage(function.language); // Note that the caller is supposed to specify required placement resources // correctly via actor_creation_options.resources. @@ -98,15 +184,16 @@ Status CoreWorkerTaskInterface::SubmitActorTask(ActorHandle &actor_handle, ObjectID::FromBinary(actor_handle.ActorID().Binary()); auto task_arguments = BuildTaskArguments(args); - auto language = core_worker_.ToTaskLanguage(function.language); + auto language = ToRayletTaskLanguage(function.language); + + std::unique_lock guard(actor_handle.mutex_); - std::vector new_actor_handles; ray::raylet::TaskSpecification spec( context.GetCurrentJobID(), context.GetCurrentTaskID(), next_task_index, ActorID::Nil(), actor_creation_dummy_object_id, 0, actor_handle.ActorID(), - actor_handle.ActorHandleID(), actor_handle.IncreaseTaskCounter(), new_actor_handles, - task_arguments, num_returns, task_options.resources, task_options.resources, - language, function.function_descriptor); + actor_handle.ActorHandleID(), actor_handle.IncreaseTaskCounter(), + actor_handle.NewActorHandles(), task_arguments, num_returns, task_options.resources, + task_options.resources, language, function.function_descriptor); std::vector execution_dependencies; execution_dependencies.push_back(actor_handle.ActorCursor()); @@ -115,6 +202,8 @@ Status CoreWorkerTaskInterface::SubmitActorTask(ActorHandle &actor_handle, actor_handle.SetActorCursor(actor_cursor); actor_handle.ClearNewActorHandles(); + guard.unlock(); + TaskSpec task(std::move(spec), execution_dependencies); auto status = task_submitters_[static_cast(TaskTransportType::RAYLET)]->SubmitTask(task); diff --git a/src/ray/core_worker/task_interface.h b/src/ray/core_worker/task_interface.h index e59934f9b..15839d72d 100644 --- a/src/ray/core_worker/task_interface.h +++ b/src/ray/core_worker/task_interface.h @@ -1,14 +1,14 @@ #ifndef RAY_CORE_WORKER_TASK_INTERFACE_H #define RAY_CORE_WORKER_TASK_INTERFACE_H -#include - #include "ray/common/buffer.h" #include "ray/common/id.h" #include "ray/common/status.h" #include "ray/core_worker/common.h" #include "ray/core_worker/transport/transport.h" +#include "ray/protobuf/core_worker.pb.h" #include "ray/raylet/task.h" +#include "ray/rpc/util.h" namespace ray { @@ -43,45 +43,65 @@ struct ActorCreationOptions { /// A handle to an actor. class ActorHandle { public: - ActorHandle(const ActorID &actor_id, const ActorHandleID &actor_handle_id) - : actor_id_(actor_id), - actor_handle_id_(actor_handle_id), - actor_cursor_(ObjectID::FromBinary(actor_id.Binary())), - task_counter_(0) {} + ActorHandle(const ActorID &actor_id, const ActorHandleID &actor_handle_id, + const ray::rpc::Language actor_language, + const std::vector &actor_creation_task_function_descriptor); + + ActorHandle(const ActorHandle &other); /// ID of the actor. - const ray::ActorID &ActorID() const { return actor_id_; }; + ray::ActorID ActorID() const; /// ID of this actor handle. - const ray::ActorHandleID &ActorHandleID() const { return actor_handle_id_; }; + ray::ActorHandleID ActorHandleID() const; + + /// Language of the actor. + ray::rpc::Language ActorLanguage() const; + + // Function descriptor of actor creation task. + std::vector ActorCreationTaskFunctionDescriptor() const; + + /// The unique id of the last return of the last task. + /// It's used as a dependency for the next task. + ObjectID ActorCursor() const; + + /// The number of tasks that have been invoked on this actor. + int64_t TaskCounter() const; + + /// The number of times that this actor handle has been forked. + /// It's used to make sure ids of actor handles are unique. + int64_t NumForks() const; + + ActorHandle Fork(); + + void Serialize(std::string *output); + + static ActorHandle Deserialize(const std::string &data); private: - /// Cursor of this actor. - const ObjectID &ActorCursor() const { return actor_cursor_; }; + ActorHandle(); /// Set actor cursor. - void SetActorCursor(const ObjectID &actor_cursor) { actor_cursor_ = actor_cursor; }; + void SetActorCursor(const ObjectID &actor_cursor); /// Increase task counter. - int IncreaseTaskCounter() { return task_counter_++; } + int64_t IncreaseTaskCounter(); - std::list GetNewActorHandle() { - // TODO(zhijunfu): implement this. - return std::list(); - } + std::vector NewActorHandles() const; - void ClearNewActorHandles() { /* TODO(zhijunfu): implement this. */ - } + void ClearNewActorHandles(); private: - /// ID of the actor. - const ray::ActorID actor_id_; - /// ID of this actor handle. - const ray::ActorHandleID actor_handle_id_; - /// ID of this actor cursor. - ObjectID actor_cursor_; - /// Counter for tasks from this handle. - int task_counter_; + /// Protobuf defined ActorHandle. + ray::rpc::ActorHandle inner_; + /// The new actor handles that were created from this handle + /// since the last task on this handle was submitted. This is + /// used to garbage-collect dummy objects that are no longer + /// necessary in the backend. + std::vector new_actor_handles_; + + /// Mutex to protect mutable fields. + std::mutex mutex_; friend class CoreWorkerTaskInterface; }; diff --git a/src/ray/protobuf/core_worker.proto b/src/ray/protobuf/core_worker.proto new file mode 100644 index 000000000..e698b4602 --- /dev/null +++ b/src/ray/protobuf/core_worker.proto @@ -0,0 +1,28 @@ +syntax = "proto3"; + +package ray.rpc; + +message ActorHandle { + // ID of the actor. + bytes actor_id = 1; + + // ID of this actor handle. + bytes actor_handle_id = 2; + + // Language of the actor. + int32 actor_language = 3; + + // Function descriptor of actor creation task. + repeated string actor_creation_task_function_descriptor = 4; + + // The unique id of the last return of the last task. + // It's used as a dependency for the next task. + bytes actor_cursor = 5; + + // The number of tasks that have been invoked on this actor. + int64 task_counter = 6; + + // The number of times that this actor handle has been forked. + // It's used to make sure ids of actor handles are unique. + int64 num_forks = 7; +}