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
2026-07-20 12:40:20 +08:00 · 2018-11-20 13:14:12 -08:00
parent b0bfd104f2
commit d3697ce4e1
6 changed files with 199 additions and 142 deletions
@@ -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<Task> &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<ResourceSet, ordered_set<TaskID>> MakeTasksWithResources(
+    const std::vector<Task> &tasks) {
+  std::unordered_map<ResourceSet, ordered_set<TaskID>> 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<TaskID> 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<ResourceSet, ordered_set<TaskID>> &tasks_with_resources) {
+  std::unordered_set<TaskID> 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<LocalClientConnection>
      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<Task>({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<Task> 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));
  }
 }

@@ -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<Task> &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<ResourceSet, ordered_set<TaskID>> &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
@@ -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 <typename TaskQueue>
+void RemoveTasksFromQueue(ray::raylet::TaskState task_state, TaskQueue &queue,
                          std::unordered_set<ray::TaskID> &task_ids,
-                          std::vector<ray::raylet::Task> &removed_tasks,
+                          std::vector<ray::raylet::Task> *removed_tasks,
                          std::vector<ray::raylet::TaskState> *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<ray::raylet::Task> &tasks) {
+template <typename TaskQueue>
+inline void QueueTasks(TaskQueue &queue, const std::vector<ray::raylet::Task> &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 <typename TaskQueue>
+inline void FilterStateFromQueue(const TaskQueue &queue,
                                 std::unordered_set<ray::TaskID> &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 <typename TaskQueue>
+inline void GetDriverTasksFromQueue(const TaskQueue &queue,
                                    const ray::DriverID &driver_id,
                                    std::unordered_set<ray::TaskID> &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 <typename TaskQueue>
+inline void GetActorTasksFromQueue(const TaskQueue &queue, const ray::ActorID &actor_id,
                                   std::unordered_set<ray::TaskID> &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<Task> *removed_tasks) {
+bool TaskQueue::RemoveTask(const TaskID &task_id, std::vector<Task> *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<Task> &SchedulingQueue::TaskQueue::GetTasks() const { return task_list_; }
+const std::list<Task> &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<Task> *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<Task> &SchedulingQueue::GetMethodsWaitingForActorCreation() const {
-  return this->methods_waiting_for_actor_creation_.GetTasks();
+  return methods_waiting_for_actor_creation_.GetTasks();
 }

 const std::list<Task> &SchedulingQueue::GetWaitingTasks() const {
-  return this->waiting_tasks_.GetTasks();
+  return waiting_tasks_.GetTasks();
 }

 const std::list<Task> &SchedulingQueue::GetPlaceableTasks() const {
-  return this->placeable_tasks_.GetTasks();
+  return placeable_tasks_.GetTasks();
 }

 const std::list<Task> &SchedulingQueue::GetReadyTasks() const {
-  return this->ready_tasks_.GetTasks();
+  return ready_tasks_.GetTasks();
 }

 const std::list<Task> &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<Task> &SchedulingQueue::GetRunningTasks() const {
-  return this->running_tasks_.GetTasks();
+  return running_tasks_.GetTasks();
 }

 const std::unordered_set<TaskID> &SchedulingQueue::GetBlockedTaskIds() const {
@@ -210,16 +221,16 @@ std::vector<Task> SchedulingQueue::RemoveTasks(std::unordered_set<TaskID> &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<TaskID> &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 "
@@ -7,6 +7,8 @@
 #include <vector>

 #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<Task> *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<Task> &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> task_list_;
+  /// A hash to speed up looking up a task.
+  std::unordered_map<TaskID, std::list<Task>::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<Task> *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<ResourceSet, ordered_set<TaskID>> &GetTasksWithResources()
+      const {
+    return tasks_with_resources_;
+  }
+
+ private:
+  /// Index from resource shape to tasks that require these resources.
+  std::unordered_map<ResourceSet, ordered_set<TaskID>> 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<Task> &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<Task> *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<Task> &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> task_list_;
-    /// A hash to speed up looking up a task.
-    std::unordered_map<TaskID, std::list<Task>::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
@@ -392,4 +392,18 @@ class SchedulingResources {

 }  // namespace ray

+namespace std {
+template <>
+struct hash<ray::raylet::ResourceSet> {
+  size_t operator()(ray::raylet::ResourceSet const &k) const {
+    size_t seed = k.GetResourceMap().size();
+    for (auto &elem : k.GetResourceMap()) {
+      seed ^= std::hash<std::string>()(elem.first);
+      seed ^= std::hash<double>()(elem.second);
+    }
+    return seed;
+  }
+};
+}
+
 #endif  // RAY_RAYLET_SCHEDULING_RESOURCES_H
@@ -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