[chiayi]

Description

Adds the standalone Python worker fork server process and unit tests to support forking workers from a warm fork server process instead of starting a new Python interpreter from scratch.

Key Changes

• Manage fork server process lifecycle and its Unix domain socket
• Pre-import Ray worker dependencies and configured preload_modules in a single-threaded process, freeze the GC heap for copy-on-write sharing, and fork workers on demand over a Unix socket
• Reset post-fork child state before initializing the worker
• Add unit tests for fork server process

Related issues

#67446 - Python worker fork server

Additional information

Wait for #67356

[gemini-code-assist]

Code Review

This pull request introduces a pre-initialized Python fork server and a cgroup bootstrap helper to manage worker processes in Ray, along with refactoring thread pool resizing logic and adding comprehensive unit tests. The review feedback highlights several key improvement opportunities: addressing a potential socket permission race condition by using os.umask, explicitly importing ray._private.worker to avoid fragile transitive imports, ensuring custom PYTHONPATH entries are correctly prioritized in sys.path, sending error responses back to clients when fork requests fail, and simplifying file operations in _cgroup_bootstrap.py to prevent potential file descriptor leaks.

[cursor]

Stale Bugbot comment from a previous run.

[cursor]

Stale Bugbot comment from a previous run.

[cursor]

Runtime env setup can replace entrypoint

Medium Severity

When a RuntimeEnvContext is supplied, the server is launched through setup_worker.py → exec_worker(). That path rewrites passthrough_args[0] via override_worker_entrypoint and can exec a container launcher, so fork_server.py never runs.

 

^{Reviewed by Cursor Bugbot for commit 4faffa0. Configure here.}

[chiayi]

We will have a followup PR that launches the fork server that will skip runtime envs that override the worker entrypoint or use containers, same as the prototype.

[cursor]

Stale Bugbot comment from a previous run.

[ryanaoleary]

SIGCHLD is set to ignore above, so crashed workers get reaped immediately and Process::Kill can hit a recycled PID on registration timeout. Should we add the SCM_RIGHTS lifetime pipe here per the fork server tracking issue so the socket protocol is settled before the raylet WorkerPool dispatch changes?

life_r, life_w = os.pipe()
pid = os.fork()
if pid > 0:
    os.close(life_w)
    socket.send_fds(conn, [f'{{"pid":{pid}}}\n'.encode("ascii")], [life_r])
    os.close(life_r)
else:
    os.close(life_r)
    # close life_w in grandchild forks via os.register_at_fork(after_in_child=...)

[chiayi]

Since passing a lifetime pipe fd changes the socket wire protocol between the fork server and the C++ WorkerPool client (recvmsg), I think it might be cleaner and make more sense to include both sides of the SCM_RIGHTS pipe handling together in the WorkerPool/Process integration PR as well so we can review and test the full lifecycle end-to-end. WDYT?

[ryanaoleary]

When RuntimeEnvAgent redirects its stdout and stderr to log-rotation pipes in main.py#L163, Popen inherits those pipes. That mixes preload and worker output into the agent logs, and forked workers keep a copy of the pipe open inside ScopedDup2Wrapper, hanging the agent on clean exit while any worker is alive.

Should we add optional log directory and stdio fd args here, switching to worker stdio or raylet log files after os.replace() so startup errors still log, and wire them into RuntimeEnvAgent in the follow-up?

ForkServerProcess(
    ...,
    logs_dir=None,         # fallback post-preload target: raylet.{out,err}
    setup_log_path=None,   # pre-fork log, default: logs_dir/fork_server_default.log
    worker_stdout_fd=None, # container stdout when RAY_LOG_TO_STDERR=1
    worker_stderr_fd=None, # container stderr when RAY_LOG_TO_STDERR=1
)

[chiayi]

Thanks for pointing this out! Sooo I'm inclined to add the logs_dir, setup_log_path, and the worker stdio fd parameters alongside the RuntimeEnvAgent plumbing all in the integration PR, that way we can review the agent/main.py fd capture and ForkServerProcess redirection together in one PR. WDYT?

[ryanaoleary]

PR_SET_PDEATHSIG fires when the spawning thread exits, killing the server and stranding its socket if called from a thread pool. Blocking __init__ for 120s on the agent event loop also trips NodeManager's 15s WaitForRuntimeEnvAgentPort timeout.

Should we switch preexec_fn to a polled stdin lifeline pipe with signal.set_wakeup_fd so SIGTERM does not fire inside os.fork(), and separate spawn from readiness polling?

server = ForkServerProcess(..., monitor_stdin=True)
server.wait_ready(timeout_s=startup_timeout_s)

[chiayi]

Done, thanks! Switched to a --monitor-stdin lifeline pipe and added wait_ready().

[ryanaoleary]

Reading the request in the parent leaves per-job env vars in the parent's heap, shared copy-on-write with later workers, and lets a slow client stall the accept loop for 2s. Read it in the child instead after checking for closed clients:

# Use POLLHUP | POLLERR (not POLLRDHUP, which fires on shutdown(SHUT_WR)).
poller = select.poll()
poller.register(conn, select.POLLHUP | select.POLLERR)
if poller.poll(0):
    continue
pid = os.fork()
# parent sends (pid, lifetime_fd) and closes conn; child reads JSON from conn

Malformed payloads fail in the child rather than returning a JSON error, and we no longer rely on wall-clock deadline_ms.

[chiayi]

Moving the request read into the child means the parent replies with the child PID before the payload is validated, so error reporting has to rely on the child's lifetime pipe closing rather than a socket error reply. I think it makes sense to pair this with the SCM_RIGHTS pipe change in the WorkerPool integration PR?

[ryanaoleary]

TORCH_DEVICE_BACKEND_AUTOLOAD=0 skips device-backend registration on import, so TorchTPU, torch_npu, and Intel XPU won't auto-register in children. Cache the entry points pre-fork to avoid a ~26ms scan per worker and load them at the end of set_visible_accelerator_ids():

# pre-fork:
_CACHED_TORCH_BACKEND_EPS = list(importlib.metadata.entry_points(group="torch.backends"))

# end of set_visible_accelerator_ids():
# if torch is imported and TORCH_DEVICE_BACKEND_AUTOLOAD != "0", load cached eps
# (fall back to torch._import_device_backends() if PYTHONPATH changed)

[chiayi]

Cached the torch.backends entry points before fork() and loaded them once the child worker's visible accelerator devices are configured

[cursor]

Stale Bugbot comment from a previous run.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

There are 2 total unresolved issues (including 1 from previous review).

^{Reviewed by Cursor Bugbot for commit 944c2ab. Configure here.}

[chiayi]

@ryanaoleary Thanks for reviewing the PR and just to summarize the comments that were addressed. These are the 2 PRs that are in mentioned as followups:

1. WorkerPool & Process C++ integrations: connects workerpool to fork server socket. This will also include the SCM_RIGHTS changes and reading the request in the child so both side of the socket protocol can be together.
2. RuntimeEnvAgent Integrations: This will wire the RuntimEnvAgent to manage the default and individual runtime_env fork servers.

Let me know if this sounds good! Thank you!