Repository: ray-project/ray
Pull request: [Core] Fix check failure due to negative available resource
<!-- 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?

The sequence of events that can trigger the check failure:
1. A node has PG resources `{"CPU_group_aaa": 2, "CPU_group_0_aaa": 1, "CPU_group_1_aaa": 1}`
2. Worker1 acquires 1 PG CPU resource, node available resource becomes `{"CPU_group_aaa": 1, "CPU_group_0_aaa": 0, "CPU_group_1_aaa": 1}`
3. Worker1 calls ray.get() which temporarily released its CPU resource, node available resource becomes  `{"CPU_group_aaa": 2, "CPU_group_0_aaa": 1, "CPU_group_1_aaa": 1}`
4. Worker2 acquires 1 PG CPU resource, node available resource becomes `{"CPU_group_aaa": 1, "CPU_group_0_aaa": 0, "CPU_group_1_aaa": 1}`
5. ray.get() returns from worker1 and it acquires back the resource, node available resource becomes `{"CPU_group_aaa": 0, "CPU_group_0_aaa": -1, "CPU_group_1_aaa": 1}`
6. Worker3 acquires 1 PG CPU resource and the check failure happens since it can get `CPU_group_1_aaa` but not `CPU_group_aaa`. The fix is that we allow `CPU_group_aaa` to go negative as well to match the indexed PG CPU resources.

<!-- Please give a short summary of the change and the problem this solves. -->

## Related issue number
Closes #50433
<!-- 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:
[Core] Crash in raylet (ray::RayLog::~RayLog)
### What happened + What you expected to happen

I am experiencing raylet crashes like this:

```
(raylet) [2025-02-11 18:42:15,859 C 30797 30797] resource_instance_set.cc:374:  Check failed: RAY_PREDICT_TRUE(_left_ >= _right_)  0 vs 10000
(raylet) *** StackTrace Information ***
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0xc17c7a) [0x559ef7d2ac7a] ray::operator<<()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0xc1a0c1) [0x559ef7d2d0c1] ray::RayLog::~RayLog()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x67b7a9) [0x559ef778e7a9] ray::NodeResourceInstanceSet::AllocateWithReference()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x67c545) [0x559ef778f545] ray::NodeResourceInstanceSet::TryAllocate()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x4b4fe1) [0x559ef75c7fe1] ray::LocalResourceManager::AllocateTaskResourceInstances()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x4ba4fd) [0x559ef75cd4fd] ray::LocalResourceManager::AllocateLocalTaskResources()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x4ba5c3) [0x559ef75cd5c3] ray::LocalResourceManager::AllocateLocalTaskResources()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x3b2fd2) [0x559ef74c5fd2] ray::raylet::LocalTaskManager::DispatchScheduledTasksToWorkers()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x3b4262) [0x559ef74c7262] ray::raylet::LocalTaskManager::ScheduleAndDispatchTasks()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x4a1011) [0x559ef75b4011] ray::raylet::ClusterTaskManager::ScheduleAndDispatchTasks()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x32f58f) [0x559ef744258f] ray::raylet::NodeManager::HandleReturnWorker()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x31ed17) [0x559ef7431d17] ray::rpc::ServerCallImpl<>::HandleRequestImpl()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x6a0248) [0x559ef77b3248] EventTracker::RecordExecution()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x69b23e) [0x559ef77ae23e] std::_Function_handler<>::_M_invoke()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x69b6b6) [0x559ef77ae6b6] boost::asio::detail::completion_handler<>::do_complete()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0xc8c2bb) [0x559ef7d9f2bb] boost::asio::detail::scheduler::do_run_one()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0xc8e849) [0x559ef7da1849] boost::asio::detail::scheduler::run()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0xc8ed62) [0x559ef7da1d62] boost::asio::io_context::run()
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x1ecb8f) [0x559ef72ffb8f] main
(raylet) /lib/x86_64-linux-gnu/libc.so.6(+0x2724a) [0x7fa042a6324a]
(raylet) /lib/x86_64-linux-gnu/libc.so.6(__libc_start_main+0x85) [0x7fa042a63305] __libc_start_main
(raylet) /usr/local/lib/python3.10/site-packages/ray/core/src/ray/raylet/raylet(+0x246ec7) [0x559ef7359ec7]
(raylet) 
Traceback (most recent call last):
  File "/code/labs/jakob/ray/test/crash.py", line 33, in <module>
    main()
  File "/code/labs/jakob/ray/test/crash.py", line 30, in main
    ray.get(tasks)
  File "/usr/local/lib/python3.10/site-packages/ray/_private/auto_init_hook.py", line 21, in auto_init_wrapper
    return fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/ray/_private/client_mode_hook.py", line 103, in wrapper
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/ray/_private/worker.py", line 2772, in get
    values, debugger_breakpoint = worker.get_objects(object_refs, timeout=timeout)
  File "/usr/local/lib/python3.10/site-packages/ray/_private/worker.py", line 921, in get_objects
    raise value
ray.exceptions.LocalRayletDiedError: The task's local raylet died. Check raylet.out for more information.
```

The above error is reproduced by the Python script provided below. To reproduce, one can simply run the script as `python3 script.py` and let Ray start up a local cluster.

However, it is worth noting that I am experiencing the same issue in my productions system on KubeRay with separate head and worker nodes.

### Versions / Dependencies

Ray version: 2.41.0

### Reproduction script

```
import os
os.environ['RAY_DEDUP_LOGS'] = "0"
import ray
from ray.util.placement_group import placement_group
from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
import time

ray.init()

# Limit number of worker nodes
NUM_WORKERS = 6
pg = placement_group([{"CPU": 1}] * NUM_WORKERS)
pg_strategy = PlacementGroupSchedulingStrategy(pg)

@ray.remote(scheduling_strategy=pg_strategy)
def g(i: int):
    print(f"g {i}")
    time.sleep(0.1)

@ray.remote(scheduling_strategy=pg_strategy)
def f():
    print("f")
    ray.get([g.remote(i) for i in range(10)])
    print("f done")

def main():
    tasks = []
    for i in range(4):
        tasks.append(f.remote())
    ray.get(tasks)

main()
```

### Issue Severity

High: It blocks me from completing my task.

Hints from issue discussion:

Commit message: up

Signed-off-by: Jiajun Yao <jeromeyjj@gmail.com>

Patch to review:
diff --git a/src/ray/common/bundle_spec.cc b/src/ray/common/bundle_spec.cc
index 833ec7c5a82ec..7624919f273e5 100644
--- a/src/ray/common/bundle_spec.cc
+++ b/src/ray/common/bundle_spec.cc
@@ -138,6 +138,23 @@ std::string GetOriginalResourceNameFromWildcardResource(const std::string &resou
   }
 }
 
+bool IsCPUOrPlacementGroupCPUResource(ResourceID resource_id) {
+  // Check whether the resource is CPU resource or CPU resource inside PG.
+  if (resource_id == ResourceID::CPU()) {
+    return true;
+  }
+
+  auto possible_pg_resource = ParsePgFormattedResource(resource_id.Binary(),
+                                                       /*for_wildcard_resource*/ true,
+                                                       /*for_indexed_resource*/ true);
+  if (possible_pg_resource.has_value() &&
+      possible_pg_resource->original_resource == ResourceID::CPU().Binary()) {
+    return true;
+  }
+
+  return false;
+}
+
 std::optional<PgFormattedResourceData> ParsePgFormattedResource(
     const std::string &resource, bool for_wildcard_resource, bool for_indexed_resource) {
   // Check if it is a wildcard pg resource.
diff --git a/src/ray/common/bundle_spec.h b/src/ray/common/bundle_spec.h
index 1330d4676fa35..b5bc329617997 100644
--- a/src/ray/common/bundle_spec.h
+++ b/src/ray/common/bundle_spec.h
@@ -126,6 +126,8 @@ std::string GetOriginalResourceName(const std::string &resource);
 // Returns "" if the resource is not a wildcard resource.
 std::string GetOriginalResourceNameFromWildcardResource(const std::string &resource);
 
+bool IsCPUOrPlacementGroupCPUResource(ResourceID resource_id);
+
 /// Parse the given resource and get the pg related information.
 ///
 /// \param resource name of the resource.
diff --git a/src/ray/common/scheduling/resource_instance_set.cc b/src/ray/common/scheduling/resource_instance_set.cc
index c9f73d09a7270..43fdf59d75f69 100644
--- a/src/ray/common/scheduling/resource_instance_set.cc
+++ b/src/ray/common/scheduling/resource_instance_set.cc
@@ -371,7 +371,12 @@ void NodeResourceInstanceSet::AllocateWithReference(
   RAY_CHECK_EQ(available.size(), ref_allocation.size());
 
   for (size_t i = 0; i < ref_allocation.size(); i++) {
-    RAY_CHECK_GE(available[i], ref_allocation[i]);
+    if (available[i] < ref_allocation[i]) {
+      // Only CPU resource can go negative due to the behavior
+      // that ray.get() will temporarily release the CPU resource.
+      // See https://github.com/ray-project/ray/pull/50517.
+      RAY_CHECK(IsCPUOrPlacementGroupCPUResource(resource_id));
+    }
     available[i] -= ref_allocation[i];
   }
 
diff --git a/src/ray/raylet/local_task_manager.cc b/src/ray/raylet/local_task_manager.cc
index 2bffd6dc5711a..08be91eb189e9 100644
--- a/src/ray/raylet/local_task_manager.cc
+++ b/src/ray/raylet/local_task_manager.cc
@@ -24,23 +24,6 @@
 namespace ray {
 namespace raylet {
 
-bool IsCPUOrPlacementGroupCPUResource(ResourceID resource_id) {
-  // Check whether the resource is CPU resource or CPU resource inside PG.
-  if (resource_id == ResourceID::CPU()) {
-    return true;
-  }
-
-  auto possible_pg_resource = ParsePgFormattedResource(resource_id.Binary(),
-                                                       /*for_wildcard_resource*/ true,
-                                                       /*for_indexed_resource*/ true);
-  if (possible_pg_resource.has_value() &&
-      possible_pg_resource->original_resource == ResourceID::CPU().Binary()) {
-    return true;
-  }
-
-  return false;
-}
-
 LocalTaskManager::LocalTaskManager(
     const NodeID &self_node_id,
     ClusterResourceScheduler &cluster_resource_scheduler,
diff --git a/src/ray/raylet/scheduling/cluster_task_manager_test.cc b/src/ray/raylet/scheduling/cluster_task_manager_test.cc
index 7fb935861b9aa..0ad63603d2257 100644
--- a/src/ray/raylet/scheduling/cluster_task_manager_test.cc
+++ b/src/ray/raylet/scheduling/cluster_task_manager_test.cc
@@ -1829,7 +1829,51 @@ TEST_F(ClusterTaskManagerTest, FeasibleToNonFeasible) {
             task1.GetTaskSpecification().TaskId());
 }
 
-TEST_F(ClusterTaskManagerTestWithGPUsAtHead, RleaseAndReturnWorkerCpuResources) {
+TEST_F(ClusterTaskManagerTest, NegativePlacementGroupCpuResources) {
+  // Add PG CPU resources.
+  scheduler_->GetLocalResourceManager().AddLocalResourceInstances(
+      scheduling::ResourceID("CPU_group_aaa"), std::vector<FixedPoint>{FixedPoint(2)});
+  scheduler_->GetLocalResourceManager().AddLocalResourceInstances(
+      scheduling::ResourceID("CPU_group_0_aaa"), std::vector<FixedPoint>{FixedPoint(1)});
+  scheduler_->GetLocalResourceManager().AddLocalResourceInstances(
+      scheduling::ResourceID("CPU_group_1_aaa"), std::vector<FixedPoint>{FixedPoint(1)});
+
+  const NodeResources &node_resources =
+      scheduler_->GetClusterResourceManager().GetNodeResources(
+          scheduling::NodeID(id_.Binary()));
+
+  auto worker1 = std::make_shared<MockWorker>(WorkerID::FromRandom(), 1234);
+  auto allocated_instances = std::make_shared<TaskResourceInstances>();
+  ASSERT_TRUE(scheduler_->GetLocalResourceManager().AllocateLocalTaskResources(
+      {{"CPU_group_aaa", 1.}, {"CPU_group_0_aaa", 1.}}, allocated_instances));
+  worker1->SetAllocatedInstances(allocated_instances);
+  // worker1 calls ray.get() and release the CPU resource
+  ASSERT_TRUE(local_task_manager_->ReleaseCpuResourcesFromBlockedWorker(worker1));
+
+  // the released CPU resource is acquired by worker2
+  auto worker2 = std::make_shared<MockWorker>(WorkerID::FromRandom(), 5678);
+  allocated_instances = std::make_shared<TaskResourceInstances>();
+  ASSERT_TRUE(scheduler_->GetLocalResourceManager().AllocateLocalTaskResources(
+      {{"CPU_group_aaa", 1.}, {"CPU_group_0_aaa", 1.}}, allocated_instances));
+  worker2->SetAllocatedInstances(allocated_instances);
+
+  // ray.get() returns and worker1 acquires the CPU resource again
+  ASSERT_TRUE(local_task_manager_->ReturnCpuResourcesToUnblockedWorker(worker1));
+  ASSERT_EQ(node_resources.available.Get(scheduling::ResourceID("CPU_group_aaa")), 0);
+  ASSERT_EQ(node_resources.available.Get(scheduling::ResourceID("CPU_group_0_aaa")), -1);
+  ASSERT_EQ(node_resources.available.Get(scheduling::ResourceID("CPU_group_1_aaa")), 1);
+
+  auto worker3 = std::make_shared<MockWorker>(WorkerID::FromRandom(), 7678);
+  allocated_instances = std::make_shared<TaskResourceInstances>();
+  ASSERT_TRUE(scheduler_->GetLocalResourceManager().AllocateLocalTaskResources(
+      {{"CPU_group_aaa", 1.}, {"CPU_group_1_aaa", 1.}}, allocated_instances));
+  worker3->SetAllocatedInstances(allocated_instances);
+  ASSERT_EQ(node_resources.available.Get(scheduling::ResourceID("CPU_group_aaa")), -1);
+  ASSERT_EQ(node_resources.available.Get(scheduling::ResourceID("CPU_group_0_aaa")), -1);
+  ASSERT_EQ(node_resources.available.Get(scheduling::ResourceID("CPU_group_1_aaa")), 0);
+}
+
+TEST_F(ClusterTaskManagerTestWithGPUsAtHead, ReleaseAndReturnWorkerCpuResources) {
   // Add PG CPU and GPU resources.
   scheduler_->GetLocalResourceManager().AddLocalResourceInstances(
       scheduling::ResourceID("CPU_group_aaa"), std::vector<FixedPoint>{FixedPoint(1)});


Code review comments:
diff_hunk: @@ -371,7 +371,12 @@ void NodeResourceInstanceSet::AllocateWithReference(
   RAY_CHECK_EQ(available.size(), ref_allocation.size());
 
   for (size_t i = 0; i < ref_allocation.size(); i++) {
-    RAY_CHECK_GE(available[i], ref_allocation[i]);
+    if (available[i] < ref_allocation[i]) {
+      // Only CPU resource can go negative due to the behavior
+      // that ray.get() will temporarily release the CPU resource.
+      // See https://github.com/ray-project/ray/pull/50517.
+      RAY_CHECK(IsCPUOrPlacementGroupCPUResource(resource_id));
path: src/ray/common/scheduling/resource_instance_set.cc
text: @user1:
Nit: The result of resource id check failure can be confusing. I'm wondering if it is better to use `RAY_CHECK_WITH_DISPLAY` and with a explicit error message to it.  

@user2:
`RAY_CHECK(...) << ...` should include more information