I have had many time where I had to stop a function in middle of a run, without having access to it, or it have hundreds of lines of code, where a if-check every line is not feasible, or it is struck somehow. That an be triggered by the code, itself.
I have used sys.settrace
to get the task done, like in this MWE:
from threading import Event, Thread
import sys
import time
_stop = Event()
def _globaltrace(frame, event, arg):
"""The global trace call handler."""
if event == 'call':
return _localtrace
else:
return None
def _localtrace(frame, event, arg):
"""The local trace call handler."""
global _stop
if _stop.is_set():
if event == 'line':
raise Exception()
return _localtrace
if __name__ == "__main__":
def long_function():
# The Function that need to be terminated before it ends.
# That we _CAN'T_ change anything in.
print("Long Function Started.")
while True:
time.sleep(1)
def simulate_external_input_callback():
global _stop
time.sleep(5)
print("Kill signal received.")
_stop.set()
sys.settrace(_globaltrace)
th = Thread(target=simulate_external_input_callback)
th.start()
try:
long_function()
except Exception:
print("Long Function Killed.")
# Try to do some cleanup.
else:
print("Long function ended normally.")
print("The program continues, like nothing happened.")
th.join() # Just here for clean up.
And then have a something bound to set the event: _stop.set()
when needed, by callback, or timer, and have a try-except to catch it.
But it is a nasty hack, and as others say it is usually unnecessary. I believe them, I can just not wrap my head around another solution.
So I hope that someone would know the "right" solution, where there are not used the sys.settrace
, to terminal the long_function
.