[Core worker] Serialize ActorHandle in core worker. Make ActorHandle thread safe. (#5034)

* Serialize ActorHandle in core worker. Make ActorHandle thread safe.

* Address comments

* Address comments

* Address comments

* Address comments

* lint

* Address comments

* Address comments

* Address comments

* Address comments

* Minor update

* Address comments

* lint

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",

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<const BYTE *>(actor_handle_id.Data()),
+                actor_handle_id.Size());
+  sha256_update(&ctx, reinterpret_cast<const BYTE *>(&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()) {                                             \

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 <typename T>
 BaseID<T>::BaseID() {
   // Using const_cast to directly change data is dangerous. The cached

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<std::string> 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<int>(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<int>(language);
+    break;
+  }
+}
+
 }  // namespace ray
 
 #endif  // RAY_CORE_WORKER_COMMON_H

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<int>(language);
-    break;
-  }
-}
-
 }  // namespace ray

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_;

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<std::string, double> &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<LocalMemoryBuffer>(array1, sizeof(array1));
 
-      RayFunction func{ray::WorkerLanguage::PYTHON, {}};
+      RayFunction func{ray::rpc::Language::PYTHON, {}};
       std::vector<TaskArg> args;
       args.emplace_back(TaskArg::PassByValue(buffer1));
 
@@ -165,7 +165,7 @@ class CoreWorkerTest : public ::testing::Test {
       std::vector<TaskArg> args;
       args.emplace_back(TaskArg::PassByReference(object_id));
 
-      RayFunction func{ray::WorkerLanguage::PYTHON, {}};
+      RayFunction func{ray::rpc::Language::PYTHON, {}};
       TaskOptions options;
 
       std::vector<ObjectID> return_ids;
@@ -184,7 +184,7 @@ class CoreWorkerTest : public ::testing::Test {
   }
 
   void TestActorTask(const std::unordered_map<std::string, double> &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<LocalMemoryBuffer>(array, sizeof(array));
 
-      RayFunction func{ray::WorkerLanguage::PYTHON, {}};
+      RayFunction func{ray::rpc::Language::PYTHON, {}};
       std::vector<TaskArg> args;
       args.emplace_back(TaskArg::PassByValue(buffer));
 
@@ -223,7 +223,7 @@ class CoreWorkerTest : public ::testing::Test {
 
       TaskOptions options{1, resources};
       std::vector<ObjectID> 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();

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,

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<std::shared_ptr<RayObject>> args;

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<std::string> &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<int>(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<std::string> 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<std::mutex> 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<std::mutex> 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<ray::ActorHandleID> 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<ObjectID> 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<ActorHandle>(
-      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<std::mutex> guard(actor_handle.mutex_);
 
-  std::vector<ActorHandleID> 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<ObjectID> 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<int>(TaskTransportType::RAYLET)]->SubmitTask(task);

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 <list>
-
 #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<std::string> &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<std::string> 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<ray::ActorHandleID> GetNewActorHandle() {
-    // TODO(zhijunfu): implement this.
-    return std::list<ray::ActorHandleID>();
-  }
+  std::vector<ray::ActorHandleID> 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<ray::ActorHandleID> new_actor_handles_;
+
+  /// Mutex to protect mutable fields.
+  std::mutex mutex_;
 
   friend class CoreWorkerTaskInterface;
 };

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;
+}