gh-142837: Use semaphore accounting for multiprocessing.Queue.empty()

Author: iamrajhans

Doc/library/multiprocessing.rst

@@ -869,10 +869,13 @@ Note that one can also create a shared queue by using a manager object -- see
    bother you then you can instead use a queue created with a
    :ref:`manager <multiprocessing-managers>`.
 
-   (1) After putting an object on an empty queue there may be an
-       infinitesimal delay before the queue's :meth:`~Queue.empty`
-       method returns :const:`False` and :meth:`~Queue.get_nowait` can
-       return without raising :exc:`queue.Empty`.
+   (1) After putting an object on an empty queue there may be a delay
+       before :meth:`~Queue.get_nowait` can return without raising
+       :exc:`queue.Empty`, because the feeder thread flushes objects to
+       the underlying pipe asynchronously. On platforms where
+       ``sem_getvalue()`` is not implemented (for example macOS), the
+       queue's :meth:`~Queue.empty` method may also remain :const:`True`
+       during this delay.
 
    (2) If multiple processes are enqueuing objects, it is possible for
        the objects to be received at the other end out-of-order.
@@ -947,8 +950,17 @@ For an example of the usage of queues for interprocess communication see
       Return ``True`` if the queue is empty, ``False`` otherwise.  Because of
       multithreading/multiprocessing semantics, this is not reliable.
 
+      On platforms where ``sem_getvalue()`` is implemented, this method
+      uses the same approximate size accounting as :meth:`~Queue.qsize`.
+      Otherwise, it may report ``True`` while items are still buffered and
+      waiting to be flushed to the underlying pipe.
+
       May raise an :exc:`OSError` on closed queues. (not guaranteed)
 
+      .. versionchanged:: 3.15
+         On platforms where ``sem_getvalue()`` is implemented, this method
+         now uses semaphore-based queue size accounting.
+
    .. method:: full()
 
       Return ``True`` if the queue is full, ``False`` otherwise.  Because of

Lib/multiprocessing/queues.py

@@ -124,7 +124,17 @@ def qsize(self):
         return self._maxsize - self._sem.get_value()
 
     def empty(self):
-        return not self._poll()
+        # Preserve the historical "closed queue may raise OSError" behavior.
+        # q.close() is a no-op for unused queues, so this only raises once the
+        # reader end has actually been closed.
+        if self._closed:
+            self._poll()
+
+        try:
+            return self._sem.get_value() == self._maxsize
+        except NotImplementedError:
+            # Fallback for platforms without sem_getvalue() (for example macOS).
+            return not self._poll()
 
     def full(self):
         return self._sem._semlock._is_zero()

Lib/test/_test_multiprocessing.py

@@ -1320,6 +1320,29 @@ def test_qsize(self):
         self.assertEqual(q.qsize(), 0)
         close_queue(q)
 
+    def test_empty_uses_semaphore_count(self):
+        if self.TYPE != 'processes':
+            self.skipTest(f'test not appropriate for {self.TYPE}')
+
+        q = self.Queue()
+        try:
+            q._sem.get_value()
+        except NotImplementedError:
+            close_queue(q)
+            self.skipTest('sem_getvalue not implemented on this platform')
+
+        q.put('sentinel')
+        original_poll = q._poll
+        q._poll = lambda timeout=0.0: False
+        try:
+            self.assertFalse(q.empty())
+        finally:
+            q._poll = original_poll
+
+        self.assertEqual(q.get(timeout=support.SHORT_TIMEOUT), 'sentinel')
+        self.assertTrue(q.empty())
+        close_queue(q)
+
     @classmethod
     def _test_task_done(cls, q):
         for obj in iter(q.get, None):