ProcessPoolExecutor队列持续增长问题求助

使用concurrent.futures.ProcessPoolExecutor实现进程池后，发现每次提交新任务时，打印pool._queue_count的值会持续增长，不清楚原因及是否会引发问题。

日志输出 #

2022-12-06 15:37:31,934 - DEBUG | Running automation 'xxx' with internal automation id 'xxx'
2022-12-06 15:37:31,934 - DEBUG | Running automation 'xxx' with internal automation id 'xxx'
2022-12-06 15:37:31,935 - DEBUG | Running automation 'xxx' with internal automation id 'xxx'
2022-12-06 15:37:31,935 - DEBUG | Pool queue size: 329
2022-12-06 15:37:31,935 - DEBUG | Pool processes: {19113: <ForkProcess name='ForkProcess-2' pid=19113 parent=19104 started>, 19114: <ForkProcess name='ForkProcess-3' pid=19114 parent=19104 started>}
2022-12-06 15:37:31,935 - DEBUG | Pool pending work: {328: <concurrent.futures.process._WorkItem object at 0x7f247f7be2e0>}
2022-12-06 15:37:41,946 - DEBUG | Running automation 'xxx' with internal automation id 'xxx'
2022-12-06 15:37:41,946 - DEBUG | Running automation 'xxx' with internal automation id 'xxx'
2022-12-06 15:37:41,946 - DEBUG | Running automation 'xxx' with internal automation id 'xxx'
2022-12-06 15:37:41,947 - DEBUG | Pool queue size: 330
2022-12-06 15:37:41,947 - DEBUG | Pool processes: {19113: <ForkProcess name='ForkProcess-2' pid=19113 parent=19104 started>, 19114: <ForkProcess name='ForkProcess-3' pid=19114 parent=19104 started>}
2022-12-06 15:37:41,947 - DEBUG | Pool pending work: {329: <concurrent.futures.process._WorkItem object at 0x7f247f7be6a0>}

注意到池队列大小每次提交任务递增1，数值偏高但测试阶段自动化任务无实际处理内容。

简化代码片段 #

futures = []
with mp.Manager() as manager:
    last_execution = time.perf_counter()
    pool = ProcessPoolExecutor()
    while True:
        current_time = time.perf_counter()
        if current_time - last_execution < 10 and not first_run:
            time.sleep(1)
        else:
            last_execution = current_time
            for automation_file in automation_files:
                with open(automation_file, "r") as f:
                    automation_config = json.load(f)
                automation_name = os.path.splitext(os.path.basename(automation_file))[0]
                automation_log = os.path.join(log_dir, f"{automation_name}.log")
                automation_type = automation_config["type"]
                if automation_type == "task":
                    automation = pyba.AutomationTask(automation_name, automation_config, automation_log, api_1, api_2)
                else:
                    logger.error(f"Unknown automation type in '{os.path.basename(automation_file)}', skipping")
                    continue
                logger.debug(f"Running automation '{automation.name}' with internal automation id '{automation._id}'")
            future = pool.submit(automation.run, args=(session_1, session_2, stop_app_event))
            futures.append(future)
            logger.debug(f"Pool queue size: {pool._queue_count}")
            logger.debug(f"Pool processes: {pool._processes}")
            logger.debug(f"Pool pending work: {pool._pending_work_items}")

逻辑是每隔10秒读取多个自动化文件，通过进程池运行任务，但测试阶段无实际任务内容。

系统资源情况 #

-bash-4.2$ ps aux | head -1; ps aux | grep -iE 'python3.9|19104' | grep -v grep | sort -rnk 4
USER       PID %CPU %MEM    VSZ   RSS TTY      STAT START   TIME COMMAND
admin    19114  0.0  0.0 225584 15648 pts/1    S+   14:42   0:00 python3.9 app.py
admin    19113  0.0  0.0 225584 15612 pts/1    S+   14:42   0:00 python3.9 app.py
admin    19107  0.0  0.0 520492 15376 pts/1    Sl+  14:42   0:01 python3.9 app.py
admin    19104  0.0  0.0 374080 20248 pts/1    Sl+  14:42   0:02 python3.9 app.py

服务器为2核，进程池包含2个进程，符合预期。

优雅停止日志 #

2022-12-06 16:16:05,505 - DEBUG | Pool queue size: 560
2022-12-06 16:16:05,506 - DEBUG | Pool processes: {19113: <ForkProcess name='ForkProcess-2' pid=19113 parent=19104 started>, 19114: <ForkProcess name='ForkProcess-3' pid=19114 parent=19104 started>}
2022-12-06 16:16:05,506 - DEBUG | Pool pending work: {559: <concurrent.futures.process._WorkItem object at 0x7f247f738160>}
2022-12-06 16:16:12,516 - DEBUG | Received a signal to stop the app, setting the stop flag
2022-12-06 16:16:12,516 - DEBUG | Cancelling all scheduled pending work
2022-12-06 16:16:12,518 - DEBUG | Shutting down the process pool
2022-12-06 16:16:12,522 - DEBUG | Process pool shut down successfully, app stopped

发送停止信号后应用可立即停止，说明队列数值高但进程未实际处理任务。

1. _queue_count持续增长的原因 #

_queue_count是ProcessPoolExecutor的内部私有计数器，它统计的是从进程池创建以来提交的总任务数，而非当前待处理的任务队列长度。每次调用pool.submit()，这个计数器就会加1，不管任务是否已经执行完成，所以它会随着提交次数持续增长，这是正常的内部行为。

从日志也能看出，pool._pending_work_items的长度始终是1，这才是当前真正等待执行的任务数，而_queue_count只是累计提交数，和实际队列积压无关。

2. 任务看似未处理的原因 #

你的代码存在明显逻辑错误：在for automation_file in automation_files循环中，每次循环都会覆盖automation变量，循环结束后只提交了最后一个automation.run任务，前面的自动化任务并没有被提交到进程池。这就导致日志打印了多个"Running automation"，但实际只提交了一个任务，而这个任务因为测试阶段无实际内容很快执行完毕，所以进程看起来没在处理任务。

3. 是否会引发问题 #

• 单纯_queue_count增长不会引发问题，它只是一个累计计数器，不占用额外资源。
• 但代码逻辑错误会导致大部分自动化任务没有被执行，这是功能性问题，需要修正。
• 另外，把所有future都存在futures列表中，长期运行会导致内存占用增加，因为每个Future对象会保留任务状态和结果，建议定期清理已完成的Future。

4. 代码修正建议 #

把pool.submit()放到for automation_file循环内部，确保每个自动化任务都被提交，同时定期清理已完成的Future：

futures = []
with mp.Manager() as manager:
    last_execution = time.perf_counter()
    pool = ProcessPoolExecutor()
    while True:
        current_time = time.perf_counter()
        if current_time - last_execution < 10 and not first_run:
            time.sleep(1)
        else:
            last_execution = current_time
            for automation_file in automation_files:
                with open(automation_file, "r") as f:
                    automation_config = json.load(f)
                automation_name = os.path.splitext(os.path.basename(automation_file))[0]
                automation_log = os.path.join(log_dir, f"{automation_name}.log")
                automation_type = automation_config["type"]
                if automation_type == "task":
                    automation = pyba.AutomationTask(automation_name, automation_config, automation_log, api_1, api_2)
                else:
                    logger.error(f"Unknown automation type in '{os.path.basename(automation_file)}', skipping")
                    continue
                logger.debug(f"Running automation '{automation.name}' with internal automation id '{automation._id}'")
                # 将submit放入循环内部，每个任务都提交到进程池
                future = pool.submit(automation.run, args=(session_1, session_2, stop_app_event))
                futures.append(future)
            # 定期清理已完成的Future，避免内存泄漏
            completed_futures = [f for f in futures if f.done()]
            for f in completed_futures:
                # 可在此处理任务结果，比如调用f.result()
                futures.remove(f)
            logger.debug(f"Pool queue size: {pool._queue_count}")
            logger.debug(f"Pool processes: {pool._processes}")
            logger.debug(f"Pool pending work: {pool._pending_work_items}")

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