Best (pythonic) way to interrupt and cancel a function call?
I want the function to foo()
perform multiple operations on the same thread with timeouts during which it periodically checks if another thread (main script) has set the variable please_stop
to True
, in which case I want to foo()
return a value immediately (for example False
).
My initial attempt looks like this:
import time
please_stop = False
def wait( n ) :
'''
Wait n seconds and return True, unless some other process
sets please_stop to True (and then return False).
'''
global please_stop
for t in range (n) :
if please_stop :
please_stop = False
return False
else :
print t
time.sleep(1)
return True
def foo() :
'''
Do a number of things, with wait() calls in between.
'''
global please_stop
print 'doing first thing'
if not wait( 5 ) : return False
print 'doing second thing'
if not wait( 5 ) : return False
print 'doing third thing'
return True
foo()
It works as it should, but I find the expressions if not wait( 5 ) : return False
very clumsy. What would be the easiest / most elegant way to achieve this behavior, perhaps instead of using a simple expression wait( 5 )
? I'm reinventing the wheel, something like interrupts will do a better job here?
Thanks in advance.
source to share
Well, you can simplify the implementation wait
by using threading.Event
instead of a global variable:
import time
import threading
def wait(event, timeout):
if event.wait(timeout):
event.clear()
return False
else:
return True
def foo(event):
'''
Do a number of things, with wait() calls in between.
'''
print 'doing first thing'
if not wait(event, 5): return False
print 'doing second thing'
if not wait(event, 5): return False
print 'doing third thing'
return True
if __name__ == "__main__":
event = threading.Event()
t = threading.Thread(target=foo, args=(event,))
t.start()
time.sleep(6)
event.set()
Output:
doing first thing
doing second thing
But you are still using the template if not wait(5) : return False
. To get rid of this, you need to throw an exception if set Event
. You probably don't want the exception to get thrown out foo
, so you either need to use a block try
/ except
around the body foo
or use a context manager :
import time
import threading
from contextlib import contextmanager
class Interrupt(Exception): pass
def wait(event, timeout):
if event.wait(timeout):
event.clear()
raise Interrupt()
else:
return True
@contextmanager
def interruptible():
try:
yield
except Interrupt:
pass
def foo(event):
'''
Do a number of things, with wait() calls in between.
'''
with interruptible():
print 'doing first thing'
wait(event, 5)
print 'doing second thing'
wait(event, 5)
print 'doing third thing'
return True
return False # This will only be reached the main thread interrupts foo
This will behave exactly the same as in the previous example, the only difference is instead of returning False
to the end foo
, we throw an exception that hits the context manager, but forces us to skip the rest of the code inside the context.
source to share