在阻塞 boost c++ 方法中，如何在 Python 中捕获中断信号？

Posted 2023-02-21

【中文标题】在阻塞 boost c++ 方法中，如何在 Python 中捕获中断信号？

【英文标题】：How do I catch an Interrupt signal in Python when inside a blocking boost c++ method?

【发布时间】：2016-10-08 09:20:30

【问题描述】：

我有一个用 C++ 编写的工具集，并为 Python 提供了 Boost 绑定。​​

最初，这段代码都是 C++，我捕获了一个 CTRL+C 中断：

signal( SIGINT, signalCallbackHandler );

和

void signalCallbackHandler(int /*signum*/)

    g_myObject->stop();

这很好用。

但是，现在我已经添加了 Python 绑定，我正在使用 Python 来初始化对象。

我最初的想法是这样做：

import signal

def handle_interrupt( signum, frame ) :
    g_myObject.stop()

signal.signal( signal.SIGINT, handle_interrupt )
g_myObject = MyObject()
g_myObject.start()

但是，这个信号处理程序永远不会被调用。

我应该如何处理这样的中断？我需要在 C++ 中完成，然后从那里调用 Python 函数吗？

【问题讨论】：

如果您的扩展代码在后台线程中运行，请确保释放 GIL。在主线程中，不要忘记在阻塞系统调用被 C 代码中的信号中断后定期调用 PyErr_CheckSignals()（同时持有 GIL）或在 errno==EINTR 上调用。相关：Cython, Python and KeyboardInterrupt ignored。这是more details (in Russian)—look at the code examples

【参考方案1】：

没有调用您的 python 信号处理程序，因为 python 将信号处理程序的执行推迟到要执行下一个字节码指令之后 - 请参阅the library documentation for signal, section 18.8.1.1：

Python 信号处理程序不会在低级 (C) 信号处理程序内执行。相反，低级信号处理程序设置一个标志，告诉虚拟机稍后执行相应的 Python 信号处理程序（例如在下一个字节码指令处）。这会产生以下后果：

捕获由 C 代码中的无效操作引起的同步错误（如 SIGFPE 或 SIGSEGV）毫无意义。 Python 将从信号处理程序返回到 C 代码，这很可能再次引发相同的信号，导致 Python 明显挂起。从 Python 3.3 开始，您可以使用 faulthandler 模块报告同步错误。 纯粹用 C 语言实现的长时间运行的计算（例如对大量文本进行正则表达式匹配）可以在任意时间内不间断地运行，无论接收到任何信号。计算完成时将调用 Python 信号处理程序。

这样做的原因是信号可以在任何时间到达，可能是在执行 python 指令的中途。 VM 开始执行信号处理程序是不安全的，因为 VM 处于未知状态。因此，python实际安装的信号处理器只是设置了一个标志，告诉VM在当前指令完成后调用信号处理器。

如果信号在您的 C++ 函数执行期间到达，则信号处理程序会设置标志并返回到您的 C++ 函数。

如果信号处理程序的主要目的是允许 C++ 函数被中断，那么我建议您放弃 Python 信号处理程序并安装一个 C++ 信号处理程序，该处理程序设置一个标志，触发 C++ 代码中的提前退出（大概返回一个表明它被中断的值）。

无论您是从 python、C++ 还是其他绑定调用代码，这种方法都允许您使用相同的代码。

【参考方案2】：

我有一个 解决方案可以解决这个问题，尽管如果我能在 Python 而不是 C++ 中捕获信号会更简洁。

我之前没有提到的一件事是 MyObject 是一个单例，所以我通过 MyObject.getObject() 得到它

在 Python 中，我有：

def signalHandler( signum ) :
    if signum == signal.SIGINT :
        MyObject.getObject().stop()

def main() :
    signal.signal( signal.SIGINT, handle_interrupt )

    myObject = MyObject.getObject()
    myObject.addSignalHandler( signal.SIGINT, signalHandler )
    myObject.start()

在我的 C++ 代码中，在一个对 Python 一无所知的领域中，我有：

class MyObject

    public :
        void addSignalHandler( int signum, void (*handler)( int, void* ), void *data = nullptr );
        void callSignalHandler( int signum );
    private :
        std::map<int, std::pair<void (*)( int, void* ), void*> > m_signalHandlers;


void signalCallbackHandler( int signum )

    MyObject::getObject()->callSignalHandler( signum );


void MyObject::addSignalHandler( int signum, void (*handler)( int, void* ), void *data )

    m_signalHandlers.insert( std::pair<int, std::pair<void (*)( int, void* ), void *> >( signum, std::make_pair( handler, data ) ) );
    signal( signum, signalCallbackHandler );


void MyObject::callSignalHandler( int signum )

    std::map<int, std::pair<void (*)( int, void* ), void*> >::iterator handler = m_signalHandlers.find( signum );
    if( handler != m_signalHandlers.end() )
    
        handler->second.first( signum, handler->second.second );
    

然后在我的 Python 绑定中：

void signalHandlerWrapper( int signum, void *data )

    if( nullptr == data )
    
        return;
    

    PyObject *func = (PyObject*)( data );
    if( PyFunction_Check( func ) )
    
        PyObject_CallFunction( func, "i", signum );
    


void addSignalHandlerWrapper( MyObject *o, int signum, PyObject *func )

    Py_INCREF( func );
    if( PyFunction_Check( func ) )
    
        o->addSignalHandler( signum, &signalHandlerWrapper, func );
    

我没有，我应该添加的是 addSignalHandlerWrapper() 中的一些东西，它将检查该信号编号是否已经存在，如果是，获取它并在添加新的之前减少引用。我还没有这样做，因为这个功能只用于结束程序，但为了完整起见，它应该到位。

无论如何，正如我在开始时所说，这比它可能涉及的更多。它也只适用于我有一个可以跟踪函数指针的单例。