diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index dc45f8401541f4..4f5c0443551ebb 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -942,8 +942,7 @@ For an example of the usage of queues for interprocess communication see standard library's :mod:`queue` module are raised to signal timeouts. :class:`Queue` implements all the methods of :class:`queue.Queue` except for - :meth:`~queue.Queue.task_done`, :meth:`~queue.Queue.join`, and - :meth:`~queue.Queue.shutdown`. + :meth:`~queue.Queue.task_done` and :meth:`~queue.Queue.join`. .. method:: qsize() @@ -980,6 +979,11 @@ For an example of the usage of queues for interprocess communication see If the queue is closed, :exc:`ValueError` is raised instead of :exc:`AssertionError`. + Raises :exc:`queue.ShutDown` if the queue has been shut down. + + .. versionchanged:: next + :exc:`queue.ShutDown` is raised if the queue has been shut down. + .. method:: put_nowait(obj) Equivalent to ``put(obj, False)``. @@ -998,10 +1002,65 @@ For an example of the usage of queues for interprocess communication see If the queue is closed, :exc:`ValueError` is raised instead of :exc:`OSError`. + Raises :exc:`queue.ShutDown` if the queue has been shut down and is + empty, or if the queue has been shut down immediately. + + .. versionchanged:: next + :exc:`queue.ShutDown` is raised if the queue has been shut down. + .. method:: get_nowait() Equivalent to ``get(False)``. + .. method:: shutdown(immediate=False) + + Put the queue into a shutdown mode, in every process which uses it. + + The queue can no longer grow. + Future calls to :meth:`put` raise :exc:`queue.ShutDown`. + Currently blocked callers of :meth:`put` will be unblocked + and will raise :exc:`queue.ShutDown` in the formerly blocked process. + + If *immediate* is false (the default), the queue can be wound + down normally with :meth:`get` calls to extract the items which have + already been 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, future calls to :meth:`get` will raise + :exc:`queue.ShutDown`. Currently blocked callers of :meth:`get` are + unblocked and raise :exc:`queue.ShutDown` as well. + + If *immediate* is true, the items remaining in the queue are discarded + instead, including those which other processes have put but not yet + written to the pipe. All callers of :meth:`get` are unblocked. + + Calling :meth:`shutdown` more than once is allowed. A queue which has + been shut down cannot be put back into service. + + Raises :exc:`ValueError` if the queue has been closed in this process + with :meth:`close`. + + .. note:: + + A getter is woken up by the getter woken before it, so the + :exc:`queue.ShutDown` exception can only reach all of the blocked + processes as long as none of them is killed, or exits without + flushing its feeder thread (see :meth:`cancel_join_thread`), as it + is being woken. + + Similarly, if a process is killed or exits without flushing its + feeder thread after putting items on the queue, the queue may never + be considered empty and :meth:`get` may block instead of raising + :exc:`queue.ShutDown`. + + .. note:: + + On Windows, an item which another process is writing to the pipe + while ``shutdown(immediate=True)`` is running may still be returned + by a later :meth:`get`. + + .. versionadded:: next + :class:`multiprocessing.Queue` has a few additional methods not found in :class:`queue.Queue`. These methods are usually unnecessary for most code: @@ -1104,6 +1163,10 @@ For an example of the usage of queues for interprocess communication see Raises a :exc:`ValueError` if called more times than there were items placed in the queue. + .. versionchanged:: next + Items discarded by ``shutdown(immediate=True)`` count as processed: + :meth:`task_done` must not be called for them. + .. method:: join() diff --git a/Doc/whatsnew/3.16.rst b/Doc/whatsnew/3.16.rst index da81d8513da99d..a3dcfea29690d1 100644 --- a/Doc/whatsnew/3.16.rst +++ b/Doc/whatsnew/3.16.rst @@ -638,6 +638,11 @@ multiprocessing ``N=buffersize-processes`` arguments). (Contributed by Oleksandr Baltian in :gh:`136871`.) +* Add :meth:`multiprocessing.Queue.shutdown` to terminate a + :class:`multiprocessing.Queue` or a :class:`multiprocessing.JoinableQueue` + in every process which uses it, like :meth:`queue.Queue.shutdown`. + (Contributed by Yves Duprat and Gregory P. Smith in :gh:`96471`.) + os -- diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py index 981599acf5ef26..0d59dfe32c3993 100644 --- a/Lib/multiprocessing/queues.py +++ b/Lib/multiprocessing/queues.py @@ -18,7 +18,7 @@ import weakref import errno -from queue import Empty, Full +from queue import Empty, Full, ShutDown from . import connection from . import context @@ -45,6 +45,24 @@ def __init__(self, maxsize=0, *, ctx): else: self._wlock = ctx.Lock() self._sem = ctx.BoundedSemaphore(maxsize) + + # State used by shutdown(). It is all kept in semaphores, which are + # only ever tested for zero, because neither ctypes nor a working + # sem_getvalue() is available on every platform. + # + # Where sem_getvalue() is broken (macOS), the zero test acquires + # the semaphore and releases it again, so a second test made at the + # same moment can wrongly find it zero. Every test whose answer + # matters is therefore made while holding _state_lock. + self._state_lock = ctx.Lock() + # Non-zero once shutdown() has been called. + self._shutdown_flag = ctx.Semaphore(0) + # Non-zero once shutdown(immediate=True) has been called. + self._immediate_flag = ctx.Semaphore(0) + # Number of items accepted by put() in any process and not yet + # received: those in the pipe and in every process's buffer. + self._outstanding = ctx.Semaphore(0) + # For use by concurrent.futures self._ignore_epipe = False self._reset() @@ -55,11 +73,15 @@ def __init__(self, maxsize=0, *, ctx): def __getstate__(self): context.assert_spawning(self) return (self._ignore_epipe, self._maxsize, self._reader, self._writer, - self._rlock, self._wlock, self._sem, self._opid) + self._rlock, self._wlock, self._sem, self._opid, + self._state_lock, self._shutdown_flag, self._immediate_flag, + self._outstanding) def __setstate__(self, state): (self._ignore_epipe, self._maxsize, self._reader, self._writer, - self._rlock, self._wlock, self._sem, self._opid) = state + self._rlock, self._wlock, self._sem, self._opid, + self._state_lock, self._shutdown_flag, self._immediate_flag, + self._outstanding) = state self._reset() def _after_fork(self): @@ -77,15 +99,52 @@ def _reset(self, after_fork=False): self._joincancelled = False self._closed = False self._close = None + # The shutdown flag is never cleared, so it can be cached once seen. + self._shutdown_seen = False self._send_bytes = self._writer.send_bytes self._recv_bytes = self._reader.recv_bytes self._poll = self._reader.poll + def _is_shutdown(self): + if not self._shutdown_seen: + with self._state_lock: + self._shutdown_seen = ( + not self._shutdown_flag._semlock._is_zero()) + return self._shutdown_seen + + def _release_slot(self): + try: + self._sem.release() + except ValueError: + # shutdown() may have released _sem once more than there were + # items, to wake the blocked putters. + if not self._is_shutdown(): + raise + + def _acquire_slot(self, block, timeout): + # Reserve room for one item and count it as outstanding, or raise + # Full or ShutDown. + if self._shutdown_seen: + raise ShutDown + if not self._sem.acquire(block, timeout): + raise ShutDown if self._is_shutdown() else Full + if not self._shutdown_seen: + # Testing the flag and counting the item must be atomic with + # respect to shutdown(): once a getter has seen the queue shut + # down with nothing outstanding, no put() may succeed. + with self._state_lock: + if self._shutdown_flag._semlock._is_zero(): + self._outstanding.release() + return + self._shutdown_seen = True + # Hand the unit on, so that the next blocked putter wakes up too. + self._release_slot() + raise ShutDown + def put(self, obj, block=True, timeout=None): if self._closed: raise ValueError(f"Queue {self!r} is closed") - if not self._sem.acquire(block, timeout): - raise Full + self._acquire_slot(block, timeout) with self._notempty: if self._thread is None: @@ -93,37 +152,201 @@ def put(self, obj, block=True, timeout=None): self._buffer.append(obj) self._notempty.notify() + def _received(self): + # Called with _rlock held, for each item read from the pipe. + if not self._outstanding.acquire(False): + # Either a zero test was holding the semaphore (see __init__), + # or the item was put by a subclass which overrides put() + # without calling _acquire_slot() and was never counted. + with self._state_lock: + self._outstanding.acquire(False) + + def _after_marker(self, block, deadline): + # Called with _rlock held, after reading the shutdown marker instead + # of an item. Return the next item, or raise ShutDown or Empty. + # + # shutdown() writes a single marker and whoever reads it must write + # it back, so that every getter, blocked or yet to come, is woken in + # its turn. Extra markers are harmless: they are only wake-ups, and + # a reader never writes back more than one. + try: + while True: + with self._state_lock: + drained = self._outstanding._semlock._is_zero() + immediate = not self._immediate_flag._semlock._is_zero() + if drained: + # No item is outstanding. Still return anything found + # in the pipe, such as an item put by a subclass which + # does not count them, rather than strand it. + if not self._poll(): + raise ShutDown + elif not block: + if not self._poll(): + raise Empty + elif deadline is not None: + if not self._poll(max(0.0, deadline - time.monotonic())): + raise Empty + # Otherwise an item is still in some process's buffer: wait + # for it. If it gets discarded instead, whoever discards it + # writes another marker. + res = self._recv_bytes() + if not res: + continue + if immediate: + self._make_discarder()(1) + continue + return res + finally: + self._put_marker() + def get(self, block=True, timeout=None): if self._closed: raise ValueError(f"Queue {self!r} is closed") if block and timeout is None: with self._rlock: res = self._recv_bytes() - self._sem.release() + if not res: + # An empty message is the shutdown marker. + res = self._after_marker(True, None) + self._received() + self._release_slot() else: if block: deadline = time.monotonic() + timeout - if not self._rlock.acquire(block, timeout): - raise Empty try: - if block: - timeout = deadline - time.monotonic() - if not self._poll(timeout): - raise Empty - elif not self._poll(): + if not self._rlock.acquire(block, timeout): raise Empty - res = self._recv_bytes() - self._sem.release() - finally: - self._rlock.release() + try: + if block: + timeout = deadline - time.monotonic() + if not self._poll(timeout): + raise Empty + elif not self._poll(): + raise Empty + res = self._recv_bytes() + if not res: + res = self._after_marker(block, + deadline if block else None) + self._received() + self._release_slot() + finally: + self._rlock.release() + except Empty: + # The pipe of a shut down queue is briefly empty while a + # getter writes the marker back. + if self._is_shutdown() and self._is_drained(): + raise ShutDown from None + raise # unserialize the data after having released the lock return _ForkingPickler.loads(res) + def _is_drained(self): + with self._state_lock: + return self._outstanding._semlock._is_zero() + + def _put_marker(self): + # The marker goes through the feeder thread so that it follows + # everything this process has already put, and so that writing it + # can never block the caller. + with self._notempty: + if self._thread is None: + self._start_thread() + self._buffer.append(_shutdown_marker) + self._notempty.notify() + + def _make_discarder(self): + # Return a function accounting for N items thrown away by + # shutdown(immediate=True). It must not refer to self: the feeder + # thread keeps it, and is only told to exit when the queue is + # garbage collected. + sem, outstanding, state_lock = ( + self._sem, self._outstanding, self._state_lock) + def discard(n): + for _ in range(n): + if not outstanding.acquire(False): + with state_lock: + outstanding.acquire(False) + try: + sem.release() + except ValueError: + pass + return discard + + def shutdown(self, immediate=False): + if self._closed: + raise ValueError(f"Queue {self!r} is closed") + with self._state_lock: + first = self._shutdown_flag._semlock._is_zero() + if first: + self._shutdown_flag.release() + if immediate and self._immediate_flag._semlock._is_zero(): + self._immediate_flag.release() + self._shutdown_seen = True + if first: + # Wake a putter blocked on a full queue; it wakes the next one. + # Putters which acquire a unit from now on give it back, so + # there is nothing to do if the queue is not full. + if self._sem._semlock._is_zero(): + self._release_slot() + # Wake a blocked getter; it wakes the next one. + self._put_marker() + if immediate: + self._discard_all() + + def _discard_all(self): + discard = self._make_discarder() + # Items this process has buffered. Feeder threads, here and in + # other processes, discard what they hold on seeing the flag. + with self._notempty: + keep = [obj for obj in self._buffer + if obj is _sentinel or obj is _shutdown_marker] + n = len(self._buffer) - len(keep) + self._buffer.clear() + self._buffer.extend(keep) + if n: + discard(n) + # A getter may be waiting for one of them. + self._put_marker() + # Items already in the pipe. + wlock = self._wlock + n = 0 + with self._rlock: + while True: + while self._poll(): + if self._recv_bytes(): + n += 1 + if wlock is None or wlock.acquire(timeout=0.01): + break + # A feeder thread is in the middle of writing, and may be + # waiting for room in the pipe: keep reading. + if wlock is None: + # Windows: there is no write lock to stop other processes + # from writing, so an item may still get in ahead of the + # marker and be returned by get(). + self._put_marker() + else: + # No one else can write now. Empty the pipe and leave the + # marker in it, ahead of any item a feeder thread had + # already decided to send: the getter which reads the marker + # throws those away before letting another getter in. + try: + while self._poll(): + if self._recv_bytes(): + n += 1 + self._send_bytes(b'') + finally: + wlock.release() + discard(n) + def qsize(self): # Raises NotImplementedError on Mac OSX because of broken sem_getvalue() - return self._maxsize - self._sem.get_value() + return max(0, self._maxsize - self._sem.get_value()) def empty(self): + if self._is_shutdown(): + # The pipe now holds the shutdown marker, so whether it can be + # read says nothing about items. + return self._is_drained() return not self._poll() def full(self): @@ -180,7 +403,9 @@ def _start_thread(self): args=(self._buffer, self._notempty, self._send_bytes, self._wlock, self._reader.close, self._writer.close, self._ignore_epipe, self._on_queue_feeder_error, - self._sem), + self._sem, self._make_discarder(), + (self._state_lock, self._shutdown_flag, + self._immediate_flag, self._outstanding)), name='QueueFeederThread', daemon=True, ) @@ -228,20 +453,35 @@ def _finalize_close(buffer, notempty): @staticmethod def _feed(buffer, notempty, send_bytes, writelock, reader_close, - writer_close, ignore_epipe, onerror, queue_sem): + writer_close, ignore_epipe, onerror, queue_sem, + discard, shutdown_state): debug('starting thread to feed data to pipe') nacquire = notempty.acquire nrelease = notempty.release nwait = notempty.wait bpopleft = buffer.popleft sentinel = _sentinel + marker = _shutdown_marker + state_lock, shutdown_flag, immediate_flag, outstanding = shutdown_state + not_immediate = immediate_flag._semlock._is_zero if sys.platform != 'win32': wacquire = writelock.acquire wrelease = writelock.release else: wacquire = None + def send(data): + if wacquire is None: + send_bytes(data) + else: + wacquire() + try: + send_bytes(data) + finally: + wrelease() + while 1: + is_item = False try: nacquire() try: @@ -249,27 +489,43 @@ def _feed(buffer, notempty, send_bytes, writelock, reader_close, nwait() finally: nrelease() + ndiscarded = 0 try: while 1: + is_item = False obj = bpopleft() - if obj is sentinel: - debug('feeder thread got sentinel -- exiting') - reader_close() - writer_close() - return - + if obj is sentinel or obj is marker: + if ndiscarded: + # A getter may be waiting for one of the + # discarded items: wake it. + discard(ndiscarded) + ndiscarded = 0 + send(b'') + if obj is sentinel: + debug('feeder thread got sentinel -- exiting') + reader_close() + writer_close() + return + # An empty message is never a valid pickle. + send(b'') + continue + + # After shutdown(immediate=True), throw away what + # has not been sent yet. This test is not made + # under state_lock, to keep the feeder cheap: at + # worst an item is sent which should not have been. + if not not_immediate(): + ndiscarded += 1 + continue + + is_item = True # serialize the data before acquiring the lock obj = _ForkingPickler.dumps(obj) - if wacquire is None: - send_bytes(obj) - else: - wacquire() - try: - send_bytes(obj) - finally: - wrelease() + send(obj) except IndexError: - pass + if ndiscarded: + discard(ndiscarded) + send(b'') except Exception as e: if ignore_epipe and getattr(e, 'errno', 0) == errno.EPIPE: return @@ -286,7 +542,14 @@ def _feed(buffer, notempty, send_bytes, writelock, reader_close, # if the object had been silently removed from the queue # and this step is necessary to have a properly working # queue. - queue_sem.release() + if is_item: + queue_sem.release() + with state_lock: + outstanding.acquire(False) + shutdown = not shutdown_flag._semlock._is_zero() + if shutdown: + # A getter may be waiting for this item. + buffer.appendleft(marker) onerror(e, obj) @staticmethod @@ -302,6 +565,9 @@ def _on_queue_feeder_error(e, obj): _sentinel = object() +# Put in a feeder thread's buffer to have the shutdown marker, an empty +# message, written to the pipe. +_shutdown_marker = object() # # A queue type which also supports join() and task_done() methods @@ -328,8 +594,7 @@ def __setstate__(self, state): def put(self, obj, block=True, timeout=None): if self._closed: raise ValueError(f"Queue {self!r} is closed") - if not self._sem.acquire(block, timeout): - raise Full + self._acquire_slot(block, timeout) with self._notempty, self._cond: if self._thread is None: @@ -350,6 +615,21 @@ def join(self): if not self._unfinished_tasks._semlock._is_zero(): self._cond.wait() + def _make_discarder(self): + discard_items = super()._make_discarder() + cond, unfinished_tasks = self._cond, self._unfinished_tasks + def discard(n): + discard_items(n) + # Only the discarded items are done: a task which a worker has + # already taken still needs its task_done(). + with cond: + for _ in range(n): + if not unfinished_tasks.acquire(False): + break + if unfinished_tasks._semlock._is_zero(): + cond.notify_all() + return discard + # # Simplified Queue type -- really just a locked pipe # diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 540189147cb4d7..80bfd79edaa23d 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -1536,6 +1536,348 @@ def test_closed_queue_put_get_exceptions(self): q.put('foo') with self.assertRaisesRegex(ValueError, 'is closed'): q.get() + + # + # shutdown() + # + # Except where they look at the internals, these tests also run on + # queue.Queue, through the 'threads' and 'manager' types: the + # behaviour must be the same. + # + + @classmethod + def _shutdown_getter(cls, q, results, ready): + ready.set() + items = [] + try: + while True: + items.append(q.get()) + except pyqueue.ShutDown: + pass + results.put(items) + + @classmethod + def _shutdown_putter(cls, q, results, ready): + ready.set() + try: + q.put('extra') + except pyqueue.ShutDown: + results.put('ShutDown') + else: + results.put('put') + + def _start_shutdown_workers(self, n, target, q): + results = self.Queue() + self.addCleanup(close_queue, results) + readies = [self.Event() for _ in range(n)] + procs = [self.Process(target=target, args=(q, results, ready)) + for ready in readies] + for p in procs: + p.daemon = True + p.start() + self.addCleanup(join_process, p) + for ready in readies: + self.assertTrue(ready.wait(support.SHORT_TIMEOUT)) + # Let the workers block in get() or put(). This is only to make + # the interesting case the likely one: the tests are valid + # whatever the workers are doing when shutdown() is called. + time.sleep(0.1) + return [functools.partial(results.get, timeout=support.SHORT_TIMEOUT) + for p in procs] + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_empty(self): + for immediate in False, True: + with self.subTest(immediate=immediate): + q = self.Queue() + q.shutdown(immediate=immediate) + with self.assertRaises(pyqueue.ShutDown): + q.put(1) + with self.assertRaises(pyqueue.ShutDown): + q.put_nowait(1) + with self.assertRaises(pyqueue.ShutDown): + q.put(1, timeout=0.01) + with self.assertRaises(pyqueue.ShutDown): + q.get() + with self.assertRaises(pyqueue.ShutDown): + q.get_nowait() + with self.assertRaises(pyqueue.ShutDown): + q.get(timeout=0.01) + self.assertTrue(queue_empty(q)) + close_queue(q) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_drain(self): + q = self.Queue() + for i in range(3): + q.put(i) + # Do not wait: the items put just before shutdown() must not be + # lost, even those the feeder thread has not written yet. + q.shutdown() + with self.assertRaises(pyqueue.ShutDown): + q.put(3) + self.assertEqual(q.get(), 0) + self.assertEqual(q.get(timeout=support.SHORT_TIMEOUT), 1) + self.assertEqual(q.get(), 2) + with self.assertRaises(pyqueue.ShutDown): + q.get() + with self.assertRaises(pyqueue.ShutDown): + q.get_nowait() + with self.assertRaises(pyqueue.ShutDown): + q.get(timeout=0.01) + self.assertTrue(queue_empty(q)) + close_queue(q) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_immediate(self): + q = self.Queue(maxsize=3) + for i in range(3): + q.put(i) + q.shutdown(immediate=True) + with self.assertRaises(pyqueue.ShutDown): + q.get() + with self.assertRaises(pyqueue.ShutDown): + q.get_nowait() + with self.assertRaises(pyqueue.ShutDown): + q.put(3) + self.assertTrue(queue_empty(q)) + self.assertFalse(queue_full(q, 3)) + self.assertReturnsIfImplemented(0, q.qsize) + close_queue(q) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_repeated(self): + q = self.Queue() + q.put(1) + q.put(2) + q.shutdown() + q.shutdown() + self.assertEqual(q.get(), 1) + # An immediate shutdown can follow a graceful one. + q.shutdown(immediate=True) + with self.assertRaises(pyqueue.ShutDown): + q.get() + q.shutdown() + self.assertTrue(queue_empty(q)) + close_queue(q) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_closed(self): + if self.TYPE != 'processes': + self.skipTest(f'test not appropriate for {self.TYPE}') + q = self.Queue() + q.close() + with self.assertRaisesRegex(ValueError, 'is closed'): + q.shutdown() + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_blocked_getters(self): + for immediate in False, True: + with self.subTest(immediate=immediate): + q = self.Queue() + results = self._start_shutdown_workers( + 4, self._shutdown_getter, q) + q.shutdown(immediate=immediate) + self.assertEqual([get() for get in results], [[]] * 4) + close_queue(q) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_blocked_getter_gets_items(self): + # The items put before shutdown() go to a getter which was already + # waiting. + q = self.Queue() + result, = self._start_shutdown_workers(1, self._shutdown_getter, q) + for i in range(10): + q.put(i) + q.shutdown() + self.assertEqual(result(), list(range(10))) + close_queue(q) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_many_getters(self): + # No getter may be left waiting for an item that another one took. + for nitems in 1, 100: + with self.subTest(nitems=nitems): + q = self.Queue() + for i in range(nitems): + q.put(i) + q.shutdown() + results = self._start_shutdown_workers( + 4, self._shutdown_getter, q) + items = [item for get in results for item in get()] + self.assertEqual(sorted(items), list(range(nitems))) + close_queue(q) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_blocked_putters(self): + for immediate in False, True: + with self.subTest(immediate=immediate): + q = self.Queue(maxsize=2) + q.put(1) + q.put(2) + results = self._start_shutdown_workers( + 4, self._shutdown_putter, q) + q.shutdown(immediate=immediate) + self.assertEqual([get() for get in results], + ['ShutDown'] * 4) + if not immediate: + self.assertEqual(q.get(), 1) + self.assertEqual(q.get(), 2) + with self.assertRaises(pyqueue.ShutDown): + q.get() + self.assertTrue(queue_empty(q)) + close_queue(q) + + @classmethod + def _shutdown_producer(cls, q, nitems, done, exit): + for i in range(nitems): + q.put((i, bytes(1000))) + done.set() + # Exiting would wait for the feeder thread to write everything to + # the pipe. Stay alive instead, so that items remain in its buffer. + exit.wait(support.LONG_TIMEOUT) + + def _start_shutdown_producer(self, q, nitems): + done = self.Event() + exit = self.Event() + self.addCleanup(exit.set) + p = self.Process(target=self._shutdown_producer, + args=(q, nitems, done, exit)) + p.daemon = True + p.start() + self.assertTrue(done.wait(support.SHORT_TIMEOUT)) + return p, exit + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_other_process_items(self): + # Most of the items are too big for the pipe: they are still in the + # buffer of the process which put them when the queue is shut down. + q = self.Queue() + nitems = 1000 + p, exit = self._start_shutdown_producer(q, nitems) + q.shutdown() + for i in range(nitems): + self.assertEqual(q.get(timeout=support.SHORT_TIMEOUT)[0], i) + with self.assertRaises(pyqueue.ShutDown): + q.get() + self.assertTrue(queue_empty(q)) + exit.set() + join_process(p) + close_queue(q) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_immediate_other_process_items(self): + if self.TYPE == 'processes' and sys.platform == 'win32': + self.skipTest('an item being written may be left on Windows') + q = self.Queue() + p, exit = self._start_shutdown_producer(q, 1000) + q.shutdown(immediate=True) + with self.assertRaises(pyqueue.ShutDown): + q.get() + self.assertTrue(queue_empty(q)) + self.assertReturnsIfImplemented(0, q.qsize) + exit.set() + join_process(p) + close_queue(q) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_feeder_error(self): + # A getter waiting for an item which the feeder thread fails to + # send is woken up. + if self.TYPE != 'processes': + self.skipTest(f'test not appropriate for {self.TYPE}') + + class NotSerializable(object): + def __reduce__(self): + raise AttributeError + q = self.Queue() + result, = self._start_shutdown_workers(1, self._shutdown_getter, q) + with test.support.captured_stderr(): + q.put(NotSerializable()) + q.shutdown() + self.assertEqual(result(), []) + close_queue(q) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_subclass_put(self): + # A subclass may override put() without calling _acquire_slot(), so + # that its items are not counted. They must not be lost. + if self.TYPE != 'processes': + self.skipTest(f'test not appropriate for {self.TYPE}') + + class MyQueue(multiprocessing.queues.Queue): + def put(self, obj, block=True, timeout=None): + if not self._sem.acquire(block, timeout): + raise pyqueue.Full + with self._notempty: + if self._thread is None: + self._start_thread() + self._buffer.append(obj) + self._notempty.notify() + + q = MyQueue(ctx=multiprocessing.get_context()) + q.put(1) + q.put(2) + self.assertEqual(q.get(), 1) + q.shutdown() + self.assertEqual(q.get(), 2) + with self.assertRaises(pyqueue.ShutDown): + q.get() + close_queue(q) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_joinable(self): + q = self.JoinableQueue() + q.put(1) + q.put(2) + q.shutdown() + self.assertEqual(q.get(), 1) + q.task_done() + self.assertEqual(q.get(), 2) + q.task_done() + q.join() + with self.assertRaises(pyqueue.ShutDown): + q.get() + close_queue(q) + + @classmethod + def _shutdown_joiner(cls, q, joined): + q.join() + joined.set() + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_immediate_joinable(self): + q = self.JoinableQueue() + for i in range(3): + q.put(i) + self.assertEqual(q.get(), 0) + q.shutdown(immediate=True) + with self.assertRaises(pyqueue.ShutDown): + q.get() + + # The discarded items are done, the one which was taken is not. + joined = self.Event() + p = self.Process(target=self._shutdown_joiner, args=(q, joined)) + p.daemon = True + p.start() + self.assertFalse(joined.wait(0.1)) + q.task_done() + self.assertTrue(joined.wait(support.SHORT_TIMEOUT)) + join_process(p) + with self.assertRaises(ValueError): + q.task_done() + close_queue(q) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_shutdown_immediate_joinable_blocked_getter(self): + # join() returns although no task_done() is ever called. + q = self.JoinableQueue() + result, = self._start_shutdown_workers(1, self._shutdown_getter, q) + q.shutdown(immediate=True) + self.assertEqual(result(), []) + q.join() + close_queue(q) + # # # diff --git a/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst b/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst new file mode 100644 index 00000000000000..9ee370289436ad --- /dev/null +++ b/Misc/NEWS.d/next/Library/2025-09-12-12-02-02.gh-issue-96471.R6sqGE.rst @@ -0,0 +1,3 @@ +Add :meth:`multiprocessing.Queue.shutdown` to terminate a +:class:`multiprocessing.Queue` or a :class:`multiprocessing.JoinableQueue` +in every process which uses it, like :meth:`queue.Queue.shutdown`.