From 193d11ab8b8018d56e5b6244a970b765ebd4d2b3 Mon Sep 17 00:00:00 2001 From: fangfengbin <869218239a@zju.edu.cn> Date: Wed, 5 Aug 2020 14:41:32 +0800 Subject: [PATCH] Optimize placement group log (#9891) --- src/ray/common/bundle_spec.cc | 8 + src/ray/common/bundle_spec.h | 2 + src/ray/common/test_util.cc | 8 + src/ray/common/test_util.h | 9 + .../test/service_based_gcs_client_test.cc | 79 ++++---- .../gcs_server/gcs_placement_group_manager.cc | 92 +++++----- .../gcs_server/gcs_placement_group_manager.h | 17 +- .../gcs_placement_group_scheduler.cc | 168 +++++++----------- .../gcs_placement_group_scheduler.h | 13 +- .../test/gcs_placement_group_manager_test.cc | 49 ++--- .../gcs_placement_group_scheduler_test.cc | 49 +++-- 11 files changed, 253 insertions(+), 241 deletions(-) diff --git a/src/ray/common/bundle_spec.cc b/src/ray/common/bundle_spec.cc index 9e83b6a26..388ada1ac 100644 --- a/src/ray/common/bundle_spec.cc +++ b/src/ray/common/bundle_spec.cc @@ -51,6 +51,14 @@ int64_t BundleSpecification::Index() const { return message_->bundle_id().bundle_index(); } +std::string BundleSpecification::DebugString() const { + std::ostringstream stream; + auto bundle_id = BundleId(); + stream << "placement group id={" << bundle_id.first << "}, bundle index={" + << bundle_id.second << "}"; + return stream.str(); +} + std::string FormatPlacementGroupResource(const std::string &original_resource_name, PlacementGroupID group_id, int64_t bundle_index) { diff --git a/src/ray/common/bundle_spec.h b/src/ray/common/bundle_spec.h index 1001af0de..c52681fc6 100644 --- a/src/ray/common/bundle_spec.h +++ b/src/ray/common/bundle_spec.h @@ -78,6 +78,8 @@ class BundleSpecification : public MessageWrapper { /// Returns the spillback bundle callback, or nullptr. const SpillbackBundleCallback &OnSpillback() const { return on_spillback_; } + std::string DebugString() const; + private: void ComputeResources(); diff --git a/src/ray/common/test_util.cc b/src/ray/common/test_util.cc index 717bf07fb..cbd27c656 100644 --- a/src/ray/common/test_util.cc +++ b/src/ray/common/test_util.cc @@ -179,6 +179,14 @@ bool WaitForCondition(std::function condition, int timeout_ms) { return false; } +void WaitForExpectedCount(std::atomic ¤t_count, int expected_count, + int timeout_ms) { + auto condition = [¤t_count, expected_count]() { + return current_count == expected_count; + }; + EXPECT_TRUE(WaitForCondition(condition, timeout_ms)); +} + void KillProcessBySocketName(std::string socket_name) { std::string pidfile_path = socket_name + ".pid"; { diff --git a/src/ray/common/test_util.h b/src/ray/common/test_util.h index 6a05a8ef5..a35afb9a5 100644 --- a/src/ray/common/test_util.h +++ b/src/ray/common/test_util.h @@ -49,6 +49,15 @@ static const int64_t SHOULD_CHECK_MESSAGE_ORDER = 123450000; /// \return Whether the condition is met. bool WaitForCondition(std::function condition, int timeout_ms); +/// Wait until the expected count is met, or timeout is reached. +/// +/// \param[in] current_count The current count. +/// \param[in] expected_count The expected count. +/// \param[in] timeout_ms Timeout in milliseconds to wait for for. +/// \return Whether the expected count is met. +void WaitForExpectedCount(std::atomic ¤t_count, int expected_count, + int timeout_ms = 60000); + /// Used to kill process whose pid is stored in `socket_name.id` file. void KillProcessBySocketName(std::string socket_name); diff --git a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc index dbfcf6ab3..a7f5141a2 100644 --- a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc +++ b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc @@ -483,13 +483,6 @@ class ServiceBasedGcsClientTest : public ::testing::Test { return status == std::future_status::ready && future.get(); } - void WaitPendingDone(std::atomic ¤t_count, int expected_count) { - auto condition = [¤t_count, expected_count]() { - return current_count == expected_count; - }; - EXPECT_TRUE(WaitForCondition(condition, timeout_ms_.count())); - } - void CheckActorData(const gcs::ActorTableData &actor, rpc::ActorTableData_ActorState expected_state) { ASSERT_TRUE(actor.state() == expected_state); @@ -524,7 +517,7 @@ TEST_F(ServiceBasedGcsClientTest, TestJobInfo) { ASSERT_TRUE(AddJob(job_table_data)); ASSERT_TRUE(MarkJobFinished(add_job_id)); - WaitPendingDone(job_updates, 2); + WaitForExpectedCount(job_updates, 2); } TEST_F(ServiceBasedGcsClientTest, TestActorInfo) { @@ -552,7 +545,7 @@ TEST_F(ServiceBasedGcsClientTest, TestActorInfo) { actor_table_data->set_state(rpc::ActorTableData::DEAD); ASSERT_TRUE(UpdateActor(actor_id, actor_table_data)); ASSERT_TRUE(GetActor(actor_id).state() == rpc::ActorTableData::DEAD); - WaitPendingDone(actor_update_count, 1); + WaitForExpectedCount(actor_update_count, 1); } TEST_F(ServiceBasedGcsClientTest, TestActorCheckpoint) { @@ -597,7 +590,7 @@ TEST_F(ServiceBasedGcsClientTest, TestActorSubscribeAll) { // Register an actor to GCS. ASSERT_TRUE(RegisterActor(actor_table_data1)); ASSERT_TRUE(RegisterActor(actor_table_data2)); - WaitPendingDone(actor_update_count, 2); + WaitForExpectedCount(actor_update_count, 2); } TEST_F(ServiceBasedGcsClientTest, TestNodeInfo) { @@ -629,7 +622,7 @@ TEST_F(ServiceBasedGcsClientTest, TestNodeInfo) { auto gcs_node2_info = Mocker::GenNodeInfo(); ClientID node2_id = ClientID::FromBinary(gcs_node2_info->node_id()); ASSERT_TRUE(RegisterNode(*gcs_node2_info)); - WaitPendingDone(register_count, 2); + WaitForExpectedCount(register_count, 2); // Get information of all nodes from GCS. std::vector node_list = GetNodeInfoList(); @@ -642,7 +635,7 @@ TEST_F(ServiceBasedGcsClientTest, TestNodeInfo) { // Cancel registration of a node to GCS. ASSERT_TRUE(UnregisterNode(node2_id)); - WaitPendingDone(unregister_count, 2); + WaitForExpectedCount(unregister_count, 2); // Get information of all nodes from GCS. node_list = GetNodeInfoList(); @@ -676,12 +669,12 @@ TEST_F(ServiceBasedGcsClientTest, TestNodeResources) { ClientID node_id = ClientID::FromBinary(node_info->node_id()); std::string key = "CPU"; ASSERT_TRUE(UpdateResources(node_id, key)); - WaitPendingDone(add_count, 1); + WaitForExpectedCount(add_count, 1); ASSERT_TRUE(GetResources(node_id).count(key)); // Delete resources of a node from GCS. ASSERT_TRUE(DeleteResources(node_id, {key})); - WaitPendingDone(remove_count, 1); + WaitForExpectedCount(remove_count, 1); ASSERT_TRUE(GetResources(node_id).empty()); } @@ -705,7 +698,7 @@ TEST_F(ServiceBasedGcsClientTest, TestNodeHeartbeat) { // Set this flag because GCS won't publish unchanged heartbeat. heartbeat->set_should_global_gc(true); ASSERT_TRUE(ReportHeartbeat(heartbeat)); - WaitPendingDone(heartbeat_batch_count, 1); + WaitForExpectedCount(heartbeat_batch_count, 1); } TEST_F(ServiceBasedGcsClientTest, TestTaskInfo) { @@ -752,7 +745,7 @@ TEST_F(ServiceBasedGcsClientTest, TestTaskInfo) { ClientID node_id = ClientID::FromRandom(); auto task_lease = Mocker::GenTaskLeaseData(task_id.Binary(), node_id.Binary()); ASSERT_TRUE(AddTaskLease(task_lease)); - WaitPendingDone(task_lease_count, 2); + WaitForExpectedCount(task_lease_count, 2); // Cancel subscription to a task lease. UnsubscribeTaskLease(task_id); @@ -793,7 +786,7 @@ TEST_F(ServiceBasedGcsClientTest, TestObjectInfo) { // Add location of object to GCS. ASSERT_TRUE(AddLocation(object_id, node_id)); - WaitPendingDone(object_add_count, 1); + WaitForExpectedCount(object_add_count, 1); // Get object's locations from GCS. auto locations = GetLocations(object_id); @@ -802,7 +795,7 @@ TEST_F(ServiceBasedGcsClientTest, TestObjectInfo) { // Remove location of object from GCS. ASSERT_TRUE(RemoveLocation(object_id, node_id)); - WaitPendingDone(object_remove_count, 1); + WaitForExpectedCount(object_remove_count, 1); ASSERT_TRUE(GetLocations(object_id).empty()); // Cancel subscription to any update of an object's location. @@ -836,14 +829,14 @@ TEST_F(ServiceBasedGcsClientTest, TestWorkerInfo) { auto worker_data = Mocker::GenWorkerTableData(); worker_data->mutable_worker_address()->set_worker_id(WorkerID::FromRandom().Binary()); ASSERT_TRUE(ReportWorkerFailure(worker_data)); - WaitPendingDone(worker_failure_count, 0); + WaitForExpectedCount(worker_failure_count, 0); // Add a worker to GCS. ASSERT_TRUE(AddWorker(worker_data)); // Report a worker failure to GCS when this worker is actually exist. ASSERT_TRUE(ReportWorkerFailure(worker_data)); - WaitPendingDone(worker_failure_count, 1); + WaitForExpectedCount(worker_failure_count, 1); } TEST_F(ServiceBasedGcsClientTest, TestErrorInfo) { @@ -866,15 +859,15 @@ TEST_F(ServiceBasedGcsClientTest, TestJobTableResubscribe) { ASSERT_TRUE(SubscribeToAllJobs(subscribe)); ASSERT_TRUE(AddJob(job_table_data)); - WaitPendingDone(job_update_count, 1); + WaitForExpectedCount(job_update_count, 1); RestartGcsServer(); // The GCS client will fetch data from the GCS server after the GCS server is restarted, // and the GCS server keeps a job record, so `job_update_count` plus one. - WaitPendingDone(job_update_count, 2); + WaitForExpectedCount(job_update_count, 2); ASSERT_TRUE(MarkJobFinished(job_id)); - WaitPendingDone(job_update_count, 3); + WaitForExpectedCount(job_update_count, 3); } TEST_F(ServiceBasedGcsClientTest, TestActorTableResubscribe) { @@ -910,8 +903,8 @@ TEST_F(ServiceBasedGcsClientTest, TestActorTableResubscribe) { ASSERT_TRUE(RegisterActor(actor_table_data)); // We should receive a new ALIVE notification from the subscribe channel. - WaitPendingDone(num_subscribe_all_notifications, 1); - WaitPendingDone(num_subscribe_one_notifications, 1); + WaitForExpectedCount(num_subscribe_all_notifications, 1); + WaitForExpectedCount(num_subscribe_one_notifications, 1); CheckActorData(subscribe_all_notifications[0], rpc::ActorTableData::ALIVE); CheckActorData(subscribe_one_notifications[0], rpc::ActorTableData::ALIVE); @@ -921,8 +914,8 @@ TEST_F(ServiceBasedGcsClientTest, TestActorTableResubscribe) { // When GCS client detects that GCS server has restarted, but the pub-sub server // didn't restart, it will fetch data again from the GCS server. So we'll receive // another notification of ALIVE state. - WaitPendingDone(num_subscribe_all_notifications, 2); - WaitPendingDone(num_subscribe_one_notifications, 2); + WaitForExpectedCount(num_subscribe_all_notifications, 2); + WaitForExpectedCount(num_subscribe_one_notifications, 2); CheckActorData(subscribe_all_notifications[1], rpc::ActorTableData::ALIVE); CheckActorData(subscribe_one_notifications[1], rpc::ActorTableData::ALIVE); @@ -931,8 +924,8 @@ TEST_F(ServiceBasedGcsClientTest, TestActorTableResubscribe) { ASSERT_TRUE(UpdateActor(actor_id, actor_table_data)); // We should receive a new DEAD notification from the subscribe channel. - WaitPendingDone(num_subscribe_all_notifications, 3); - WaitPendingDone(num_subscribe_one_notifications, 3); + WaitForExpectedCount(num_subscribe_all_notifications, 3); + WaitForExpectedCount(num_subscribe_one_notifications, 3); CheckActorData(subscribe_all_notifications[2], rpc::ActorTableData::DEAD); CheckActorData(subscribe_one_notifications[2], rpc::ActorTableData::DEAD); } @@ -961,9 +954,9 @@ TEST_F(ServiceBasedGcsClientTest, TestObjectTableResubscribe) { })); ASSERT_TRUE(AddLocation(object1_id, node_id)); - WaitPendingDone(object1_change_count, 1); + WaitForExpectedCount(object1_change_count, 1); ASSERT_TRUE(AddLocation(object2_id, node_id)); - WaitPendingDone(object2_change_count, 1); + WaitForExpectedCount(object2_change_count, 1); // Cancel subscription to any update of an object's location. UnsubscribeToLocations(object1_id); @@ -974,13 +967,13 @@ TEST_F(ServiceBasedGcsClientTest, TestObjectTableResubscribe) { // When GCS client detects that GCS server has restarted, but the pub-sub server // didn't restart, it will fetch the subscription data again from the GCS server, so // `object2_change_count` plus 1. - WaitPendingDone(object2_change_count, 2); + WaitForExpectedCount(object2_change_count, 2); // Add location of object to GCS again and check if resubscribe works. ASSERT_TRUE(AddLocation(object1_id, node_id)); - WaitPendingDone(object1_change_count, 1); + WaitForExpectedCount(object1_change_count, 1); ASSERT_TRUE(AddLocation(object2_id, node_id)); - WaitPendingDone(object2_change_count, 3); + WaitForExpectedCount(object2_change_count, 3); } TEST_F(ServiceBasedGcsClientTest, TestNodeTableResubscribe) { @@ -1019,7 +1012,7 @@ TEST_F(ServiceBasedGcsClientTest, TestNodeTableResubscribe) { // Set this flag because GCS won't publish unchanged heartbeat. heartbeat->set_should_global_gc(true); ASSERT_TRUE(ReportHeartbeat(heartbeat)); - WaitPendingDone(batch_heartbeat_count, 1); + WaitForExpectedCount(batch_heartbeat_count, 1); RestartGcsServer(); @@ -1030,9 +1023,9 @@ TEST_F(ServiceBasedGcsClientTest, TestNodeTableResubscribe) { heartbeat->set_client_id(node_info->node_id()); ASSERT_TRUE(ReportHeartbeat(heartbeat)); - WaitPendingDone(node_change_count, 2); - WaitPendingDone(resource_change_count, 2); - WaitPendingDone(batch_heartbeat_count, 2); + WaitForExpectedCount(node_change_count, 2); + WaitForExpectedCount(resource_change_count, 2); + WaitForExpectedCount(batch_heartbeat_count, 2); } TEST_F(ServiceBasedGcsClientTest, TestTaskTableResubscribe) { @@ -1061,8 +1054,8 @@ TEST_F(ServiceBasedGcsClientTest, TestTaskTableResubscribe) { ClientID node_id = ClientID::FromRandom(); auto task_lease = Mocker::GenTaskLeaseData(task_id.Binary(), node_id.Binary()); ASSERT_TRUE(AddTaskLease(task_lease)); - WaitPendingDone(task_count, 1); - WaitPendingDone(task_lease_count, 1); + WaitForExpectedCount(task_count, 1); + WaitForExpectedCount(task_lease_count, 1); UnsubscribeTask(task_id); RestartGcsServer(); @@ -1070,8 +1063,8 @@ TEST_F(ServiceBasedGcsClientTest, TestTaskTableResubscribe) { node_id = ClientID::FromRandom(); task_lease = Mocker::GenTaskLeaseData(task_id.Binary(), node_id.Binary()); ASSERT_TRUE(AddTaskLease(task_lease)); - WaitPendingDone(task_lease_count, 3); - WaitPendingDone(task_count, 1); + WaitForExpectedCount(task_lease_count, 3); + WaitForExpectedCount(task_count, 1); } TEST_F(ServiceBasedGcsClientTest, TestWorkerTableResubscribe) { @@ -1093,7 +1086,7 @@ TEST_F(ServiceBasedGcsClientTest, TestWorkerTableResubscribe) { // Report a worker failure to GCS and check if resubscribe works. ASSERT_TRUE(ReportWorkerFailure(worker_data)); - WaitPendingDone(worker_failure_count, 1); + WaitForExpectedCount(worker_failure_count, 1); } TEST_F(ServiceBasedGcsClientTest, TestGcsTableReload) { diff --git a/src/ray/gcs/gcs_server/gcs_placement_group_manager.cc b/src/ray/gcs/gcs_server/gcs_placement_group_manager.cc index 4110d98e6..18315fd44 100644 --- a/src/ray/gcs/gcs_server/gcs_placement_group_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_placement_group_manager.cc @@ -41,8 +41,8 @@ std::string GcsPlacementGroup::GetName() const { std::vector> GcsPlacementGroup::GetBundles() const { auto bundles = placement_group_table_data_.bundles(); std::vector> ret_bundles; - for (auto iter = bundles.begin(); iter != bundles.end(); iter++) { - ret_bundles.push_back(std::make_shared(*iter)); + for (auto &bundle : bundles) { + ret_bundles.push_back(std::make_shared(bundle)); } return ret_bundles; } @@ -62,25 +62,25 @@ GcsPlacementGroupManager::GcsPlacementGroupManager( std::shared_ptr scheduler, std::shared_ptr gcs_table_storage) : gcs_placement_group_scheduler_(std::move(scheduler)), - gcs_table_storage_(gcs_table_storage), + gcs_table_storage_(std::move(gcs_table_storage)), reschedule_timer_(io_context) {} void GcsPlacementGroupManager::RegisterPlacementGroup( - const ray::rpc::CreatePlacementGroupRequest &request, - std::function)> callback) { + const ray::rpc::CreatePlacementGroupRequest &request, EmptyCallback callback) { RAY_CHECK(callback); const auto &placement_group_spec = request.placement_group_spec(); auto placement_group_id = PlacementGroupID::FromBinary(placement_group_spec.placement_group_id()); - auto placement_group = std::make_shared(request); + + // TODO(ffbin): If GCS is restarted, GCS client will repeatedly send + // `CreatePlacementGroup` requests, + // which will lead to resource leakage, we will solve it in next pr. // Mark the callback as pending and invoke it after the placement_group has been // successfully created. - placement_group_to_register_callbacks_[placement_group_id].emplace_back( - std::move(callback)); - RAY_CHECK(registered_placement_groups_ - .emplace(placement_group->GetPlacementGroupID(), placement_group) - .second); + placement_group_to_register_callback_[placement_group_id] = std::move(callback); + registered_placement_groups_.emplace(placement_group->GetPlacementGroupID(), + placement_group); pending_placement_groups_.emplace_back(std::move(placement_group)); SchedulePendingPlacementGroups(); } @@ -88,10 +88,9 @@ void GcsPlacementGroupManager::RegisterPlacementGroup( PlacementGroupID GcsPlacementGroupManager::GetPlacementGroupIDByName( const std::string &name) { PlacementGroupID placement_group_id = PlacementGroupID::Nil(); - for (auto placement_group_pair : registered_placement_groups_) { - auto placement_group = placement_group_pair.second; - if (placement_group->GetName() == name) { - placement_group_id = placement_group_pair.first; + for (const auto &iter : registered_placement_groups_) { + if (iter.second->GetName() == name) { + placement_group_id = iter.first; break; } } @@ -100,6 +99,8 @@ PlacementGroupID GcsPlacementGroupManager::GetPlacementGroupIDByName( void GcsPlacementGroupManager::OnPlacementGroupCreationFailed( std::shared_ptr placement_group) { + RAY_LOG(WARNING) << "Failed to create placement group " << placement_group->GetName() + << ", try again."; // We will attempt to schedule this placement_group once an eligible node is // registered. pending_placement_groups_.emplace_back(std::move(placement_group)); @@ -109,27 +110,24 @@ void GcsPlacementGroupManager::OnPlacementGroupCreationFailed( void GcsPlacementGroupManager::OnPlacementGroupCreationSuccess( std::shared_ptr placement_group) { - auto placement_group_id = placement_group->GetPlacementGroupID(); - RAY_CHECK(registered_placement_groups_.count(placement_group_id) > 0); + RAY_LOG(INFO) << "Successfully created placement group " << placement_group->GetName(); placement_group->UpdateState(rpc::PlacementGroupTableData::ALIVE); - - auto placement_group_table_data = placement_group->GetPlacementGroupTableData(); - // The backend storage is reliable in the future, so the status must be ok. + auto placement_group_id = placement_group->GetPlacementGroupID(); RAY_CHECK_OK(gcs_table_storage_->PlacementGroupTable().Put( - placement_group_id, placement_group_table_data, [](Status status) {})); + placement_group_id, placement_group->GetPlacementGroupTableData(), + [this, placement_group_id](Status status) { + RAY_CHECK_OK(status); - // Invoke all callbacks for all registration requests of this placement_group - // (duplicated requests are included) and remove all of them from - // placement_group_to_register_callbacks_. - auto iter = placement_group_to_register_callbacks_.find(placement_group_id); - if (iter != placement_group_to_register_callbacks_.end()) { - for (auto &callback : iter->second) { - callback(placement_group); - } - placement_group_to_register_callbacks_.erase(iter); - } - is_creating_ = false; - SchedulePendingPlacementGroups(); + // Invoke callback for registration request of this placement_group + // and remove it from placement_group_to_register_callback_. + auto iter = placement_group_to_register_callback_.find(placement_group_id); + if (iter != placement_group_to_register_callback_.end()) { + iter->second(); + placement_group_to_register_callback_.erase(iter); + } + is_creating_ = false; + SchedulePendingPlacementGroups(); + })); } void GcsPlacementGroupManager::SchedulePendingPlacementGroups() { @@ -138,7 +136,7 @@ void GcsPlacementGroupManager::SchedulePendingPlacementGroups() { } is_creating_ = true; gcs_placement_group_scheduler_->Schedule( - *pending_placement_groups_.begin(), + pending_placement_groups_.front(), [this](std::shared_ptr placement_group) { OnPlacementGroupCreationFailed(std::move(placement_group)); }, @@ -154,29 +152,31 @@ void GcsPlacementGroupManager::HandleCreatePlacementGroup( ray::rpc::SendReplyCallback send_reply_callback) { auto placement_group_id = PlacementGroupID::FromBinary(request.placement_group_spec().placement_group_id()); - const auto &strategy = request.placement_group_spec().strategy(); const auto &name = request.placement_group_spec().name(); + const auto &strategy = request.placement_group_spec().strategy(); RAY_LOG(INFO) << "Registering placement group, placement group id = " << placement_group_id << ", name = " << name << ", strategy = " << PlacementStrategy_Name(strategy); - RegisterPlacementGroup( - request, [reply, send_reply_callback, placement_group_id]( - std::shared_ptr placement_group) { - RAY_LOG(INFO) << "Registered placement group, placement group id = " - << placement_group_id << ", name = " << placement_group->GetName() - << ", strategy = " << placement_group->GetStrategy(); - GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK()); - }); auto placement_group = std::make_shared(request); - auto placement_group_table_data = placement_group->GetPlacementGroupTableData(); RAY_CHECK_OK(gcs_table_storage_->PlacementGroupTable().Put( - placement_group_id, placement_group_table_data, [](Status status) {})); + placement_group_id, placement_group->GetPlacementGroupTableData(), + [this, request, reply, send_reply_callback, placement_group_id, name, + strategy](Status status) { + RAY_CHECK_OK(status); + RegisterPlacementGroup(request, [reply, send_reply_callback, placement_group_id, + name, strategy]() { + RAY_LOG(INFO) << "Finished registering placement group, placement group id = " + << placement_group_id << ", name = " << name + << ", strategy = " << strategy; + GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK()); + }); + })); } void GcsPlacementGroupManager::ScheduleTick() { reschedule_timer_.expires_from_now(boost::posix_time::milliseconds(500)); reschedule_timer_.async_wait([this](const boost::system::error_code &error) { - if (error == boost::system::errc::operation_canceled) { + if (error == boost::asio::error::operation_aborted) { return; } else { SchedulePendingPlacementGroups(); diff --git a/src/ray/gcs/gcs_server/gcs_placement_group_manager.h b/src/ray/gcs/gcs_server/gcs_placement_group_manager.h index 3a1cf0f17..a64e78006 100644 --- a/src/ray/gcs/gcs_server/gcs_placement_group_manager.h +++ b/src/ray/gcs/gcs_server/gcs_placement_group_manager.h @@ -81,6 +81,7 @@ class GcsPlacementGroup { using RegisterPlacementGroupCallback = std::function)>; + /// GcsPlacementGroupManager is responsible for managing the lifecycle of all placement /// group. This class is not thread-safe. /// The placementGroup will be added into queue and set the status as pending first and @@ -114,7 +115,7 @@ class GcsPlacementGroupManager : public rpc::PlacementGroupInfoHandler { /// `registered_placement_groups_` and its state is `ALIVE`. The callback will not be /// called in this case. void RegisterPlacementGroup(const rpc::CreatePlacementGroupRequest &request, - RegisterPlacementGroupCallback callback); + EmptyCallback callback); /// Schedule placement_groups in the `pending_placement_groups_` queue. /// This function is exposed for testing only. @@ -144,17 +145,9 @@ class GcsPlacementGroupManager : public rpc::PlacementGroupInfoHandler { /// Schedule another tick after a short time. void ScheduleTick(); - /// Callbacks of placement_group registration requests that are not yet flushed. - /// This map is used to filter duplicated messages from a Driver/Worker caused by some - /// network problems. - /// - /// Since the GRPC message received by the GCS side is out of order, it can not be - /// determined that the last callback is the valid one. Therefore, the repeated - /// callbacks are recorded in the form of vector without distinction. When the operation - /// is successful, all callbacks will be triggered. One of them must be valid, and the - /// rest invalid callbacks will not have any effect even if they are called. - absl::flat_hash_map> - placement_group_to_register_callbacks_; + /// Callback of placement_group registration requests that are not yet flushed. + absl::flat_hash_map + placement_group_to_register_callback_; /// All registered placement_groups (pending placement_groups are also included). absl::flat_hash_map> registered_placement_groups_; diff --git a/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc b/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc index 76cf87b72..47ae66395 100644 --- a/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc +++ b/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc @@ -26,7 +26,7 @@ GcsPlacementGroupScheduler::GcsPlacementGroupScheduler( const gcs::GcsNodeManager &gcs_node_manager, ReserveResourceClientFactoryFn lease_client_factory) : return_timer_(io_context), - gcs_table_storage_(gcs_table_storage), + gcs_table_storage_(std::move(gcs_table_storage)), gcs_node_manager_(gcs_node_manager), lease_client_factory_(std::move(lease_client_factory)) { scheduler_strategies_.push_back(std::make_shared()); @@ -76,73 +76,59 @@ void GcsPlacementGroupScheduler::Schedule( std::shared_ptr placement_group, std::function)> schedule_failure_handler, std::function)> schedule_success_handler) { + RAY_LOG(INFO) << "Scheduling placement group " << placement_group->GetName(); auto bundles = placement_group->GetBundles(); auto strategy = placement_group->GetStrategy(); - auto alive_nodes = gcs_node_manager_.GetAllAliveNodes(); - /// If the placement group don't have bundle, the placement group creates success. - if (bundles.empty()) { - schedule_success_handler(placement_group); - return; - } + auto selected_nodes = + scheduler_strategies_[strategy]->Schedule(bundles, gcs_node_manager_); - // If alive_node is empty, the the placement group creates fail. - if (alive_nodes.empty()) { + // If no nodes are available, scheduling fails. + if (selected_nodes.empty()) { + RAY_LOG(WARNING) << "Failed to schedule placement group " + << placement_group->GetName() << ", because no nodes are available."; schedule_failure_handler(placement_group); return; } - auto schedule_map = - scheduler_strategies_[strategy]->Schedule(bundles, gcs_node_manager_); + // If schedule success, the decision will be set as schedule_map[bundles[pos]] // else will be set ClientID::Nil(). - auto decision = std::shared_ptr>( - new std::unordered_map()); + auto bundle_locations = std::make_shared>, pair_hash>>(); // To count how many scheduler have been return, which include success and failure. - auto finish_count = std::make_shared(); + auto finished_count = std::make_shared(); /// TODO(AlisaWu): Change the strategy when reserve resource failed. - for (size_t pos = 0; pos < bundles.size(); pos++) { - RAY_CHECK(node_to_bundles_when_leasing_[schedule_map.at(bundles[pos]->BundleId())] - .emplace(bundles[pos]->BundleId()) - .second); + for (auto &bundle : bundles) { + const auto &bundle_id = bundle->BundleId(); + const auto &node_id = selected_nodes[bundle_id]; + RAY_CHECK(node_to_bundles_when_leasing_[node_id].emplace(bundle_id).second); ReserveResourceFromNode( - bundles[pos], - gcs_node_manager_.GetNode(schedule_map.at(bundles[pos]->BundleId())), - [this, pos, bundles, schedule_map, placement_group, decision, finish_count, - schedule_failure_handler, schedule_success_handler]( - const Status &status, const rpc::RequestResourceReserveReply &reply) { - (*finish_count)++; - if (status.ok() && reply.success()) { - (*decision)[bundles[pos]->BundleId()] = - schedule_map.at(bundles[pos]->BundleId()); - } else { - (*decision)[bundles[pos]->BundleId()] = ClientID::Nil(); + bundle, gcs_node_manager_.GetNode(node_id), + [this, bundle_id, bundle, bundles, node_id, placement_group, bundle_locations, + finished_count, schedule_failure_handler, + schedule_success_handler](const Status &status) { + if (status.ok()) { + (*bundle_locations)[bundle_id] = std::make_pair(node_id, bundle); } - if ((*finish_count) == bundles.size()) { - bool lease_success = true; - for (size_t i = 0; i < (*finish_count); i++) { - if ((*decision)[bundles[i]->BundleId()] == ClientID::Nil()) { - lease_success = false; - break; - } - } - if (lease_success) { + + if (++(*finished_count) == bundles.size()) { + if (bundle_locations->size() == bundles.size()) { rpc::ScheduleData data; - for (size_t i = 0; i < bundles.size(); i++) { + for (const auto &iter : bundles) { // TODO(ekl) this is a hack to get a string key for the proto - auto key = bundles[i]->PlacementGroupId().Hex() + "_" + - std::to_string(bundles[i]->Index()); + auto key = + iter->PlacementGroupId().Hex() + "_" + std::to_string(iter->Index()); data.mutable_schedule_plan()->insert( - {key, (*decision)[bundles[i]->BundleId()].Binary()}); + {key, (*bundle_locations)[iter->BundleId()].first.Binary()}); } RAY_CHECK_OK(gcs_table_storage_->PlacementGroupScheduleTable().Put( - placement_group->GetPlacementGroupID(), data, [](Status status) {})); - schedule_success_handler(placement_group); + placement_group->GetPlacementGroupID(), data, + [schedule_success_handler, placement_group](Status status) { + schedule_success_handler(placement_group); + })); } else { - for (size_t i = 0; i < (*finish_count); i++) { - if ((*decision)[bundles[i]->BundleId()] != ClientID::Nil()) { - CancelResourceReserve(bundles[i], - gcs_node_manager_.GetNode( - schedule_map.at(bundles[pos]->BundleId()))); - } + for (auto &iter : *bundle_locations) { + CancelResourceReserve(iter.second.second, + gcs_node_manager_.GetNode(node_id)); } schedule_failure_handler(placement_group); } @@ -152,77 +138,59 @@ void GcsPlacementGroupScheduler::Schedule( } void GcsPlacementGroupScheduler::ReserveResourceFromNode( - std::shared_ptr bundle, - std::shared_ptr node, ReserveResourceCallback callback) { - RAY_CHECK(node); - + const std::shared_ptr &bundle, + const std::shared_ptr &node, const StatusCallback &callback) { rpc::Address remote_address; remote_address.set_raylet_id(node->node_id()); remote_address.set_ip_address(node->node_manager_address()); remote_address.set_port(node->node_manager_port()); auto node_id = ClientID::FromBinary(node->node_id()); auto lease_client = GetOrConnectLeaseClient(remote_address); - RAY_LOG(DEBUG) << "Start leasing resource from node " << node_id << " for bundle " - << bundle->BundleId().first << bundle->BundleId().second; + RAY_LOG(INFO) << "Leasing resource from node " << node_id + << " for bundle: " << bundle->DebugString(); lease_client->RequestResourceReserve( - *bundle, [this, node_id, bundle, node, callback]( + *bundle, [this, node_id, bundle, callback]( const Status &status, const rpc::RequestResourceReserveReply &reply) { // TODO(AlisaWu): Add placement group cancel. - auto iter = node_to_bundles_when_leasing_.find(node_id); - if (iter != node_to_bundles_when_leasing_.end()) { - auto bundle_iter = iter->second.find(bundle->BundleId()); - RAY_CHECK(bundle_iter != iter->second.end()); - if (status.ok()) { - if (reply.success()) { - RAY_LOG(DEBUG) << "Finished leasing resource from " << node_id - << " for bundle " << bundle->BundleId().first - << bundle->BundleId().second; - } else { - RAY_LOG(DEBUG) << "Failed leasing resource from " << node_id - << " for bundle " << bundle->BundleId().first - << bundle->BundleId().second; - } - // Remove the bundle from the leasing map as the reply is returned from the - // remote node. - iter->second.erase(bundle_iter); - if (iter->second.empty()) { - node_to_bundles_when_leasing_.erase(iter); - } - callback(status, reply); - } + auto result = reply.success() ? Status::OK() + : Status::IOError("Failed to reserve resource"); + auto bundles = node_to_bundles_when_leasing_.find(node_id); + RAY_CHECK(bundles != node_to_bundles_when_leasing_.end()); + auto bundle_iter = bundles->second.find(bundle->BundleId()); + RAY_CHECK(bundle_iter != bundles->second.end()); + if (result.ok()) { + RAY_LOG(INFO) << "Finished leasing resource from " << node_id + << " for bundle: " << bundle->DebugString(); + } else { + RAY_LOG(WARNING) << "Failed to lease resource from " << node_id + << " for bundle: " << bundle->DebugString(); } + // Remove the bundle from the leasing map as the reply is returned from the + // remote node. + bundles->second.erase(bundle_iter); + if (bundles->second.empty()) { + node_to_bundles_when_leasing_.erase(bundles); + } + callback(result); }); } void GcsPlacementGroupScheduler::CancelResourceReserve( - std::shared_ptr bundle_spec, - std::shared_ptr node) { - RAY_CHECK(node); - + const std::shared_ptr &bundle_spec, + const std::shared_ptr &node) { auto node_id = ClientID::FromBinary(node->node_id()); - RAY_LOG(DEBUG) << "Start returning resource for node " << node_id << " for bundle " - << bundle_spec->BundleId().first << bundle_spec->BundleId().second; - + RAY_LOG(INFO) << "Cancelling the resource reserved for bundle: " + << bundle_spec->DebugString() << " at node " << node_id; rpc::Address remote_address; remote_address.set_raylet_id(node->node_id()); remote_address.set_ip_address(node->node_manager_address()); remote_address.set_port(node->node_manager_port()); auto return_client = GetOrConnectLeaseClient(remote_address); return_client->CancelResourceReserve( - *bundle_spec, - [this, bundle_spec, node](const Status &status, - const rpc::CancelResourceReserveReply &reply) { - if (!status.ok()) { - return_timer_.expires_from_now(boost::posix_time::milliseconds(5)); - return_timer_.async_wait( - [this, bundle_spec, node](const boost::system::error_code &error) { - if (error == boost::asio::error::operation_aborted) { - return; - } else { - CancelResourceReserve(bundle_spec, node); - } - }); - } + *bundle_spec, [bundle_spec, node_id](const Status &status, + const rpc::CancelResourceReserveReply &reply) { + RAY_LOG(INFO) << "Finished cancelling the resource reserved for bundle: " + << bundle_spec->DebugString() << " at node " << node_id; }); } diff --git a/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.h b/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.h index 572703f9a..afd8acec6 100644 --- a/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.h +++ b/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.h @@ -33,9 +33,6 @@ namespace gcs { using ReserveResourceClientFactoryFn = std::function(const rpc::Address &address)>; -using ReserveResourceCallback = - std::function; - typedef std::pair BundleID; struct pair_hash { template @@ -109,16 +106,16 @@ class GcsPlacementGroupScheduler : public GcsPlacementGroupSchedulerInterface { protected: /// Lease resource from the specified node for the specified bundle. - void ReserveResourceFromNode(std::shared_ptr bundle, - std::shared_ptr node, - ReserveResourceCallback callback); + void ReserveResourceFromNode(const std::shared_ptr &bundle, + const std::shared_ptr &node, + const StatusCallback &callback); /// return resource for the specified node for the specified bundle. /// /// \param bundle A description of the bundle to return. /// \param node The node that the worker will be returned for. - void CancelResourceReserve(std::shared_ptr bundle_spec, - std::shared_ptr node); + void CancelResourceReserve(const std::shared_ptr &bundle_spec, + const std::shared_ptr &node); /// Get an existing lease client or connect a new one. std::shared_ptr GetOrConnectLeaseClient( diff --git a/src/ray/gcs/gcs_server/test/gcs_placement_group_manager_test.cc b/src/ray/gcs/gcs_server/test/gcs_placement_group_manager_test.cc index 12fa81df6..fcd99fccc 100644 --- a/src/ray/gcs/gcs_server/test/gcs_placement_group_manager_test.cc +++ b/src/ray/gcs/gcs_server/test/gcs_placement_group_manager_test.cc @@ -46,6 +46,20 @@ class GcsPlacementGroupManagerTest : public ::testing::Test { io_service_, mock_placement_group_scheduler_, gcs_table_storage_)); } + void SetUp() override { + thread_io_service_.reset(new std::thread([this] { + std::unique_ptr work( + new boost::asio::io_service::work(io_service_)); + io_service_.run(); + })); + } + + void TearDown() override { + io_service_.stop(); + thread_io_service_->join(); + } + + std::unique_ptr thread_io_service_; boost::asio::io_service io_service_; std::shared_ptr gcs_table_storage_; std::shared_ptr mock_placement_group_scheduler_; @@ -54,34 +68,28 @@ class GcsPlacementGroupManagerTest : public ::testing::Test { TEST_F(GcsPlacementGroupManagerTest, TestBasic) { auto create_placement_group_request = Mocker::GenCreatePlacementGroupRequest(); - std::vector> finished_placement_groups; + std::atomic finished_placement_group_count(0); gcs_placement_group_manager_->RegisterPlacementGroup( create_placement_group_request, - [&finished_placement_groups]( - std::shared_ptr placement_group) { - finished_placement_groups.emplace_back(placement_group); - }); - ASSERT_EQ(finished_placement_groups.size(), 0); + [&finished_placement_group_count]() { ++finished_placement_group_count; }); + ASSERT_EQ(finished_placement_group_count, 0); ASSERT_EQ(mock_placement_group_scheduler_->placement_groups.size(), 1); auto placement_group = mock_placement_group_scheduler_->placement_groups.back(); mock_placement_group_scheduler_->placement_groups.pop_back(); gcs_placement_group_manager_->OnPlacementGroupCreationSuccess(placement_group); - ASSERT_EQ(finished_placement_groups.size(), 1); + WaitForExpectedCount(finished_placement_group_count, 1); ASSERT_EQ(placement_group->GetState(), rpc::PlacementGroupTableData::ALIVE); } TEST_F(GcsPlacementGroupManagerTest, TestSchedulingFailed) { auto create_placement_group_request = Mocker::GenCreatePlacementGroupRequest(); - std::vector> finished_placement_groups; + std::atomic finished_placement_group_count(0); gcs_placement_group_manager_->RegisterPlacementGroup( create_placement_group_request, - [&finished_placement_groups]( - std::shared_ptr placement_group) { - finished_placement_groups.emplace_back(placement_group); - }); + [&finished_placement_group_count]() { ++finished_placement_group_count; }); - ASSERT_EQ(finished_placement_groups.size(), 0); + ASSERT_EQ(finished_placement_group_count, 0); ASSERT_EQ(mock_placement_group_scheduler_->placement_groups.size(), 1); auto placement_group = mock_placement_group_scheduler_->placement_groups.back(); mock_placement_group_scheduler_->placement_groups.clear(); @@ -90,32 +98,29 @@ TEST_F(GcsPlacementGroupManagerTest, TestSchedulingFailed) { gcs_placement_group_manager_->SchedulePendingPlacementGroups(); ASSERT_EQ(mock_placement_group_scheduler_->placement_groups.size(), 1); mock_placement_group_scheduler_->placement_groups.clear(); - ASSERT_EQ(finished_placement_groups.size(), 0); + ASSERT_EQ(finished_placement_group_count, 0); // Check that the placement_group is in state `ALIVE`. gcs_placement_group_manager_->OnPlacementGroupCreationSuccess(placement_group); - ASSERT_EQ(finished_placement_groups.size(), 1); + WaitForExpectedCount(finished_placement_group_count, 1); ASSERT_EQ(placement_group->GetState(), rpc::PlacementGroupTableData::ALIVE); } TEST_F(GcsPlacementGroupManagerTest, TestGetPlacementGroupIDByName) { auto create_placement_group_request = Mocker::GenCreatePlacementGroupRequest("test_name"); - std::vector> finished_placement_groups; + std::atomic finished_placement_group_count(0); gcs_placement_group_manager_->RegisterPlacementGroup( create_placement_group_request, - [&finished_placement_groups]( - std::shared_ptr placement_group) { - finished_placement_groups.emplace_back(placement_group); - }); + [&finished_placement_group_count]() { ++finished_placement_group_count; }); - ASSERT_EQ(finished_placement_groups.size(), 0); + ASSERT_EQ(finished_placement_group_count, 0); ASSERT_EQ(mock_placement_group_scheduler_->placement_groups.size(), 1); auto placement_group = mock_placement_group_scheduler_->placement_groups.back(); mock_placement_group_scheduler_->placement_groups.pop_back(); gcs_placement_group_manager_->OnPlacementGroupCreationSuccess(placement_group); - ASSERT_EQ(finished_placement_groups.size(), 1); + WaitForExpectedCount(finished_placement_group_count, 1); ASSERT_EQ(placement_group->GetState(), rpc::PlacementGroupTableData::ALIVE); ASSERT_EQ( gcs_placement_group_manager_->GetPlacementGroupIDByName("test_name"), diff --git a/src/ray/gcs/gcs_server/test/gcs_placement_group_scheduler_test.cc b/src/ray/gcs/gcs_server/test/gcs_placement_group_scheduler_test.cc index e90dea612..70413007c 100644 --- a/src/ray/gcs/gcs_server/test/gcs_placement_group_scheduler_test.cc +++ b/src/ray/gcs/gcs_server/test/gcs_placement_group_scheduler_test.cc @@ -23,8 +23,14 @@ namespace ray { class GcsPlacementGroupSchedulerTest : public ::testing::Test { public: void SetUp() override { + thread_io_service_.reset(new std::thread([this] { + std::unique_ptr work( + new boost::asio::io_service::work(io_service_)); + io_service_.run(); + })); + raylet_client_ = std::make_shared(); - gcs_table_storage_ = std::make_shared(redis_client_); + gcs_table_storage_ = std::make_shared(io_service_); gcs_pub_sub_ = std::make_shared(redis_client_); gcs_node_manager_ = std::make_shared( io_service_, io_service_, gcs_pub_sub_, gcs_table_storage_); @@ -37,7 +43,24 @@ class GcsPlacementGroupSchedulerTest : public ::testing::Test { [this](const rpc::Address &address) { return raylet_client_; }); } + void TearDown() override { + io_service_.stop(); + thread_io_service_->join(); + } + + template + void WaitPendingDone(const std::vector &data, int expected_count) { + auto condition = [this, &data, expected_count]() { + absl::MutexLock lock(&vector_mutex_); + return (int)data.size() == expected_count; + }; + EXPECT_TRUE(WaitForCondition(condition, timeout_ms_.count())); + } + protected: + const std::chrono::milliseconds timeout_ms_{6000}; + absl::Mutex vector_mutex_; + std::unique_ptr thread_io_service_; boost::asio::io_service io_service_; std::shared_ptr store_client_; @@ -90,9 +113,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupSuccess) { gcs_placement_group_scheduler_->Schedule( placement_group, [this](std::shared_ptr placement_group) { + absl::MutexLock lock(&vector_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }, [this](std::shared_ptr placement_group) { + absl::MutexLock lock(&vector_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }); @@ -100,8 +125,8 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupSuccess) { ASSERT_EQ(2, raylet_client_->lease_callbacks.size()); ASSERT_TRUE(raylet_client_->GrantResourceReserve()); ASSERT_TRUE(raylet_client_->GrantResourceReserve()); - ASSERT_EQ(0, failure_placement_groups_.size()); - ASSERT_EQ(1, success_placement_groups_.size()); + WaitPendingDone(failure_placement_groups_, 0); + WaitPendingDone(success_placement_groups_, 1); ASSERT_EQ(placement_group, success_placement_groups_.front()); } @@ -119,9 +144,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupFailed) { gcs_placement_group_scheduler_->Schedule( placement_group, [this](std::shared_ptr placement_group) { + absl::MutexLock lock(&vector_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }, [this](std::shared_ptr placement_group) { + absl::MutexLock lock(&vector_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }); @@ -129,10 +156,10 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupFailed) { ASSERT_EQ(2, raylet_client_->lease_callbacks.size()); ASSERT_TRUE(raylet_client_->GrantResourceReserve(false)); ASSERT_TRUE(raylet_client_->GrantResourceReserve(false)); - // // Reply the placement_group creation request, then the placement_group should be - // scheduled successfully. - ASSERT_EQ(1, failure_placement_groups_.size()); - ASSERT_EQ(0, success_placement_groups_.size()); + // Reply the placement_group creation request, then the placement_group should be + // scheduled successfully. + WaitPendingDone(failure_placement_groups_, 1); + WaitPendingDone(success_placement_groups_, 0); ASSERT_EQ(placement_group, failure_placement_groups_.front()); } @@ -150,9 +177,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupReturnResource) gcs_placement_group_scheduler_->Schedule( placement_group, [this](std::shared_ptr placement_group) { + absl::MutexLock lock(&vector_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }, [this](std::shared_ptr placement_group) { + absl::MutexLock lock(&vector_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }); @@ -163,9 +192,9 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupReturnResource) ASSERT_TRUE(raylet_client_->GrantResourceReserve(false)); ASSERT_EQ(1, raylet_client_->num_return_requested); // Reply the placement_group creation request, then the placement_group should be - // scheduled successfully. - ASSERT_EQ(1, failure_placement_groups_.size()); - ASSERT_EQ(0, success_placement_groups_.size()); + // scheduled successfully. + WaitPendingDone(failure_placement_groups_, 1); + WaitPendingDone(success_placement_groups_, 0); ASSERT_EQ(placement_group, failure_placement_groups_.front()); }