`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! ✨ 🍰 ✨