Implement resource deadlock detection for new scheduler (#12961)

python/ray/test_utils.py

@@ -414,7 +414,7 @@ def init_error_pubsub():
     return p
 
 
-def get_error_message(pub_sub, num, error_type=None, timeout=5):
+def get_error_message(pub_sub, num, error_type=None, timeout=20):
     """Get errors through pub/sub."""
     start_time = time.time()
     msgs = []

python/ray/tests/test_failure.py

@@ -17,7 +17,8 @@ import ray.ray_constants as ray_constants
 from ray.exceptions import RayTaskError
 from ray.cluster_utils import Cluster
 from ray.test_utils import (wait_for_condition, SignalActor, init_error_pubsub,
-                            get_error_message, Semaphore)
+                            get_error_message, Semaphore,
+                            new_scheduler_enabled)
 
 
 def test_failed_task(ray_start_regular, error_pubsub):
@@ -632,11 +633,12 @@ def test_export_large_objects(ray_start_regular, error_pubsub):
     assert errors[0].type == ray_constants.PICKLING_LARGE_OBJECT_PUSH_ERROR
 
 
-@pytest.mark.skip(reason="TODO detect resource deadlock")
-def test_warning_for_resource_deadlock(error_pubsub, shutdown_only):
-    p = error_pubsub
-    # Check that we get warning messages for infeasible tasks.
-    ray.init(num_cpus=1)
+@pytest.mark.skipif(
+    new_scheduler_enabled(), reason="Supposed to deadlock, but it doesn't")
+def test_warning_all_tasks_blocked(shutdown_only):
+    ray.init(
+        num_cpus=1, _system_config={"debug_dump_period_milliseconds": 500})
+    p = init_error_pubsub()
 
     @ray.remote(num_cpus=1)
     class Foo:
@@ -646,7 +648,7 @@ def test_warning_for_resource_deadlock(error_pubsub, shutdown_only):
     @ray.remote
     def f():
         # Creating both actors is not possible.
-        actors = [Foo.remote() for _ in range(2)]
+        actors = [Foo.remote() for _ in range(3)]
         for a in actors:
             ray.get(a.f.remote())
 
@@ -657,6 +659,46 @@ def test_warning_for_resource_deadlock(error_pubsub, shutdown_only):
     assert errors[0].type == ray_constants.RESOURCE_DEADLOCK_ERROR
 
 
+def test_warning_actor_waiting_on_actor(shutdown_only):
+    ray.init(
+        num_cpus=1, _system_config={"debug_dump_period_milliseconds": 500})
+    p = init_error_pubsub()
+
+    @ray.remote(num_cpus=1)
+    class Actor:
+        pass
+
+    a = Actor.remote()  # noqa
+    b = Actor.remote()  # noqa
+
+    errors = get_error_message(p, 1, ray_constants.RESOURCE_DEADLOCK_ERROR)
+    assert len(errors) == 1
+    assert errors[0].type == ray_constants.RESOURCE_DEADLOCK_ERROR
+
+
+def test_warning_task_waiting_on_actor(shutdown_only):
+    ray.init(
+        num_cpus=1, _system_config={"debug_dump_period_milliseconds": 500})
+    p = init_error_pubsub()
+
+    @ray.remote(num_cpus=1)
+    class Actor:
+        pass
+
+    a = Actor.remote()  # noqa
+
+    @ray.remote(num_cpus=1)
+    def f():
+        print("f running")
+        time.sleep(999)
+
+    ids = [f.remote()]  # noqa
+
+    errors = get_error_message(p, 1, ray_constants.RESOURCE_DEADLOCK_ERROR)
+    assert len(errors) == 1
+    assert errors[0].type == ray_constants.RESOURCE_DEADLOCK_ERROR
+
+
 def test_warning_for_infeasible_tasks(ray_start_regular, error_pubsub):
     p = error_pubsub
     # Check that we get warning messages for infeasible tasks.

python/ray/tests/test_global_gc.py

@@ -9,7 +9,7 @@ import pytest
 
 import ray
 import ray.cluster_utils
-from ray.test_utils import wait_for_condition, new_scheduler_enabled
+from ray.test_utils import wait_for_condition
 from ray.internal.internal_api import global_gc
 
 logger = logging.getLogger(__name__)
@@ -166,9 +166,9 @@ def test_global_gc_when_full(shutdown_only):
         gc.enable()
 
 
-@pytest.mark.skipif(new_scheduler_enabled(), reason="hangs")
 def test_global_gc_actors(shutdown_only):
-    ray.init(num_cpus=1)
+    ray.init(
+        num_cpus=1, _system_config={"debug_dump_period_milliseconds": 500})
 
     try:
         gc.disable()
@@ -179,8 +179,7 @@ def test_global_gc_actors(shutdown_only):
                 return "Ok"
 
         # Try creating 3 actors. Unless python GC is triggered to break
-        # reference cycles, this won't be possible. Note this test takes 20s
-        # to run due to the 10s delay before checking of infeasible tasks.
+        # reference cycles, this won't be possible.
         for i in range(3):
             a = A.remote()
             cycle = [a]

src/ray/raylet/node_manager.cc

@@ -655,37 +655,47 @@ void NodeManager::HandleReleaseUnusedBundles(
 // debug_dump_period_ milliseconds.
 // See https://github.com/ray-project/ray/issues/5790 for details.
 void NodeManager::WarnResourceDeadlock() {
-  // Check if any progress is being made on this raylet.
-  for (const auto &task : local_queues_.GetTasks(TaskState::RUNNING)) {
-    // Ignore blocked tasks.
-    if (local_queues_.GetBlockedTaskIds().count(task.GetTaskSpecification().TaskId())) {
-      continue;
-    }
-    // Progress is being made, don't warn.
-    resource_deadlock_warned_ = 0;
-    return;
-  }
-
-  // The node is full of actors and no progress has been made for some time.
-  // If there are any pending tasks, build a warning.
-  std::ostringstream error_message;
   ray::Task exemplar;
   bool any_pending = false;
   int pending_actor_creations = 0;
   int pending_tasks = 0;
+  std::string available_resources;
 
-  // See if any tasks are blocked trying to acquire resources.
-  for (const auto &task : local_queues_.GetTasks(TaskState::READY)) {
-    const TaskSpecification &spec = task.GetTaskSpecification();
-    if (spec.IsActorCreationTask()) {
-      pending_actor_creations += 1;
-    } else {
-      pending_tasks += 1;
+  // Check if any progress is being made on this raylet.
+  for (const auto &worker : worker_pool_.GetAllRegisteredWorkers()) {
+    if (!worker->IsDead() && !worker->GetAssignedTaskId().IsNil() &&
+        !worker->IsBlocked() && worker->GetActorId().IsNil()) {
+      // Progress is being made in a task, don't warn.
+      resource_deadlock_warned_ = 0;
+      return;
     }
-    if (!any_pending) {
-      exemplar = task;
-      any_pending = true;
+  }
+
+  if (new_scheduler_enabled_) {
+    // Check if any tasks are blocked on resource acquisition.
+    if (!cluster_task_manager_->AnyPendingTasks(
+            &exemplar, &any_pending, &pending_actor_creations, &pending_tasks)) {
+      // No pending tasks, no need to warn.
+      resource_deadlock_warned_ = 0;
+      return;
     }
+    available_resources = new_resource_scheduler_->GetLocalResourceViewString();
+  } else {
+    // See if any tasks are blocked trying to acquire resources.
+    for (const auto &task : local_queues_.GetTasks(TaskState::READY)) {
+      const TaskSpecification &spec = task.GetTaskSpecification();
+      if (spec.IsActorCreationTask()) {
+        pending_actor_creations += 1;
+      } else {
+        pending_tasks += 1;
+      }
+      if (!any_pending) {
+        exemplar = task;
+        any_pending = true;
+      }
+    }
+    SchedulingResources &local_resources = cluster_resource_map_[self_node_id_];
+    available_resources = local_resources.GetAvailableResources().ToString();
   }
 
   // Push an warning to the driver that a task is blocked trying to acquire resources.
@@ -693,6 +703,7 @@ void NodeManager::WarnResourceDeadlock() {
   // case resource_deadlock_warned_:  0 => first time, don't do anything yet
   // case resource_deadlock_warned_:  1 => second time, print a warning
   // case resource_deadlock_warned_: >1 => global gc but don't print any warnings
+  std::ostringstream error_message;
   if (any_pending && resource_deadlock_warned_++ > 0) {
     // Actor references may be caught in cycles, preventing them from being deleted.
     // Trigger global GC to hopefully free up resource slots.
@@ -703,15 +714,13 @@ void NodeManager::WarnResourceDeadlock() {
       return;
     }
 
-    SchedulingResources &local_resources = cluster_resource_map_[self_node_id_];
     error_message
         << "The actor or task with ID " << exemplar.GetTaskSpecification().TaskId()
         << " cannot be scheduled right now. It requires "
         << exemplar.GetTaskSpecification().GetRequiredPlacementResources().ToString()
-        << " for placement, but this node only has remaining "
-        << local_resources.GetAvailableResources().ToString() << ". In total there are "
-        << pending_tasks << " pending tasks and " << pending_actor_creations
-        << " pending actors on this node. "
+        << " for placement, but this node only has remaining " << available_resources
+        << ". In total there are " << pending_tasks << " pending tasks and "
+        << pending_actor_creations << " pending actors on this node. "
         << "This is likely due to all cluster resources being claimed by actors. "
         << "To resolve the issue, consider creating fewer actors or increase the "
         << "resources available to this Ray cluster. You can ignore this message "
@@ -2164,6 +2173,7 @@ void NodeManager::HandleDirectCallTaskBlocked(
   auto const cpu_resource_ids = worker->ReleaseTaskCpuResources();
   local_available_resources_.Release(cpu_resource_ids);
   cluster_resource_map_[self_node_id_].Release(cpu_resource_ids.ToResourceSet());
+
   worker->MarkBlocked();
   DispatchTasks(local_queues_.GetReadyTasksByClass());
 }

src/ray/raylet/scheduling/cluster_resource_data.cc

@@ -232,6 +232,61 @@ std::string NodeResources::DebugString(StringIdMap string_to_in_map) const {
   return buffer.str();
 }
 
+const std::string format_resource(std::string resource_name, double quantity) {
+  if (resource_name == "object_store_memory" || resource_name == "memory") {
+    // Convert to 50MiB chunks and then to GiB
+    return std::to_string(quantity * (50 * 1024 * 1024) / (1024 * 1024 * 1024)) + " GiB";
+  }
+  return std::to_string(quantity);
+}
+
+std::string NodeResources::DictString(StringIdMap string_to_in_map) const {
+  std::stringstream buffer;
+  bool first = true;
+  buffer << "{";
+  for (size_t i = 0; i < this->predefined_resources.size(); i++) {
+    if (this->predefined_resources[i].total <= 0) {
+      continue;
+    }
+    if (first) {
+      first = false;
+    } else {
+      buffer << ", ";
+    }
+    std::string name = "";
+    switch (i) {
+    case CPU:
+      name = "CPU";
+      break;
+    case MEM:
+      name = "memory";
+      break;
+    case GPU:
+      name = "GPU";
+      break;
+    case TPU:
+      name = "TPU";
+      break;
+    default:
+      RAY_CHECK(false) << "This should never happen.";
+      break;
+    }
+    buffer << format_resource(name, this->predefined_resources[i].available.Double())
+           << "/";
+    buffer << format_resource(name, this->predefined_resources[i].total.Double());
+    buffer << " " << name;
+  }
+  for (auto it = this->custom_resources.begin(); it != this->custom_resources.end();
+       ++it) {
+    auto name = string_to_in_map.Get(it->first);
+    buffer << ", " << format_resource(name, it->second.available.Double()) << "/"
+           << format_resource(name, it->second.total.Double());
+    buffer << " " << name;
+  }
+  buffer << "}" << std::endl;
+  return buffer.str();
+}
+
 bool NodeResourceInstances::operator==(const NodeResourceInstances &other) {
   for (size_t i = 0; i < PredefinedResources_MAX; i++) {
     if (!EqualVectors(this->predefined_resources[i].total,

src/ray/raylet/scheduling/cluster_resource_data.h

@@ -164,6 +164,8 @@ class NodeResources {
   bool operator!=(const NodeResources &other);
   /// Returns human-readable string for these resources.
   std::string DebugString(StringIdMap string_to_int_map) const;
+  /// Returns compact dict-like string.
+  std::string DictString(StringIdMap string_to_int_map) const;
 };
 
 /// Total and available capacities of each resource instance.

src/ray/raylet/scheduling/cluster_resource_scheduler.cc

@@ -518,6 +518,12 @@ void ClusterResourceScheduler::InitResourceInstances(
   }
 }
 
+std::string ClusterResourceScheduler::GetLocalResourceViewString() const {
+  const auto &node_it = nodes_.find(local_node_id_);
+  RAY_CHECK(node_it != nodes_.end());
+  return node_it->second.GetLocalView().DictString(string_to_int_map_);
+}
+
 void ClusterResourceScheduler::InitLocalResources(const NodeResources &node_resources) {
   local_resources_.predefined_resources.resize(PredefinedResources_MAX);
 

src/ray/raylet/scheduling/cluster_resource_scheduler.h

@@ -191,6 +191,9 @@ class ClusterResourceScheduler {
   /// Return local resources.
   NodeResourceInstances GetLocalResources() { return local_resources_; };
 
+  /// Return local resources in human-readable string form.
+  std::string GetLocalResourceViewString() const;
+
   /// Create instances for each resource associated with the local node, given
   /// the node's resources.
   ///

src/ray/raylet/scheduling/cluster_task_manager.cc

@@ -242,7 +242,10 @@ void ClusterTaskManager::TasksUnblocked(const std::vector<TaskID> ready_ids) {
       const auto &scheduling_key = task.GetTaskSpecification().GetSchedulingClass();
       RAY_LOG(DEBUG) << "Args ready, task can be dispatched "
                      << task.GetTaskSpecification().TaskId();
-      tasks_to_dispatch_[scheduling_key].push_back(work);
+      // Note: we transition tasks back to the scheduling queue instead of directly
+      // to dispatch. This allows AnyPendingTasks() to simply check the scheduling
+      // queue to see if any tasks are blocked on resource availability: see #12438
+      tasks_to_schedule_[scheduling_key].push_back(work);
       waiting_tasks_.erase(it);
     }
   }
@@ -500,6 +503,32 @@ void ClusterTaskManager::FillResourceUsage(
   }
 }
 
+bool ClusterTaskManager::AnyPendingTasks(Task *exemplar, bool *any_pending,
+                                         int *num_pending_actor_creation,
+                                         int *num_pending_tasks) const {
+  // We are guaranteed that these tasks are blocked waiting for resources after a
+  // call to ScheduleAndDispatch(). Note that tasks that transition to waiting
+  // move back to the tasks_to_schedule_ queue after their deps are satisfied.
+  for (const auto &shapes_it : tasks_to_schedule_) {
+    auto &work_queue = shapes_it.second;
+    for (const auto &work_it : work_queue) {
+      const auto &task = std::get<0>(work_it);
+      if (task.GetTaskSpecification().IsActorCreationTask()) {
+        *num_pending_actor_creation += 1;
+      } else {
+        *num_pending_tasks += 1;
+      }
+
+      if (!*any_pending) {
+        *exemplar = task;
+        *any_pending = true;
+      }
+    }
+  }
+  // If there's any pending task, at this point, there's no progress being made.
+  return *any_pending;
+}
+
 std::string ClusterTaskManager::DebugString() const {
   std::stringstream buffer;
   buffer << "========== Node: " << self_node_id_ << " =================\n";

src/ray/raylet/scheduling/cluster_task_manager.h

@@ -115,6 +115,16 @@ class ClusterTaskManager {
   void FillResourceUsage(bool light_report_resource_usage_enabled,
                          std::shared_ptr<rpc::ResourcesData> data) const;
 
+  /// Return if any tasks are pending resource acquisition.
+  ///
+  /// \param[in] exemplar An example task that is deadlocking.
+  /// \param[in] num_pending_actor_creation Number of pending actor creation tasks.
+  /// \param[in] num_pending_tasks Number of pending tasks.
+  /// \param[in] any_pending True if there's any pending exemplar.
+  /// \return True if any progress is any tasks are pending.
+  bool AnyPendingTasks(Task *exemplar, bool *any_pending, int *num_pending_actor_creation,
+                       int *num_pending_tasks) const;
+
   std::string DebugString() const;
 
  private:
@@ -147,11 +157,11 @@ class ClusterTaskManager {
   std::unordered_map<SchedulingClass, std::deque<Work>> tasks_to_schedule_;
 
   /// Queue of lease requests that should be scheduled onto workers.
-  /// Tasks move from scheduled | waiting -> dispatch.
+  /// Tasks move from scheduled -> dispatch.
   std::unordered_map<SchedulingClass, std::deque<Work>> tasks_to_dispatch_;
 
   /// Tasks waiting for arguments to be transferred locally.
-  /// Tasks move from waiting -> dispatch.
+  /// Tasks move (back) from waiting -> scheduled.
   absl::flat_hash_map<TaskID, Work> waiting_tasks_;
 
   /// Queue of lease requests that are infeasible.

src/ray/raylet/scheduling/cluster_task_manager_test.cc

@@ -250,7 +250,9 @@ TEST_F(ClusterTaskManagerTest, ResourceTakenWhileResolving) {
   /* First task is unblocked now, but resources are no longer available */
   auto id = task.GetTaskSpecification().TaskId();
   std::vector<TaskID> unblocked = {id};
+  dependencies_fulfilled_ = true;
   task_manager_.TasksUnblocked(unblocked);
+  task_manager_.SchedulePendingTasks();
   task_manager_.DispatchScheduledTasksToWorkers(pool_, leased_workers_);
 
   ASSERT_EQ(num_callbacks, 1);
@@ -261,6 +263,7 @@ TEST_F(ClusterTaskManagerTest, ResourceTakenWhileResolving) {
   leased_workers_.clear();
   task_manager_.HandleTaskFinished(worker);
 
+  task_manager_.SchedulePendingTasks();
   task_manager_.DispatchScheduledTasksToWorkers(pool_, leased_workers_);
 
   // Task2 is now done so task can run.
@@ -676,6 +679,46 @@ TEST_F(ClusterTaskManagerTest, TestMultipleInfeasibleTasksWarnOnce) {
   ASSERT_EQ(announce_infeasible_task_calls_, 1);
 }
 
+TEST_F(ClusterTaskManagerTest, TestAnyPendingTasks) {
+  /*
+    Check if the manager can correctly identify pending tasks.
+   */
+
+  // task1: running
+  Task task = CreateTask({{ray::kCPU_ResourceLabel, 6}});
+  rpc::RequestWorkerLeaseReply reply;
+  std::shared_ptr<bool> callback_occurred = std::make_shared<bool>(false);
+  auto callback = [callback_occurred]() { *callback_occurred = true; };
+  task_manager_.QueueTask(task, &reply, callback);
+  task_manager_.SchedulePendingTasks();
+  std::shared_ptr<MockWorker> worker =
+      std::make_shared<MockWorker>(WorkerID::FromRandom(), 1234);
+  pool_.PushWorker(std::dynamic_pointer_cast<WorkerInterface>(worker));
+  task_manager_.DispatchScheduledTasksToWorkers(pool_, leased_workers_);
+  ASSERT_TRUE(*callback_occurred);
+  ASSERT_EQ(leased_workers_.size(), 1);
+  ASSERT_EQ(pool_.workers.size(), 0);
+
+  // task1: running. Progress is made, and there's no deadlock.
+  ray::Task exemplar;
+  bool any_pending = false;
+  int pending_actor_creations = 0;
+  int pending_tasks = 0;
+  ASSERT_FALSE(task_manager_.AnyPendingTasks(&exemplar, &any_pending,
+                                             &pending_actor_creations, &pending_tasks));
+
+  // task1: running, task2: queued.
+  Task task2 = CreateTask({{ray::kCPU_ResourceLabel, 6}});
+  rpc::RequestWorkerLeaseReply reply2;
+  std::shared_ptr<bool> callback_occurred2 = std::make_shared<bool>(false);
+  auto callback2 = [callback_occurred2]() { *callback_occurred2 = true; };
+  task_manager_.QueueTask(task2, &reply2, callback2);
+  task_manager_.SchedulePendingTasks();
+  ASSERT_FALSE(*callback_occurred2);
+  ASSERT_TRUE(task_manager_.AnyPendingTasks(&exemplar, &any_pending,
+                                            &pending_actor_creations, &pending_tasks));
+}
+
 int main(int argc, char **argv) {
   ::testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();