diff --git a/python/ray/test_utils.py b/python/ray/test_utils.py index 648b4f4f9..7f6aaa360 100644 --- a/python/ray/test_utils.py +++ b/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 = [] diff --git a/python/ray/tests/test_failure.py b/python/ray/tests/test_failure.py index 84904a5e1..b3505bbf5 100644 --- a/python/ray/tests/test_failure.py +++ b/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. diff --git a/python/ray/tests/test_global_gc.py b/python/ray/tests/test_global_gc.py index 247516450..a3039f14d 100644 --- a/python/ray/tests/test_global_gc.py +++ b/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] diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 49bf6a6af..2e13bbb91 100644 --- a/src/ray/raylet/node_manager.cc +++ b/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()); } diff --git a/src/ray/raylet/scheduling/cluster_resource_data.cc b/src/ray/raylet/scheduling/cluster_resource_data.cc index 551b5a980..a9fa3af39 100644 --- a/src/ray/raylet/scheduling/cluster_resource_data.cc +++ b/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, diff --git a/src/ray/raylet/scheduling/cluster_resource_data.h b/src/ray/raylet/scheduling/cluster_resource_data.h index 96b4c4359..46a5cbe1b 100644 --- a/src/ray/raylet/scheduling/cluster_resource_data.h +++ b/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. diff --git a/src/ray/raylet/scheduling/cluster_resource_scheduler.cc b/src/ray/raylet/scheduling/cluster_resource_scheduler.cc index 66047d258..bcca8862a 100644 --- a/src/ray/raylet/scheduling/cluster_resource_scheduler.cc +++ b/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); diff --git a/src/ray/raylet/scheduling/cluster_resource_scheduler.h b/src/ray/raylet/scheduling/cluster_resource_scheduler.h index 470c97c38..9e480b4c8 100644 --- a/src/ray/raylet/scheduling/cluster_resource_scheduler.h +++ b/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. /// diff --git a/src/ray/raylet/scheduling/cluster_task_manager.cc b/src/ray/raylet/scheduling/cluster_task_manager.cc index 74437a4a1..c11d818ef 100644 --- a/src/ray/raylet/scheduling/cluster_task_manager.cc +++ b/src/ray/raylet/scheduling/cluster_task_manager.cc @@ -242,7 +242,10 @@ void ClusterTaskManager::TasksUnblocked(const std::vector 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"; diff --git a/src/ray/raylet/scheduling/cluster_task_manager.h b/src/ray/raylet/scheduling/cluster_task_manager.h index 995273ed5..b71593f8a 100644 --- a/src/ray/raylet/scheduling/cluster_task_manager.h +++ b/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 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> 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> 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 waiting_tasks_; /// Queue of lease requests that are infeasible. diff --git a/src/ray/raylet/scheduling/cluster_task_manager_test.cc b/src/ray/raylet/scheduling/cluster_task_manager_test.cc index 023390632..24018dbc8 100644 --- a/src/ray/raylet/scheduling/cluster_task_manager_test.cc +++ b/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 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 callback_occurred = std::make_shared(false); + auto callback = [callback_occurred]() { *callback_occurred = true; }; + task_manager_.QueueTask(task, &reply, callback); + task_manager_.SchedulePendingTasks(); + std::shared_ptr worker = + std::make_shared(WorkerID::FromRandom(), 1234); + pool_.PushWorker(std::dynamic_pointer_cast(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 callback_occurred2 = std::make_shared(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();