nemo_curator.core.serve.subprocess_mgr

Track a detached Ray actor and the subprocess it owns.

The instance methods hide the ray.get(self.actor.X.remote()) boilerplate callers would otherwise repeat for every lifecycle operation. Teardown is parallelised via stop_many so a whole replica (or the infra trio) can be reaped with one round-trip.

nemo_curator.core.serve.subprocess_mgr.ManagedSubprocess.is_alive() -> bool

nemo_curator.core.serve.subprocess_mgr.ManagedSubprocess.pid() -> int

nemo_curator.core.serve.subprocess_mgr.ManagedSubprocess.read_log_tail(
    num_bytes: int = 8192
) -> str

nemo_curator.core.serve.subprocess_mgr.ManagedSubprocess.spawn(
    label: str,
    pg: ray.util.placement_group.PlacementGroup,
    bundle_index: int,
    num_gpus: int,
    command: list[str] | None = None,
    python_args: list[str] | None = None,
    runtime_dir: str | None = None,
    actor_name_prefix: str = '',
    subprocess_env: dict[str, str] | None = None,
    runtime_env: dict[str, typing.Any] | None = None
) -> nemo_curator.core.serve.subprocess_mgr.ManagedSubprocess

nemo_curator.core.serve.subprocess_mgr.ManagedSubprocess.stop(
    timeout_s: float | None = None
) -> None

nemo_curator.core.serve.subprocess_mgr.ManagedSubprocess.stop_many(
    procs: list[nemo_curator.core.serve.subprocess_mgr.ManagedSubprocess],
    timeout_s: float | None = None
) -> None

nemo_curator.core.serve.subprocess_mgr.ManagedSubprocess.wait(
    timeout: float | None = None
) -> int

Bases: RuntimeError

Raised when a managed subprocess fails during inference server lifecycle.

Carries an optional debug_context dict so callers can surface subprocess logs, PIDs, or any other diagnostic state alongside the user-facing message.

Return True iff the actor-side stop() completed successfully.

Raise if name is not found on $PATH.

Return a Ray remote actor class named actor_type.

Each call produces a class whose __name__ and __qualname__ are set to actor_type so the Ray dashboard shows descriptive labels (e.g. Dynamo_ETCD, Dynamo_DP0_Qwen3-0.6B).

Lifecycle:

1. initialize Popens the subprocess with start_new_session=True so the whole subtree shares one process group (pgid == popen.pid).
2. run blocks until the subprocess exits; the returned ObjectRef resolves on exit so ray.wait can be used for liveness.
3. stop / force_sigkill_subprocess reap that process group.

Teardown is driver-initiated via graceful_stop_actors so the actor can killpg its subprocess group before ray.kill hard-kills the actor process itself — otherwise the Popen-launched child reparents to init and keeps running. A Python atexit hook is registered as a backup for clean actor exits (e.g. a normal raylet shutdown); it does not fire on ray.kill / PG removal / SIGKILL.

Assemble the Ray detached-actor name for a given prefix and label.

Both ManagedSubprocess.spawn (when naming a new actor) and reacquire_detached_actor_handles (when re-looking it up in a later session) must agree on this formula, so it lives in one place.

Best-effort dispatch of actor.stop.remote() for one actor.

SIGTERM the whole process group, wait briefly, then SIGKILL.

Because every managed subprocess is launched with start_new_session=True, pgid == pid for the launcher and every descendant that did not explicitly call setsid() remains in that group. The group persists while any member is alive, so this still works even if the launcher itself already exited.

Reap the subprocess’s process group and return its exit code.

Assumes proc was launched with start_new_session=True, so proc.pid is the process-group id to target.

Block until a TCP connection to host:port succeeds.

Stop many detached actors in parallel.

Prefer ManagedSubprocess.stop / ManagedSubprocess.stop_many. This primitive is for the raw-actor case (e.g. the reconnecting driver looking up orphaned actors by name).

Escalation order per actor:

1. actor.stop.remote() — actor-driven SIGTERM on the subprocess group with a bounded wait, then SIGKILL if needed.
2. If stop did not drain in time, actor.force_sigkill_subprocess.remote() — host-side SIGKILL on the subprocess group. This runs on the actor’s node (because actor methods always do), so it works multi-node without any driver-side connectivity assumption.
3. ray.kill(actor) — tear down the actor itself.

Step 2 matters because ray.kill SIGKILLs the actor process; its Popen child (the launcher) reparents to init and keeps running, and that launcher’s own children (e.g. vLLM EngineCore) stay alive too. Sending killpg first ensures the whole subprocess tree is gone before the actor itself is torn down. Each dispatch is guarded individually so a single invalid handle (e.g. a stale cross-ray.init()-session handle) does not abort the batch.

Refresh each proc.actor in place by re-looking up its detached actor.

ActorHandle objects are pinned to the Ray job that created them, so any handle stored across a with ray.init() boundary raises ActorHandleNotFoundError on .remote(). Backends that use the start/stop attach-detach pattern must re-acquire their handles inside the stop-side with ray.init() before calling any actor method.

Actor names follow the ManagedSubprocess.spawn convention: f"{actor_name_prefix}_{label}", or just label when the prefix is empty.

Returns a filtered list containing only procs whose actor was successfully re-acquired; entries whose actor is already gone are dropped so callers can safely pass the result to ManagedSubprocess.stop_many.

Graceful-stop any ALIVE detached actor whose name starts with prefix.

Intended for orphan cleanup after a driver restart: detached actors outlive the driver that created them, so a reconnecting driver (in the same namespace) can list them by name, re-acquire handles, and reap them + their subprocess trees via graceful_stop_actors. Returns the number of orphans reaped. Mirrors remove_named_pgs_with_prefix in placement.