wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-18 12:40:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

[PlacementGroup]Add guarded by in placement group scheduler ut (#11306)

Browse Source 
* 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
This commit is contained in:
DK.Pino
2020-10-19 18:54:35 -07:00
committed by GitHub
parent 0a9cc9cce5
commit 1b3b009f7a
1 changed files with 123 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/gcs/gcs_server/test/gcs_placement_group_scheduler_test.cc
+123 -90
View File
@@ -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 <typename Data>
void WaitPendingDone(const std::vector<Data> &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<gcs::GcsPlacementGroup> 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 <typename Data>
void WaitPendingDone(const std::list<Data> &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<gcs::GcsPlacementGroup> placement_group) {
absl::MutexLock lock(&placement_group_requests_mutex_);
failure_placement_groups_.emplace_back(std::move(placement_group));
},
[this](std::shared_ptr<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup>(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<std::thread> thread_io_service_;
boost::asio::io_service io_service_;
std::shared_ptr<gcs::StoreClient> store_client_;
@@ -175,8 +204,10 @@ class GcsPlacementGroupSchedulerTest : public ::testing::Test {
std::vector<std::shared_ptr<GcsServerMocker::MockRayletResourceClient>> raylet_clients_;
std::shared_ptr<gcs::GcsNodeManager> gcs_node_manager_;
std::shared_ptr<GcsServerMocker::MockedGcsPlacementGroupScheduler> scheduler_;
std::vector<std::shared_ptr<gcs::GcsPlacementGroup>> success_placement_groups_;
std::vector<std::shared_ptr<gcs::GcsPlacementGroup>> failure_placement_groups_;
std::vector<std::shared_ptr<gcs::GcsPlacementGroup>> success_placement_groups_
GUARDED_BY(placement_group_requests_mutex_);
std::vector<std::shared_ptr<gcs::GcsPlacementGroup>> failure_placement_groups_
GUARDED_BY(placement_group_requests_mutex_);
std::shared_ptr<GcsServerMocker::MockGcsPubSub> gcs_pub_sub_;
std::shared_ptr<gcs::GcsTableStorage> gcs_table_storage_;
std::shared_ptr<gcs::RedisClient> redis_client_;
@@ -223,11 +254,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupReplyFailure) {
scheduler_->ScheduleUnplacedBundles(
placement_group,
[this](std::shared_ptr<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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