如何让Jupyter中断运行ctypes共享库的进程

本质原因是：终端按^C会给整个进程组发送SIGINT信号，Python和底层C代码都能响应；但Jupyter的“中断内核”只会给Python内核进程发SIGINT，而当Python调用C共享库时，GIL（全局解释器锁）会被释放，Python的信号处理函数要等C++代码执行完才能触发，导致无法实时中断。

以下是可行的解决办法：

1. 在C++库中直接处理SIGINT信号 #

在C++代码中添加信号处理逻辑，让耗时任务能响应中断：

• 定义一个全局volatile布尔变量（确保跨线程可见）作为中断标志
• 注册SIGINT信号处理函数，触发时将标志设为true
• 在C++的耗时循环/任务中，定期检查该标志，一旦为true就提前退出并释放资源

示例代码（C++）：

#include <csignal>
#include <atomic>

std::atomic<bool> g_interrupted(false);

void sigint_handler(int signum) {
    g_interrupted = true;
}

// 你的耗时函数
extern "C" void long_running_task() {
    // 注册信号处理
    signal(SIGINT, sigint_handler);
    
    for (int i = 0; i < 1000000; ++i) {
        // 检查中断标志
        if (g_interrupted) {
            // 清理资源
            printf("Task interrupted\n");
            return;
        }
        // 执行耗时操作
        heavy_computation();
    }
}

注意：如果C++代码是多线程的，要确保每个工作线程都能检查中断标志，或者在主线程中统一处理后通知子线程退出。

2. 在Python层用线程包装C++调用 #

将C++库的调用放到子线程中，主线程监听Jupyter的中断信号，主动终止任务：

• 用threading模块创建子线程执行C++调用
• 主线程通过try-except捕获KeyboardInterrupt，然后触发C++库的中断逻辑（比如通过全局变量或回调）

示例代码（Python）：

import threading
import ctypes
import time

# 加载你的C++库
lib = ctypes.CDLL('./your_library.so')

# 定义全局中断标志（要和C++端同步的话，也可以用ctypes共享内存）
interrupted = False

def run_cpp_task():
    global interrupted
    try:
        lib.long_running_task()
    except Exception:
        pass

# 启动子线程
thread = threading.Thread(target=run_cpp_task)
thread.start()

# 主线程等待并监听中断
try:
    while thread.is_alive():
        time.sleep(0.1)
except KeyboardInterrupt:
    # 通知C++库中断（这里假设C++端有设置全局标志的接口）
    lib.set_interrupted(True)
    thread.join()
    print("任务已中断")

3. 给C++库添加Python回调接口 #

在C++的耗时任务中定期调用Python回调函数，让Python有机会处理信号：

• 用ctypes.CFUNCTYPE定义回调类型
• 在C++代码中每隔一段时间调用该回调，此时Python会重新获取GIL，信号处理函数就能被触发

示例代码：

Python端 #

import ctypes

# 定义回调函数类型
CallbackType = ctypes.CFUNCTYPE(None)

def py_callback():
    # 空回调即可，目的是让Python有机会处理信号
    pass

# 加载库并设置回调
lib = ctypes.CDLL('./your_library.so')
lib.set_callback(CallbackType(py_callback))

# 调用耗时函数
try:
    lib.long_running_task()
except KeyboardInterrupt:
    print("任务已中断")

C++端 #

#include <csignal>
#include <atomic>
#include <Python.h>

std::atomic<bool> g_interrupted(false);
void (*g_py_callback)() = nullptr;

void sigint_handler(int signum) {
    g_interrupted = true;
}

extern "C" void set_callback(void (*callback)()) {
    g_py_callback = callback;
}

extern "C" void long_running_task() {
    signal(SIGINT, sigint_handler);
    
    for (int i = 0; i < 1000000; ++i) {
        if (g_interrupted) {
            printf("Task interrupted\n");
            return;
        }
        // 每执行N次操作调用一次Python回调
        if (i % 1000 == 0 && g_py_callback != nullptr) {
            // 确保GIL状态正确
            PyGILState_STATE gstate = PyGILState_Ensure();
            g_py_callback();
            PyGILState_Release(gstate);
        }
        heavy_computation();
    }
}

内容的提问来源于stack exchange，提问作者Ethan Roubenoff