YvesDup · Mar 15, 2021 · Mar 15, 2021 · Mar 15, 2021 · Mar 16, 2021 · Mar 16, 2021
@@ -1,4 +1,5 @@
-__all__ = ('Queue', 'PriorityQueue', 'LifoQueue', 'QueueFull', 'QueueEmpty')
+__all__ = ('Queue', 'PriorityQueue', 'LifoQueue', 'QueueFull', 'QueueEmpty',
+           'QueueFullWithPendingPutTasks', 'QueueEmptyWithPendingGetTasks')
 
 import collections
 import heapq
@@ -12,11 +13,17 @@ class QueueEmpty(Exception):
     """Raised when Queue.get_nowait() is called on an empty Queue."""
     pass
 
+class QueueEmptyWithPendingGetTasks(QueueEmpty):
+    """Raised when the Queue.get_nowait() method is called on not empty Queue._getters"""
+    pass
 
 class QueueFull(Exception):
     """Raised when the Queue.put_nowait() method is called on a full Queue."""
     pass
 
+class QueueFullWithPendingPutTasks(QueueFull):
+    """Raised when the Queue.put_nowait() method is called on not empty Queue._putters"""
+    pass
 
 class Queue(mixins._LoopBoundMixin):
     """A queue, useful for coordinating producer and consumer coroutines.
@@ -57,8 +64,9 @@ def _put(self, item):
 
     def _wakeup_next(self, waiters):
         # Wake up the next waiter (if any) that isn't cancelled.
-        while waiters:
-            waiter = waiters.popleft()
+        # from gh-83055: don't remove future from waiters,
+        # do it in get() or put() when `waiter` wake up.
+        for waiter in waiters:
             if not waiter.done():
                 waiter.set_result(None)
                 break
@@ -111,13 +119,16 @@ async def put(self, item):
         """Put an item into the queue.
 
         Put an item into the queue. If the queue is full, wait until a free
-        slot is available before adding item.
+        slot is available before adding item. If there are still item moving
+        to a free or waiting for free slots, wait for its time.
         """
-        while self.full():
+        if self.full() or self._putters:
             putter = self._get_loop().create_future()
             self._putters.append(putter)
             try:
                 await putter
+                # see gh-83055, clean self._putters.
+                self._putters.remove(putter)
             except:
                 putter.cancel()  # Just in case putter is not done yet.
                 try:
@@ -132,15 +143,26 @@ async def put(self, item):
                     # the call.  Wake up the next in line.
                     self._wakeup_next(self._putters)
                 raise
-        return self.put_nowait(item)
+        # Don't run tests from self._put_nowait().
+        return self._put_and_wakeup_next(item)
 
     def put_nowait(self, item):
         """Put an item into the queue without blocking.
 
         If no free slot is immediately available, raise QueueFull.
+        If there are pending put tasks, raise QueueFullWithPendingPutTasks.
         """
         if self.full():
             raise QueueFull
+        if self._putters:
+            raise QueueFullWithPendingPutTasks
+        self._put_and_wakeup_next(item)
+
+    def _put_and_wakeup_next(self, item):
+        """Put an item into the queue.
+
+        Wakes up next tasks waiting for slots.
+        """
         self._put(item)
         self._unfinished_tasks += 1
         self._finished.clear()
@@ -150,12 +172,16 @@ async def get(self):
         """Remove and return an item from the queue.
 
         If queue is empty, wait until an item is available.
+        If queue is not empty and there are still tasks waiting for items,
+        task has to wait for its time.
         """
-        while self.empty():
+        if self.empty() or self._getters:
             getter = self._get_loop().create_future()
             self._getters.append(getter)
             try:
                 await getter
+                # see gh-83055, clean self._getters.
+                self._getters.remove(getter)
             except:
                 getter.cancel()  # Just in case getter is not done yet.
                 try:
@@ -170,15 +196,26 @@ async def get(self):
                     # the call.  Wake up the next in line.
                     self._wakeup_next(self._getters)
                 raise
-        return self.get_nowait()
+        # Don't run tests from self._get_nowait().
+        return self._get_and_wakeup_next()
 
     def get_nowait(self):
         """Remove and return an item from the queue.
 
-        Return an item if one is immediately available, else raise QueueEmpty.
+        Return an item if one is immediately available, else raise QueueEmpty
+        Return an item if there is no pending get task, else raise QueueEmptyWithPendingGetTasks
         """
         if self.empty():
             raise QueueEmpty
+        if self._getters:
+            raise QueueEmptyWithPendingGetTasks
+        return self._get_and_wakeup_next()
+
+    def _get_and_wakeup_next(self):
+        """Remove and return an item from the queue.
+
+        Wakes up next tasks waiting for slots.
+        """
         item = self._get()
         self._wakeup_next(self._putters)
         return item