Python SafeQuitProcess: On termination the process also propagates the termination signal to all its children
If a process has children, kill the parent won’t be enough to guarantee “on orphaned process”. You basically have to recursively kill all the children process as well.
Here is the way to do it.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
psutil |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from multiprocessing import Process | |
class SafeQuitProcess(Process): | |
def __init__(self, target, args=[], kwargs=dict()): | |
super(SafeQuitProcess, self).__init__(target=target, | |
args=args, | |
kwargs=kwargs) | |
def terminate(self): | |
''' | |
Recursively terminates all the child processes | |
''' | |
import psutil, signal | |
p_process = psutil.Process(self.pid) | |
children = p_process.children(recursive=True) | |
for process in children: | |
process.send_signal(signal.SIGTERM) | |
super(SafeQuitProcess, self).terminate() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
class SafeQuitProcessTest(unittest.TestCase): | |
def test_terminate(self): | |
from multiprocessing import Queue | |
err = Queue() | |
def dummy(): | |
import time | |
time.sleep(1) | |
err.put(True) | |
p = SafeQuitProcess(target=dummy) | |
p.start() | |
p.terminate() | |
p.join() | |
self.assertTrue(err.empty()) | |
def test_terminate_recursively(self): | |
from multiprocessing import Queue | |
all_processes = Queue() | |
def dummy(): | |
import os | |
all_processes.put(os.getpid()) | |
import time | |
time.sleep(0.5) | |
def complex_dummy(): | |
p = Process(target=dummy) | |
p.start() | |
p.join() | |
def run(): | |
import time | |
p = SafeQuitProcess(target=complex_dummy) | |
p.start() | |
time.sleep(0.1) | |
p.terminate() | |
p.join() | |
if all_processes.empty(): | |
raise Exception('the dummy has not been spawned') | |
else: | |
dummy_pid = all_processes.get() | |
import os | |
# it should not be able to kill a non-existing process | |
self.assertRaises(OSError, os.kill, dummy_pid, 0) | |
run() |