Actor manager refactor and unit tests (#8224)

* parametrize test

* Regression test and logging

* Test no restart after actor deletion

* Unit tests

* Refactor to subscribe to and lookup from worker failure table

* Refactor ActorManager to remove dependencies

* Revert "Regression test and logging"

This reverts commit 835e1a9091b51ca8efb00392d4cc4a665145de24.

* Revert "parametrize test"

This reverts commit f31272082831ba1a494816dd5511d87b24eca4c9.

* Revert "Test no restart after actor deletion"

This reverts commit 114a83de14329aa6ab787c80cd5757cf074a9072.

* doc

* merge

* Revert "Refactor to subscribe to and lookup from worker failure table"

This reverts commit 6aa13a05178d0b9aa1db9dee5c978c911b74fa3a.

* Use actor ID instead of shared_ptr

* TODO and lint

* Update src/ray/gcs/gcs_server/gcs_actor_scheduler.h

Co-authored-by: Edward Oakes <ed.nmi.oakes@gmail.com>

* Fix build

* doc

* Build

Co-authored-by: Edward Oakes <ed.nmi.oakes@gmail.com>

src/ray/gcs/gcs_server/gcs_actor_manager.cc

@@ -66,42 +66,10 @@ const rpc::ActorTableData &GcsActor::GetActorTableData() const {
 rpc::ActorTableData *GcsActor::GetMutableActorTableData() { return &actor_table_data_; }
 
 /////////////////////////////////////////////////////////////////////////////////////////
-GcsActorManager::GcsActorManager(boost::asio::io_context &io_context,
-                                 gcs::ActorInfoAccessor &actor_info_accessor,
-                                 gcs::GcsNodeManager &gcs_node_manager,
-                                 LeaseClientFactoryFn lease_client_factory,
-                                 rpc::ClientFactoryFn client_factory)
-    : actor_info_accessor_(actor_info_accessor),
-      gcs_actor_scheduler_(new gcs::GcsActorScheduler(
-          io_context, actor_info_accessor, gcs_node_manager,
-          /*schedule_failure_handler=*/
-          [this](std::shared_ptr<GcsActor> actor) {
-            // When there are no available nodes to schedule the actor the
-            // gcs_actor_scheduler will treat it as failed and invoke this handler. In
-            // this case, the actor should be appended to the `pending_actors_` and wait
-            // for the registration of new node.
-            pending_actors_.emplace_back(std::move(actor));
-          },
-          /*schedule_success_handler=*/
-          [this](std::shared_ptr<GcsActor> actor) {
-            OnActorCreateSuccess(std::move(actor));
-          },
-          std::move(lease_client_factory), std::move(client_factory))) {
-  RAY_LOG(INFO) << "Initializing GcsActorManager.";
-  gcs_node_manager.AddNodeAddedListener(
-      [this](const std::shared_ptr<rpc::GcsNodeInfo> &) {
-        // Because a new node has been added, we need to try to schedule the pending
-        // actors.
-        SchedulePendingActors();
-      });
-
-  gcs_node_manager.AddNodeRemovedListener([this](std::shared_ptr<rpc::GcsNodeInfo> node) {
-    // All of the related actors should be reconstructed when a node is removed from the
-    // GCS.
-    ReconstructActorsOnNode(ClientID::FromBinary(node->node_id()));
-  });
-  RAY_LOG(INFO) << "Finished initialing GcsActorManager.";
-}
+GcsActorManager::GcsActorManager(std::shared_ptr<GcsActorSchedulerInterface> scheduler,
+                                 gcs::ActorInfoAccessor &actor_info_accessor)
+    : gcs_actor_scheduler_(std::move(scheduler)),
+      actor_info_accessor_(actor_info_accessor) {}
 
 void GcsActorManager::RegisterActor(
     const ray::rpc::CreateActorRequest &request,
@@ -141,32 +109,24 @@ void GcsActorManager::RegisterActor(
 void GcsActorManager::ReconstructActorOnWorker(const ray::ClientID &node_id,
                                                const ray::WorkerID &worker_id,
                                                bool need_reschedule) {
-  std::shared_ptr<GcsActor> actor;
-  // Cancel the scheduling of the related actor.
-  auto actor_id = gcs_actor_scheduler_->CancelOnWorker(node_id, worker_id);
-  if (!actor_id.IsNil()) {
-    auto iter = registered_actors_.find(actor_id);
-    RAY_CHECK(iter != registered_actors_.end());
-    actor = iter->second;
-  } else {
-    // Find from worker_to_created_actor_.
-    auto iter = worker_to_created_actor_.find(worker_id);
-    if (iter != worker_to_created_actor_.end()) {
-      actor = std::move(iter->second);
-      // Remove the created actor from worker_to_created_actor_.
-      worker_to_created_actor_.erase(iter);
-      // remove the created actor from node_to_created_actors_.
-      auto node_iter = node_to_created_actors_.find(node_id);
-      RAY_CHECK(node_iter != node_to_created_actors_.end());
-      RAY_CHECK(node_iter->second.erase(actor->GetActorID()) != 0);
-      if (node_iter->second.empty()) {
-        node_to_created_actors_.erase(node_iter);
-      }
+  ActorID actor_id;
+  // Find from worker_to_created_actor_.
+  auto iter = created_actors_.find(node_id);
+  if (iter != created_actors_.end() && iter->second.count(worker_id)) {
+    actor_id = iter->second[worker_id];
+    iter->second.erase(worker_id);
+    if (iter->second.empty()) {
+      created_actors_.erase(iter);
     }
+  } else {
+    actor_id = gcs_actor_scheduler_->CancelOnWorker(node_id, worker_id);
   }
-  if (actor != nullptr) {
+
+  if (!actor_id.IsNil()) {
+    RAY_LOG(INFO) << "Worker " << worker_id << " on node " << node_id
+                  << " failed, reconstructing actor " << actor_id;
     // Reconstruct the actor.
-    ReconstructActor(actor, need_reschedule);
+    ReconstructActor(actor_id, need_reschedule);
   }
 }
 
@@ -174,30 +134,25 @@ void GcsActorManager::ReconstructActorsOnNode(const ClientID &node_id) {
   // Cancel the scheduling of all related actors.
   auto scheduling_actor_ids = gcs_actor_scheduler_->CancelOnNode(node_id);
   for (auto &actor_id : scheduling_actor_ids) {
-    auto iter = registered_actors_.find(actor_id);
-    if (iter != registered_actors_.end()) {
-      // Reconstruct the canceled actor.
-      ReconstructActor(iter->second);
-    }
+    // Reconstruct the canceled actor.
+    ReconstructActor(actor_id);
   }
 
   // Find all actors that were created on this node.
-  auto iter = node_to_created_actors_.find(node_id);
-  if (iter != node_to_created_actors_.end()) {
+  auto iter = created_actors_.find(node_id);
+  if (iter != created_actors_.end()) {
     auto created_actors = std::move(iter->second);
     // Remove all created actors from node_to_created_actors_.
-    node_to_created_actors_.erase(iter);
+    created_actors_.erase(iter);
     for (auto &entry : created_actors) {
-      // Remove the actor from worker_to_created_actor_.
-      RAY_CHECK(worker_to_created_actor_.erase(entry.second->GetWorkerID()) != 0);
       // Reconstruct the removed actor.
-      ReconstructActor(std::move(entry.second));
+      ReconstructActor(entry.second);
     }
   }
 }
 
-void GcsActorManager::ReconstructActor(std::shared_ptr<GcsActor> actor,
-                                       bool need_reschedule) {
+void GcsActorManager::ReconstructActor(const ActorID &actor_id, bool need_reschedule) {
+  auto &actor = registered_actors_[actor_id];
   RAY_CHECK(actor != nullptr);
   auto node_id = actor->GetNodeID();
   auto worker_id = actor->GetWorkerID();
@@ -218,11 +173,9 @@ void GcsActorManager::ReconstructActor(std::shared_ptr<GcsActor> actor,
     auto actor_table_data =
         std::make_shared<rpc::ActorTableData>(*mutable_actor_table_data);
     // The backend storage is reliable in the future, so the status must be ok.
-    RAY_CHECK_OK(actor_info_accessor_.AsyncUpdate(actor->GetActorID(), actor_table_data,
-                                                  [this, actor](Status status) {
-                                                    RAY_CHECK_OK(status);
-                                                    gcs_actor_scheduler_->Schedule(actor);
-                                                  }));
+    RAY_CHECK_OK(
+        actor_info_accessor_.AsyncUpdate(actor->GetActorID(), actor_table_data, nullptr));
+    gcs_actor_scheduler_->Schedule(actor);
   } else {
     mutable_actor_table_data->set_state(rpc::ActorTableData::DEAD);
     auto actor_table_data =
@@ -233,16 +186,15 @@ void GcsActorManager::ReconstructActor(std::shared_ptr<GcsActor> actor,
   }
 }
 
-void GcsActorManager::OnActorCreateSuccess(std::shared_ptr<GcsActor> actor) {
-  auto worker_id = actor->GetWorkerID();
-  RAY_CHECK(!worker_id.IsNil());
-  RAY_CHECK(worker_to_created_actor_.emplace(worker_id, actor).second);
+void GcsActorManager::OnActorCreationFailed(std::shared_ptr<GcsActor> actor) {
+  // We will attempt to schedule this actor once an eligible node is
+  // registered.
+  pending_actors_.emplace_back(std::move(actor));
+}
 
+void GcsActorManager::OnActorCreationSuccess(std::shared_ptr<GcsActor> actor) {
   auto actor_id = actor->GetActorID();
-  auto node_id = actor->GetNodeID();
-  RAY_CHECK(!node_id.IsNil());
-  RAY_CHECK(node_to_created_actors_[node_id].emplace(actor_id, actor).second);
-
+  RAY_CHECK(registered_actors_.count(actor_id) > 0);
   actor->UpdateState(rpc::ActorTableData::ALIVE);
   auto actor_table_data =
       std::make_shared<rpc::ActorTableData>(actor->GetActorTableData());
@@ -258,6 +210,12 @@ void GcsActorManager::OnActorCreateSuccess(std::shared_ptr<GcsActor> actor) {
     }
     actor_to_register_callbacks_.erase(iter);
   }
+
+  auto worker_id = actor->GetWorkerID();
+  auto node_id = actor->GetNodeID();
+  RAY_CHECK(!worker_id.IsNil());
+  RAY_CHECK(!node_id.IsNil());
+  RAY_CHECK(created_actors_[node_id].emplace(worker_id, actor_id).second);
 }
 
 void GcsActorManager::SchedulePendingActors() {

src/ray/gcs/gcs_server/gcs_actor_manager.h

@@ -104,23 +104,12 @@ class GcsActorManager {
  public:
   /// Create a GcsActorManager
   ///
-  /// \param io_context The main event loop.
+  /// \param scheduler Used to schedule actor creation tasks.
   /// \param actor_info_accessor Used to flush actor data to storage.
-  /// \param gcs_node_manager The actor manager needs to listen to the node change events
-  /// inside gcs_node_manager.
-  /// \param lease_client_factory Factory to create remote lease client, it will be passed
-  /// through to the constructor of gcs_actor_scheduler, the gcs_actor_scheduler will use
-  /// default factory inside itself if it is not set.
-  /// \param client_factory Factory to create remote core worker client, it will be passed
-  /// through to the constructor of gcs_actor_scheduler, the gcs_actor_scheduler will use
-  /// default factory inside itself if it is not set.
-  explicit GcsActorManager(boost::asio::io_context &io_context,
-                           gcs::ActorInfoAccessor &actor_info_accessor,
-                           gcs::GcsNodeManager &gcs_node_manager,
-                           LeaseClientFactoryFn lease_client_factory = nullptr,
-                           rpc::ClientFactoryFn client_factory = nullptr);
+  GcsActorManager(std::shared_ptr<GcsActorSchedulerInterface> scheduler,
+                  gcs::ActorInfoAccessor &actor_info_accessor);
 
-  virtual ~GcsActorManager() = default;
+  ~GcsActorManager() = default;
 
   /// Register actor asynchronously.
   ///
@@ -131,6 +120,11 @@ class GcsActorManager {
   void RegisterActor(const rpc::CreateActorRequest &request,
                      RegisterActorCallback callback);
 
+  /// Schedule actors in the `pending_actors_` queue.
+  /// This method should be called when new nodes are registered or resources
+  /// change.
+  void SchedulePendingActors();
+
   /// Reconstruct all actors associated with the specified node id, including actors which
   /// are scheduled or have been created on this node. Triggered when the given node goes
   /// down.
@@ -149,43 +143,45 @@ class GcsActorManager {
   void ReconstructActorOnWorker(const ClientID &node_id, const WorkerID &worker_id,
                                 bool need_reschedule = true);
 
- protected:
-  /// Schedule actors in the `pending_actors_` queue.
-  /// This method is triggered when new nodes are registered or resources change.
-  void SchedulePendingActors();
+  /// Handle actor creation task failure. This should be called when scheduling
+  /// an actor creation task is infeasible.
+  ///
+  /// \param actor The actor whose creation task is infeasible.
+  void OnActorCreationFailed(std::shared_ptr<GcsActor> actor);
 
+  /// Handle actor creation task success. This should be called when the actor
+  /// creation task has been scheduled successfully.
+  ///
+  /// \param actor The actor that has been created.
+  void OnActorCreationSuccess(std::shared_ptr<GcsActor> actor);
+
+ private:
   /// Reconstruct the specified actor.
   ///
   /// \param actor The target actor to be reconstructed.
   /// \param need_reschedule Whether to reschedule the actor creation task, sometimes
   /// users want to kill an actor intentionally and don't want it to be reconstructed
   /// again.
-  void ReconstructActor(std::shared_ptr<GcsActor> actor, bool need_reschedule = true);
+  void ReconstructActor(const ActorID &actor_id, bool need_reschedule = true);
 
-  /// This method is a callback of gcs_actor_scheduler when actor is created successfully.
-  /// It will update the state of actor as well as the worker_to_created_actor_ and
-  /// node_to_created_actors_ and flush the actor data to the storage.
-  void OnActorCreateSuccess(std::shared_ptr<GcsActor> actor);
-
- protected:
   /// Callbacks of actor registration requests that are not yet flushed.
   /// This map is used to filter duplicated messages from a Driver/Worker caused by some
   /// network problems.
   absl::flat_hash_map<ActorID, std::vector<RegisterActorCallback>>
       actor_to_register_callbacks_;
   /// All registered actors (pending actors are also included).
+  /// TODO(swang): Use unique_ptr instead of shared_ptr.
   absl::flat_hash_map<ActorID, std::shared_ptr<GcsActor>> registered_actors_;
   /// The pending actors which will not be scheduled until there's a resource change.
   std::vector<std::shared_ptr<GcsActor>> pending_actors_;
-  /// Map contains the relationship of worker and created actor.
-  absl::flat_hash_map<WorkerID, std::shared_ptr<GcsActor>> worker_to_created_actor_;
-  /// Map contains the relationship of node and created actors.
-  absl::flat_hash_map<ClientID, absl::flat_hash_map<ActorID, std::shared_ptr<GcsActor>>>
-      node_to_created_actors_;
-  /// The access info accessor.
-  gcs::ActorInfoAccessor &actor_info_accessor_;
+  /// Map contains the relationship of node and created actors. Each node ID
+  /// maps to a map from worker ID to the actor created on that worker.
+  absl::flat_hash_map<ClientID, absl::flat_hash_map<WorkerID, ActorID>> created_actors_;
+
   /// The scheduler to schedule all registered actors.
-  std::unique_ptr<gcs::GcsActorScheduler> gcs_actor_scheduler_;
+  std::shared_ptr<gcs::GcsActorSchedulerInterface> gcs_actor_scheduler_;
+  /// Actor table. Used to update actor information upon creation, deletion, etc.
+  gcs::ActorInfoAccessor &actor_info_accessor_;
 };
 
 }  // namespace gcs

src/ray/gcs/gcs_server/gcs_actor_scheduler.cc

@@ -28,7 +28,6 @@ GcsActorScheduler::GcsActorScheduler(
     std::function<void(std::shared_ptr<GcsActor>)> schedule_success_handler,
     LeaseClientFactoryFn lease_client_factory, rpc::ClientFactoryFn client_factory)
     : io_context_(io_context),
-      client_call_manager_(io_context_),
       actor_info_accessor_(actor_info_accessor),
       gcs_node_manager_(gcs_node_manager),
       schedule_failure_handler_(std::move(schedule_failure_handler)),
@@ -36,19 +35,6 @@ GcsActorScheduler::GcsActorScheduler(
       lease_client_factory_(std::move(lease_client_factory)),
       client_factory_(std::move(client_factory)) {
   RAY_CHECK(schedule_failure_handler_ != nullptr && schedule_success_handler_ != nullptr);
-  if (lease_client_factory_ == nullptr) {
-    lease_client_factory_ = [this](const rpc::Address &address) {
-      auto node_manager_worker_client = rpc::NodeManagerWorkerClient::make(
-          address.ip_address(), address.port(), client_call_manager_);
-      return std::make_shared<raylet::RayletClient>(
-          std::move(node_manager_worker_client));
-    };
-  }
-  if (client_factory_ == nullptr) {
-    client_factory_ = [this](const rpc::Address &address) {
-      return std::make_shared<rpc::CoreWorkerClient>(address, client_call_manager_);
-    };
-  }
 }
 
 void GcsActorScheduler::Schedule(std::shared_ptr<GcsActor> actor) {

src/ray/gcs/gcs_server/gcs_actor_scheduler.h

@@ -36,9 +36,33 @@ using LeaseClientFactoryFn =
     std::function<std::shared_ptr<WorkerLeaseInterface>(const rpc::Address &address)>;
 
 class GcsActor;
+
+class GcsActorSchedulerInterface {
+ public:
+  /// Schedule the specified actor.
+  ///
+  /// \param actor to be scheduled.
+  virtual void Schedule(std::shared_ptr<GcsActor> actor) = 0;
+
+  /// Cancel all actors that are being scheduled to the specified node.
+  ///
+  /// \param node_id ID of the node where the worker is located.
+  /// \return ID list of actors associated with the specified node id.
+  virtual std::vector<ActorID> CancelOnNode(const ClientID &node_id) = 0;
+
+  /// Cancel the actor that is being scheduled to the specified worker.
+  ///
+  /// \param node_id ID of the node where the worker is located.
+  /// \param worker_id ID of the worker that the actor is creating on.
+  /// \return ID of actor associated with the specified node id and worker id.
+  virtual ActorID CancelOnWorker(const ClientID &node_id, const WorkerID &worker_id) = 0;
+
+  virtual ~GcsActorSchedulerInterface() {}
+};
+
 /// GcsActorScheduler is responsible for scheduling actors registered to GcsActorManager.
 /// This class is not thread-safe.
-class GcsActorScheduler {
+class GcsActorScheduler : public GcsActorSchedulerInterface {
  public:
   /// Create a GcsActorScheduler
   ///
@@ -67,20 +91,20 @@ class GcsActorScheduler {
   /// triggered, otherwise the actor will be scheduled until succeed or canceled.
   ///
   /// \param actor to be scheduled.
-  void Schedule(std::shared_ptr<GcsActor> actor);
+  void Schedule(std::shared_ptr<GcsActor> actor) override;
 
   /// Cancel all actors that are being scheduled to the specified node.
   ///
   /// \param node_id ID of the node where the worker is located.
   /// \return ID list of actors associated with the specified node id.
-  std::vector<ActorID> CancelOnNode(const ClientID &node_id);
+  std::vector<ActorID> CancelOnNode(const ClientID &node_id) override;
 
   /// Cancel the actor that is being scheduled to the specified worker.
   ///
   /// \param node_id ID of the node where the worker is located.
   /// \param worker_id ID of the worker that the actor is creating on.
   /// \return ID of actor associated with the specified node id and worker id.
-  ActorID CancelOnWorker(const ClientID &node_id, const WorkerID &worker_id);
+  ActorID CancelOnWorker(const ClientID &node_id, const WorkerID &worker_id) override;
 
  protected:
   /// The GcsLeasedWorker is kind of abstraction of remote leased worker inside raylet. It
@@ -202,11 +226,9 @@ class GcsActorScheduler {
       const rpc::Address &worker_address);
 
  protected:
-  /// The io loop which is used to construct `client_call_manager_` and delay execution of
-  /// tasks(e.g. execute_after).
+  /// The io loop that is used to delay execution of tasks (e.g.,
+  /// execute_after).
   boost::asio::io_context &io_context_;
-  /// The `ClientCallManager` object that is shared by all `NodeManagerWorkerClient`s.
-  rpc::ClientCallManager client_call_manager_;
   /// The actor info accessor.
   gcs::ActorInfoAccessor &actor_info_accessor_;
   /// Map from node ID to the set of actors for whom we are trying to acquire a lease from

src/ray/gcs/gcs_server/gcs_server.cc

@@ -31,7 +31,8 @@ namespace gcs {
 GcsServer::GcsServer(const ray::gcs::GcsServerConfig &config)
     : config_(config),
       rpc_server_(config.grpc_server_name, config.grpc_server_port,
-                  config.grpc_server_thread_num) {}
+                  config.grpc_server_thread_num),
+      client_call_manager_(main_service_) {}
 
 GcsServer::~GcsServer() { Stop(); }
 
@@ -133,8 +134,55 @@ void GcsServer::InitGcsNodeManager() {
 
 void GcsServer::InitGcsActorManager() {
   RAY_CHECK(redis_gcs_client_ != nullptr && gcs_node_manager_ != nullptr);
-  gcs_actor_manager_ = std::make_shared<GcsActorManager>(
-      main_service_, redis_gcs_client_->Actors(), *gcs_node_manager_);
+  auto scheduler = std::make_shared<GcsActorScheduler>(
+      main_service_, redis_gcs_client_->Actors(), *gcs_node_manager_,
+      /*schedule_failure_handler=*/
+      [this](std::shared_ptr<GcsActor> actor) {
+        // When there are no available nodes to schedule the actor the
+        // gcs_actor_scheduler will treat it as failed and invoke this handler. In
+        // this case, the actor manager should schedule the actor once an
+        // eligible node is registered.
+        gcs_actor_manager_->OnActorCreationFailed(std::move(actor));
+      },
+      /*schedule_success_handler=*/
+      [this](std::shared_ptr<GcsActor> actor) {
+        gcs_actor_manager_->OnActorCreationSuccess(std::move(actor));
+      },
+      /*lease_client_factory=*/
+      [this](const rpc::Address &address) {
+        auto node_manager_worker_client = rpc::NodeManagerWorkerClient::make(
+            address.ip_address(), address.port(), client_call_manager_);
+        return std::make_shared<ray::raylet::RayletClient>(
+            std::move(node_manager_worker_client));
+      },
+      /*client_factory=*/
+      [this](const rpc::Address &address) {
+        return std::make_shared<rpc::CoreWorkerClient>(address, client_call_manager_);
+      });
+  gcs_actor_manager_ =
+      std::make_shared<GcsActorManager>(scheduler, redis_gcs_client_->Actors());
+  gcs_node_manager_->AddNodeAddedListener(
+      [this](const std::shared_ptr<rpc::GcsNodeInfo> &) {
+        // Because a new node has been added, we need to try to schedule the pending
+        // actors.
+        gcs_actor_manager_->SchedulePendingActors();
+      });
+
+  gcs_node_manager_->AddNodeRemovedListener([this](
+                                                std::shared_ptr<rpc::GcsNodeInfo> node) {
+    // All of the related actors should be reconstructed when a node is removed from the
+    // GCS.
+    gcs_actor_manager_->ReconstructActorsOnNode(ClientID::FromBinary(node->node_id()));
+  });
+  RAY_CHECK_OK(redis_gcs_client_->Workers().AsyncSubscribeToWorkerFailures(
+      [this](const WorkerID &id, const rpc::WorkerFailureData &worker_failure_data) {
+        auto &worker_address = worker_failure_data.worker_address();
+        WorkerID worker_id = WorkerID::FromBinary(worker_address.worker_id());
+        ClientID node_id = ClientID::FromBinary(worker_address.raylet_id());
+        auto needs_restart = !worker_failure_data.intentional_disconnect();
+        gcs_actor_manager_->ReconstructActorOnWorker(node_id, worker_id, needs_restart);
+      },
+      /*done_callback=*/nullptr));
 }
 
 std::unique_ptr<rpc::JobInfoHandler> GcsServer::InitJobInfoHandler() {
@@ -181,8 +229,8 @@ std::unique_ptr<rpc::ErrorInfoHandler> GcsServer::InitErrorInfoHandler() {
 }
 
 std::unique_ptr<rpc::WorkerInfoHandler> GcsServer::InitWorkerInfoHandler() {
-  return std::unique_ptr<rpc::DefaultWorkerInfoHandler>(new rpc::DefaultWorkerInfoHandler(
-      *redis_gcs_client_, *gcs_actor_manager_, gcs_pub_sub_));
+  return std::unique_ptr<rpc::DefaultWorkerInfoHandler>(
+      new rpc::DefaultWorkerInfoHandler(*redis_gcs_client_, gcs_pub_sub_));
 }
 
 }  // namespace gcs

src/ray/gcs/gcs_server/gcs_server.h

@@ -17,6 +17,7 @@
 
 #include <ray/gcs/pubsub/gcs_pub_sub.h>
 #include <ray/gcs/redis_gcs_client.h>
+#include <ray/rpc/client_call.h>
 #include <ray/rpc/gcs_server/gcs_rpc_server.h>
 #include "ray/gcs/gcs_server/gcs_redis_failure_detector.h"
 
@@ -112,6 +113,8 @@ class GcsServer {
   rpc::GrpcServer rpc_server_;
   /// The main io service to drive event posted from grpc threads.
   boost::asio::io_context main_service_;
+  /// The `ClientCallManager` object that is shared by all `NodeManagerWorkerClient`s.
+  rpc::ClientCallManager client_call_manager_;
   /// The gcs node manager.
   std::shared_ptr<GcsNodeManager> gcs_node_manager_;
   /// The gcs redis failure detector.

src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc

@@ -20,151 +20,174 @@
 
 namespace ray {
 
-class MockedGcsActorManager : public gcs::GcsActorManager {
+using ::testing::_;
+
+class MockActorScheduler : public gcs::GcsActorSchedulerInterface {
  public:
-  explicit MockedGcsActorManager(boost::asio::io_context &io_context,
-                                 gcs::ActorInfoAccessor &actor_info_accessor,
-                                 gcs::GcsNodeManager &gcs_node_manager,
-                                 gcs::LeaseClientFactoryFn lease_client_factory = nullptr,
-                                 rpc::ClientFactoryFn client_factory = nullptr)
-      : gcs::GcsActorManager(io_context, actor_info_accessor, gcs_node_manager,
-                             lease_client_factory, client_factory) {
-    gcs_actor_scheduler_.reset(new GcsServerMocker::MockedGcsActorScheduler(
-        io_context, actor_info_accessor, gcs_node_manager,
-        /*schedule_failure_handler=*/
-        [this](std::shared_ptr<gcs::GcsActor> actor) {
-          // When there are no available nodes to schedule the actor the
-          // gcs_actor_scheduler will treat it as failed and invoke this handler. In
-          // this case, the actor should be appended to the `pending_actors_` and wait
-          // for the registration of new node.
-          pending_actors_.emplace_back(std::move(actor));
-        },
-        /*schedule_success_handler=*/
-        [this](std::shared_ptr<gcs::GcsActor> actor) {
-          OnActorCreateSuccess(std::move(actor));
-        },
-        std::move(lease_client_factory), std::move(client_factory)));
-  }
+  MockActorScheduler() {}
 
- public:
-  void ResetLeaseClientFactory(gcs::LeaseClientFactoryFn lease_client_factory) {
-    auto gcs_actor_scheduler = dynamic_cast<GcsServerMocker::MockedGcsActorScheduler *>(
-        gcs_actor_scheduler_.get());
-    gcs_actor_scheduler->ResetLeaseClientFactory(std::move(lease_client_factory));
-  }
+  void Schedule(std::shared_ptr<gcs::GcsActor> actor) { actors.push_back(actor); }
 
-  void ResetClientFactory(rpc::ClientFactoryFn client_factory) {
-    auto gcs_actor_scheduler = dynamic_cast<GcsServerMocker::MockedGcsActorScheduler *>(
-        gcs_actor_scheduler_.get());
-    gcs_actor_scheduler->ResetClientFactory(std::move(client_factory));
-  }
+  MOCK_METHOD1(CancelOnNode, std::vector<ActorID>(const ClientID &node_id));
+  MOCK_METHOD2(CancelOnWorker,
+               ActorID(const ClientID &node_id, const WorkerID &worker_id));
 
-  const absl::flat_hash_map<ActorID, std::shared_ptr<gcs::GcsActor>>
-      &GetAllRegisteredActors() const {
-    return registered_actors_;
-  }
-
-  const std::vector<std::shared_ptr<gcs::GcsActor>> &GetAllPendingActors() const {
-    return pending_actors_;
-  }
+  std::vector<std::shared_ptr<gcs::GcsActor>> actors;
 };
 
 class GcsActorManagerTest : public ::testing::Test {
  public:
-  void SetUp() override {
-    raylet_client_ = std::make_shared<GcsServerMocker::MockRayletClient>();
-    worker_client_ = std::make_shared<GcsServerMocker::MockWorkerClient>();
-    gcs_node_manager_ = std::make_shared<gcs::GcsNodeManager>(
-        io_service_, node_info_accessor_, error_info_accessor_);
-    gcs_actor_manager_ = std::make_shared<MockedGcsActorManager>(
-        io_service_, actor_info_accessor_, *gcs_node_manager_,
-        /*lease_client_factory=*/
-        [this](const rpc::Address &address) { return raylet_client_; },
-        /*client_factory=*/
-        [this](const rpc::Address &address) { return worker_client_; });
-  }
+  GcsActorManagerTest()
+      : mock_actor_scheduler_(new MockActorScheduler()),
+        gcs_actor_manager_(mock_actor_scheduler_, actor_info_accessor_) {}
 
- protected:
-  boost::asio::io_service io_service_;
   GcsServerMocker::MockedActorInfoAccessor actor_info_accessor_;
-  GcsServerMocker::MockedNodeInfoAccessor node_info_accessor_;
-  GcsServerMocker::MockedErrorInfoAccessor error_info_accessor_;
-
-  std::shared_ptr<GcsServerMocker::MockRayletClient> raylet_client_;
-  std::shared_ptr<GcsServerMocker::MockWorkerClient> worker_client_;
-  std::shared_ptr<gcs::GcsNodeManager> gcs_node_manager_;
-  std::shared_ptr<MockedGcsActorManager> gcs_actor_manager_;
+  std::shared_ptr<MockActorScheduler> mock_actor_scheduler_;
+  gcs::GcsActorManager gcs_actor_manager_;
 };
 
-TEST_F(GcsActorManagerTest, TestNormalFlow) {
-  gcs_actor_manager_->ResetLeaseClientFactory([this](const rpc::Address &address) {
-    raylet_client_->auto_grant_node_id = ClientID::FromBinary(address.raylet_id());
-    return raylet_client_;
-  });
-  gcs_actor_manager_->ResetClientFactory([this](const rpc::Address &address) {
-    worker_client_->enable_auto_reply = true;
-    return worker_client_;
-  });
-
+TEST_F(GcsActorManagerTest, TestBasic) {
   auto job_id = JobID::FromInt(1);
-  auto create_actor_request =
-      Mocker::GenCreateActorRequest(job_id, /*max_reconstructions=*/2);
+  auto create_actor_request = Mocker::GenCreateActorRequest(job_id);
   std::vector<std::shared_ptr<gcs::GcsActor>> finished_actors;
-  gcs_actor_manager_->RegisterActor(
+  gcs_actor_manager_.RegisterActor(
       create_actor_request, [&finished_actors](std::shared_ptr<gcs::GcsActor> actor) {
         finished_actors.emplace_back(actor);
       });
 
-  ASSERT_EQ(0, finished_actors.size());
-  ASSERT_EQ(1, gcs_actor_manager_->GetAllRegisteredActors().size());
-  ASSERT_EQ(1, gcs_actor_manager_->GetAllPendingActors().size());
+  ASSERT_EQ(finished_actors.size(), 0);
+  ASSERT_EQ(mock_actor_scheduler_->actors.size(), 1);
+  auto actor = mock_actor_scheduler_->actors.back();
+  mock_actor_scheduler_->actors.pop_back();
 
-  auto actor = gcs_actor_manager_->GetAllRegisteredActors().begin()->second;
-  ASSERT_EQ(rpc::ActorTableData::PENDING, actor->GetState());
+  gcs_actor_manager_.OnActorCreationFailed(actor);
+  gcs_actor_manager_.SchedulePendingActors();
+  ASSERT_EQ(mock_actor_scheduler_->actors.size(), 1);
+  mock_actor_scheduler_->actors.clear();
+  ASSERT_EQ(finished_actors.size(), 0);
 
-  // Add node_1 and then check if the actor is in state `ALIVE`
-  auto node_1 = Mocker::GenNodeInfo();
-  auto node_id_1 = ClientID::FromBinary(node_1->node_id());
-  gcs_node_manager_->AddNode(node_1);
-  ASSERT_EQ(1, finished_actors.size());
-  ASSERT_EQ(1, gcs_node_manager_->GetAllAliveNodes().size());
-  ASSERT_EQ(0, gcs_actor_manager_->GetAllPendingActors().size());
-  ASSERT_EQ(rpc::ActorTableData::ALIVE, actor->GetState());
-  ASSERT_EQ(node_id_1, actor->GetNodeID());
+  // Check that the actor is in state `ALIVE`.
+  rpc::Address address;
+  auto node_id = ClientID::FromRandom();
+  auto worker_id = WorkerID::FromRandom();
+  address.set_raylet_id(node_id.Binary());
+  address.set_worker_id(worker_id.Binary());
+  actor->UpdateAddress(address);
+  gcs_actor_manager_.OnActorCreationSuccess(actor);
+  ASSERT_EQ(finished_actors.size(), 1);
 
-  // Remove node_1 and then check if the actor is in state `RECONSTRUCTING`
-  gcs_node_manager_->RemoveNode(node_id_1);
-  ASSERT_EQ(0, gcs_node_manager_->GetAllAliveNodes().size());
-  ASSERT_EQ(1, gcs_actor_manager_->GetAllPendingActors().size());
-  ASSERT_EQ(rpc::ActorTableData::RECONSTRUCTING, actor->GetState());
+  // Killing another worker does not affect this actor.
+  EXPECT_CALL(*mock_actor_scheduler_, CancelOnWorker(node_id, _));
+  gcs_actor_manager_.ReconstructActorOnWorker(node_id, WorkerID::FromRandom());
+  ASSERT_EQ(actor->GetState(), rpc::ActorTableData::ALIVE);
 
-  // Add node_2 and then check if the actor is alive again.
-  auto node_2 = Mocker::GenNodeInfo();
-  auto node_id_2 = ClientID::FromBinary(node_2->node_id());
-  gcs_node_manager_->AddNode(node_2);
-  ASSERT_EQ(1, gcs_node_manager_->GetAllAliveNodes().size());
-  ASSERT_EQ(0, gcs_actor_manager_->GetAllPendingActors().size());
-  ASSERT_EQ(rpc::ActorTableData::ALIVE, actor->GetState());
-  ASSERT_EQ(node_id_2, actor->GetNodeID());
+  // Remove worker and then check that the actor is dead.
+  gcs_actor_manager_.ReconstructActorOnWorker(node_id, worker_id);
+  ASSERT_EQ(actor->GetState(), rpc::ActorTableData::DEAD);
+  // No more actors to schedule.
+  gcs_actor_manager_.SchedulePendingActors();
+  ASSERT_EQ(mock_actor_scheduler_->actors.size(), 0);
+}
 
-  // Add node_3.
-  auto node_3 = Mocker::GenNodeInfo();
-  auto node_id_3 = ClientID::FromBinary(node_3->node_id());
-  gcs_node_manager_->AddNode(node_3);
-  ASSERT_EQ(2, gcs_node_manager_->GetAllAliveNodes().size());
+TEST_F(GcsActorManagerTest, TestBasicNodeFailure) {
+  auto job_id = JobID::FromInt(1);
+  auto create_actor_request = Mocker::GenCreateActorRequest(job_id);
+  std::vector<std::shared_ptr<gcs::GcsActor>> finished_actors;
+  gcs_actor_manager_.RegisterActor(
+      create_actor_request, [&finished_actors](std::shared_ptr<gcs::GcsActor> actor) {
+        finished_actors.emplace_back(actor);
+      });
 
-  // Remove node_2 and then check if the actor drift to node_3.
-  gcs_node_manager_->RemoveNode(node_id_2);
-  ASSERT_EQ(1, gcs_node_manager_->GetAllAliveNodes().size());
-  ASSERT_EQ(0, gcs_actor_manager_->GetAllPendingActors().size());
-  ASSERT_EQ(rpc::ActorTableData::ALIVE, actor->GetState());
-  ASSERT_EQ(node_id_3, actor->GetNodeID());
+  ASSERT_EQ(finished_actors.size(), 0);
+  ASSERT_EQ(mock_actor_scheduler_->actors.size(), 1);
+  auto actor = mock_actor_scheduler_->actors.back();
+  mock_actor_scheduler_->actors.pop_back();
 
-  // Remove node_3 and then check if the actor is dead.
-  gcs_node_manager_->RemoveNode(node_id_3);
-  ASSERT_EQ(0, gcs_node_manager_->GetAllAliveNodes().size());
-  ASSERT_EQ(0, gcs_actor_manager_->GetAllPendingActors().size());
-  ASSERT_EQ(rpc::ActorTableData::DEAD, actor->GetState());
+  gcs_actor_manager_.OnActorCreationFailed(actor);
+  gcs_actor_manager_.SchedulePendingActors();
+  ASSERT_EQ(mock_actor_scheduler_->actors.size(), 1);
+  mock_actor_scheduler_->actors.clear();
+  ASSERT_EQ(finished_actors.size(), 0);
+
+  // Check that the actor is in state `ALIVE`.
+  rpc::Address address;
+  auto node_id = ClientID::FromRandom();
+  auto worker_id = WorkerID::FromRandom();
+  address.set_raylet_id(node_id.Binary());
+  address.set_worker_id(worker_id.Binary());
+  actor->UpdateAddress(address);
+  gcs_actor_manager_.OnActorCreationSuccess(actor);
+  ASSERT_EQ(finished_actors.size(), 1);
+
+  // Killing another node does not affect this actor.
+  EXPECT_CALL(*mock_actor_scheduler_, CancelOnNode(_));
+  gcs_actor_manager_.ReconstructActorsOnNode(ClientID::FromRandom());
+  ASSERT_EQ(actor->GetState(), rpc::ActorTableData::ALIVE);
+
+  // Remove node and then check that the actor is dead.
+  EXPECT_CALL(*mock_actor_scheduler_, CancelOnNode(node_id));
+  gcs_actor_manager_.ReconstructActorsOnNode(node_id);
+  ASSERT_EQ(actor->GetState(), rpc::ActorTableData::DEAD);
+  // No more actors to schedule.
+  gcs_actor_manager_.SchedulePendingActors();
+  ASSERT_EQ(mock_actor_scheduler_->actors.size(), 0);
+}
+
+TEST_F(GcsActorManagerTest, TestActorReconstruction) {
+  auto job_id = JobID::FromInt(1);
+  auto create_actor_request = Mocker::GenCreateActorRequest(
+      job_id, /*max_reconstructions=*/1, /*detached=*/false);
+  std::vector<std::shared_ptr<gcs::GcsActor>> finished_actors;
+  gcs_actor_manager_.RegisterActor(
+      create_actor_request, [&finished_actors](std::shared_ptr<gcs::GcsActor> actor) {
+        finished_actors.emplace_back(actor);
+      });
+
+  ASSERT_EQ(finished_actors.size(), 0);
+  ASSERT_EQ(mock_actor_scheduler_->actors.size(), 1);
+  auto actor = mock_actor_scheduler_->actors.back();
+  mock_actor_scheduler_->actors.pop_back();
+
+  // Check that the actor is in state `ALIVE`.
+  rpc::Address address;
+  auto node_id = ClientID::FromRandom();
+  auto worker_id = WorkerID::FromRandom();
+  address.set_raylet_id(node_id.Binary());
+  address.set_worker_id(worker_id.Binary());
+  actor->UpdateAddress(address);
+  gcs_actor_manager_.OnActorCreationSuccess(actor);
+  ASSERT_EQ(finished_actors.size(), 1);
+
+  // Remove node and then check that the actor is being restarted.
+  EXPECT_CALL(*mock_actor_scheduler_, CancelOnNode(node_id));
+  gcs_actor_manager_.ReconstructActorsOnNode(node_id);
+  ASSERT_EQ(actor->GetState(), rpc::ActorTableData::RECONSTRUCTING);
+
+  // Add node and check that the actor is restarted.
+  gcs_actor_manager_.SchedulePendingActors();
+  ASSERT_EQ(mock_actor_scheduler_->actors.size(), 1);
+  mock_actor_scheduler_->actors.clear();
+  ASSERT_EQ(finished_actors.size(), 1);
+  auto node_id2 = ClientID::FromRandom();
+  address.set_raylet_id(node_id2.Binary());
+  actor->UpdateAddress(address);
+  gcs_actor_manager_.OnActorCreationSuccess(actor);
+  ASSERT_EQ(finished_actors.size(), 1);
+  ASSERT_EQ(actor->GetState(), rpc::ActorTableData::ALIVE);
+  ASSERT_EQ(actor->GetNodeID(), node_id2);
+
+  // Killing another node does not affect this actor.
+  EXPECT_CALL(*mock_actor_scheduler_, CancelOnNode(_));
+  gcs_actor_manager_.ReconstructActorsOnNode(ClientID::FromRandom());
+  ASSERT_EQ(actor->GetState(), rpc::ActorTableData::ALIVE);
+
+  // Remove node and then check that the actor is dead.
+  EXPECT_CALL(*mock_actor_scheduler_, CancelOnNode(node_id2));
+  gcs_actor_manager_.ReconstructActorsOnNode(node_id2);
+  ASSERT_EQ(actor->GetState(), rpc::ActorTableData::DEAD);
+  // No more actors to schedule.
+  gcs_actor_manager_.SchedulePendingActors();
+  ASSERT_EQ(mock_actor_scheduler_->actors.size(), 0);
 }
 
 }  // namespace ray

src/ray/gcs/gcs_server/test/gcs_server_test_util.h

@@ -35,9 +35,6 @@ struct GcsServerMocker {
         std::unique_ptr<rpc::PushTaskRequest> request,
         const rpc::ClientCallback<rpc::PushTaskReply> &callback) override {
       callbacks.push_back(callback);
-      if (enable_auto_reply) {
-        ReplyPushTask();
-      }
       return Status::OK();
     }
 
@@ -55,7 +52,6 @@ struct GcsServerMocker {
       return true;
     }
 
-    bool enable_auto_reply = false;
     std::list<rpc::ClientCallback<rpc::PushTaskReply>> callbacks;
   };
 
@@ -76,10 +72,6 @@ struct GcsServerMocker {
         const rpc::ClientCallback<rpc::RequestWorkerLeaseReply> &callback) override {
       num_workers_requested += 1;
       callbacks.push_back(callback);
-      if (!auto_grant_node_id.IsNil()) {
-        GrantWorkerLease("", 0, WorkerID::FromRandom(), auto_grant_node_id,
-                         ClientID::Nil());
-      }
       return Status::OK();
     }
 
@@ -91,6 +83,10 @@ struct GcsServerMocker {
       return Status::OK();
     }
 
+    bool GrantWorkerLease() {
+      return GrantWorkerLease("", 0, WorkerID::FromRandom(), node_id, ClientID::Nil());
+    }
+
     // Trigger reply to RequestWorkerLease.
     bool GrantWorkerLease(const std::string &address, int port, const WorkerID &worker_id,
                           const ClientID &raylet_id, const ClientID &retry_at_raylet_id,
@@ -136,7 +132,7 @@ struct GcsServerMocker {
     int num_workers_returned = 0;
     int num_workers_disconnected = 0;
     int num_leases_canceled = 0;
-    ClientID auto_grant_node_id;
+    ClientID node_id = ClientID::FromRandom();
     std::list<rpc::ClientCallback<rpc::RequestWorkerLeaseReply>> callbacks = {};
     std::list<rpc::ClientCallback<rpc::CancelWorkerLeaseReply>> cancel_callbacks = {};
   };

src/ray/gcs/gcs_server/worker_info_handler_impl.cc

@@ -20,15 +20,11 @@ namespace rpc {
 void DefaultWorkerInfoHandler::HandleReportWorkerFailure(
     const ReportWorkerFailureRequest &request, ReportWorkerFailureReply *reply,
     SendReplyCallback send_reply_callback) {
-  Address worker_address = request.worker_failure().worker_address();
+  const Address worker_address = request.worker_failure().worker_address();
   RAY_LOG(DEBUG) << "Reporting worker failure, " << worker_address.DebugString();
   auto worker_failure_data = std::make_shared<WorkerFailureData>();
   worker_failure_data->CopyFrom(request.worker_failure());
-  auto need_reschedule = !worker_failure_data->intentional_disconnect();
-  auto node_id = ClientID::FromBinary(worker_address.raylet_id());
-  auto worker_id = WorkerID::FromBinary(worker_address.worker_id());
-  gcs_actor_manager_.ReconstructActorOnWorker(node_id, worker_id, need_reschedule);
-
+  const auto worker_id = WorkerID::FromBinary(worker_address.worker_id());
   auto on_done = [this, worker_address, worker_id, worker_failure_data, reply,
                   send_reply_callback](const Status &status) {
     if (!status.ok()) {

src/ray/gcs/gcs_server/worker_info_handler_impl.h

@@ -15,7 +15,6 @@
 #ifndef RAY_GCS_WORKER_INFO_HANDLER_IMPL_H
 #define RAY_GCS_WORKER_INFO_HANDLER_IMPL_H
 
-#include "gcs_actor_manager.h"
 #include "ray/gcs/pubsub/gcs_pub_sub.h"
 #include "ray/gcs/redis_gcs_client.h"
 #include "ray/rpc/gcs_server/gcs_rpc_server.h"
@@ -27,11 +26,8 @@ namespace rpc {
 class DefaultWorkerInfoHandler : public rpc::WorkerInfoHandler {
  public:
   explicit DefaultWorkerInfoHandler(gcs::RedisGcsClient &gcs_client,
-                                    gcs::GcsActorManager &gcs_actor_manager,
                                     std::shared_ptr<gcs::GcsPubSub> &gcs_pub_sub)
-      : gcs_client_(gcs_client),
-        gcs_actor_manager_(gcs_actor_manager),
-        gcs_pub_sub_(gcs_pub_sub) {}
+      : gcs_client_(gcs_client), gcs_pub_sub_(gcs_pub_sub) {}
 
   void HandleReportWorkerFailure(const ReportWorkerFailureRequest &request,
                                  ReportWorkerFailureReply *reply,
@@ -43,7 +39,6 @@ class DefaultWorkerInfoHandler : public rpc::WorkerInfoHandler {
 
  private:
   gcs::RedisGcsClient &gcs_client_;
-  gcs::GcsActorManager &gcs_actor_manager_;
   std::shared_ptr<gcs::GcsPubSub> gcs_pub_sub_;
 };
 

src/ray/gcs/test/gcs_test_util.h

@@ -17,6 +17,7 @@
 
 #include <memory>
 #include <utility>
+#include "gmock/gmock.h"
 
 #include "src/ray/common/task/task.h"
 #include "src/ray/common/task/task_util.h"
@@ -29,7 +30,8 @@ namespace ray {
 
 struct Mocker {
   static TaskSpecification GenActorCreationTask(const JobID &job_id,
-                                                int max_reconstructions = 100) {
+                                                int max_reconstructions, bool detached,
+                                                const rpc::Address &owner_address) {
     TaskSpecBuilder builder;
     rpc::Address empty_address;
     ray::FunctionDescriptor empty_descriptor =
@@ -37,15 +39,24 @@ struct Mocker {
     auto actor_id = ActorID::Of(job_id, RandomTaskId(), 0);
     auto task_id = TaskID::ForActorCreationTask(actor_id);
     builder.SetCommonTaskSpec(task_id, Language::PYTHON, empty_descriptor, job_id,
-                              TaskID::Nil(), 0, TaskID::Nil(), empty_address, 1, {}, {});
-    builder.SetActorCreationTaskSpec(actor_id, max_reconstructions);
+                              TaskID::Nil(), 0, TaskID::Nil(), owner_address, 1, {}, {});
+    builder.SetActorCreationTaskSpec(actor_id, max_reconstructions, {}, 1, detached);
     return builder.Build();
   }
 
   static rpc::CreateActorRequest GenCreateActorRequest(const JobID &job_id,
-                                                       int max_reconstructions = 100) {
+                                                       int max_reconstructions = 0,
+                                                       bool detached = false) {
     rpc::CreateActorRequest request;
-    auto actor_creation_task_spec = GenActorCreationTask(job_id, max_reconstructions);
+    rpc::Address owner_address;
+    if (owner_address.raylet_id().empty()) {
+      owner_address.set_raylet_id(ClientID::FromRandom().Binary());
+      owner_address.set_ip_address("1234");
+      owner_address.set_port(5678);
+      owner_address.set_worker_id(WorkerID::FromRandom().Binary());
+    }
+    auto actor_creation_task_spec =
+        GenActorCreationTask(job_id, max_reconstructions, detached, owner_address);
     request.mutable_task_spec()->CopyFrom(actor_creation_task_spec.GetMessage());
     return request;
   }