Repository: ray-project/ray
Pull request: [Core] Logic Added to Cancel Infeasible Tasks in GCS Based on Autoscaler State
<!-- Thank you for your contribution! Please review https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR. If you don't have the access to it, we will shortly find a reviewer and assign them to your PR. -->

## Why are these changes needed?

This is the last PR for the feature of early termination of infeasible tasks. 

The previous PRs:
(1) added logic in GCS to obtain the per node infeasible resource requests from the autoscaler state; 
(2) added the new API in raylet_client, node_manager and cluster_task_manager to cancel tasks with certain resource request shapes

This PR added the integration of the above PRs:
(1) added the logic to call the cancel resource shape API based on the per node infeasible requests from the autoscaler
(2) put the feature behind a ray config and make the default to be on
(3) small improvements on previous PRs (logging, comments, messages, add an early exit when getting the per node infeasible requests)
(4) added integration tests for both normal task scheduling case and actor creation case 

With the change, the infeasible tasks (both normal tasks and the actor creation tasks) can be early terminated by default.

<!-- Please give a short summary of the change and the problem this solves. -->

## Related issue number
Closes #45909
<!-- For example: "Closes #1234" -->

## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR.
- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for https://docs.ray.io/en/master/.
    - [ ] I've added any new APIs to the API Reference. For example, if I added a
           method in Tune, I've added it in `doc/source/tune/api/` under the
           corresponding `.rst` file.
- [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(


Problem statement:
Make infeasible tasks error much more obvious
Many times I see users run into common error scenarios that end up hanging the entire workload.

1. Scheduling something on the head node when head node `num_cpus=0`
2. Scheduling a task that requests more cpus or gpus then is available in the cluster. (And autoscaling is disabled)

Although we show a warning in the events. Those can be hard to catch. We should make infeasible things errors and have the task itself return an error.

Hints from issue discussion:
adding for consideration in next sprint
can we consider this for next sprint? This comes up in ray data as well.
@jjyao what would be the tshirt size for this?
not done yet

Commit message: remove unused include

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

Patch to review:
diff --git a/python/ray/tests/test_autoscaler_e2e.py b/python/ray/tests/test_autoscaler_e2e.py
index 45fb8e6f6eb97..559bc693c770f 100644
--- a/python/ray/tests/test_autoscaler_e2e.py
+++ b/python/ray/tests/test_autoscaler_e2e.py
@@ -266,6 +266,105 @@ def test_node_launch_exception_serialization(shutdown_only):
     assert after_serialization.src_exc_info is None
 
 
+def test_infeasible_task_early_cancellation_normal_tasks(shutdown_only):
+    cluster = AutoscalingCluster(
+        head_resources={"CPU": 0},
+        worker_node_types={
+            "type-i": {
+                "resources": {"CPU": 1},
+                "node_config": {},
+                "min_workers": 1,
+                "max_workers": 1,
+            },
+        },
+        autoscaler_v2=True,  # Test with V2 autoscaler only
+    )
+
+    try:
+        import time
+
+        cluster.start()
+        ray.init(address="auto")
+
+        @ray.remote(num_cpus=1)
+        def f():
+            print("Starting f")
+            time.sleep(1)
+            print("Ending f")
+            return 1
+
+        @ray.remote(num_cpus=10)
+        def g():
+            return 2
+
+        f_obj = f.remote()
+        g_obj = g.remote()
+
+        try:
+            [ray.get(f_obj, timeout=5), ray.get(g_obj, timeout=5)]
+        except Exception as e:
+            assert isinstance(e, ray.exceptions.TaskUnschedulableError)
+            assert (
+                "The task is not schedulable: Tasks or actors with resource shapes [{CPU: 100000}] failed to schedule because there are not enough resources for the tasks or actors on the whole cluster."
+                in str(e)
+            )
+
+    finally:
+        cluster.shutdown()
+
+
+def test_infeasible_task_early_cancellation_actor_creation(shutdown_only):
+    cluster = AutoscalingCluster(
+        head_resources={"CPU": 0},
+        worker_node_types={
+            "type-i": {
+                "resources": {"CPU": 1},
+                "node_config": {},
+                "min_workers": 1,
+                "max_workers": 1,
+            },
+        },
+        autoscaler_v2=True,  # Test with V2 autoscaler only
+    )
+
+    try:
+        import time
+
+        cluster.start()
+        ray.init(address="auto")
+
+        @ray.remote(num_cpus=1)
+        class Actor1:
+            def f(self):
+                print("Starting f")
+                time.sleep(1)
+                print("Ending f")
+                return 1
+
+        @ray.remote(num_cpus=10)
+        class Actor2:
+            def f(eslf):
+                return 1
+
+        try:
+            actor_1 = Actor1.remote()
+            actor_2 = Actor2.remote()
+            [
+                ray.get(actor_1.f.remote(), timeout=5),
+                ray.get(actor_2.f.remote(), timeout=5),
+            ]
+        except Exception as e:
+            print(e)
+            assert isinstance(e, ray.exceptions.ActorUnschedulableError)
+            assert (
+                "The actor is not schedulable: Tasks or actors with resource shapes [{CPU: 100000}] failed to schedule because there are not enough resources for the tasks or actors on the whole cluster."
+                in str(e)
+            )
+
+    finally:
+        cluster.shutdown()
+
+
 if __name__ == "__main__":
     import os
     import pytest
diff --git a/src/ray/common/grpc_util.h b/src/ray/common/grpc_util.h
index c47933b459316..7077d2d3c8b14 100644
--- a/src/ray/common/grpc_util.h
+++ b/src/ray/common/grpc_util.h
@@ -156,6 +156,23 @@ inline absl::flat_hash_map<K, V> MapFromProtobuf(
   return absl::flat_hash_map<K, V>(pb_map.begin(), pb_map.end());
 }
 
+/// Debug string for a google protobuf map.
+template <class K, class V>
+inline std::string DebugString(const ::google::protobuf::Map<K, V> &pb_map) {
+  std::stringstream ss;
+  ss << "{";
+  bool first = true;
+  for (const auto &pair : pb_map) {
+    if (!first) {
+      ss << ",";
+      first = false;
+    }
+    ss << pair.first << ":" << pair.second;
+  }
+  ss << "}";
+  return ss.str();
+}
+
 /// Check whether 2 google::protobuf::Map are equal. This function assumes that the
 /// value of the map is either a simple type that supports operator== or a protobuf
 /// message that can be compared using
diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h
index fbdbf4a4d4854..84ad62978397f 100644
--- a/src/ray/common/ray_config_def.h
+++ b/src/ray/common/ray_config_def.h
@@ -942,3 +942,10 @@ RAY_CONFIG(std::vector<std::string>, enable_export_api_write_config, {})
 
 // Configuration for pipe logger buffer size.
 RAY_CONFIG(uint64_t, pipe_logger_read_buf_size, 1024)
+
+// Configuration to enable or disable the infeasible task early termination feature. If
+// set to true, whenever the autoscaler detects that a task is infeasible, the task will
+// be cancelled with SCHEDULING_CANCELLED_UNSCHEDULABLE state. If set to false, the task
+// will be waiting for the required resources to become available achieved mainly by
+// updating the cluster config to add new nodes.
+RAY_CONFIG(bool, enable_infeasible_task_early_exit, true);
diff --git a/src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.cc b/src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.cc
index aa6faa52dd6ad..4d3ace6e42bd4 100644
--- a/src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.cc
+++ b/src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.cc
@@ -14,8 +14,6 @@
 
 #include "ray/gcs/gcs_server/gcs_autoscaler_state_manager.h"
 
-#include <type_traits>
-
 #include "ray/gcs/gcs_server/gcs_actor_manager.h"
 #include "ray/gcs/gcs_server/gcs_node_manager.h"
 #include "ray/gcs/gcs_server/gcs_placement_group_manager.h"
@@ -61,8 +59,6 @@ void GcsAutoscalerStateManager::HandleReportAutoscalingState(
     rpc::autoscaler::ReportAutoscalingStateReply *reply,
     rpc::SendReplyCallback send_reply_callback) {
   RAY_CHECK(thread_checker_.IsOnSameThread());
-  // TODO(rickyx): We should handle the infeasible requests in the future.
-  // Right now, this info will only be used for observability, i.e. ray status.
 
   // Never seen any autoscaling state before - so just takes this.
   if (!autoscaling_state_.has_value()) {
@@ -71,6 +67,16 @@ void GcsAutoscalerStateManager::HandleReportAutoscalingState(
     return;
   }
 
+  // Cancel the infeasible requests if the feature is enabled
+  std::function<void()> callback = [this]() {
+    bool enable_infeasible_task_early_exit =
+        RayConfig::instance().enable_infeasible_task_early_exit();
+
+    if (enable_infeasible_task_early_exit) {
+      this->CancelInfeasibleRequests();
+    }
+  };
+
   // We have a state cached. We discard the incoming state if it's older than the
   // cached state.
   if (request.autoscaling_state().autoscaler_state_version() <
@@ -80,13 +86,13 @@ void GcsAutoscalerStateManager::HandleReportAutoscalingState(
                   << ", received version: "
                   << request.autoscaling_state().autoscaler_state_version()
                   << ". Discarding incoming request.";
-    send_reply_callback(ray::Status::OK(), nullptr, nullptr);
+    send_reply_callback(ray::Status::OK(), callback, nullptr);
     return;
   }
 
   // We should overwrite the cache version.
   autoscaling_state_ = std::move(*request.mutable_autoscaling_state());
-  send_reply_callback(ray::Status::OK(), nullptr, nullptr);
+  send_reply_callback(ray::Status::OK(), callback, nullptr);
 }
 
 void GcsAutoscalerStateManager::HandleRequestClusterResourceConstraint(
@@ -466,37 +472,45 @@ GcsAutoscalerStateManager::GetPerNodeInfeasibleResourceRequests() const {
     return per_node_infeasible_requests;
   }
 
-  // obtain the infeasible requests from the autoscaler state
+  // Early return if there is no infeasible resource requests
+  auto infeasible_resource_shapes_size =
+      autoscaling_state_.value().infeasible_resource_requests_size();
+  if (infeasible_resource_shapes_size == 0) {
+    return per_node_infeasible_requests;
+  }
+
+  // Obtain the infeasible requests from the autoscaler state
   std::vector<google::protobuf::Map<std::string, double>>
       autoscaler_infeasible_resource_shapes;
-  autoscaler_infeasible_resource_shapes.reserve(
-      autoscaling_state_.value().infeasible_resource_requests_size());
-  for (int i = 0; i < autoscaling_state_.value().infeasible_resource_requests_size();
-       i++) {
+  autoscaler_infeasible_resource_shapes.reserve(infeasible_resource_shapes_size);
+  for (int i = 0; i < infeasible_resource_shapes_size; i++) {
     autoscaler_infeasible_resource_shapes.emplace_back(
         autoscaling_state_.value().infeasible_resource_requests(i).resources_bundle());
   }
 
-  // collect the infeasible requests per node
+  // Collect the infeasible requests per node
   for (const auto &[node_id, time_resource_data_pair] : node_resource_info_) {
+    // Iterate through the resource load on each nodes
     const auto &resource_load_by_shape =
         time_resource_data_pair.second.resource_load_by_shape();
 
     for (int i = 0; i < resource_load_by_shape.resource_demands_size(); i++) {
       bool is_infeasible_shape = false;
-      auto &infeasible_resource_shape = autoscaler_infeasible_resource_shapes.at(0);
+      google::protobuf::Map<std::string, double> const *infeasible_resource_shape;
+
+      // Check with each infeasible resource shapes from the autoscaler state
       for (const auto &shape : autoscaler_infeasible_resource_shapes) {
         const auto &resource_demand = resource_load_by_shape.resource_demands(i);
         if (resource_demand.num_infeasible_requests_queued() > 0 &&
             MapEqual(shape, resource_demand.shape())) {
           is_infeasible_shape = true;
-          infeasible_resource_shape = shape;
+          infeasible_resource_shape = &shape;
           break;
         }
       }
 
       if (is_infeasible_shape) {
-        per_node_infeasible_requests[node_id].emplace_back(infeasible_resource_shape);
+        per_node_infeasible_requests[node_id].emplace_back(*infeasible_resource_shape);
       }
     }
   }
@@ -512,9 +526,41 @@ void GcsAutoscalerStateManager::CancelInfeasibleRequests() const {
     return;
   }
 
-  // Cancel the infeasible requests
-  // NOTE(mengjin): This is just a stub for now, but will soon be implemented to support
-  // cancel infeasible resource requests
+  // Cancel the infeasible requests for each nodes
+  for (const auto &node_infeasible_request_pair : per_node_infeasible_requests) {
+    const auto &node_id = node_infeasible_request_pair.first;
+    const auto &infeasible_shapes = node_infeasible_request_pair.second;
+    const auto raylet_client = raylet_client_pool_.GetOrConnectByID(node_id);
+
+    if (raylet_client.has_value()) {
+      std::stringstream resource_shape_str;
+      resource_shape_str << "[";
+      bool first = true;
+      for (const auto &shape : infeasible_shapes) {
+        if (!first) {
+          resource_shape_str << ", ";
+        }
+        resource_shape_str << ray::DebugString(shape);
+        first = false;
+      }
+      resource_shape_str << "]";
+
+      RAY_LOG(INFO) << "Canceling infeasible requests on node " << node_id
+                    << " with infeasible_shapes=" << resource_shape_str.str();
+
+      (*raylet_client)
+          ->CancelTasksWithResourceShapes(
+              infeasible_shapes,
+              [node_id](
+                  const Status &status /*unused*/,
+                  const rpc::CancelTasksWithResourceShapesReply &&reply /*unused*/) {
+                RAY_LOG(INFO) << "Infeasible tasks cancelled on node " << node_id;
+              });
+    } else {
+      RAY_LOG(WARNING) << "Failed to cancel infeasible requests on node " << node_id
+                       << ". Raylet client to the node is not available.";
+    }
+  }
 }
 
 }  // namespace gcs
diff --git a/src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.h b/src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.h
index 5ef4413367ee4..f672825544f21 100644
--- a/src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.h
+++ b/src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.h
@@ -217,7 +217,10 @@ class GcsAutoscalerStateManager : public rpc::autoscaler::AutoscalerStateHandler
   ThreadChecker thread_checker_;
 
   FRIEND_TEST(GcsAutoscalerStateManagerTest, TestReportAutoscalingState);
-  FRIEND_TEST(GcsAutoscalerStateManagerTest, TestGetPerNodeInfeasibleResourceRequests);
+  FRIEND_TEST(GcsAutoscalerStateManagerTest,
+              TestGetPerNodeInfeasibleResourceRequests_NoInfeasibleRequests);
+  FRIEND_TEST(GcsAutoscalerStateManagerTest,
+              TestGetPerNodeInfeasibleResourceRequests_WithInfeasibleRequests);
 };
 
 }  // namespace gcs
diff --git a/src/ray/gcs/gcs_server/test/gcs_autoscaler_state_manager_test.cc b/src/ray/gcs/gcs_server/test/gcs_autoscaler_state_manager_test.cc
index 9e05383ec5887..161a482d3bff1 100644
--- a/src/ray/gcs/gcs_server/test/gcs_autoscaler_state_manager_test.cc
+++ b/src/ray/gcs/gcs_server/test/gcs_autoscaler_state_manager_test.cc
@@ -851,7 +851,67 @@ TEST_F(GcsAutoscalerStateManagerTest, TestGcsKvManagerInternalConfig) {
   EXPECT_EQ(reply.config(), kRayletConfig);
 }
 
-TEST_F(GcsAutoscalerStateManagerTest, TestGetPerNodeInfeasibleResourceRequests) {
+TEST_F(GcsAutoscalerStateManagerTest,
+       TestGetPerNodeInfeasibleResourceRequests_NoInfeasibleRequests) {
+  // Prepare
+  auto node_1 = Mocker::GenNodeInfo();
+  auto node_2 = Mocker::GenNodeInfo();
+
+  // Add nodes
+  {
+    node_1->mutable_resources_total()->insert({"CPU", 2});
+    node_1->set_instance_id("instance_1");
+    AddNode(node_1);
+    node_2->mutable_resources_total()->insert({"CPU", 1});
+    node_2->set_instance_id("instance_2");
+    AddNode(node_2);
+  }
+
+  // Update resource usages
+  {
+    UpdateResourceLoads(node_1->node_id(),
+                        {Mocker::GenResourceDemand({{"GPU", 1}},
+                                                   /* nun_ready_queued */ 1,
+                                                   /* nun_infeasible */ 1,
+                                                   /* num_backlog */ 0),
+                         Mocker::GenResourceDemand({{"CPU", 1}},
+                                                   /* nun_ready_queued */ 1,
+                                                   /* nun_infeasible */ 0,
+                                                   /* num_backlog */ 1),
+                         Mocker::GenResourceDemand({{"CPU", 3}},
+                                                   /* num_ready_queued */ 0,
+                                                   /* num_infeasible */ 1,
+                                                   /* num_backlog */ 1)});
+    UpdateResourceLoads(node_2->node_id(),
+                        {Mocker::GenResourceDemand({{"CPU", 2}},
+                                                   /* nun_ready_queued */ 1,
+                                                   /* nun_infeasible */ 0,
+                                                   /* num_backlog */ 1)});
+  }
+
+  // Update autoscaling state
+  {
+    rpc::autoscaler::AutoscalingState actual_state;
+    actual_state.set_autoscaler_state_version(1);
+    ReportAutoscalingState(actual_state);
+  }
+
+  // Execute
+  const auto per_node_infeasible_requests =
+      gcs_autoscaler_state_manager_->GetPerNodeInfeasibleResourceRequests();
+
+  // Verify
+  { ASSERT_TRUE(per_node_infeasible_requests.empty()); }
+
+  // Reset
+  {
+    RemoveNode(node_1);
+    RemoveNode(node_2);
+  }
+}
+
+TEST_F(GcsAutoscalerStateManagerTest,
+       TestGetPerNodeInfeasibleResourceRequests_WithInfeasibleRequests) {
   // Prepare
   auto node_1 = Mocker::GenNodeInfo();
   auto node_2 = Mocker::GenNodeInfo();
diff --git a/src/ray/raylet/scheduling/cluster_task_manager.cc b/src/ray/raylet/scheduling/cluster_task_manager.cc
index c2b5b552b9042..99be772e28d98 100644
--- a/src/ray/raylet/scheduling/cluster_task_manager.cc
+++ b/src/ray/raylet/scheduling/cluster_task_manager.cc
@@ -147,11 +147,15 @@ bool ClusterTaskManager::CancelTasksWithResourceShapes(
   }
   resource_shapes_str << "]";
 
+  RAY_LOG(INFO) << "Cancelling infeasible tasks with resource shapes "
+                << resource_shapes_str.str();
+
   return CancelTasks(
       predicate,
       rpc::RequestWorkerLeaseReply::SCHEDULING_CANCELLED_UNSCHEDULABLE,
-      "Canceling tasks with resource shapes " + resource_shapes_str.str() +
-          " because there are not enough resources for the tasks on the whole cluster.");
+      "Tasks or actors with resource shapes " + resource_shapes_str.str() +
+          " failed to schedule because there are not enough resources for the tasks "
+          "or actors on the whole cluster.");
 }
 
 bool ClusterTaskManager::IsWorkWithResourceShape(


Code review comments:
diff_hunk: @@ -266,6 +266,105 @@ def test_node_launch_exception_serialization(shutdown_only):
     assert after_serialization.src_exc_info is None
 
 
+def test_infeasible_task_early_cancellation_normal_tasks(shutdown_only):
+    cluster = AutoscalingCluster(
+        head_resources={"CPU": 0},
+        worker_node_types={
+            "type-i": {
+                "resources": {"CPU": 1},
+                "node_config": {},
+                "min_workers": 1,
+                "max_workers": 1,
+            },
+        },
+        autoscaler_v2=True,  # Test with V2 autoscaler only
+    )
+
+    try:
+        import time
+
+        cluster.start()
+        ray.init(address="auto")
+
+        @ray.remote(num_cpus=1)
+        def f():
+            print("Starting f")
+            time.sleep(1)
+            print("Ending f")
+            return 1
+
+        @ray.remote(num_cpus=10)
+        def g():
+            return 2
+
+        f_obj = f.remote()
+        g_obj = g.remote()
+
+        try:
+            [ray.get(f_obj, timeout=5), ray.get(g_obj, timeout=5)]
+        except Exception as e:
+            assert isinstance(e, ray.exceptions.TaskUnschedulableError)
+            assert (
+                "The task is not schedulable: Tasks or actors with resource shapes [{CPU: 100000}] failed to schedule because there are not enough resources for the tasks or actors on the whole cluster."
+                in str(e)
+            )
+
+    finally:
+        cluster.shutdown()
+
+
+def test_infeasible_task_early_cancellation_actor_creation(shutdown_only):
+    cluster = AutoscalingCluster(
+        head_resources={"CPU": 0},
+        worker_node_types={
+            "type-i": {
+                "resources": {"CPU": 1},
+                "node_config": {},
+                "min_workers": 1,
+                "max_workers": 1,
+            },
+        },
+        autoscaler_v2=True,  # Test with V2 autoscaler only
+    )
+
+    try:
+        import time
+
+        cluster.start()
+        ray.init(address="auto")
+
+        @ray.remote(num_cpus=1)
+        class Actor1:
+            def f(self):
+                print("Starting f")
+                time.sleep(1)
+                print("Ending f")
+                return 1
+
+        @ray.remote(num_cpus=10)
+        class Actor2:
+            def f(eslf):
path: python/ray/tests/test_autoscaler_e2e.py
text: @user1:
```suggestion
            def f(self):
```

diff_hunk: @@ -266,6 +266,105 @@ def test_node_launch_exception_serialization(shutdown_only):
     assert after_serialization.src_exc_info is None
 
 
+def test_infeasible_task_early_cancellation_normal_tasks(shutdown_only):
+    cluster = AutoscalingCluster(
+        head_resources={"CPU": 0},
+        worker_node_types={
+            "type-i": {
+                "resources": {"CPU": 1},
+                "node_config": {},
+                "min_workers": 1,
+                "max_workers": 1,
+            },
+        },
+        autoscaler_v2=True,  # Test with V2 autoscaler only
+    )
+
+    try:
+        import time
+
+        cluster.start()
+        ray.init(address="auto")
+
+        @ray.remote(num_cpus=1)
+        def f():
+            print("Starting f")
+            time.sleep(1)
+            print("Ending f")
+            return 1
path: python/ray/tests/test_autoscaler_e2e.py
text: @user1:
shouldn't this one succeed as written? and why is there a `time.sleep(1)` in it?

@author:
replied in the above comment

diff_hunk: @@ -266,6 +266,105 @@ def test_node_launch_exception_serialization(shutdown_only):
     assert after_serialization.src_exc_info is None
 
 
+def test_infeasible_task_early_cancellation_normal_tasks(shutdown_only):
+    cluster = AutoscalingCluster(
+        head_resources={"CPU": 0},
+        worker_node_types={
+            "type-i": {
+                "resources": {"CPU": 1},
+                "node_config": {},
+                "min_workers": 1,
+                "max_workers": 1,
+            },
+        },
+        autoscaler_v2=True,  # Test with V2 autoscaler only
+    )
+
+    try:
+        import time
+
+        cluster.start()
+        ray.init(address="auto")
+
+        @ray.remote(num_cpus=1)
+        def f():
+            print("Starting f")
+            time.sleep(1)
+            print("Ending f")
+            return 1
+
+        @ray.remote(num_cpus=10)
+        def g():
+            return 2
+
+        f_obj = f.remote()
+        g_obj = g.remote()
+
+        try:
+            [ray.get(f_obj, timeout=5), ray.get(g_obj, timeout=5)]
+        except Exception as e:
+            assert isinstance(e, ray.exceptions.TaskUnschedulableError)
+            assert (
+                "The task is not schedulable: Tasks or actors with resource shapes [{CPU: 100000}] failed to schedule because there are not enough resources for the tasks or actors on the whole cluster."
+                in str(e)
+            )
path: python/ray/tests/test_autoscaler_e2e.py
text: @user1:
```suggestion
        with pytest.raises(ray.exceptions.TaskUnschedulableError, match="The task is not schedulable: ..."):
```

diff_hunk: @@ -942,3 +942,10 @@ RAY_CONFIG(std::vector<std::string>, enable_export_api_write_config, {})
 
 // Configuration for pipe logger buffer size.
 RAY_CONFIG(uint64_t, pipe_logger_read_buf_size, 1024)
+
+// Configuration to enable or disable the infeasible task early termination feature. If
+// set to true, whenever the autoscaler detects that a task is infeasible, the task will
+// be cancelled with SCHEDULING_CANCELLED_UNSCHEDULABLE state. If set to false, the task
+// will be waiting for the required resources to become available achieved mainly by
+// updating the cluster config to add new nodes.
+RAY_CONFIG(bool, enable_infeasible_task_early_exit, true);
path: src/ray/common/ray_config_def.h
text: @user1:
Will this be a breaking behavior change for any workloads? If so, we will need to do a slow rollout and start with it off by default.

@author:
Yes. This will be a breaking change for any workloads using autoscaler V2. 

Will change to false by default.

diff_hunk: @@ -512,9 +526,41 @@ void GcsAutoscalerStateManager::CancelInfeasibleRequests() const {
     return;
   }
 
-  // Cancel the infeasible requests
-  // NOTE(mengjin): This is just a stub for now, but will soon be implemented to support
-  // cancel infeasible resource requests
+  // Cancel the infeasible requests for each nodes
+  for (const auto &node_infeasible_request_pair : per_node_infeasible_requests) {
+    const auto &node_id = node_infeasible_request_pair.first;
+    const auto &infeasible_shapes = node_infeasible_request_pair.second;
+    const auto raylet_client = raylet_client_pool_.GetOrConnectByID(node_id);
+
+    if (raylet_client.has_value()) {
+      std::stringstream resource_shape_str;
+      resource_shape_str << "[";
+      bool first = true;
+      for (const auto &shape : infeasible_shapes) {
+        if (!first) {
+          resource_shape_str << ", ";
+        }
+        resource_shape_str << ray::DebugString(shape);
+        first = false;
+      }
+      resource_shape_str << "]";
path: src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.cc
text: @user1:
put this in a util for readability pls

diff_hunk: @@ -147,11 +147,15 @@ bool ClusterTaskManager::CancelTasksWithResourceShapes(
   }
   resource_shapes_str << "]";
 
+  RAY_LOG(INFO) << "Cancelling infeasible tasks with resource shapes "
path: src/ray/raylet/scheduling/cluster_task_manager.cc
text: @user1:
this should be warning level IMO

diff_hunk: @@ -266,6 +266,105 @@ def test_node_launch_exception_serialization(shutdown_only):
     assert after_serialization.src_exc_info is None
 
 
+def test_infeasible_task_early_cancellation_normal_tasks(shutdown_only):
+    cluster = AutoscalingCluster(
+        head_resources={"CPU": 0},
+        worker_node_types={
+            "type-i": {
+                "resources": {"CPU": 1},
+                "node_config": {},
+                "min_workers": 1,
+                "max_workers": 1,
+            },
+        },
+        autoscaler_v2=True,  # Test with V2 autoscaler only
+    )
+
+    try:
+        import time
+
+        cluster.start()
+        ray.init(address="auto")
+
+        @ray.remote(num_cpus=1)
+        def f():
+            print("Starting f")
+            time.sleep(1)
+            print("Ending f")
+            return 1
+
+        @ray.remote(num_cpus=10)
+        def g():
+            return 2
+
+        f_obj = f.remote()
+        g_obj = g.remote()
+
+        try:
+            [ray.get(f_obj, timeout=5), ray.get(g_obj, timeout=5)]
+        except Exception as e:
+            assert isinstance(e, ray.exceptions.TaskUnschedulableError)
+            assert (
+                "The task is not schedulable: Tasks or actors with resource shapes [{CPU: 100000}] failed to schedule because there are not enough resources for the tasks or actors on the whole cluster."
+                in str(e)
+            )
+
+    finally:
+        cluster.shutdown()
+
+
+def test_infeasible_task_early_cancellation_actor_creation(shutdown_only):
+    cluster = AutoscalingCluster(
+        head_resources={"CPU": 0},
+        worker_node_types={
+            "type-i": {
+                "resources": {"CPU": 1},
+                "node_config": {},
+                "min_workers": 1,
+                "max_workers": 1,
+            },
+        },
+        autoscaler_v2=True,  # Test with V2 autoscaler only
+    )
+
+    try:
+        import time
+
+        cluster.start()
+        ray.init(address="auto")
+
+        @ray.remote(num_cpus=1)
+        class Actor1:
+            def f(self):
+                print("Starting f")
+                time.sleep(1)
+                print("Ending f")
path: python/ray/tests/test_autoscaler_e2e.py
text: @user1:
remove prints

@user1:
why is there a body to this function at all?

@author:
This particular function is supposed to be a feasible task. Here I want to test when we cancel the infeasible tasks, whether the feasible ones will be executed until complete. 

At the same time, agree that I should remove the prints and the sleep.

@user1:
we should not rely on sleeping behavior in any tests unless absolutely necessary

if you want to test this behavior, what you can do is:

1. start the feasible task and have it block waiting on a `SignalActor` when it begins running (from test_utils)
2. in the driver, wait until that signal is set, then spawn the infeasible task
3. check that the infeasible task is cancelled
4. unblock the feasible task by setting the signal it is blocked on, then `ray.get` it and verify it succeeded

diff_hunk: @@ -266,6 +266,105 @@ def test_node_launch_exception_serialization(shutdown_only):
     assert after_serialization.src_exc_info is None
 
 
+def test_infeasible_task_early_cancellation_normal_tasks(shutdown_only):
+    cluster = AutoscalingCluster(
+        head_resources={"CPU": 0},
+        worker_node_types={
+            "type-i": {
+                "resources": {"CPU": 1},
+                "node_config": {},
+                "min_workers": 1,
+                "max_workers": 1,
+            },
+        },
+        autoscaler_v2=True,  # Test with V2 autoscaler only
+    )
+
+    try:
+        import time
+
+        cluster.start()
+        ray.init(address="auto")
+
+        @ray.remote(num_cpus=1)
+        def f():
+            print("Starting f")
+            time.sleep(1)
+            print("Ending f")
+            return 1
+
+        @ray.remote(num_cpus=10)
+        def g():
+            return 2
+
+        f_obj = f.remote()
+        g_obj = g.remote()
+
+        try:
+            [ray.get(f_obj, timeout=5), ray.get(g_obj, timeout=5)]
+        except Exception as e:
+            assert isinstance(e, ray.exceptions.TaskUnschedulableError)
+            assert (
+                "The task is not schedulable: Tasks or actors with resource shapes [{CPU: 100000}] failed to schedule because there are not enough resources for the tasks or actors on the whole cluster."
+                in str(e)
+            )
+
+    finally:
+        cluster.shutdown()
path: python/ray/tests/test_autoscaler_e2e.py
text: @user1:
all cleanup should be handled in a fixture to guard against resource leaks when the code fails in unexpected ways (or if someone modifies in it in the future)

the `shutdown_only` future might already be doing what you want

@author:
Here I was mainly follow what other tests in the same file are doing. 

Also, I checked the code. From what I see, `cluster.shutdown()` does `ray stop --force` and the `shutdown_only` fixture does `ray.shutdown()` and `ray._private.utils.reset_ray_address()`. 

The 2 functions seems to handle the cleanup of different init method. `cluster.shutdown()` mainly kills the processes that created by `cluster.start()` which uses `ray start`. And `shutdown_only` fixture mainly handles the cleanup for `ray.init()`. The 2 shutdown methods are in different code paths and I'll need to further check whether they are fully overlapped or there are some differences. 

Also, if there are any differences, do you think it is worthwhile we put the cleanup of the cluster also in a python fixture?

@user1:
Yes let's move the cleanup logic to a new fixture if needed

Perhaps we already have one that creates the cluster and passes it in as an argument?

@author:
Looks like we don't have one and it seems that all autoscaling e2e tests are using the same pattern since this PR: https://github.com/ray-project/ray/pull/18987

Let me create a fixture and update the existing tests as well.

diff_hunk: @@ -147,11 +147,15 @@ bool ClusterTaskManager::CancelTasksWithResourceShapes(
   }
   resource_shapes_str << "]";
 
+  RAY_LOG(INFO) << "Cancelling infeasible tasks with resource shapes "
+                << resource_shapes_str.str();
path: src/ray/raylet/scheduling/cluster_task_manager.cc
text: @user1:
save the str to a variable so you don't have to create it the second time

diff_hunk: @@ -512,9 +526,41 @@ void GcsAutoscalerStateManager::CancelInfeasibleRequests() const {
     return;
   }
 
-  // Cancel the infeasible requests
-  // NOTE(mengjin): This is just a stub for now, but will soon be implemented to support
-  // cancel infeasible resource requests
+  // Cancel the infeasible requests for each nodes
+  for (const auto &node_infeasible_request_pair : per_node_infeasible_requests) {
+    const auto &node_id = node_infeasible_request_pair.first;
+    const auto &infeasible_shapes = node_infeasible_request_pair.second;
+    const auto raylet_client = raylet_client_pool_.GetOrConnectByID(node_id);
+
+    if (raylet_client.has_value()) {
+      std::stringstream resource_shape_str;
+      resource_shape_str << "[";
+      bool first = true;
+      for (const auto &shape : infeasible_shapes) {
+        if (!first) {
+          resource_shape_str << ", ";
+        }
+        resource_shape_str << ray::DebugString(shape);
+        first = false;
+      }
+      resource_shape_str << "]";
+
+      RAY_LOG(INFO) << "Canceling infeasible requests on node " << node_id
+                    << " with infeasible_shapes=" << resource_shape_str.str();
path: src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.cc
text: @user1:
this should be warning level IMO.

Also, this log line might spam many times if there are many nodes in the cluster. Should we log it once per shape instead of once per node?

@author:
Warning level sounds good. 

Regarding whether to log at per node or per shape basis, I'm not very familiar of our workloads, do you think it will be the same infeasible tasks spreading across different nodes in most cases? If that's the case, I think logging per shape makes more sense than per node.

diff_hunk: @@ -466,37 +472,45 @@ GcsAutoscalerStateManager::GetPerNodeInfeasibleResourceRequests() const {
     return per_node_infeasible_requests;
   }
 
-  // obtain the infeasible requests from the autoscaler state
+  // Early return if there is no infeasible resource requests
+  auto infeasible_resource_shapes_size =
+      autoscaling_state_.value().infeasible_resource_requests_size();
+  if (infeasible_resource_shapes_size == 0) {
+    return per_node_infeasible_requests;
+  }
+
+  // Obtain the infeasible requests from the autoscaler state
   std::vector<google::protobuf::Map<std::string, double>>
       autoscaler_infeasible_resource_shapes;
-  autoscaler_infeasible_resource_shapes.reserve(
-      autoscaling_state_.value().infeasible_resource_requests_size());
-  for (int i = 0; i < autoscaling_state_.value().infeasible_resource_requests_size();
-       i++) {
+  autoscaler_infeasible_resource_shapes.reserve(infeasible_resource_shapes_size);
+  for (int i = 0; i < infeasible_resource_shapes_size; i++) {
     autoscaler_infeasible_resource_shapes.emplace_back(
         autoscaling_state_.value().infeasible_resource_requests(i).resources_bundle());
   }
 
-  // collect the infeasible requests per node
+  // Collect the infeasible requests per node
   for (const auto &[node_id, time_resource_data_pair] : node_resource_info_) {
+    // Iterate through the resource load on each nodes
     const auto &resource_load_by_shape =
         time_resource_data_pair.second.resource_load_by_shape();
 
     for (int i = 0; i < resource_load_by_shape.resource_demands_size(); i++) {
       bool is_infeasible_shape = false;
-      auto &infeasible_resource_shape = autoscaler_infeasible_resource_shapes.at(0);
+      google::protobuf::Map<std::string, double> const *infeasible_resource_shape;
+
+      // Check with each infeasible resource shapes from the autoscaler state
       for (const auto &shape : autoscaler_infeasible_resource_shapes) {
         const auto &resource_demand = resource_load_by_shape.resource_demands(i);
         if (resource_demand.num_infeasible_requests_queued() > 0 &&
             MapEqual(shape, resource_demand.shape())) {
           is_infeasible_shape = true;
-          infeasible_resource_shape = shape;
+          infeasible_resource_shape = &shape;
path: src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.cc
text: @user1:
just do `per_node_infeasible_requests[node_id].push_back(move(shape))` here. You'll be able to move the shape and simplifies a bit

diff_hunk: @@ -147,11 +147,15 @@ bool ClusterTaskManager::CancelTasksWithResourceShapes(
   }
   resource_shapes_str << "]";
 
+  RAY_LOG(INFO) << "Cancelling infeasible tasks with resource shapes "
+                << resource_shapes_str.str();
+
   return CancelTasks(
       predicate,
       rpc::RequestWorkerLeaseReply::SCHEDULING_CANCELLED_UNSCHEDULABLE,
-      "Canceling tasks with resource shapes " + resource_shapes_str.str() +
-          " because there are not enough resources for the tasks on the whole cluster.");
+      "Tasks or actors with resource shapes " + resource_shapes_str.str() +
+          " failed to schedule because there are not enough resources for the tasks "
+          "or actors on the whole cluster.");
path: src/ray/raylet/scheduling/cluster_task_manager.cc
text: @user1:
prefer absl::StrCat https://abseil.io/docs/cpp/guides/strings#abslstrcat more efficient with multiple ops

diff_hunk: @@ -512,9 +526,41 @@ void GcsAutoscalerStateManager::CancelInfeasibleRequests() const {
     return;
   }
 
-  // Cancel the infeasible requests
-  // NOTE(mengjin): This is just a stub for now, but will soon be implemented to support
-  // cancel infeasible resource requests
+  // Cancel the infeasible requests for each nodes
+  for (const auto &node_infeasible_request_pair : per_node_infeasible_requests) {
+    const auto &node_id = node_infeasible_request_pair.first;
+    const auto &infeasible_shapes = node_infeasible_request_pair.second;
+    const auto raylet_client = raylet_client_pool_.GetOrConnectByID(node_id);
+
+    if (raylet_client.has_value()) {
+      std::stringstream resource_shape_str;
+      resource_shape_str << "[";
+      bool first = true;
+      for (const auto &shape : infeasible_shapes) {
+        if (!first) {
+          resource_shape_str << ", ";
+        }
+        resource_shape_str << ray::DebugString(shape);
+        first = false;
+      }
+      resource_shape_str << "]";
+
+      RAY_LOG(INFO) << "Canceling infeasible requests on node " << node_id
+                    << " with infeasible_shapes=" << resource_shape_str.str();
+
+      (*raylet_client)
+          ->CancelTasksWithResourceShapes(
+              infeasible_shapes,
+              [node_id](
+                  const Status &status /*unused*/,
+                  const rpc::CancelTasksWithResourceShapesReply &&reply /*unused*/) {
path: src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.cc
text: @user1:
you can just leave out the variable names to signify unused i think