wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-12 20:45:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

State transition diagram documentation. (#2502)

Browse Source 
* Added description of transition diagram and a few name changes for imporved clarity.

* rename some methods and update task_states.rst
This commit is contained in:
Ion
2018-07-28 22:28:45 -07:00
committed by Philipp Moritz
parent 0ea7a6abf0
commit 80db69d245
4 changed files with 73 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/raylet/design_docs/task_states.rst
+41
View File
@@ -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
src/ray/raylet/node_manager.cc
+21 -3
View File
@@ -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<TaskID> 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<TaskID> 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> 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> 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> 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>({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>({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>({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<TaskID> 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();
src/ray/raylet/scheduling_queue.cc
+6 -5
View File
@@ -45,8 +45,8 @@ namespace ray {
namespace raylet {
const std::list<Task> &SchedulingQueue::GetUncreatedActorMethods() const {
return this->uncreated_actor_methods_;
const std::list<Task> &SchedulingQueue::GetMethodsWaitingForActorCreation() const {
return this->methods_waiting_for_actor_creation_;
}
const std::list<Task> &SchedulingQueue::GetWaitingTasks() const {
@@ -108,7 +108,7 @@ std::vector<Task> SchedulingQueue::RemoveTasks(std::unordered_set<TaskID> &task_
std::vector<Task> 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<TaskID> &task_ids, TaskState
}
}
void SchedulingQueue::QueueUncreatedActorMethods(const std::vector<Task> &tasks) {
QueueTasks(uncreated_actor_methods_, tasks);
void SchedulingQueue::QueueMethodsWaitingForActorCreation(
const std::vector<Task> &tasks) {
QueueTasks(methods_waiting_for_actor_creation_, tasks);
}
void SchedulingQueue::QueueWaitingTasks(const std::vector<Task> &tasks) {
src/ray/raylet/scheduling_queue.h
+5 -9
View File
@@ -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<Task> &GetUncreatedActorMethods() const;
const std::list<Task> &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<Task> &tasks);
void QueueMethodsWaitingForActorCreation(const std::vector<Task> &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<Task> uncreated_actor_methods_;
std::list<Task> methods_waiting_for_actor_creation_;
/// Tasks that are waiting for an object dependency to appear locally.
std::list<Task> waiting_tasks_;
/// Tasks whose object dependencies are locally available, but that are