GeezerVison is showing github.com — Queue shutdown · Issue #96471 · python/cpython
DescriptionAdd 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:
Linked PRs
Activitygvanrossum commented on May 9, 2023 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 commented on May 10, 2023 ContributorAuthor More actions
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 commented on May 20, 2023 ContributorAuthor More actions 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 commented on May 20, 2023 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()? EpicWink commented on May 22, 2023 ContributorAuthor More actions
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, and queue.put) are all unblocked
Edit: notify -> notify_all gvanrossum commented on May 22, 2023 Not sure if join should be unconditionally unblocked. What if there's a lagging thread that got an item from the queue before the shutdown happened and is still working on it? It will eventually call task_done(). EpicWink commented on May 22, 2023 ContributorAuthor More actions
Not sure if join should be unconditionally unblocked. What if there's a lagging thread that got an item from the queue before the shutdown happened and is still working on it? It will eventually call task_done().
You're right, shutdown should only take away from unfinished_tasks as many as it consumes. I'll fix that later today (turns out this makes a few tests hang, investigating) EpicWink commented on Feb 8, 2024 ContributorAuthor More actions After investigation, the issue was the tests calling task_done without ever calling get, breaking my assumptions and making unfinished_tasks go negative. The solution I went with was to never make unfinished_tasks go below zero. Jean-Luc-Picard-2021 commented on Aug 18, 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 whether
my feature request has a duplicate and whether this is the right place. gvanrossum commented on Aug 18, 2026 Jean-Luc-Picard-2021 commented on Aug 21, 2026 Thx, 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 move
to 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 are
used for feature requests and/or roadmaps. Because of cohesion. EpicWink commented on Aug 21, 2026 ContributorAuthor More actions
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.
MetadataMetadataDevelopmentNo branches or pull requestsIssue actions
|