Remove unused scheduler states (#10318)

* remove unused state

* remove unused states

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<TaskID> 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.

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<int>(ray::raylet::TaskState::kNumTaskQueues),
               "Must specify a TaskState name for every task queue");
@@ -242,9 +240,6 @@ void SchedulingQueue::FilterState(std::unordered_set<TaskID> &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<TaskID> &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<Task> SchedulingQueue::RemoveTasks(std::unordered_set<TaskID> &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<TaskID> &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<std::underlying_type<TaskState>::type>(src_state);
@@ -423,9 +408,6 @@ void SchedulingQueue::MoveTasks(std::unordered_set<TaskID> &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<std::underlying_type<TaskState>::type>(dst_state);
@@ -461,23 +443,6 @@ std::unordered_set<TaskID> SchedulingQueue::GetTaskIdsForJob(const JobID &job_id
   return task_ids;
 }
 
-std::unordered_set<TaskID> SchedulingQueue::GetTaskIdsForActor(
-    const ActorID &actor_id) const {
-  std::unordered_set<TaskID> task_ids;
-  int swap = static_cast<int>(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);

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<int>(task_state)] = ready_queue_;
@@ -321,12 +309,6 @@ class SchedulingQueue {
   /// \return All the tasks that have the given job ID.
   std::unordered_set<TaskID> 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<TaskID> GetTaskIdsForActor(const ActorID &actor_id) const;
-
   /// Returns the number of running tasks in this class.
   ///
   /// \return int.