From 1b3b009f7ae254d4ea07972f21aca28f9d4f6ab4 Mon Sep 17 00:00:00 2001 From: "DK.Pino" Date: Tue, 20 Oct 2020 09:54:35 +0800 Subject: [PATCH] [PlacementGroup]Add guarded by in placement group scheduler ut (#11306) * add GUARDED_BY for success_placement_groups_ and failure_placement_groups_ vector * update lint * update lint * update logical * update lint * change int to unsigned int * update lint * rename vector_mutex_ to placement_group_requests_mutex_ * resolve comment * add int() for windows --- .../gcs_placement_group_scheduler_test.cc | 213 ++++++++++-------- 1 file changed, 123 insertions(+), 90 deletions(-) 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 765430c04..5e6fb352b 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 @@ -20,6 +20,11 @@ namespace ray { +enum class GcsPlacementGroupStatus : int32_t { + SUCCESS = 0, + FAILURE = 1, +}; + class GcsPlacementGroupSchedulerTest : public ::testing::Test { public: void SetUp() override { @@ -50,19 +55,41 @@ class GcsPlacementGroupSchedulerTest : public ::testing::Test { 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; + void WaitPlacementGroupPendingDone(int expected_count, + const GcsPlacementGroupStatus status) { + auto condition = [this, expected_count, status]() { + absl::MutexLock lock(&placement_group_requests_mutex_); + return status == GcsPlacementGroupStatus::SUCCESS + ? (int)success_placement_groups_.size() == expected_count + : (int)failure_placement_groups_.size() == expected_count; }; EXPECT_TRUE(WaitForCondition(condition, timeout_ms_.count())); } + void CheckPlacementGroupSize(int expected_count, const GcsPlacementGroupStatus status) { + absl::MutexLock lock(&placement_group_requests_mutex_); + if (status == GcsPlacementGroupStatus::SUCCESS) { + ASSERT_EQ(expected_count, success_placement_groups_.size()); + } else { + ASSERT_EQ(expected_count, failure_placement_groups_.size()); + } + } + + void CheckEqWithPlacementGroupFront( + std::shared_ptr placement_group, + const GcsPlacementGroupStatus status) { + absl::MutexLock lock(&placement_group_requests_mutex_); + if (status == GcsPlacementGroupStatus::SUCCESS) { + ASSERT_EQ(placement_group, success_placement_groups_.front()); + } else { + ASSERT_EQ(placement_group, failure_placement_groups_.front()); + } + } + template void WaitPendingDone(const std::list &data, int expected_count) { auto condition = [this, &data, expected_count]() { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); return (int)data.size() == expected_count; }; EXPECT_TRUE(WaitForCondition(condition, timeout_ms_.count())); @@ -85,18 +112,20 @@ class GcsPlacementGroupSchedulerTest : public ::testing::Test { scheduler_->ScheduleUnplacedBundles( placement_group, [this](std::shared_ptr placement_group) { + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }, [this](std::shared_ptr placement_group) { + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }); // The lease request should not be send and the scheduling of placement_group should // fail as there are no available nodes. ASSERT_EQ(raylet_clients_[0]->num_lease_requested, 0); - ASSERT_EQ(0, success_placement_groups_.size()); - ASSERT_EQ(1, failure_placement_groups_.size()); - ASSERT_EQ(placement_group, failure_placement_groups_.front()); + CheckPlacementGroupSize(0, GcsPlacementGroupStatus::SUCCESS); + CheckPlacementGroupSize(1, GcsPlacementGroupStatus::FAILURE); + CheckEqWithPlacementGroupFront(placement_group, GcsPlacementGroupStatus::FAILURE); } void SchedulePlacementGroupSuccessTest(rpc::PlacementStrategy strategy) { @@ -112,11 +141,11 @@ class GcsPlacementGroupSchedulerTest : public ::testing::Test { scheduler_->ScheduleUnplacedBundles( placement_group, [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }, [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }); @@ -127,21 +156,21 @@ class GcsPlacementGroupSchedulerTest : public ::testing::Test { WaitPendingDone(raylet_clients_[0]->commit_callbacks, 2); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); - WaitPendingDone(failure_placement_groups_, 0); - WaitPendingDone(success_placement_groups_, 1); - ASSERT_EQ(placement_group, success_placement_groups_.front()); + WaitPlacementGroupPendingDone(0, GcsPlacementGroupStatus::FAILURE); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); + CheckEqWithPlacementGroupFront(placement_group, GcsPlacementGroupStatus::SUCCESS); } void ReschedulingWhenNodeAddTest(rpc::PlacementStrategy strategy) { AddNode(Mocker::GenNodeInfo(0), 1); auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; @@ -150,8 +179,8 @@ class GcsPlacementGroupSchedulerTest : public ::testing::Test { auto placement_group = std::make_shared(request); scheduler_->ScheduleUnplacedBundles(placement_group, failure_handler, success_handler); - WaitPendingDone(failure_placement_groups_, 1); - ASSERT_EQ(0, success_placement_groups_.size()); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::FAILURE); + CheckPlacementGroupSize(0, GcsPlacementGroupStatus::SUCCESS); // A new node is added, and the rescheduling is successful. AddNode(Mocker::GenNodeInfo(0), 2); @@ -162,12 +191,12 @@ class GcsPlacementGroupSchedulerTest : public ::testing::Test { WaitPendingDone(raylet_clients_[0]->commit_callbacks, 2); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); - WaitPendingDone(success_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); } protected: const std::chrono::milliseconds timeout_ms_{6000}; - absl::Mutex vector_mutex_; + absl::Mutex placement_group_requests_mutex_; std::unique_ptr thread_io_service_; boost::asio::io_service io_service_; std::shared_ptr store_client_; @@ -175,8 +204,10 @@ class GcsPlacementGroupSchedulerTest : public ::testing::Test { std::vector> raylet_clients_; std::shared_ptr gcs_node_manager_; std::shared_ptr scheduler_; - std::vector> success_placement_groups_; - std::vector> failure_placement_groups_; + std::vector> success_placement_groups_ + GUARDED_BY(placement_group_requests_mutex_); + std::vector> failure_placement_groups_ + GUARDED_BY(placement_group_requests_mutex_); std::shared_ptr gcs_pub_sub_; std::shared_ptr gcs_table_storage_; std::shared_ptr redis_client_; @@ -223,11 +254,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupReplyFailure) { scheduler_->ScheduleUnplacedBundles( placement_group, [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }, [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }); @@ -237,20 +268,21 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupReplyFailure) { // Reply failure, so the placement group scheduling failed. ASSERT_TRUE(raylet_clients_[0]->GrantPrepareBundleResources(false)); ASSERT_TRUE(raylet_clients_[0]->GrantPrepareBundleResources(false)); - WaitPendingDone(failure_placement_groups_, 1); - WaitPendingDone(success_placement_groups_, 0); - ASSERT_EQ(placement_group, failure_placement_groups_.front()); + + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::FAILURE); + WaitPlacementGroupPendingDone(0, GcsPlacementGroupStatus::SUCCESS); + CheckEqWithPlacementGroupFront(placement_group, GcsPlacementGroupStatus::FAILURE); } TEST_F(GcsPlacementGroupSchedulerTest, TestSpreadStrategyResourceCheck) { auto node = Mocker::GenNodeInfo(0); AddNode(node, 2); auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; auto request = @@ -259,12 +291,12 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSpreadStrategyResourceCheck) { scheduler_->ScheduleUnplacedBundles(placement_group, failure_handler, success_handler); // The node resource is not enough, scheduling failed. - WaitPendingDone(failure_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::FAILURE); scheduler_->ScheduleUnplacedBundles(placement_group, failure_handler, success_handler); // The node resource is not enough, scheduling failed. - WaitPendingDone(failure_placement_groups_, 2); + WaitPlacementGroupPendingDone(2, GcsPlacementGroupStatus::FAILURE); } TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupReturnResource) { @@ -280,11 +312,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupReturnResource) scheduler_->ScheduleUnplacedBundles( placement_group, [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }, [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }); @@ -296,20 +328,20 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupReturnResource) ASSERT_EQ(1, raylet_clients_[0]->num_return_requested); // 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()); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::FAILURE); + WaitPlacementGroupPendingDone(0, GcsPlacementGroupStatus::SUCCESS); + CheckEqWithPlacementGroupFront(placement_group, GcsPlacementGroupStatus::FAILURE); } TEST_F(GcsPlacementGroupSchedulerTest, TestStrictPackStrategyBalancedScheduling) { AddNode(Mocker::GenNodeInfo(0)); AddNode(Mocker::GenNodeInfo(1)); auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; @@ -338,7 +370,8 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestStrictPackStrategyBalancedScheduling) }; EXPECT_TRUE(WaitForCondition(condition, timeout_ms_.count())); } - WaitPendingDone(success_placement_groups_, 10); + WaitPlacementGroupPendingDone(10, GcsPlacementGroupStatus::SUCCESS); + ASSERT_EQ(node_select_count[0], 5); ASSERT_EQ(node_select_count[1], 5); } @@ -351,11 +384,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestStrictPackStrategyResourceCheck) { auto node0 = Mocker::GenNodeInfo(0); AddNode(node0); auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; auto request = @@ -367,7 +400,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestStrictPackStrategyResourceCheck) { WaitPendingDone(raylet_clients_[0]->commit_callbacks, 2); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); - WaitPendingDone(success_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); // Node1 has less number of bundles, but it doesn't satisfy the resource // requirement. In this case, the bundles should be scheduled on Node0. @@ -383,7 +416,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestStrictPackStrategyResourceCheck) { WaitPendingDone(raylet_clients_[0]->commit_callbacks, 2); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); - WaitPendingDone(success_placement_groups_, 2); + WaitPlacementGroupPendingDone(2, GcsPlacementGroupStatus::SUCCESS); } TEST_F(GcsPlacementGroupSchedulerTest, DestroyPlacementGroup) { @@ -400,11 +433,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, DestroyPlacementGroup) { scheduler_->ScheduleUnplacedBundles( placement_group, [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }, [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }); ASSERT_TRUE(raylet_clients_[0]->GrantPrepareBundleResources()); @@ -412,8 +445,8 @@ TEST_F(GcsPlacementGroupSchedulerTest, DestroyPlacementGroup) { WaitPendingDone(raylet_clients_[0]->commit_callbacks, 2); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); - WaitPendingDone(failure_placement_groups_, 0); - WaitPendingDone(success_placement_groups_, 1); + WaitPlacementGroupPendingDone(0, GcsPlacementGroupStatus::FAILURE); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); const auto &placement_group_id = placement_group->GetPlacementGroupID(); scheduler_->DestroyPlacementGroupBundleResourcesIfExists(placement_group_id); ASSERT_TRUE(raylet_clients_[0]->GrantCancelResourceReserve()); @@ -439,11 +472,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, DestroyCancelledPlacementGroup) { scheduler_->ScheduleUnplacedBundles( placement_group, [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }, [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }); @@ -453,7 +486,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, DestroyCancelledPlacementGroup) { ASSERT_TRUE(raylet_clients_[0]->GrantPrepareBundleResources()); ASSERT_TRUE(raylet_clients_[0]->GrantCancelResourceReserve()); ASSERT_TRUE(raylet_clients_[0]->GrantCancelResourceReserve()); - WaitPendingDone(failure_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::FAILURE); } TEST_F(GcsPlacementGroupSchedulerTest, PlacementGroupCancelledDuringCommit) { @@ -471,11 +504,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, PlacementGroupCancelledDuringCommit) { scheduler_->ScheduleUnplacedBundles( placement_group, [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }, [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }); @@ -488,7 +521,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, PlacementGroupCancelledDuringCommit) { ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); ASSERT_TRUE(raylet_clients_[0]->GrantCancelResourceReserve()); ASSERT_TRUE(raylet_clients_[0]->GrantCancelResourceReserve()); - WaitPendingDone(failure_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::FAILURE); } TEST_F(GcsPlacementGroupSchedulerTest, TestPackStrategyReschedulingWhenNodeAdd) { @@ -499,11 +532,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestPackStrategyLargeBundlesScheduling) { AddNode(Mocker::GenNodeInfo(0)); AddNode(Mocker::GenNodeInfo(1)); auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; @@ -532,7 +565,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestPackStrategyLargeBundlesScheduling) { for (int index = 0; index < raylet_clients_[1]->num_commit_requested; ++index) { ASSERT_TRUE(raylet_clients_[1]->GrantCommitBundleResources()); } - WaitPendingDone(success_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); } TEST_F(GcsPlacementGroupSchedulerTest, TestRescheduleWhenNodeDead) { @@ -548,11 +581,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestRescheduleWhenNodeDead) { // Schedule the placement group successfully. auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; @@ -563,7 +596,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestRescheduleWhenNodeDead) { WaitPendingDone(raylet_clients_[1]->commit_callbacks, 1); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); ASSERT_TRUE(raylet_clients_[1]->GrantCommitBundleResources()); - WaitPendingDone(success_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); auto bundles_on_node0 = scheduler_->GetBundlesOnNode(NodeID::FromBinary(node0->node_id())); @@ -580,7 +613,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestRescheduleWhenNodeDead) { // pr. auto commit_ready = [this]() { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); return raylet_clients_[0]->commit_callbacks.size() == 1 || raylet_clients_[1]->commit_callbacks.size() == 1; }; @@ -589,18 +622,18 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestRescheduleWhenNodeDead) { EXPECT_TRUE(WaitForCondition(commit_ready, timeout_ms_.count())); raylet_clients_[0]->GrantCommitBundleResources(); raylet_clients_[1]->GrantCommitBundleResources(); - WaitPendingDone(success_placement_groups_, 2); + WaitPlacementGroupPendingDone(2, GcsPlacementGroupStatus::SUCCESS); } TEST_F(GcsPlacementGroupSchedulerTest, TestStrictSpreadStrategyResourceCheck) { auto node0 = Mocker::GenNodeInfo(0); AddNode(node0); auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; auto request = Mocker::GenCreatePlacementGroupRequest( @@ -609,13 +642,13 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestStrictSpreadStrategyResourceCheck) { scheduler_->ScheduleUnplacedBundles(placement_group, failure_handler, success_handler); // The number of nodes is less than the number of bundles, scheduling failed. - WaitPendingDone(failure_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::FAILURE); // Node1 resource is insufficient, scheduling failed. auto node1 = Mocker::GenNodeInfo(1); AddNode(node1, 1); scheduler_->ScheduleUnplacedBundles(placement_group, failure_handler, success_handler); - WaitPendingDone(failure_placement_groups_, 2); + WaitPlacementGroupPendingDone(2, GcsPlacementGroupStatus::FAILURE); // The node2 resource is enough and the scheduling is successful. auto node2 = Mocker::GenNodeInfo(2); @@ -627,7 +660,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestStrictSpreadStrategyResourceCheck) { WaitPendingDone(raylet_clients_[2]->commit_callbacks, 1); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); ASSERT_TRUE(raylet_clients_[2]->GrantCommitBundleResources()); - WaitPendingDone(success_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); } TEST_F(GcsPlacementGroupSchedulerTest, TestBundleLocationIndex) { @@ -720,12 +753,12 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestNodeDeadDuringPreparingResources) { // Schedule the placement group. // One node is dead, so one bundle failed to schedule. auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); ASSERT_TRUE(placement_group->GetUnplacedBundles().size() == 2); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; @@ -736,7 +769,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestNodeDeadDuringPreparingResources) { ASSERT_TRUE(raylet_clients_[1]->GrantPrepareBundleResources(false)); ASSERT_TRUE(raylet_clients_[0]->commit_callbacks.size() == 0); ASSERT_TRUE(raylet_clients_[1]->commit_callbacks.size() == 0); - WaitPendingDone(failure_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::FAILURE); } TEST_F(GcsPlacementGroupSchedulerTest, @@ -756,12 +789,12 @@ TEST_F(GcsPlacementGroupSchedulerTest, // Schedule the placement group. // One node is dead, so one bundle failed to schedule. auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); ASSERT_TRUE(placement_group->GetUnplacedBundles().size() == 1); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; @@ -779,7 +812,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, ASSERT_FALSE(raylet_clients_[1]->GrantCommitBundleResources()); // In this case, we treated the placement group creation successful. Instead, // we will reschedule them. - WaitPendingDone(failure_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::FAILURE); } TEST_F(GcsPlacementGroupSchedulerTest, TestNodeDeadDuringCommittingResources) { @@ -796,12 +829,12 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestNodeDeadDuringCommittingResources) { // Schedule the placement group. // One node is dead, so one bundle failed to schedule. auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); ASSERT_TRUE(placement_group->GetUnplacedBundles().size() == 2); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; @@ -814,7 +847,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestNodeDeadDuringCommittingResources) { ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); // Commit will fail because the node is dead. ASSERT_TRUE(raylet_clients_[1]->GrantCommitBundleResources(false)); - WaitPendingDone(success_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); } TEST_F(GcsPlacementGroupSchedulerTest, TestNodeDeadDuringRescheduling) { @@ -830,11 +863,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestNodeDeadDuringRescheduling) { // Schedule the placement group successfully. auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; @@ -845,7 +878,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestNodeDeadDuringRescheduling) { WaitPendingDone(raylet_clients_[1]->commit_callbacks, 1); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); ASSERT_TRUE(raylet_clients_[1]->GrantCommitBundleResources()); - WaitPendingDone(success_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); auto bundles_on_node0 = scheduler_->GetBundlesOnNode(NodeID::FromBinary(node0->node_id())); @@ -867,8 +900,8 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestNodeDeadDuringRescheduling) { ASSERT_TRUE(raylet_clients_[0]->commit_callbacks.size() == 0); ASSERT_TRUE(raylet_clients_[1]->commit_callbacks.size() == 0); - WaitPendingDone(success_placement_groups_, 1); - WaitPendingDone(failure_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::FAILURE); } TEST_F(GcsPlacementGroupSchedulerTest, TestPGCancelledDuringReschedulingCommit) { @@ -884,11 +917,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestPGCancelledDuringReschedulingCommit) // Schedule the placement group successfully. auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; @@ -899,7 +932,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestPGCancelledDuringReschedulingCommit) WaitPendingDone(raylet_clients_[1]->commit_callbacks, 1); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); ASSERT_TRUE(raylet_clients_[1]->GrantCommitBundleResources()); - WaitPendingDone(success_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); auto bundles_on_node0 = scheduler_->GetBundlesOnNode(NodeID::FromBinary(node0->node_id())); @@ -923,8 +956,8 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestPGCancelledDuringReschedulingCommit) // After commits are granted the placement group will be removed. ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); ASSERT_TRUE(raylet_clients_[1]->GrantCommitBundleResources()); - WaitPendingDone(success_placement_groups_, 1); - WaitPendingDone(failure_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::FAILURE); } TEST_F(GcsPlacementGroupSchedulerTest, TestPGCancelledDuringReschedulingCommitPrepare) { @@ -940,11 +973,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestPGCancelledDuringReschedulingCommitPr // Schedule the placement group successfully. auto failure_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); failure_placement_groups_.emplace_back(std::move(placement_group)); }; auto success_handler = [this](std::shared_ptr placement_group) { - absl::MutexLock lock(&vector_mutex_); + absl::MutexLock lock(&placement_group_requests_mutex_); success_placement_groups_.emplace_back(std::move(placement_group)); }; @@ -955,7 +988,7 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestPGCancelledDuringReschedulingCommitPr WaitPendingDone(raylet_clients_[1]->commit_callbacks, 1); ASSERT_TRUE(raylet_clients_[0]->GrantCommitBundleResources()); ASSERT_TRUE(raylet_clients_[1]->GrantCommitBundleResources()); - WaitPendingDone(success_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); auto bundles_on_node0 = scheduler_->GetBundlesOnNode(NodeID::FromBinary(node0->node_id())); @@ -977,8 +1010,8 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestPGCancelledDuringReschedulingCommitPr WaitPendingDone(raylet_clients_[0]->commit_callbacks, 0); WaitPendingDone(raylet_clients_[1]->commit_callbacks, 0); // Make sure the placement group creation has failed. - WaitPendingDone(success_placement_groups_, 1); - WaitPendingDone(failure_placement_groups_, 1); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::SUCCESS); + WaitPlacementGroupPendingDone(1, GcsPlacementGroupStatus::FAILURE); } } // namespace ray