[core] Fix all variable shadowing for core worker (#51672)

This PR fixes all variable shadowing for core worker.

Variable shadowing is known to be error prune --- I personally just
spent 20 minute debugging an issue caused by it.
Need to address all issues before
#51669

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

Author: dentiny

src/ray/common/bundle_location_index.cc

@@ -30,12 +30,12 @@ void BundleLocationIndex::AddBundleLocations(
   }
 
   // Update `node_to_leased_bundles_`.
-  for (auto iter : *bundle_locations) {
-    const auto &node_id = iter.second.first;
+  for (auto bundle_iter : *bundle_locations) {
+    const auto &node_id = bundle_iter.second.first;
     if (!node_to_leased_bundles_.contains(node_id)) {
       node_to_leased_bundles_[node_id] = std::make_shared<BundleLocations>();
     }
-    node_to_leased_bundles_[node_id]->emplace(iter.first, iter.second);
+    node_to_leased_bundles_[node_id]->emplace(bundle_iter.first, bundle_iter.second);
   }
 }
 

src/ray/common/ray_config.h

@@ -51,8 +51,8 @@ template <>
 inline std::vector<std::string> ConvertValue<std::vector<std::string>>(
     const std::string &type_string, const std::string &value) {
   std::vector<std::string> split = absl::StrSplit(value, ',');
-  for (auto &value : split) {
-    boost::algorithm::trim(value);
+  for (auto &part : split) {
+    boost::algorithm::trim(part);
   }
   return split;
 }

src/ray/common/scheduling/cluster_resource_data.cc

@@ -63,21 +63,19 @@ NodeResources ResourceMapToNodeResources(
 float NodeResources::CalculateCriticalResourceUtilization() const {
   float highest = 0;
   for (const auto &i : {CPU, MEM, OBJECT_STORE_MEM}) {
-    const auto &total = this->total.Get(ResourceID(i));
-    if (total == 0) {
+    const auto &cur_total = this->total.Get(ResourceID(i));
+    if (cur_total == 0) {
       continue;
     }
-    auto available = this->available.Get(ResourceID(i)).Double();
+    auto cur_available = this->available.Get(ResourceID(i)).Double();
     // Gcs scheduler handles the `normal_task_resources` specifically. So when calculating
     // the available resources, we have to take one more step to take that into account.
     // For raylet scheduling, the `normal_task_resources` is always empty.
     if (this->normal_task_resources.Has(ResourceID(i))) {
-      available -= this->normal_task_resources.Get(ResourceID(i)).Double();
-      if (available < 0) {
-        available = 0;
-      }
+      cur_available -= this->normal_task_resources.Get(ResourceID(i)).Double();
+      cur_available = std::max<float>(0, cur_available);
     }
-    float utilization = 1 - (available / total.Double());
+    float utilization = 1 - (cur_available / cur_total.Double());
     if (utilization > highest) {
       highest = utilization;
     }

src/ray/common/scheduling/cluster_resource_data.h

@@ -266,9 +266,9 @@ class TaskResourceInstances {
         buffer << resource[0];
       } else {
         buffer << "[";
-        for (size_t i = 0; i < resource.size(); i++) {
-          buffer << resource[i];
-          if (i < resource.size() - 1) {
+        for (size_t j = 0; j < resource.size(); j++) {
+          buffer << resource[j];
+          if (j < resource.size() - 1) {
             buffer << ", ";
           }
         }

src/ray/core_worker/actor_manager.cc

@@ -263,7 +263,7 @@ std::vector<ObjectID> ActorManager::GetActorHandleIDsFromHandles() {
 
 ActorID ActorManager::GetCachedNamedActorID(const std::string &actor_name) {
   {
-    absl::MutexLock lock(&cache_mutex_);
+    absl::MutexLock cache_lock(&cache_mutex_);
     auto it = cached_actor_name_to_ids_.find(actor_name);
     if (it != cached_actor_name_to_ids_.end()) {
       absl::MutexLock lock(&mutex_);

src/ray/core_worker/transport/normal_task_submitter.cc

@@ -611,9 +611,10 @@ void NormalTaskSubmitter::PushNormalTask(
                                             get_task_failure_cause_reply_status,
                                             get_task_failure_cause_reply);
                 };
-            auto &lease_entry = worker_to_lease_entry_[addr];
-            RAY_CHECK(lease_entry.lease_client);
-            lease_entry.lease_client->GetTaskFailureCause(lease_entry.task_id, callback);
+            auto &cur_lease_entry = worker_to_lease_entry_[addr];
+            RAY_CHECK(cur_lease_entry.lease_client);
+            cur_lease_entry.lease_client->GetTaskFailureCause(cur_lease_entry.task_id,
+                                                              callback);
           }
 
           if (!status.ok() || !is_actor_creation || reply.worker_exiting()) {

src/ray/gcs/pubsub/gcs_pub_sub.cc

@@ -358,10 +358,12 @@ Status PythonGcsSubscriber::DoPoll(int64_t timeout_ms, rpc::PubMessage *message)
     if (status.error_code() == grpc::StatusCode::DEADLINE_EXCEEDED ||
         status.error_code() == grpc::StatusCode::UNAVAILABLE) {
       return Status::OK();
-    } else if (status.error_code() == grpc::StatusCode::CANCELLED) {
+    }
+    if (status.error_code() == grpc::StatusCode::CANCELLED) {
       // This channel was shut down via Close()
       return Status::OK();
-    } else if (status.error_code() != grpc::StatusCode::OK) {
+    }
+    if (status.error_code() != grpc::StatusCode::OK) {
       return Status::Invalid(status.error_message());
     }
 
@@ -376,18 +378,18 @@ Status PythonGcsSubscriber::DoPoll(int64_t timeout_ms, rpc::PubMessage *message)
       max_processed_sequence_id_ = 0;
     }
     last_batch_size_ = reply.pub_messages().size();
-    for (auto &message : reply.pub_messages()) {
-      if (message.sequence_id() <= max_processed_sequence_id_) {
-        RAY_LOG(WARNING) << "Ignoring out of order message " << message.sequence_id();
+    for (auto &cur_pub_msg : reply.pub_messages()) {
+      if (cur_pub_msg.sequence_id() <= max_processed_sequence_id_) {
+        RAY_LOG(WARNING) << "Ignoring out of order message " << cur_pub_msg.sequence_id();
         continue;
       }
-      max_processed_sequence_id_ = message.sequence_id();
-      if (message.channel_type() != channel_type_) {
+      max_processed_sequence_id_ = cur_pub_msg.sequence_id();
+      if (cur_pub_msg.channel_type() != channel_type_) {
         RAY_LOG(WARNING) << "Ignoring message from unsubscribed channel "
-                         << message.channel_type();
+                         << cur_pub_msg.channel_type();
         continue;
       }
-      queue_.emplace_back(std::move(message));
+      queue_.emplace_back(std::move(cur_pub_msg));
     }
   }