Repository navigation
Queue shutdown #96471
Description
Activity
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.)
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 openI have opened up a new discussionWhy 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.
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
shutdownmethod - 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
- the loop is in the
So it would still need the
immediate=Trueflag onshutdown(), 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()?I'm not sure I follow "everything is notified at the end of the loop" -- is this about task_done()?
Basically (for threading):
self.not_empty.notify_all() self.not_full.notify_all() self.all_tasks_done.notify_all()
So consumers (ie callers of
queue.join,queue.get, andqueue.put) are all unblockedEdit:
notify->notify_allNot sure if
joinshould be unconditionally unblocked. What if there's a lagging thread that got an item from the queue before theshutdownhappened and is still working on it? It will eventually calltask_done().Not sure if
joinshould be unconditionally unblocked. What if there's a lagging thread that got an item from the queue before theshutdownhappened and is still working on it? It will eventually calltask_done().You're right,
shutdownshould only take away fromunfinished_tasksas many as it consumes. I'll fix thatlater today(turns out this makes a few tests hang, investigating)After investigation, the issue was the tests calling
task_donewithout ever callingget, breaking my assumptions and makingunfinished_tasksgo negative. The solution I went with was to never makeunfinished_tasksgo below zero.9 remaining items
- added 4 commits that reference this issue
on Apr 17, 2024 - added a commit that references this issue
on Aug 10, 2025 - added a commit that references this issue
on Apr 26, 2026 Would it be possibly to add "shutdown" also to Threading.join etc..
Basically replicating the exception signature throws InterruptedException
from Java. Or what is the Pythonesk approach to get out of Threading.join ?Similarly for Semaphores. Basically extending the concept introduced
here from Queues, to other concurrency elements such as the more
simpler Semaphores and even Threads itself. BTW: Didn't check whethermy feature request has a duplicate and whether this is the right place.
@Jean-Luc-Picard-2021 Please start a discussion on Discord if you want your suggestion to be heard.
Reacted by Laurie OThx, I have 99 problems but not getting heard is none of mine.
Proof: See comment above from former BDFL himself.
But it would be more helpful to hint why I should moveto discord. Does Python not anymore use GitHub, Discourse, etc..?
BTW: I don't have a Discord account, because of social media
inflation, and sometimes ticket systems and not discussions areused for feature requests and/or roadmaps. Because of cohesion.
Does Python not anymore use GitHub
@Jean-Luc-Picard-2021 CPython and most PSF projects use Discourse to discuss proposals. Discord is a quick way to talk to people for design suggestions before creating a Discourse topic, but in my opinion can be skipped if you have a concrete enough proposal to debate with.
Add a
shutdownmethod to queue class (threadingqueue,multiprocessing,asyncio) which causes all future puts to raise (aqueue.QueueShutdown) and all future gets once the queue is empty to also raise, unblocking all waiters. An optional argumentimmediate=Truewill 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:
Linked PRs