[core] add get pid rpc to node manager

Signed-off-by: tianyi-ge <tianyig@outlook.com>

Author: tianyi-ge

python/ray/dashboard/modules/reporter/reporter_agent.py

@@ -896,18 +896,19 @@ def _get_disk_io_stats():
                 stats.write_count,
             )
 
-    def _get_worker_pids_from_raylet(self):
+    async def _get_worker_pids_from_raylet(self):
         channel = ray._private.utils.init_grpc_channel(
             self._node_manager_address, GLOBAL_GRPC_OPTIONS, asynchronous=True
         )
         timeout = NODE_MANAGER_RPC_TIMEOUT_SECONDS
         stub = node_manager_pb2_grpc.NodeManagerServiceStub(channel)
         try:
-            reply = await stub.GetDriverAndWorkers(
-                node_manager_pb2.GetDriverAndWorkersRequest(), timeout=timeout
+            reply = await stub.GetDriverAndWorkerPids(
+                node_manager_pb2.GetDriverAndWorkerPidsRequest(), timeout=timeout
             )
             return reply.pids
-        except Exception:
+        except Exception as e:
+            logger.debug(f"Failed to get worker pids from raylet via gRPC: {e}")
             return None
 
     def _get_agent_proc(self) -> psutil.Process:
@@ -919,7 +920,7 @@ def _generate_worker_key(self, proc: psutil.Process) -> Tuple[int, float]:
         return (proc.pid, proc.create_time())
 
     def _get_worker_processes(self):
-        pids = self._get_worker_pids_from_raylet()
+        pids = asyncio.run(self._get_worker_pids_from_raylet())
         if pids is not None:
             workers = {}
             for pid in pids:

src/ray/raylet/node_manager.cc

@@ -2728,14 +2728,10 @@ void NodeManager::HandleGetDriverAndWorkerPids(
     rpc::GetDriverAndWorkerPidsReply *reply,
     rpc::SendReplyCallback send_reply_callback) {
   auto all_workers = worker_pool_.GetAllRegisteredWorkers(/* filter_dead_worker */ true);
-  for (const auto &driver :
-       worker_pool_.GetAllRegisteredDrivers(/* filter_dead_driver */ true)) {
-    all_workers.push_back(driver);
-  }
+  const auto &drivers =
+      worker_pool_.GetAllRegisteredDrivers(/* filter_dead_driver */ true);
+  all_workers.insert(all_workers.end(), drivers.begin(), drivers.end());
   for (const auto &worker : all_workers) {
-    if (worker->IsDead()) {
-      continue;
-    }
     reply->add_pids(worker->GetProcess().GetId());
   }
   send_reply_callback(Status::OK(), nullptr, nullptr);