Interrupt function execution from another function in Python

Question

I have a function a doing some tasks and another function b being a callback to some events. Whenever an event happens, function b is called and I would like to make it able to interrupt the execution of function a. Both functions are declared inside the same class.

Function a is not supposed to call function b. Function b is totally independent, it is a callback to an external event like "user face detected" coming from ROS: robot operating system.

what I need is basically something like Ctrl+C that can be called from within Python and which only aborts a targeted function and not the whole program.

Can this be done in Python?

I am using ROS (robot operating system) where you can subscribe any function to some events. — Mehdi, Jul 30 '14 at 06:16
Something like this??? chk = True def function_a(): if chk: do_process() def function_b(): global chk chk = False — Selva, Jul 30 '14 at 06:41
Yes I thought about that but then I would need to put if chk on every block of the script do_process so that the program can quit even during the do_process() and not after it finishes. — Mehdi, Jul 30 '14 at 06:48

Michael Innes · Answer 1 · 2017-12-08T02:41:37.547

It's generally recommended not to use exception calling for flow-control. Instead, look to python stdlib's threading.Event, even if you only plan on using a single thread (even the most basic Python program uses at least one thread).

This answer https://stackoverflow.com/a/46346184/914778 has a good explanation of how calling one function (function b) could interrupt another (function a).

Here's a few important parts, summarized from that other answer.

Set up your threading libraries:

from threading import Event
global exit
exit = Event()

This is a good replacement for time.sleep(60), as it can be interrupt:

exit.wait(60)

This code will execute, until you change exit to "set":

while not exit.is_set():
    do_a_thing()

This will cause exit.wait(60) to stop waiting, and exit.is_set() will return True:

exit.set()

This will enable execution again, and exit.is_set() will return False:

exit.clear()

this is a lot more clear than most examples I've been finding on the interwebs, ty — James Robert Albert, Aug 26 '22 at 19:45

score 5 · Answer 2 · answered Jul 30 '14 at 06:09

I would do the following:

define a custom exception
call the callback function within an appropriate try/catch block
if the callback function decides to break the execution, it will raise exception and the caller will catch it and handle it as needed.

Here's some pseudo-code:

class InterruptExecution (Exception):
    pass

def function_a():
    while some_condition_is_true():
        do_something()
        if callback_time():
            try:
                function_b()
            except InterruptExecution:
                break
        do_something_else()
    do_final_stuff()


def function_b():
    do_this_and_that()
    if interruption_needed():
        raise (InterruptExecution('Stop the damn thing'))

thanks for your fast answer, unfortunately misusing the term "callback" in my initial question has led to misunderstandings. function a does not call function b. function b should be able to interrupt function a from outside. — Mehdi, Jul 30 '14 at 06:22

score 0 · Answer 3 · answered Jul 30 '14 at 09:37

I had done using Threading.

    import threading
class myThread (threading.Thread):
    def __init__(self, threadID, name, counter):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.counter = counter
    def run(self):
        # Get lock to synchronize threads
        #threadLock.acquire()
        if self.name == 'a':
            function_a(self.name, self.counter, 3)
        if self.name == 'b':
            function_b(self.name, self.counter, 3)

def function_a(threadName, delay, counter):
    name = raw_input("Name")
    print name

def function_b(threadName, delay, counter):
    global thread1
    thread1.shutdown = True
    thread1._Thread__stop()

# Create new threads
thread1 = myThread(1, "a", 0)
thread2 = myThread(2, "b", 0)

# Start new Threads
thread1.start()
thread2.start()

function_a stopped executing when thread1 is stopped

Interrupt function execution from another function in Python

3 Answers3

Linked