如何判断Pdb调试的Python进程是运行中还是等待stdin输入？

核心思路 #

要区分两种状态，关键在于：

• Pdb等待命令时，会输出(Pdb) 提示符，此时子进程处于等待Pdb命令输入的状态
• 被调试程序等待stdin输入时，子进程处于等待程序输入的状态，此时Pdb被挂起，不会响应调试命令

采用超时检测+试探性命令的方案，无需修改被调试程序，通用性更强：在读取不到输出时，发送无副作用的echo命令，若收到(Pdb) 提示符则是Pdb等待状态；若无响应则是程序在等待输入。

修改后的调试代码 #

import os, fire, re
from subprocess import Popen, PIPE, TimeoutExpired

def wait(proc, timeout):
    try:
        return proc.wait(timeout) is not None
    except TimeoutExpired:
        return False

def send(proc, msg):
    proc.stdin.write((msg + '\n').encode())
    proc.stdin.flush()

def recv_nonblock(proc, timeout=0.5):
    """非阻塞读取子进程输出，超时返回已读取内容"""
    os.set_blocking(proc.stdout.fileno(), False)
    output = []
    start_time = os.times()[4]
    while os.times()[4] - start_time < timeout:
        try:
            chunk = proc.stdout.read(1024)
            if not chunk:
                break
            output.append(chunk.decode())
        except BlockingIOError:
            continue
    os.set_blocking(proc.stdout.fileno(), True)
    return ''.join(output)

def is_pdb_waiting(proc):
    """判断子进程是否处于Pdb等待命令的状态"""
    send(proc, 'echo')  # 发送无副作用的Pdb命令
    response = recv_nonblock(proc, timeout=0.3)
    return response.endswith('(Pdb) ')

def parse_array(s):
    s = s.strip('[]')
    return [item.strip() for item in s.split(',') if item.strip()]

def get_code_status(proc):
    if not is_pdb_waiting(proc):
        return None, None  # 不是Pdb状态，无法获取变量
    
    send(proc, 'locals().keys()')
    output = recv_nonblock(proc)
    m = re.match(r'.*dict_keys\((\[.*\])\).*', output)
    if not m or len(m.groups()) != 1:
        return None, None
    
    names = []
    for var in parse_array(m.group(1)):
        stripped = var.strip("'")
        if not re.match(r'__.*__', stripped):
            names.append(f'{var}:{stripped}')
    
    if names:
        send(proc, '{' + ','.join(names) + '}')
        status = recv_nonblock(proc).splitlines()[0]
        send(proc, 'll')
        code = recv_nonblock(proc)
        # 清理最后残留的(Pdb)提示符
        if code.endswith('(Pdb) '):
            code = code[:-len('(Pdb) ')]
        return code, status
    return None, None

def main(code, data, output):
    code_path = os.path.abspath(code)
    proc = Popen(
        ["python", "-m", "pdb", code_path],
        stdin=PIPE,
        stdout=PIPE,
        stderr=PIPE,
        text=False  # 保持字节流处理，避免编码问题
    )
    wait(proc, 1)  # 等待Pdb初始化
    
    with open(data, 'r') as f:
        input_lines = f.read().splitlines()
    
    prev_line = 0
    for step in range(100000):
        output_content = recv_nonblock(proc)
        
        if (step + 1) % 10 == 0:
            code_snippet, status = get_code_status(proc)
            if code_snippet:
                print("当前代码片段：")
                print(code_snippet)
                print("变量状态：")
                print(status)
            break
        
        if not output_content:
            # 无输出，判断是程序等待输入还是Pdb等待命令
            if is_pdb_waiting(proc):
                # Pdb等待状态，执行步进命令
                print("Pdb等待命令，执行步进...")
                # 解析当前代码位置
                send(proc, 'where')
                where_output = recv_nonblock(proc)
                cmd = 's'
                for line in where_output.splitlines():
                    if line.startswith('> '):
                        m = re.match(r'> (.*)\((\d+)\).*', line)
                        if m and m.group(1) == code_path:
                            cur_line = int(m.group(2))
                            cmd = 'r' if prev_line == cur_line else 's'
                            prev_line = cur_line
                        break
                send(proc, cmd)
                print(f"发送命令：{cmd}")
            else:
                # 程序等待输入，发送预设数据
                if not input_lines:
                    print("无更多输入数据，退出循环")
                    break
                send_line = input_lines.pop(0)
                print(f"发送程序输入：{send_line}")
                send(proc, send_line)
        else:
            print(output_content, end='')
            # 如果输出包含(Pdb)提示符，说明处于调试状态
            if output_content.endswith('(Pdb) '):
                # 解析代码位置，决定步进命令
                cmd = 'r'
                for line in output_content.splitlines():
                    if line.startswith('> '):
                        m = re.match(r'> (.*)\((\d+)\).*', line)
                        if m and m.group(1) == code_path:
                            cur_line = int(m.group(2))
                            cmd = 'r' if prev_line == cur_line else 's'
                            prev_line = cur_line
                        break
                send(proc, cmd)
                print(f"发送命令：{cmd}")

if __name__ == '__main__':
    fire.Fire(main)

关键改进点 #

• 新增is_pdb_waiting函数：通过echo命令检测(Pdb) 提示符，精准判断Pdb等待状态
• 改进recv_nonblock函数：带超时的非阻塞读取，避免无输出时的永久阻塞
• 优化分支逻辑：明确区分无输出场景的两种状态，分别处理调试命令与程序输入
• 增强鲁棒性：添加命令响应校验，避免正则匹配失败导致的异常

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