[core] Fix all gcs variable shadowing (#51704)

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

Author: dentiny

BUILD.bazel

@@ -3054,7 +3054,8 @@ pyx_library(
     # shared lib, see python/ray/__init__.py.
     cc_kwargs = dict(
         srcs = PYX_SRCS,
-        copts = COPTS + PYX_COPTS,
+        # cython code is auto-generated, which is out of our control.
+        copts = COPTS + PYX_COPTS + ["-Wno-shadow"],
         # see https://github.com/tensorflow/tensorflow/blob/r2.1/tensorflow/lite/BUILD#L444
         linkopts = select({
             "@platforms//os:osx": [

cpp/src/ray/runtime/task/task_executor.cc

@@ -335,8 +335,8 @@ void TaskExecutor::Invoke(
     }
   } catch (std::exception &e) {
     auto result = PackError(e.what());
-    auto data = std::make_shared<msgpack::sbuffer>(std::move(result));
-    runtime->Put(std::move(data), task_spec.ReturnId(0));
+    auto result_data = std::make_shared<msgpack::sbuffer>(std::move(result));
+    runtime->Put(std::move(result_data), task_spec.ReturnId(0));
   }
 }
 

src/ray/gcs/gcs_server/gcs_job_manager.cc

@@ -357,27 +357,27 @@ void GcsJobManager::HandleGetAllJobInfo(rpc::GetAllJobInfoRequest request,
           num_finished_tasks->fetch_add(job_count) + job_count;
       try_send_reply(updated_finished_tasks);
     } else {
-      for (int i = 0; i < reply->job_info_list_size(); i++) {
-        const auto &data = reply->job_info_list(i);
+      for (int jj = 0; jj < reply->job_info_list_size(); jj++) {
+        const auto &data = reply->job_info_list(jj);
         auto job_id = JobID::FromBinary(data.job_id());
         WorkerID worker_id = WorkerID::FromBinary(data.driver_address().worker_id());
 
         // If job is dead, no need to get.
         if (data.is_dead()) {
-          reply->mutable_job_info_list(i)->set_is_running_tasks(false);
+          reply->mutable_job_info_list(jj)->set_is_running_tasks(false);
           core_worker_clients_.Disconnect(worker_id);
           size_t updated_finished_tasks = num_finished_tasks->fetch_add(1) + 1;
           try_send_reply(updated_finished_tasks);
         } else {
           // Get is_running_tasks from the core worker for the driver.
           auto client = core_worker_clients_.GetOrConnect(data.driver_address());
-          auto request = std::make_unique<rpc::NumPendingTasksRequest>();
+          auto pending_task_req = std::make_unique<rpc::NumPendingTasksRequest>();
           constexpr int64_t kNumPendingTasksRequestTimeoutMs = 1000;
           RAY_LOG(DEBUG) << "Send NumPendingTasksRequest to worker " << worker_id
                          << ", timeout " << kNumPendingTasksRequestTimeoutMs << " ms.";
           client->NumPendingTasks(
-              std::move(request),
-              [job_id, worker_id, reply, i, num_finished_tasks, try_send_reply](
+              std::move(pending_task_req),
+              [job_id, worker_id, reply, jj, num_finished_tasks, try_send_reply](
                   const Status &status,
                   const rpc::NumPendingTasksReply &num_pending_tasks_reply) {
                 RAY_LOG(DEBUG).WithField(worker_id)
@@ -386,10 +386,11 @@ void GcsJobManager::HandleGetAllJobInfo(rpc::GetAllJobInfoRequest request,
                   RAY_LOG(WARNING).WithField(job_id).WithField(worker_id)
                       << "Failed to get num_pending_tasks from core worker: " << status
                       << ", is_running_tasks is unset.";
-                  reply->mutable_job_info_list(i)->clear_is_running_tasks();
+                  reply->mutable_job_info_list(jj)->clear_is_running_tasks();
                 } else {
                   bool is_running_tasks = num_pending_tasks_reply.num_pending_tasks() > 0;
-                  reply->mutable_job_info_list(i)->set_is_running_tasks(is_running_tasks);
+                  reply->mutable_job_info_list(jj)->set_is_running_tasks(
+                      is_running_tasks);
                 }
                 size_t updated_finished_tasks = num_finished_tasks->fetch_add(1) + 1;
                 try_send_reply(updated_finished_tasks);

src/ray/gcs/gcs_server/gcs_worker_manager.cc

@@ -268,10 +268,10 @@ void GcsWorkerManager::HandleUpdateWorkerDebuggerPort(
           auto worker_data = std::make_shared<rpc::WorkerTableData>();
           worker_data->CopyFrom(*result);
           worker_data->set_debugger_port(debugger_port);
-          Status status = gcs_table_storage_.WorkerTable().Put(
+          Status put_status = gcs_table_storage_.WorkerTable().Put(
               worker_id, *worker_data, {on_worker_update_done, io_context_});
-          if (!status.ok()) {
-            GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
+          if (!put_status.ok()) {
+            GCS_RPC_SEND_REPLY(send_reply_callback, reply, put_status);
           }
         }
       };
@@ -325,10 +325,10 @@ void GcsWorkerManager::HandleUpdateWorkerNumPausedThreads(
           worker_data->has_num_paused_threads() ? worker_data->num_paused_threads() : 0;
       worker_data->set_num_paused_threads(current_num_paused_threads +
                                           num_paused_threads_delta);
-      Status status = gcs_table_storage_.WorkerTable().Put(
+      Status put_status = gcs_table_storage_.WorkerTable().Put(
           worker_id, *worker_data, {on_worker_update_done, io_context_});
-      if (!status.ok()) {
-        GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
+      if (!put_status.ok()) {
+        GCS_RPC_SEND_REPLY(send_reply_callback, reply, put_status);
       }
     }
   };