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.

psutil
view raw requirements.txt hosted with ❤ by GitHub
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()
view raw safequitprocess.py hosted with ❤ by GitHub
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()
view raw test.py hosted with ❤ by GitHub