Python subprocess返回错误退出码：并行单元测试脚本问题求助

我来帮你排查下这个问题——从批量静态调度改成动态任务调度后出现subprocess错误退出码，大概率是这几个常见的坑导致的，咱们一个个捋：

1. 先搞清楚：退出码是测试失败还是脚本执行错误？ #

首先别把**测试用例本身失败（合理返回非0退出码）和子进程启动/执行出错（比如命令找不到、权限问题）**搞混了！

建议你先给每个任务加上详细的日志输出，把退出码、标准输出、错误输出都打出来，比如用这段代码替换你原来的subprocess调用：

import subprocess

# 假设task是你的任务命令，task_id是任务标识
result = subprocess.run(task, capture_output=True, text=True, check=False)
print(f"任务[{task_id}]执行完成 - 退出码: {result.returncode}")
if result.stdout:
    print(f"标准输出:\n{result.stdout}")
if result.stderr:
    print(f"错误输出:\n{result.stderr}")

通过日志你能快速区分：比如退出码127是命令不存在，1可能是测试用例不通过，2是命令参数错误，这样能精准定位问题根源。

2. 动态调度的进程管理逻辑可能有漏洞 #

如果你是手动维护活跃进程列表（比如自己跟踪哪些进程在运行，结束后分配新任务），很容易出现这些问题：

• 进程状态判断错误：比如用wait()会阻塞当前线程，导致你漏检其他进程的结束状态；或者误把已经结束的进程当成活跃进程重复分配任务
• 输出缓冲区阻塞：如果测试用例输出大量内容，而你没有及时读取stdout/stderr，子进程会因为缓冲区满而被迫退出，返回异常码

解决方法推荐用subprocess.Popen结合进程状态检查，同时把输出重定向到文件避免阻塞：

import time
import subprocess

proc = subprocess.Popen(task, stdout=open(f"task_{task_id}_log.txt", "w"), 
                       stderr=subprocess.STDOUT)
# 之后在循环里检查进程是否结束
while proc.poll() is None:
    # 可以做些其他事情，比如定期检查其他进程
    time.sleep(0.1)
# 进程结束后获取退出码
returncode = proc.returncode

3. 用线程安全的任务队列替代手动任务分配 #

手动管理任务列表很容易出现索引越界、重复分配、遗漏任务的问题，建议用Python内置的queue.Queue来做任务调度，它是线程安全的，能自动处理任务的分发和完成确认：

import queue
import threading
import subprocess

def worker(task_queue):
    """工作线程：从队列取任务并执行"""
    while not task_queue.empty():
        task_id, task_cmd = task_queue.get()
        try:
            result = subprocess.run(task_cmd, check=False, capture_output=True, text=True)
            print(f"任务[{task_id}]完成，退出码: {result.returncode}")
        except Exception as e:
            print(f"任务[{task_id}]执行异常: {str(e)}")
        finally:
            # 标记任务完成，让队列知道可以等待所有任务结束
            task_queue.task_done()

def main():
    num_workers = 4
    # 构造你的任务列表，这里用unittest命令举例
    tasks = [(i, ["python", "-m", "unittest", f"tests.test_case_{i}"]) for i in range(10)]
    
    # 初始化任务队列
    task_queue = queue.Queue()
    for task in tasks:
        task_queue.put(task)
    
    # 启动工作线程
    threads = []
    for _ in range(num_workers):
        t = threading.Thread(target=worker, args=(task_queue,))
        t.daemon = True
        t.start()
        threads.append(t)
    
    # 等待所有任务完成
    task_queue.join()
    print("所有任务执行完毕！")

if __name__ == "__main__":
    main()

这种方式完全不用你手动跟踪进程状态，队列会自动把任务分发给空闲的工作线程，避免了手动管理的各种错误。

4. 检查系统资源限制 #

动态调度时如果短时间内启动过多进程，可能触发系统的进程数、内存或CPU限制，导致子进程无法正常启动，返回错误码。你可以用resource模块查看当前的进程数限制：

import resource
print("当前系统进程数限制:", resource.getrlimit(resource.RLIMIT_NPROC))

如果限制过低，可以临时调整（需要root权限），或者确保同时运行的进程数严格不超过num_workers。

先按照上面的方法排查，重点先看每个任务的退出码和输出，大概率能快速找到问题所在！

内容的提问来源于stack exchange，提问作者Jose Javier Gonzalez Ortiz