Current section

Files

Jump to
snakepit lib snakepit process_killer.ex
Raw

lib/snakepit/process_killer.ex

defmodule Snakepit.ProcessKiller do
@moduledoc """
Robust OS process management using Erlang primitives.
No shell commands, pure Erlang/Elixir.
This module provides POSIX-compliant process management that works
across Linux, macOS, and BSD systems without relying on shell-specific
features like pkill.
"""
alias Snakepit.Logger, as: SLog
@kill_command_candidates ["/bin/kill", "/usr/bin/kill"]
@ps_command_candidates ["/bin/ps", "/usr/bin/ps"]
@log_category :shutdown
@doc """
Returns true if the platform supports process group kill semantics.
"""
def process_group_supported? do
case :os.type() do
{:unix, _} ->
true
_ ->
false
end
end
@doc """
Returns the path to the setsid executable, or {:error, :not_found}.
"""
def setsid_executable do
case System.find_executable("setsid") do
nil -> {:error, :not_found}
path -> {:ok, path}
end
end
@doc """
Returns the setsid executable path or raises if not available.
"""
def setsid_executable! do
case setsid_executable() do
{:ok, path} -> path
{:error, _} -> raise "setsid executable not found"
end
end
@doc """
Kills a process by PID using proper Erlang signals.
## Parameters
- `os_pid`: OS process ID (integer)
- `signal`: :sigterm | :sigkill | :sighup
## Returns
- `:ok` if kill succeeded
- `{:error, reason}` if kill failed
"""
def kill_process(os_pid, signal \\ :sigterm) when is_integer(os_pid) do
signal_num = signal_to_number(signal)
# DEBUG: Log all kills to find who's killing workers during startup
caller = Process.info(self(), :registered_name)
SLog.debug(
@log_category,
"ProcessKiller.kill_process: PID=#{os_pid}, signal=#{signal}, caller=#{inspect(caller)}"
)
with {:ok, kill_path} <- require_executable("kill", @kill_command_candidates),
{:ok, output, code} <-
run_command(kill_path, ["-#{signal_num}", Integer.to_string(os_pid)]) do
trimmed = String.trim(output || "")
cond do
code == 0 ->
:ok
String.contains?(trimmed, "No such process") ->
:ok
true ->
{:error, if(trimmed == "", do: {:exit_status, code}, else: trimmed)}
end
else
{:error, reason} ->
SLog.warning(@log_category, "Failed to execute kill command: #{inspect(reason)}")
{:error, reason}
end
end
@doc """
Kills a process group by PGID using proper Erlang signals.
## Parameters
- `pgid`: Process group ID (integer)
- `signal`: :sigterm | :sigkill | :sighup
"""
def kill_process_group(pgid, signal \\ :sigterm) when is_integer(pgid) do
signal_num = signal_to_number(signal)
with {:ok, kill_path} <- require_executable("kill", @kill_command_candidates),
{:ok, output, code} <-
run_command(kill_path, ["-#{signal_num}", "--", "-#{pgid}"]) do
trimmed = String.trim(output || "")
cond do
code == 0 ->
:ok
String.contains?(trimmed, "No such process") ->
:ok
true ->
{:error, if(trimmed == "", do: {:exit_status, code}, else: trimmed)}
end
else
{:error, reason} ->
SLog.warning(@log_category, "Failed to execute kill command: #{inspect(reason)}")
{:error, reason}
end
end
@doc """
Checks if a process is alive.
Uses kill -0 (signal 0) which doesn't kill but checks existence.
"""
def process_alive?(os_pid) when is_integer(os_pid) do
case :os.type() do
{:unix, :linux} ->
process_alive_via_proc(os_pid)
{:unix, _} ->
process_alive_via_ps(os_pid)
_ ->
false
end
end
def process_alive?(_), do: false
defp process_alive_via_ps(os_pid) do
with {:ok, ps_path} <- require_executable("ps", @ps_command_candidates),
{:ok, output, 0} <-
run_command(ps_path, ["-p", Integer.to_string(os_pid), "-o", "pid="]) do
String.trim(output || "") != ""
else
_ -> false
end
end
defp process_alive_via_proc(os_pid) do
stat_path = "/proc/#{os_pid}/stat"
case File.read(stat_path) do
{:ok, content} ->
case parse_proc_state(content) do
{:ok, state} -> state not in ["Z", "X", "x"]
:error -> File.exists?(stat_path)
end
{:error, :enoent} ->
false
{:error, _} ->
File.exists?(stat_path)
end
end
defp parse_proc_state(content) when is_binary(content) do
case String.split(content, ") ", parts: 2) do
[_prefix, rest] ->
case String.split(rest, " ", parts: 2) do
[state | _] -> {:ok, state}
_ -> :error
end
_ ->
:error
end
end
@doc """
Gets the command line of a process.
POSIX-compliant using /proc on Linux, ps on macOS/BSD.
"""
def get_process_command(os_pid) when is_integer(os_pid) do
# Try Linux /proc first (fastest)
proc_file = "/proc/#{os_pid}/cmdline"
if File.exists?(proc_file) do
# Linux: Read /proc/PID/cmdline
case File.read(proc_file) do
{:ok, content} ->
# cmdline uses null bytes as separators
command = content |> String.split(<<0>>) |> Enum.join(" ") |> String.trim()
{:ok, command}
{:error, _} ->
get_process_command_ps(os_pid)
end
else
# macOS/BSD: Use ps command
get_process_command_ps(os_pid)
end
end
@doc """
Gets the process group ID (PGID) for a process.
"""
def get_process_group_id(os_pid) when is_integer(os_pid) do
case :os.type() do
{:unix, :linux} ->
get_process_group_id_proc(os_pid)
{:unix, _} ->
get_process_group_id_ps(os_pid)
_ ->
{:error, :not_supported}
end
end
defp get_process_group_id_proc(os_pid) do
stat_path = "/proc/#{os_pid}/stat"
case File.read(stat_path) do
{:ok, content} -> parse_stat_pgid(content)
{:error, _} -> {:error, :not_found}
end
end
defp parse_stat_pgid(content) do
case String.split(content, ") ", parts: 2) do
[_prefix, rest] -> extract_pgid_from_stat(rest)
_ -> {:error, :parse_error}
end
end
defp extract_pgid_from_stat(rest) do
case String.split(rest, " ", trim: true) do
[_state, _ppid, pgrp | _] -> parse_pgid(pgrp)
_ -> {:error, :parse_error}
end
end
defp parse_pgid(pgrp) do
case Integer.parse(pgrp) do
{pgid, ""} -> {:ok, pgid}
_ -> {:error, :parse_error}
end
end
defp get_process_group_id_ps(os_pid) do
with {:ok, ps_path} <- require_executable("ps", @ps_command_candidates),
{:ok, output, 0} <-
run_command(ps_path, ["-p", Integer.to_string(os_pid), "-o", "pgid="]) do
case Integer.parse(String.trim(output || "")) do
{pgid, ""} -> {:ok, pgid}
_ -> {:error, :not_found}
end
else
{:error, {:executable_not_found, _cmd}} ->
{:error, :not_found}
_ ->
{:error, :not_found}
end
end
defp get_process_command_ps(os_pid) do
with {:ok, ps_path} <- require_executable("ps", @ps_command_candidates),
{:ok, output, 0} <-
run_command(ps_path, ["-p", Integer.to_string(os_pid), "-o", "args="]) do
case String.trim(output || "") do
"" -> {:error, :not_found}
cmd -> {:ok, cmd}
end
else
{:error, {:executable_not_found, _cmd}} ->
{:error, :not_found}
_ ->
{:error, :not_found}
end
end
@doc """
Kills all processes matching a run ID.
Pure Erlang implementation, no pkill.
"""
def kill_by_run_id(run_id) when is_binary(run_id) do
SLog.warning(@log_category, "🔪 Killing all processes with run_id: #{run_id}")
SLog.debug(@log_category, "kill_by_run_id called at: #{System.monotonic_time(:millisecond)}")
caller_info = Process.info(self(), [:registered_name, :current_stacktrace])
SLog.debug(@log_category, "Called from: #{inspect(caller_info)}")
# Get all Python processes
python_pids = find_python_processes()
# Filter by run_id in command line
# Support both --snakepit-run-id (current) and --run-id (future)
matching_pids =
python_pids
|> Enum.filter(fn pid ->
case get_process_command(pid) do
{:ok, cmd} ->
has_grpc_server = String.contains?(cmd, "grpc_server.py")
has_old_format = String.contains?(cmd, "--snakepit-run-id #{run_id}")
has_new_format = String.contains?(cmd, "--run-id #{run_id}")
has_grpc_server and (has_old_format or has_new_format)
_ ->
false
end
end)
SLog.info(@log_category, "Found #{length(matching_pids)} processes to kill")
# Kill with escalation
killed_count =
Enum.reduce(matching_pids, 0, fn pid, acc ->
case kill_with_escalation(pid) do
:ok ->
acc + 1
{:error, reason} ->
SLog.warning(@log_category, "Failed to kill #{pid}: #{inspect(reason)}")
acc
end
end)
{:ok, killed_count}
end
defp find_python_processes_linux do
case File.ls("/proc") do
{:ok, entries} ->
entries
|> Enum.reduce([], &collect_python_pid/2)
|> Enum.uniq()
{:error, _} ->
find_python_processes_posix()
end
end
defp collect_python_pid(entry, acc) do
case Integer.parse(entry) do
{pid, ""} ->
if python_command?(pid), do: [pid | acc], else: acc
_ ->
acc
end
end
defp python_command?(pid) do
comm_path = "/proc/#{pid}/comm"
cmdline_path = "/proc/#{pid}/cmdline"
if File.exists?(comm_path) do
case File.read(comm_path) do
{:ok, comm} ->
comm
|> String.trim()
|> String.downcase()
|> String.contains?("python")
_ ->
python_cmdline?(cmdline_path)
end
else
python_cmdline?(cmdline_path)
end
rescue
_ -> false
end
defp python_cmdline?(path) do
case File.read(path) do
{:ok, content} ->
try do
content
|> String.replace(<<0>>, " ")
|> String.downcase()
|> String.contains?("python")
rescue
ArgumentError -> false
end
_ ->
false
end
end
defp find_python_processes_posix do
with {:ok, ps_path} <- require_executable("ps", @ps_command_candidates),
{:ok, output, 0} <- run_command(ps_path, ["-eo", "pid,comm"]) do
output
|> String.split("\n", trim: true)
|> Enum.reduce([], fn line, acc ->
trimmed = String.trim_leading(line)
case Regex.split(~r/\s+/, trimmed, parts: 2) do
[pid_str, command] ->
with {pid, ""} <- Integer.parse(pid_str),
true <- String.contains?(String.downcase(command), "python") do
[pid | acc]
else
_ -> acc
end
_ ->
acc
end
end)
|> Enum.reverse()
else
{:error, {:executable_not_found, _cmd}} ->
SLog.warning(@log_category, "ps command not available; skipping python process discovery")
[]
_ ->
[]
end
end
@doc """
Finds all Python processes on the system.
Returns a list of OS PIDs.
"""
def find_python_processes do
case :os.type() do
{:unix, :linux} -> find_python_processes_linux()
{:unix, _} -> find_python_processes_posix()
_ -> []
end
end
@doc """
Kills a process with escalation: SIGTERM -> wait -> SIGKILL
"""
def kill_with_escalation(os_pid, timeout_ms \\ 2000) do
# Try SIGTERM first (graceful)
case kill_process(os_pid, :sigterm) do
:ok ->
# Wait for process to die
if wait_for_death(os_pid, timeout_ms) do
SLog.debug(@log_category, "✅ Process #{os_pid} terminated gracefully")
:ok
else
# Escalate to SIGKILL
SLog.warning(@log_category, "⏰ Process #{os_pid} didn't die, escalating to SIGKILL")
kill_process(os_pid, :sigkill)
end
error ->
error
end
end
@doc """
Kills a process group with escalation: SIGTERM -> wait -> SIGKILL.
"""
def kill_process_group_with_escalation(pgid, timeout_ms \\ 2000) when is_integer(pgid) do
case kill_process_group(pgid, :sigterm) do
:ok ->
if wait_for_death(pgid, timeout_ms) do
SLog.debug(@log_category, "✅ Process group #{pgid} terminated gracefully")
:ok
else
SLog.warning(@log_category, "⏰ Process group #{pgid} didn't die, escalating to SIGKILL")
kill_process_group(pgid, :sigkill)
end
error ->
error
end
end
defp wait_for_death(os_pid, timeout_ms) do
deadline = System.monotonic_time(:millisecond) + timeout_ms
wait_for_death_loop(os_pid, deadline, 1)
end
# Non-blocking polling with exponential backoff using receive after.
# Starts at 1ms, doubles to 2ms, 4ms, 8ms, capping at 100ms.
# This is the OTP-correct way to implement timed waits without blocking the scheduler.
defp wait_for_death_loop(os_pid, deadline, backoff) do
if System.monotonic_time(:millisecond) >= deadline do
false
else
if process_alive?(os_pid) do
delay = min(backoff, 100)
# OTP-idiomatic non-blocking wait - integrates with process mailbox and scheduler
receive do
after
delay -> :ok
end
wait_for_death_loop(os_pid, deadline, backoff * 2)
else
true
end
end
end
defp signal_to_number(:sigterm), do: 15
defp signal_to_number(:sigkill), do: 9
defp signal_to_number(:sighup), do: 1
defp signal_to_number(n) when is_integer(n), do: n
defp require_executable(cmd, fallback_paths) when is_list(fallback_paths) do
case System.find_executable(cmd) do
nil ->
fallback_paths
|> Enum.find(&File.exists?/1)
|> case do
nil -> {:error, {:executable_not_found, cmd}}
path -> {:ok, path}
end
path ->
{:ok, path}
end
end
defp run_command(path, args) when is_binary(path) and is_list(args) do
{output, status} = System.cmd(path, args, stderr_to_stdout: true)
{:ok, output, status}
rescue
error -> {:error, error}
end
end