Skip to content

Commit a5bf5f3

Browse files
gpsheadclaude
andcommitted
gh-96471: Reimplement multiprocessing.Queue.shutdown()
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 <noreply@anthropic.com>
1 parent 11d4089 commit a5bf5f3

5 files changed

Lines changed: 651 additions & 584 deletions

File tree

‎Doc/library/multiprocessing.rst‎

Lines changed: 58 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -942,7 +942,7 @@ For an example of the usage of queues for interprocess communication see
942942
standard library's :mod:`queue` module are raised to signal timeouts.
943943

944944
:class:`Queue` implements all the methods of :class:`queue.Queue` except for
945-
:meth:`~queue.Queue.task_done`, :meth:`~queue.Queue.join`.
945+
:meth:`~queue.Queue.task_done` and :meth:`~queue.Queue.join`.
946946

947947
.. method:: qsize()
948948

@@ -979,9 +979,10 @@ For an example of the usage of queues for interprocess communication see
979979
If the queue is closed, :exc:`ValueError` is raised instead of
980980
:exc:`AssertionError`.
981981

982-
Can raise the :exc:`queue.ShutDown` when queue shuts down.
982+
Raises :exc:`queue.ShutDown` if the queue has been shut down.
983983

984-
.. versionadded:: next
984+
.. versionchanged:: next
985+
:exc:`queue.ShutDown` is raised if the queue has been shut down.
985986

986987
.. method:: put_nowait(obj)
987988

@@ -1001,24 +1002,64 @@ For an example of the usage of queues for interprocess communication see
10011002
If the queue is closed, :exc:`ValueError` is raised instead of
10021003
:exc:`OSError`.
10031004

1004-
Can raise the :exc:`queue.ShutDown` when queue shuts down and is empty.
1005+
Raises :exc:`queue.ShutDown` if the queue has been shut down and is
1006+
empty, or if the queue has been shut down immediately.
10051007

1006-
.. versionadded:: next
1008+
.. versionchanged:: next
1009+
:exc:`queue.ShutDown` is raised if the queue has been shut down.
10071010

10081011
.. method:: get_nowait()
10091012

10101013
Equivalent to ``get(False)``.
10111014

1012-
.. method:: shutdown([immediate])
1015+
.. method:: shutdown(immediate=False)
1016+
1017+
Put the queue into a shutdown mode, in every process which uses it.
1018+
1019+
The queue can no longer grow.
1020+
Future calls to :meth:`put` raise :exc:`queue.ShutDown`.
1021+
Currently blocked callers of :meth:`put` will be unblocked
1022+
and will raise :exc:`queue.ShutDown` in the formerly blocked process.
1023+
1024+
If *immediate* is false (the default), the queue can be wound
1025+
down normally with :meth:`get` calls to extract the items which have
1026+
already been put, including those which a process has put but which
1027+
its feeder thread has not yet written to the pipe.
1028+
1029+
Once the queue is empty, future calls to :meth:`get` will raise
1030+
:exc:`queue.ShutDown`. Currently blocked callers of :meth:`get` are
1031+
unblocked and raise :exc:`queue.ShutDown` as well.
1032+
1033+
If *immediate* is true, the items remaining in the queue are discarded
1034+
instead, including those which other processes have put but not yet
1035+
written to the pipe. All callers of :meth:`get` are unblocked.
1036+
1037+
Calling :meth:`shutdown` more than once is allowed. A queue which has
1038+
been shut down cannot be put back into service.
10131039

1014-
Shuts down the queue. If optional args *immediate* is ``False``
1015-
(the default), all remaining items could be retrieved until queue is empty.
1016-
If optional args *immediate* is ``True``, all remaining items from the
1017-
queue are erased. When queue is empty, get an item raises
1018-
a :exc:`queue.ShutDown` exception. Adding a new item into the queue
1019-
also raises this same exception.
1040+
Raises :exc:`ValueError` if the queue has been closed in this process
1041+
with :meth:`close`.
10201042

1021-
.. versionadded:: 3.16
1043+
.. note::
1044+
1045+
A getter is woken up by the getter woken before it, so the
1046+
:exc:`queue.ShutDown` exception can only reach all of the blocked
1047+
processes as long as none of them is killed, or exits without
1048+
flushing its feeder thread (see :meth:`cancel_join_thread`), as it
1049+
is being woken.
1050+
1051+
Similarly, if a process is killed or exits without flushing its
1052+
feeder thread after putting items on the queue, the queue may never
1053+
be considered empty and :meth:`get` may block instead of raising
1054+
:exc:`queue.ShutDown`.
1055+
1056+
.. note::
1057+
1058+
On Windows, an item which another process is writing to the pipe
1059+
while ``shutdown(immediate=True)`` is running may still be returned
1060+
by a later :meth:`get`.
1061+
1062+
.. versionadded:: next
10221063

10231064
:class:`multiprocessing.Queue` has a few additional methods not found in
10241065
:class:`queue.Queue`. These methods are usually unnecessary for most
@@ -1122,6 +1163,10 @@ For an example of the usage of queues for interprocess communication see
11221163
Raises a :exc:`ValueError` if called more times than there were items
11231164
placed in the queue.
11241165

1166+
.. versionchanged:: next
1167+
Items discarded by ``shutdown(immediate=True)`` count as processed:
1168+
:meth:`task_done` must not be called for them.
1169+
11251170

11261171
.. method:: join()
11271172

‎Doc/whatsnew/3.16.rst‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -638,6 +638,11 @@ multiprocessing
638638
``N=buffersize-processes`` arguments).
639639
(Contributed by Oleksandr Baltian in :gh:`136871`.)
640640

641+
* Add :meth:`multiprocessing.Queue.shutdown` to terminate a
642+
:class:`multiprocessing.Queue` or a :class:`multiprocessing.JoinableQueue`
643+
in every process which uses it, like :meth:`queue.Queue.shutdown`.
644+
(Contributed by Yves Duprat and Gregory P. Smith in :gh:`96471`.)
645+
641646

642647
os
643648
--

0 commit comments

Comments
 (0)