[YvesDup]

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

• Issue: Queue shutdown #96471

[gvanrossum]

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.

[read-the-docs-community]

Documentation build overview

📚 cpython-previews | 🛠️ Build #35827780 | 📁 Comparing a5bf5f3 against main (8270ae5)

  🔍 Preview build  

460 files changed · + 13 added · ± 446 modified · - 1 deleted

+ Added

• builtins/constants.html
• builtins/exceptions.html
• builtins/functions.html
• builtins/index.html
• builtins/stdtypes.html
• builtins/threadsafety.html
• builtins/time-complexity.html
• c-api/pyexpat.html
• library/asyncio-tools.html
• library/time-complexity.html
• and 3 more...

± Modified

• glossary.html
• index.html
• license.html
• py-modindex.html
• _static/tachyon-example-heatmap.html
• c-api/arg.html
• c-api/bool.html
• c-api/buffer.html
• c-api/bytearray.html
• c-api/bytes.html
• and 436 more...

- Deleted

• library/asyncio-policy.html

[gpshead]

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.