# Uncomment test calls at the end from threading import Thread, Event from time import sleep, perf_counter as pc done = Event() # 'thread finished' flag def work(t): sleep(t) done.set() def test_default_handler(twork, twait): t = Thread(target=work, args=(twork,)) done.clear() t0 = pc() try: t.start() print('started. Press Ctrl-C now') t.join(twait) except KeyboardInterrupt: log('Ctrl-C', t0, t) t.join() # this hangs if twork < twait log('finish', t0, t) def test_custom_handler(twork, twait): def sig_handler(signal, frame): log('Ctrl-C', t0, t) import signal signal.signal(signal.SIGINT, sig_handler) t = Thread(target=work, args=(twork,)) done.clear() t0 = pc() t.start() print('started. Press Ctrl-C now') t.join(twait) t.join() # this does not hang log('finish', t0, t) # debug tool # format: event, seconds_passed, thread finished?, thread alive? def log(ev, t0, thr): print('%s [%.2f] done=%s alive=%s' % (ev, pc() - t0, done.is_set(), thr.is_alive())) # Test calls below. #1 hangs, #2-4 do not. # Note how #1 and #3 print done=True and alive=True, # a state when thread has finished its code but is still alive # 1. Work is shorter than timeout. test_default_handler(twork=3, twait=4) # 2. Work is longer than timeout: #test_default_handler(twork=4, twait=3) # 3. #test_custom_handler(twork=3, twait=4) # 4. #test_custom_handler(twork=4, twait=3)