Linux下signal.SIGTERM无法终止子进程，仅SIGKILL生效的问题求助

问题描述 #

我通过dict存储子进程实例，希望安全终止处于运行状态的子进程。在Linux环境下，尝试使用p.terminate()或os.kill(p.pid, signal.SIGTERM)均无法终止子进程，仅os.kill(p.pid, signal.SIGKILL)能生效，现寻求安全终止进程的可行方案。

相关代码 #

进程终止逻辑 #

def __remove_progress_without_lock(self, k):
    import os
    import signal
    if k in self.progress_map:
        p = self.progress_map[k]
        if p.is_alive():
            # p.terminate()
            # print("stop 6")
            # p.join()
            os.kill(p.pid, signal.SIGTERM)
            p.join()
            # os.kill(p.pid, signal.SIGKILL)
            # p.join()
            exit_code = p.exitcode
            print(f"stop 7 exit_code:{exit_code}")
        del self.progress_map[k]
        print("stop 8")

其他关联代码 #

def add_progress(self, k, progress):
    with SingletonProgressRecorder.__lock:
        self.progress_map[k] = progress

def test(para: CreatePara):
    print(para)
    for i in range(para.time_s):
        print(f"{para.sign}: {i}")
        time.sleep(1)
    return para

@app.post("/add")
def add_test(para: AddPara):
    try:
        export_recorder = SingletonProgressRecorder()
        rc_args = (para,)
        p = Process(target=test, args=rc_args)
        export_recorder.add_progress(para.sign, p)
        p.start()
        return {"status": "success", "data": "Process started successfully"}
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=repr(e))

@app.post("/stop")
def stop_test(para: StopPara):
    print("stop_test")
    try:
        export_recorder = SingletonProgressRecorder()
        export_recorder.remove_progress(para.sign)
        return {"status": "success", "data": "Process stopped successfully"}
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=repr(e))

问题原因 #

SIGTERM无法终止子进程的核心原因有两个：

1. 子进程中的test函数使用time.sleep(1)长时间阻塞，Python在sleep期间会暂停信号处理，导致SIGTERM无法被及时响应。
2. 子进程未注册SIGTERM的信号处理函数，即便信号被送达，进程也不会主动执行退出逻辑。

解决方法 #

1. 给子进程注册SIGTERM信号处理函数 #

在子进程的执行函数（test）内部注册信号处理逻辑，收到SIGTERM时设置退出标志，让进程主动终止。

2. 替换阻塞式sleep为可中断的等待 #

将time.sleep(1)拆分为多次短时间sleep，每次sleep后检查退出标志，避免长时间阻塞信号处理。

修改后的代码示例 #

子进程执行函数（test）

import signal
import time

exit_flag = False

def handle_sigterm(signum, frame):
    global exit_flag
    exit_flag = True

def test(para: CreatePara):
    # 在子进程内部注册SIGTERM信号处理器
    signal.signal(signal.SIGTERM, handle_sigterm)
    print(para)
    for i in range(para.time_s):
        if exit_flag:
            print(f"{para.sign}: 收到终止信号，主动退出")
            break
        print(f"{para.sign}: {i}")
        # 分段sleep，每0.1秒检查一次退出标志
        for _ in range(10):
            if exit_flag:
                break
            time.sleep(0.1)
    return para

进程终止逻辑

def __remove_progress_without_lock(self, k):
    if k in self.progress_map:
        p = self.progress_map[k]
        if p.is_alive():
            p.terminate()  # 发送SIGTERM信号
            # 设置超时时间，避免子进程无响应
            p.join(timeout=5)
            if p.is_alive():
                print("子进程未响应SIGTERM，执行强制终止")
                p.kill()  # 发送SIGKILL兜底
                p.join()
            exit_code = p.exitcode
            print(f"进程退出码: {exit_code}")
        del self.progress_map[k]

补充说明 #

• 信号处理函数必须在子进程内部注册，父进程的信号配置不会被子进程继承。
• 分段sleep的目的是让子进程有机会在等待期间处理信号，保证SIGTERM能被及时响应。
• 加入超时兜底逻辑，在子进程不响应SIGTERM时再用SIGKILL强制终止，平衡安全性和可靠性。

内容的提问来源于stack exchange，提问作者x xi