@willingc ready for review

asyncio-only changes from #102499 (on top of #96474)

I assume the "on top of ..." part is obsolete, since that PR was closed without merging. I'll add @willingc as a reviewer per your request.

@EpicWink Could you merge main?

Could you merge main?

Done, but like #104750, we may want to modify the implementation of immediate=True to simply consume the queue rather than having a separate state.

Eh, @EpicWink, did you see the hubbub about the hanging test in your previous PR? The test was disabled because it kept hanging in CI on various platforms. Could you look into what's wrong with it? (At least the feature wasn't reverted.)

Eh, @EpicWink, did you see the hubbub about the hanging test in your previous PR? The test was disabled because it kept hanging in CI on various platforms. Could you look into what's wrong with it?

I did see the follow-on issue and pull-requests. Yves and I are working on fixing it (you can follow the changes here: main...EpicWink:cpython:fix-thread-queue-shutdown-test)

Awesome! (An acknowledgement that you were working on it would have lessened my stress. :-)

Looking at this with an eye towards review, I realized that the threaded-queue implementation doesn't appear to match its documentation: It appears that q.join() does not raise ShutDown when the queue is shutdown immediately. The tests don't seem to be testing for this either. (Come to think of it, the test never seems to take the except self.queue.ShutDown path -- if I put a breakpoint there it never gets hit during any of the tests. The test logic is pretty convoluted, which is my only excuse for not having caught this in the review of gh-104750.)

Now that the immediate-consume implementation of the threading queue shutdown has been accepted, I'm going to do the same here and for multiprocessing. I'll personally rewrite the tests to be more readable and obvious.

And what about queue.Queue.join()? Should it raise after shutdown(immediate=True)? Either way some changes are needed -- either in the implementation or in the docs. And the tests there too (what you wrote above appears to apply to the asyncio and multiprocessing versions only.)

And what about queue.Queue.join()? Should it raise after shutdown(immediate=True)? Either way some changes are needed -- either in the implementation or in the docs. And the tests there too (what you wrote above appears to apply to the asyncio and multiprocessing versions only.)

The behaviour should (and I intend to implement it to) be the same between the threading, multiprocessing and asyncio queues

By the way, should task_done() raise an exception after shutdown or not ?
update; see https://docs.python.org/3.13/library/queue.html#queue.Queue.task_done

By the way, should task_done() raise an exception after shutdown or not ?

@YvesDup No (except the usual ValueError); see #115838

I want to trust that you two have done the right thing there

@gvanrossum please don't trust, have you seen xz-utils's backdoor?

but not push until the big test run completes? (Buildbot tests can take many hours, sorry.)

Oops, sorry. I forgot to hold off my most recent push

ALso, why is the DO-NOT-MERGE label set? Is it because of #104228?

No, I think it was because I hadn't updated the implementation to match the new agreed-upon behaviour (this PR was in draft at that time anyway, so I don't think the label was needed).

wondering why you always wake up all getters -- and came to the conclusion that it is needed

I should probably add a comment

there is the same bug on threading queue shutdown. I will open an issue.

@YvesDup I've created issue #117531, and fix in PR #117532

there is the same bug on threading queue shutdown. I will open an issue.

@YvesDup I've created issue #117531, and fix in PR #117532

I have seen and commented

I think it's good to go

wondering why you always wake up all getters -- and came to the conclusion that it is needed

I should probably add a comment

I could do this, but I don't think it's worth revoking the PR's approval

Sorry for merging prematurely. @EpicWink Can you prepare a follow-up PR? I will merge it as soon as @YvesDup approves it. (Or, alternatively, @YvesDup can make a PR and I will merge when @EpicWink approves.)

@gvanrossum see #117621

I'd like to add that the shutdown() method and task groups makes certain patterns really easy to express. I was struggling with how to properly make a generic async map function with a fixed number of workers (mostly as an exercise), but when I saw the new shutdown method, it came together really easily:

async def async_map[T, R](
    func: Callable[[T], Awaitable[R]],
    iterable: AsyncIterable[T],
    *,
    limit: int,
    maxsize: int = -1,
) -> AsyncIterator[R]:
    if maxsize < 0:
        maxsize = limit

    arguments_queue = Queue[T](maxsize=maxsize)
    results_queue = Queue[R](maxsize=maxsize)

    async def drain():
        async for argument in iterable:
            await arguments_queue.put(argument)
        arguments_queue.shutdown(immediate=False)

    async def worker():
        while True:
            try:
                argument = await arguments_queue.get()
            except QueueShutDown:
                break
            await results_queue.put(await func(argument))

    async def background():
        async with asyncio.TaskGroup() as background_task_group:
            background_task_group.create_task(drain())
            for _ in range(limit):
                background_task_group.create_task(worker())
        results_queue.shutdown(immediate=False)
 
    async with asyncio.TaskGroup() as task_group:
        task_group.create_task(background())

        while True:
            try:
                yield await results_queue.get()
            except QueueShutDown:
                break

I believe this has a common issue that a lot of async generators do, in that if you don't consume the entire generator, it will still continue processing and end up with a lot of futures never awaited.

I know there's some backports. Is there a backports package for asyncio stuff already?