diff --git a/src/ray/raylet/design_docs/task_states.rst b/src/ray/raylet/design_docs/task_states.rst new file mode 100644 index 000000000..c17b8f36a --- /dev/null +++ b/src/ray/raylet/design_docs/task_states.rst @@ -0,0 +1,41 @@ +Task State: Definitions & Transition Diagram +============================================ + +A task can be in one of the following states: + +- **Placeable**: the task is ready to be placed at the node where is going to be + executed. This can be either local or a remote node. The decision is based on + resource availability (the location and size of the task's arguments are + ignore). If the local node has enough resources to satisfy task's demand, then + the task is placed locally, otherwise is forwarded to another node. + +- **WaitForActorCreation**: an actor method (task) is waiting for its actor to get + instantiated. Once the actor is created, the task transitions into the + waiting state, if the actor is local, or it is forwarded to the remote machine + running the actor. + +- **Waiting**: the task is waiting for its argument dependencies to be satisfied, + i.e., for its arguments to be transferred to the local object store. + +- **Ready**: the task is ready to run, that is, all task's arguments are in the + local object store. + +- **Running**: the task has been dispatched and it is running on a local + worker/actor. + +- **Blocked**: the task is being blocked as some data objects it depends on are not + available, e.g., because the task has launched another task and it waits + for the results, ore because of failures. + +:: + + forward + ------ + | | resource arguments actor/worker + | v available local available + Placeable ----------> Waiting --------> Ready ---------> Running + | ^ ^ | ^ + actor | | actor | actor worker | | worker + created | | created | created blocked | | unblocked + v | (remote) | (local) v | + WaitForActorCreation--------- Blocked diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index b4d225863..bdb6c389e 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -341,7 +341,8 @@ void NodeManager::HandleActorCreation(const ActorID &actor_id, } else { // The actor's location is now known. Dequeue any methods that were // submitted before the actor's location was known. - const auto &methods = local_queues_.GetUncreatedActorMethods(); + // (See design_docs/task_states.rst for the state transition diagram.) + const auto &methods = local_queues_.GetMethodsWaitingForActorCreation(); std::unordered_set created_actor_method_ids; for (const auto &method : methods) { if (method.GetTaskSpecification().ActorId() == actor_id) { @@ -377,7 +378,8 @@ void NodeManager::CleanUpTasksForDeadActor(const ActorID &actor_id) { // SchedulingQueue API. std::unordered_set tasks_to_remove; - GetActorTasksFromList(actor_id, local_queues_.GetUncreatedActorMethods(), + // (See design_docs/task_states.rst for the state transition diagram.) + GetActorTasksFromList(actor_id, local_queues_.GetMethodsWaitingForActorCreation(), tasks_to_remove); GetActorTasksFromList(actor_id, local_queues_.GetWaitingTasks(), tasks_to_remove); GetActorTasksFromList(actor_id, local_queues_.GetPlaceableTasks(), tasks_to_remove); @@ -400,6 +402,7 @@ void NodeManager::ProcessNewClient(LocalClientConnection &client) { void NodeManager::DispatchTasks() { // Work with a copy of scheduled tasks. + // (See design_docs/task_states.rst for the state transition diagram.) auto scheduled_tasks = local_queues_.GetReadyTasks(); // Return if there are no tasks to schedule. if (scheduled_tasks.empty()) { @@ -415,6 +418,7 @@ void NodeManager::DispatchTasks() { } // We have enough resources for this task. Assign task. // TODO(atumanov): perform the task state/queue transition inside AssignTask. + // (See design_docs/task_states.rst for the state transition diagram.) auto dispatched_task = local_queues_.RemoveTask(task.GetTaskSpecification().TaskId()); AssignTask(dispatched_task); } @@ -461,6 +465,7 @@ void NodeManager::ProcessClientMessage( // The client is a worker. Handle the case where the worker is killed // while executing a task. Clean up the assigned task's resources, push // an error to the driver. + // (See design_docs/task_states.rst for the state transition diagram.) const TaskID &task_id = worker->GetAssignedTaskId(); if (!task_id.is_nil()) { auto const &running_tasks = local_queues_.GetRunningTasks(); @@ -713,6 +718,7 @@ void NodeManager::ScheduleTasks() { local_task_ids.insert(task_id); } else { // TODO(atumanov): need a better interface for task exit on forward. + // (See design_docs/task_states.rst for the state transition diagram.) const auto task = local_queues_.RemoveTask(task_id); // Attempt to forward the task. If this fails to forward the task, // the task will be resubmit locally. @@ -816,7 +822,8 @@ void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineag RAY_CHECK_OK(gcs_client_->actor_table().Lookup(JobID::nil(), spec.ActorId(), lookup_callback)); // Keep the task queued until we discover the actor's location. - local_queues_.QueueUncreatedActorMethods({task}); + // (See design_docs/task_states.rst for the state transition diagram.) + local_queues_.QueueMethodsWaitingForActorCreation({task}); } } else { // This is a non-actor task. Queue the task for a placement decision or for dispatch @@ -825,6 +832,7 @@ void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineag // Check for local dependencies and enqueue as waiting or ready for dispatch. EnqueuePlaceableTask(task); } else { + // (See design_docs/task_states.rst for the state transition diagram.) local_queues_.QueuePlaceableTasks({task}); ScheduleTasks(); } @@ -840,6 +848,7 @@ void NodeManager::HandleWorkerBlocked(std::shared_ptr worker) { // it acquired for its assigned task while it is blocked. The resources will // be acquired again once the worker is unblocked. RAY_CHECK(!worker->GetAssignedTaskId().is_nil()); + // (See design_docs/task_states.rst for the state transition diagram.) const auto task = local_queues_.RemoveTask(worker->GetAssignedTaskId()); // Get the CPU resources required by the running task. const auto required_resources = task.GetTaskSpecification().GetRequiredResources(); @@ -868,6 +877,7 @@ void NodeManager::HandleWorkerUnblocked(std::shared_ptr worker) { return; } + // (See design_docs/task_states.rst for the state transition diagram.) const auto task = local_queues_.RemoveTask(worker->GetAssignedTaskId()); // Get the CPU resources required by the running task. const auto required_resources = task.GetTaskSpecification().GetRequiredResources(); @@ -899,6 +909,7 @@ void NodeManager::HandleWorkerUnblocked(std::shared_ptr worker) { } // Mark the task as running again. + // (See design_docs/task_states.rst for the state transition diagram.) local_queues_.QueueRunningTasks({task}); worker->MarkUnblocked(); } @@ -924,6 +935,7 @@ void NodeManager::EnqueuePlaceableTask(const Task &task) { task.GetTaskSpecification().TaskId(), task.GetDependencies()); // Enqueue the task. If all dependencies are available, then the task is queued // in the READY state, else the WAITING state. + // (See design_docs/task_states.rst for the state transition diagram.) if (args_ready) { local_queues_.QueueReadyTasks({task}); // Try to dispatch the newly ready task. @@ -955,6 +967,7 @@ void NodeManager::AssignTask(Task &task) { } // Queue this task for future assignment. The task will be assigned to a // worker once one becomes available. + // (See design_docs/task_states.rst for the state transition diagram.) local_queues_.QueueReadyTasks(std::vector({task})); return; } @@ -1014,6 +1027,7 @@ void NodeManager::AssignTask(Task &task) { // We started running the task, so the task is ready to write to GCS. lineage_cache_.AddReadyTask(task); // Mark the task as running. + // (See design_docs/task_states.rst for the state transition diagram.) local_queues_.QueueRunningTasks(std::vector({task})); // Notify the task dependency manager that we no longer need this task's // object dependencies. @@ -1026,6 +1040,7 @@ void NodeManager::AssignTask(Task &task) { nullptr); // Queue this task for future assignment. The task will be assigned to a // worker once one becomes available. + // (See design_docs/task_states.rst for the state transition diagram.) local_queues_.QueueReadyTasks(std::vector({task})); } } @@ -1033,6 +1048,8 @@ void NodeManager::AssignTask(Task &task) { void NodeManager::FinishAssignedTask(Worker &worker) { TaskID task_id = worker.GetAssignedTaskId(); RAY_LOG(DEBUG) << "Finished task " << task_id; + + // (See design_docs/task_states.rst for the state transition diagram.) const auto task = local_queues_.RemoveTask(task_id); if (task.GetTaskSpecification().IsActorCreationTask()) { @@ -1096,6 +1113,7 @@ void NodeManager::HandleObjectLocal(const ObjectID &object_id) { std::unordered_set ready_task_id_set(ready_task_ids.begin(), ready_task_ids.end()); // Transition tasks from waiting to scheduled. + // (See design_docs/task_states.rst for the state transition diagram.) local_queues_.MoveTasks(ready_task_id_set, TaskState::WAITING, TaskState::READY); // New scheduled tasks appeared in the queue, try to dispatch them. DispatchTasks(); diff --git a/src/ray/raylet/scheduling_queue.cc b/src/ray/raylet/scheduling_queue.cc index bf9555c5d..b6cd51b92 100644 --- a/src/ray/raylet/scheduling_queue.cc +++ b/src/ray/raylet/scheduling_queue.cc @@ -45,8 +45,8 @@ namespace ray { namespace raylet { -const std::list &SchedulingQueue::GetUncreatedActorMethods() const { - return this->uncreated_actor_methods_; +const std::list &SchedulingQueue::GetMethodsWaitingForActorCreation() const { + return this->methods_waiting_for_actor_creation_; } const std::list &SchedulingQueue::GetWaitingTasks() const { @@ -108,7 +108,7 @@ std::vector SchedulingQueue::RemoveTasks(std::unordered_set &task_ std::vector removed_tasks; // Try to find the tasks to remove from the queues. - RemoveTasksFromQueue(uncreated_actor_methods_, task_ids, removed_tasks); + RemoveTasksFromQueue(methods_waiting_for_actor_creation_, task_ids, removed_tasks); RemoveTasksFromQueue(waiting_tasks_, task_ids, removed_tasks); RemoveTasksFromQueue(placeable_tasks_, task_ids, removed_tasks); RemoveTasksFromQueue(ready_tasks_, task_ids, removed_tasks); @@ -174,8 +174,9 @@ void SchedulingQueue::MoveTasks(std::unordered_set &task_ids, TaskState } } -void SchedulingQueue::QueueUncreatedActorMethods(const std::vector &tasks) { - QueueTasks(uncreated_actor_methods_, tasks); +void SchedulingQueue::QueueMethodsWaitingForActorCreation( + const std::vector &tasks) { + QueueTasks(methods_waiting_for_actor_creation_, tasks); } void SchedulingQueue::QueueWaitingTasks(const std::vector &tasks) { diff --git a/src/ray/raylet/scheduling_queue.h b/src/ray/raylet/scheduling_queue.h index e44459ac8..768bff29e 100644 --- a/src/ray/raylet/scheduling_queue.h +++ b/src/ray/raylet/scheduling_queue.h @@ -16,12 +16,8 @@ enum class TaskState { INIT, PLACEABLE, WAITING, READY, RUNNING, BLOCKED, DRIVER /// \class SchedulingQueue /// -/// Encapsulates task queues. Each queue represents a scheduling state for a -/// task. The scheduling state is one of -/// (1) placeable: the task is ready for a placement decision, -/// (2) waiting: waiting for object dependencies to become locally available, -/// (3) ready: the task is ready for local dispatch, with all arguments locally ready, -/// (4) running: the task has been dispatched and is running on a worker. +/// Encapsulates task queues. +// (See design_docs/task_states.rst for the state transition diagram.) class SchedulingQueue { public: /// Create a scheduling queue. @@ -35,7 +31,7 @@ class SchedulingQueue { /// /// \return A const reference to the queue of tasks that are destined for /// actors that have not yet been created. - const std::list &GetUncreatedActorMethods() const; + const std::list &GetMethodsWaitingForActorCreation() const; /// Get the queue of tasks in the waiting state. /// @@ -97,7 +93,7 @@ class SchedulingQueue { /// Queue tasks that are destined for actors that have not yet been created. /// /// \param tasks The tasks to queue. - void QueueUncreatedActorMethods(const std::vector &tasks); + void QueueMethodsWaitingForActorCreation(const std::vector &tasks); /// Queue tasks in the waiting state. These are tasks that cannot yet be /// dispatched since they are blocked on a missing data dependency. @@ -153,7 +149,7 @@ class SchedulingQueue { private: /// Tasks that are destined for actors that have not yet been created. - std::list uncreated_actor_methods_; + std::list methods_waiting_for_actor_creation_; /// Tasks that are waiting for an object dependency to appear locally. std::list waiting_tasks_; /// Tasks whose object dependencies are locally available, but that are