import { spawn, spawnSync } from "node:child_process"; import type { ChildProcess } from "node:child_process"; import { createWriteStream } from "node:fs"; import type { WriteStream } from "node:fs"; import { createInterface } from "node:readline"; import { setTimeout as delayTimeout } from "node:timers/promises"; import type { Config } from "../../../config/env"; import { delay } from "../../../core/async"; import { cleanupLogFiles, getLogCleanupDefaultsFromEnvironment, primaryLogPathFor, } from "../../../core/log-files"; import type { Logger } from "../../../core/logger"; import type { LaunchResult, ProcessInfo, Recipe } from "../../models/types"; import type { EventManager } from "../../system/event-manager"; import { buildBackendCommand } from "./backend-builder"; import { buildEnvironment, collectChildren, detectBackend, extractFlag, listProcesses, pidExists, buildProcessTree, } from "./process-utilities"; export interface ProcessManager { findInferenceProcess: (port: number) => Promise; launchModel: (recipe: Recipe) => Promise; evictModel: (force: boolean) => Promise; killProcess: (pid: number, force: boolean) => Promise; } export const createProcessManager = ( config: Config, logger: Logger, eventManager?: EventManager ): ProcessManager => { type ProcessTableEntry = { pid: number; ppid: number; stat: string; command: string; }; const findInferenceProcess = async (port: number): Promise => { const processes = listProcesses(); for (const proc of processes) { const backend = detectBackend(proc.args); if (!backend) { continue; } const flagPort = extractFlag(proc.args, "--port"); if (flagPort && Number(flagPort) !== port) { continue; } else if (!flagPort && !(backend === "vllm" && port === 8000)) { continue; } let modelPath = extractFlag(proc.args, "--model") || extractFlag(proc.args, "--model-path"); if (!modelPath && backend === "llamacpp") { modelPath = extractFlag(proc.args, "-m"); } if (!modelPath && backend === "sglang") { const launchServerIndex = proc.args.findIndex( (argument) => argument === "sglang.launch_server" ); const candidate = launchServerIndex >= 0 ? proc.args[launchServerIndex + 1] : undefined; if (candidate && !candidate.startsWith("-")) { modelPath = candidate; } } const servedModelName = extractFlag(proc.args, "--served-model-name") || extractFlag(proc.args, "--alias") || extractFlag(proc.args, "-a"); if (!modelPath) { const serveIndex = proc.args.indexOf("serve"); if (serveIndex >= 0 && serveIndex + 1 < proc.args.length) { const candidate = proc.args[serveIndex + 1]; if (candidate && !candidate.startsWith("-")) { modelPath = candidate; } } } return { pid: proc.pid, backend, model_path: modelPath ?? null, port, served_model_name: servedModelName ?? null, }; } return null; }; const killProcess = async (pid: number, force: boolean): Promise => { if (!pidExists(pid)) { return true; } const tree = buildProcessTree(); const children = new Set(); collectChildren(tree, pid, children); const allPids = [...children, pid]; // Docker-backed recipes often leave the actual server inside a container whose // host process tree does not reliably die when the docker CLI process is // signalled. Stop/kill the named container first, then signal the process tree. stopDockerContainersForProcesses(allPids, force); const signal = force ? "SIGKILL" : "SIGTERM"; for (const childPid of allPids) { sendSignal(childPid, signal); } const deadline = Date.now() + (force ? 15_000 : 10_000); while (Date.now() < deadline) { if (!pidExists(pid)) { break; } await delayTimeout(250); } if (pidExists(pid)) { stopDockerContainersForProcesses(allPids, true); if (!sendSignal(pid, "SIGKILL")) { return false; } const finalDeadline = Date.now() + 5_000; while (Date.now() < finalDeadline) { if (!pidExists(pid)) { break; } await delayTimeout(250); } } await delay(force ? 500 : 1000); return !pidExists(pid); }; const stopDockerContainersForProcesses = (pids: number[], force: boolean): void => { const pidSet = new Set(pids); const names = new Set(); const inferencePorts = new Set(); const processes = listProcesses(); for (const proc of processes) { if (!pidSet.has(proc.pid)) continue; const port = Number(extractFlag(proc.args, "--port")); if (Number.isFinite(port) && port > 0) inferencePorts.add(port); const dockerIndex = proc.args.findIndex( (argument) => argument === "docker" || argument.endsWith("/docker") ); if (dockerIndex < 0 || proc.args[dockerIndex + 1] !== "run") continue; const name = extractFlag(proc.args.slice(dockerIndex + 2), "--name"); if (name) names.add(name); } // With Docker + host process visibility, the Python server process is often // parented under containerd-shim rather than the `docker run` CLI process. If // `findInferenceProcess()` found the in-container Python PID, match the // sibling docker-run command by inference port so the container is stopped too. if (inferencePorts.size > 0) { for (const proc of processes) { const dockerIndex = proc.args.findIndex( (argument) => argument === "docker" || argument.endsWith("/docker") ); if (dockerIndex < 0 || proc.args[dockerIndex + 1] !== "run") continue; const dockerPort = Number(extractFlag(proc.args, "--port")); if (!inferencePorts.has(dockerPort)) continue; const name = extractFlag(proc.args.slice(dockerIndex + 2), "--name"); if (name) names.add(name); } } for (const name of names) { const action = force ? "kill" : "stop"; const args = force ? [action, name] : [action, "--time", "2", name]; let result = spawnSync("docker", args, { stdio: "ignore" }); if (result.status !== 0) { result = spawnSync("sudo", ["-n", "docker", ...args], { stdio: "ignore" }); } if (result.status !== 0) { logger.warn("Failed to stop docker inference container", { name, action }); } } }; const sendSignal = (pid: number, signal: NodeJS.Signals): boolean => { try { process.kill(pid, signal); return true; } catch { const result = spawnSync("sudo", ["-n", "kill", `-${signal}`, String(pid)], { stdio: "ignore", }); return result.status === 0; } }; const listProcessTable = (): ProcessTableEntry[] => { try { const result = spawnSync("ps", ["-eo", "pid=,ppid=,stat=,args="]); if (result.status !== 0) { return []; } const output = result.stdout.toString("utf-8").trim(); if (!output) { return []; } return output .split("\n") .map((line): ProcessTableEntry | null => { const match = line.trim().match(/^(\d+)\s+(\d+)\s+(\S+)\s+(.*)$/); if (!match) { return null; } const command = match[4] ?? ""; return { pid: Number(match[1]), ppid: Number(match[2]), stat: match[3] ?? "", command, }; }) .filter((entry): entry is ProcessTableEntry => entry !== null && entry.pid > 0); } catch { return []; } }; const isOrphanedInferenceWorker = (entry: ProcessTableEntry): boolean => { if (entry.ppid !== 1 || entry.stat.includes("Z")) { return false; } return entry.command.includes("VLLM::Worker"); }; const cleanupOrphanedInferenceWorkers = async (reason: string): Promise => { const workers = listProcessTable().filter(isOrphanedInferenceWorker); if (workers.length === 0) { return 0; } for (const worker of workers) { logger.warn("Killing orphaned inference worker", { pid: worker.pid, reason, command: worker.command, }); sendSignal(worker.pid, "SIGTERM"); } const deadline = Date.now() + 2_000; while (Date.now() < deadline && workers.some((worker) => pidExists(worker.pid))) { await delayTimeout(200); } for (const worker of workers) { if (pidExists(worker.pid)) { logger.warn("Force killing orphaned inference worker", { pid: worker.pid, reason, command: worker.command, }); sendSignal(worker.pid, "SIGKILL"); } } return workers.length; }; const launchModel = async (recipe: Recipe): Promise => { const updatedRecipe: Recipe = { ...recipe, port: config.inference_port, }; let command: string[] | null = null; try { command = buildBackendCommand(updatedRecipe, config); } catch (error) { const message = error instanceof Error ? error.message : String(error); return { success: false, pid: null, message, log_file: primaryLogPathFor(config.data_dir, updatedRecipe.id), }; } if (!command) { return { success: false, pid: null, message: "Invalid launch command", log_file: primaryLogPathFor(config.data_dir, updatedRecipe.id), }; } await cleanupOrphanedInferenceWorkers("before-launch"); const logFile = primaryLogPathFor(config.data_dir, updatedRecipe.id); // Best-effort retention to prevent unbounded growth over long-running installs. cleanupLogFiles(config.data_dir, { ...getLogCleanupDefaultsFromEnvironment(), excludePaths: new Set([logFile]), }); const env = buildEnvironment(updatedRecipe); try { const entry = command[0]; if (!entry) { return { success: false, pid: null, message: "Invalid launch command", log_file: logFile, }; } let spawnError: string | null = null; const child = spawn(entry, command.slice(1), { stdio: ["ignore", "pipe", "pipe"], env, detached: true, }) as ChildProcess; child.on("error", (error) => { spawnError = String(error); }); let logStream: WriteStream | null = null; try { logStream = createWriteStream(logFile, { flags: "a" }); } catch (logError) { logger.warn("Failed to open log file", { error: String(logError), }); } if (child.stdout) { const rl = createInterface({ input: child.stdout, crlfDelay: Infinity, }); rl.on("line", (line) => { if (logStream) { logStream.write(line + "\n"); } if (eventManager) { eventManager.publishLogLine(updatedRecipe.id, line).catch(() => {}); } }); } if (child.stderr) { const rl = createInterface({ input: child.stderr, crlfDelay: Infinity, }); rl.on("line", (line) => { if (logStream) { logStream.write(line + "\n"); } if (eventManager) { eventManager.publishLogLine(updatedRecipe.id, line).catch(() => {}); } }); } child.on("exit", () => { if (logStream) { logStream.end(); } }); child.unref(); await delay(3000); if (spawnError) { if (logStream) { logStream.end(); } return { success: false, pid: null, message: spawnError, log_file: logFile, }; } if (child.exitCode !== null) { if (logStream) { logStream.end(); } return { success: false, pid: null, message: "Process exited early", log_file: logFile, }; } return { success: true, pid: child.pid ?? null, message: "Process started", log_file: logFile, }; } catch (error) { logger.error("Launch failed", { error: String(error) }); return { success: false, pid: null, message: String(error), log_file: logFile, }; } }; const evictModel = async (force: boolean): Promise => { const current = await findInferenceProcess(config.inference_port); if (!current) { await cleanupOrphanedInferenceWorkers("evict-without-active-process"); return null; } await killProcess(current.pid, force); await cleanupOrphanedInferenceWorkers("after-evict"); return current.pid; }; return { findInferenceProcess, launchModel, evictModel, killProcess, }; };