diff --git a/src/ray/raylet/lineage_cache.cc b/src/ray/raylet/lineage_cache.cc index 1d23bacff..d1caa8cd6 100644 --- a/src/ray/raylet/lineage_cache.cc +++ b/src/ray/raylet/lineage_cache.cc @@ -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); diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index c5651aa10..be849263d 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -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 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 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})); + local_queues_.QueueReadyTasks(std::vector({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})); + local_queues_.QueueReadyTasks(std::vector({task})); } } @@ -873,10 +881,10 @@ void NodeManager::HandleObjectLocal(const ObjectID &object_id) { if (ready_task_ids.size() > 0) { std::unordered_set 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(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(); } } diff --git a/src/ray/raylet/node_manager.h b/src/ray/raylet/node_manager.h index 640322728..0e897bf07 100644 --- a/src/ray/raylet/node_manager.h +++ b/src/ray/raylet/node_manager.h @@ -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 cluster_resource_map_; /// A pool of workers. WorkerPool worker_pool_; diff --git a/src/ray/raylet/scheduling_policy.cc b/src/ray/raylet/scheduling_policy.cc index 9f3de5424..0ff0ee631 100644 --- a/src/ray/raylet/scheduling_policy.cc +++ b/src/ray/raylet/scheduling_policy.cc @@ -25,7 +25,7 @@ std::unordered_map 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(); diff --git a/src/ray/raylet/scheduling_queue.cc b/src/ray/raylet/scheduling_queue.cc index a1d0a48c1..a9365bcc3 100644 --- a/src/ray/raylet/scheduling_queue.cc +++ b/src/ray/raylet/scheduling_queue.cc @@ -14,12 +14,12 @@ const std::list &SchedulingQueue::GetWaitingTasks() const { return this->waiting_tasks_; } -const std::list &SchedulingQueue::GetReadyTasks() const { - return this->ready_tasks_; +const std::list &SchedulingQueue::GetPlaceableTasks() const { + return this->placeable_tasks_; } -const std::list &SchedulingQueue::GetScheduledTasks() const { - return this->scheduled_tasks_; +const std::list &SchedulingQueue::GetReadyTasks() const { + return this->ready_tasks_; } const std::list &SchedulingQueue::GetRunningTasks() const { @@ -30,10 +30,6 @@ const std::list &SchedulingQueue::GetBlockedTasks() const { return this->blocked_tasks_; } -const std::list &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 &queue, std::unordered_set &task_ids, @@ -62,8 +58,8 @@ std::vector SchedulingQueue::RemoveTasks(std::unordered_set 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 SchedulingQueue::RemoveTasks(std::unordered_set task_i return removed_tasks; } +void SchedulingQueue::MoveTasks(std::unordered_set task_ids, TaskState src_state, + TaskState dst_state) { + // TODO(atumanov): check the states first to ensure the move is transactional. + std::vector 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 &tasks) { queueTasks(uncreated_actor_methods_, tasks); } @@ -80,12 +116,12 @@ void SchedulingQueue::QueueWaitingTasks(const std::vector &tasks) { queueTasks(waiting_tasks_, tasks); } -void SchedulingQueue::QueueReadyTasks(const std::vector &tasks) { - queueTasks(ready_tasks_, tasks); +void SchedulingQueue::QueuePlaceableTasks(const std::vector &tasks) { + queueTasks(placeable_tasks_, tasks); } -void SchedulingQueue::QueueScheduledTasks(const std::vector &tasks) { - queueTasks(scheduled_tasks_, tasks); +void SchedulingQueue::QueueReadyTasks(const std::vector &tasks) { + queueTasks(ready_tasks_, tasks); } void SchedulingQueue::QueueRunningTasks(const std::vector &tasks) { diff --git a/src/ray/raylet/scheduling_queue.h b/src/ray/raylet/scheduling_queue.h index 7da3050f3..e8a7fbca7 100644 --- a/src/ray/raylet/scheduling_queue.h +++ b/src/ray/raylet/scheduling_queue.h @@ -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 &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 &GetReadyTasks() const; + const std::list &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 &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 &GetScheduledTasks() const; + const std::list &GetReadyTasks() const; /// Get the queue of tasks in the running state. /// @@ -85,21 +80,21 @@ class SchedulingQueue { void QueueUncreatedActorMethods(const std::vector &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 &tasks); + /// Queue tasks in the placeable state. + /// + /// \param tasks The tasks to queue. + void QueuePlaceableTasks(const std::vector &tasks); + /// Queue tasks in the ready state. /// /// \param tasks The tasks to queue. void QueueReadyTasks(const std::vector &tasks); - /// Queue tasks in the scheduled state. - /// - /// \param tasks The tasks to queue. - void QueueScheduledTasks(const std::vector &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 &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 tasks, TaskState src_state, + TaskState dst_state); + private: /// Tasks that are destined for actors that have not yet been created. std::list uncreated_actor_methods_; @@ -119,9 +123,9 @@ class SchedulingQueue { std::list waiting_tasks_; /// Tasks whose object dependencies are locally available, but that are /// waiting to be scheduled. + std::list placeable_tasks_; + /// Tasks ready for dispatch, but that are waiting for a worker. std::list ready_tasks_; - /// Tasks that have been scheduled to run, but that are waiting for a worker. - std::list scheduled_tasks_; /// Tasks that are running on a worker. std::list running_tasks_; /// Tasks that were dispatched to a worker but are blocked on a data