Python中time.sleep会阻塞子进程吗？C++子进程测试困境求解

一、先解决子进程IO阻塞的核心问题 #

你遇到的C进程定时输出停止、文件未处理的根本原因，大概率是子进程的stdout缓冲区被填满导致阻塞。你的伪代码里将C程序的stdout直接重定向到文件，当输出内容（比如每秒的"I am Alive"）填满系统缓冲区后，C++的输出函数会被阻塞，进而导致整个进程卡住，无法执行文件处理逻辑。

解决这个问题的关键是让Python实时读取子进程的输出，避免缓冲区积压：

代码示例：用线程读取子进程输出 #

import subprocess
import threading
import time

def read_subprocess_output(process, output_file):
    """线程函数：实时读取子进程的stdout并写入文件"""
    with open(output_file, 'w') as f:
        for line in iter(process.stdout.readline, b''):
            f.write(line.decode())
            f.flush()  # 立即写入文件，避免缓冲区积压

# 启动C++子进程，用PIPE获取stdout
cpppath = "./your_cpp_binary"
cfg = "./config.cfg"
output_log = "SomeFile.txt"

prc = subprocess.Popen(
    [cpppath, "-v", "-c", cfg],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,  # 可选：把stderr也合并到stdout一起处理
    text=True  # 直接按文本模式读取，不用手动decode
)

# 启动线程读取输出
output_thread = threading.Thread(target=read_subprocess_output, args=(prc, output_log))
output_thread.daemon = True  # 线程随主进程退出
output_thread.start()

# 向监控目录添加测试文件
# ... 你的添加文件代码 ...

二、替换固定sleep，使用可靠的等待机制 #

解决IO阻塞后，再替换time.sleep(5)这种不可靠的固定等待，下面是几种可行方案：

方案1：轮询检查目标输出文件 #

等待目标目录中出现预期的处理后文件，并且确认文件已经完全写入（避免C++还在写入时就开始验证）：

import os
import time

def wait_for_processed_file(target_file, timeout=30, check_interval=0.5):
    """等待目标文件出现且内容稳定"""
    start_time = time.time()
    last_size = -1
    while time.time() - start_time < timeout:
        if os.path.exists(target_file):
            current_size = os.path.getsize(target_file)
            # 连续两次检查文件大小不变，说明写入完成
            if current_size == last_size:
                return True
            last_size = current_size
        time.sleep(check_interval)
    return False

# 添加测试文件后调用
target_processed_file = "./output_dir/processed_file.txt"
if not wait_for_processed_file(target_processed_file):
    raise TimeoutError("C++程序未在超时时间内生成处理后的文件")

# 验证文件内容
# ... 你的验证代码 ...

方案2：基于SIGUSR1的无阻塞等待 #

如果坚持用信号机制，可以用Python的signal模块注册信号处理函数，配合短间隔循环等待，同时保证子进程的IO能正常处理（因为之前已经用线程读取输出，不会阻塞）：

import signal

signal_received = False

def handle_sigusr1(signum, frame):
    global signal_received
    signal_received = True

# 注册SIGUSR1信号处理函数
signal.signal(signal.SIGUSR1, handle_sigusr1)

# 添加测试文件后开始等待信号
start_time = time.time()
timeout = 30
while not signal_received and time.time() - start_time < timeout:
    time.sleep(0.1)  # 短间隔sleep，避免长时间阻塞

if not signal_received:
    raise TimeoutError("未收到C++程序发送的SIGUSR1信号")

# 验证文件内容
# ... 你的验证代码 ...

方案3：让C++程序生成标记文件 #

修改C++程序，当完成文件处理后，在指定目录生成一个标记文件（比如processing_done.flag），Python测试用例只需等待这个标记文件出现即可：

def wait_for_flag(flag_file, timeout=30, check_interval=0.5):
    start_time = time.time()
    while time.time() - start_time < timeout:
        if os.path.exists(flag_file):
            # 可选：删除标记文件，避免影响下一次测试
            os.remove(flag_file)
            return True
        time.sleep(check_interval)
    return False

# 添加测试文件后调用
flag_file = "./monitor_dir/processing_done.flag"
if not wait_for_flag(flag_file):
    raise TimeoutError("C++程序未完成文件处理")

# 验证文件内容
# ... 你的验证代码 ...

三、总结 #

1. 优先解决子进程IO阻塞问题：必须实时读取C++程序的输出，避免缓冲区满导致进程卡住；
2. 选择适合的等待机制：轮询目标文件或标记文件实现简单可靠，信号机制适合需要精确同步的场景，但需要C++程序配合发送信号。

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