diff --git a/ci/regression_test/stress_tests/placement_group/test_placement_group.py b/ci/regression_test/stress_tests/placement_group/test_placement_group.py index 3f7c9cc6c..2e4f4387e 100644 --- a/ci/regression_test/stress_tests/placement_group/test_placement_group.py +++ b/ci/regression_test/stress_tests/placement_group/test_placement_group.py @@ -15,7 +15,7 @@ from ray.util.placement_group import (placement_group, remove_placement_group) # TODO(sang): Increase the number in the actual stress test. # This number should be divisible by 3. -resource_quantity = 999 +resource_quantity = 666 num_nodes = 5 custom_resources = {"pg_custom": resource_quantity} # Create pg that uses 1 resource of cpu & custom resource. @@ -73,19 +73,22 @@ assert ray.cluster_resources()["pg_custom"] == num_nodes * resource_quantity # - Make sure jobs are done without breaking GCS server. # - Make sure all the resources are recovered after the job is done. # - Measure the creation latency in the stressful environment. -@ray.remote +@ray.remote(num_cpus=0, num_gpus=1, max_calls=0) def mock_task(): time.sleep(0.1) return True -@ray.remote +@ray.remote(num_cpus=0, num_gpus=1, max_restarts=0) class MockActor: def __init__(self): pass + def ping(self): + pass -@ray.remote + +@ray.remote(num_cpus=0) def pg_launcher(pre_created_pgs, num_pgs_to_create): pgs = [] pgs += pre_created_pgs @@ -102,21 +105,29 @@ def pg_launcher(pre_created_pgs, num_pgs_to_create): else: pgs_unremoved.append(pg) + tasks = [] + max_actor_cnt = 5 + actor_cnt = 0 + actors = [] # Randomly schedule tasks or actors on placement groups that # are not removed. for pg in pgs_unremoved: # TODO(sang): Comment in this line causes GCS actor management # failure. We need to fix it. - # if random() < .5: - mock_task.options(placement_group=pg).remote() - # else: - # MockActor.options(placement_group=pg).remote() + if random() < .5: + tasks.append(mock_task.options(placement_group=pg).remote()) + else: + if actor_cnt < max_actor_cnt: + actors.append(MockActor.options(placement_group=pg).remote()) + actor_cnt += 1 # Remove the rest of placement groups. for pg in pgs_removed: remove_placement_group(pg) - ray.get([pg.ready() for pg in pgs_unremoved], timeout=10) + ray.get([pg.ready() for pg in pgs_unremoved]) + ray.get(tasks) + ray.get([actor.ping.remote() for actor in actors]) # Since placement groups are scheduled, remove them. for pg in pgs_unremoved: remove_placement_group(pg) @@ -124,7 +135,6 @@ def pg_launcher(pre_created_pgs, num_pgs_to_create): pre_created_num_pgs = round(num_pg * 0.3) num_pgs_to_create = num_pg - pre_created_num_pgs - pg_launchers = [] for i in range(3): pre_created_pgs = [ diff --git a/src/ray/gcs/gcs_server/gcs_placement_group_manager.cc b/src/ray/gcs/gcs_server/gcs_placement_group_manager.cc index 3521dd225..997d0e50d 100644 --- a/src/ray/gcs/gcs_server/gcs_placement_group_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_placement_group_manager.cc @@ -294,7 +294,8 @@ void GcsPlacementGroupManager::RemovePlacementGroup( // that the creation of placement group has failed. auto it = placement_group_to_register_callback_.find(placement_group_id); if (it != placement_group_to_register_callback_.end()) { - it->second(Status::NotFound("Placement group is removed before it is created")); + it->second( + Status::NotFound("Placement group is removed before it is created.")); placement_group_to_register_callback_.erase(it); } on_placement_group_removed(status); 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 bfeb209f3..f162a8894 100644 --- a/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc +++ b/src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc @@ -252,8 +252,6 @@ void GcsPlacementGroupScheduler::DestroyPlacementGroupBundleResourcesIfExists( // Check if we can find committed bundle locations. const auto &maybe_bundle_locations = committed_bundle_location_index_.GetBundleLocations(placement_group_id); - // If bundle location has been already removed, it means bundles - // are already destroyed. Do nothing. if (maybe_bundle_locations.has_value()) { committed_bundle_locations = maybe_bundle_locations.value(); } @@ -267,6 +265,8 @@ void GcsPlacementGroupScheduler::DestroyPlacementGroupBundleResourcesIfExists( } // Cancel all resource reservation. + RAY_LOG(INFO) << "Cancelling all bundles of a placement group, id is " + << placement_group_id; for (const auto &iter : *(committed_bundle_locations)) { auto &bundle_spec = iter.second.second; auto &node_id = iter.second.first; @@ -296,6 +296,7 @@ void GcsPlacementGroupScheduler::PrepareResources( callback(Status::NotFound("Node is already dead.")); return; } + const auto lease_client = GetLeaseClientFromNode(node.value()); const auto node_id = NodeID::FromBinary(node.value()->node_id()); RAY_LOG(DEBUG) << "Preparing resource from node " << node_id @@ -323,6 +324,7 @@ void GcsPlacementGroupScheduler::CommitResources( RAY_CHECK(node.has_value()); const auto lease_client = GetLeaseClientFromNode(node.value()); const auto node_id = NodeID::FromBinary(node.value()->node_id()); + RAY_LOG(DEBUG) << "Committing resource to a node " << node_id << " for a bundle: " << bundle->DebugString(); lease_client->CommitBundleResources( @@ -353,6 +355,7 @@ void GcsPlacementGroupScheduler::CancelResourceReserve( RAY_LOG(DEBUG) << "Cancelling the resource reserved for bundle: " << bundle_spec->DebugString() << " at node " << node_id; const auto return_client = GetLeaseClientFromNode(node.value()); + return_client->CancelResourceReserve( *bundle_spec, [bundle_spec, node_id](const Status &status, const rpc::CancelResourceReserveReply &reply) { diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 52b23357e..b3c3352fd 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -1842,11 +1842,10 @@ void NodeManager::HandleCancelResourceReserve( } // Return bundle resources. - if (ReturnBundleResources(bundle_spec)) { - // Call task dispatch to assign work to the released resources. - TryLocalInfeasibleTaskScheduling(); - DispatchTasks(local_queues_.GetReadyTasksByClass()); - } + ReturnBundleResources(bundle_spec); + TryLocalInfeasibleTaskScheduling(); + DispatchTasks(local_queues_.GetReadyTasksByClass()); + send_reply_callback(Status::OK(), nullptr, nullptr); } @@ -2158,10 +2157,10 @@ void NodeManager::ScheduleTasks( << task.GetTaskSpecification().GetRequiredResources().ToString() << " for execution and " << task.GetTaskSpecification().GetRequiredPlacementResources().ToString() - << " for placement, however there are no nodes in the cluster that can " - << "provide the requested resources. To resolve this issue, consider " - << "reducing the resource requests of this task or add nodes that " - << "can fit the task."; + << " for placement, however the cluster currently cannot provide the requested " + "resources. The required resources may be added as autoscaling takes place " + "or placement groups are scheduled. Otherwise, consider reducing the " + "resource requirements of the task."; auto error_data_ptr = gcs::CreateErrorTableData(type, error_message.str(), current_time_ms(), task.GetTaskSpecification().JobId()); @@ -3258,8 +3257,11 @@ void NodeManager::HandleGetNodeStats(const rpc::GetNodeStatsRequest &node_stats_ if (status.ok()) { worker_stats->mutable_core_worker_stats()->MergeFrom(r.core_worker_stats()); } else { - RAY_LOG(ERROR) << "Failed to send get core worker stats request: " - << status.ToString(); + RAY_LOG(WARNING) << "Failed to send get core worker stats request, " + << "worker id is " << worker->WorkerId() << ", status is " + << status.ToString() + << ". This is likely since the worker has died before the " + "request was sent."; worker_stats->set_fetch_error(status.ToString()); } if (reply->num_workers() == all_workers.size()) {