[core worker] Submit Python actor tasks through core worker (#5750)

* Submit actor tasks through core worker

* Fix java

* add comment

* Remove task builder

* Check negative

* Increase -> Increment

* pass by reference

* fix signal

* Clean up c++ actor handle

* more cleanup

* Clean up headers

* Fix unique_ptr construction

* Fix java

* Move profiling to c++

* dedup

* fix error

* comments

* fix java

* Fix tests

* wait for actor to exit

* Start after constructor

* ignore java build

* fix comment

* always init logging

* Fix logging

* fix logging issue

* shared_ptr for profiler

* DEBUG -> WARNING

* fix killed_ init

* Fix flaky checkpointing tests

* -v flag for tune tests

* Fix checkpoint test logic

* Fix exception matching

* timeout exception

* Fix test exception info

* Fix import

* fix build

* Fix test

* shared_ptr

src/ray/common/id.cc

@@ -346,8 +346,8 @@ ObjectID ObjectID::GenerateObjectId(const std::string &task_id_binary,
   return ret;
 }
 
-const ActorHandleID ComputeNextActorHandleId(const ActorHandleID &actor_handle_id,
-                                             int64_t num_forks) {
+const ActorHandleID ComputeForkedActorHandleId(const ActorHandleID &actor_handle_id,
+                                               int64_t num_forks) {
   // Compute hashes.
   SHA256_CTX ctx;
   sha256_init(&ctx);
@@ -362,6 +362,23 @@ const ActorHandleID ComputeNextActorHandleId(const ActorHandleID &actor_handle_i
   return ActorHandleID::FromBinary(std::string(buff, buff + ActorHandleID::Size()));
 }
 
+const ActorHandleID ComputeSerializedActorHandleId(const ActorHandleID &actor_handle_id,
+                                                   const TaskID &current_task_id) {
+  // 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 *>(current_task_id.Data()),
+                current_task_id.Size());
+
+  // 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()));
+}
+
 JobID JobID::FromInt(uint32_t value) {
   std::vector<uint8_t> data(JobID::Size(), 0);
   std::memcpy(data.data(), &value, JobID::Size());

src/ray/common/id.h

@@ -366,16 +366,25 @@ std::ostream &operator<<(std::ostream &os, const ObjectID &id);
 
 #undef DEFINE_UNIQUE_ID
 
-// Restore the compiler alignment to defult (8 bytes).
+// Restore the compiler alignment to default (8 bytes).
 #pragma pack(pop)
 
-/// Compute the next actor handle ID of a new actor handle during a fork operation.
+/// Compute an actor handle ID for a newly forked actor handle.
 ///
-/// \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);
+/// \param actor_handle_id The actor handle ID of the existing actor handle.
+/// \param num_forks The number of forks of the existing actor handle.
+/// \return Generated actor handle ID.
+const ActorHandleID ComputeForkedActorHandleId(const ActorHandleID &actor_handle_id,
+                                               int64_t num_forks);
+
+/// Compute an actor handle ID for a new actor handle created by an
+/// out-of-band serialization mechanism.
+///
+/// \param actor_handle_id The actor handle ID of the existing actor handle.
+/// \param current_task_id The current task ID.
+/// \return Generated actor handle ID.
+const ActorHandleID ComputeSerializedActorHandleId(const ActorHandleID &actor_handle_id,
+                                                   const TaskID &current_task_id);
 
 template <typename T>
 BaseID<T>::BaseID() {

src/ray/core_worker/actor_handle.cc

@@ -0,0 +1,107 @@
+#include <memory>
+
+#include "ray/core_worker/actor_handle.h"
+
+namespace ray {
+
+ActorHandle::ActorHandle(
+    const class ActorID &actor_id, const class ActorHandleID &actor_handle_id,
+    const class JobID &job_id, const ObjectID &initial_cursor,
+    const Language actor_language, bool is_direct_call,
+    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_creation_job_id(job_id.Data(), job_id.Size());
+  inner_.set_actor_language(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(initial_cursor.Binary());
+  inner_.set_is_direct_call(is_direct_call);
+  // Increment the task counter to account for the actor creation task.
+  task_counter_++;
+}
+
+std::unique_ptr<ActorHandle> ActorHandle::Fork() {
+  std::unique_lock<std::mutex> guard(mutex_);
+  std::unique_ptr<ActorHandle> child =
+      std::unique_ptr<ActorHandle>(new ActorHandle(inner_));
+  child->inner_ = inner_;
+  const class ActorHandleID new_actor_handle_id =
+      ComputeForkedActorHandleId(GetActorHandleID(), num_forks_++);
+  // Notify the backend to expect this new actor handle. The backend will
+  // not release the cursor for any new handles until the first task for
+  // each of the new handles is submitted.
+  // NOTE(swang): There is currently no garbage collection for actor
+  // handles until the actor itself is removed.
+  new_actor_handles_.push_back(new_actor_handle_id);
+  guard.unlock();
+
+  child->inner_.set_actor_handle_id(new_actor_handle_id.Data(),
+                                    new_actor_handle_id.Size());
+  return child;
+}
+
+std::unique_ptr<ActorHandle> ActorHandle::ForkForSerialization() {
+  std::unique_lock<std::mutex> guard(mutex_);
+  std::unique_ptr<ActorHandle> child =
+      std::unique_ptr<ActorHandle>(new ActorHandle(inner_));
+  child->inner_ = inner_;
+  // The execution dependency for a serialized actor handle is never safe
+  // to release, since it could be deserialized and submit another
+  // dependent task at any time. Therefore, we notify the backend of a
+  // random handle ID that will never actually be used.
+  new_actor_handles_.push_back(ActorHandleID::FromRandom());
+  guard.unlock();
+
+  // We set the actor handle ID to nil to signal that this actor handle was
+  // created by an out-of-band fork. A new actor handle ID will be computed
+  // when the handle is deserialized.
+  const class ActorHandleID new_actor_handle_id = ActorHandleID::Nil();
+  child->inner_.set_actor_handle_id(new_actor_handle_id.Data(),
+                                    new_actor_handle_id.Size());
+  return child;
+}
+
+ActorHandle::ActorHandle(const std::string &serialized, const TaskID &current_task_id) {
+  inner_.ParseFromString(serialized);
+  // If the actor handle ID is nil, this serialized handle was created by an out-of-band
+  // mechanism (see fork constructor above), so we compute a new actor handle ID.
+  // TODO(pcm): This still leads to a lot of actor handles being
+  // created, there should be a better way to handle serialized
+  // actor handles.
+  // TODO(swang): Deserializing the same actor handle twice in the same
+  // task will break the application, and deserializing it twice in the
+  // same actor is likely a performance bug. We should consider
+  // logging a warning in these cases.
+  if (ActorHandleID::FromBinary(inner_.actor_handle_id()).IsNil()) {
+    const class ActorHandleID new_actor_handle_id = ComputeSerializedActorHandleId(
+        ActorHandleID::FromBinary(inner_.actor_handle_id()), current_task_id);
+    inner_.set_actor_handle_id(new_actor_handle_id.Data(), new_actor_handle_id.Size());
+  }
+}
+
+void ActorHandle::SetActorTaskSpec(TaskSpecBuilder &builder,
+                                   const TaskTransportType transport_type,
+                                   const ObjectID new_cursor) {
+  std::unique_lock<std::mutex> guard(mutex_);
+  // Build actor task spec.
+  const TaskID actor_creation_task_id = TaskID::ForActorCreationTask(GetActorID());
+  const ObjectID actor_creation_dummy_object_id =
+      ObjectID::ForTaskReturn(actor_creation_task_id, /*index=*/1,
+                              /*transport_type=*/static_cast<int>(transport_type));
+  builder.SetActorTaskSpec(GetActorID(), GetActorHandleID(),
+                           actor_creation_dummy_object_id,
+                           /*previous_actor_task_dummy_object_id=*/ActorCursor(),
+                           task_counter_++, new_actor_handles_);
+
+  inner_.set_actor_cursor(new_cursor.Binary());
+  new_actor_handles_.clear();
+}
+
+void ActorHandle::Serialize(std::string *output) {
+  std::unique_lock<std::mutex> guard(mutex_);
+  inner_.SerializeToString(output);
+}
+
+}  // namespace ray

src/ray/core_worker/actor_handle.h

@@ -0,0 +1,85 @@
+#ifndef RAY_CORE_WORKER_ACTOR_HANDLE_H
+#define RAY_CORE_WORKER_ACTOR_HANDLE_H
+
+#include <gtest/gtest_prod.h>
+
+#include "ray/common/id.h"
+#include "ray/common/task/task_util.h"
+#include "ray/core_worker/common.h"
+#include "ray/core_worker/context.h"
+#include "ray/protobuf/core_worker.pb.h"
+
+namespace ray {
+
+class ActorHandle {
+ public:
+  ActorHandle(ray::rpc::ActorHandle inner) : inner_(inner) {}
+
+  // Constructs a new ActorHandle as part of the actor creation process.
+  ActorHandle(const ActorID &actor_id, const ActorHandleID &actor_handle_id,
+              const JobID &job_id, const ObjectID &initial_cursor,
+              const Language actor_language, bool is_direct_call,
+              const std::vector<std::string> &actor_creation_task_function_descriptor);
+
+  /// Constructs an ActorHandle from a serialized string.
+  ActorHandle(const std::string &serialized, const TaskID &current_task_id);
+
+  /// Forks a child ActorHandle. This will modify the handle to account for the newly
+  /// forked child handle. This should only be used for forks that are part of a Ray
+  /// API call (e.g., passing an actor handle into a remote function).
+  std::unique_ptr<ActorHandle> Fork();
+
+  /// Forks a child ActorHandle. This will *not* modify the handle to account for the
+  /// newly forked child handle. This should be used by application-level code for
+  /// serialization in order to pass an actor handle for uses not covered by the Ray API.
+  std::unique_ptr<ActorHandle> ForkForSerialization();
+
+  ActorID GetActorID() const { return ActorID::FromBinary(inner_.actor_id()); };
+
+  ActorHandleID GetActorHandleID() const {
+    return ActorHandleID::FromBinary(inner_.actor_handle_id());
+  };
+
+  /// ID of the job that created the actor (it is possible that the handle
+  /// exists on a job with a different job ID).
+  JobID CreationJobID() const { return JobID::FromBinary(inner_.creation_job_id()); };
+
+  Language ActorLanguage() const { return inner_.actor_language(); };
+
+  std::vector<std::string> ActorCreationTaskFunctionDescriptor() const {
+    return VectorFromProtobuf(inner_.actor_creation_task_function_descriptor());
+  };
+
+  ObjectID ActorCursor() const { return ObjectID::FromBinary(inner_.actor_cursor()); }
+
+  bool IsDirectCallActor() const { return inner_.is_direct_call(); }
+
+  void SetActorTaskSpec(TaskSpecBuilder &builder, const TaskTransportType transport_type,
+                        const ObjectID new_cursor);
+
+  void Serialize(std::string *output);
+
+ private:
+  // Protobuf-defined persistent state of the actor handle.
+  ray::rpc::ActorHandle inner_;
+
+  // Number of times this handle has been forked.
+  uint64_t num_forks_ = 0;
+
+  // Number of tasks that have been submitted on this handle.
+  uint64_t task_counter_ = 0;
+
+  /// 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_;
+
+  std::mutex mutex_;
+
+  FRIEND_TEST(ZeroNodeTest, TestActorHandle);
+};
+
+}  // namespace ray
+
+#endif  // RAY_CORE_WORKER_ACTOR_HANDLE_H

src/ray/core_worker/core_worker.cc

@@ -20,7 +20,7 @@ CoreWorker::CoreWorker(
       io_work_(io_service_) {
   // Initialize logging if log_dir is passed. Otherwise, it must be initialized
   // and cleaned up by the caller.
-  if (!log_dir_.empty()) {
+  if (log_dir_ != "") {
     std::stringstream app_name;
     app_name << LanguageString(language_) << "-" << WorkerTypeString(worker_type_) << "-"
              << worker_context_.GetWorkerID();

src/ray/core_worker/lib/java/org_ray_runtime_actor_NativeRayActor.cc

@@ -20,8 +20,7 @@ extern "C" {
  */
 JNIEXPORT jlong JNICALL Java_org_ray_runtime_actor_NativeRayActor_nativeFork(
     JNIEnv *env, jclass o, jlong nativeActorHandle) {
-  auto new_actor_handle = GetActorHandle(nativeActorHandle).Fork();
-  return reinterpret_cast<jlong>(new ray::ActorHandle(new_actor_handle));
+  return reinterpret_cast<jlong>(GetActorHandle(nativeActorHandle).Fork().release());
 }
 
 /*
@@ -32,7 +31,7 @@ JNIEXPORT jlong JNICALL Java_org_ray_runtime_actor_NativeRayActor_nativeFork(
 JNIEXPORT jbyteArray JNICALL Java_org_ray_runtime_actor_NativeRayActor_nativeGetActorId(
     JNIEnv *env, jclass o, jlong nativeActorHandle) {
   return IdToJavaByteArray<ray::ActorID>(env,
-                                         GetActorHandle(nativeActorHandle).ActorID());
+                                         GetActorHandle(nativeActorHandle).GetActorID());
 }
 
 /*
@@ -44,7 +43,7 @@ JNIEXPORT jbyteArray JNICALL
 Java_org_ray_runtime_actor_NativeRayActor_nativeGetActorHandleId(
     JNIEnv *env, jclass o, jlong nativeActorHandle) {
   return IdToJavaByteArray<ray::ActorHandleID>(
-      env, GetActorHandle(nativeActorHandle).ActorHandleID());
+      env, GetActorHandle(nativeActorHandle).GetActorHandleID());
 }
 
 /*
@@ -62,7 +61,8 @@ JNIEXPORT jint JNICALL Java_org_ray_runtime_actor_NativeRayActor_nativeGetLangua
  * Method:    nativeIsDirectCallActor
  * Signature: (J)Z
  */
-JNIEXPORT jboolean JNICALL Java_org_ray_runtime_actor_NativeRayActor_nativeIsDirectCallActor(
+JNIEXPORT jboolean JNICALL
+Java_org_ray_runtime_actor_NativeRayActor_nativeIsDirectCallActor(
     JNIEnv *env, jclass o, jlong nativeActorHandle) {
   return GetActorHandle(nativeActorHandle).IsDirectCallActor();
 }
@@ -104,8 +104,7 @@ JNIEXPORT jlong JNICALL Java_org_ray_runtime_actor_NativeRayActor_nativeDeserial
   auto buffer = JavaByteArrayToNativeBuffer(env, data);
   RAY_CHECK(buffer->Size() > 0);
   auto binary = std::string(reinterpret_cast<char *>(buffer->Data()), buffer->Size());
-  return reinterpret_cast<jlong>(
-      new ray::ActorHandle(ray::ActorHandle::Deserialize(binary)));
+  return reinterpret_cast<jlong>(new ray::ActorHandle(binary, TaskID::Nil()));
 }
 
 /*

src/ray/core_worker/lib/java/org_ray_runtime_task_NativeTaskSubmitter.cc

@@ -84,7 +84,7 @@ inline ray::ActorCreationOptions ToActorCreationOptions(JNIEnv *env,
     max_reconstructions = static_cast<uint64_t>(env->GetIntField(
         actorCreationOptions, java_actor_creation_options_max_reconstructions));
     use_direct_call = env->GetBooleanField(actorCreationOptions,
-                                          java_actor_creation_options_use_direct_call);
+                                           java_actor_creation_options_use_direct_call);
     jobject java_resources =
         env->GetObjectField(actorCreationOptions, java_base_task_options_resources);
     resources = ToResources(env, java_resources);
@@ -101,7 +101,7 @@ inline ray::ActorCreationOptions ToActorCreationOptions(JNIEnv *env,
   }
 
   ray::ActorCreationOptions action_creation_options{
-      static_cast<uint64_t>(max_reconstructions), use_direct_call, resources,
+      static_cast<uint64_t>(max_reconstructions), use_direct_call, resources, resources,
       dynamic_worker_options};
   return action_creation_options;
 }

src/ray/core_worker/task_interface.cc

@@ -1,103 +1,11 @@
 #include "ray/core_worker/task_interface.h"
 #include "ray/core_worker/context.h"
-#include "ray/core_worker/core_worker.h"
 #include "ray/core_worker/task_interface.h"
 #include "ray/core_worker/transport/direct_actor_transport.h"
 #include "ray/core_worker/transport/raylet_transport.h"
 
 namespace ray {
 
-ActorHandle::ActorHandle(
-    const class ActorID &actor_id, const class ActorHandleID &actor_handle_id,
-    const Language actor_language, bool is_direct_call,
-    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(actor_language);
-  *inner_.mutable_actor_creation_task_function_descriptor() = {
-      actor_creation_task_function_descriptor.begin(),
-      actor_creation_task_function_descriptor.end()};
-  const auto &actor_creation_task_id = TaskID::ForActorCreationTask(actor_id);
-  const auto &actor_creation_dummy_object_id =
-      ObjectID::ForTaskReturn(actor_creation_task_id, /*index=*/1, /*transport_type=*/0);
-  inner_.set_actor_cursor(actor_creation_dummy_object_id.Data(),
-                          actor_creation_dummy_object_id.Size());
-  inner_.set_is_direct_call(is_direct_call);
-}
-
-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());
-};
-
-Language ActorHandle::ActorLanguage() const { return inner_.actor_language(); };
-
-std::vector<std::string> ActorHandle::ActorCreationTaskFunctionDescriptor() const {
-  return 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(); };
-
-bool ActorHandle::IsDirectCallActor() const { return inner_.is_direct_call(); }
-
-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(
     WorkerContext &worker_context, std::unique_ptr<RayletClient> &raylet_client,
     CoreWorkerObjectInterface &object_interface, boost::asio::io_service &io_service,
@@ -115,16 +23,17 @@ CoreWorkerTaskInterface::CoreWorkerTaskInterface(
 }
 
 void CoreWorkerTaskInterface::BuildCommonTaskSpec(
-    TaskSpecBuilder &builder, const TaskID &task_id, const int task_index,
-    const RayFunction &function, const std::vector<TaskArg> &args, uint64_t num_returns,
+    TaskSpecBuilder &builder, const JobID &job_id, const TaskID &task_id,
+    const int task_index, const RayFunction &function, const std::vector<TaskArg> &args,
+    uint64_t num_returns,
     const std::unordered_map<std::string, double> &required_resources,
     const std::unordered_map<std::string, double> &required_placement_resources,
     TaskTransportType transport_type, std::vector<ObjectID> *return_ids) {
   // Build common task spec.
-  builder.SetCommonTaskSpec(
-      task_id, function.GetLanguage(), function.GetFunctionDescriptor(),
-      worker_context_.GetCurrentJobID(), worker_context_.GetCurrentTaskID(), task_index,
-      num_returns, required_resources, required_placement_resources);
+  builder.SetCommonTaskSpec(task_id, function.GetLanguage(),
+                            function.GetFunctionDescriptor(), job_id,
+                            worker_context_.GetCurrentTaskID(), task_index, num_returns,
+                            required_resources, required_placement_resources);
   // Set task arguments.
   for (const auto &arg : args) {
     if (arg.IsPassedByReference()) {
@@ -152,9 +61,9 @@ Status CoreWorkerTaskInterface::SubmitTask(const RayFunction &function,
   const auto task_id =
       TaskID::ForNormalTask(worker_context_.GetCurrentJobID(),
                             worker_context_.GetCurrentTaskID(), next_task_index);
-  BuildCommonTaskSpec(builder, task_id, next_task_index, function, args,
-                      task_options.num_returns, task_options.resources, {},
-                      TaskTransportType::RAYLET, return_ids);
+  BuildCommonTaskSpec(builder, worker_context_.GetCurrentJobID(), task_id,
+                      next_task_index, function, args, task_options.num_returns,
+                      task_options.resources, {}, TaskTransportType::RAYLET, return_ids);
   return task_submitters_[TaskTransportType::RAYLET]->SubmitTask(builder.Build());
 }
 
@@ -167,20 +76,21 @@ Status CoreWorkerTaskInterface::CreateActor(
       ActorID::Of(worker_context_.GetCurrentJobID(), worker_context_.GetCurrentTaskID(),
                   next_task_index);
   const TaskID actor_creation_task_id = TaskID::ForActorCreationTask(actor_id);
+  const JobID job_id = worker_context_.GetCurrentJobID();
   std::vector<ObjectID> return_ids;
   TaskSpecBuilder builder;
-  BuildCommonTaskSpec(builder, actor_creation_task_id, next_task_index, function, args, 1,
-                      actor_creation_options.resources, actor_creation_options.resources,
+  BuildCommonTaskSpec(builder, job_id, actor_creation_task_id, next_task_index, function,
+                      args, 1, actor_creation_options.resources,
+                      actor_creation_options.placement_resources,
                       TaskTransportType::RAYLET, &return_ids);
   builder.SetActorCreationTaskSpec(actor_id, actor_creation_options.max_reconstructions,
                                    actor_creation_options.dynamic_worker_options,
                                    actor_creation_options.is_direct_call);
 
   *actor_handle = std::unique_ptr<ActorHandle>(new ActorHandle(
-      actor_id, ActorHandleID::Nil(), function.GetLanguage(),
-      actor_creation_options.is_direct_call, function.GetFunctionDescriptor()));
-  (*actor_handle)->IncreaseTaskCounter();
-  (*actor_handle)->SetActorCursor(return_ids[0]);
+      actor_id, ActorHandleID::Nil(), job_id, /*actor_cursor=*/return_ids[0],
+      function.GetLanguage(), actor_creation_options.is_direct_call,
+      function.GetFunctionDescriptor()));
 
   return task_submitters_[TaskTransportType::RAYLET]->SubmitTask(builder.Build());
 }
@@ -191,46 +101,29 @@ Status CoreWorkerTaskInterface::SubmitActorTask(ActorHandle &actor_handle,
                                                 const TaskOptions &task_options,
                                                 std::vector<ObjectID> *return_ids) {
   // Add one for actor cursor object id for tasks.
-  const auto num_returns = task_options.num_returns + 1;
+  const int num_returns = task_options.num_returns + 1;
 
   const bool is_direct_call = actor_handle.IsDirectCallActor();
-  const auto transport_type =
+  const TaskTransportType transport_type =
       is_direct_call ? TaskTransportType::DIRECT_ACTOR : TaskTransportType::RAYLET;
 
   // Build common task spec.
   TaskSpecBuilder builder;
   const int next_task_index = worker_context_.GetNextTaskIndex();
-  const auto actor_task_id = TaskID::ForActorTask(
+  const TaskID actor_task_id = TaskID::ForActorTask(
       worker_context_.GetCurrentJobID(), worker_context_.GetCurrentTaskID(),
-      next_task_index, actor_handle.ActorID());
-  BuildCommonTaskSpec(builder, actor_task_id, next_task_index, function, args,
-                      num_returns, task_options.resources, {}, transport_type,
-                      return_ids);
+      next_task_index, actor_handle.GetActorID());
+  BuildCommonTaskSpec(builder, actor_handle.CreationJobID(), actor_task_id,
+                      next_task_index, function, args, num_returns,
+                      task_options.resources, {}, transport_type, return_ids);
 
-  std::unique_lock<std::mutex> guard(actor_handle.mutex_);
-  // Build actor task spec.
-  const auto actor_creation_task_id =
-      TaskID::ForActorCreationTask(actor_handle.ActorID());
-  const auto actor_creation_dummy_object_id =
-      ObjectID::ForTaskReturn(actor_creation_task_id, /*index=*/1,
-                              /*transport_type=*/static_cast<int>(transport_type));
-  builder.SetActorTaskSpec(
-      actor_handle.ActorID(), actor_handle.ActorHandleID(),
-      actor_creation_dummy_object_id,
-      /*previous_actor_task_dummy_object_id=*/actor_handle.ActorCursor(),
-      actor_handle.IncreaseTaskCounter(), actor_handle.NewActorHandles());
-
-  // Manipulate actor handle state.
-  auto actor_cursor = (*return_ids).back();
-  actor_handle.SetActorCursor(actor_cursor);
-  actor_handle.ClearNewActorHandles();
-
-  guard.unlock();
+  const ObjectID new_cursor = return_ids->back();
+  actor_handle.SetActorTaskSpec(builder, transport_type, new_cursor);
 
   // Submit task.
   auto status = task_submitters_[transport_type]->SubmitTask(builder.Build());
   // Remove cursor from return ids.
-  (*return_ids).pop_back();
+  return_ids->pop_back();
 
   return status;
 }

src/ray/core_worker/task_interface.h

@@ -8,6 +8,7 @@
 #include "ray/common/task/task.h"
 #include "ray/common/task/task_spec.h"
 #include "ray/common/task/task_util.h"
+#include "ray/core_worker/actor_handle.h"
 #include "ray/core_worker/common.h"
 #include "ray/core_worker/context.h"
 #include "ray/core_worker/object_interface.h"
@@ -17,8 +18,6 @@
 
 namespace ray {
 
-class CoreWorker;
-
 /// Options of a non-actor-creation task.
 struct TaskOptions {
   TaskOptions() {}
@@ -36,10 +35,12 @@ struct ActorCreationOptions {
   ActorCreationOptions() {}
   ActorCreationOptions(uint64_t max_reconstructions, bool is_direct_call,
                        const std::unordered_map<std::string, double> &resources,
+                       const std::unordered_map<std::string, double> &placement_resources,
                        const std::vector<std::string> &dynamic_worker_options)
       : max_reconstructions(max_reconstructions),
         is_direct_call(is_direct_call),
         resources(resources),
+        placement_resources(placement_resources),
         dynamic_worker_options(dynamic_worker_options) {}
 
   /// Maximum number of times that the actor should be reconstructed when it dies
@@ -50,81 +51,13 @@ struct ActorCreationOptions {
   const bool is_direct_call = false;
   /// Resources required by the whole lifetime of this actor.
   const std::unordered_map<std::string, double> resources;
+  /// Resources required to place this actor.
+  const std::unordered_map<std::string, double> placement_resources;
   /// The dynamic options used in the worker command when starting a worker process for
   /// an actor creation task.
   const std::vector<std::string> dynamic_worker_options;
 };
 
-/// A handle to an actor.
-class ActorHandle {
- public:
-  ActorHandle(const ActorID &actor_id, const ActorHandleID &actor_handle_id,
-              const Language actor_language, bool is_direct_call,
-              const std::vector<std::string> &actor_creation_task_function_descriptor);
-
-  ActorHandle(const ActorHandle &other);
-
-  /// ID of the actor.
-  ray::ActorID ActorID() const;
-
-  /// ID of this actor handle.
-  ray::ActorHandleID ActorHandleID() const;
-
-  /// Language of the actor.
-  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;
-
-  /// Whether direct call is used. If this is true, then the tasks
-  /// are submitted directly to the actor without going through raylet.
-  bool IsDirectCallActor() const;
-
-  ActorHandle Fork();
-
-  void Serialize(std::string *output);
-
-  static ActorHandle Deserialize(const std::string &data);
-
- private:
-  ActorHandle();
-
-  /// Set actor cursor.
-  void SetActorCursor(const ObjectID &actor_cursor);
-
-  /// Increase task counter.
-  int64_t IncreaseTaskCounter();
-
-  std::vector<ray::ActorHandleID> NewActorHandles() const;
-
-  void ClearNewActorHandles();
-
- private:
-  /// 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;
-};
-
 /// The interface that contains all `CoreWorker` methods that are related to task
 /// submission.
 class CoreWorkerTaskInterface {
@@ -173,6 +106,7 @@ class CoreWorkerTaskInterface {
   /// Build common attributes of the task spec, and compute return ids.
   ///
   /// \param[in] builder Builder to build a `TaskSpec`.
+  /// \param[in] job_id The ID of the job submitting the task.
   /// \param[in] task_id The ID of this task.
   /// \param[in] task_index The task index used to build this task.
   /// \param[in] function The remote function to execute.
@@ -185,8 +119,9 @@ class CoreWorkerTaskInterface {
   /// \param[out] return_ids Return IDs.
   /// \return Void.
   void BuildCommonTaskSpec(
-      TaskSpecBuilder &builder, const TaskID &task_id, const int task_index,
-      const RayFunction &function, const std::vector<TaskArg> &args, uint64_t num_returns,
+      TaskSpecBuilder &builder, const JobID &job_id, const TaskID &task_id,
+      const int task_index, const RayFunction &function, const std::vector<TaskArg> &args,
+      uint64_t num_returns,
       const std::unordered_map<std::string, double> &required_resources,
       const std::unordered_map<std::string, double> &required_placement_resources,
       TaskTransportType transport_type, std::vector<ObjectID> *return_ids);

src/ray/core_worker/test/core_worker_test.cc

@@ -61,7 +61,8 @@ std::unique_ptr<ActorHandle> CreateActorHelper(
   std::vector<TaskArg> args;
   args.emplace_back(TaskArg::PassByValue(std::make_shared<RayObject>(buffer, nullptr)));
 
-  ActorCreationOptions actor_options{max_reconstructions, is_direct_call, resources, {}};
+  ActorCreationOptions actor_options{
+      max_reconstructions, is_direct_call, resources, resources, {}};
 
   // Create an actor.
   RAY_CHECK_OK(worker.Tasks().CreateActor(func, args, actor_options, &actor_handle));
@@ -358,7 +359,7 @@ void CoreWorkerTest::TestActorReconstruction(
   auto actor_handle = CreateActorHelper(driver, resources, is_direct_call, 1000);
 
   // Wait for actor alive event.
-  ASSERT_TRUE(WaitForDirectCallActorState(driver, actor_handle->ActorID(), true,
+  ASSERT_TRUE(WaitForDirectCallActorState(driver, actor_handle->GetActorID(), true,
                                           30 * 1000 /* 30s */));
   RAY_LOG(INFO) << "actor has been created";
 
@@ -373,9 +374,9 @@ void CoreWorkerTest::TestActorReconstruction(
         ASSERT_EQ(system("pkill mock_worker"), 0);
 
         // Wait for actor restruction event, and then for alive event.
-        ASSERT_TRUE(WaitForDirectCallActorState(driver, actor_handle->ActorID(), false,
+        ASSERT_TRUE(WaitForDirectCallActorState(driver, actor_handle->GetActorID(), false,
                                                 30 * 1000 /* 30s */));
-        ASSERT_TRUE(WaitForDirectCallActorState(driver, actor_handle->ActorID(), true,
+        ASSERT_TRUE(WaitForDirectCallActorState(driver, actor_handle->GetActorID(), true,
                                                 30 * 1000 /* 30s */));
 
         RAY_LOG(INFO) << "actor has been reconstructed";
@@ -649,10 +650,12 @@ TEST_F(ZeroNodeTest, TestTaskSpecPerf) {
   args.emplace_back(TaskArg::PassByValue(std::make_shared<RayObject>(buffer, nullptr)));
 
   std::unordered_map<std::string, double> resources;
-  ActorCreationOptions actor_options{0, /*is_direct_call*/ true, resources, {}};
+  ActorCreationOptions actor_options{
+      0, /*is_direct_call*/ true, resources, resources, {}};
   const auto job_id = NextJobId();
   ActorHandle actor_handle(ActorID::Of(job_id, TaskID::ForDriverTask(job_id), 1),
-                           ActorHandleID::Nil(), function.GetLanguage(), true,
+                           ActorHandleID::Nil(), job_id, ObjectID::FromRandom(),
+                           function.GetLanguage(), true,
                            function.GetFunctionDescriptor());
 
   // Manually create `num_tasks` task specs, and for each of them create a
@@ -679,13 +682,8 @@ TEST_F(ZeroNodeTest, TestTaskSpecPerf) {
       }
     }
 
-    const auto actor_creation_dummy_object_id =
-        ObjectID::ForTaskReturn(TaskID::ForActorCreationTask(actor_handle.ActorID()),
-                                /*index=*/1, /*transport_type=*/0);
-    builder.SetActorTaskSpec(
-        actor_handle.ActorID(), actor_handle.ActorHandleID(),
-        actor_creation_dummy_object_id,
-        /*previous_actor_task_dummy_object_id=*/actor_handle.ActorCursor(), 0, {});
+    actor_handle.SetActorTaskSpec(builder, TaskTransportType::RAYLET,
+                                  ObjectID::FromRandom());
 
     const auto &task_spec = builder.Build();
 
@@ -712,11 +710,12 @@ TEST_F(SingleNodeTest, TestDirectActorTaskSubmissionPerf) {
   args.emplace_back(TaskArg::PassByValue(std::make_shared<RayObject>(buffer, nullptr)));
 
   std::unordered_map<std::string, double> resources;
-  ActorCreationOptions actor_options{0, /*is_direct_call*/ true, resources, {}};
+  ActorCreationOptions actor_options{
+      0, /*is_direct_call*/ true, resources, resources, {}};
   // Create an actor.
   RAY_CHECK_OK(driver.Tasks().CreateActor(func, args, actor_options, &actor_handle));
   // wait for actor creation finish.
-  ASSERT_TRUE(WaitForDirectCallActorState(driver, actor_handle->ActorID(), true,
+  ASSERT_TRUE(WaitForDirectCallActorState(driver, actor_handle->GetActorID(), true,
                                           30 * 1000 /* 30s */));
   // Test submitting some tasks with by-value args for that actor.
   int64_t start_ms = current_time_ms();
@@ -777,37 +776,75 @@ TEST_F(ZeroNodeTest, TestWorkerContext) {
 }
 
 TEST_F(ZeroNodeTest, TestActorHandle) {
-  const auto job_id = NextJobId();
-  ActorHandle handle1(ActorID::Of(job_id, TaskID::ForDriverTask(job_id), 1),
-                      ActorHandleID::FromRandom(), Language::JAVA, false,
-                      {"org.ray.exampleClass", "exampleMethod", "exampleSignature"});
+  const JobID job_id = NextJobId();
+  const TaskID task_id = TaskID::ForDriverTask(job_id);
+  const ActorHandleID actor_handle_id = ActorHandleID::FromRandom();
+  ActorHandle parent(ActorID::Of(job_id, task_id, 1), actor_handle_id, job_id,
+                     ObjectID::FromRandom(), Language::JAVA, false,
+                     {"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());
+  // Test in-band forking logic.
+  std::unique_ptr<ActorHandle> forkedHandle1 = parent.Fork();
+  ASSERT_EQ(1, parent.num_forks_);
+  ASSERT_EQ(parent.GetActorID(), forkedHandle1->GetActorID());
+  ASSERT_EQ(actor_handle_id, parent.GetActorHandleID());
+  ASSERT_NE(parent.GetActorHandleID(), forkedHandle1->GetActorHandleID());
+  ASSERT_EQ(parent.ActorLanguage(), forkedHandle1->ActorLanguage());
+  ASSERT_EQ(parent.ActorCreationTaskFunctionDescriptor(),
+            forkedHandle1->ActorCreationTaskFunctionDescriptor());
+  ASSERT_EQ(parent.ActorCursor(), forkedHandle1->ActorCursor());
+  ASSERT_EQ(0, forkedHandle1->task_counter_);
+  ASSERT_EQ(0, forkedHandle1->num_forks_);
+  ASSERT_EQ(parent.new_actor_handles_.size(), 1);
+  ASSERT_EQ(parent.new_actor_handles_.back(), forkedHandle1->GetActorHandleID());
+  parent.new_actor_handles_.clear();
 
-  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());
+  std::unique_ptr<ActorHandle> forkedHandle2 = parent.Fork();
+  ASSERT_EQ(2, parent.num_forks_);
+  ASSERT_EQ(0, forkedHandle2->task_counter_);
+  ASSERT_EQ(0, forkedHandle2->num_forks_);
+  ASSERT_EQ(parent.new_actor_handles_.size(), 1);
+  ASSERT_EQ(parent.new_actor_handles_.back(), forkedHandle2->GetActorHandleID());
+  parent.new_actor_handles_.clear();
+
+  // Test serialization and deserialization for in-band fork.
+  std::string buffer1;
+  forkedHandle2->Serialize(&buffer1);
+  ActorHandle deserializedHandle1(buffer1, task_id);
+  ASSERT_EQ(forkedHandle2->GetActorID(), deserializedHandle1.GetActorID());
+  ASSERT_EQ(forkedHandle2->GetActorHandleID(), deserializedHandle1.GetActorHandleID());
+  ASSERT_EQ(forkedHandle2->ActorLanguage(), deserializedHandle1.ActorLanguage());
+  ASSERT_EQ(forkedHandle2->ActorCreationTaskFunctionDescriptor(),
+            deserializedHandle1.ActorCreationTaskFunctionDescriptor());
+  ASSERT_EQ(forkedHandle2->ActorCursor(), deserializedHandle1.ActorCursor());
+
+  // Test out-of-band forking logic.
+  std::unique_ptr<ActorHandle> forkedHandle3 = parent.ForkForSerialization();
+  ASSERT_EQ(2, parent.num_forks_);
+  ASSERT_EQ(parent.GetActorID(), forkedHandle3->GetActorID());
+  ASSERT_EQ(actor_handle_id, parent.GetActorHandleID());
+  ASSERT_NE(parent.GetActorHandleID(), forkedHandle3->GetActorHandleID());
+  ASSERT_NE(forkedHandle2->GetActorHandleID(), forkedHandle3->GetActorHandleID());
+  ASSERT_EQ(parent.ActorLanguage(), forkedHandle3->ActorLanguage());
+  ASSERT_EQ(parent.ActorCreationTaskFunctionDescriptor(),
+            forkedHandle3->ActorCreationTaskFunctionDescriptor());
+  ASSERT_EQ(parent.ActorCursor(), forkedHandle3->ActorCursor());
+  ASSERT_EQ(0, forkedHandle3->task_counter_);
+  ASSERT_EQ(0, forkedHandle3->num_forks_);
+  ASSERT_EQ(parent.new_actor_handles_.size(), 1);
+  ASSERT_NE(parent.new_actor_handles_.back(), forkedHandle3->GetActorHandleID());
+  parent.new_actor_handles_.clear();
+
+  // Test serialization and deserialization for out-of-band fork.
+  std::string buffer2;
+  forkedHandle3->Serialize(&buffer2);
+  ActorHandle deserializedHandle2(buffer2, task_id);
+  ASSERT_EQ(forkedHandle3->GetActorID(), deserializedHandle2.GetActorID());
+  ASSERT_NE(forkedHandle3->GetActorHandleID(), deserializedHandle2.GetActorHandleID());
+  ASSERT_EQ(forkedHandle3->ActorLanguage(), deserializedHandle2.ActorLanguage());
+  ASSERT_EQ(forkedHandle3->ActorCreationTaskFunctionDescriptor(),
+            deserializedHandle2.ActorCreationTaskFunctionDescriptor());
+  ASSERT_EQ(forkedHandle3->ActorCursor(), deserializedHandle2.ActorCursor());
 }
 
 TEST_F(SingleNodeTest, TestMemoryStoreProvider) {

src/ray/gcs/redis_gcs_client_test.cc

@@ -95,7 +95,7 @@ std::shared_ptr<TaskTableData> CreateTaskTableData(const TaskID &task_id,
   return data;
 }
 
-/// A helper function that compare wether 2 `TaskTableData` objects are equal.
+/// A helper function that compare whether 2 `TaskTableData` objects are equal.
 /// Note, this function only compares fields set by `CreateTaskTableData`.
 bool TaskTableDataEqual(const TaskTableData &data1, const TaskTableData &data2) {
   const auto &spec1 = data1.task().task_spec();

src/ray/protobuf/core_worker.proto

@@ -4,6 +4,7 @@ package ray.rpc;
 
 import "src/ray/protobuf/common.proto";
 
+// Persistent state of an ActorHandle.
 message ActorHandle {
   // ID of the actor.
   bytes actor_id = 1;
@@ -11,23 +12,20 @@ message ActorHandle {
   // ID of this actor handle.
   bytes actor_handle_id = 2;
 
+  // ID of the job that created the actor (it is possible that the handle
+  // exists on a job with a different job ID).
+  bytes creation_job_id = 3;
+
   // Language of the actor.
-  Language actor_language = 3;
+  Language actor_language = 4;
 
   // Function descriptor of actor creation task.
-  repeated string actor_creation_task_function_descriptor = 4;
+  repeated string actor_creation_task_function_descriptor = 5;
 
   // 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;
+  bytes actor_cursor = 6;
 
   // Whether direct actor call is used.
-  bool is_direct_call = 8;
+  bool is_direct_call = 7;
 }