From 8625e09067a0f87b9951b2b41a7c7e0fead66f6e Mon Sep 17 00:00:00 2001 From: Stephanie Wang Date: Mon, 4 May 2020 10:16:52 -0700 Subject: [PATCH] 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 * Fix build * doc * Build Co-authored-by: Edward Oakes --- src/ray/gcs/gcs_server/gcs_actor_manager.cc | 128 +++------ src/ray/gcs/gcs_server/gcs_actor_manager.h | 64 ++--- src/ray/gcs/gcs_server/gcs_actor_scheduler.cc | 14 - src/ray/gcs/gcs_server/gcs_actor_scheduler.h | 38 ++- src/ray/gcs/gcs_server/gcs_server.cc | 58 +++- src/ray/gcs/gcs_server/gcs_server.h | 3 + .../gcs_server/test/gcs_actor_manager_test.cc | 263 ++++++++++-------- .../gcs_server/test/gcs_server_test_util.h | 14 +- .../gcs_server/worker_info_handler_impl.cc | 8 +- .../gcs/gcs_server/worker_info_handler_impl.h | 7 +- src/ray/gcs/test/gcs_test_util.h | 21 +- 11 files changed, 326 insertions(+), 292 deletions(-) diff --git a/src/ray/gcs/gcs_server/gcs_actor_manager.cc b/src/ray/gcs/gcs_server/gcs_actor_manager.cc index f9fa2ae8b..8c7a21e49 100644 --- a/src/ray/gcs/gcs_server/gcs_actor_manager.cc +++ b/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 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 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 &) { - // 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 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 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 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 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 actor, auto actor_table_data = std::make_shared(*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 actor, } } -void GcsActorManager::OnActorCreateSuccess(std::shared_ptr 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 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 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(actor->GetActorTableData()); @@ -258,6 +210,12 @@ void GcsActorManager::OnActorCreateSuccess(std::shared_ptr 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() { diff --git a/src/ray/gcs/gcs_server/gcs_actor_manager.h b/src/ray/gcs/gcs_server/gcs_actor_manager.h index c4a1dc7c6..c16803979 100644 --- a/src/ray/gcs/gcs_server/gcs_actor_manager.h +++ b/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 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 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 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 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 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> 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> registered_actors_; /// The pending actors which will not be scheduled until there's a resource change. std::vector> pending_actors_; - /// Map contains the relationship of worker and created actor. - absl::flat_hash_map> worker_to_created_actor_; - /// Map contains the relationship of node and created actors. - absl::flat_hash_map>> - 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> created_actors_; + /// The scheduler to schedule all registered actors. - std::unique_ptr gcs_actor_scheduler_; + std::shared_ptr gcs_actor_scheduler_; + /// Actor table. Used to update actor information upon creation, deletion, etc. + gcs::ActorInfoAccessor &actor_info_accessor_; }; } // namespace gcs diff --git a/src/ray/gcs/gcs_server/gcs_actor_scheduler.cc b/src/ray/gcs/gcs_server/gcs_actor_scheduler.cc index 9fe965dec..f0c4569ef 100644 --- a/src/ray/gcs/gcs_server/gcs_actor_scheduler.cc +++ b/src/ray/gcs/gcs_server/gcs_actor_scheduler.cc @@ -28,7 +28,6 @@ GcsActorScheduler::GcsActorScheduler( std::function)> 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( - std::move(node_manager_worker_client)); - }; - } - if (client_factory_ == nullptr) { - client_factory_ = [this](const rpc::Address &address) { - return std::make_shared(address, client_call_manager_); - }; - } } void GcsActorScheduler::Schedule(std::shared_ptr actor) { diff --git a/src/ray/gcs/gcs_server/gcs_actor_scheduler.h b/src/ray/gcs/gcs_server/gcs_actor_scheduler.h index 4022a4a8c..1f3efc1fd 100644 --- a/src/ray/gcs/gcs_server/gcs_actor_scheduler.h +++ b/src/ray/gcs/gcs_server/gcs_actor_scheduler.h @@ -36,9 +36,33 @@ using LeaseClientFactoryFn = std::function(const rpc::Address &address)>; class GcsActor; + +class GcsActorSchedulerInterface { + public: + /// Schedule the specified actor. + /// + /// \param actor to be scheduled. + virtual void Schedule(std::shared_ptr 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 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 actor); + void Schedule(std::shared_ptr 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 CancelOnNode(const ClientID &node_id); + std::vector 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 diff --git a/src/ray/gcs/gcs_server/gcs_server.cc b/src/ray/gcs/gcs_server/gcs_server.cc index b26a78f18..b89853a0e 100644 --- a/src/ray/gcs/gcs_server/gcs_server.cc +++ b/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( - main_service_, redis_gcs_client_->Actors(), *gcs_node_manager_); + auto scheduler = std::make_shared( + main_service_, redis_gcs_client_->Actors(), *gcs_node_manager_, + /*schedule_failure_handler=*/ + [this](std::shared_ptr 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 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( + std::move(node_manager_worker_client)); + }, + /*client_factory=*/ + [this](const rpc::Address &address) { + return std::make_shared(address, client_call_manager_); + }); + gcs_actor_manager_ = + std::make_shared(scheduler, redis_gcs_client_->Actors()); + gcs_node_manager_->AddNodeAddedListener( + [this](const std::shared_ptr &) { + // 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 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 GcsServer::InitJobInfoHandler() { @@ -181,8 +229,8 @@ std::unique_ptr GcsServer::InitErrorInfoHandler() { } std::unique_ptr GcsServer::InitWorkerInfoHandler() { - return std::unique_ptr(new rpc::DefaultWorkerInfoHandler( - *redis_gcs_client_, *gcs_actor_manager_, gcs_pub_sub_)); + return std::unique_ptr( + new rpc::DefaultWorkerInfoHandler(*redis_gcs_client_, gcs_pub_sub_)); } } // namespace gcs diff --git a/src/ray/gcs/gcs_server/gcs_server.h b/src/ray/gcs/gcs_server/gcs_server.h index bf27a7f8a..ccbdb904b 100644 --- a/src/ray/gcs/gcs_server/gcs_server.h +++ b/src/ray/gcs/gcs_server/gcs_server.h @@ -17,6 +17,7 @@ #include #include +#include #include #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 gcs_node_manager_; /// The gcs redis failure detector. diff --git a/src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc b/src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc index 48c50741a..8ec69a203 100644 --- a/src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc +++ b/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 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 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( - gcs_actor_scheduler_.get()); - gcs_actor_scheduler->ResetLeaseClientFactory(std::move(lease_client_factory)); - } + void Schedule(std::shared_ptr actor) { actors.push_back(actor); } - void ResetClientFactory(rpc::ClientFactoryFn client_factory) { - auto gcs_actor_scheduler = dynamic_cast( - gcs_actor_scheduler_.get()); - gcs_actor_scheduler->ResetClientFactory(std::move(client_factory)); - } + MOCK_METHOD1(CancelOnNode, std::vector(const ClientID &node_id)); + MOCK_METHOD2(CancelOnWorker, + ActorID(const ClientID &node_id, const WorkerID &worker_id)); - const absl::flat_hash_map> - &GetAllRegisteredActors() const { - return registered_actors_; - } - - const std::vector> &GetAllPendingActors() const { - return pending_actors_; - } + std::vector> actors; }; class GcsActorManagerTest : public ::testing::Test { public: - void SetUp() override { - raylet_client_ = std::make_shared(); - worker_client_ = std::make_shared(); - gcs_node_manager_ = std::make_shared( - io_service_, node_info_accessor_, error_info_accessor_); - gcs_actor_manager_ = std::make_shared( - 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 raylet_client_; - std::shared_ptr worker_client_; - std::shared_ptr gcs_node_manager_; - std::shared_ptr gcs_actor_manager_; + std::shared_ptr 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> finished_actors; - gcs_actor_manager_->RegisterActor( + gcs_actor_manager_.RegisterActor( create_actor_request, [&finished_actors](std::shared_ptr 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> finished_actors; + gcs_actor_manager_.RegisterActor( + create_actor_request, [&finished_actors](std::shared_ptr 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> finished_actors; + gcs_actor_manager_.RegisterActor( + create_actor_request, [&finished_actors](std::shared_ptr 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 diff --git a/src/ray/gcs/gcs_server/test/gcs_server_test_util.h b/src/ray/gcs/gcs_server/test/gcs_server_test_util.h index cb467cfa8..b1ef49365 100644 --- a/src/ray/gcs/gcs_server/test/gcs_server_test_util.h +++ b/src/ray/gcs/gcs_server/test/gcs_server_test_util.h @@ -35,9 +35,6 @@ struct GcsServerMocker { std::unique_ptr request, const rpc::ClientCallback &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> callbacks; }; @@ -76,10 +72,6 @@ struct GcsServerMocker { const rpc::ClientCallback &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> callbacks = {}; std::list> cancel_callbacks = {}; }; diff --git a/src/ray/gcs/gcs_server/worker_info_handler_impl.cc b/src/ray/gcs/gcs_server/worker_info_handler_impl.cc index f7c740450..7a4f13bfd 100644 --- a/src/ray/gcs/gcs_server/worker_info_handler_impl.cc +++ b/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(); 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()) { diff --git a/src/ray/gcs/gcs_server/worker_info_handler_impl.h b/src/ray/gcs/gcs_server/worker_info_handler_impl.h index 1ba87c699..1cf33b217 100644 --- a/src/ray/gcs/gcs_server/worker_info_handler_impl.h +++ b/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_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_pub_sub_; }; diff --git a/src/ray/gcs/test/gcs_test_util.h b/src/ray/gcs/test/gcs_test_util.h index 94f383700..13b0d18d4 100644 --- a/src/ray/gcs/test/gcs_test_util.h +++ b/src/ray/gcs/test/gcs_test_util.h @@ -17,6 +17,7 @@ #include #include +#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; }