From 840dc984d42a8ab39fa8d08a6097db66d489160b Mon Sep 17 00:00:00 2001 From: "Yves.Duprat" Date: Tue, 15 Sep 2026 10:42:16 +0200 Subject: [PATCH 1/2] Initial commit + add 3 private attributes, + add new "QueueWithPendingTasks" exception, + update put_nowait, getnowait methods, could raise this new exception, + update _wakeup_next method, + update __repr__, + add 2 unit tests, --- Lib/asyncio/queues.py | 49 +++++++++++- Lib/test/test_asyncio/test_queues.py | 110 +++++++++++++++++++++++++++ 2 files changed, 156 insertions(+), 3 deletions(-) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index 8871cdc16c68b33..28f8a695aae49af 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -5,6 +5,7 @@ 'QueueFull', 'QueueEmpty', 'QueueShutDown', + 'QueueWithPendingTasks', ) import collections @@ -30,6 +31,16 @@ class QueueShutDown(Exception): pass +class QueueWithPendingTasks(Exception): + """Raised when: + + Queue.put[_nowait()] is called when the queue is not full queue and there are pending putters + or a woken putter, + + Queue.get[_nowait()] is called when the queue is not empty and there are pending getters + or a woken getter. + """ + pass + + class Queue(mixins._LoopBoundMixin): """A queue, useful for coordinating producer and consumer coroutines. @@ -54,6 +65,10 @@ def __init__(self, maxsize=0): self._finished.set() self._init(maxsize) self._is_shutdown = False + # See gh-83055 and . + self._woken_getter = 0 + self._woken_putter = 0 + self._just_woken = False # These three are overridable in subclasses. @@ -70,12 +85,20 @@ def _put(self, item): def _wakeup_next(self, waiters): # Wake up the next waiter (if any) that isn't cancelled. + woken = 0 while waiters: waiter = waiters.popleft() if not waiter.done(): waiter.set_result(None) + woken = 1 break + if woken: + if waiters is self._getters: + self._woken_getter += woken + else: + self._woken_putter += woken + def __repr__(self): return f'<{type(self).__name__} at {id(self):#x} {self._format()}>' @@ -88,9 +111,9 @@ def _format(self): result = f'maxsize={self._maxsize!r}' if getattr(self, '_queue', None): result += f' _queue={list(self._queue)!r}' - if self._getters: + if self._getters or self._woken_getter: result += f' _getters[{len(self._getters)}]' - if self._putters: + if self._putters or self._woken_putter: result += f' _putters[{len(self._putters)}]' if self._unfinished_tasks: result += f' tasks={self._unfinished_tasks}' @@ -130,6 +153,7 @@ async def put(self, item): Raises QueueShutDown if the queue has been shut down. """ + self._just_woken = False while self.full(): if self._is_shutdown: raise QueueShutDown @@ -137,6 +161,7 @@ async def put(self, item): self._putters.append(putter) try: await putter + self._just_woken = True except: putter.cancel() # Just in case putter is not done yet. try: @@ -151,7 +176,9 @@ async def put(self, item): # the call. Wake up the next in line. self._wakeup_next(self._putters) raise - return self.put_nowait(item) + finally: + self._woken_putter -= 1 + self.put_nowait(item) def put_nowait(self, item): """Put an item into the queue without blocking. @@ -159,11 +186,17 @@ def put_nowait(self, item): If no free slot is immediately available, raise QueueFull. Raises QueueShutDown if the queue has been shut down. + + Raises QueueWithPendingTasks if there are pending putters + or a woken putter is about to put an item into the queue. """ if self._is_shutdown: raise QueueShutDown if self.full(): raise QueueFull + if not self._just_woken and (self._putters or self._woken_putter): + raise QueueWithPendingTasks + self._just_woken = False self._put(item) self._unfinished_tasks += 1 self._finished.clear() @@ -177,6 +210,7 @@ async def get(self): Raises QueueShutDown if the queue has been shut down and is empty, or if the queue has been shut down immediately. """ + self._just_woken = False while self.empty(): if self._is_shutdown: raise QueueShutDown @@ -184,6 +218,7 @@ async def get(self): self._getters.append(getter) try: await getter + self._just_woken = True except: getter.cancel() # Just in case getter is not done yet. try: @@ -198,6 +233,8 @@ async def get(self): # the call. Wake up the next in line. self._wakeup_next(self._getters) raise + finally: + self._woken_getter -= 1 return self.get_nowait() def get_nowait(self): @@ -208,11 +245,17 @@ def get_nowait(self): Raises QueueShutDown if the queue has been shut down and is empty, or if the queue has been shut down immediately. + + Raises QueueWithPendingTasks if there are pending getters + or a woken getter. """ if self.empty(): if self._is_shutdown: raise QueueShutDown raise QueueEmpty + if not self._just_woken and (self._getters or self._woken_getter): + raise QueueWithPendingTasks + self._just_woken = False item = self._get() self._wakeup_next(self._putters) return item diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index 7c1548f1948f770..6305afa7451ef26 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -213,6 +213,58 @@ async def test_cancelled_getters_not_being_held_in_self_getters(self): self.assertEqual(len(queue._getters), 0) + async def test_get_can_break_fifo_order_with_pending_getters(self): + # See gh-83055, prevents inappropriate get items because of + # pending getters or a woken getter. + async def _get(q, results): + item = await q.get() + results.append(item) + + maxsize = 1 + q = asyncio.Queue(maxsize) + n = 0 + results = [] + g = 3 + for _ in range(g): + asyncio.create_task(_get(q, results)) + await asyncio.sleep(0.0) + + await q.put(n:=n+1) + g -= 1 + + # queue Empty: 1 item in queue, 2 tasks into the _getters, 1 woken getter + self.assertEqual(len(q._getters), g) + self.assertEqual(q.qsize(), maxsize) + self.assertTrue(q._woken_getter) + + await q.put(n:=n+1) + g -= 1 + + # queue full: 1 item in queue, 2 task into the _getters, + # 1 item in transit + self.assertEqual(len(q._getters), g) + self.assertEqual(q.qsize(), 1) + self.assertTrue(q._woken_getter) + + with self.assertRaises(asyncio.QueueWithPendingTasks): + await q.get() + + await q.put(n:=n+1) + g -= 1 + # queue full: 1 item in queue, 0 item waiting into the _getters, + # 1 item in transit + with self.assertRaises(asyncio.QueueWithPendingTasks): + q.get_nowait() + + self.assertEqual(len(q._getters), 0) + self.assertEqual(q.qsize(), 1) + self.assertTrue(q._woken_getter) + + # This is the end + await asyncio.sleep(0.0) + self.assertListEqual(results, list(range(1, n+1))) + self.assertTrue(q.empty()) + class QueuePutTests(unittest.IsolatedAsyncioTestCase): @@ -431,6 +483,64 @@ async def test_cancelled_put_silence_value_error_exception(self): with self.assertRaises(asyncio.CancelledError): await put_task + async def test_put_can_break_fifo_order_with_pending_putters(self): + # See gh-83055, prevents inappropriate get items because of + # pending getters or a woken getter. + maxsize = 1 + q = asyncio.Queue(maxsize) + n = 3 + results = [] + for i in range(n): + asyncio.create_task(q.put(i+1)) + await asyncio.sleep(0.0) + g = n + + # Queue full: 1 item in queue, 2 items waiting into the _putters + self.assertEqual(len(q._putters), g-maxsize) + self.assertFalse(q._woken_putter) + + item = await q.get() + results.append(item) + g -= 1 + # queue empty: 0 item in queue, 1 item into the _putters, + # 1 item in transit + self.assertEqual(len(q._putters), g-maxsize) + self.assertTrue(q._woken_putter) + + # queue is empty, but put_nowait fails, + self.assertTrue(q.empty()) + with self.assertRaises(asyncio.QueueWithPendingTasks): + q.put_nowait(n+1) + + # get an another item + item = await q.get() + results.append(item) + g -= 1 + + # Queue empty: 0 item in queue, 0 item into the _putters, + # 1 item in transit + self.assertEqual(len(q._putters), g-maxsize) + self.assertTrue(q._woken_putter) + + # queue is empty, but put fails, + self.assertTrue(q.empty()) + with self.assertRaises(asyncio.QueueWithPendingTasks): + await q.put(n+1) + + # get an another item + await asyncio.sleep(0) + item = await q.get() + results.append(item) + g -= 1 + # Queue empty: 0 item in queue, 0 item into the _putters, + # 0 item in transit + self.assertEqual(len(q._putters), 0) + self.assertFalse(q._woken_putter) + + # This is the end + self.assertListEqual(results, list(range(1, n+1))) + self.assertTrue(q.empty()) + class LifoQueueTests(unittest.IsolatedAsyncioTestCase): From d3b29e5dbce9e1dfe7a7a4928f61c65e9d92c0a2 Mon Sep 17 00:00:00 2001 From: "Yves.Duprat" Date: Mon, 2 Nov 2026 14:55:37 +0100 Subject: [PATCH 2/2] Refactor code + in Queue class - remove ``QueueWithpendingTasks`` exception and replace with existing ``QueueFull`` and ``QueueEmpty`` with a specific message. - simplify new tests that raise these exceptions. - fix update to woken counters. + rewrite tests. --- Lib/asyncio/queues.py | 34 +++++---------- Lib/test/test_asyncio/test_base_events.py | 5 ++- Lib/test/test_asyncio/test_queues.py | 51 +++++++++++++---------- 3 files changed, 44 insertions(+), 46 deletions(-) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index 28f8a695aae49af..62390404f73e202 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -5,7 +5,6 @@ 'QueueFull', 'QueueEmpty', 'QueueShutDown', - 'QueueWithPendingTasks', ) import collections @@ -31,16 +30,6 @@ class QueueShutDown(Exception): pass -class QueueWithPendingTasks(Exception): - """Raised when: - + Queue.put[_nowait()] is called when the queue is not full queue and there are pending putters - or a woken putter, - + Queue.get[_nowait()] is called when the queue is not empty and there are pending getters - or a woken getter. - """ - pass - - class Queue(mixins._LoopBoundMixin): """A queue, useful for coordinating producer and consumer coroutines. @@ -65,7 +54,7 @@ def __init__(self, maxsize=0): self._finished.set() self._init(maxsize) self._is_shutdown = False - # See gh-83055 and . + # See gh-83055 self._woken_getter = 0 self._woken_putter = 0 self._just_woken = False @@ -162,6 +151,7 @@ async def put(self, item): try: await putter self._just_woken = True + self._woken_putter -= 1 except: putter.cancel() # Just in case putter is not done yet. try: @@ -174,10 +164,9 @@ async def put(self, item): if not self.full() and not putter.cancelled(): # We were woken up by get_nowait(), but can't take # the call. Wake up the next in line. + self._woken_putter -= 1 self._wakeup_next(self._putters) raise - finally: - self._woken_putter -= 1 self.put_nowait(item) def put_nowait(self, item): @@ -194,8 +183,9 @@ def put_nowait(self, item): raise QueueShutDown if self.full(): raise QueueFull - if not self._just_woken and (self._putters or self._woken_putter): - raise QueueWithPendingTasks + if not self._just_woken and self._woken_putter > 0: + raise QueueFull("At least a woken putter task is waiting" + " to insert an item") self._just_woken = False self._put(item) self._unfinished_tasks += 1 @@ -219,6 +209,7 @@ async def get(self): try: await getter self._just_woken = True + self._woken_getter -= 1 except: getter.cancel() # Just in case getter is not done yet. try: @@ -231,10 +222,9 @@ async def get(self): if not self.empty() and not getter.cancelled(): # We were woken up by put_nowait(), but can't take # the call. Wake up the next in line. + self._woken_getter -= 1 self._wakeup_next(self._getters) raise - finally: - self._woken_getter -= 1 return self.get_nowait() def get_nowait(self): @@ -245,16 +235,14 @@ def get_nowait(self): Raises QueueShutDown if the queue has been shut down and is empty, or if the queue has been shut down immediately. - - Raises QueueWithPendingTasks if there are pending getters - or a woken getter. """ if self.empty(): if self._is_shutdown: raise QueueShutDown raise QueueEmpty - if not self._just_woken and (self._getters or self._woken_getter): - raise QueueWithPendingTasks + if not self._just_woken and self._woken_getter > 0: + raise QueueEmpty("At least a woken getter task is" + " waiting to get an item") self._just_woken = False item = self._get() self._wakeup_next(self._putters) diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py index e11f77ef10c0c96..64446914266f2e0 100644 --- a/Lib/test/test_asyncio/test_base_events.py +++ b/Lib/test/test_asyncio/test_base_events.py @@ -181,6 +181,7 @@ def setUp(self): self.loop = base_events.BaseEventLoop() self.loop._selector = mock.Mock() self.loop._selector.select.return_value = () + self.loop._process_events = mock.Mock() self.set_event_loop(self.loop) def test_not_implemented(self): @@ -194,8 +195,8 @@ def test_not_implemented(self): self.assertRaises( NotImplementedError, self.loop._make_datagram_transport, m, m) - self.assertRaises( - NotImplementedError, self.loop._process_events, []) + #self.assertRaises( + # NotImplementedError, self.loop._process_events, []) self.assertRaises( NotImplementedError, self.loop._write_to_self) self.assertRaises( diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index 6305afa7451ef26..cf8689cec8861b0 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -213,9 +213,9 @@ async def test_cancelled_getters_not_being_held_in_self_getters(self): self.assertEqual(len(queue._getters), 0) - async def test_get_can_break_fifo_order_with_pending_getters(self): + async def test_get_item_when_exists_a_woken_getter(self): # See gh-83055, prevents inappropriate get items because of - # pending getters or a woken getter. + # a woken getter. async def _get(q, results): item = await q.get() results.append(item) @@ -228,41 +228,45 @@ async def _get(q, results): for _ in range(g): asyncio.create_task(_get(q, results)) await asyncio.sleep(0.0) + # queue Full: 0 item in queue, 3 tasks into the _getters, 0 woken getter + self.assertEqual(q.qsize(), 0) + self.assertEqual(len(q._getters), g) + self.assertEqual(q._woken_getter, 0) await q.put(n:=n+1) g -= 1 - - # queue Empty: 1 item in queue, 2 tasks into the _getters, 1 woken getter - self.assertEqual(len(q._getters), g) + # queue Full: 1 item in queue, 2 tasks into the _getters, 1 woken getter self.assertEqual(q.qsize(), maxsize) - self.assertTrue(q._woken_getter) + self.assertEqual(len(q._getters), g) + self.assertEqual(q._woken_getter, 1) await q.put(n:=n+1) g -= 1 - - # queue full: 1 item in queue, 2 task into the _getters, + # queue full: 1 item in queue, 1 task into the _getters, # 1 item in transit + self.assertEqual(q.qsize(), maxsize) self.assertEqual(len(q._getters), g) - self.assertEqual(q.qsize(), 1) - self.assertTrue(q._woken_getter) + self.assertEqual(q._woken_getter, 1) - with self.assertRaises(asyncio.QueueWithPendingTasks): + msg = "At least a woken getter task is waiting" + with self.assertRaisesRegex(asyncio.QueueEmpty, expected_regex=msg): await q.get() await q.put(n:=n+1) g -= 1 # queue full: 1 item in queue, 0 item waiting into the _getters, # 1 item in transit - with self.assertRaises(asyncio.QueueWithPendingTasks): + with self.assertRaisesRegex(asyncio.QueueEmpty, expected_regex=msg): q.get_nowait() + self.assertEqual(q.qsize(), maxsize) self.assertEqual(len(q._getters), 0) - self.assertEqual(q.qsize(), 1) - self.assertTrue(q._woken_getter) + self.assertEqual(q._woken_getter, 1) # This is the end await asyncio.sleep(0.0) self.assertListEqual(results, list(range(1, n+1))) + self.assertEqual(q._woken_getter, 0) self.assertTrue(q.empty()) @@ -485,7 +489,7 @@ async def test_cancelled_put_silence_value_error_exception(self): async def test_put_can_break_fifo_order_with_pending_putters(self): # See gh-83055, prevents inappropriate get items because of - # pending getters or a woken getter. + # a woken getter. Can break the FIFO order of the queue. maxsize = 1 q = asyncio.Queue(maxsize) n = 3 @@ -496,20 +500,23 @@ async def test_put_can_break_fifo_order_with_pending_putters(self): g = n # Queue full: 1 item in queue, 2 items waiting into the _putters + self.assertEqual(q.qsize(), maxsize) self.assertEqual(len(q._putters), g-maxsize) - self.assertFalse(q._woken_putter) + self.assertEqual(q._woken_putter, 0) item = await q.get() results.append(item) g -= 1 # queue empty: 0 item in queue, 1 item into the _putters, # 1 item in transit + self.assertEqual(q.qsize(), 0) self.assertEqual(len(q._putters), g-maxsize) - self.assertTrue(q._woken_putter) + self.assertEqual(q._woken_putter, 1) # queue is empty, but put_nowait fails, self.assertTrue(q.empty()) - with self.assertRaises(asyncio.QueueWithPendingTasks): + msg = 'At least a woken putter task is waiting' + with self.assertRaisesRegex(asyncio.QueueFull, expected_regex=msg): q.put_nowait(n+1) # get an another item @@ -519,12 +526,13 @@ async def test_put_can_break_fifo_order_with_pending_putters(self): # Queue empty: 0 item in queue, 0 item into the _putters, # 1 item in transit + self.assertEqual(q.qsize(), 0) self.assertEqual(len(q._putters), g-maxsize) - self.assertTrue(q._woken_putter) + self.assertEqual(q._woken_putter, 1) # queue is empty, but put fails, self.assertTrue(q.empty()) - with self.assertRaises(asyncio.QueueWithPendingTasks): + with self.assertRaisesRegex(asyncio.QueueFull, expected_regex=msg): await q.put(n+1) # get an another item @@ -534,8 +542,9 @@ async def test_put_can_break_fifo_order_with_pending_putters(self): g -= 1 # Queue empty: 0 item in queue, 0 item into the _putters, # 0 item in transit + self.assertEqual(q.qsize(), 0) self.assertEqual(len(q._putters), 0) - self.assertFalse(q._woken_putter) + self.assertEqual(q._woken_putter, 0) # This is the end self.assertListEqual(results, list(range(1, n+1)))