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 摘要。