Avoid modifying leased_workers during iterate, block tasks from being…

… dispatched if priority is too low

src/ray/core_worker/reference_count.h

@@ -476,9 +476,6 @@ class ReferenceCounter : public ReferenceCounterInterface,
       LOCKS_EXCLUDED(mutex_);
 
  private:
-  // TODO: Add priority to the Reference struct.
-  // TODO: Track how many tasks have been submitted so far that depend on an
-  // object, use for DFS-based priority calculation.
   struct Reference {
     /// Constructor for a reference whose origin is unknown.
     Reference() {}

src/ray/raylet/scheduling/cluster_task_manager.cc

@@ -78,8 +78,9 @@ bool ClusterTaskManager::SchedulePendingTasks() {
       // tasks from being scheduled.
       Priority task_priority = work_it->first.first;
       if(task_priority >= block_requested_priority_){
-		return did_schedule;
+        return did_schedule;
       }
+
       const std::shared_ptr<Work> &work = work_it->second;
       RayTask task = work->task;
       RAY_LOG(DEBUG) << "Scheduling pending task "
@@ -294,6 +295,13 @@ void ClusterTaskManager::DispatchScheduledTasksToWorkers(
         continue;
       }
 
+      // Block tasks of a lower priority.
+      Priority task_priority = work_it->first.first;
+      if(task_priority >= block_requested_priority_){
+        break;
+      }
+
+
       bool args_missing = false;
       bool success = PinTaskArgsIfMemoryAvailable(spec, &args_missing);
       // An argument was evicted since this task was added to the dispatch
@@ -1199,17 +1207,23 @@ bool ClusterTaskManager::ReturnCpuResourcesToBlockedWorker(
 
 bool ClusterTaskManager::EvictTasks(Priority base_priority) {
   bool should_spill = true;
+  std::vector<std::shared_ptr<WorkerInterface>> workers_to_kill;
   for (auto &entry : leased_workers_) {
     std::shared_ptr<WorkerInterface> worker = entry.second;
     Priority priority = worker->GetAssignedTask().GetTaskSpecification().GetPriority();
     //Smaller priority have higher priority
     //Does not have less than check it
-    if(priority >= base_priority){
-      //Consider Using CancelTask instead of DestroyWorker
-      destroy_worker_(worker, rpc::WorkerExitType::INTENDED_EXIT);
-	  should_spill = false;
+    if (priority >= base_priority){
+      workers_to_kill.push_back(worker);
+      should_spill = false;
     }
   }
+
+  for (auto &worker : workers_to_kill) {
+    //Consider Using CancelTask instead of DestroyWorker
+    destroy_worker_(worker, rpc::WorkerExitType::INTENDED_EXIT);
+  }
+
   //Check Deadlock corner cases
   //Finer granularity preemption is not considered, kill all the lower priorities
   return should_spill;