From e9c1a3e381f5d919d837eba9ade321719d679f18 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 11:31:55 +0200 Subject: [PATCH 01/88] Add `asyncio.Queue.__aiter__()` Co-authored-by: Guido van Rossum --- Lib/asyncio/queues.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index 2f3865114a84f95..9bc07e1e0f70dcb 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -30,6 +30,22 @@ class QueueShutDown(Exception): pass +class AsyncQueueIterator: + def __init__(self, queue): + self.queue = queue + + def __aiter__(self): + return self + + async def __anext__(self): + try: + item = await self.queue.get() + except QueueShutDown: + raise StopAsyncIteration + else: + return item + + class Queue(mixins._LoopBoundMixin): """A queue, useful for coordinating producer and consumer coroutines. @@ -76,6 +92,9 @@ def _wakeup_next(self, waiters): waiter.set_result(None) break + def __aiter__(self): + return AsyncQueueIterator(self) + def __repr__(self): return f'<{type(self).__name__} at {id(self):#x} {self._format()}>' @@ -200,6 +219,7 @@ async def get(self): raise return self.get_nowait() + def get_nowait(self): """Remove and return an item from the queue. From 3cd59ee95949448cef6e9adb68897c4788390ed5 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 11:47:40 +0200 Subject: [PATCH 02/88] Reduce diff --- Lib/asyncio/queues.py | 1 - 1 file changed, 1 deletion(-) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index 9bc07e1e0f70dcb..393a08c2135fa71 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -219,7 +219,6 @@ async def get(self): raise return self.get_nowait() - def get_nowait(self): """Remove and return an item from the queue. From 16fc429f43738f546db8b94613c5b48f8e0c3cda Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 12:18:57 +0200 Subject: [PATCH 03/88] Add test --- Lib/test/test_asyncio/test_queues.py | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index 5019e9a293525d1..3da6c0c0f8c1379 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -471,27 +471,23 @@ async def test_task_done(self): # Two workers get items from the queue and call task_done after each. # Join the queue and assert all items have been processed. - running = True async def worker(): nonlocal accumulator - while running: - item = await q.get() + async for item in q: accumulator += item q.task_done() async with asyncio.TaskGroup() as tg: - tasks = [tg.create_task(worker()) - for index in range(2)] + for _ in range(2): + tg.create_task(worker()) await q.join() self.assertEqual(sum(range(100)), accumulator) # close running generators - running = False - for i in range(len(tasks)): - q.put_nowait(0) + q.shutdown() async def test_join_empty_queue(self): q = self.q_class() From f13ddb3912e1e6efbf613f80ea7b51993a86f246 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 12:24:04 +0200 Subject: [PATCH 04/88] Simplify code --- Lib/asyncio/queues.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index 393a08c2135fa71..c82147da02ac72d 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -39,11 +39,9 @@ def __aiter__(self): async def __anext__(self): try: - item = await self.queue.get() + return await self.queue.get() except QueueShutDown: raise StopAsyncIteration - else: - return item class Queue(mixins._LoopBoundMixin): From afd7ad4a0ea2d3535946ba775d3fc2cb0cf04629 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 12:26:00 +0200 Subject: [PATCH 05/88] Add to `__all__` --- Lib/asyncio/queues.py | 1 + 1 file changed, 1 insertion(+) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index c82147da02ac72d..33abb84b8cce0d6 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -5,6 +5,7 @@ 'QueueFull', 'QueueEmpty', 'QueueShutDown', + 'AsyncQueueIterator', ) import collections From 214548c4b2a774b0fef2098ec59a51d7337fbccc Mon Sep 17 00:00:00 2001 From: "blurb-it[bot]" <43283697+blurb-it[bot]@users.noreply.github.com> Date: Fri, 14 Jun 2024 10:30:44 +0000 Subject: [PATCH 06/88] =?UTF-8?q?=F0=9F=93=9C=F0=9F=A4=96=20Added=20by=20b?= =?UTF-8?q?lurb=5Fit.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst | 1 + 1 file changed, 1 insertion(+) create mode 100644 Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst diff --git a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst new file mode 100644 index 000000000000000..21d2f98fb6b6226 --- /dev/null +++ b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst @@ -0,0 +1 @@ +Make :class:`asyncio.Queue` an asynchronous iterable. From 9824b639a8b001db0f3b94e2a2319dc32f2d5110 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 12:50:10 +0200 Subject: [PATCH 07/88] Simplify example --- Doc/library/asyncio-queue.rst | 72 ++++++++++++++++------------------- 1 file changed, 32 insertions(+), 40 deletions(-) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index 9b579cc1d5fdfe5..2f403f0457ec590 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -191,50 +191,42 @@ concurrent tasks:: async def worker(name, queue): - while True: - # Get a "work item" out of the queue. - sleep_for = await queue.get() + # Get a "work item" out of the queue. + async for sleep_for in queue: + # Sleep for the "sleep_for" seconds. + await asyncio.sleep(sleep_for) - # Sleep for the "sleep_for" seconds. - await asyncio.sleep(sleep_for) - - # Notify the queue that the "work item" has been processed. - queue.task_done() - - print(f'{name} has slept for {sleep_for:.2f} seconds') + print(f'{name} has slept for {sleep_for:.2f} seconds') async def main(): - # Create a queue that we will use to store our "workload". - queue = asyncio.Queue() - - # Generate random timings and put them into the queue. - total_sleep_time = 0 - for _ in range(20): - sleep_for = random.uniform(0.05, 1.0) - total_sleep_time += sleep_for - queue.put_nowait(sleep_for) - - # Create three worker tasks to process the queue concurrently. - tasks = [] - for i in range(3): - task = asyncio.create_task(worker(f'worker-{i}', queue)) - tasks.append(task) - - # Wait until the queue is fully processed. - started_at = time.monotonic() - await queue.join() - total_slept_for = time.monotonic() - started_at - - # Cancel our worker tasks. - for task in tasks: - task.cancel() - # Wait until all worker tasks are cancelled. - await asyncio.gather(*tasks, return_exceptions=True) - - print('====') - print(f'3 workers slept in parallel for {total_slept_for:.2f} seconds') - print(f'total expected sleep time: {total_sleep_time:.2f} seconds') + # Create a queue that we will use to store our "workload". + queue = asyncio.Queue() + + # Generate random timings and put them into the queue. + total_sleep_time = 0 + for _ in range(20): + sleep_for = random.uniform(0.05, 1.0) + total_sleep_time += sleep_for + queue.put_nowait(sleep_for) + + # All tasks have been queued + queue.shutdown() + + # Create three worker tasks to process the queue concurrently. + tasks = [] + for i in range(3): + task = asyncio.create_task(worker(f'worker-{i}', queue)) + tasks.append(task) + + # Wait until the queue is fully processed. + started_at = time.monotonic() + await asyncio.gather(*tasks, return_exceptions=True) + total_slept_for = time.monotonic() - started_at + + print('====') + print(f'3 workers slept in parallel for {total_slept_for:.2f} seconds') + print(f'total expected sleep time: {total_sleep_time:.2f} seconds') asyncio.run(main()) From 6358b8ec6bc4b8665b3a47a8a48cfdfe2b8875c6 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 12:57:49 +0200 Subject: [PATCH 08/88] Fix indentation --- Doc/library/asyncio-queue.rst | 64 +++++++++++++++++------------------ 1 file changed, 32 insertions(+), 32 deletions(-) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index 2f403f0457ec590..ec27bbf2c80ef7e 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -191,42 +191,42 @@ concurrent tasks:: async def worker(name, queue): - # Get a "work item" out of the queue. - async for sleep_for in queue: - # Sleep for the "sleep_for" seconds. - await asyncio.sleep(sleep_for) + # Get a "work item" out of the queue. + async for sleep_for in queue: + # Sleep for the "sleep_for" seconds. + await asyncio.sleep(sleep_for) - print(f'{name} has slept for {sleep_for:.2f} seconds') + print(f'{name} has slept for {sleep_for:.2f} seconds') async def main(): - # Create a queue that we will use to store our "workload". - queue = asyncio.Queue() - - # Generate random timings and put them into the queue. - total_sleep_time = 0 - for _ in range(20): - sleep_for = random.uniform(0.05, 1.0) - total_sleep_time += sleep_for - queue.put_nowait(sleep_for) - - # All tasks have been queued - queue.shutdown() - - # Create three worker tasks to process the queue concurrently. - tasks = [] - for i in range(3): - task = asyncio.create_task(worker(f'worker-{i}', queue)) - tasks.append(task) - - # Wait until the queue is fully processed. - started_at = time.monotonic() - await asyncio.gather(*tasks, return_exceptions=True) - total_slept_for = time.monotonic() - started_at - - print('====') - print(f'3 workers slept in parallel for {total_slept_for:.2f} seconds') - print(f'total expected sleep time: {total_sleep_time:.2f} seconds') + # Create a queue that we will use to store our "workload". + queue = asyncio.Queue() + + # Generate random timings and put them into the queue. + total_sleep_time = 0 + for _ in range(20): + sleep_for = random.uniform(0.05, 1.0) + total_sleep_time += sleep_for + queue.put_nowait(sleep_for) + + # All tasks have been queued + queue.shutdown() + + # Create three worker tasks to process the queue concurrently. + tasks = [] + for i in range(3): + task = asyncio.create_task(worker(f'worker-{i}', queue)) + tasks.append(task) + + # Wait until the queue is fully processed. + started_at = time.monotonic() + await asyncio.gather(*tasks, return_exceptions=True) + total_slept_for = time.monotonic() - started_at + + print('====') + print(f'3 workers slept in parallel for {total_slept_for:.2f} seconds') + print(f'total expected sleep time: {total_sleep_time:.2f} seconds') asyncio.run(main()) From 316b8006ef5c675a65758c660e00b700e654138a Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 13:11:22 +0200 Subject: [PATCH 09/88] Improve order --- Lib/test/test_asyncio/test_queues.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index 3da6c0c0f8c1379..859f746d791b35b 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -467,6 +467,9 @@ async def test_task_done(self): for i in range(100): q.put_nowait(i) + # All tasks have been queued + q.shutdown() + accumulator = 0 # Two workers get items from the queue and call task_done after each. @@ -486,9 +489,6 @@ async def worker(): await q.join() self.assertEqual(sum(range(100)), accumulator) - # close running generators - q.shutdown() - async def test_join_empty_queue(self): q = self.q_class() From 92f32954db74d624f37306d427b6d710b588dc42 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 13:21:45 +0200 Subject: [PATCH 10/88] Test `StopAsyncIteration` --- Lib/test/test_asyncio/test_queues.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index 859f746d791b35b..6e230d649cd361b 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -116,6 +116,30 @@ async def putter(): self.assertTrue(t.done()) self.assertTrue(t.result()) + async def test_aiter(self): + q = asyncio.Queue() + for i in range(100): + q.put_nowait(i) + + # All tasks have been queued + q.shutdown() + + accumulator = 0 + + # Two workers get items from the queue and call task_done after each. + # Join the queue and assert all items have been processed. + + async def worker(): + nonlocal accumulator + + async for item in q: + accumulator += item + + async with asyncio.TaskGroup() as tg: + tasks = [tg.create_task(worker()) for _ in range(2)] + await asyncio.gather(*tasks, return_exceptions=True) + self.assertEqual(sum(range(100)), accumulator) + class QueueGetTests(unittest.IsolatedAsyncioTestCase): From a18767df829a4f65fe921a3a043c44810f5dbbd2 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 13:28:47 +0200 Subject: [PATCH 11/88] Document `AsyncQueueIterator` --- Doc/library/asyncio-queue.rst | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index ec27bbf2c80ef7e..dece95247dc3427 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -133,6 +133,21 @@ Queue items placed in the queue. +Async Queue Iterator +==================== + +.. class:: AsyncQueueIterator(queue) + + The :term:`asynchronous iterator` returned by calling :func:`aiter` + on a :class:`Queue`. + +.. attribute:: queue + + The queue backed by the ``AsyncQueueIterator``. + +.. versionadded:: 3.14 + + Priority Queue ============== From e826a01248a840f05182c97536f85e089ec584cf Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 15:28:10 +0200 Subject: [PATCH 12/88] Add `queue.Queue.__iter__()` --- Lib/queue.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/Lib/queue.py b/Lib/queue.py index 25beb46e30d6bd9..a52c3984f404f0d 100644 --- a/Lib/queue.py +++ b/Lib/queue.py @@ -275,6 +275,13 @@ def _put(self, item): def _get(self): return self.queue.popleft() + def __iter__(self): + while True: + try: + yield self.get_nowait() + except (Empty, ShutDown): + return + __class_getitem__ = classmethod(types.GenericAlias) From c8cbe4d0b1a48d31bfaa2dba3f3874c679a02dbe Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 15:34:19 +0200 Subject: [PATCH 13/88] Add `asyncio.Queue.__iter__()` --- Lib/asyncio/queues.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index 2f3865114a84f95..bc1bc1142526564 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -76,6 +76,13 @@ def _wakeup_next(self, waiters): waiter.set_result(None) break + def __iter__(self): + while True: + try: + yield self.get_nowait() + except (QueueEmpty, QueueShutDown): + return + def __repr__(self): return f'<{type(self).__name__} at {id(self):#x} {self._format()}>' From 940e3c0ee41835d62771bfaf00dfc3168ea1206c Mon Sep 17 00:00:00 2001 From: "blurb-it[bot]" <43283697+blurb-it[bot]@users.noreply.github.com> Date: Fri, 14 Jun 2024 13:37:47 +0000 Subject: [PATCH 14/88] =?UTF-8?q?=F0=9F=93=9C=F0=9F=A4=96=20Added=20by=20b?= =?UTF-8?q?lurb=5Fit.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst | 1 + 1 file changed, 1 insertion(+) create mode 100644 Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst diff --git a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst new file mode 100644 index 000000000000000..bda2559146f42a0 --- /dev/null +++ b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst @@ -0,0 +1 @@ +Make :class:`queue.Queue` and `asyncio.Queue` iterables. From 8fca2888da1ae26bf442723b922617be76b2f1e2 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Fri, 14 Jun 2024 17:59:19 +0200 Subject: [PATCH 15/88] Fix lint --- .../2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst index bda2559146f42a0..779814d180fbe45 100644 --- a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst +++ b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst @@ -1 +1 @@ -Make :class:`queue.Queue` and `asyncio.Queue` iterables. +Make :class:`queue.Queue` and :class:`asyncio.Queue` iterables. From 499457c1993c69f6cf7ff33ac94dcf082fc6b039 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Fri, 14 Jun 2024 19:36:33 +0200 Subject: [PATCH 16/88] Wait --- Lib/asyncio/queues.py | 7 ------- Lib/queue.py | 4 ++-- 2 files changed, 2 insertions(+), 9 deletions(-) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index bc1bc1142526564..2f3865114a84f95 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -76,13 +76,6 @@ def _wakeup_next(self, waiters): waiter.set_result(None) break - def __iter__(self): - while True: - try: - yield self.get_nowait() - except (QueueEmpty, QueueShutDown): - return - def __repr__(self): return f'<{type(self).__name__} at {id(self):#x} {self._format()}>' diff --git a/Lib/queue.py b/Lib/queue.py index a52c3984f404f0d..a01ccea6cb3b8cb 100644 --- a/Lib/queue.py +++ b/Lib/queue.py @@ -278,8 +278,8 @@ def _get(self): def __iter__(self): while True: try: - yield self.get_nowait() - except (Empty, ShutDown): + yield self.get() + except ShutDown: return __class_getitem__ = classmethod(types.GenericAlias) From 7f206d98a7928a4531a84082d5874e8b6296d0f6 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Fri, 14 Jun 2024 19:37:02 +0200 Subject: [PATCH 17/88] Update 2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst --- .../2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst index 779814d180fbe45..112317b8ac4b91a 100644 --- a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst +++ b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst @@ -1 +1 @@ -Make :class:`queue.Queue` and :class:`asyncio.Queue` iterables. +Make :class:`queue.Queue` an iterable. From 320691cd422e5a3f0c323985d5db399a524badcf Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Sat, 15 Jun 2024 06:36:54 +0200 Subject: [PATCH 18/88] Apply suggestions from code review Co-authored-by: Zac Hatfield-Dodds --- Doc/library/asyncio-queue.rst | 2 +- Lib/test/test_asyncio/test_queues.py | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index dece95247dc3427..3fbbdf8a8e42515 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -143,7 +143,7 @@ Async Queue Iterator .. attribute:: queue - The queue backed by the ``AsyncQueueIterator``. + The queue which backs this ``AsyncQueueIterator``. .. versionadded:: 3.14 diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index 6e230d649cd361b..ca9a92b8e79e54d 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -136,9 +136,9 @@ async def worker(): accumulator += item async with asyncio.TaskGroup() as tg: - tasks = [tg.create_task(worker()) for _ in range(2)] - await asyncio.gather(*tasks, return_exceptions=True) - self.assertEqual(sum(range(100)), accumulator) + tg.create_task(worker()) + tg.create_task(worker()) + self.assertEqual(sum(range(100)), accumulator) class QueueGetTests(unittest.IsolatedAsyncioTestCase): From f43bdb17235fee0bdf6680854d01affba49d0ac5 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sat, 15 Jun 2024 06:47:18 +0200 Subject: [PATCH 19/88] Unpack for loop --- Lib/test/test_asyncio/test_queues.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index ca9a92b8e79e54d..9211ecefba6dd9a 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -507,9 +507,8 @@ async def worker(): q.task_done() async with asyncio.TaskGroup() as tg: - for _ in range(2): - tg.create_task(worker()) - + tg.create_task(worker()) + tg.create_task(worker()) await q.join() self.assertEqual(sum(range(100)), accumulator) From 4ca4a93e18a239521da1a359bd8af6e9020a873c Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sat, 15 Jun 2024 06:48:44 +0200 Subject: [PATCH 20/88] Insert newline --- Lib/test/test_asyncio/test_queues.py | 1 + 1 file changed, 1 insertion(+) diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index 9211ecefba6dd9a..4f9abcd0b6a9730 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -138,6 +138,7 @@ async def worker(): async with asyncio.TaskGroup() as tg: tg.create_task(worker()) tg.create_task(worker()) + self.assertEqual(sum(range(100)), accumulator) From fffacb2cbfd5ca588bfa348068ca101edd628925 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sat, 15 Jun 2024 07:00:22 +0200 Subject: [PATCH 21/88] Simplify example --- Doc/library/asyncio-queue.rst | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index 3fbbdf8a8e42515..047c45a533ec14b 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -229,14 +229,11 @@ concurrent tasks:: queue.shutdown() # Create three worker tasks to process the queue concurrently. - tasks = [] - for i in range(3): - task = asyncio.create_task(worker(f'worker-{i}', queue)) - tasks.append(task) - - # Wait until the queue is fully processed. started_at = time.monotonic() - await asyncio.gather(*tasks, return_exceptions=True) + async with asyncio.TaskGroup() as tg: + for i in range(3): + tg.create_task(worker(f'worker-{i}', queue)) + total_slept_for = time.monotonic() - started_at print('====') From 92d7b16d0f511b49e40a709a4144d2b416572db7 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sat, 15 Jun 2024 07:03:16 +0200 Subject: [PATCH 22/88] Indent correctly --- Doc/library/asyncio-queue.rst | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index 047c45a533ec14b..1a5926183da0c2d 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -141,11 +141,11 @@ Async Queue Iterator The :term:`asynchronous iterator` returned by calling :func:`aiter` on a :class:`Queue`. -.. attribute:: queue + .. versionadded:: 3.14 - The queue which backs this ``AsyncQueueIterator``. + .. attribute:: queue -.. versionadded:: 3.14 + The queue which backs this ``AsyncQueueIterator``. Priority Queue From ce23fcedf1bb3686ed85ab4f9b10e0337ca88409 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sat, 15 Jun 2024 07:29:52 +0200 Subject: [PATCH 23/88] Document `__aiter__()` --- Doc/library/asyncio-queue.rst | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index 1a5926183da0c2d..6a11cabde809cd3 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -46,6 +46,18 @@ Queue Number of items allowed in the queue. + .. method:: __aiter__() + Return an :class:`AsyncQueueIterator` which iterates asynchronously + over the items in this queue until :meth:`shutdown` is called and + the queue is empty:: + + ... # add items to the queue + queue.shutdown() + async for item in queue: + ... # process item + + .. versionadded:: 3.14 + .. method:: empty() Return ``True`` if the queue is empty, ``False`` otherwise. From 640d2a8a59f7385ccd806462cf2a255df032311f Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sat, 15 Jun 2024 07:34:58 +0200 Subject: [PATCH 24/88] Fix indent attempt 1 --- Doc/library/asyncio-queue.rst | 1 + 1 file changed, 1 insertion(+) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index 6a11cabde809cd3..57e73984a259828 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -47,6 +47,7 @@ Queue Number of items allowed in the queue. .. method:: __aiter__() + Return an :class:`AsyncQueueIterator` which iterates asynchronously over the items in this queue until :meth:`shutdown` is called and the queue is empty:: From 3cad1810cbe41fd88c1e8ce5244d152b5156e028 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Tue, 18 Jun 2024 07:49:15 +0200 Subject: [PATCH 25/88] Move while loop inside try/except block --- Lib/queue.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/Lib/queue.py b/Lib/queue.py index a01ccea6cb3b8cb..e392bc66e077833 100644 --- a/Lib/queue.py +++ b/Lib/queue.py @@ -276,11 +276,11 @@ def _get(self): return self.queue.popleft() def __iter__(self): - while True: - try: + try: + while True: yield self.get() - except ShutDown: - return + except ShutDown: + return __class_getitem__ = classmethod(types.GenericAlias) From cc9f7ef91dd4d89165ba8410a6ad144e222915b2 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 08:24:51 +0200 Subject: [PATCH 26/88] test iteration --- Lib/test/test_queue.py | 42 +++++++++++++++++++++++++++++------------- 1 file changed, 29 insertions(+), 13 deletions(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index d5927fbf39142b3..511861ea5dfd3e9 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -152,21 +152,34 @@ def basic_queue_test(self, q): self.do_blocking_test(q.get, (True, 10), q.put, ('empty',)) - def worker(self, q): - while True: - x = q.get() - if x < 0: - q.task_done() - return + def worker(self, q, task_done): + for x in q: with self.cumlock: self.cum += x - q.task_done() + if task_done: + q.task_done() + + def queue_iter_test(self, _): + q = self.type2test() + self.cum = 0 + threads = [] + for i in range(2): + thread = threading.Thread(target=self.worker, args=(q, False)) + thread.start() + threads.append(thread) + for i in range(100): + q.put(i) + q.shutdown() + for thread in threads: + thread.join() + self.assertEqual(self.cum, sum(range(100))) - def queue_join_test(self, q): + def queue_join_test(self, _): + q = self.type2test() self.cum = 0 threads = [] - for i in (0,1): - thread = threading.Thread(target=self.worker, args=(q,)) + for i in range(2): + thread = threading.Thread(target=self.worker, args=(q, True)) thread.start() threads.append(thread) for i in range(100): @@ -174,9 +187,12 @@ def queue_join_test(self, q): q.join() self.assertEqual(self.cum, sum(range(100)), "q.join() did not block until all tasks were done") - for i in (0,1): - q.put(-1) # instruct the threads to close - q.join() # verify that you can join twice + for i in range(100, 200): + q.put(i) + q.shutdown() + q.join() # verify that you can join twice + self.assertEqual(self.cum, sum(range(200)), + "q.join() did not block until all tasks were done") for thread in threads: thread.join() From 92d7dc025519c526caa92c4bf8a851db5ffcd884 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 08:26:06 +0200 Subject: [PATCH 27/88] Reduce diff --- Lib/test/test_queue.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index 511861ea5dfd3e9..0d9913ced5e30b5 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -163,7 +163,7 @@ def queue_iter_test(self, _): q = self.type2test() self.cum = 0 threads = [] - for i in range(2): + for i in (0,1): thread = threading.Thread(target=self.worker, args=(q, False)) thread.start() threads.append(thread) @@ -178,7 +178,7 @@ def queue_join_test(self, _): q = self.type2test() self.cum = 0 threads = [] - for i in range(2): + for i in (0,1): thread = threading.Thread(target=self.worker, args=(q, True)) thread.start() threads.append(thread) From a92e0318f86d017b831aabf58def4b1feb3fd861 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 08:36:04 +0200 Subject: [PATCH 28/88] Run test --- Lib/test/test_queue.py | 23 +++++++++++------------ 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index 0d9913ced5e30b5..b23c3c12d5a19df 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -159,7 +159,7 @@ def worker(self, q, task_done): if task_done: q.task_done() - def queue_iter_test(self, _): + def test_iter(self): q = self.type2test() self.cum = 0 threads = [] @@ -174,8 +174,7 @@ def queue_iter_test(self, _): thread.join() self.assertEqual(self.cum, sum(range(100))) - def queue_join_test(self, _): - q = self.type2test() + def queue_join_test(self, q): self.cum = 0 threads = [] for i in (0,1): @@ -209,15 +208,15 @@ def test_queue_task_done(self): def test_queue_join(self): # Test that a queue join()s successfully, and before anything else # (done twice for insurance). - q = self.type2test() - self.queue_join_test(q) - self.queue_join_test(q) - try: - q.task_done() - except ValueError: - pass - else: - self.fail("Did not detect task count going negative") + for _ in range(2): + q = self.type2test() + self.queue_join_test(q) + try: + q.task_done() + except ValueError: + pass + else: + self.fail("Did not detect task count going negative") def test_basic(self): # Do it a couple of times on the same queue. From 3621cc169d06928f11870944d72089dceb533638 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 08:38:36 +0200 Subject: [PATCH 29/88] Reduce diff --- Lib/test/test_queue.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index b23c3c12d5a19df..6ddd0278afcfc0d 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -189,7 +189,7 @@ def queue_join_test(self, q): for i in range(100, 200): q.put(i) q.shutdown() - q.join() # verify that you can join twice + q.join() # verify that you can join twice self.assertEqual(self.cum, sum(range(200)), "q.join() did not block until all tasks were done") for thread in threads: From 98252c6e925d85fc988cd19b51621d3d478b0970 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 09:18:03 +0200 Subject: [PATCH 30/88] Adapt example from asyncio --- Doc/library/queue.rst | 71 +++++++++++++++++++++++++++---------------- 1 file changed, 45 insertions(+), 26 deletions(-) diff --git a/Doc/library/queue.rst b/Doc/library/queue.rst index fbbebcf4ed8f927..b9871e2bc91cd6e 100644 --- a/Doc/library/queue.rst +++ b/Doc/library/queue.rst @@ -132,6 +132,51 @@ provide the public methods described below. will not block. Similarly, if full() returns ``False`` it doesn't guarantee that a subsequent call to put() will not block. +.. method:: Queue.__iter__() + + Return an :term:`iterator` which iterates over the items in this queue + until :meth:`Queue.shutdown` is called and the queue is empty. + + .. versionadded:: 3.14 + + +Example of how to wait for enqueued tasks to be completed:: + + import concurrent.futures, queue, random, time + + def worker(name, queue): + # Get a "work item" out of the queue. + for sleep_for in queue: + # Sleep for the "sleep_for" seconds. + time.sleep(sleep_for) + + print(f'{name} has slept for {sleep_for:.2f} seconds') + + # Create a queue that we will use to store our "workload". + q = queue.Queue() + + # Generate random timings and put them into the queue. + total_sleep_time = 0 + for _ in range(20): + sleep_for = random.uniform(0.05, 1.0) + total_sleep_time += sleep_for + q.put_nowait(sleep_for) + + # All tasks have been queued + q.shutdown() + + # Create three worker tasks to process the queue concurrently. + started_at = time.monotonic() + with concurrent.futures.ThreadPoolExecutor() as tp: + for i in range(3): + tp.submit(worker, f'worker-{i}', q) + + total_slept_for = time.monotonic() - started_at + + print('====') + print(f'3 workers slept in parallel for {total_slept_for:.2f} seconds') + print(f'total expected sleep time: {total_sleep_time:.2f} seconds') + .. method:: Queue.put(item, block=True, timeout=None) @@ -204,32 +249,6 @@ fully processed by daemon consumer threads. count of unfinished tasks drops to zero, :meth:`join` unblocks. -Example of how to wait for enqueued tasks to be completed:: - - import threading - import queue - - q = queue.Queue() - - def worker(): - while True: - item = q.get() - print(f'Working on {item}') - print(f'Finished {item}') - q.task_done() - - # Turn-on the worker thread. - threading.Thread(target=worker, daemon=True).start() - - # Send thirty task requests to the worker. - for item in range(30): - q.put(item) - - # Block until all tasks are done. - q.join() - print('All work completed') - - Terminating queues ^^^^^^^^^^^^^^^^^^ From 4b1ff4ba47578fdb68e91b6a05325a00ddbd78c3 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 09:23:18 +0200 Subject: [PATCH 31/88] Simpler example --- Doc/library/queue.rst | 39 ++++++++++++++------------------------- 1 file changed, 14 insertions(+), 25 deletions(-) diff --git a/Doc/library/queue.rst b/Doc/library/queue.rst index b9871e2bc91cd6e..667696fb48ab8a2 100644 --- a/Doc/library/queue.rst +++ b/Doc/library/queue.rst @@ -142,40 +142,29 @@ provide the public methods described below. Example of how to wait for enqueued tasks to be completed:: - import concurrent.futures, queue, random, time + import concurrent.futures + import queue - def worker(name, queue): - # Get a "work item" out of the queue. - for sleep_for in queue: - # Sleep for the "sleep_for" seconds. - time.sleep(sleep_for) - - print(f'{name} has slept for {sleep_for:.2f} seconds') - - # Create a queue that we will use to store our "workload". q = queue.Queue() - # Generate random timings and put them into the queue. - total_sleep_time = 0 - for _ in range(20): - sleep_for = random.uniform(0.05, 1.0) - total_sleep_time += sleep_for - q.put_nowait(sleep_for) + # Queue thirty tasks. + for item in range(30): + q.put(item) # All tasks have been queued q.shutdown() - # Create three worker tasks to process the queue concurrently. - started_at = time.monotonic() - with concurrent.futures.ThreadPoolExecutor() as tp: - for i in range(3): - tp.submit(worker, f'worker-{i}', q) + def worker(): + for item in q: + print(f'Working on {item}') + print(f'Finished {item}') - total_slept_for = time.monotonic() - started_at + # Create 2 worker threads. + with concurrent.futures.ThreadPoolExecutor() as tp: + tp.submit(worker) + tp.submit(worker) - print('====') - print(f'3 workers slept in parallel for {total_slept_for:.2f} seconds') - print(f'total expected sleep time: {total_sleep_time:.2f} seconds') + print('All work completed') .. method:: Queue.put(item, block=True, timeout=None) From 5e9f48b76d5d652447761601833b5ad61ed61760 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 09:40:32 +0200 Subject: [PATCH 32/88] Put definition first --- Doc/library/queue.rst | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/Doc/library/queue.rst b/Doc/library/queue.rst index 667696fb48ab8a2..2a1aa44d43c39f1 100644 --- a/Doc/library/queue.rst +++ b/Doc/library/queue.rst @@ -145,6 +145,11 @@ Example of how to wait for enqueued tasks to be completed:: import concurrent.futures import queue + def worker(): + for item in q: + print(f'Working on {item}') + print(f'Finished {item}') + q = queue.Queue() # Queue thirty tasks. @@ -154,11 +159,6 @@ Example of how to wait for enqueued tasks to be completed:: # All tasks have been queued q.shutdown() - def worker(): - for item in q: - print(f'Working on {item}') - print(f'Finished {item}') - # Create 2 worker threads. with concurrent.futures.ThreadPoolExecutor() as tp: tp.submit(worker) From 4b382ec300702fe85a60f49ab5e78e98df8bd360 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 09:47:40 +0200 Subject: [PATCH 33/88] Print worker name --- Doc/library/queue.rst | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/Doc/library/queue.rst b/Doc/library/queue.rst index 2a1aa44d43c39f1..3f5b61ac8382805 100644 --- a/Doc/library/queue.rst +++ b/Doc/library/queue.rst @@ -144,11 +144,12 @@ Example of how to wait for enqueued tasks to be completed:: import concurrent.futures import queue + import time - def worker(): + def worker(name, q): for item in q: - print(f'Working on {item}') - print(f'Finished {item}') + time.sleep(.01) + print(f'{name} finished {item}') q = queue.Queue() @@ -159,10 +160,10 @@ Example of how to wait for enqueued tasks to be completed:: # All tasks have been queued q.shutdown() - # Create 2 worker threads. + # Create 3 worker threads. with concurrent.futures.ThreadPoolExecutor() as tp: - tp.submit(worker) - tp.submit(worker) + for i in range(3): + tp.submit(worker, f'worker-{i}', q) print('All work completed') From 076a7a617f5c62d558f364453003a95e51c05bf5 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Tue, 18 Jun 2024 19:44:14 +0200 Subject: [PATCH 34/88] Restore example --- Doc/library/queue.rst | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/Doc/library/queue.rst b/Doc/library/queue.rst index 3f5b61ac8382805..39a1ced798c5b96 100644 --- a/Doc/library/queue.rst +++ b/Doc/library/queue.rst @@ -239,6 +239,31 @@ fully processed by daemon consumer threads. count of unfinished tasks drops to zero, :meth:`join` unblocks. +Example of how to wait for enqueued tasks to be completed:: + + import threading + import queue + + q = queue.Queue() + + def worker(): + for item in q: + print(f'Working on {item}') + print(f'Finished {item}') + q.task_done() + + # Turn-on the worker thread. + threading.Thread(target=worker, daemon=True).start() + + # Send thirty task requests to the worker. + for item in range(30): + q.put(item) + + # Block until all tasks are done. + q.join() + print('All work completed') + + Terminating queues ^^^^^^^^^^^^^^^^^^ From 6132f5a854eb3caa771f6fc66aa1d0f9262fd59b Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 19:56:27 +0200 Subject: [PATCH 35/88] Indent example --- Doc/library/queue.rst | 40 +++++++++++++++++----------------------- 1 file changed, 17 insertions(+), 23 deletions(-) diff --git a/Doc/library/queue.rst b/Doc/library/queue.rst index 39a1ced798c5b96..3c35f1a1c19a5ce 100644 --- a/Doc/library/queue.rst +++ b/Doc/library/queue.rst @@ -137,35 +137,29 @@ provide the public methods described below. Return an :term:`iterator` which iterates over the items in this queue until :meth:`Queue.shutdown` is called and the queue is empty. - .. versionadded:: 3.14 - - -Example of how to wait for enqueued tasks to be completed:: - - import concurrent.futures - import queue - import time + Example:: - def worker(name, q): - for item in q: - time.sleep(.01) - print(f'{name} finished {item}') + import concurrent.futures + import queue + import time - q = queue.Queue() + def worker(name, q): + for item in q: + time.sleep(.01) + print(f'{name} finished {item}') - # Queue thirty tasks. - for item in range(30): - q.put(item) + q = queue.Queue() + for item in range(30): + q.put(item) - # All tasks have been queued - q.shutdown() + q.shutdown() + with concurrent.futures.ThreadPoolExecutor() as tp: + for i in range(3): + tp.submit(worker, f'worker-{i}', q) - # Create 3 worker threads. - with concurrent.futures.ThreadPoolExecutor() as tp: - for i in range(3): - tp.submit(worker, f'worker-{i}', q) + print('All work completed') - print('All work completed') + .. versionadded:: 3.14 .. method:: Queue.put(item, block=True, timeout=None) From 6c9aa6d6e7ec3668e3b12c4869ec7bac063e6c9d Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 20:07:07 +0200 Subject: [PATCH 36/88] Make `AsyncQueueIterator` private --- Doc/library/asyncio-queue.rst | 26 +++----------------------- Lib/asyncio/queues.py | 5 ++--- 2 files changed, 5 insertions(+), 26 deletions(-) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index 57e73984a259828..8dfa3f56d080d25 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -48,14 +48,9 @@ Queue .. method:: __aiter__() - Return an :class:`AsyncQueueIterator` which iterates asynchronously - over the items in this queue until :meth:`shutdown` is called and - the queue is empty:: - - ... # add items to the queue - queue.shutdown() - async for item in queue: - ... # process item + Return an iterator which iterates asynchronously over the items + in this queue until :meth:`shutdown` is called and the queue is + empty. .. versionadded:: 3.14 @@ -146,21 +141,6 @@ Queue items placed in the queue. -Async Queue Iterator -==================== - -.. class:: AsyncQueueIterator(queue) - - The :term:`asynchronous iterator` returned by calling :func:`aiter` - on a :class:`Queue`. - - .. versionadded:: 3.14 - - .. attribute:: queue - - The queue which backs this ``AsyncQueueIterator``. - - Priority Queue ============== diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index 33abb84b8cce0d6..9d5d5c453132e05 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -5,7 +5,6 @@ 'QueueFull', 'QueueEmpty', 'QueueShutDown', - 'AsyncQueueIterator', ) import collections @@ -31,7 +30,7 @@ class QueueShutDown(Exception): pass -class AsyncQueueIterator: +class _AsyncQueueIterator: def __init__(self, queue): self.queue = queue @@ -92,7 +91,7 @@ def _wakeup_next(self, waiters): break def __aiter__(self): - return AsyncQueueIterator(self) + return _AsyncQueueIterator(self) def __repr__(self): return f'<{type(self).__name__} at {id(self):#x} {self._format()}>' From 9a28bc508f5afb28946c40b7581ff3c54afc9102 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 20:46:47 +0200 Subject: [PATCH 37/88] Cleanup tests --- Lib/test/test_queue.py | 69 +++++++++++++++++++++--------------------- 1 file changed, 35 insertions(+), 34 deletions(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index 6ddd0278afcfc0d..81ece265c463387 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -1,5 +1,6 @@ # Some simple queue module tests, plus some failure conditions # to ensure the Queue locks remain stable. +import concurrent.futures import itertools import random import sys @@ -151,49 +152,49 @@ def basic_queue_test(self, q): self.do_blocking_test(q.get, (), q.put, ('empty',)) self.do_blocking_test(q.get, (True, 10), q.put, ('empty',)) - - def worker(self, q, task_done): - for x in q: - with self.cumlock: - self.cum += x - if task_done: - q.task_done() - def test_iter(self): q = self.type2test() - self.cum = 0 - threads = [] - for i in (0,1): - thread = threading.Thread(target=self.worker, args=(q, False)) - thread.start() - threads.append(thread) for i in range(100): q.put(i) + q.shutdown() - for thread in threads: - thread.join() + self.cum = 0 + def worker(): + for x in q: + with self.cumlock: + self.cum += x + + with concurrent.futures.ThreadPoolExecutor as tp: + tp.submit(worker) + tp.submit(worker) + self.assertEqual(self.cum, sum(range(100))) def queue_join_test(self, q): self.cum = 0 - threads = [] - for i in (0,1): - thread = threading.Thread(target=self.worker, args=(q, True)) - thread.start() - threads.append(thread) - for i in range(100): - q.put(i) - q.join() - self.assertEqual(self.cum, sum(range(100)), - "q.join() did not block until all tasks were done") - for i in range(100, 200): - q.put(i) - q.shutdown() - q.join() # verify that you can join twice - self.assertEqual(self.cum, sum(range(200)), - "q.join() did not block until all tasks were done") - for thread in threads: - thread.join() + def worker(): + for x in q: + with self.cumlock: + self.cum += x + + q.task_done() + + with concurrent.futures.ThreadPoolExecutor as tp: + tp.submit(worker) + tp.submit(worker) + for i in range(100): + q.put(i) + + q.join() + self.assertEqual(self.cum, sum(range(100)), + "q.join() didn't block until all tasks were done") + for i in range(100, 200): + q.put(i) + + q.join() # verify that you can join twice + self.assertEqual(self.cum, sum(range(200)), + "q.join() didn't block until all tasks were done") + q.shutdown() def test_queue_task_done(self): # Test to make sure a queue task completed successfully. From 0a69fd9752bc02aa11ac6a65b6cd8a22cd486b1c Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 20:48:41 +0200 Subject: [PATCH 38/88] Instantiate --- Lib/test/test_queue.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index 81ece265c463387..0a0e05f9d180177 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -164,7 +164,7 @@ def worker(): with self.cumlock: self.cum += x - with concurrent.futures.ThreadPoolExecutor as tp: + with concurrent.futures.ThreadPoolExecutor() as tp: tp.submit(worker) tp.submit(worker) @@ -179,7 +179,7 @@ def worker(): q.task_done() - with concurrent.futures.ThreadPoolExecutor as tp: + with concurrent.futures.ThreadPoolExecutor() as tp: tp.submit(worker) tp.submit(worker) for i in range(100): From e6a5849ac8db0269c27e5d36717b1c339340491e Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 20:50:33 +0200 Subject: [PATCH 39/88] Use term --- Doc/library/asyncio-queue.rst | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index 8dfa3f56d080d25..6d50db026a05202 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -48,9 +48,9 @@ Queue .. method:: __aiter__() - Return an iterator which iterates asynchronously over the items - in this queue until :meth:`shutdown` is called and the queue is - empty. + Return an :term:`iterator` which iterates asynchronously over + the items in this queue until :meth:`shutdown` is called and + the queue is empty. .. versionadded:: 3.14 From 4407eedd34f008d08b64377594a24f1d871422bc Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 21:00:23 +0200 Subject: [PATCH 40/88] Move inside with block --- Lib/test/test_queue.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index 0a0e05f9d180177..695cc6070062bee 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -176,8 +176,7 @@ def worker(): for x in q: with self.cumlock: self.cum += x - - q.task_done() + q.task_done() with concurrent.futures.ThreadPoolExecutor() as tp: tp.submit(worker) From 1e500abcd7aa2e4c2636d67846c4550eba75fd6a Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 21:14:49 +0200 Subject: [PATCH 41/88] Add comment --- Lib/test/test_queue.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index 695cc6070062bee..be39b47735502a1 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -193,6 +193,8 @@ def worker(): q.join() # verify that you can join twice self.assertEqual(self.cum, sum(range(200)), "q.join() didn't block until all tasks were done") + + # instruct the threads to close q.shutdown() def test_queue_task_done(self): From 583f642bb2d5044c863d77cdaf6c9568e372bb3d Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 21:15:12 +0200 Subject: [PATCH 42/88] Re-order --- Lib/test/test_asyncio/test_queues.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index 4f9abcd0b6a9730..082bd08237f6704 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -492,9 +492,6 @@ async def test_task_done(self): for i in range(100): q.put_nowait(i) - # All tasks have been queued - q.shutdown() - accumulator = 0 # Two workers get items from the queue and call task_done after each. @@ -513,6 +510,9 @@ async def worker(): await q.join() self.assertEqual(sum(range(100)), accumulator) + # close running generators + q.shutdown() + async def test_join_empty_queue(self): q = self.q_class() From 1b6e78135503dfa4feafc83bf4f728d853286993 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 21:18:29 +0200 Subject: [PATCH 43/88] Remove comment --- Lib/test/test_asyncio/test_queues.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index 082bd08237f6704..c5f8e4b36141d12 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -126,9 +126,6 @@ async def test_aiter(self): accumulator = 0 - # Two workers get items from the queue and call task_done after each. - # Join the queue and assert all items have been processed. - async def worker(): nonlocal accumulator From 101409f316e6fd93ccfa139336eaa0a0857b6ef8 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Tue, 18 Jun 2024 21:19:56 +0200 Subject: [PATCH 44/88] Separate function definition --- Lib/test/test_queue.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index be39b47735502a1..bbe3c934d0c3174 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -159,6 +159,7 @@ def test_iter(self): q.shutdown() self.cum = 0 + def worker(): for x in q: with self.cumlock: @@ -172,6 +173,7 @@ def worker(): def queue_join_test(self, q): self.cum = 0 + def worker(): for x in q: with self.cumlock: From 2d5998ada601b06a84158b16ee0e4db56b72f477 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Wed, 19 Jun 2024 12:03:54 +0200 Subject: [PATCH 45/88] Update whatsnew --- Doc/whatsnew/3.14.rst | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index b357553735e8bbf..4c058c3a029c29f 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -92,6 +92,12 @@ ast Added :func:`ast.compare` for comparing two ASTs. (Contributed by Batuhan Taskaya and Jeremy Hylton in :issue:`15987`.) +asyncio +------- + +Made :class:`asyncio.Queue` an asynchronous iterable. +(Contributed by Wannes Boeykens in :gh:`120491`.) + os -- From 841d50c8dbfa49117f5d90fa4e0b0a67ffff7588 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Wed, 19 Jun 2024 12:04:28 +0200 Subject: [PATCH 46/88] Update blurb --- .../2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst | 1 + 1 file changed, 1 insertion(+) diff --git a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst index 21d2f98fb6b6226..7c1110b60fc2738 100644 --- a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst +++ b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst @@ -1 +1,2 @@ Make :class:`asyncio.Queue` an asynchronous iterable. +(Contributed by Wannes Boeykens in :gh:`120491`.) From bcdb21ff04df988175b9d4edc53ae22c2a00d4c3 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Wed, 19 Jun 2024 12:09:31 +0200 Subject: [PATCH 47/88] Update whatsnew --- Doc/whatsnew/3.14.rst | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index b357553735e8bbf..6219dad1ec3c040 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -100,6 +100,12 @@ os by :func:`os.unsetenv`, or made outside Python in the same process. (Contributed by Victor Stinner in :gh:`120057`.) +queue +----- + +Made :class:`queue.Queue` an :term:`iterable`. +(Contributed by Wannes Boeykens in :gh:`120503`.) + symtable -------- From 0743257f2dba19d6b163124361da3ff9f0a1ac52 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Wed, 19 Jun 2024 12:10:13 +0200 Subject: [PATCH 48/88] Update blurb --- .../2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst index 112317b8ac4b91a..e7cf8f084e60736 100644 --- a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst +++ b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst @@ -1 +1,2 @@ -Make :class:`queue.Queue` an iterable. +Made :class:`queue.Queue` an :term:`iterable`. +(Contributed by Wannes Boeykens in :gh:`120503`.) From d624beaa4925d99a30900df2ac7a705876c324a3 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Wed, 19 Jun 2024 12:12:04 +0200 Subject: [PATCH 49/88] Apply suggestions from code review --- Doc/whatsnew/3.14.rst | 2 +- .../2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index 4c058c3a029c29f..c7f4db10c9fafc7 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -95,7 +95,7 @@ Added :func:`ast.compare` for comparing two ASTs. asyncio ------- -Made :class:`asyncio.Queue` an asynchronous iterable. +Made :class:`asyncio.Queue` an :term:`asynchronous iterable`. (Contributed by Wannes Boeykens in :gh:`120491`.) os diff --git a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst index 7c1110b60fc2738..38bd9c61c334220 100644 --- a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst +++ b/Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst @@ -1,2 +1,2 @@ -Make :class:`asyncio.Queue` an asynchronous iterable. +Make :class:`asyncio.Queue` an :term:`asynchronous iterable`. (Contributed by Wannes Boeykens in :gh:`120491`.) From 0a8a72b59b42abc67c8084053bf3e4cf2259bbb7 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Wed, 19 Jun 2024 12:13:21 +0200 Subject: [PATCH 50/88] Update Doc/library/asyncio-queue.rst --- Doc/library/asyncio-queue.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index 6d50db026a05202..1fc9797135d3892 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -48,7 +48,7 @@ Queue .. method:: __aiter__() - Return an :term:`iterator` which iterates asynchronously over + Return an :term:`asynchronous iterator` which iterates over the items in this queue until :meth:`shutdown` is called and the queue is empty. From b3199dfad4fdc8d5ea6383e7d61145011a885ced Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Wed, 19 Jun 2024 12:33:28 +0200 Subject: [PATCH 51/88] Rename 2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst to 2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst --- .../2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename Misc/NEWS.d/next/{Core and Builtins => Library}/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst (100%) diff --git a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst b/Misc/NEWS.d/next/Library/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst similarity index 100% rename from Misc/NEWS.d/next/Core and Builtins/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst rename to Misc/NEWS.d/next/Library/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst From 5e37cd5ea45504c870e192e77312bf3b17c446d9 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Wed, 19 Jun 2024 12:34:30 +0200 Subject: [PATCH 52/88] Rename 2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst to 2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst --- .../2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename Misc/NEWS.d/next/{Core and Builtins => Library}/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst (100%) diff --git a/Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst b/Misc/NEWS.d/next/Library/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst similarity index 100% rename from Misc/NEWS.d/next/Core and Builtins/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst rename to Misc/NEWS.d/next/Library/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst From d71eb91e9474111af26dc8f3e4dce94ba1be8c31 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Wed, 19 Jun 2024 12:35:17 +0200 Subject: [PATCH 53/88] Make queue a private attribute Co-authored-by: Kumar Aditya --- Lib/asyncio/queues.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index 9d5d5c453132e05..f799efa0534c400 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -32,14 +32,14 @@ class QueueShutDown(Exception): class _AsyncQueueIterator: def __init__(self, queue): - self.queue = queue + self._queue = queue def __aiter__(self): return self async def __anext__(self): try: - return await self.queue.get() + return await self._queue.get() except QueueShutDown: raise StopAsyncIteration From 23e37f4e7b47779920ae8222f4554442ea669df2 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Thu, 20 Jun 2024 08:03:11 +0200 Subject: [PATCH 54/88] Clarify behaviour --- Doc/library/asyncio-queue.rst | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index 1fc9797135d3892..eb55e8258292c52 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -49,8 +49,10 @@ Queue .. method:: __aiter__() Return an :term:`asynchronous iterator` which iterates over - the items in this queue until :meth:`shutdown` is called and - the queue is empty. + the queue of items until :meth:`shutdown` is called continues + iteration until the queue is empty. + + ``shutdown(immediate=True)`` stops iteration immediately. .. versionadded:: 3.14 From 3bb4e6585c0b8a27d521cfbfce37d936a5176718 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Thu, 20 Jun 2024 08:04:13 +0200 Subject: [PATCH 55/88] Fix grammar --- Doc/library/asyncio-queue.rst | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index eb55e8258292c52..62c7bac734bd4a9 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -49,8 +49,8 @@ Queue .. method:: __aiter__() Return an :term:`asynchronous iterator` which iterates over - the queue of items until :meth:`shutdown` is called continues - iteration until the queue is empty. + the queue of items until :meth:`shutdown` is called and + continues iteration until the queue is empty. ``shutdown(immediate=True)`` stops iteration immediately. From b23a1b326791edd0b4310867c12bce008e7b2656 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Fri, 21 Jun 2024 09:06:56 +0200 Subject: [PATCH 56/88] Clarify documentation --- Doc/library/queue.rst | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/Doc/library/queue.rst b/Doc/library/queue.rst index 3c35f1a1c19a5ce..5f17124c6a61cfc 100644 --- a/Doc/library/queue.rst +++ b/Doc/library/queue.rst @@ -134,8 +134,9 @@ provide the public methods described below. .. method:: Queue.__iter__() - Return an :term:`iterator` which iterates over the items in this queue - until :meth:`Queue.shutdown` is called and the queue is empty. + Return an :term:`iterator` which iterates over the queue of items until + :meth:`shutdown` is called and continues iteration until the queue is + empty. Example:: From b4e5c602808216fec9a00f05b872d7eebbcba9ed Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Sun, 23 Jun 2024 19:04:06 +0200 Subject: [PATCH 57/88] Delete Misc/NEWS.d/next/Library/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst --- .../next/Library/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst | 2 -- 1 file changed, 2 deletions(-) delete mode 100644 Misc/NEWS.d/next/Library/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst diff --git a/Misc/NEWS.d/next/Library/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst b/Misc/NEWS.d/next/Library/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst deleted file mode 100644 index e7cf8f084e60736..000000000000000 --- a/Misc/NEWS.d/next/Library/2024-06-14-13-37-45.gh-issue-120499.JK7Mv8.rst +++ /dev/null @@ -1,2 +0,0 @@ -Made :class:`queue.Queue` an :term:`iterable`. -(Contributed by Wannes Boeykens in :gh:`120503`.) From f6ba54eee63b67e4b659ff9fed4b7acf5846a8b2 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Sun, 23 Jun 2024 19:05:24 +0200 Subject: [PATCH 58/88] Update 3.14.rst --- Doc/whatsnew/3.14.rst | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index e8cd8b86a922060..5f360b4a53141f8 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -110,8 +110,8 @@ pathlib queue ----- -Made :class:`queue.Queue` an :term:`iterable`. -(Contributed by Wannes Boeykens in :gh:`120503`.) +Add :meth:`queue.Queue.iter()`. +(Contributed by Wannes Boeykens in :gh:`???`.) symtable -------- From a50f12f90411b069ee5ae6c9f69f963b92934d9e Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 20:14:46 +0200 Subject: [PATCH 59/88] Explicit method --- Doc/library/queue.rst | 23 +++++++++++++++++------ Lib/queue.py | 34 +++++++++++++++++++++++++++++++--- Lib/test/test_queue.py | 26 ++++++++++++++++++++++++-- 3 files changed, 72 insertions(+), 11 deletions(-) diff --git a/Doc/library/queue.rst b/Doc/library/queue.rst index 5f17124c6a61cfc..fe1c1b3acb1761d 100644 --- a/Doc/library/queue.rst +++ b/Doc/library/queue.rst @@ -132,11 +132,18 @@ provide the public methods described below. will not block. Similarly, if full() returns ``False`` it doesn't guarantee that a subsequent call to put() will not block. -.. method:: Queue.__iter__() +.. method:: Queue.iter(block=True, timeout=None) - Return an :term:`iterator` which iterates over the queue of items until - :meth:`shutdown` is called and continues iteration until the queue is - empty. + Return an :term:`iterator` which iterates over the queue of items. If optional + args *block* is true and *timeout* is ``None`` (the default), block if necessary + until the next item is available. If *timeout* is a positive number, it blocks + at most *timeout* seconds and stops iteration if no item was available within that + time, but if the first item is not available the :exc:`Empty` exception is raised. + Otherwise (*block* is false), iterate over all item which are immediately + available, but raise the :exc:`Empty` exception if none are (*timeout* is ignored + in that case). + + :meth:`task_done` is called automatically. Example:: @@ -145,7 +152,7 @@ provide the public methods described below. import time def worker(name, q): - for item in q: + for item in q.iter(): time.sleep(.01) print(f'{name} finished {item}') @@ -162,6 +169,11 @@ provide the public methods described below. .. versionadded:: 3.14 +.. method:: Queue.iter_nowait() + + Equivalent to ``iter(False)``. + + .. versionadded:: 3.14 .. method:: Queue.put(item, block=True, timeout=None) @@ -245,7 +257,6 @@ Example of how to wait for enqueued tasks to be completed:: for item in q: print(f'Working on {item}') print(f'Finished {item}') - q.task_done() # Turn-on the worker thread. threading.Thread(target=worker, daemon=True).start() diff --git a/Lib/queue.py b/Lib/queue.py index e392bc66e077833..17033a991eee8c4 100644 --- a/Lib/queue.py +++ b/Lib/queue.py @@ -275,13 +275,41 @@ def _put(self, item): def _get(self): return self.queue.popleft() - def __iter__(self): + def iter(self, block=True, timeout=None): + '''Return an iterator which iterates over the queue of items. + + If optional args 'block' is true and 'timeout' is None (the default), + block if necessary until the next item is available. If 'timeout' is a + positive number, it blocks at most 'timeout' seconds and stops + iteration if no item was available within that time, but if the first + item is not available the Empty exception is raised. Otherwise ('block' + is false), iterate over all item which are immediately available, but + raise the Empty exception if none are ('timeout' is ignored in that + case). + + task_done() is called automatically. + ''' try: - while True: - yield self.get() + yield self.get(block=block, timeout=timeout) except ShutDown: return + try: + while True: + self.task_done() + yield self.get(block=block, timeout=timeout) + except (Empty, ShutDown): + return + + def iter_nowait(self): + '''Return an iterator which iterates over the queue of items without + blocking. + + Only iterate over the items which are immediately available, but raise + the Empty exception if none are. + ''' + return self.iter(block=False) + __class_getitem__ = classmethod(types.GenericAlias) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index 6fa8daffb8bc92d..c1ad3db13546435 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -160,7 +160,27 @@ def test_iter(self): self.cum = 0 def worker(): - for x in q: + for x in q.iter(): + with self.cumlock: + self.cum += x + + with concurrent.futures.ThreadPoolExecutor() as tp: + tp.submit(worker) + tp.submit(worker) + + self.assertEqual(self.cum, sum(range(100))) + for _ in q.iter(): + pass + + def test_iter_nowait(self): + q = self.type2test() + for i in range(100): + q.put(i) + + self.cum = 0 + + def worker(): + for x in q.iter_nowait(): with self.cumlock: self.cum += x @@ -169,6 +189,9 @@ def worker(): tp.submit(worker) self.assertEqual(self.cum, sum(range(100))) + with self.assertRaises(self.queue.Empty): + for _ in q.iter_nowait(): + pass def queue_join_test(self, q): self.cum = 0 @@ -177,7 +200,6 @@ def worker(): for x in q: with self.cumlock: self.cum += x - q.task_done() with concurrent.futures.ThreadPoolExecutor() as tp: tp.submit(worker) From 33af2a6f4647d0dabc448f2bc087124d61827431 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 20:34:09 +0200 Subject: [PATCH 60/88] Re-order --- Doc/whatsnew/3.14.rst | 4 ++-- Lib/queue.py | 3 ++- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index 5f360b4a53141f8..41294915e510399 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -110,8 +110,8 @@ pathlib queue ----- -Add :meth:`queue.Queue.iter()`. -(Contributed by Wannes Boeykens in :gh:`???`.) +Add :meth:`queue.Queue.iter()` and :meth:`queue.Queue.iter_nowait()`. +(Contributed by Wannes Boeykens in ``gh-???``.) symtable -------- diff --git a/Lib/queue.py b/Lib/queue.py index 17033a991eee8c4..d5cf52cd7cc5908 100644 --- a/Lib/queue.py +++ b/Lib/queue.py @@ -291,13 +291,14 @@ def iter(self, block=True, timeout=None): ''' try: yield self.get(block=block, timeout=timeout) + self.task_done() except ShutDown: return try: while True: - self.task_done() yield self.get(block=block, timeout=timeout) + self.task_done() except (Empty, ShutDown): return From 99db1509e2b40c2bf4bbbef78cf6de95c1cd993e Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 20:54:39 +0200 Subject: [PATCH 61/88] Revert calling `task_done()` --- Doc/library/queue.rst | 3 +-- Lib/queue.py | 4 ---- Lib/test/test_queue.py | 1 + 3 files changed, 2 insertions(+), 6 deletions(-) diff --git a/Doc/library/queue.rst b/Doc/library/queue.rst index fe1c1b3acb1761d..1f6aedae82ae0c9 100644 --- a/Doc/library/queue.rst +++ b/Doc/library/queue.rst @@ -143,8 +143,6 @@ provide the public methods described below. available, but raise the :exc:`Empty` exception if none are (*timeout* is ignored in that case). - :meth:`task_done` is called automatically. - Example:: import concurrent.futures @@ -257,6 +255,7 @@ Example of how to wait for enqueued tasks to be completed:: for item in q: print(f'Working on {item}') print(f'Finished {item}') + q.task_done() # Turn-on the worker thread. threading.Thread(target=worker, daemon=True).start() diff --git a/Lib/queue.py b/Lib/queue.py index d5cf52cd7cc5908..185fc97e0096745 100644 --- a/Lib/queue.py +++ b/Lib/queue.py @@ -286,19 +286,15 @@ def iter(self, block=True, timeout=None): is false), iterate over all item which are immediately available, but raise the Empty exception if none are ('timeout' is ignored in that case). - - task_done() is called automatically. ''' try: yield self.get(block=block, timeout=timeout) - self.task_done() except ShutDown: return try: while True: yield self.get(block=block, timeout=timeout) - self.task_done() except (Empty, ShutDown): return diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index c1ad3db13546435..23d70abf18465c3 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -200,6 +200,7 @@ def worker(): for x in q: with self.cumlock: self.cum += x + q.task_done() with concurrent.futures.ThreadPoolExecutor() as tp: tp.submit(worker) From fd96d7db6a053961bb70c58f8567fb9bad8b3332 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 21:03:54 +0200 Subject: [PATCH 62/88] Add more tests --- Lib/test/test_queue.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index 23d70abf18465c3..31760788c302a95 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -169,8 +169,6 @@ def worker(): tp.submit(worker) self.assertEqual(self.cum, sum(range(100))) - for _ in q.iter(): - pass def test_iter_nowait(self): q = self.type2test() @@ -189,9 +187,6 @@ def worker(): tp.submit(worker) self.assertEqual(self.cum, sum(range(100))) - with self.assertRaises(self.queue.Empty): - for _ in q.iter_nowait(): - pass def queue_join_test(self, q): self.cum = 0 @@ -257,6 +252,8 @@ def test_negative_timeout_raises_exception(self): q.put(1, timeout=-1) with self.assertRaises(ValueError): q.get(1, timeout=-1) + with self.assertRaises(ValueError): + list(q.iter(1, timeout=-1)) def test_nowait(self): q = self.type2test(QUEUE_SIZE) @@ -269,6 +266,8 @@ def test_nowait(self): q.get_nowait() with self.assertRaises(self.queue.Empty): q.get_nowait() + with self.assertRaises(self.queue.Empty): + list(q.iter_nowait()) def test_shrinking_queue(self): # issue 10110 @@ -290,6 +289,7 @@ def test_shutdown_empty(self): q.put("data") with self.assertRaises(self.queue.ShutDown): q.get() + list(q.iter()) def test_shutdown_nonempty(self): q = self.type2test() @@ -298,6 +298,8 @@ def test_shutdown_nonempty(self): q.get() with self.assertRaises(self.queue.ShutDown): q.get() + with self.assertRaises(self.queue.ShutDown): + list(q.iter()) def test_shutdown_immediate(self): q = self.type2test() @@ -305,6 +307,8 @@ def test_shutdown_immediate(self): q.shutdown(immediate=True) with self.assertRaises(self.queue.ShutDown): q.get() + with self.assertRaises(self.queue.ShutDown): + list(q.iter()) def test_shutdown_allowed_transitions(self): # allowed transitions would be from alive via shutdown to immediate From 3fb6f8854f7973cd023da756de3f91ac219ab3fc Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 21:05:56 +0200 Subject: [PATCH 63/88] Call `iter()` --- Lib/test/test_queue.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index 31760788c302a95..f29106807ea6ef1 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -192,7 +192,7 @@ def queue_join_test(self, q): self.cum = 0 def worker(): - for x in q: + for x in q.iter(): with self.cumlock: self.cum += x q.task_done() From d2bf0a13dc381ab1ca587a4dcd1830022618352c Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 21:10:48 +0200 Subject: [PATCH 64/88] Fix tests --- Lib/test/test_queue.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index f29106807ea6ef1..ab94e5e29d4e9a9 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -289,7 +289,7 @@ def test_shutdown_empty(self): q.put("data") with self.assertRaises(self.queue.ShutDown): q.get() - list(q.iter()) + self.assertEqual(list(q.iter()), []) def test_shutdown_nonempty(self): q = self.type2test() @@ -298,8 +298,7 @@ def test_shutdown_nonempty(self): q.get() with self.assertRaises(self.queue.ShutDown): q.get() - with self.assertRaises(self.queue.ShutDown): - list(q.iter()) + self.assertEqual(list(q.iter()), []) def test_shutdown_immediate(self): q = self.type2test() @@ -307,8 +306,7 @@ def test_shutdown_immediate(self): q.shutdown(immediate=True) with self.assertRaises(self.queue.ShutDown): q.get() - with self.assertRaises(self.queue.ShutDown): - list(q.iter()) + self.assertEqual(list(q.iter()), []) def test_shutdown_allowed_transitions(self): # allowed transitions would be from alive via shutdown to immediate From e2f6ad8f5a046acf51757d86d7eeca374e8c79e4 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 21:15:27 +0200 Subject: [PATCH 65/88] Add paragraph about `shutdown()` --- Doc/library/queue.rst | 3 +++ Lib/queue.py | 3 +++ 2 files changed, 6 insertions(+) diff --git a/Doc/library/queue.rst b/Doc/library/queue.rst index 1f6aedae82ae0c9..d25d7b24c36ff21 100644 --- a/Doc/library/queue.rst +++ b/Doc/library/queue.rst @@ -143,6 +143,9 @@ provide the public methods described below. available, but raise the :exc:`Empty` exception if none are (*timeout* is ignored in that case). + Stops iteration if the queue has been shut down and is empty, or if the queue has + been shut down immediately. + Example:: import concurrent.futures diff --git a/Lib/queue.py b/Lib/queue.py index 185fc97e0096745..afab1303415e6fb 100644 --- a/Lib/queue.py +++ b/Lib/queue.py @@ -286,6 +286,9 @@ def iter(self, block=True, timeout=None): is false), iterate over all item which are immediately available, but raise the Empty exception if none are ('timeout' is ignored in that case). + + Stops iteration if the queue has been shut down and is empty, or if the + queue has been shut down immediately. ''' try: yield self.get(block=block, timeout=timeout) From 10188a3ee6c360db6a0524d4fd8081943c6109f4 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 21:19:01 +0200 Subject: [PATCH 66/88] Test blocking --- Lib/test/test_queue.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index ab94e5e29d4e9a9..f6346e724f16cf4 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -152,11 +152,6 @@ def basic_queue_test(self, q): self.do_blocking_test(q.get, (True, 10), q.put, ('empty',)) def test_iter(self): - q = self.type2test() - for i in range(100): - q.put(i) - - q.shutdown() self.cum = 0 def worker(): @@ -167,14 +162,15 @@ def worker(): with concurrent.futures.ThreadPoolExecutor() as tp: tp.submit(worker) tp.submit(worker) + q = self.type2test() + for i in range(100): + q.put(i) + + q.shutdown() self.assertEqual(self.cum, sum(range(100))) def test_iter_nowait(self): - q = self.type2test() - for i in range(100): - q.put(i) - self.cum = 0 def worker(): @@ -183,6 +179,10 @@ def worker(): self.cum += x with concurrent.futures.ThreadPoolExecutor() as tp: + q = self.type2test() + for i in range(100): + q.put(i) + tp.submit(worker) tp.submit(worker) From ef063835fe74879c022b541e2c00d51daa2e9e11 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 21:21:28 +0200 Subject: [PATCH 67/88] Ensure queue is defined --- Lib/test/test_queue.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index f6346e724f16cf4..14d3d7d966b435f 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -152,6 +152,7 @@ def basic_queue_test(self, q): self.do_blocking_test(q.get, (True, 10), q.put, ('empty',)) def test_iter(self): + q = self.type2test() self.cum = 0 def worker(): @@ -162,7 +163,6 @@ def worker(): with concurrent.futures.ThreadPoolExecutor() as tp: tp.submit(worker) tp.submit(worker) - q = self.type2test() for i in range(100): q.put(i) @@ -171,6 +171,7 @@ def worker(): self.assertEqual(self.cum, sum(range(100))) def test_iter_nowait(self): + q = self.type2test() self.cum = 0 def worker(): @@ -179,7 +180,6 @@ def worker(): self.cum += x with concurrent.futures.ThreadPoolExecutor() as tp: - q = self.type2test() for i in range(100): q.put(i) From ae06d694705f8f7b5530160eaf1e697af2f08a43 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 21:34:46 +0200 Subject: [PATCH 68/88] Update whatsnew --- Doc/whatsnew/3.14.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index 41294915e510399..82a1eb79345b9b3 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -111,7 +111,7 @@ queue ----- Add :meth:`queue.Queue.iter()` and :meth:`queue.Queue.iter_nowait()`. -(Contributed by Wannes Boeykens in ``gh-???``.) +(Contributed by Wannes Boeykens in :gh:`120925`.) symtable -------- From 8613f61e3032d0ec9166fd28449a4b162a7a52a5 Mon Sep 17 00:00:00 2001 From: "blurb-it[bot]" <43283697+blurb-it[bot]@users.noreply.github.com> Date: Sun, 23 Jun 2024 19:35:16 +0000 Subject: [PATCH 69/88] =?UTF-8?q?=F0=9F=93=9C=F0=9F=A4=96=20Added=20by=20b?= =?UTF-8?q?lurb=5Fit.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst | 2 ++ 1 file changed, 2 insertions(+) create mode 100644 Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst diff --git a/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst b/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst new file mode 100644 index 000000000000000..c3f1ccc02fd6e4d --- /dev/null +++ b/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst @@ -0,0 +1,2 @@ +Add :meth:`queue.Queue.iter()` and :meth:`queue.Queue.iter_nowait()`. +(Contributed by Wannes Boeykens in :gh:`120925`.) From ba8b13416e9f089b27feffdf311c53165cbda4f7 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 21:54:54 +0200 Subject: [PATCH 70/88] Explicit method --- Doc/library/asyncio-queue.rst | 24 +++++++++++---------- Lib/asyncio/queues.py | 19 +++++++++++++++-- Lib/test/test_asyncio/test_queues.py | 31 +++++++++++++++++++++------- 3 files changed, 54 insertions(+), 20 deletions(-) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index 62c7bac734bd4a9..9f2eb1053f2eb3e 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -46,16 +46,6 @@ Queue Number of items allowed in the queue. - .. method:: __aiter__() - - Return an :term:`asynchronous iterator` which iterates over - the queue of items until :meth:`shutdown` is called and - continues iteration until the queue is empty. - - ``shutdown(immediate=True)`` stops iteration immediately. - - .. versionadded:: 3.14 - .. method:: empty() Return ``True`` if the queue is empty, ``False`` otherwise. @@ -80,6 +70,18 @@ Queue Return an item if one is immediately available, else raise :exc:`QueueEmpty`. + .. method:: iter() + + .. TODO(Nice Zombies) + + .. versionadded:: 3.14 + + .. method:: iter_nowait() + + .. TODO(Nice Zombies) + + .. versionadded:: 3.14 + .. coroutinemethod:: join() Block until all items in the queue have been received and processed. @@ -202,7 +204,7 @@ concurrent tasks:: async def worker(name, queue): # Get a "work item" out of the queue. - async for sleep_for in queue: + async for sleep_for in queue.iter(): # Sleep for the "sleep_for" seconds. await asyncio.sleep(sleep_for) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index f799efa0534c400..2efcbeeba05233d 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -90,8 +90,6 @@ def _wakeup_next(self, waiters): waiter.set_result(None) break - def __aiter__(self): - return _AsyncQueueIterator(self) def __repr__(self): return f'<{type(self).__name__} at {id(self):#x} {self._format()}>' @@ -233,6 +231,23 @@ def get_nowait(self): self._wakeup_next(self._putters) return item + def iter(self): + # TODO(Nice Zombies) + return _AsyncQueueIterator(self) + + def iter_nowait(self): + # TODO(Nice Zombies) + try: + yield self.get_nowait() + except QueueShutDown: + return + + try: + while True: + yield self.get_nowait() + except (QueueEmpty, QueueShutDown): + return + def task_done(self): """Indicate that a formerly enqueued task is complete. diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index c5f8e4b36141d12..56c07d0936f5b22 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -116,23 +116,40 @@ async def putter(): self.assertTrue(t.done()) self.assertTrue(t.result()) - async def test_aiter(self): + async def test_iter(self): q = asyncio.Queue() - for i in range(100): - q.put_nowait(i) + accumulator = 0 - # All tasks have been queued - q.shutdown() + async def worker(): + nonlocal accumulator + + async for item in q.iter(): + accumulator += item + + async with asyncio.TaskGroup() as tg: + tg.create_task(worker()) + tg.create_task(worker()) + for i in range(100): + q.put_nowait(i) + + q.shutdown() + self.assertEqual(sum(range(100)), accumulator) + + async def test_iter_nowait(self): + q = asyncio.Queue() accumulator = 0 async def worker(): nonlocal accumulator - async for item in q: + for item in q.iter_nowait(): accumulator += item async with asyncio.TaskGroup() as tg: + for i in range(100): + q.put_nowait(i) + tg.create_task(worker()) tg.create_task(worker()) @@ -497,7 +514,7 @@ async def test_task_done(self): async def worker(): nonlocal accumulator - async for item in q: + async for item in q.iter(): accumulator += item q.task_done() From 5d6681a231e8624bc90fb572e3497a3d72d37205 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 21:57:11 +0200 Subject: [PATCH 71/88] Fix lint --- Lib/asyncio/queues.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index 2efcbeeba05233d..c6bd8b902f7e2cb 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -234,7 +234,7 @@ def get_nowait(self): def iter(self): # TODO(Nice Zombies) return _AsyncQueueIterator(self) - + def iter_nowait(self): # TODO(Nice Zombies) try: From 823389f024af72cb11a596553dbd91be6716a0bb Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 22:08:44 +0200 Subject: [PATCH 72/88] Don't use multiple threads --- Lib/test/test_asyncio/test_queues.py | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index 56c07d0936f5b22..f22df5e54cb05f2 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -139,19 +139,11 @@ async def worker(): async def test_iter_nowait(self): q = asyncio.Queue() accumulator = 0 + for i in range(100): + q.put_nowait(i) - async def worker(): - nonlocal accumulator - - for item in q.iter_nowait(): - accumulator += item - - async with asyncio.TaskGroup() as tg: - for i in range(100): - q.put_nowait(i) - - tg.create_task(worker()) - tg.create_task(worker()) + for item in q.iter_nowait(): + accumulator += item self.assertEqual(sum(range(100)), accumulator) @@ -206,6 +198,8 @@ def test_nonblocking_get(self): def test_nonblocking_get_exception(self): q = asyncio.Queue() self.assertRaises(asyncio.QueueEmpty, q.get_nowait) + with self.assertRaises(asyncio.QueueEmpty): + list(q.iter_nowait()) async def test_get_cancelled_race(self): q = asyncio.Queue() @@ -600,6 +594,7 @@ async def test_shutdown_empty(self): await q.get() with self.assertRaisesShutdown(): q.get_nowait() + self.assertEqual(list(q.iter_nowait()), []) async def test_shutdown_nonempty(self): # Test shutting down a non-empty queue @@ -644,6 +639,7 @@ async def test_shutdown_nonempty(self): await q.get() with self.assertRaisesShutdown(): q.get_nowait() + self.assertEqual(list(q.iter_nowait()), []) # Ensure there is 1 unfinished task, and join() task succeeds q.task_done() @@ -686,6 +682,7 @@ async def test_shutdown_immediate(self): await q.get() with self.assertRaisesShutdown(): q.get_nowait() + self.assertEqual(list(q.iter_nowait()), []) # Ensure there are no unfinished tasks with self.assertRaises( From a1c8ebbe565e92e00ca2fda639eaf015e84c4e65 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 22:09:12 +0200 Subject: [PATCH 73/88] Don't use multiple threads --- Lib/test/test_queue.py | 15 ++++----------- 1 file changed, 4 insertions(+), 11 deletions(-) diff --git a/Lib/test/test_queue.py b/Lib/test/test_queue.py index 14d3d7d966b435f..cf2e59ba4eb83e3 100644 --- a/Lib/test/test_queue.py +++ b/Lib/test/test_queue.py @@ -173,18 +173,11 @@ def worker(): def test_iter_nowait(self): q = self.type2test() self.cum = 0 + for i in range(100): + q.put(i) - def worker(): - for x in q.iter_nowait(): - with self.cumlock: - self.cum += x - - with concurrent.futures.ThreadPoolExecutor() as tp: - for i in range(100): - q.put(i) - - tp.submit(worker) - tp.submit(worker) + for x in q.iter_nowait(): + self.cum += x self.assertEqual(self.cum, sum(range(100))) From 104423a06a9d582b4baf58b380af23cb9dca7367 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 22:21:15 +0200 Subject: [PATCH 74/88] Add docstrings --- Doc/library/asyncio-queue.rst | 12 ++++++++++-- Lib/asyncio/queues.py | 14 ++++++++++++-- 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/Doc/library/asyncio-queue.rst b/Doc/library/asyncio-queue.rst index 9f2eb1053f2eb3e..fe44157a9ef249c 100644 --- a/Doc/library/asyncio-queue.rst +++ b/Doc/library/asyncio-queue.rst @@ -72,13 +72,21 @@ Queue .. method:: iter() - .. TODO(Nice Zombies) + Return an :term:`asynchronous iterator` which iterates over the queue + of items. If queue is empty, wait until the next item is available. + + Stops iteration if the queue has been shut down and is empty, or if + the queue has been shut down immediately. .. versionadded:: 3.14 .. method:: iter_nowait() - .. TODO(Nice Zombies) + Return an :term:`iterator` which iterates over the queue of items + without blocking. + + Only iterate over the items which are immediately available, but raise + the :exc:`QueueEmpty` exception if none are. .. versionadded:: 3.14 diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index c6bd8b902f7e2cb..e42584f95326d0d 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -232,11 +232,21 @@ def get_nowait(self): return item def iter(self): - # TODO(Nice Zombies) + """Return an asynchronous iterator which iterates over the queue of + items. If queue is empty, wait until the next item is available. + + Stops iteration if the queue has been shut down and is empty, or if the + queue has been shut down immediately. + """ return _AsyncQueueIterator(self) def iter_nowait(self): - # TODO(Nice Zombies) + """Return an iterator which iterates over the queue of items without + blocking. + + Only iterate over the items which are immediately available, but raise + the QueueEmpty exception if none are. + """ try: yield self.get_nowait() except QueueShutDown: From d3e06d4119008cd2bb9f8f7c3e28794efe44791c Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 22:23:09 +0200 Subject: [PATCH 75/88] Revert newline --- Lib/asyncio/queues.py | 1 - 1 file changed, 1 deletion(-) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index e42584f95326d0d..94767edc2495435 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -90,7 +90,6 @@ def _wakeup_next(self, waiters): waiter.set_result(None) break - def __repr__(self): return f'<{type(self).__name__} at {id(self):#x} {self._format()}>' From f5aeeacb6c30cad9c9ca5945b0a755d4b573d19d Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Sun, 23 Jun 2024 22:23:32 +0200 Subject: [PATCH 76/88] Delete Misc/NEWS.d/next/Library/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst --- .../next/Library/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst | 2 -- 1 file changed, 2 deletions(-) delete mode 100644 Misc/NEWS.d/next/Library/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst diff --git a/Misc/NEWS.d/next/Library/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst b/Misc/NEWS.d/next/Library/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst deleted file mode 100644 index 38bd9c61c334220..000000000000000 --- a/Misc/NEWS.d/next/Library/2024-06-14-10-30-43.gh-issue-119154.1MHWA-.rst +++ /dev/null @@ -1,2 +0,0 @@ -Make :class:`asyncio.Queue` an :term:`asynchronous iterable`. -(Contributed by Wannes Boeykens in :gh:`120491`.) From efc7d8e500a1fa4cfe83b8cd9ebb92c8fa5a0788 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 22:25:46 +0200 Subject: [PATCH 77/88] Remove redundant parentheses --- Doc/whatsnew/3.14.rst | 2 +- .../next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index 82a1eb79345b9b3..a41ef297f43627b 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -110,7 +110,7 @@ pathlib queue ----- -Add :meth:`queue.Queue.iter()` and :meth:`queue.Queue.iter_nowait()`. +Add :meth:`queue.Queue.iter` and :meth:`queue.Queue.iter_nowait`. (Contributed by Wannes Boeykens in :gh:`120925`.) symtable diff --git a/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst b/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst index c3f1ccc02fd6e4d..98994927b6dcaec 100644 --- a/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst +++ b/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst @@ -1,2 +1,2 @@ -Add :meth:`queue.Queue.iter()` and :meth:`queue.Queue.iter_nowait()`. +Add :meth:`queue.Queue.iter` and :meth:`queue.Queue.iter_nowait`. (Contributed by Wannes Boeykens in :gh:`120925`.) From fb98fa68b6a9d10d756e72f484bde1e868245d6d Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 22:26:26 +0200 Subject: [PATCH 78/88] Update whatsnew --- Doc/whatsnew/3.14.rst | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index c7f4db10c9fafc7..7863d0378e675ff 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -95,8 +95,8 @@ Added :func:`ast.compare` for comparing two ASTs. asyncio ------- -Made :class:`asyncio.Queue` an :term:`asynchronous iterable`. -(Contributed by Wannes Boeykens in :gh:`120491`.) +Add :meth:`asyncio.Queue.iter` and :meth:`asyncio.Queue.iter_nowait`. +(Contributed by Wannes Boeykens in ``gh-???`120925``.) os -- From a56505dabbafd0a2eae6f55055f6f5a0a679c801 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 22:26:51 +0200 Subject: [PATCH 79/88] Fix typo --- Doc/whatsnew/3.14.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index 7863d0378e675ff..0d3b2979c3915da 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -96,7 +96,7 @@ asyncio ------- Add :meth:`asyncio.Queue.iter` and :meth:`asyncio.Queue.iter_nowait`. -(Contributed by Wannes Boeykens in ``gh-???`120925``.) +(Contributed by Wannes Boeykens in ``gh-???``.) os -- From 6f146f5282d78118e7c4ad7b90b9755877454d53 Mon Sep 17 00:00:00 2001 From: "blurb-it[bot]" <43283697+blurb-it[bot]@users.noreply.github.com> Date: Sun, 23 Jun 2024 20:32:00 +0000 Subject: [PATCH 80/88] =?UTF-8?q?=F0=9F=93=9C=F0=9F=A4=96=20Added=20by=20b?= =?UTF-8?q?lurb=5Fit.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../next/Library/2024-06-23-20-31-59.gh-issue-120924.2AiErS.rst | 2 ++ 1 file changed, 2 insertions(+) create mode 100644 Misc/NEWS.d/next/Library/2024-06-23-20-31-59.gh-issue-120924.2AiErS.rst diff --git a/Misc/NEWS.d/next/Library/2024-06-23-20-31-59.gh-issue-120924.2AiErS.rst b/Misc/NEWS.d/next/Library/2024-06-23-20-31-59.gh-issue-120924.2AiErS.rst new file mode 100644 index 000000000000000..b9bb86ca9dab74d --- /dev/null +++ b/Misc/NEWS.d/next/Library/2024-06-23-20-31-59.gh-issue-120924.2AiErS.rst @@ -0,0 +1,2 @@ +Add :meth:`asyncio.Queue.iter` and :meth:`asyncio.Queue.iter_nowait`. +(Contributed by Wannes Boeykens in :gh:`120927`.) From 0bbcfddc94101b0d412e94dcf5b1f4ce2961df22 Mon Sep 17 00:00:00 2001 From: Nineteendo Date: Sun, 23 Jun 2024 22:32:41 +0200 Subject: [PATCH 81/88] Update whatsnew --- Doc/whatsnew/3.14.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index 0d3b2979c3915da..7c5c033a348253f 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -96,7 +96,7 @@ asyncio ------- Add :meth:`asyncio.Queue.iter` and :meth:`asyncio.Queue.iter_nowait`. -(Contributed by Wannes Boeykens in ``gh-???``.) +(Contributed by Wannes Boeykens in :gh:`120927`.) os -- From 81da0e6a17cbe29de1fa3b32808e0d0e2ebfae86 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Mon, 24 Jun 2024 09:17:43 +0200 Subject: [PATCH 82/88] Update 3.14.rst --- Doc/whatsnew/3.14.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index 7c5c033a348253f..c75af9edce20654 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -96,7 +96,7 @@ asyncio ------- Add :meth:`asyncio.Queue.iter` and :meth:`asyncio.Queue.iter_nowait`. -(Contributed by Wannes Boeykens in :gh:`120927`.) +(Contributed by Wannes Boeykens in :gh:`120925`.) os -- From 461d1e164e690e3dc8c1b842de94cda3db3435ed Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Mon, 24 Jun 2024 09:18:00 +0200 Subject: [PATCH 83/88] Delete Misc/NEWS.d/next/Library/2024-06-23-20-31-59.gh-issue-120924.2AiErS.rst --- .../next/Library/2024-06-23-20-31-59.gh-issue-120924.2AiErS.rst | 2 -- 1 file changed, 2 deletions(-) delete mode 100644 Misc/NEWS.d/next/Library/2024-06-23-20-31-59.gh-issue-120924.2AiErS.rst diff --git a/Misc/NEWS.d/next/Library/2024-06-23-20-31-59.gh-issue-120924.2AiErS.rst b/Misc/NEWS.d/next/Library/2024-06-23-20-31-59.gh-issue-120924.2AiErS.rst deleted file mode 100644 index b9bb86ca9dab74d..000000000000000 --- a/Misc/NEWS.d/next/Library/2024-06-23-20-31-59.gh-issue-120924.2AiErS.rst +++ /dev/null @@ -1,2 +0,0 @@ -Add :meth:`asyncio.Queue.iter` and :meth:`asyncio.Queue.iter_nowait`. -(Contributed by Wannes Boeykens in :gh:`120927`.) From 4d7e55ba5f41d1ee0c195938176f6a54d5377f8e Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Mon, 24 Jun 2024 09:21:30 +0200 Subject: [PATCH 84/88] Update 2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst --- .../Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst b/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst index 98994927b6dcaec..b25f9ef6af1b949 100644 --- a/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst +++ b/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst @@ -1,2 +1,3 @@ +Add :meth:`asyncio.Queue.iter` and :meth:`asyncio.Queue.iter_nowait`. Add :meth:`queue.Queue.iter` and :meth:`queue.Queue.iter_nowait`. -(Contributed by Wannes Boeykens in :gh:`120925`.) +(Contributed by Wannes Boeykens.) From 888ae9e51b975ee4f7b6c6c8021f875ba46adeda Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Mon, 24 Jun 2024 09:43:35 +0200 Subject: [PATCH 85/88] Remove parentheses --- .../next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst b/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst index b25f9ef6af1b949..a5c571572c5de62 100644 --- a/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst +++ b/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst @@ -1,3 +1,3 @@ Add :meth:`asyncio.Queue.iter` and :meth:`asyncio.Queue.iter_nowait`. Add :meth:`queue.Queue.iter` and :meth:`queue.Queue.iter_nowait`. -(Contributed by Wannes Boeykens.) +Contributed by Wannes Boeykens. From a413d1370b631b9ec955ab7b1ffe2786ad3d7d0e Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Mon, 24 Jun 2024 10:34:24 +0200 Subject: [PATCH 86/88] Link to github issue --- Doc/whatsnew/3.14.rst | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index cf69ef9b24761ce..823efb3362ae744 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -96,7 +96,7 @@ asyncio ------- Add :meth:`asyncio.Queue.iter` and :meth:`asyncio.Queue.iter_nowait`. -(Contributed by Wannes Boeykens in :gh:`120925`.) +(Contributed by Wannes Boeykens in :gh:`120924`.) os -- @@ -117,7 +117,7 @@ queue ----- Add :meth:`queue.Queue.iter` and :meth:`queue.Queue.iter_nowait`. -(Contributed by Wannes Boeykens in :gh:`120925`.) +(Contributed by Wannes Boeykens in :gh:`120924`.) symtable -------- From dcdd17e2a33bfb9cefeada9dd016f827bb9567e8 Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Tue, 25 Jun 2024 10:22:53 +0200 Subject: [PATCH 87/88] Rename 2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst to 2024-06-23-19-35-14.gh-issue-119154.7CXFaX.rst --- ....7CXFaX.rst => 2024-06-23-19-35-14.gh-issue-119154.7CXFaX.rst} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename Misc/NEWS.d/next/Library/{2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst => 2024-06-23-19-35-14.gh-issue-119154.7CXFaX.rst} (100%) diff --git a/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst b/Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-119154.7CXFaX.rst similarity index 100% rename from Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-120924.7CXFaX.rst rename to Misc/NEWS.d/next/Library/2024-06-23-19-35-14.gh-issue-119154.7CXFaX.rst From 525effeb30c0638b5a549c5cf504c67e4ba54dae Mon Sep 17 00:00:00 2001 From: Nice Zombies Date: Tue, 25 Jun 2024 10:23:50 +0200 Subject: [PATCH 88/88] Update issue number --- Doc/whatsnew/3.14.rst | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst index 823efb3362ae744..61ec8519b9d3c7a 100644 --- a/Doc/whatsnew/3.14.rst +++ b/Doc/whatsnew/3.14.rst @@ -96,7 +96,7 @@ asyncio ------- Add :meth:`asyncio.Queue.iter` and :meth:`asyncio.Queue.iter_nowait`. -(Contributed by Wannes Boeykens in :gh:`120924`.) +(Contributed by Wannes Boeykens in :gh:`119154`.) os -- @@ -117,7 +117,7 @@ queue ----- Add :meth:`queue.Queue.iter` and :meth:`queue.Queue.iter_nowait`. -(Contributed by Wannes Boeykens in :gh:`120924`.) +(Contributed by Wannes Boeykens in :gh:`119154`.) symtable --------