From d3697ce4e1b435066b40270b19f479cb708a2540 Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Tue, 20 Nov 2018 13:14:12 -0800 Subject: [PATCH] Ready queue refactor to make Dispatching tasks more efficient (#3324) * put queues outside * working version, still needs to be optimized * implement round robin * proper round robin * fix spillback * update * fix * cleanup * more cleanups * fix * fix * add documentation * explanation for hash combiner * speed it up * cleanup and linting * linting * comments * Update scheduling_queue.h * temp commit * fixes * update * fix * cleanup * cleanup * lint * more prints * more prints * increase sleep * documentation * sleep * fix * fix * sleep longer * update * fix * fix * fix * Add ordered_set container. * Fix * Linting * Constructors * Remove O(n) call to list.size(). * fixes * use ordered set * Fix. * Add documentation. * Add iterators to ordered_set container implementation. * iterator_type -> iterator * Make typedefs private * Add const_iterator * fix * fix test * linting * lint * update * add documentation * linting --- src/ray/raylet/node_manager.cc | 76 ++++++-------- src/ray/raylet/node_manager.h | 14 +-- src/ray/raylet/scheduling_queue.cc | 88 +++++++++------- src/ray/raylet/scheduling_queue.h | 145 ++++++++++++++++---------- src/ray/raylet/scheduling_resources.h | 14 +++ test/actor_test.py | 4 +- 6 files changed, 199 insertions(+), 142 deletions(-) diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 1cf320953..1efa3a487 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -59,7 +59,6 @@ NodeManager::NodeManager(boost::asio::io_service &io_service, local_available_resources_(config.resource_config), worker_pool_(config.num_initial_workers, config.num_workers_per_process, config.maximum_startup_concurrency, config.worker_commands), - local_queues_(SchedulingQueue()), scheduling_policy_(local_queues_), reconstruction_policy_( io_service_, @@ -532,34 +531,36 @@ void NodeManager::ProcessNewClient(LocalClientConnection &client) { client.ProcessMessages(); } -void NodeManager::DispatchTasks(const std::list &ready_tasks) { - // Return if there are no tasks to schedule. - if (ready_tasks.empty()) { - return; +// A helper function to create a mapping from resource shapes to +// tasks with that resource shape from a given list of tasks. +std::unordered_map> MakeTasksWithResources( + const std::vector &tasks) { + std::unordered_map> result; + for (const auto &task : tasks) { + auto spec = task.GetTaskSpecification(); + result[spec.GetRequiredResources()].push_back(spec.TaskId()); } + return result; +} - // Remember ids of the task we need to remove from ready queue. - std::unordered_set removed_task_ids = {}; - - for (const auto &task : ready_tasks) { - const auto &task_resources = task.GetTaskSpecification().GetRequiredResources(); - if (!local_available_resources_.Contains(task_resources)) { - // Not enough local resources for this task right now, skip this task. - // TODO(rkn): We should always skip node managers that have 0 CPUs. - continue; - } - // We have enough resources for this task. Assign task. - if (AssignTask(task)) { - // We were successful in assigning this task on a local worker, so - // remember to remove it from ready queue. If for some reason the - // scheduling of this task fails later, we will add it back to the - // ready queue. - removed_task_ids.insert(task.GetTaskSpecification().TaskId()); +void NodeManager::DispatchTasks( + const std::unordered_map> &tasks_with_resources) { + std::unordered_set removed_task_ids; + for (const auto &it : tasks_with_resources) { + for (const auto &task_id : it.second) { + const auto &task = local_queues_.GetReadyQueue().GetTask(task_id); + const auto &task_resources = task.GetTaskSpecification().GetRequiredResources(); + if (!local_available_resources_.Contains(task_resources)) { + // All the tasks in it.second have the same resource shape, so + // once the first task is not feasible, we can break out of this loop + break; + } + if (AssignTask(task)) { + removed_task_ids.insert(task_id); + } } } - if (!removed_task_ids.empty()) { - local_queues_.RemoveTasks(removed_task_ids); - } + local_queues_.RemoveTasks(removed_task_ids); } void NodeManager::ProcessClientMessage( @@ -642,7 +643,7 @@ void NodeManager::ProcessRegisterClientRequestMessage( if (message->is_worker()) { // Register the new worker. worker_pool_.RegisterWorker(std::move(worker)); - DispatchTasks(local_queues_.GetReadyTasks()); + DispatchTasks(local_queues_.GetReadyQueue().GetTasksWithResources()); } else { // Register the new driver. Note that here the driver_id in RegisterClientRequest // message is actually the ID of the driver task, while client_id represents the @@ -671,7 +672,7 @@ void NodeManager::ProcessGetTaskMessage( cluster_resource_map_[local_client_id].SetLoadResources( local_queues_.GetResourceLoad()); // Call task dispatch to assign work to the new worker. - DispatchTasks(local_queues_.GetReadyTasks()); + DispatchTasks(local_queues_.GetReadyQueue().GetTasksWithResources()); } void NodeManager::ProcessDisconnectClientMessage( @@ -768,7 +769,7 @@ void NodeManager::ProcessDisconnectClientMessage( << "driver_id: " << worker->GetAssignedDriverId(); // Since some resources may have been released, we can try to dispatch more tasks. - DispatchTasks(local_queues_.GetReadyTasks()); + DispatchTasks(local_queues_.GetReadyQueue().GetTasksWithResources()); } else if (is_driver) { // The client is a driver. RAY_CHECK_OK(gcs_client_->driver_table().AppendDriverData(client->GetClientId(), @@ -1165,7 +1166,7 @@ void NodeManager::HandleTaskBlocked(const std::shared_ptr worker->MarkBlocked(); // Try dispatching tasks since we may have released some resources. - DispatchTasks(local_queues_.GetReadyTasks()); + DispatchTasks(local_queues_.GetReadyQueue().GetTasksWithResources()); } } else { // The client is a driver. Drivers do not hold resources, so we simply mark @@ -1259,11 +1260,7 @@ void NodeManager::EnqueuePlaceableTask(const Task &task) { // (See design_docs/task_states.rst for the state transition diagram.) if (args_ready) { local_queues_.QueueReadyTasks({task}); - // Dispatch just the new who was added to the ready task. - // The other tasks in the ready queue can be ignored as no new resources - // have been added and no new worker has became available since the last - // time DispatchTasks() was called. - DispatchTasks({task}); + DispatchTasks(MakeTasksWithResources({task})); } else { local_queues_.QueueWaitingTasks({task}); } @@ -1375,8 +1372,8 @@ bool NodeManager::AssignTask(const Task &task) { // DispatchTasks() removed it from the ready queue. 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({assigned_task})); - DispatchTasks({assigned_task}); + local_queues_.QueueReadyTasks({assigned_task}); + DispatchTasks(MakeTasksWithResources({assigned_task})); } }); @@ -1525,12 +1522,7 @@ void NodeManager::HandleObjectLocal(const ObjectID &object_id) { // Queue and dispatch the tasks that are ready to run (i.e., WAITING). auto ready_tasks = local_queues_.RemoveTasks(ready_task_id_set); local_queues_.QueueReadyTasks(ready_tasks); - const std::list ready_tasks_list(ready_tasks.begin(), ready_tasks.end()); - // Dispatch only the new ready tasks whose dependencies were fulfilled. - // The other tasks in the ready queue can be ignored as no new resources - // have been added and no new worker has became available since the last - // time DispatchTasks() was called. - DispatchTasks(ready_tasks_list); + DispatchTasks(MakeTasksWithResources(ready_tasks)); } } diff --git a/src/ray/raylet/node_manager.h b/src/ray/raylet/node_manager.h index 8a2661661..6d52b734c 100644 --- a/src/ray/raylet/node_manager.h +++ b/src/ray/raylet/node_manager.h @@ -16,6 +16,7 @@ #include "ray/raylet/reconstruction_policy.h" #include "ray/raylet/task_dependency_manager.h" #include "ray/raylet/worker_pool.h" +#include "ray/util/ordered_set.h" // clang-format on namespace ray { @@ -210,7 +211,7 @@ class NodeManager { /// Dispatch locally scheduled tasks. This attempts the transition from "scheduled" to /// "running" task state. /// - /// This function is called in one of the following cases: + /// This function is called in the following cases: /// (1) A set of new tasks is added to the ready queue. /// (2) New resources are becoming available on the local node. /// (3) A new worker becomes available. @@ -218,11 +219,12 @@ class NodeManager { /// ready queue, as we know that the old tasks in the ready queue cannot /// be scheduled (We checked those tasks last time new resources or /// workers became available, and nothing changed since then.) In this case, - /// task_queue contains only the newly added tasks to the ready queue; - /// Otherwise, task_queue points to entire ready queue. - /// - /// \param ready_tasks Tasks to be dispatched, a subset from ready queue. - void DispatchTasks(const std::list &ready_tasks); + /// tasks_with_resources contains only the newly added tasks to the + /// ready queue. Otherwise, tasks_with_resources points to entire ready queue. + /// \param tasks_with_resources Mapping from resource shapes to tasks with + /// that resource shape. + void DispatchTasks( + const std::unordered_map> &tasks_with_resources); /// Handle a task that is blocked. This could be a task assigned to a worker, /// an out-of-band task (e.g., a thread created by the application), or a diff --git a/src/ray/raylet/scheduling_queue.cc b/src/ray/raylet/scheduling_queue.cc index 7f9dda8ec..b9045e579 100644 --- a/src/ray/raylet/scheduling_queue.cc +++ b/src/ray/raylet/scheduling_queue.cc @@ -8,13 +8,13 @@ namespace { // 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(ray::raylet::TaskState task_state, - ray::raylet::SchedulingQueue::TaskQueue &queue, +template +void RemoveTasksFromQueue(ray::raylet::TaskState task_state, TaskQueue &queue, std::unordered_set &task_ids, - std::vector &removed_tasks, + std::vector *removed_tasks, std::vector *task_states = nullptr) { for (auto it = task_ids.begin(); it != task_ids.end();) { - if (queue.RemoveTask(*it, &removed_tasks)) { + if (queue.RemoveTask(*it, removed_tasks)) { it = task_ids.erase(it); if (task_states != nullptr) { task_states->push_back(task_state); @@ -26,15 +26,16 @@ void RemoveTasksFromQueue(ray::raylet::TaskState task_state, } // Helper function to queue the given tasks to the given queue. -inline void QueueTasks(ray::raylet::SchedulingQueue::TaskQueue &queue, - const std::vector &tasks) { +template +inline void QueueTasks(TaskQueue &queue, const std::vector &tasks) { for (const auto &task : tasks) { queue.AppendTask(task.GetTaskSpecification().TaskId(), task); } } // Helper function to filter out tasks of a given state. -inline void FilterStateFromQueue(const ray::raylet::SchedulingQueue::TaskQueue &queue, +template +inline void FilterStateFromQueue(const TaskQueue &queue, std::unordered_set &task_ids, ray::raylet::TaskState filter_state) { for (auto it = task_ids.begin(); it != task_ids.end();) { @@ -47,7 +48,8 @@ inline void FilterStateFromQueue(const ray::raylet::SchedulingQueue::TaskQueue & } // Helper function to get tasks for a driver from a given state. -inline void GetDriverTasksFromQueue(const ray::raylet::SchedulingQueue::TaskQueue &queue, +template +inline void GetDriverTasksFromQueue(const TaskQueue &queue, const ray::DriverID &driver_id, std::unordered_set &task_ids) { const auto &tasks = queue.GetTasks(); @@ -60,8 +62,8 @@ inline void GetDriverTasksFromQueue(const ray::raylet::SchedulingQueue::TaskQueu } // Helper function to get tasks for an actor from a given state. -inline void GetActorTasksFromQueue(const ray::raylet::SchedulingQueue::TaskQueue &queue, - const ray::ActorID &actor_id, +template +inline void GetActorTasksFromQueue(const TaskQueue &queue, const ray::ActorID &actor_id, std::unordered_set &task_ids) { const auto &tasks = queue.GetTasks(); for (const auto &task : tasks) { @@ -78,12 +80,7 @@ namespace ray { namespace raylet { -SchedulingQueue::TaskQueue::~TaskQueue() { - task_map_.clear(); - task_list_.clear(); -} - -bool SchedulingQueue::TaskQueue::AppendTask(const TaskID &task_id, const Task &task) { +bool TaskQueue::AppendTask(const TaskID &task_id, const Task &task) { RAY_CHECK(task_map_.find(task_id) == task_map_.end()); auto list_iterator = task_list_.insert(task_list_.end(), task); task_map_[task_id] = list_iterator; @@ -92,8 +89,7 @@ bool SchedulingQueue::TaskQueue::AppendTask(const TaskID &task_id, const Task &t return true; } -bool SchedulingQueue::TaskQueue::RemoveTask(const TaskID &task_id, - std::vector *removed_tasks) { +bool TaskQueue::RemoveTask(const TaskID &task_id, std::vector *removed_tasks) { auto task_found_iterator = task_map_.find(task_id); if (task_found_iterator == task_map_.end()) { return false; @@ -111,34 +107,49 @@ bool SchedulingQueue::TaskQueue::RemoveTask(const TaskID &task_id, return true; } -bool SchedulingQueue::TaskQueue::HasTask(const TaskID &task_id) const { +bool TaskQueue::HasTask(const TaskID &task_id) const { return task_map_.find(task_id) != task_map_.end(); } -const std::list &SchedulingQueue::TaskQueue::GetTasks() const { return task_list_; } +const std::list &TaskQueue::GetTasks() const { return task_list_; } -const ResourceSet &SchedulingQueue::TaskQueue::GetCurrentResourceLoad() const { +const ResourceSet &TaskQueue::GetCurrentResourceLoad() const { return current_resource_load_; } +bool ReadyQueue::AppendTask(const TaskID &task_id, const Task &task) { + const auto &resources = task.GetTaskSpecification().GetRequiredResources(); + tasks_with_resources_[resources].push_back(task_id); + return TaskQueue::AppendTask(task_id, task); +} + +bool ReadyQueue::RemoveTask(const TaskID &task_id, std::vector *removed_tasks) { + if (task_map_.find(task_id) != task_map_.end()) { + const auto &resources = + task_map_[task_id]->GetTaskSpecification().GetRequiredResources(); + tasks_with_resources_[resources].erase(task_id); + } + return TaskQueue::RemoveTask(task_id, removed_tasks); +} + const std::list &SchedulingQueue::GetMethodsWaitingForActorCreation() const { - return this->methods_waiting_for_actor_creation_.GetTasks(); + return methods_waiting_for_actor_creation_.GetTasks(); } const std::list &SchedulingQueue::GetWaitingTasks() const { - return this->waiting_tasks_.GetTasks(); + return waiting_tasks_.GetTasks(); } const std::list &SchedulingQueue::GetPlaceableTasks() const { - return this->placeable_tasks_.GetTasks(); + return placeable_tasks_.GetTasks(); } const std::list &SchedulingQueue::GetReadyTasks() const { - return this->ready_tasks_.GetTasks(); + return ready_tasks_.GetTasks(); } const std::list &SchedulingQueue::GetInfeasibleTasks() const { - return this->infeasible_tasks_.GetTasks(); + return infeasible_tasks_.GetTasks(); } ResourceSet SchedulingQueue::GetReadyQueueResources() const { @@ -151,7 +162,7 @@ ResourceSet SchedulingQueue::GetResourceLoad() const { } const std::list &SchedulingQueue::GetRunningTasks() const { - return this->running_tasks_.GetTasks(); + return running_tasks_.GetTasks(); } const std::unordered_set &SchedulingQueue::GetBlockedTaskIds() const { @@ -210,16 +221,16 @@ std::vector SchedulingQueue::RemoveTasks(std::unordered_set &task_ // Try to find the tasks to remove from the queues. RemoveTasksFromQueue(TaskState::WAITING_FOR_ACTOR, methods_waiting_for_actor_creation_, - task_ids, removed_tasks, task_states); - RemoveTasksFromQueue(TaskState::WAITING, waiting_tasks_, task_ids, removed_tasks, + task_ids, &removed_tasks, task_states); + RemoveTasksFromQueue(TaskState::WAITING, waiting_tasks_, task_ids, &removed_tasks, task_states); - RemoveTasksFromQueue(TaskState::PLACEABLE, placeable_tasks_, task_ids, removed_tasks, + RemoveTasksFromQueue(TaskState::PLACEABLE, placeable_tasks_, task_ids, &removed_tasks, task_states); - RemoveTasksFromQueue(TaskState::READY, ready_tasks_, task_ids, removed_tasks, + RemoveTasksFromQueue(TaskState::READY, ready_tasks_, task_ids, &removed_tasks, task_states); - RemoveTasksFromQueue(TaskState::RUNNING, running_tasks_, task_ids, removed_tasks, + RemoveTasksFromQueue(TaskState::RUNNING, running_tasks_, task_ids, &removed_tasks, task_states); - RemoveTasksFromQueue(TaskState::INFEASIBLE, infeasible_tasks_, task_ids, removed_tasks, + RemoveTasksFromQueue(TaskState::INFEASIBLE, infeasible_tasks_, task_ids, &removed_tasks, task_states); RAY_CHECK(task_ids.size() == 0); @@ -251,20 +262,21 @@ void SchedulingQueue::MoveTasks(std::unordered_set &task_ids, TaskState // Remove the tasks from the specified source queue. switch (src_state) { case TaskState::PLACEABLE: - RemoveTasksFromQueue(TaskState::PLACEABLE, placeable_tasks_, task_ids, removed_tasks); + RemoveTasksFromQueue(TaskState::PLACEABLE, placeable_tasks_, task_ids, + &removed_tasks); break; case TaskState::WAITING: - RemoveTasksFromQueue(TaskState::WAITING, waiting_tasks_, task_ids, removed_tasks); + RemoveTasksFromQueue(TaskState::WAITING, waiting_tasks_, task_ids, &removed_tasks); break; case TaskState::READY: - RemoveTasksFromQueue(TaskState::READY, ready_tasks_, task_ids, removed_tasks); + RemoveTasksFromQueue(TaskState::READY, ready_tasks_, task_ids, &removed_tasks); break; case TaskState::RUNNING: - RemoveTasksFromQueue(TaskState::RUNNING, running_tasks_, task_ids, removed_tasks); + RemoveTasksFromQueue(TaskState::RUNNING, running_tasks_, task_ids, &removed_tasks); break; case TaskState::INFEASIBLE: RemoveTasksFromQueue(TaskState::INFEASIBLE, infeasible_tasks_, task_ids, - removed_tasks); + &removed_tasks); break; default: RAY_LOG(FATAL) << "Attempting to move tasks from unrecognized state " diff --git a/src/ray/raylet/scheduling_queue.h b/src/ray/raylet/scheduling_queue.h index ef9f8ca6c..bdd065fa1 100644 --- a/src/ray/raylet/scheduling_queue.h +++ b/src/ray/raylet/scheduling_queue.h @@ -7,6 +7,8 @@ #include #include "ray/raylet/task.h" +#include "ray/util/logging.h" +#include "ray/util/ordered_set.h" namespace ray { @@ -40,6 +42,89 @@ enum class TaskState { WAITING_FOR_ACTOR, }; +class TaskQueue { + public: + /// \brief Append a task to queue. + /// + /// \param task_id The task ID for the task to append. + /// \param task The task to append to the queue. + /// \return Whether the append operation succeeds. + bool AppendTask(const TaskID &task_id, const Task &task); + + /// \brief Remove a task from queue. + /// + /// \param task_id The task ID for the task to remove from the queue. + /// \param removed_tasks If the task specified by task_id is successfully + /// removed from the queue, the task data is appended to the vector. Can + /// be a nullptr, in which case nothing is appended. + /// \return Whether the removal succeeds. + bool RemoveTask(const TaskID &task_id, std::vector *removed_tasks = nullptr); + + /// \brief Check if the queue contains a specific task id. + /// + /// \param task_id The task ID for the task. + /// \return Whether the task_id exists in this queue. + bool HasTask(const TaskID &task_id) const; + + /// \brief Return the task list of the queue. + /// \return A list of tasks contained in this queue. + const std::list &GetTasks() const; + + /// \brief Get the total resources required by the tasks in the queue. + /// \return Total resources required by the tasks in the queue. + const ResourceSet &GetCurrentResourceLoad() const; + + protected: + /// A list of tasks. + std::list task_list_; + /// A hash to speed up looking up a task. + std::unordered_map::iterator> task_map_; + /// Aggregate resources of all the tasks in this queue. + ResourceSet current_resource_load_; +}; + +class ReadyQueue : public TaskQueue { + public: + ReadyQueue(){}; + + ReadyQueue(const ReadyQueue &other) = delete; + + /// \brief Append a task to queue. + /// + /// \param task_id The task ID for the task to append. + /// \param task The task to append to the queue. + /// \return Whether the append operation succeeds. + bool AppendTask(const TaskID &task_id, const Task &task); + + /// \brief Remove a task from queue. + /// + /// \param task_id The task ID for the task to remove from the queue. + /// \return Whether the removal succeeds. + bool RemoveTask(const TaskID &task_id, std::vector *removed_tasks); + + /// \brief Get task associated to task_id in this queue. + /// + /// \param task_id The task ID for the task to get. + /// \return The task corresponding to task_id. + const Task &GetTask(const TaskID &task_id) const { + auto it = task_map_.find(task_id); + RAY_CHECK(it != task_map_.end()); + return *it->second; + } + + /// \brief Get a mapping from resource shape to tasks. + /// + /// \return Mapping from resource set to task IDs with these resource requirements. + const std::unordered_map> &GetTasksWithResources() + const { + return tasks_with_resources_; + } + + private: + /// Index from resource shape to tasks that require these resources. + std::unordered_map> tasks_with_resources_; +}; + /// \class SchedulingQueue /// /// Encapsulates task queues. @@ -95,6 +180,11 @@ class SchedulingQueue { /// to execute but that are waiting for a worker. const std::list &GetReadyTasks() const; + /// Get a reference to the queue of ready tasks. + /// + /// \return A reference to the queue of ready tasks. + const ReadyQueue &GetReadyQueue() const { return ready_tasks_; } + /// Get the queue of tasks in the running state. /// /// \return A const reference to the queue of tasks that are currently @@ -222,59 +312,6 @@ class SchedulingQueue { /// \return Aggregate resource demand from ready tasks. ResourceSet GetReadyQueueResources() const; - class TaskQueue { - public: - /// Creating a task queue. - TaskQueue() {} - - /// Destructor for task queue. - ~TaskQueue(); - - /// \brief Append a task to queue. - /// - /// \param task_id The task ID for the task to append. - /// \param task The task to append to the queue. - /// \return Whether the append operation succeeds. - bool AppendTask(const TaskID &task_id, const Task &task); - - /// \brief Remove a task from queue. - /// - /// \param task_id The task ID for the task to remove from the queue. - /// \return Whether the removal succeeds. - bool RemoveTask(const TaskID &task_id); - - /// \brief Remove a task from queue. - /// - /// \param task_id The task ID for the task to remove from the queue. - /// \param removed_tasks If the task specified by task_id is successfully - /// removed from the queue, the task data is appended to the vector. Can - /// be a nullptr, in which case nothing is appended. - /// \return Whether the removal succeeds. - bool RemoveTask(const TaskID &task_id, std::vector *removed_tasks = nullptr); - - /// \brief Check if the queue contains a specific task id. - /// - /// \param task_id The task ID for the task. - /// \return Whether the task_id exists in this queue. - bool HasTask(const TaskID &task_id) const; - - /// \brief Remove the task list of the queue. - /// \return A list of tasks contained in this queue. - const std::list &GetTasks() const; - - /// \brief Get the total resources required by the tasks in the queue. - /// \return Total resources required by the tasks in the queue. - const ResourceSet &GetCurrentResourceLoad() const; - - private: - /// A list of tasks. - std::list task_list_; - /// A hash to speed up looking up a task. - std::unordered_map::iterator> task_map_; - /// Aggregate resources of all the tasks in this queue. - ResourceSet current_resource_load_; - }; - /// Returns debug string for class. /// /// \return string. @@ -289,7 +326,7 @@ class SchedulingQueue { /// waiting to be scheduled. TaskQueue placeable_tasks_; /// Tasks ready for dispatch, but that are waiting for a worker. - TaskQueue ready_tasks_; + ReadyQueue ready_tasks_; /// Tasks that are running on a worker. TaskQueue running_tasks_; /// Tasks that were dispatched to a worker but are blocked on a data diff --git a/src/ray/raylet/scheduling_resources.h b/src/ray/raylet/scheduling_resources.h index ba6fce2da..80b2abea6 100644 --- a/src/ray/raylet/scheduling_resources.h +++ b/src/ray/raylet/scheduling_resources.h @@ -392,4 +392,18 @@ class SchedulingResources { } // namespace ray +namespace std { +template <> +struct hash { + size_t operator()(ray::raylet::ResourceSet const &k) const { + size_t seed = k.GetResourceMap().size(); + for (auto &elem : k.GetResourceMap()) { + seed ^= std::hash()(elem.first); + seed ^= std::hash()(elem.second); + } + return seed; + } +}; +} + #endif // RAY_RAYLET_SCHEDULING_RESOURCES_H diff --git a/test/actor_test.py b/test/actor_test.py index 8981292e0..dc83d3d90 100644 --- a/test/actor_test.py +++ b/test/actor_test.py @@ -1020,7 +1020,7 @@ def test_actors_and_tasks_with_gpus(shutdown_only): @ray.remote(num_gpus=1) def f1(): t1 = time.monotonic() - time.sleep(0.1) + time.sleep(0.2) t2 = time.monotonic() gpu_ids = ray.get_gpu_ids() assert len(gpu_ids) == 1 @@ -1031,7 +1031,7 @@ def test_actors_and_tasks_with_gpus(shutdown_only): @ray.remote(num_gpus=2) def f2(): t1 = time.monotonic() - time.sleep(0.1) + time.sleep(0.2) t2 = time.monotonic() gpu_ids = ray.get_gpu_ids() assert len(gpu_ids) == 2