Skip to content

Instantly share code, notes, and snippets.

Created September 1, 2015 21:09
Show Gist options
  • Save anonymous/fbb0719b8658a6441c10 to your computer and use it in GitHub Desktop.
Save anonymous/fbb0719b8658a6441c10 to your computer and use it in GitHub Desktop.
Kill all child process on program exit. Also catches (sets up a hande) the SIGTERM signal to process it gracefully.
import os, signal, atexit
# Signletone made as a class with static methonds.
# The "init()" method must be called before any child process are created.
# "killAllChildrenOnce" or "killAllChildren" methods could be called
# any time to kill all chil process.
# Creates a new process group during initialization and uses it to send
# the SIGTERM signal to all process 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.__killAllChildren()
#------------------------------------------------------------------------
# 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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment