Merge pull request #2097 from gevent/issue1957

jamadden · web-flow · commit cd9e22d72530 · 2025-04-16T15:20:23.000-05:00
Mnkey-patching the ``queue`` module (done by default in ``patch_all``) now patches ``Queue``, ``PriorityQueue``, and ``LifoQueue`
diff --git a/docs/changes/1957.bugfix.rst b/docs/changes/1957.bugfix.rst
@@ -0,0 +1,14 @@
+Monkey-patching the ``queue`` module (done by default in
+``patch_all``) now patches ``Queue``, ``PriorityQueue``, and
+``LifoQueue``. In addition to the general benefits of making all those
+classes cooperative, this is known to solve a non-deterministic
+deadlock with ``urllib3``.
+
+In addition, ``Queue`` was renamed to ``SimpleQueue``; previously
+``SimpleQueue`` was an alias for the undocumented
+``queue._PySimpleQueue``. This makes ``SimpleQueue`` cooperative even
+without monkey-patching.
+
+Likewise, ``JoinableQueue`` was renamed to ``Queue``, providing the
+``join`` method to all ``Queue`` objects, thus matching the standard
+library. The old name remains for backwards compatibility.
diff --git a/src/gevent/_gevent_cqueue.pxd b/src/gevent/_gevent_cqueue.pxd
@@ -36,7 +36,7 @@ cdef inline void greenlet_init():
 @cython.final
 cdef _safe_remove(deq, item)
 
-cdef class Queue:
+cdef class SimpleQueue:
     cdef __weakref__
     cdef readonly hub
     cdef readonly queue
@@ -66,9 +66,18 @@ cdef class Queue:
     cpdef get_nowait(self)
     cpdef peek(self, block=*, timeout=*)
     cpdef peek_nowait(self)
-    cpdef shutdown(self, immediate=*)
+
 
     cdef _schedule_unlock(self)
+
+
+cdef class Queue(SimpleQueue):
+    cdef Event _cond
+    cdef readonly int unfinished_tasks
+    cdef _did_put_task(self)
+
+
+    cpdef shutdown(self, immediate=*)
     cdef _drain_for_immediate_shutdown(self)
 
 
@@ -78,27 +87,14 @@ cdef class ItemWaiter(Waiter):
     cdef readonly item
     cdef readonly Queue queue
 
-###
-# XXX: Disabling Cython.final here pending a release > Cython 3.0.11
-# because it breaks on GCC. See https://github.com/gevent/gevent/issues/2049#issuecomment-2404700280
-# Restore when new cython is released.
-#
-# @cython.final
-###
+
 cdef class UnboundQueue(Queue):
     pass
 
 cdef class PriorityQueue(Queue):
     pass
 
-
-cdef class JoinableQueue(Queue):
-    cdef Event _cond
-    cdef readonly int unfinished_tasks
-    cdef _did_put_task(self)
-
-
-cdef class LifoQueue(JoinableQueue):
+cdef class LifoQueue(Queue):
     pass
 
 cdef class Channel:
diff --git a/src/gevent/monkey/__init__.py b/src/gevent/monkey/__init__.py
@@ -265,14 +265,20 @@ def patch_queue():
     """
     Patch objects in :mod:`queue`.
 
-
-    Currently, this just replaces :class:`queue.SimpleQueue` (implemented
-    in C) with its Python counterpart, but the details may change at any time.
+    This replaces ``SimpleQueue``, ``PriorityQueue``, ``Queue``
+    and ``LifoQueue`` with their gevent equivalents.
 
     .. versionadded:: 1.3.5
+
+    .. versionchanged:: NEXT
+       In addition to ``SimpleQueue``, now also patches
+       ``Queue``, ``PriorityQueue`` and ``LifoQueue``.`
     """
     _patch_module('queue', items=[
         'SimpleQueue',
+        'PriorityQueue',
+        'LifoQueue',
+        'Queue',
     ])
 
 
diff --git a/src/gevent/queue.py b/src/gevent/queue.py
@@ -26,6 +26,15 @@
 .. versionchanged:: 1.0
        ``Queue(0)`` now means queue of infinite size, not a channel. A :exc:`DeprecationWarning`
        will be issued with this argument.
+
+.. versionchanged:: NEXT
+   :class:`Queue` was renamed to :class:`SimpleQueue`, while :class:`JoinableQueue` was
+   renamed to :class:`Queue` (`JoinableQueue` remains a backwards compatible alias).
+   This adds the ability to ``join()`` all queues, like the standard library.
+
+   Previously ``SimpleQueue`` was an alias for the undocumented Python
+   implementation ``queue._PySimpleQueue``; now it is gevent's own implementation.
+   This ensures that it is cooperative even without monkey-patching.
 """
 
 
@@ -47,16 +56,10 @@
 from gevent.exceptions import InvalidSwitchError
 
 __all__ = []
-__implements__ = ['Queue', 'PriorityQueue', 'LifoQueue']
+__implements__ = ['Queue', 'PriorityQueue', 'LifoQueue', 'SimpleQueue']
 __extensions__ = ['JoinableQueue', 'Channel']
 __imports__ = ['Empty', 'Full']
 
-
-__all__.append('SimpleQueue')
-# SimpleQueue is implemented in C and directly allocates locks
-# unaffected by monkey patching. We need the Python version.
-SimpleQueue = __queue__._PySimpleQueue # pylint:disable=no-member
-
 if hasattr(__queue__, 'ShutDown'): # New in 3.13
     ShutDown = __queue__.ShutDown
     __imports__.append('ShutDown')
@@ -105,7 +108,7 @@ def put_and_switch(self):
         self.item = None
         return self.switch(self)
 
-class Queue(object):
+class SimpleQueue(object):
     """
     Create a queue object with a given maximum size.
 
@@ -126,6 +129,10 @@ class Queue(object):
        previously anyway, but that wasn't the case for PyPy.
     .. versionchanged:: 24.10.1
        Implement the ``shutdown`` methods from Python 3.13.
+    .. versionchanged:: NEXT
+       Renamed from ``Queue`` to ``SimpleQueue`` to better match the standard library.
+       While this class no longer has a ``shutdown`` method, the new ``Queue`` class
+       (previously ``JoinableQueue``) continues to have it.
     """
 
     __slots__ = (
@@ -449,81 +456,15 @@ def __next__(self):
             raise result
         return result
 
-    def shutdown(self, immediate=False):
-        """
-        "Shut-down the queue, making queue gets and puts raise
-        `ShutDown`.
-
-        By default, gets will only raise once the queue is empty. Set
-        *immediate* to True to make gets raise immediately instead.
-
-        All blocked callers of `put` and `get` will be unblocked.
-
-        In joinable queues, if *immediate*, a task is marked as done
-        for each item remaining in the queue, which may unblock
-        callers of `join`.
-        """
-        self.is_shutdown = True
-        if immediate:
-            self._drain_for_immediate_shutdown()
-        getters = list(self.getters)
-        putters = list(self.putters)
-        self.getters.clear()
-        self.putters.clear()
-        for waiter in getters + putters:
-            self.hub.loop.run_callback(waiter.throw, ShutDown)
-
-    def _drain_for_immediate_shutdown(self):
-        while self.qsize():
-            self.get()
-
-class UnboundQueue(Queue):
-    # A specialization of Queue that knows it can never
-    # be bound. Changing its maxsize has no effect.
-
-    __slots__ = ()
-
-    def __init__(self, maxsize=None, items=()):
-        if maxsize is not None:
-            raise ValueError("UnboundQueue has no maxsize")
-        Queue.__init__(self, maxsize, items)
-        self.putters = None # Will never be used.
-
-    def put(self, item, block=True, timeout=None):
-        self._put(item)
-        if self.getters:
-            self._schedule_unlock()
-
-
-class PriorityQueue(Queue):
-    '''A subclass of :class:`Queue` that retrieves entries in priority order (lowest first).
-
-    Entries are typically tuples of the form: ``(priority number, data)``.
 
-    .. versionchanged:: 1.2a1
-       Any *items* given to the constructor will now be passed through
-       :func:`heapq.heapify` to ensure the invariants of this class hold.
-       Previously it was just assumed that they were already a heap.
-    '''
-
-    __slots__ = ()
-
-    def _create_queue(self, items=()):
-        q = list(items)
-        _heapify(q)
-        return q
-
-    def _put(self, item):
-        _heappush(self.queue, item)
-
-    def _get(self):
-        return _heappop(self.queue)
-
-
-class JoinableQueue(Queue):
+class Queue(SimpleQueue):
     """
-    A subclass of :class:`Queue` that additionally has
+    A subclass of :class:`SimpleQueue` that additionally has
     :meth:`task_done` and :meth:`join` methods.
+
+    .. versionchanged:: NEXT
+       Renamed from ``JoinablQueue`` to simply ``Queue`` to better
+       match the capability of the standard library :class:`queue.Queue`.
     """
 
     __slots__ = (
@@ -539,7 +480,7 @@ def __init__(self, maxsize=None, items=(), unfinished_tasks=None):
            (if any) will be considered unfinished.
 
         """
-        Queue.__init__(self, maxsize, items, _warn_depth=3)
+        SimpleQueue.__init__(self, maxsize, items, _warn_depth=3)
 
         from gevent.event import Event
         self._cond = Event()
@@ -559,13 +500,13 @@ def copy(self):
         return type(self)(self.maxsize, self.queue, self.unfinished_tasks)
 
     def _format(self):
-        result = Queue._format(self)
+        result = SimpleQueue._format(self)
         if self.unfinished_tasks:
             result += ' tasks=%s _cond=%s' % (self.unfinished_tasks, self._cond)
         return result
 
     def _put(self, item):
-        Queue._put(self, item)
+        SimpleQueue._put(self, item)
         self._did_put_task()
 
     def _did_put_task(self):
@@ -574,8 +515,8 @@ def _did_put_task(self):
 
     def task_done(self):
         '''Indicate that a formerly enqueued task is complete. Used by queue consumer threads.
-        For each :meth:`get <Queue.get>` used to fetch a task, a subsequent call to :meth:`task_done` tells the queue
-        that the processing on the task is complete.
+        For each :meth:`get <Queue.get>` used to fetch a task, a subsequent call to
+        :meth:`task_done` tells the queue that the processing on the task is complete.
 
         If a :meth:`join` is currently blocking, it will resume when all items have been processed
         (meaning that a :meth:`task_done` call was received for every item that had been
@@ -608,13 +549,84 @@ def join(self, timeout=None):
         '''
         return self._cond.wait(timeout=timeout)
 
+    def shutdown(self, immediate=False):
+        """
+        "Shut-down the queue, making queue gets and puts raise
+        `ShutDown`.
+
+        By default, gets will only raise once the queue is empty. Set
+        *immediate* to True to make gets raise immediately instead.
+
+        All blocked callers of `put` and `get` will be unblocked.
+
+        In joinable queues, if *immediate*, a task is marked as done
+        for each item remaining in the queue, which may unblock
+        callers of `join`.
+        """
+        self.is_shutdown = True
+        if immediate:
+            self._drain_for_immediate_shutdown()
+        getters = list(self.getters)
+        putters = list(self.putters)
+        self.getters.clear()
+        self.putters.clear()
+        for waiter in getters + putters:
+            self.hub.loop.run_callback(waiter.throw, ShutDown)
+
     def _drain_for_immediate_shutdown(self):
         while self.qsize():
             self.get()
             self.task_done()
 
+# .. versionchanged:: NEXT
+#  Now a BWC alias
+JoinableQueue = Queue
+
+class UnboundQueue(Queue):
+    # A specialization of Queue that knows it can never
+    # be bound. Changing its maxsize has no effect.
+
+    __slots__ = ()
+
+    def __init__(self, maxsize=None, items=()):
+        if maxsize is not None:
+            raise ValueError("UnboundQueue has no maxsize")
+        Queue.__init__(self, maxsize, items)
+        self.putters = None # Will never be used.
+
+    def put(self, item, block=True, timeout=None):
+        self._put(item)
+        if self.getters:
+            self._schedule_unlock()
+
+
+class PriorityQueue(Queue):
+    '''A subclass of :class:`Queue` that retrieves entries in priority order (lowest first).
+
+    Entries are typically tuples of the form: ``(priority number, data)``.
+
+    .. versionchanged:: 1.2a1
+       Any *items* given to the constructor will now be passed through
+       :func:`heapq.heapify` to ensure the invariants of this class hold.
+       Previously it was just assumed that they were already a heap.
+    '''
+
+    __slots__ = ()
+
+    def _create_queue(self, items=()):
+        q = list(items)
+        _heapify(q)
+        return q
+
+    def _put(self, item):
+        _heappush(self.queue, item)
+        self._did_put_task()
+
+    def _get(self):
+        return _heappop(self.queue)
+
 
-class LifoQueue(JoinableQueue):
+class LifoQueue(Queue):
     """
     A subclass of :class:`JoinableQueue` that retrieves most recently added entries first.
 
diff --git a/src/gevent/tests/test__monkey.py b/src/gevent/tests/test__monkey.py
@@ -152,17 +152,14 @@ def veto(event):
                              and e.module_name == 'ssl')
 
     def test_patch_queue(self):
-        try:
-            import queue
-        except ImportError:
-            # Python 2 called this Queue. Note that having
-            # python-future installed gives us a queue module on
-            # Python 2 as well.
-            queue = None
-        if not hasattr(queue, 'SimpleQueue'):
-            raise unittest.SkipTest("Needs SimpleQueue")
+        import queue
+        import gevent.queue as gq
+
         # pylint:disable=no-member
-        self.assertIs(queue.SimpleQueue, queue._PySimpleQueue)
+        self.assertIs(queue.SimpleQueue, gq.SimpleQueue)
+        self.assertIs(queue.LifoQueue, gq.LifoQueue)
+        self.assertIs(queue.Queue, gq.Queue)
+        self.assertIs(queue.PriorityQueue, gq.PriorityQueue)
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/src/gevent/tests/test__queue.py b/src/gevent/tests/test__queue.py
