Bug report

Bug description

When buffersize is passed to Pool.imap() or Pool.imap_unordered(), calling close() before the iterator is exhausted makes the iterator stop early. No exception is raised. The items that were not submitted to the workers yet are dropped.

Without buffersize, the same code returns all results.

import multiprocessing as mp


def sq(x):
    return x * x


if __name__ == "__main__":
    for kw in ({}, {"buffersize": 4}):
        pool = mp.Pool(2)
        it = pool.imap(sq, range(20), **kw)
        first = next(it)
        pool.close()
        rest = list(it)
        pool.join()
        print(kw, "got", 1 + len(rest), "of 20")

Output:

{} got 20 of 20
{'buffersize': 4} got 4 of 20

ThreadPool and imap_unordered() are affected in the same way.

Expected: close() is documented as preventing new tasks from being submitted. Items of an imap() call that was already made should still be processed, as they are without buffersize.

Cause: This comes from the fix for gh-155477 (gh-155478). close() now releases the buffersize semaphores, and _guarded_task_generation() stops as soon as it sees self._state != RUN. The task handler then treats the generator as exhausted and calls set_length(), so the iterator ends with a normal StopIteration.

The test added in gh-155478 only checks that close() + join() does not hang. It does not check the number of results.

Proposed fix: Keep throttling after close(), and only stop throttling in join(), where the remaining items are submitted without waiting for the consumer. That keeps gh-155477 fixed and does not lose results.

I would like to work on this. I have a patch with tests and will open a PR.

buffersize was added in gh-64192 and is new in Python 3.16, so no released version is affected.

CPython versions tested on

CPython main branch

Operating systems tested on

macOS

Linked PRs

• gh-158675: Fix Pool.imap(buffersize=...) dropping results after close() #158676