Here is an example code snippet that executes 'last rites'. My current use-case is to handle final tasks on a cluster job management system when the script violates some execution criterion e.g. exceeding memory quota. It's helpful if the job could perform some task(s) that indicate that it had to be terminated.
The key module here is the Python signal
package which provides a way to handle various system signals. The example here sends an email whenever the script is terminated. To use signal
you need to register one or more signals to have a handle i.e. a callback to be executed if that signal is raised. For example:
signal.signal(signal.SIGINT, handler)
registers the handler
function when the script receives an keyboard interrupt (Control-C) event.
You can view a full list of signals here.
# -*- coding: utf-8 -*-
import os
import signal
import smtplib
import socket
import sys
# import time
from email.mime.text import MIMEText
signal_dict = {getattr(signal, s): s for s in dir(signal) if s.startswith('SIG') and not s.startswith('SIG_')}
def send_email(t, s, b):
"""Send email"""
f = "send@from.email"
msg = MIMEText(b)
msg['To'] = t
msg['From'] = f
msg['Subject'] = s
s = smtplib.SMTP(
"email.hostname",
"email.port",
)
s.sendmail(f, [t], msg.as_string())
s.quit()
def handler(signum, frame):
"""Handle signal"""
send_email(
'send@to.email',
'received signal {} on {}'.format(signal_dict[signum], socket.gethostname()),
'terminated',
)
raise Exception
def main():
# register signal handlers
signal.signal(signal.SIGINT, handler)
signal.signal(signal.SIGQUIT, handler)
signal.signal(signal.SIGILL, handler)
signal.signal(signal.SIGABRT, handler)
signal.signal(signal.SIGFPE, handler)
signal.signal(signal.SIGSEGV, handler)
signal.signal(signal.SIGPIPE, handler)
signal.signal(signal.SIGALRM, handler)
signal.signal(signal.SIGTERM, handler)
# do something
send_email(
'send@to.email',
'starting on host {}'.format(socket.gethostname()),
'started...',
)
# simulate a long running task that gets manually killed
# time.sleep(100)
# simulate a task that violates the memory limit
with open(sys.argv[1], 'rb') as f:
g = f.readlines()
h = g[::]
i = h[::]
send_email(
'send@to.email',
'successfully completed on host {}'.format(socket.gethostname()),
'completed...',
)
return os.EX_OK
if __name__ == "__main__":
sys.exit(main())