Queue shutdown

[EpicWink]

Add a shutdown method to queue class (threading queue, multiprocessing, asyncio) which causes all future puts to raise (a queue.QueueShutdown) and all future gets once the queue is empty to also raise, unblocking all waiters. An optional argument immediate=True will skip the requirement for the queue to be empty.

This will enable producers and consumers to use the queue to know when to stop. This is important because both producers and consumers can be blocked waiting on the queue.

Previous discussion:

• queue termination on Discourse
• Add a shutdown method to threading and asyncio Queue classes #81515

Linked PRs

• gh-96471: Add queue shutdown #96474
• gh-96471: Add queue shutdown, next step. #102499
• gh-96471: Add threading queue shutdown #104225
• gh-96471: Add asyncio queue shutdown #104228
• gh-96471: Add multiprocessing queue shutdown #104230
• gh-96471: Add threading queue shutdown #104750
• Revert "gh-96471: Add shutdown() method to queue.Queue (#104750)" #115265
• gh-96471: Correct docs for queue shutdown raises #115838
• gh-96471: Add ShutDown to threading module '__all__' #116699
• gh-96471: Correct documentation for asyncio queue shutdown #117621

[gvanrossum]

I wonder if we need a more complete specification and motivation before starting to approve PRs. Just reading the docs for the asyncio PR (#104228) I have tons of questions, e.g.

• Why do we need immediate=True? Couldn't you get the same effect by calling shutdown() and then depleting the queue by calling get() in a tight loop until it raises? Having two variants causes a fair amount of extra code. (From an early discussion on the topic it appears it's for atomicity, but I'm not too sure we had considered this solution.)
• What should full() and empty() return when the queue is shut down?
• Do we need a new inquiry method (is_alive()?) to tell whether a queue has been shut down?
• When the queue is shut down and empty, should get_nowait() raise QueueShutDown or QueueEmpty?
• How do task_done() and join() interact with shutting down?
• What did I miss? (If there's already a spec somewhere, please link here.)

[EpicWink]

I wonder if we need a more complete specification and motivation before starting to approve PRs.

I agree. There were some things that needed to be considered but weren't discussed in the Discourse thread. I'll open I have opened up a new discussion

Why do we need immediate=True? Couldn't you get the same effect by calling shutdown() and then depleting the queue by calling get() in a tight loop until it raises?

During that tight loop a consumer may finish and get a new item to process, increasing the time before all consumers have exited, especially if the queue size is large.

• What should full() and empty() return when the queue is shut down?
• Do we need a new inquiry method (is_alive()?) to tell whether a queue has been shut down?
• When the queue is shut down and empty, should get_nowait() raise QueueShutDown or QueueEmpty?
• How do task_done() and join() interact with shutting down?

To be included in aforementioned new discussion.

[EpicWink]

Actually, now that I think about it, you could have a tight loop consuming all items for immediate=True, if:

• the loop is in the shutdown method
• state is set to 'shut-down` before the loop
• the lock is held for the entire loop
• everything is notified at the end of the loop

[gvanrossum]

So it would still need the immediate=True flag on shutdown(), but the rest of the code would not have to distinguish between shut-down and immediately-shut-down. That's much better!

I'm not sure I follow "everything is notified at the end of the loop" -- is this about task_done()?