python / python/cpython

`ProcessPoolExecutor` fails to notify when called with `shutdown(cancel_futures=True)`

未关闭
#136,655 5 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看

还没有人认领这个 Issue。

stdlib topic-multiprocessing type-bug
主要语言
Python
星标
77.2k
派生
35.9k
PR 合并指标
PR 指标待抓取

描述

Bug report

Bug description:

I am running a small processor that handles large tasks. When a task encounters an exception I want to store all errors (including other tasks errors) and cancel any other queued work.

However, it seems like the ProcessPoolExecutor's _ExecutorManagerThread does not notify any waiting threads of the future cancelations: making the idiom as_completed hang indefinitely.

Minimal Reproducable Example

import concurrent.futures
import time


def task(n: int) -> int:
    if n == 2:
        raise Exception("Not gonna do it")
    else:
        time.sleep(0.1)
        return n


def main() -> None:
    with concurrent.futures.ProcessPoolExecutor(
        max_workers=2,
    ) as executor:
        futures = [executor.submit(task, i) for i in range(1, 16)]

        for future in concurrent.futures.as_completed(futures):
            try:
                result = future.result()
            except Exception as e:
                print(f"Exception: {e}")
                executor.shutdown(wait=False, cancel_futures=True)
                continue
            print(f"Result: {result}")

            # Look for 'CANCELLED' here:
            print("Other futures states:", [f._state for f in futures])


if __name__ == "__main__":
    main()

Workaround

A workaround is to break after the first future that completed with a failure and do a post hoc gathering of other exceptions of non-cancelled futures.

Suggested Fix

In https://github.com/python/cpython/blob/a68ddea3bf7e9bb77d096c613bce2ec1e67a28f4/Lib/concurrent/futures/process.py#L520-L522

One could replace it with:

for work_id, work_item in self.pending_work_items.items():
    canceled = work_item.future.cancel()
    if canceled:
        work_item.future.set_running_or_notify_cancel()
    else:
        new_pending_work_items[work_id] = work_item

Similar like:

CPython versions tested on:

3.13

Operating systems tested on:

macOS

Linked PRs
  • gh-140021

贡献指南

打开贡献指南

从这里开始

  1. 先读完整个 Issue,再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库,在一个分支上完成修改。
  4. 提交 Pull Request,并在描述里引用这个 Issue 编号。

调研方向

从 Lib/concurrent/futures/process.py 第 520–522 行附近开始,使用最小的 ProcessPoolExecutor 示例重现挂起。将提议的行为与 PR 134618 进行比较,然后验证 shutdown(cancel_futures=True) 会通知已取消的 future,使 as_completed 不会挂起。

由索引模型根据 Issue 内容生成。

评估

技术栈
python
领域
backend
Issue 类型
缺陷
难度
3/5
预计耗时
1-2 天
活跃度
停滞
描述清晰度
基本清楚
新手友好度
38/100

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。