concurrent.futures.Executor.map does not cancel the currently waited on future

[graingert]

Bug report

import contextlib
import functools
import concurrent.futures
import threading
import sys


def fn(num, stop_event):
    if num == 1:
        stop_event.wait()
        return "done 1"

    if num == 2:
        return "done 2"


def main():
    stop_event = threading.Event()
    log = []

    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:

        def print_n_wait(ident):
            log.append(f"{ident=} started")
            try:
                stop_event.wait()
            finally:
                log.append(f"{ident=} stopped")

        fut = pool.submit(print_n_wait, ident="first")
        try:
            with contextlib.closing(pool.map(print_n_wait, ["second", "third"], timeout=1)) as gen:
                try:
                    next(gen)
                except concurrent.futures.TimeoutError:
                    print("timed out")
                else:
                    raise RuntimeError("timeout expected")
        finally:
            stop_event.set()

    assert log == ["ident='first' started", "ident='first' stopped"], f"{log=} is wrong"

if __name__ == "__main__":
    sys.exit(main())

result in:

timed out
Traceback (most recent call last):
  File "/home/graingert/projects/executor_map.py", line 45, in <module>
    sys.exit(main())
  File "/home/graingert/projects/executor_map.py", line 42, in main
    assert log == ["ident='first' started", "ident='first' stopped"], f"{log=} is wrong"
AssertionError: log=["ident='first' started", "ident='first' stopped", "ident='second' started", "ident='second' stopped"] is wrong

Your environment

• CPython versions tested on:
• Operating system and architecture:

[ambv]

Fixed in 3.10 - 3.12. Thanks, Thomas! ✨ 🍰 ✨