ProcessPoolExecutor.map may hang forever when the submitted iterable is far greater than max_tasks_per_child

[alirzaev]

Bug report

Bug description:

Summary

ProcessPoolExecutor.map may hang forever when the submitted iterable is far greater than max_tasks_per_child

Internally ProcessPoolExecutor.map submits to the executor all items from the iterable (via submit). The worker process exits when the first max_tasks_per_child work items are processed but a new worker process isn't created because ProcessPoolExecutor._adjust_process_count can be invoked only in submit

Reproducible example

import time
from concurrent.futures import ProcessPoolExecutor


def work(i):
    time.sleep(1)

    return i * i


def main():
    with ProcessPoolExecutor(max_workers=1) as executor:
        _ = list(executor.map(work, range(100)))
        # completes

    with ProcessPoolExecutor(max_workers=1, max_tasks_per_child=5) as executor:
        _ = list(executor.map(work, range(100)))
        # hangs forever


if __name__ == '__main__':
    main()

Expected behavior

Program exits

Actual behavor

Program hangs forever

CPython versions tested on:

3.11

Operating systems tested on:

Windows

[cwindolf]

I'm observing the same issue (using the same test case, with some added prints to make sure the first map completes) on Python 3.11.4 and Python 3.12.1 on Linux as well. Using range(10) is enough to see the issue. By adding print(f"work {i=}") inside the work function, I can observe that the second map only makes it to i=4, indicating that only one child process was created.

In other words, the test case:

import time
from concurrent.futures import ProcessPoolExecutor


def work(i):
    print(f"work {i=}", flush=True)
    time.sleep(1)
    return i * i


def main():
    print("main")
    with ProcessPoolExecutor(max_workers=1) as executor:
        print("start A", flush=True)
        _ = list(executor.map(work, range(10)))
        # completes
    print("done A", flush=True)

    with ProcessPoolExecutor(max_workers=1, max_tasks_per_child=5) as executor:
        print("start B", flush=True)
        _ = list(executor.map(work, range(10)))
        # hangs forever
    print("done B", flush=True)


if __name__ == '__main__':
    main()

produces the output

$ python test.py
main
start A
work i=0
work i=1
work i=2
work i=3
work i=4
work i=5
work i=6
work i=7
work i=8
work i=9
done A
start B
work i=0
work i=1
work i=2
work i=3
work i=4

and hangs.