使 Python 中的 C++ 函数超时

Question

我实现了 python-cpp 绑定（使用 boost-python），以便从 Python 调用

foop()

运行 C++ 函数

fooc()

。我想从 Python 设置一个超时，以便

foop

在

t

秒后返回。这里的 解决方案 适用于 Python 函数，但不适用于

foop

b/c 我无法中断 C++ 代码 - 下面是调用

run_for_timeout(foop)

的示例。有没有办法从 Python 执行此操作（即无需在 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

Answer 1

要使用 boost-python 为从 Python 调用的 C++ 函数实现超时，您可以利用 Python 的

multiprocessing

模块创建一个单独的进程来调用 C++ 函数，然后您可以使用

timeout

方法设置进程运行的最长时间。如果进程在指定的超时时间内没有完成，您可以终止它并将其视为超时事件。以下是如何修改代码来实现此目的：

join()

此修改后的代码创建一个用于调用 C++ 函数的新进程，并使用进程上的 

import signal
from multiprocessing import Process, Queue

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 = {}

    result_queue = Queue()

    def target():
        try:
            result = func(*args, **kwargs)
            result_queue.put(result)
        except Exception as e:
            result_queue.put(e)

    process = Process(target=target)

    try:
        process.start()
        process.join(timeout)
        if process.is_alive():
            process.terminate()
            process.join()
            raise TimeoutError()
        else:
            result = result_queue.get()
            if isinstance(result, Exception):
                raise result
            return result
    finally:
        process.close()
        signal.alarm(0)

# Example usage:
# Assuming you have a C++ function fooc() and its Python wrapper foop().
# result = run_for_timeout(foop, args=(...), kwargs={...}, timeout=5)

方法设置指定的超时。如果该过程未在指定时间内完成，则会引发

join()

，如果及时完成，则会检索并返回结果。请注意，此方法假设 C++ 函数调用不是阻塞操作，因为它在单独的进程中运行。如果 C++ 函数有任何阻塞调用，则可能需要额外的同步机制来正确处理超时。