У меня было много раз, когда мне приходилось останавливать функцию в середине выполнения, не имея доступа к ней, или она имела сотни строк кода, где проверка if каждой строки неосуществима илиударил как-то. Это может быть вызвано самим кодом.
Я использовал sys.settrace
, чтобы выполнить задачу, как в этом 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.
А затем нужно что-то связать, чтобы установитьсобытие: «_stop.set()
», когда необходимо, с помощью обратного вызова или таймера, и есть попытка, за исключением, чтобы поймать его.
Но это неприятный хак, и, как говорят другие это обычно не нужно. Я верю им, я просто не могу обернуть голову вокруг другого решения.
Поэтому я надеюсь, что кто-то узнает «правильное» решение, где не используются sys.settrace
, для терминации long_function
.