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

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:

• https://github.com/python/cpython/pull/134618

CPython versions tested on:

3.13

Operating systems tested on:

macOS

Linked PRs

• gh-140021

Bắt đầu tại Lib/concurrent/futures/process.py, khoảng dòng 520–522, và tái hiện tình trạng treo bằng ví dụ ProcessPoolExecutor tối giản. So sánh hành vi được đề xuất với PR 134618, sau đó xác minh rằng shutdown(cancel_futures=True) thông báo cho các future đã bị hủy để as_completed không bị treo.