Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Save Sap333/b80c5e0a375d1b93e3ff to your computer and use it in GitHub Desktop.
Save Sap333/b80c5e0a375d1b93e3ff to your computer and use it in GitHub Desktop.
Kill all child processes on program exit. Also catches (sets up a hande) the SIGTERM signal to process it gracefully. Python. Linux/Posix only.
#!/usr/bin/python3
# It is for Python3 but may work for modern 2.x versions as well.
import os, signal, atexit
# Signletone which is made as a class with static methods.
# The "init()" method must be called before any child process are created.
# "killAllChildrenOnce" or "killAllChildren" methods could be called
# any time to kill all child processes.
# Creates a new process group during initialization and uses it to send
# the SIGTERM signal to all processes in that group on exit.
class KillAllChildrenProcesses:
__initComplete=False # It can be initialized only once - this is the flag.
__childrenKilled=False # Childrens initialized only once - this is the flag.
@classmethod
def __atExit(cls):
# Disable SIGTERM handler.
signal.pthread_sigmask(signal.SIG_BLOCK, (signal.SIGTERM,))
# Kill all children processes.
cls.killAllChildrenOnce()
#------------------------------------------------------------------------
# Kills all child processes. Does it only on the first call.
# Does nothing on subsequent calls.
@classmethod
def killAllChildrenOnce(cls):
if cls.__childrenKilled: return
cls.__childrenKilled=True
cls.killAllChildren()
os.killpg( os.getpid(), signal.SIGTERM )
#------------------------------------------------------------------------
# Kills all child processes. Does it every time is called.
@classmethod
def killAllChildren(cls):
# Kill all process in out process group.
os.killpg( os.getpid(), signal.SIGTERM )
#------------------------------------------------------------------------
# Initializes this singletone. Does it only once on the first call.
@classmethod
def init(cls):
if cls.__initComplete: return
cls.__initComplete = True
os.setsid() # Create our own process group.
atexit.register(cls.__atExit) # Register an exit handler.
#------------------------------------------------------------------------
#----------------------------------------------------------------------------
# The TERM signal handler.
def onSIGTERM( signum, frame ):
# Must be executed only once - set the "executed" attribute as a flag.
if hasattr(onSIGTERM, "executed"): return
onSIGTERM.executed = True # Create a flag attribute - we've been allready executed.
# Disable SIGTERM handling - just for sure.
signal.pthread_sigmask(signal.SIG_BLOCK, (signal.SIGTERM,))
# Replace the next line with custom code if you want other actions on SITERM.
quit()
#-----------------------------------------------------------------------------
if __name__ == '__main__' :
# Setup the custom SIGTERM handler - might be not necessary.
signal.signal(signal.SIGTERM, onSIGTERM)
# Only for non-blocking code!
# Uncomment the next line if you want SIGTERM not to interrupt system calls -
#signal.siginterrupt(signal.SIGTERM, False)
# Initialize our singleton - all children will be killed on exit since now.
KillAllChildrenProcesses.init()
#-----------------------------------------------------------------------------
@Ashark
Copy link

Ashark commented May 19, 2024

You cannot do

os.setsid() # Create our own process group.

instead, use

os.setpgrp()  # Create our own process group.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment