54 lines
1.3 KiB
Python
54 lines
1.3 KiB
Python
import time
|
|
from typing import List
|
|
|
|
import psutil
|
|
|
|
CURRENT_PROCESS = psutil.Process()
|
|
|
|
|
|
def terminate_processes(procs: List[psutil.Process], timeout=3):
|
|
# https://psutil.readthedocs.io/en/latest/#terminate-my-children
|
|
for p in procs:
|
|
p.terminate()
|
|
gone, alive = psutil.wait_procs(procs, timeout=timeout)
|
|
|
|
if alive:
|
|
# send SIGKILL
|
|
for p in alive:
|
|
p.kill()
|
|
gone, alive = psutil.wait_procs(alive, timeout=timeout)
|
|
return alive
|
|
|
|
|
|
def terminate_subprocesses(parent: psutil.Process = CURRENT_PROCESS):
|
|
terminate_processes(parent.children(recursive=True))
|
|
|
|
|
|
def get_root_process() -> psutil.Process:
|
|
proc = CURRENT_PROCESS
|
|
while proc.parent():
|
|
proc = proc.parent()
|
|
return proc
|
|
|
|
|
|
def is_process_running(pid: int) -> bool:
|
|
try:
|
|
psutil.Process(pid)
|
|
return True
|
|
except psutil.NoSuchProcess:
|
|
return False
|
|
|
|
|
|
def wait_for_process_terminate(pid: int):
|
|
while is_process_running(pid):
|
|
time.sleep(0.1)
|
|
|
|
|
|
def get_running_process_for(process):
|
|
for proc in get_root_process().children(recursive=True):
|
|
try:
|
|
if proc.environ().get("CATFISH_IDENT") == process.ident:
|
|
return proc
|
|
except psutil.AccessDenied:
|
|
continue
|