wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-09-10 12:38:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

[xray] raylet task queue transition discipline (#2302)

Browse Source 
* add queueing interface to move tasks between queues internally
* queueing discipline change: ready->waiting->scheduled->running
* rename task states : ready -> placeable; update documentation
* rename task states : scheduled -> ready; update documentation
* cleanup comments
* cleanup; transition placeable actor tasks
* minor comment cleanup
* addressing comments
* linting
This commit is contained in:
Alexey Tumanov
2018-06-27 14:23:41 -07:00
committed by GitHub
parent 44f5f0520b
commit 965e182384
6 changed files with 122 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/raylet/lineage_cache.cc
+1 -1
View File
@@ -326,7 +326,7 @@ bool LineageCache::FlushTask(const TaskID &task_id) {
}
void LineageCache::Flush() {
// Iterate through all tasks that are READY.
// Iterate through all tasks that are PLACEABLE.
for (auto it = uncommitted_ready_tasks_.begin();
it != uncommitted_ready_tasks_.end();) {
bool flushed = FlushTask(*it);
src/ray/raylet/node_manager.cc
+37 -29
View File
@@ -320,7 +320,7 @@ void NodeManager::ProcessNewClient(LocalClientConnection &client) {
void NodeManager::DispatchTasks() {
// Work with a copy of scheduled tasks.
auto scheduled_tasks = local_queues_.GetScheduledTasks();
auto scheduled_tasks = local_queues_.GetReadyTasks();
// Return if there are no tasks to schedule.
if (scheduled_tasks.empty()) {
return;
@@ -577,7 +577,7 @@ void NodeManager::ProcessNodeManagerMessage(TcpClientConnection &node_manager_cl
const Task &task = uncommitted_lineage.GetEntry(task_id)->TaskData();
RAY_LOG(DEBUG) << "got task " << task.GetTaskSpecification().TaskId()
<< " spillback=" << task.GetTaskExecutionSpecReadonly().NumForwards();
SubmitTask(task, uncommitted_lineage);
SubmitTask(task, uncommitted_lineage, /* forwarded = */ true);
} break;
case protocol::MessageType::DisconnectClient: {
// TODO(rkn): We need to do some cleanup here.
@@ -591,16 +591,17 @@ void NodeManager::ProcessNodeManagerMessage(TcpClientConnection &node_manager_cl
}
void NodeManager::ScheduleTasks() {
// This method performs the transition of tasks from PENDING to SCHEDULED.
auto policy_decision = scheduling_policy_.Schedule(
cluster_resource_map_, gcs_client_->client_table().GetLocalClientId(),
remote_clients_);
#ifndef NDEBUG
RAY_LOG(DEBUG) << "[NM ScheduleTasks] policy decision:";
for (const auto &pair : policy_decision) {
TaskID task_id = pair.first;
ClientID client_id = pair.second;
RAY_LOG(DEBUG) << task_id << " --> " << client_id;
}
#endif
// Extract decision for this local scheduler.
std::unordered_set<TaskID> local_task_ids;
@@ -611,28 +612,26 @@ void NodeManager::ScheduleTasks() {
if (client_id == gcs_client_->client_table().GetLocalClientId()) {
local_task_ids.insert(task_id);
} else {
// TODO(atumanov): need a better interface for task exit on forward.
auto tasks = local_queues_.RemoveTasks({task_id});
RAY_CHECK(1 == tasks.size());
Task &task = tasks.front();
// TODO(swang): Handle forward task failure.
RAY_CHECK_OK(ForwardTask(task, client_id));
}
// Notify the task dependency manager that we no longer need this task's
// object dependencies.
// NOTE(swang): For local tasks, the scheduled task's dependencies may get
// evicted before it can be assigned to a worker.
task_dependency_manager_.UnsubscribeDependencies(task_id);
}
// Transition locally scheduled tasks to SCHEDULED and dispatch scheduled tasks.
// Transition locally placed tasks to waiting or ready for dispatch.
if (local_task_ids.size() > 0) {
std::vector<Task> tasks = local_queues_.RemoveTasks(local_task_ids);
local_queues_.QueueScheduledTasks(tasks);
DispatchTasks();
for (const auto &t : tasks) {
EnqueuePlaceableTask(t);
}
}
}
void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineage) {
void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineage,
bool forwarded) {
// Add the task and its uncommitted lineage to the lineage cache.
lineage_cache_.AddWaitingTask(task, uncommitted_lineage);
// Mark the task as pending. Once the task has finished execution, or once it
@@ -648,8 +647,8 @@ void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineag
// We have a known location for the actor.
auto node_manager_id = actor_entry->second.GetNodeManagerId();
if (node_manager_id == gcs_client_->client_table().GetLocalClientId()) {
// The actor is local. Queue the task for local execution.
QueueTask(task);
// The actor is local. Queue the task for local execution, bypassing placement.
EnqueuePlaceableTask(task);
} else {
// The actor is remote. Forward the task to the node manager that owns
// the actor.
@@ -680,8 +679,15 @@ void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineag
local_queues_.QueueUncreatedActorMethods({task});
}
} else {
// This is a non-actor task. Queue the task for local execution.
QueueTask(task);
// This is a non-actor task. Queue the task for a placement decision or for dispatch
// if the task was forwarded.
if (forwarded) {
// Check for local dependencies and enqueue as waiting or ready for dispatch.
EnqueuePlaceableTask(task);
} else {
local_queues_.QueuePlaceableTasks({task});
ScheduleTasks();
}
}
}
@@ -698,16 +704,18 @@ void NodeManager::HandleRemoteDependencyCanceled(const ObjectID &dependency_id)
// TODO(swang): Cancel reconstruction of the object.
}
void NodeManager::QueueTask(const Task &task) {
void NodeManager::EnqueuePlaceableTask(const Task &task) {
// TODO(atumanov): add task lookup hashmap and change EnqueuePlaceableTask to take
// a vector of TaskIDs. Trigger MoveTask internally.
// Subscribe to the task's dependencies.
bool ready = task_dependency_manager_.SubscribeDependencies(
bool args_ready = task_dependency_manager_.SubscribeDependencies(
task.GetTaskSpecification().TaskId(), task.GetDependencies());
// Queue the task. If all dependencies are available, then the task is queued
// in the READY state, else the WAITING.
if (ready) {
// Enqueue the task. If all dependencies are available, then the task is queued
// in the READY state, else the WAITING state.
if (args_ready) {
local_queues_.QueueReadyTasks({task});
// Try to schedule the newly ready task.
ScheduleTasks();
// Try to dispatch the newly ready task.
DispatchTasks();
} else {
local_queues_.QueueWaitingTasks({task});
}
@@ -735,7 +743,7 @@ void NodeManager::AssignTask(Task &task) {
}
// Queue this task for future assignment. The task will be assigned to a
// worker once one becomes available.
local_queues_.QueueScheduledTasks(std::vector<Task>({task}));
local_queues_.QueueReadyTasks(std::vector<Task>({task}));
return;
}
@@ -803,7 +811,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.
local_queues_.QueueScheduledTasks(std::vector<Task>({task}));
local_queues_.QueueReadyTasks(std::vector<Task>({task}));
}
}
@@ -873,10 +881,10 @@ void NodeManager::HandleObjectLocal(const ObjectID &object_id) {
if (ready_task_ids.size() > 0) {
std::unordered_set<TaskID> ready_task_id_set(ready_task_ids.begin(),
ready_task_ids.end());
auto ready_tasks = local_queues_.RemoveTasks(ready_task_id_set);
local_queues_.QueueReadyTasks(std::vector<Task>(ready_tasks));
// Schedule the newly ready tasks.
ScheduleTasks();
// Transition tasks from waiting to scheduled.
local_queues_.MoveTasks(ready_task_id_set, WAITING, READY);
// New scheduled tasks appeared in the queue, try to dispatch them.
DispatchTasks();
}
}
src/ray/raylet/node_manager.h
+6 -6
View File
@@ -69,15 +69,16 @@ class NodeManager {
const HeartbeatTableDataT &data);
/// Methods for task scheduling.
// Queue a task for local execution.
void QueueTask(const Task &task);
/// Submit a task to this node.
void SubmitTask(const Task &task, const Lineage &uncommitted_lineage);
/// Enqueue a placeable task to wait on object dependencies or be ready for dispatch.
void EnqueuePlaceableTask(const Task &task);
/// Handle specified task's submission to the local node manager.
void SubmitTask(const Task &task, const Lineage &uncommitted_lineage,
bool forwarded = false);
/// Assign a task. The task is assumed to not be queued in local_queues_.
void AssignTask(Task &task);
/// Handle a worker finishing its assigned task.
void FinishAssignedTask(Worker &worker);
/// Schedule tasks.
/// Perform a placement decision on placeable tasks.
void ScheduleTasks();
/// Resubmit a task whose return value needs to be reconstructed.
void ResubmitTask(const TaskID &task_id);
@@ -118,7 +119,6 @@ class NodeManager {
const SchedulingResources local_resources_;
/// The resources (and specific resource IDs) that are currently available.
ResourceIdSet local_available_resources_;
// TODO(atumanov): Add resource information from other nodes.
std::unordered_map<ClientID, SchedulingResources> cluster_resource_map_;
/// A pool of workers.
WorkerPool worker_pool_;
src/ray/raylet/scheduling_policy.cc
+1 -1
View File
@@ -25,7 +25,7 @@ std::unordered_map<TaskID, ClientID> SchedulingPolicy::Schedule(
}
// Iterate over running tasks, get their resource demand and try to schedule.
for (const auto &t : scheduling_queue_.GetReadyTasks()) {
for (const auto &t : scheduling_queue_.GetPlaceableTasks()) {
// Get task's resource demand
const auto &resource_demand = t.GetTaskSpecification().GetRequiredResources();
const TaskID &task_id = t.GetTaskSpecification().TaskId();
src/ray/raylet/scheduling_queue.cc
+49 -13
View File
@@ -14,12 +14,12 @@ const std::list<Task> &SchedulingQueue::GetWaitingTasks() const {
return this->waiting_tasks_;
}
const std::list<Task> &SchedulingQueue::GetReadyTasks() const {
return this->ready_tasks_;
const std::list<Task> &SchedulingQueue::GetPlaceableTasks() const {
return this->placeable_tasks_;
}
const std::list<Task> &SchedulingQueue::GetScheduledTasks() const {
return this->scheduled_tasks_;
const std::list<Task> &SchedulingQueue::GetReadyTasks() const {
return this->ready_tasks_;
}
const std::list<Task> &SchedulingQueue::GetRunningTasks() const {
@@ -30,10 +30,6 @@ const std::list<Task> &SchedulingQueue::GetBlockedTasks() const {
return this->blocked_tasks_;
}
const std::list<Task> &SchedulingQueue::GetReadyMethods() const {
throw std::runtime_error("Method not implemented");
}
// Helper function to remove tasks in the given set of task_ids from a
// queue, and append them to the given vector removed_tasks.
void removeTasksFromQueue(std::list<Task> &queue, std::unordered_set<TaskID> &task_ids,
@@ -62,8 +58,8 @@ std::vector<Task> SchedulingQueue::RemoveTasks(std::unordered_set<TaskID> task_i
// Try to find the tasks to remove from the waiting tasks.
removeTasksFromQueue(uncreated_actor_methods_, 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);
removeTasksFromQueue(scheduled_tasks_, task_ids, removed_tasks);
removeTasksFromQueue(running_tasks_, task_ids, removed_tasks);
removeTasksFromQueue(blocked_tasks_, task_ids, removed_tasks);
// TODO(swang): Remove from running methods.
@@ -72,6 +68,46 @@ std::vector<Task> SchedulingQueue::RemoveTasks(std::unordered_set<TaskID> task_i
return removed_tasks;
}
void SchedulingQueue::MoveTasks(std::unordered_set<TaskID> task_ids, TaskState src_state,
TaskState dst_state) {
// TODO(atumanov): check the states first to ensure the move is transactional.
std::vector<Task> removed_tasks;
// Remove the tasks from the specified source queue.
switch (src_state) {
case PLACEABLE:
removeTasksFromQueue(placeable_tasks_, task_ids, removed_tasks);
break;
case WAITING:
removeTasksFromQueue(waiting_tasks_, task_ids, removed_tasks);
break;
case READY:
removeTasksFromQueue(ready_tasks_, task_ids, removed_tasks);
break;
case RUNNING:
removeTasksFromQueue(running_tasks_, task_ids, removed_tasks);
break;
default:
RAY_LOG(ERROR) << "Attempting to move tasks from unrecognized state " << src_state;
}
// Add the tasks to the specified destination queue.
switch (dst_state) {
case PLACEABLE:
queueTasks(placeable_tasks_, removed_tasks);
break;
case WAITING:
queueTasks(waiting_tasks_, removed_tasks);
break;
case READY:
queueTasks(ready_tasks_, removed_tasks);
break;
case RUNNING:
queueTasks(running_tasks_, removed_tasks);
break;
default:
RAY_LOG(ERROR) << "Attempting to move tasks to unrecognized state " << dst_state;
}
}
void SchedulingQueue::QueueUncreatedActorMethods(const std::vector<Task> &tasks) {
queueTasks(uncreated_actor_methods_, tasks);
}
@@ -80,12 +116,12 @@ void SchedulingQueue::QueueWaitingTasks(const std::vector<Task> &tasks) {
queueTasks(waiting_tasks_, tasks);
}
void SchedulingQueue::QueueReadyTasks(const std::vector<Task> &tasks) {
queueTasks(ready_tasks_, tasks);
void SchedulingQueue::QueuePlaceableTasks(const std::vector<Task> &tasks) {
queueTasks(placeable_tasks_, tasks);
}
void SchedulingQueue::QueueScheduledTasks(const std::vector<Task> &tasks) {
queueTasks(scheduled_tasks_, tasks);
void SchedulingQueue::QueueReadyTasks(const std::vector<Task> &tasks) {
queueTasks(ready_tasks_, tasks);
}
void SchedulingQueue::QueueRunningTasks(const std::vector<Task> &tasks) {
src/ray/raylet/scheduling_queue.h
+28 -24
View File
@@ -12,14 +12,15 @@ namespace ray {
namespace raylet {
enum TaskState { INIT, PLACEABLE, WAITING, READY, RUNNING };
/// \class SchedulingQueue
///
/// Encapsulates task queues. Each queue represents a scheduling state for a
/// task. The scheduling state is one of (1) waiting: for object dependencies
/// to become available, (2) ready: object dependencies are available and the
/// task is ready to be scheduled, (3) scheduled: the task has been scheduled
/// but is waiting for a worker, or (4) running: the task has been scheduled
/// and is running on a worker.
/// 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.
class SchedulingQueue {
public:
/// Create a scheduling queue.
@@ -41,23 +42,17 @@ class SchedulingQueue {
/// object dependencies to become available.
const std::list<Task> &GetWaitingTasks() const;
/// Get the queue of tasks in the ready state.
/// Get the queue of tasks in the placeable state.
///
/// \return A const reference to the queue of tasks that have all
/// dependencies local and that are waiting to be scheduled.
const std::list<Task> &GetReadyTasks() const;
const std::list<Task> &GetPlaceableTasks() const;
/// Get the queue of actor methods in the ready state.
/// Get the queue of tasks in the ready state.
///
/// \return A const reference to the queue of actor methods that have all
/// dependencies local and that are waiting to be scheduled.
const std::list<Task> &GetReadyMethods() const;
/// Get the queue of tasks in the scheduled state.
///
/// \return A const reference to the queue of tasks that have been scheduled
/// \return A const reference to the queue of tasks ready
/// to execute but that are waiting for a worker.
const std::list<Task> &GetScheduledTasks() const;
const std::list<Task> &GetReadyTasks() const;
/// Get the queue of tasks in the running state.
///
@@ -85,21 +80,21 @@ class SchedulingQueue {
void QueueUncreatedActorMethods(const std::vector<Task> &tasks);
/// Queue tasks in the waiting state. These are tasks that cannot yet be
/// scheduled since they are blocked on a missing data dependency.
/// dispatched since they are blocked on a missing data dependency.
///
/// \param tasks The tasks to queue.
void QueueWaitingTasks(const std::vector<Task> &tasks);
/// Queue tasks in the placeable state.
///
/// \param tasks The tasks to queue.
void QueuePlaceableTasks(const std::vector<Task> &tasks);
/// Queue tasks in the ready state.
///
/// \param tasks The tasks to queue.
void QueueReadyTasks(const std::vector<Task> &tasks);
/// Queue tasks in the scheduled state.
///
/// \param tasks The tasks to queue.
void QueueScheduledTasks(const std::vector<Task> &tasks);
/// Queue tasks in the running state.
///
/// \param tasks The tasks to queue.
@@ -112,6 +107,15 @@ class SchedulingQueue {
/// \param tasks The tasks to queue.
void QueueBlockedTasks(const std::vector<Task> &tasks);
/// \brief Move the specified tasks from the source state to the destination state.
///
/// \param tasks The set of task IDs to move.
/// \param src_state Source state, which corresponds to one of the internal task queues.
/// \param dst_state Destination state, corresponding to one of the internal task
/// queues.
void MoveTasks(std::unordered_set<TaskID> tasks, TaskState src_state,
TaskState dst_state);
private:
/// Tasks that are destined for actors that have not yet been created.
std::list<Task> uncreated_actor_methods_;
@@ -119,9 +123,9 @@ class SchedulingQueue {
std::list<Task> waiting_tasks_;
/// Tasks whose object dependencies are locally available, but that are
/// waiting to be scheduled.
std::list<Task> placeable_tasks_;
/// Tasks ready for dispatch, but that are waiting for a worker.
std::list<Task> ready_tasks_;
/// Tasks that have been scheduled to run, but that are waiting for a worker.
std::list<Task> scheduled_tasks_;
/// Tasks that are running on a worker.
std::list<Task> running_tasks_;
/// Tasks that were dispatched to a worker but are blocked on a data