使用ThreadPool/ProcessPoolExecutor批改作业首次运行超时问题求助

问题分析 #

首次运行测试超时、第二次正常的核心原因是：首次运行时，200份作业的并行测试请求瞬间触发大量子进程启动、Python解释器初始化、依赖加载等操作，导致系统CPU、内存、磁盘IO资源过载；而第二次运行时，系统缓存（文件系统缓存、Python字节码缓存）已经生效，资源开销大幅降低，因此速度正常。

针对疑问的解决方案 #

1. 必须限制并行任务数 #

当前设置threads = int(os.cpu_count() * 1.5)完全不适用于200份作业的场景——即使是线程池，每个线程都会启动一个独立的subprocess执行测试，瞬间200+进程会直接把系统资源打满。

• 建议将并行数设置为CPU核心数的1倍或更低（比如threads = max(1, os.cpu_count())），对于IO密集型的测试场景，最多不超过CPU核心数的1.2倍。
• 如果系统内存较小（<16G），可以进一步降低并行数（比如CPU核心数的50%），避免内存不足导致的进程阻塞。

2. 修改子进程创建与预热方式 #

之前的compileall预热只解决了字节码编译问题，未覆盖首次运行的核心瓶颈：

• 提前预热虚拟环境：在创建虚拟环境后，执行一次空的Python命令（python -c ""），触发解释器初始化和基础缓存，避免测试时重复执行这些操作。
• 动态调整超时时间：首次运行时将subprocess.run的timeout临时调整为120秒，后续运行恢复为59秒，给系统足够的初始化时间。
• 复用线程池：当前代码每个task都创建新线程池，改为全局复用一个线程池，减少线程初始化开销。

3. 更优的多虚拟环境并行方案 #

• 分组批量处理：将200份作业分成若干组（比如每组15-20份），串行处理每组，并行执行组内作业，平衡并行效率和系统负载。
• 进程池+线程池组合：如果测试是CPU密集型，用ProcessPoolExecutor（进程数等于CPU核心数）替代线程池，避免GIL限制；如果是IO密集型，保持线程池但严格控制并行数。
• 复用Python解释器进程：如果所有作业依赖一致，可考虑用pexpect或subprocess.Popen保持Python进程存活，多次执行测试命令，减少进程启动的重复开销（注意作业间的文件隔离，避免污染）。

代码调整示例 #

修改并行数与添加预热逻辑 #

def grade_all_submissions(tasks: list, submissions_root: Path) -> None:
    # 保守设置并行数，适配200份作业的场景
    threads = max(1, os.cpu_count())

    for task in tasks:
        config = TASK_CONFIG.get(task)
        if not config:
            continue

        submissions = [
            submission for submission in submissions_root.iterdir()
            if submission.is_dir() and submission.name.startswith("Portfolio")
        ]

        # 预热所有虚拟环境的Python解释器
        for submission in submissions:
            # 根据系统调整虚拟环境Python路径
            venv_python = submission / "venv" / "bin" / "python"  # Linux/macOS
            # venv_python = submission / "venv" / "Scripts" / "python.exe"  # Windows
            if venv_python.exists():
                # 执行空命令触发初始化
                subprocess.run([str(venv_python), "-c", ""], capture_output=True, timeout=10)

        with ThreadPoolExecutor(max_workers=threads) as executor:
            future_to_submission = {
                executor.submit(grade_single_submission, task, submission): submission
                for submission in submissions
            }

            for future in as_completed(future_to_submission):
                submission = future_to_submission[future]
                try:
                    future.result()
                except Exception as e:
                    print(f"Error in {submission.name} for {task}: {e}")

动态调整超时的run_python修改 #

def run_python(self, args, cwd, is_first_run=False) -> str:
        pythonPath = str(self.get_python_path())
        command = [pythonPath] + args
        # 首次运行放宽超时限制
        timeout = 120.0 if is_first_run else 59.0
        result = subprocess.run(
            command,
            capture_output=True,
            text=True,
            cwd = str(cwd) if cwd else None,
            timeout=timeout
        )
        return result.stdout + result.stderr

额外建议 #

• 监控系统负载：首次运行时用top（Linux）或任务管理器（Windows）查看CPU、内存、磁盘IO使用率，确认是否资源耗尽。
• 细化日志：在grade_single_submission中记录每个作业的开始、结束时间，定位具体耗时环节。
• 隔离作业文件：确保每份作业的测试目录独立，避免文件锁或读写冲突导致的阻塞。

内容的提问来源于stack exchange，提问作者Ebrahim Al-Aghbari