如何在 Python 中向 C 子进程组发送 SIGINT 信号-Python教程-PHP中文网

如何在 Python 中向 C 子进程组发送 SIGINT 信号

本文介绍通过创建独立进程组并捕获 sigint，在 python 主进程中将中断信号广播至所有关联的 c/c++ 子进程，解决信号被延迟处理或丢失的问题。

本文介绍通过创建独立进程组并捕获 sigint，在 python 主进程中将中断信号广播至所有关联的 c/c++ 子进程，解决信号被延迟处理或丢失的问题。

在使用 Python 调用 C/C++ 扩展库（如通过 PyBind11、Cython 或 ctypes）时，若该库内部启动了长期运行的子进程（例如后台服务、计算任务或守护进程），常会遇到一个典型问题：按下 Ctrl+C 后，Python 并未立即响应，而是等到 C++ 代码执行返回后才触发 SIGINT 处理器——此时子进程可能已脱离控制、僵死或继续运行，导致无法优雅终止。

根本原因在于：默认情况下，子进程与父进程共享同一进程组（process group），而 Python 的 signal 模块仅对主 Python 进程本身注册处理器；当 C++ 代码处于阻塞调用（如 wait()、read() 或第三方库的同步 API）时，Python 的信号处理器不会被调度，SIGINT 实际被挂起，直至控制权交还解释器。

✅ 正确解法是显式创建新进程组，并将所有衍生子进程纳入其中，再由 Python 主进程统一捕获并转发信号：

✅ 关键步骤

在 Python 主进程启动早期调用 os.setpgrp()
创建新的会话和进程组（等价于 os.setsid() + os.setpgrp()），确保后续 fork/exec 启动的子进程（包括 C 库内部创建的进程）默认归属该组（前提是 C 库未主动调用 setsid() 或 setpgid(0, 0)）。
注册自定义信号处理器，使用 os.killpg(0, sig) 广播信号
os.killpg(0, sig) 中的 0 表示“当前进程组”，可一次性向整个组内所有进程发送指定信号（如 SIGINT、SIGTERM）。

python学习
Python 编程学习助手 skill，面向零基础用户。当用户表达以下意图时触发： - 学习 Python 基础概念（变量、数据类型、循环、函数、类等） - 请求 Python 练习题或测验 - 请求审查或调试 Python 代码 - 请求 Python 项目实战指导（爬虫、数据分析、Web 开发等入门项目） - 泛化表述如"教我 Python""我想学编程""Python 怎么写""帮我学 Python" 此 skill

下载
避免信号递归与竞态
在处理器中临时忽略同类型信号（signal.signal(sig, signal.SIG_IGN)），防止重复触发导致栈溢出或不可预测行为。

✅ 完整可运行示例

import os
import signal
import sys
import time
from typing import NoReturn

# ? 关键：创建独立进程组（必须在任何子进程启动前调用）
os.setpgrp()

def handle_sigint(signum: int, frame) -> NoReturn:
    # 防止递归：忽略当前信号，避免重复进入 handler
    signal.signal(signum, signal.SIG_IGN)
    print(f"\n[INFO] Received {signal.Signals(signum).name}, forwarding to process group...")

    try:
        # 向当前进程组广播 SIGINT（0 表示当前组）
        os.killpg(0, signal.SIGINT)
    except ProcessLookupError:
        # 进程组可能已空（无存活子进程），属正常情况
        pass
    except PermissionError:
        # 权限不足（罕见，多见于容器或特殊沙箱环境）
        print("[WARN] Cannot send signal to process group (permission denied)")

    sys.exit(0)

# 注册信号处理器
signal.signal(signal.SIGINT, handle_sigint)

# ? 模拟 C 库启动子进程的行为（实际中由 C++ 代码完成）
# 以下仅为演示：Python 中 fork 出子进程模拟 C 子进程
if os.fork() == 0:
    # 子进程：模拟 C 子进程持续运行
    print(f"[CHILD] PID={os.getpid()}, in process group {os.getpgrp()}")
    while True:
        time.sleep(1)
else:
    # 父进程：保持运行，等待 Ctrl+C
    print(f"[PARENT] PID={os.getpid()}, process group={os.getpgrp()}")
    print("Press Ctrl+C to terminate all processes...")
    try:
        while True:
            time.sleep(5)
    except KeyboardInterrupt:
        # 不会到达此处 —— 已由 signal handler 拦截
        pass

? 验证方法：运行后按 Ctrl+C，观察父子进程是否同时退出（可通过 ps -o pid,pgid,comm 查看进程组关系）。

⚠️ 注意事项

os.setpgrp() 必须在任何子进程派生前调用，否则子进程将继承原始进程组。
若 C++ 库内部调用了 setsid()（创建新会话），则子进程将脱离当前进程组 —— 此时需改用其他机制（如共享文件描述符、命名信号量或 IPC 通知）。
os.killpg(0, ...) 对非直接子进程（如孙子进程）同样有效，前提是它们未脱离本组。
在容器（Docker）、systemd 或某些嵌入式环境中，初始进程可能已是 session leader，此时 os.setpgrp() 可能失败（PermissionError），应捕获异常并降级为逐个 kill(pid, sig)（需提前记录 PID）。