Timeout a C++ function from Python

Question

I have python-cpp bindings implemented (using boost-python) such that calling foop() from Python runs a C++ function fooc(). I would like to set a timeout from Python such that foop returns after t seconds. The solutions here work fine for Python functions, but not with foop b/c I'm unable to interrupt the C++ code -- example below for calling run_for_timeout(foop). Is there a way to do this from Python (i.e. without implementing the timer functionality in C++)?

import signal

class CallbackValueError(ValueError):
    """Raise for improper data values with callback functions and their utils."""
    pass

class TimeoutError(RuntimeError):
    pass

def run_for_timeout(func, args=(), kwargs=None, timeout=5):
    """Run a function until it times-out.

    Note that ``timeout`` = 0 does not imply no timeout, but rather there is
    no time for the function to run and this function raises an error

    Parameters
    ----------
    func : function
    args : tuple
    kwargs : dict | None
    timeout : int
        (seconds)

    Returns
    -------
    result : object | None
        Return object from function, or None if it timed out

    Raises
    ------
    CallbackValueError

    """
    if timeout <= 0:
        raise CallbackValueError("{}s is a nonsensical value for the "
                                 "timeout function".format(timeout))

    def handler(signum, frame):
        raise TimeoutError()

    # Set the timeout handler
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(timeout)

    if kwargs is None:
        kwargs = {}

    try:
        result = func(*args, **kwargs)
    except TimeoutError as e:
        result = None
    finally:
        # Function returned before timeout, so cancel the timer
        signal.alarm(0)

    return result

The `multiprocessing` solution from your list should work just fine. — n. m. could be an AI, Jul 25 '23 at 12:08

score 0 · Answer 1 · answered Aug 01 '23 at 08:21

Since the C++ function fooc() is running through Python's foop() and the C++ code itself is not directly accessible from Python, it is challenging to directly interrupt the C++ code using Python's signal handling mechanisms.

import multiprocessing
import time

# C++ function wrapped by foop()
def fooc():
    time.sleep(10)  # Simulate a long-running C++ computation (10 seconds)
    return "Done"

# Wrapper function to run fooc() with a timeout
def foop_with_timeout(timeout=5):
    result_queue = multiprocessing.Queue()

    def fooc_wrapper(queue):
        try:
            result = fooc()
            queue.put(result)
        except Exception as e:
            queue.put(e)

    process = multiprocessing.Process(target=fooc_wrapper, args=(result_queue,))
    process.start()
    process.join(timeout)

    if process.is_alive():
        process.terminate()
        process.join()
        return None  # Return None if the timeout was reached

    result = result_queue.get()
    if isinstance(result, Exception):
        raise result
    return result

if __name__ == "__main__":
    try:
        result = foop_with_timeout(5)
        if result is None:
            print("Function timed out.")
        else:
            print("Result:", result)
    except Exception as e:
        print("An error occurred:", e)

To make this work, I had to put `fooc_wrapper()` outside `foop_with_timeout()` - but then it worked like a charm in contrast to many other solutions I tried. — user2296653, Aug 01 '23 at 08:41

Timeout a C++ function from Python

1 Answers1