[core] Fix all raylet variable shadowing (#51689)

Signed-off-by: dentiny <dentinyhao@gmail.com>

Author: dentiny

src/ray/object_manager/plasma/store.cc

@@ -380,7 +380,6 @@ Status PlasmaStore::ProcessClientMessage(std::shared_ptr<Client> client,
   // TODO(suquark): We should convert these interfaces to const later.
   const uint8_t *input = const_cast<uint8_t *>(message.data());
   size_t input_size = message.size();
-  ObjectID object_id;
 
   // Process the different types of requests.
   switch (type) {
@@ -425,6 +424,7 @@ Status PlasmaStore::ProcessClientMessage(std::shared_ptr<Client> client,
     ReplyToCreateClient(client, object_id, request->request_id());
   } break;
   case fb::MessageType::PlasmaAbortRequest: {
+    ObjectID object_id;
     RAY_RETURN_NOT_OK(ReadAbortRequest(input, input_size, &object_id));
     RAY_CHECK(AbortObject(object_id, client) == 1) << "To abort an object, the only "
                                                       "client currently using it "
@@ -443,6 +443,7 @@ Status PlasmaStore::ProcessClientMessage(std::shared_ptr<Client> client,
     // May unmap: client knows a fallback-allocated fd is involved.
     // Should unmap: server finds refcnt == 0 -> need to be unmapped.
     bool may_unmap;
+    ObjectID object_id;
     RAY_RETURN_NOT_OK(ReadReleaseRequest(input, input_size, &object_id, &may_unmap));
     bool should_unmap = ReleaseObject(object_id, client);
     if (!may_unmap) {
@@ -468,6 +469,7 @@ Status PlasmaStore::ProcessClientMessage(std::shared_ptr<Client> client,
     RAY_RETURN_NOT_OK(SendDeleteReply(client, object_ids, error_codes));
   } break;
   case fb::MessageType::PlasmaContainsRequest: {
+    ObjectID object_id;
     RAY_RETURN_NOT_OK(ReadContainsRequest(input, input_size, &object_id));
     if (object_lifecycle_mgr_.IsObjectSealed(object_id)) {
       RAY_RETURN_NOT_OK(SendContainsReply(client, object_id, 1));
@@ -476,6 +478,7 @@ Status PlasmaStore::ProcessClientMessage(std::shared_ptr<Client> client,
     }
   } break;
   case fb::MessageType::PlasmaSealRequest: {
+    ObjectID object_id;
     RAY_RETURN_NOT_OK(ReadSealRequest(input, input_size, &object_id));
     SealObjects({object_id});
     RAY_RETURN_NOT_OK(SendSealReply(client, object_id, PlasmaError::OK));

src/ray/raylet/dependency_manager.cc

@@ -97,19 +97,19 @@ void DependencyManager::StartOrUpdateWaitRequest(
 
 void DependencyManager::CancelWaitRequest(const WorkerID &worker_id) {
   RAY_LOG(DEBUG) << "Canceling wait request for worker " << worker_id;
-  auto it = wait_requests_.find(worker_id);
-  if (it == wait_requests_.end()) {
+  auto req_iter = wait_requests_.find(worker_id);
+  if (req_iter == wait_requests_.end()) {
     return;
   }
 
-  for (const auto &obj_id : it->second) {
-    auto it = required_objects_.find(obj_id);
-    RAY_CHECK(it != required_objects_.end());
-    it->second.dependent_wait_requests.erase(worker_id);
-    RemoveObjectIfNotNeeded(it);
+  for (const auto &obj_id : req_iter->second) {
+    auto obj_iter = required_objects_.find(obj_id);
+    RAY_CHECK(obj_iter != required_objects_.end());
+    obj_iter->second.dependent_wait_requests.erase(worker_id);
+    RemoveObjectIfNotNeeded(obj_iter);
   }
 
-  wait_requests_.erase(it);
+  wait_requests_.erase(req_iter);
 }
 
 void DependencyManager::StartOrUpdateGetRequest(
@@ -152,23 +152,23 @@ void DependencyManager::StartOrUpdateGetRequest(
 
 void DependencyManager::CancelGetRequest(const WorkerID &worker_id) {
   RAY_LOG(DEBUG) << "Canceling get request for worker " << worker_id;
-  auto it = get_requests_.find(worker_id);
-  if (it == get_requests_.end()) {
+  auto req_iter = get_requests_.find(worker_id);
+  if (req_iter == get_requests_.end()) {
     return;
   }
 
   RAY_LOG(DEBUG) << "Canceling pull for get request from worker " << worker_id
-                 << " request: " << it->second.second;
-  object_manager_.CancelPull(it->second.second);
-
-  for (const auto &obj_id : it->second.first) {
-    auto it = required_objects_.find(obj_id);
-    RAY_CHECK(it != required_objects_.end());
-    it->second.dependent_get_requests.erase(worker_id);
-    RemoveObjectIfNotNeeded(it);
+                 << " request: " << req_iter->second.second;
+  object_manager_.CancelPull(req_iter->second.second);
+
+  for (const auto &obj_id : req_iter->second.first) {
+    auto obj_iter = required_objects_.find(obj_id);
+    RAY_CHECK(obj_iter != required_objects_.end());
+    obj_iter->second.dependent_get_requests.erase(worker_id);
+    RemoveObjectIfNotNeeded(obj_iter);
   }
 
-  get_requests_.erase(it);
+  get_requests_.erase(req_iter);
 }
 
 /// Request dependencies for a queued task.

src/ray/raylet/local_task_manager.cc

@@ -172,14 +172,14 @@ void LocalTaskManager::DispatchScheduledTasksToWorkers() {
     // requests.
     size_t num_classes_with_cpu = 0;
     for (const auto &entry : tasks_to_dispatch_) {
-      const auto &dispatch_queue = entry.second;
-      for (const auto &work : dispatch_queue) {
+      const auto &cur_dispatch_queue = entry.second;
+      for (const auto &work : cur_dispatch_queue) {
         const auto &task_spec = work->task.GetTaskSpecification();
         auto cpu_request_ =
             task_spec.GetRequiredResources().Get(scheduling::ResourceID::CPU()).Double();
         if (cpu_request_ > 0) {
           num_classes_with_cpu++;
-          total_cpu_requests_ += dispatch_queue.size() * cpu_request_;
+          total_cpu_requests_ += cur_dispatch_queue.size() * cpu_request_;
           break;
         }
       }
@@ -201,9 +201,10 @@ void LocalTaskManager::DispatchScheduledTasksToWorkers() {
       size_t total_cpu_running_tasks = 0;
       for (auto &entry : info_by_sched_cls_) {
         // Only consider CPU requests
-        const auto &sched_cls_desc =
+        const auto &cur_sched_cls_desc =
             TaskSpecification::GetSchedulingClassDescriptor(entry.first);
-        if (sched_cls_desc.resource_set.Get(scheduling::ResourceID::CPU()).Double() > 0) {
+        if (cur_sched_cls_desc.resource_set.Get(scheduling::ResourceID::CPU()).Double() >
+            0) {
           total_cpu_running_tasks += entry.second.running_tasks.size();
         }
       }
@@ -812,13 +813,12 @@ void LocalTaskManager::PinTaskArgs(const TaskSpecification &spec,
   // TODO(swang): This should really be an assertion, but we can sometimes
   // receive a duplicate task request if there is a failure and the original
   // version of the task has not yet been canceled.
-  auto inserted = executing_task_args_.emplace(spec.TaskId(), deps).second;
-  if (inserted) {
+  auto executed_task_inserted = executing_task_args_.emplace(spec.TaskId(), deps).second;
+  if (executed_task_inserted) {
     for (size_t i = 0; i < deps.size(); i++) {
-      auto inserted =
+      auto [it, pinned_task_inserted] =
           pinned_task_arguments_.emplace(deps[i], std::make_pair(std::move(args[i]), 0));
-      auto it = inserted.first;
-      if (inserted.second) {
+      if (pinned_task_inserted) {
         // This is the first task that needed this argument.
         pinned_task_arguments_bytes_ += it->second.first->GetSize();
       }

src/ray/raylet/scheduling/cluster_task_manager.cc

@@ -253,8 +253,8 @@ void ClusterTaskManager::ScheduleAndDispatchTasks() {
     if (is_infeasible) {
       RAY_CHECK(!work_queue.empty());
       // Only announce the first item as infeasible.
-      auto &work_queue = shapes_it->second;
-      const auto &work = work_queue[0];
+      auto &cur_work_queue = shapes_it->second;
+      const auto &work = cur_work_queue[0];
       const RayTask task = work->task;
       if (announce_infeasible_task_) {
         announce_infeasible_task_(task);

src/ray/raylet/scheduling/policy/bundle_scheduling_policy.cc

@@ -306,11 +306,12 @@ SchedulingResult BundlePackSchedulingPolicy::Schedule(
   }
 
   // Releasing the resources temporarily deducted from `cluster_resource_manager_`.
-  for (size_t index = 0; index < result_nodes.size(); index++) {
+  for (size_t res_node_idx = 0; res_node_idx < result_nodes.size(); res_node_idx++) {
     // If `PackSchedule` fails, the id of some nodes may be nil.
-    if (!result_nodes[index].IsNil()) {
+    if (!result_nodes[res_node_idx].IsNil()) {
       RAY_CHECK(cluster_resource_manager_.AddNodeAvailableResources(
-          result_nodes[index], (*sorted_resource_request_list[index]).GetResourceSet()));
+          result_nodes[res_node_idx],
+          (*sorted_resource_request_list[res_node_idx]).GetResourceSet()));
     }
   }
 
@@ -361,10 +362,10 @@ SchedulingResult BundleSpreadSchedulingPolicy::Schedule(
       selected_nodes.emplace(best_node);
     } else {
       // Scheduling from selected nodes.
-      auto best_node = GetBestNode(*resource_request,
-                                   selected_nodes,
-                                   options,
-                                   available_cpus_before_bundle_scheduling);
+      best_node = GetBestNode(*resource_request,
+                              selected_nodes,
+                              options,
+                              available_cpus_before_bundle_scheduling);
       if (!best_node.first.IsNil()) {
         result_nodes.emplace_back(best_node.first);
         RAY_CHECK(cluster_resource_manager_.SubtractNodeAvailableResources(

src/ray/raylet/worker_pool.cc

@@ -1583,10 +1583,11 @@ void WorkerPool::DisconnectWorker(const std::shared_ptr<WorkerInterface> &worker
     return;
   }
 
-  for (auto it = idle_of_all_languages_.begin(); it != idle_of_all_languages_.end();
-       it++) {
-    if (it->worker == worker) {
-      idle_of_all_languages_.erase(it);
+  for (auto idle_worker_iter = idle_of_all_languages_.begin();
+       idle_worker_iter != idle_of_all_languages_.end();
+       idle_worker_iter++) {
+    if (idle_worker_iter->worker == worker) {
+      idle_of_all_languages_.erase(idle_worker_iter);
       break;
     }
   }