diff --git a/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc b/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc index 52daa01c0..122dc877e 100644 --- a/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc +++ b/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc @@ -386,17 +386,23 @@ void GcsPlacementGroupScheduler::CommitAllBundles( const auto &node = gcs_node_manager_.GetNode(node_id); const auto &bundle = bundle_to_commit.second.second; - // TODO(sang) Handle the case nodes are dead. - CommitResources( - bundle, node, - [this, lease_status_tracker, bundle, node_id, schedule_failure_handler, - schedule_success_handler](const Status &status) { - lease_status_tracker->MarkCommitRequestReturned(node_id, bundle, status); - if (lease_status_tracker->AllCommitRequestReturned()) { - OnAllBundleCommitRequestReturned( - lease_status_tracker, schedule_failure_handler, schedule_success_handler); - } - }); + auto commit_resources_callback = [this, lease_status_tracker, bundle, node_id, + schedule_failure_handler, + schedule_success_handler](const Status &status) { + lease_status_tracker->MarkCommitRequestReturned(node_id, bundle, status); + if (lease_status_tracker->AllCommitRequestReturned()) { + OnAllBundleCommitRequestReturned(lease_status_tracker, schedule_failure_handler, + schedule_success_handler); + } + }; + + if (node != nullptr) { + CommitResources(bundle, node, commit_resources_callback); + } else { + RAY_LOG(INFO) << "Failed to commit resources because the node is dead, node id = " + << node_id; + commit_resources_callback(Status::Interrupted("Node is dead")); + } } } @@ -682,7 +688,8 @@ bool LeaseStatusTracker::AllCommitRequestsSuccessful() const { // We don't check cancel state here because we shouldn't destroy bundles when // commit requests failed. Cancel state should be treated separately. return AllCommitRequestReturned() && - preparing_bundle_locations_->size() == bundles_to_schedule_.size(); + preparing_bundle_locations_->size() == bundles_to_schedule_.size() && + uncommitted_bundle_locations_->empty(); } const std::shared_ptr &LeaseStatusTracker::GetPlacementGroup() const { 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 3ab438eb2..2104aedb2 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 @@ -616,6 +616,36 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestBundleLocationIndex) { ASSERT_TRUE(bundle_location_index.GetBundleLocationsOnNode(node2).value()->size() == 1); } +TEST_F(GcsPlacementGroupSchedulerTest, TestNodeDeadDuringCommitResources) { + auto node0 = Mocker::GenNodeInfo(0); + auto node1 = Mocker::GenNodeInfo(1); + AddNode(node0); + AddNode(node1); + ASSERT_EQ(2, gcs_node_manager_->GetAllAliveNodes().size()); + + auto create_placement_group_request = Mocker::GenCreatePlacementGroupRequest(); + auto placement_group = + std::make_shared(create_placement_group_request); + + // 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_); + 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_); + success_placement_groups_.emplace_back(std::move(placement_group)); + }; + + scheduler_->ScheduleUnplacedBundles(placement_group, failure_handler, success_handler); + ASSERT_TRUE(raylet_clients_[0]->GrantPrepareBundleResources()); + gcs_node_manager_->RemoveNode(ClientID::FromBinary(node1->node_id())); + ASSERT_TRUE(raylet_clients_[1]->GrantPrepareBundleResources()); + WaitPendingDone(failure_placement_groups_, 1); +} + } // namespace ray int main(int argc, char **argv) {