如何中断休眠的Python进程？自定义超时上下文管理器失效排查

核心原因分析 #

你的超时管理器在time.sleep、requests.get这类阻塞系统调用场景下失效，本质是Python信号处理机制与系统调用的重启行为导致的：

1. Python解释器仅在执行字节码的间隙检查并处理信号，当代码阻塞在系统调用（如sleep、socket连接）时，解释器暂停运行，信号要等系统调用自然完成后才会被处理。
2. 更关键的是，Python的signal.signal()默认会给信号处理函数加上SA_RESTART标志——这意味着被信号中断的系统调用会自动重启，而不是提前返回。你的信号确实发送了，但系统调用被中断后立刻重新执行，直到自然结束才会回到Python层面触发CustomInterrupt。

对应你的观察：

• 无限循环能生效，是因为它一直在执行Python字节码，解释器会频繁检查信号，超时后能立刻触发异常。
• Ctrl+C（SIGINT）能终止程序，是因为Python对SIGINT的处理默认未设置SA_RESTART，信号会直接中断系统调用并返回。

改进方案1：修改信号处理标志（线程方案保留） #

通过signal.siginterrupt()取消自定义信号的SA_RESTART行为，让系统调用被信号中断后直接返回，Python就能立刻处理信号并抛出异常。

修改后的完整代码：

import signal
import contextlib
from threading import Thread
from _thread import interrupt_main  # 3.10+ 支持
import time


class CustomInterrupt(Exception):
    SIG = 63  # Linux下可用的自定义信号


def handle_custom_interrupt(signum, _):
    raise CustomInterrupt


signal.signal(CustomInterrupt.SIG, handle_custom_interrupt)
# 关键修改：让该信号可以中断系统调用，不自动重启
signal.siginterrupt(CustomInterrupt.SIG, True)


@contextlib.contextmanager
def timeout(seconds: float = 3.0):
    class TimedOut:
        def __init__(self, seconds_: float):
            self.seconds: float = seconds_
            self.state: bool | None = None
            self.thread = Thread(target=self._task)

        def _task(self):
            time.sleep(self.seconds)
            if self.state is None:
                interrupt_main(CustomInterrupt.SIG)

        def start(self):
            self.thread.start()

        def __bool__(self):
            if self.state is None:
                raise ValueError("Can't decide yet, value is only meaningful on exit.")
            return self.state

    timed_out = TimedOut(seconds)
    try:
        timed_out.start()
        yield timed_out
        timed_out.state = False
    except CustomInterrupt:
        timed_out.state = True
    finally:
        timed_out.thread.join()

测试验证 #

• 中断休眠场景：

  with timeout(1) as t_o:
      time.sleep(5)
  print(f"escaped! timed out: {bool(t_o)}")  # 1秒后触发超时，而非5秒
  

• 中断无响应请求场景：

  import requests
  with timeout(1) as t_o:
      requests.get("http://www.google.com:81/")
  print(f"escaped! timed out: {bool(t_o)}")  # 1秒后触发超时，而非130秒
  

备选方案：使用signal.alarm()（无线程实现） #

如果不想依赖线程，可以用进程级的signal.alarm()，它会在指定时间后发送SIGALRM信号，Python对该信号的处理默认不会重启系统调用：

import signal
import contextlib

class TimeoutError(Exception):
    pass

def handle_alarm(signum, frame):
    raise TimeoutError

@contextlib.contextmanager
def timeout(seconds):
    original_handler = signal.signal(signal.SIGALRM, handle_alarm)
    try:
        signal.alarm(seconds)
        yield
    except TimeoutError:
        pass
    finally:
        signal.alarm(0)  # 取消未触发的闹钟
        signal.signal(signal.SIGALRM, original_handler)

注意：alarm()是进程级的，同一时间只能有一个闹钟生效，不适合嵌套或多并发超时场景。

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