[Fix][Core] Close unused pipe file descriptor of child processes of Raylet (#52700)

- We rely on a pipe-based mechanism to monitor the parent process
(raylet)'s health:

https://github.com/ray-project/ray/blob/07a00f20bb830c61ccf8fb2fddcfae8fa4c418dd/python/ray/_private/process_watcher.py#L127-L149
- Ideally, when raylet dies, the OS will close all its file descriptors.
With no process holding the write end of the pipe, so the `readline()`
in
https://github.com/ray-project/ray/blob/07a00f20bb830c61ccf8fb2fddcfae8fa4c418dd/python/ray/_private/process_watcher.py#L138
will return an empty string, indicating that the raylet has exited.
- However, this only works if *no other process* holds the pipe's write
end.
- Raylet launches several child processes via `fork + exec`. For
example, `DashboardAgent` and `RuntimeEnvAgent`:

https://github.com/ray-project/ray/blob/c2a6de384217e5db6f055d0607ca3e531deed56c/src/ray/raylet/node_manager.cc#L392-L393
  - These child processes are spawned using `ProcessFD::spawnvpe`:  

https://github.com/ray-project/ray/blob/a80f02f2c1c53d79eb94ecafaf59ae51f87d0734/src/ray/util/process.cc#L130
This function sets up a pipe and closes unused ends in both parent and
child:

https://github.com/ray-project/ray/blob/6ca0da17300e6087ae6bc9cb9faaa7e9a071b33e/src/ray/util/process.cc#L178-L226
- Raylet (parent process) creates `DashboardAgent` first, holding the
write end (`parent_lifetime_pipe[1]`, say fd 40), and closes the read
end.
- The dashboard agent (child process 1) holds the read end
(`parent_lifetime_pipe[0]`, say fd 39), and closes the write end.
- When raylet later forks `RuntimeEnvAgent` (child process 2), all open
fds (including fd 40) are inherited by the new child process.
- As a result, even if raylet dies, `RuntimeEnvAgent` still holds fd 40
(write end), preventing `DashboardAgent` from detecting raylet's death.
- Therefore, `readline()` in the dashboard agent hangs forever instead
of returning an empty string.

**Solution:** Ensure unused pipe write ends are closed on `exec` by
setting `FD_CLOEXEC`.

### Changes in this PR

- Revert the changes in #52388
because it is only a hotfix.
- Move `SetFdCloseOnExec` to `process.h` and `process.cc` and make it a
public util function.
- Call `SetFdCloseOnExec` in the parent process to make sure the pipe fd
is closed in subsequent `exec`.

### Testing

Tested manually. Before this PR, in about 20% of the runs the agent
process did not exit after the raylet died. After this PR, I ran it 20
times consecutively and the agent process successfully exited in all
runs.

---------

Signed-off-by: Chi-Sheng Liu <[email protected]>
Co-authored-by: Edward Oakes <[email protected]>

Author: MortalHappiness

python/ray/dashboard/modules/job/tests/test_sdk.py

@@ -197,9 +197,7 @@ def get_register_agents_number(gcs_client):
     ],
     indirect=True,
 )
-def test_job_head_choose_job_agent_E2E(
-    ray_start_cluster_head_with_env_vars, call_ray_stop_only
-):
+def test_job_head_choose_job_agent_E2E(ray_start_cluster_head_with_env_vars):
     cluster = ray_start_cluster_head_with_env_vars
     assert wait_until_server_available(cluster.webui_url) is True
     webui_url = cluster.webui_url

src/ray/common/client_connection.cc

@@ -30,6 +30,7 @@
 
 #include "ray/common/event_stats.h"
 #include "ray/common/ray_config.h"
+#include "ray/util/process.h"
 #include "ray/util/util.h"
 
 #if defined(_WIN32)
@@ -42,48 +43,16 @@
 
 namespace ray {
 
-namespace {
-
-#if defined(_WIN32)
-// Don't care what exact type is in windows... Looks like to be an asio specific type.
-template <typename NativeHandleType>
-void SetFdCloseOnExec(const NativeHandleType &handle) {
-  // In Windows we don't need to do anything, because in CreateProcess we pass
-  // bInheritHandles = false which means we don't inherit handles or sockets.
-  // https://github.com/ray-project/ray/blob/928183b3acab3c4ad73ef3001203a7aaf009bc87/src/ray/util/process.cc#L148
-  // https://learn.microsoft.com/en-us/windows/win32/sysinfo/handle-inheritance
-  return;
-}
-#else
-
-// Sets the flag FD_CLOEXEC to a file descriptor.
-// This means when the process is forked, this fd would be closed in the child process
-// side. Raylet forks to create core workers and we don't want the Unix Socket FDs to be
-// inherited by the core workers. Leaking these FDs would have performance implications.
-//
-// Idempotent. Calling twice == calling once.
-// Not thread safe.
-// See https://github.com/ray-project/ray/issues/40813
-void SetFdCloseOnExec(int fd) {
-  if (fd < 0) {
-    return;
-  }
-  int flags = fcntl(fd, F_GETFD, 0);
-  RAY_CHECK_NE(flags, -1) << "fcntl error: errno = " << errno << ", fd = " << fd;
-  const int ret = fcntl(fd, F_SETFD, flags | FD_CLOEXEC);
-  RAY_CHECK_NE(ret, -1) << "fcntl error: errno = " << errno << ", fd = " << fd;
-  RAY_LOG(DEBUG) << "set FD_CLOEXEC to fd " << fd;
-}
-#endif
-
-}  // namespace
-
 void SetCloseOnExec(local_stream_socket &socket) {
+#if !defined(_WIN32)
   SetFdCloseOnExec(socket.native_handle());
+#endif
 }
 
 void SetCloseOnExec(boost::asio::basic_socket_acceptor<local_stream_protocol> &acceptor) {
+#if !defined(_WIN32)
   SetFdCloseOnExec(acceptor.native_handle());
+#endif
 }
 
 Status ConnectSocketRetry(local_stream_socket &socket,

src/ray/util/process.cc

@@ -22,6 +22,7 @@
 #include <Winternl.h>
 #include <process.h>
 #else
+#include <fcntl.h>
 #include <poll.h>
 #include <signal.h>
 #include <stddef.h>
@@ -68,6 +69,19 @@ int execvpe(const char *program, char *const argv[], char *const envp[]) {
 
 namespace ray {
 
+#if !defined(_WIN32)
+void SetFdCloseOnExec(int fd) {
+  if (fd < 0) {
+    return;
+  }
+  int flags = fcntl(fd, F_GETFD, 0);
+  RAY_CHECK_NE(flags, -1) << "fcntl error: errno = " << errno << ", fd = " << fd;
+  const int ret = fcntl(fd, F_SETFD, flags | FD_CLOEXEC);
+  RAY_CHECK_NE(ret, -1) << "fcntl error: errno = " << errno << ", fd = " << fd;
+  RAY_LOG(DEBUG) << "set FD_CLOEXEC to fd " << fd;
+}
+#endif
+
 bool EnvironmentVariableLess::operator()(char a, char b) const {
   // TODO(mehrdadn): This is only used on Windows due to current lack of Unicode support.
   // It should be changed when Process adds Unicode support on Windows.
@@ -202,6 +216,8 @@ class ProcessFD {
     if (pid != 0 && pipefds[1] != -1) {
       close(pipefds[1]);  // not the child, so close the write end of the pipe
       pipefds[1] = -1;
+      // make sure the read end of the pipe is closed on exec
+      SetFdCloseOnExec(pipefds[0]);
     }
 
     // Create a pipe and redirect the read pipe to a child's stdin.
@@ -213,11 +229,14 @@ class ProcessFD {
         // Child. Close sthe write end of the pipe from child.
         close(parent_lifetime_pipe[1]);
         parent_lifetime_pipe[1] = -1;
+        SetFdCloseOnExec(parent_lifetime_pipe[0]);
       }
       if (pid != 0 && parent_lifetime_pipe[0] != -1) {
         // Parent. Close the read end of the pipe.
         close(parent_lifetime_pipe[0]);
         parent_lifetime_pipe[0] = -1;
+        // Make sure the write end of the pipe is closed on exec.
+        SetFdCloseOnExec(parent_lifetime_pipe[1]);
       }
     } else {
       // parent_lifetime_pipe pipes are not used.

src/ray/util/process.h

@@ -40,6 +40,17 @@ enum { PID_MAX_LIMIT = 1 << 22 };
 
 namespace ray {
 
+#if !defined(_WIN32)
+/// Sets the FD_CLOEXEC flag on a file descriptor.
+/// This means when the process is forked, this fd would be closed in the child process
+/// side.
+///
+/// Idempotent.
+/// Not thread safe.
+/// See https://github.com/ray-project/ray/issues/40813
+void SetFdCloseOnExec(int fd);
+#endif
+
 class EnvironmentVariableLess {
  public:
   bool operator()(char a, char b) const;