@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 ofimmediate=True to simply consume the queue rather than having a separate state.

Eh,@EpicWink, did you see the hubbub about the hanging test in yourprevious 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 yourprevious 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 thatq.join() doesnot raiseShutDown 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 theexcept self.queue.ShutDown path -- if I put a breakpoint there it never gets hit duringany of the tests. The test logic is pretty convoluted, which is my only excuse for not having caught this in the review ofgh-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 aboutqueue.Queue.join()? Should it raise aftershutdown(immediate=True)? Either way some changes are needed -- either in the implementation or in the docs. And the tests there too (what you wrote aboveappears to apply to the asyncio and multiprocessing versions only.)

And what aboutqueue.Queue.join()? Should it raise aftershutdown(immediate=True)? Either way some changes are needed -- either in the implementation or in the docs. And the tests there too (what you wrote aboveappears 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, shouldtask_done() raise an exception aftershutdown or not ?
update; seehttps://docs.python.org/3.13/library/queue.html#queue.Queue.task_done

By the way, shouldtask_done() raise an exception aftershutdown or not ?

@YvesDup No (except the usualValueError); see#115838

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

@gvanrossum please don't trust, have you seenxz-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

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 theshutdown() 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:

asyncdefasync_map[T,R](func:Callable[[T],Awaitable[R]],iterable:AsyncIterable[T],*,limit:int,maxsize:int=-1,)->AsyncIterator[R]:ifmaxsize<0:maxsize=limitarguments_queue=Queue[T](maxsize=maxsize)results_queue=Queue[R](maxsize=maxsize)asyncdefdrain():asyncforargumentiniterable:awaitarguments_queue.put(argument)arguments_queue.shutdown(immediate=False)asyncdefworker():whileTrue:try:argument=awaitarguments_queue.get()exceptQueueShutDown:breakawaitresults_queue.put(awaitfunc(argument))asyncdefbackground():asyncwithasyncio.TaskGroup()asbackground_task_group:background_task_group.create_task(drain())for_inrange(limit):background_task_group.create_task(worker())results_queue.shutdown(immediate=False)asyncwithasyncio.TaskGroup()astask_group:task_group.create_task(background())whileTrue:try:yieldawaitresults_queue.get()exceptQueueShutDown: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 somebackports. Is there a backports package for asyncio stuff already?