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

Optimize placement group log (#9891)

Browse Source
This commit is contained in:
fangfengbin
2020-08-05 14:41:32 +08:00
committed by GitHub
parent 5af7d24f66
commit 193d11ab8b
11 changed files with 253 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/common/bundle_spec.cc
+8
View File
@@ -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) {
src/ray/common/bundle_spec.h
+2
View File
@@ -78,6 +78,8 @@ class BundleSpecification : public MessageWrapper<rpc::Bundle> {
/// Returns the spillback bundle callback, or nullptr.
const SpillbackBundleCallback &OnSpillback() const { return on_spillback_; }
std::string DebugString() const;
private:
void ComputeResources();
src/ray/common/test_util.cc
+8
View File
@@ -179,6 +179,14 @@ bool WaitForCondition(std::function<bool()> condition, int timeout_ms) {
return false;
}
void WaitForExpectedCount(std::atomic<int> &current_count, int expected_count,
int timeout_ms) {
auto condition = [&current_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";
{
src/ray/common/test_util.h
+9
View File
@@ -49,6 +49,15 @@ static const int64_t SHOULD_CHECK_MESSAGE_ORDER = 123450000;
/// \return Whether the condition is met.
bool WaitForCondition(std::function<bool()> 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<int> &current_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);
src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc
+36 -43
View File
@@ -483,13 +483,6 @@ class ServiceBasedGcsClientTest : public ::testing::Test {
return status == std::future_status::ready && future.get();
}
void WaitPendingDone(std::atomic<int> &current_count, int expected_count) {
auto condition = [&current_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<rpc::GcsNodeInfo> 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) {
src/ray/gcs/gcs_server/gcs_placement_group_manager.cc
+46 -46
View File
@@ -41,8 +41,8 @@ std::string GcsPlacementGroup::GetName() const {
std::vector<std::shared_ptr<BundleSpecification>> GcsPlacementGroup::GetBundles() const {
auto bundles = placement_group_table_data_.bundles();
std::vector<std::shared_ptr<BundleSpecification>> ret_bundles;
for (auto iter = bundles.begin(); iter != bundles.end(); iter++) {
ret_bundles.push_back(std::make_shared<BundleSpecification>(*iter));
for (auto &bundle : bundles) {
ret_bundles.push_back(std::make_shared<BundleSpecification>(bundle));
}
return ret_bundles;
}
@@ -62,25 +62,25 @@ GcsPlacementGroupManager::GcsPlacementGroupManager(
std::shared_ptr<GcsPlacementGroupSchedulerInterface> scheduler,
std::shared_ptr<gcs::GcsTableStorage> 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<void(std::shared_ptr<GcsPlacementGroup>)> 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<GcsPlacementGroup>(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<GcsPlacementGroup> 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<GcsPlacementGroup> 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<GcsPlacementGroup> 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<gcs::GcsPlacementGroup> 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<GcsPlacementGroup>(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();
src/ray/gcs/gcs_server/gcs_placement_group_manager.h
+5 -12
View File
@@ -81,6 +81,7 @@ class GcsPlacementGroup {
using RegisterPlacementGroupCallback =
std::function<void(std::shared_ptr<GcsPlacementGroup>)>;
/// 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<PlacementGroupID, std::vector<RegisterPlacementGroupCallback>>
placement_group_to_register_callbacks_;
/// Callback of placement_group registration requests that are not yet flushed.
absl::flat_hash_map<PlacementGroupID, EmptyCallback>
placement_group_to_register_callback_;
/// All registered placement_groups (pending placement_groups are also included).
absl::flat_hash_map<PlacementGroupID, std::shared_ptr<GcsPlacementGroup>>
registered_placement_groups_;
src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc
+68 -100
View File
@@ -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<GcsPackStrategy>());
@@ -76,73 +76,59 @@ void GcsPlacementGroupScheduler::Schedule(
std::shared_ptr<GcsPlacementGroup> placement_group,
std::function<void(std::shared_ptr<GcsPlacementGroup>)> schedule_failure_handler,
std::function<void(std::shared_ptr<GcsPlacementGroup>)> 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<std::unordered_map<BundleID, ClientID, pair_hash>>(
new std::unordered_map<BundleID, ClientID, pair_hash>());
auto bundle_locations = std::make_shared<std::unordered_map<
BundleID, std::pair<ClientID, std::shared_ptr<BundleSpecification>>, pair_hash>>();
// To count how many scheduler have been return, which include success and failure.
auto finish_count = std::make_shared<size_t>();
auto finished_count = std::make_shared<size_t>();
/// 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<BundleSpecification> bundle,
std::shared_ptr<ray::rpc::GcsNodeInfo> node, ReserveResourceCallback callback) {
RAY_CHECK(node);
const std::shared_ptr<BundleSpecification> &bundle,
const std::shared_ptr<ray::rpc::GcsNodeInfo> &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<BundleSpecification> bundle_spec,
std::shared_ptr<ray::rpc::GcsNodeInfo> node) {
RAY_CHECK(node);
const std::shared_ptr<BundleSpecification> &bundle_spec,
const std::shared_ptr<ray::rpc::GcsNodeInfo> &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;
});
}
src/ray/gcs/gcs_server/gcs_placement_group_scheduler.h
+5 -8
View File
@@ -33,9 +33,6 @@ namespace gcs {
using ReserveResourceClientFactoryFn =
std::function<std::shared_ptr<ResourceReserveInterface>(const rpc::Address &address)>;
using ReserveResourceCallback =
std::function<void(const Status &, const rpc::RequestResourceReserveReply &)>;
typedef std::pair<PlacementGroupID, int64_t> BundleID;
struct pair_hash {
template <class T1, class T2>
@@ -109,16 +106,16 @@ class GcsPlacementGroupScheduler : public GcsPlacementGroupSchedulerInterface {
protected:
/// Lease resource from the specified node for the specified bundle.
void ReserveResourceFromNode(std::shared_ptr<BundleSpecification> bundle,
std::shared_ptr<ray::rpc::GcsNodeInfo> node,
ReserveResourceCallback callback);
void ReserveResourceFromNode(const std::shared_ptr<BundleSpecification> &bundle,
const std::shared_ptr<ray::rpc::GcsNodeInfo> &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<BundleSpecification> bundle_spec,
std::shared_ptr<ray::rpc::GcsNodeInfo> node);
void CancelResourceReserve(const std::shared_ptr<BundleSpecification> &bundle_spec,
const std::shared_ptr<ray::rpc::GcsNodeInfo> &node);
/// Get an existing lease client or connect a new one.
std::shared_ptr<ResourceReserveInterface> GetOrConnectLeaseClient(
src/ray/gcs/gcs_server/test/gcs_placement_group_manager_test.cc
+27 -22
View File
@@ -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<boost::asio::io_service::work> 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<std::thread> thread_io_service_;
boost::asio::io_service io_service_;
std::shared_ptr<gcs::GcsTableStorage> gcs_table_storage_;
std::shared_ptr<MockPlacementGroupScheduler> 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<std::shared_ptr<gcs::GcsPlacementGroup>> finished_placement_groups;
std::atomic<int> finished_placement_group_count(0);
gcs_placement_group_manager_->RegisterPlacementGroup(
create_placement_group_request,
[&finished_placement_groups](
std::shared_ptr<gcs::GcsPlacementGroup> 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<std::shared_ptr<gcs::GcsPlacementGroup>> finished_placement_groups;
std::atomic<int> finished_placement_group_count(0);
gcs_placement_group_manager_->RegisterPlacementGroup(
create_placement_group_request,
[&finished_placement_groups](
std::shared_ptr<gcs::GcsPlacementGroup> 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<std::shared_ptr<gcs::GcsPlacementGroup>> finished_placement_groups;
std::atomic<int> finished_placement_group_count(0);
gcs_placement_group_manager_->RegisterPlacementGroup(
create_placement_group_request,
[&finished_placement_groups](
std::shared_ptr<gcs::GcsPlacementGroup> 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"),
src/ray/gcs/gcs_server/test/gcs_placement_group_scheduler_test.cc
+39 -10
View File
@@ -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<boost::asio::io_service::work> work(
new boost::asio::io_service::work(io_service_));
io_service_.run();
}));
raylet_client_ = std::make_shared<GcsServerMocker::MockRayletResourceClient>();
gcs_table_storage_ = std::make_shared<gcs::RedisGcsTableStorage>(redis_client_);
gcs_table_storage_ = std::make_shared<gcs::InMemoryGcsTableStorage>(io_service_);
gcs_pub_sub_ = std::make_shared<GcsServerMocker::MockGcsPubSub>(redis_client_);
gcs_node_manager_ = std::make_shared<gcs::GcsNodeManager>(
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 <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;
};
EXPECT_TRUE(WaitForCondition(condition, timeout_ms_.count()));
}
protected:
const std::chrono::milliseconds timeout_ms_{6000};
absl::Mutex vector_mutex_;
std::unique_ptr<std::thread> thread_io_service_;
boost::asio::io_service io_service_;
std::shared_ptr<gcs::StoreClient> store_client_;
@@ -90,9 +113,11 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestSchedulePlacementGroupSuccess) {
gcs_placement_group_scheduler_->Schedule(
placement_group,
[this](std::shared_ptr<gcs::GcsPlacementGroup> placement_group) {
absl::MutexLock lock(&vector_mutex_);
failure_placement_groups_.emplace_back(std::move(placement_group));
},
[this](std::shared_ptr<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> placement_group) {
absl::MutexLock lock(&vector_mutex_);
failure_placement_groups_.emplace_back(std::move(placement_group));
},
[this](std::shared_ptr<gcs::GcsPlacementGroup> 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<gcs::GcsPlacementGroup> placement_group) {
absl::MutexLock lock(&vector_mutex_);
failure_placement_groups_.emplace_back(std::move(placement_group));
},
[this](std::shared_ptr<gcs::GcsPlacementGroup> 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());
}