gh-96471: Add asyncio queue shutdown

[EpicWink]

asyncio-only changes from #102499 (which supercedes #96474), updated to match the API introduced by #104750

• Issue: Queue shutdown #96471

---

📚 Documentation preview 📚: https://cpython-previews--104228.org.readthedocs.build/

[EpicWink]

@willingc ready for review

[gvanrossum]

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.

[gvanrossum]

@EpicWink Could you merge main?

[EpicWink]

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.

[gvanrossum]

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.)

[EpicWink]

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)

[gvanrossum]

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

[gvanrossum]

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.)

[EpicWink]

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. Laurie

…

________________________________ From: Guido van Rossum ***@***.***> Sent: Thursday, February 22, 2024 11:00:09 AM To: python/cpython ***@***.***> Cc: Laurie O ***@***.***>; Mention ***@***.***> Subject: Re: [python/cpython] gh-96471: Add asyncio queue shutdown (PR #104228) 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<#104750>.) — Reply to this email directly, view it on GitHub<#104228 (comment)>, or unsubscribe<https://github.com/notifications/unsubscribe-auth/AF72GRNKMPJRCRIJHOB4NUDYU2KBTAVCNFSM6AAAAAAXX3PWIGVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMYTSNJYGQ3TENZYGQ>. You are receiving this because you were mentioned.Message ID: ***@***.***>

[gvanrossum]

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.)

[gvanrossum]

I'll wait reviewing until you've rewritten shutdown, but here's one doc markup nit.

Also, please mark PRs that aren't ready for review as Draft.

[EpicWink]

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

[YvesDup]

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

[EpicWink]

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

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

[EpicWink]

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

[EpicWink]

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

[YvesDup]

I suggest:

q.put_nowait("data1")
q.put_nowait("data2")

[YvesDup]

and testing on "data1".

[EpicWink]

Except when q is an instance of LifoQueue, when the value should be "data2". Is it necessary to test queue ordering in the shutdown tests?

[YvesDup]

I forgot the LifoQueue class. Is this worth commenting on?

[EpicWink]

I have implemented your suggestion, but I have not committed. I am considering it. I don't think it's necessary.

[YvesDup]

The test case (thus including LifoQueue) is implicit. A comment would be enough.

[YvesDup]

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

[gvanrossum]

Yup! This version looks great. Let me know if you agree -- I've removed the DO-NOT-MERGE label, but I'll wait until you and @YvesDup are happy too.

[EpicWink]

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

[YvesDup]

Sorry but I have a doubt, shouldn't this documentation block be rather:

 All blocked callers of :meth:`~Queue.put` and :meth:`~Queue.get` will be unblocked. If *immediate* is true, also unblock callers of :meth:`~Queue.join`.

In event of change, the docstring of the shutdown method must be updated.

[EpicWink]

join callers aren't necessarily even unblocked anyway, if consumers are processing any items. I should probably say that a task is marked as done for each item in the queue if immediate shutdown.

Also, I think the threading queue docs are the same.

[YvesDup]

a task is marked as done for each item in the queue if immediate shutdown.

It's very precise, better.

Also, I think the threading queue docs are the same.

Yes, I commented here so as not to forget (see #117532 (comment)).

English is your native language, I think it'is best if you update documentations and docstrings.

Update: but I can create the follow-up PR.

[EpicWink]

I've made a PR: #117621

[YvesDup]

Docstring to modify depending of agree/disagree on my first remark about `blocked callers'.
Changes could be:

All blocked callers of put() and get() will be unblocked, and
also join() if 'immediate

[YvesDup]

We've worked well together, I'd be nice if you'd mention me.
:-)

[gvanrossum]

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.)

[EpicWink]

@gvanrossum see #117621

[john-parton]

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:

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 some backports. Is there a backports package for asyncio stuff already?