From 1e99b814f0a0b929dd6a0310fd3ec6c4d4bdfde6 Mon Sep 17 00:00:00 2001 From: Edward Oakes Date: Tue, 25 Aug 2020 20:56:21 -0500 Subject: [PATCH] Remove unused scheduler states (#10318) * remove unused state * remove unused states --- src/ray/raylet/node_manager.cc | 45 ------------------------------ src/ray/raylet/scheduling_queue.cc | 39 ++------------------------ src/ray/raylet/scheduling_queue.h | 18 ------------ 3 files changed, 2 insertions(+), 100 deletions(-) diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 060c0ff1c..2c9159e6c 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -1043,50 +1043,6 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id, // stop listening for the actor creation task. This is needed because we use // `ListenAndMaybeReconstruct` to reconstruct the actor. reconstruction_policy_.Cancel(actor_registration.GetActorCreationDependency()); - // The actor's location is now known. Dequeue any methods that were - // submitted before the actor's location was known. - // (See design_docs/task_states.rst for the state transition diagram.) - const auto &methods = local_queues_.GetTasks(TaskState::WAITING_FOR_ACTOR_CREATION); - std::unordered_set created_actor_method_ids; - for (const auto &method : methods) { - if (method.GetTaskSpecification().ActorId() == actor_id) { - created_actor_method_ids.insert(method.GetTaskSpecification().TaskId()); - } - } - // Resubmit the methods that were submitted before the actor's location was - // known. - auto created_actor_methods = local_queues_.RemoveTasks(created_actor_method_ids); - for (const auto &method : created_actor_methods) { - // Maintain the invariant that if a task is in the - // MethodsWaitingForActorCreation queue, then it is subscribed to its - // respective actor creation task. Since the actor location is now known, - // we can remove the task from the queue and forget its dependency on the - // actor creation task. - RAY_CHECK(task_dependency_manager_.UnsubscribeGetDependencies( - method.GetTaskSpecification().TaskId())); - // The task's uncommitted lineage was already added to the local lineage - // cache upon the initial submission, so it's okay to resubmit it with an - // empty lineage this time. - SubmitTask(method); - } - } else if (actor_registration.GetState() == ActorTableData::DEAD) { - // When an actor dies, loop over all of the queued tasks for that actor - // and treat them as failed. - auto tasks_to_remove = local_queues_.GetTaskIdsForActor(actor_id); - auto removed_tasks = local_queues_.RemoveTasks(tasks_to_remove); - for (auto const &task : removed_tasks) { - TreatTaskAsFailed(task, ErrorType::ACTOR_DIED); - } - } else if (actor_registration.GetState() == ActorTableData::RESTARTING) { - RAY_LOG(DEBUG) << "Actor is being restarted: " << actor_id; - // When an actor fails but can be restarted, resubmit all of the queued - // tasks for that actor. This will mark the tasks as waiting for actor - // creation. - auto tasks_to_remove = local_queues_.GetTaskIdsForActor(actor_id); - auto removed_tasks = local_queues_.RemoveTasks(tasks_to_remove); - for (auto const &task : removed_tasks) { - SubmitTask(task); - } } } @@ -2810,7 +2766,6 @@ void NodeManager::HandleObjectLocal(const ObjectID &object_id) { local_queues_.FilterState(ready_task_id_set, TaskState::BLOCKED); local_queues_.FilterState(ready_task_id_set, TaskState::RUNNING); local_queues_.FilterState(ready_task_id_set, TaskState::DRIVER); - local_queues_.FilterState(ready_task_id_set, TaskState::WAITING_FOR_ACTOR_CREATION); // Make sure that the remaining tasks are all WAITING or direct call // actors. diff --git a/src/ray/raylet/scheduling_queue.cc b/src/ray/raylet/scheduling_queue.cc index 4b3137951..d5dfe87ed 100644 --- a/src/ray/raylet/scheduling_queue.cc +++ b/src/ray/raylet/scheduling_queue.cc @@ -21,10 +21,8 @@ namespace { -static constexpr const char *task_state_strings[] = { - "placeable", "waiting", "ready", - "running", "infeasible", "waiting for actor creation", - "swap"}; +static constexpr const char *task_state_strings[] = {"placeable", "waiting", "ready", + "running", "infeasible"}; static_assert(sizeof(task_state_strings) / sizeof(const char *) == static_cast(ray::raylet::TaskState::kNumTaskQueues), "Must specify a TaskState name for every task queue"); @@ -242,9 +240,6 @@ void SchedulingQueue::FilterState(std::unordered_set &task_ids, case TaskState::PLACEABLE: FilterStateFromQueue(task_ids, TaskState::PLACEABLE); break; - case TaskState::WAITING_FOR_ACTOR_CREATION: - FilterStateFromQueue(task_ids, TaskState::WAITING_FOR_ACTOR_CREATION); - break; case TaskState::WAITING: FilterStateFromQueue(task_ids, TaskState::WAITING); break; @@ -257,9 +252,6 @@ void SchedulingQueue::FilterState(std::unordered_set &task_ids, case TaskState::INFEASIBLE: FilterStateFromQueue(task_ids, TaskState::INFEASIBLE); break; - case TaskState::SWAP: - FilterStateFromQueue(task_ids, TaskState::SWAP); - break; case TaskState::BLOCKED: { const auto blocked_ids = GetBlockedTaskIds(); for (auto it = task_ids.begin(); it != task_ids.end();) { @@ -326,8 +318,6 @@ std::vector SchedulingQueue::RemoveTasks(std::unordered_set &task_ TaskState::READY, TaskState::RUNNING, TaskState::INFEASIBLE, - TaskState::WAITING_FOR_ACTOR_CREATION, - TaskState::SWAP, }) { RemoveTasksFromQueue(task_state, task_ids, &removed_tasks); } @@ -347,8 +337,6 @@ bool SchedulingQueue::RemoveTask(const TaskID &task_id, Task *removed_task, TaskState::READY, TaskState::RUNNING, TaskState::INFEASIBLE, - TaskState::WAITING_FOR_ACTOR_CREATION, - TaskState::SWAP, }) { RemoveTasksFromQueue(task_state, task_id_set, &removed_tasks); if (task_id_set.empty()) { @@ -395,9 +383,6 @@ void SchedulingQueue::MoveTasks(std::unordered_set &task_ids, TaskState case TaskState::INFEASIBLE: RemoveTasksFromQueue(TaskState::INFEASIBLE, task_ids, &removed_tasks); break; - case TaskState::SWAP: - RemoveTasksFromQueue(TaskState::SWAP, task_ids, &removed_tasks); - break; default: RAY_LOG(FATAL) << "Attempting to move tasks from unrecognized state " << static_cast::type>(src_state); @@ -423,9 +408,6 @@ void SchedulingQueue::MoveTasks(std::unordered_set &task_ids, TaskState case TaskState::INFEASIBLE: QueueTasks(removed_tasks, TaskState::INFEASIBLE); break; - case TaskState::SWAP: - QueueTasks(removed_tasks, TaskState::SWAP); - break; default: RAY_LOG(FATAL) << "Attempting to move tasks to unrecognized state " << static_cast::type>(dst_state); @@ -461,23 +443,6 @@ std::unordered_set SchedulingQueue::GetTaskIdsForJob(const JobID &job_id return task_ids; } -std::unordered_set SchedulingQueue::GetTaskIdsForActor( - const ActorID &actor_id) const { - std::unordered_set task_ids; - int swap = static_cast(TaskState::SWAP); - int i = 0; - for (const auto &task_queue : task_queues_) { - // This is a hack to make sure that we don't remove tasks from the SWAP - // queue, since these are always guaranteed to be removed and eventually - // resubmitted if necessary by the node manager. - if (i != swap) { - GetActorTasksFromQueue(*task_queue, actor_id, task_ids); - } - i++; - } - return task_ids; -} - void SchedulingQueue::AddBlockedTaskId(const TaskID &task_id) { RAY_LOG(DEBUG) << "Added blocked task " << task_id; auto inserted = blocked_task_ids_.insert(task_id); diff --git a/src/ray/raylet/scheduling_queue.h b/src/ray/raylet/scheduling_queue.h index 2e32ecdbe..35e49741e 100644 --- a/src/ray/raylet/scheduling_queue.h +++ b/src/ray/raylet/scheduling_queue.h @@ -44,16 +44,6 @@ enum class TaskState { // The task has resources that cannot be satisfied by any node, as far as we // know. INFEASIBLE, - // The task is an actor method and is waiting to learn where the actor was - // created. - WAITING_FOR_ACTOR_CREATION, - // Swap queue for tasks that are in between states. This can happen when a - // task is removed from one queue, and an async callback is responsible for - // re-queuing the task. For example, a READY task that has just been assigned - // to a worker will get moved to the SWAP queue while waiting for a response - // from the worker. If the worker accepts the task, the task will be added to - // the RUNNING queue, else it will be returned to READY. - SWAP, // The number of task queues. All states that precede this enum must have an // associated TaskQueue in SchedulingQueue. All states that succeed // this enum do not have an associated TaskQueue, since the tasks @@ -177,8 +167,6 @@ class SchedulingQueue { TaskState::READY, TaskState::RUNNING, TaskState::INFEASIBLE, - TaskState::WAITING_FOR_ACTOR_CREATION, - TaskState::SWAP, }) { if (task_state == TaskState::READY) { task_queues_[static_cast(task_state)] = ready_queue_; @@ -321,12 +309,6 @@ class SchedulingQueue { /// \return All the tasks that have the given job ID. std::unordered_set GetTaskIdsForJob(const JobID &job_id) const; - /// \brief Get all the task IDs for an actor. - /// - /// \param actor_id All the tasks that have the given actor_id are returned. - /// \return All the tasks that have the given actor ID. - std::unordered_set GetTaskIdsForActor(const ActorID &actor_id) const; - /// Returns the number of running tasks in this class. /// /// \return int.