GeezerVison is showing github.com — gh-96471: Add multiprocessing.Queue.shutdown() by YvesDup · Pull Request #138828 · python/cpython
Conversation
YvesDup
commented
Sep 12, 2025
•
edited by gpshead
Loading
Uh oh!
There was an error while loading. Please reload this page.
This adds shutdown(immediate=False) to multiprocessing.Queue and multiprocessing.JoinableQueue, with the behaviour of queue.Queue.shutdown() (gh-104750) and asyncio.Queue.shutdown() (gh-104228), applied in every process which uses the queue.
Behaviour
After shutdown():
put() raises queue.ShutDown. Callers blocked in put() are unblocked and raise it too.
get() returns the items already put, including those which a process has put but which its feeder thread has not yet written to the pipe. Once the queue is empty, get() raises queue.ShutDown, and callers blocked in get() are unblocked.
- With
immediate=True, the remaining items are discarded instead, in the pipe and in every process's feeder thread buffer. For a JoinableQueue, the unfinished task count goes down by the number of items discarded, so tasks which a worker has already taken still need their task_done().
shutdown() may be called more than once, and an immediate shutdown may follow a graceful one.
Implementation details from Claude
Two constraints shape it: ctypes is not available on every platform, so
no multiprocessing.Value, and sem_getvalue() is broken on macOS, so a
semaphore can only be tested for zero. The state is four SemLocks:
- a flag for "shut down" and a flag for "shut down immediately";
- a count of the items accepted by
put() in any process and not yet
received. This is what tells a getter that the queue is really empty,
rather than that the pipe happens to be empty;
- a lock. On macOS the zero test acquires and releases the semaphore, so two
tests at the same moment can give a wrong answer. Every test whose answer
matters is made under this lock, as is the "check the flag, count the
item" step of put().
Nothing needs to know how many processes are waiting:
- Getters.
shutdown() sends one empty message through the pipe, which
can never be a valid pickle. The getter which reads it sends it back for
the next one, and raises ShutDown only if nothing is outstanding;
otherwise it waits for the item which is still on its way.
- Putters.
shutdown() releases the queue's semaphore once if the queue
is full. A putter which acquires it and finds the queue shut down releases
it again, which wakes the next one.
- Immediate. The caller discards its own buffer and what is in the pipe.
Feeder threads in other processes discard their buffers when they see the
flag. On Unix the caller holds the write lock while it empties the pipe
and leaves the marker in it, so an item which a feeder thread had already
decided to send lands behind the marker and is thrown away by the next
getter.
A queue which never calls shutdown() pays for one lock and two semaphore
operations per put(), and one semaphore operation per get(). A
subclass which overrides put() without counting its items keeps working.
Limitations (documented)
- A getter is woken by the getter before it. If a process is killed, or exits without flushing its feeder thread, while it is being woken, the getters behind it stay blocked.
- If a process is killed, or exits without flushing its feeder thread, after putting items, the queue may never be considered empty and
get()
may block instead of raising ShutDown.
- On Windows there is no write lock, so an item which another process is writing while
shutdown(immediate=True) runs may still be returned by a later get().
Performance
Cost per item for a queue which is never shut down, producer process to consumer process, small items:
|
main |
this PR |
| Linux |
1.95 µs |
2.1–2.3 µs |
| macOS |
3.5 µs |
4.6 µs |
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Docs is not updated, so ref to `shutdown`method is wrong.
|
Thank you! Unfortunately I don't know the multiprocessing code at all and I can't offer to review this. Maybe you can ask for help in the core development category on discuss.python.org.
|
YvesDup reacted with thumbs up emoji
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
+ add ``Queue.shutdown` method,
+ update ``Queue.get``and ``Queue.put`` methods
+ add versionadded to ``get``and ``put``
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Replace the pending-getter and pending-putter counters with a design
which does not need to know who is waiting:
* A count of the items accepted by put() in any process and not yet
received tells a getter whether the queue is really empty, including
the items still in another process's feeder thread buffer.
* shutdown() writes a single empty message to the pipe. The getter
which reads it writes it back, so every getter is woken in its turn,
and raises ShutDown only when nothing is outstanding.
* shutdown() releases the queue's semaphore once if the queue is full.
A putter which acquires it and finds the queue shut down releases it
again, waking the next one.
* shutdown(immediate=True) discards the items in the pipe and in every
feeder thread buffer, accounting for each of them. The unfinished
task count of a JoinableQueue only goes down by the number of items
discarded, as in queue.Queue.
Items put just before shutdown() are no longer lost, getters and
shutdown() itself no longer hang, and calling shutdown() again is
allowed, as in queue.Queue.
The tests are rewritten. Most of them also run on queue.Queue, through
the 'threads' and 'manager' types, to check the behaviour is the same.
Co-Authored-By: Claude Fable 5.2 <[email protected]>
gpshead
changed the title
gh-96471: Add multiprocessing queue shutdown
gh-96471: Add multiprocessing.Queue.shutdown()
Nov 12, 2026
|
Thanks for your patience on this one.
While reviewing I found a few cases the pending-getter/putter counters could not handle, for example:
q.put('x'); q.shutdown(); q.get() could raise ShutDown and lose 'x';
- a consumer already blocked in
get() lost the items put just before shutdown();
- several getters draining a shut-down queue, or
shutdown(immediate=True) with a waiting getter, could hang.
These came from the design rather than from individual bugs, so rather than ask you for another round I pushed a5bf5f30ffc to your branch, which reimplements shutdown() on top of your work (thanks Claude Fable 5.2!); the PR description was updated to detail the approach.
Your docs, and NEWS entry are kept, the tests are rewritten to also cover the cases above.
Reviews welcome.
|
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Reviewers
AA-Turner
Awaiting requested review from AA-Turner
AA-Turner is a code owner
gpshead
Awaiting requested review from gpshead
gpshead is a code owner
3 more reviewers
Reviewers whose approvals may not affect merge requirements
Uh oh!
There was an error while loading. Please reload this page.
Milestone
No milestone
Development
Successfully merging this pull request may close these issues.
Uh oh!
There was an error while loading. Please reload this page.
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
|