Investigate signals and threads/processes

signal_exp_01.py

#!/usr/bin/env python3
#
# In this code we investigate how to "gracefully" exit all child processes
# and threads. "Gracefully" implies "each thread/process gets through the
# whole release work before exitting, minimizing the lisk of resource leaks or
# other unexpected behaviors"
#
# In the real world, it may be OK just to use thread.daemon = True or
# process.daemon = True, especially if you do not care of long term stability.
# Container technology may allow that kind of rough edges in your codes.
#
# This code only takes care of KeyboardInterrupt and a few signals;
# SIGTERM and SIGUSR1.
#
# Key points:
#
# - KeyboardInterrupt will be sent not only to main process but to subprocesses
# - KeyboardInterrupt will NOT be sent to child threads
# -- Main thread should take care of it
# - Signals will NOT be sent to subprocess (unless explicitly done so)
# -- os.kill() will be your friend.
# - Signals to a process will be sent to its child threads
# - signal.Signals(signum) seems introduced in 3.5 and somewhat useful.
# -- Not explicitly documented though.
#

import multiprocessing
import os
import signal
import sys
import threading
import time
import traceback

SLEEP_TIME = 2
SHOW_STACK_TRACE = False


class AppError(RuntimeError):
    pass


def _run_thread(do_exit):
    try:
        while not do_exit.is_set():
            print('(Thread still running)', file=sys.stderr)
            time.sleep(SLEEP_TIME)
    except KeyboardInterrupt as e:
        print('Thread received keyboardInterrupt', file=sys.stderr)
    except Exception as e:
        print(f'Thread received Exception: {e}', file=sys.stderr)
        if SHOW_STACK_TRACE:
            tb = e.__traceback__
            for line in traceback.format_tb(tb):
                print(line.rstrip(), file=sys.stderr)
    print('Thread exitting')


def _run_subprocess():
    try:
        def _handle_signal(signum, frame):
            print(f'Subprocess received signum: {signum}', file=sys.stderr)
            sig_name = signal.Signals(signum).name  # available since 3.5?
            raise AppError('Received {}'.format(sig_name))
        
        signal.signal(signal.SIGTERM, _handle_signal)
        signal.signal(signal.SIGUSR1, _handle_signal)

        while True:
            print('(Subprocess still running)', file=sys.stderr)
            time.sleep(SLEEP_TIME)
    except KeyboardInterrupt as e:
        print('Subprocess received keyboardInterrupt', file=sys.stderr)
    except Exception as e:
        print(f'Subprocess received Exception: {e}', file=sys.stderr)
        if SHOW_STACK_TRACE:
            tb = e.__traceback__
            for line in traceback.format_tb(tb):
                print(line.rstrip(), file=sys.stderr)
    print('Subprocess exitting')


def main():
    try:
        do_exit = threading.Event()
        t = threading.Thread(name='Thread', target=_run_thread,
                             args=(do_exit,))
        p = multiprocessing.Process(name='Subprocess', target=_run_subprocess)
        def _handle_signal(signum, frame):
            print(f'Main thread received signum: {signum}', file=sys.stderr)
            # Let thread exit.
            do_exit.set()
            # Let subprocess exit (Main thread won't receive this)
            print('Main process manually terminate subprocess')
            if p.is_alive():
                os.kill(p.pid, signum)
                # p.terminate()
            sig_name = signal.Signals(signum).name  # after 3.5
            # Send Exception to the main thread itself to graceful exit
            raise AppError('Received {}'.format(sig_name))

        signal.signal(signal.SIGTERM, _handle_signal)
        signal.signal(signal.SIGUSR1, _handle_signal)

        t.start()
        p.start()

        while True:
            print('(Main process still running)', file=sys.stderr)
            time.sleep(SLEEP_TIME)
        print('Exitting main()', file=sys.stderr)
    except KeyboardInterrupt:
        print('Main process received keyboardInterrupt',
              file=sys.stderr)
        do_exit.set()
    except Exception as e:
        print(f'Main process received Exception: {e}', file=sys.stderr)
        if SHOW_STACK_TRACE:
            tb = e.__traceback__
            for line in traceback.format_tb(tb):
                print(line.rstrip(), file=sys.stderr)

    if p.is_alive():
        print('Joining subprocess (again)', file=sys.stderr)
        p.join(3)
    if t.is_alive():
        print('Joining thread (again)', file=sys.stderr)
        t.join(3)

    print('Main process exitting ({}, {})'
          .format(t.is_alive(), p.is_alive()),
          file=sys.stderr)
    return 0


if __name__ == '__main__':
    sys.exit(main())