From 4e22f397969ba8a16b34ceb18bc7b2ffcd9b2cc9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Melero=20Fern=C3=A1ndez?= Date: Tue, 16 Jan 2018 19:57:59 +0000 Subject: [PATCH 1/4] bpo-32574: asyncio.Queue, put() leaks memory --- Lib/asyncio/queues.py | 6 ++++-- Lib/test/test_asyncio/test_queues.py | 20 ++++++++++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index 512ea6037f8abb1..e578ac09b6b8942 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -121,6 +121,10 @@ async def put(self, item): await putter except: putter.cancel() # Just in case putter is not done yet. + try: + self._putters.remove(putter) + except ValueError: + pass if not self.full() and not putter.cancelled(): # We were woken up by get_nowait(), but can't take # the call. Wake up the next in line. @@ -152,12 +156,10 @@ async def get(self): await getter except: getter.cancel() # Just in case getter is not done yet. - try: self._getters.remove(getter) except ValueError: pass - if not self.empty() and not getter.cancelled(): # We were woken up by put_nowait(), but can't take # the call. Wake up the next in line. diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index 8d78546318accbe..a05da513985e25d 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -520,6 +520,26 @@ async def getter(): self.loop.run_until_complete( asyncio.gather(getter(), t0, t1, t2, t3, loop=self.loop)) + def test_cancelled_putters_not_being_held_in_self_putters(self): + def a_generator(): + yield 0.1 + yield 0.2 + yield 0.3 + + self.loop = self.new_test_loop(a_generator) + + async def produce(queue, num_items): + for i in range(num_items): + try: + await asyncio.wait_for(queue.put(i), 0.1, loop=self.loop) + except asyncio.TimeoutError: + pass + + queue = asyncio.Queue(loop=self.loop, maxsize=1) + coro = produce(queue, num_items=2) + self.loop.run_until_complete(self.loop.create_task(coro)) + self.assertEqual(len(queue._putters), 0) + class LifoQueueTests(_QueueTestBase): From 51bdc7e0f08f28af9a7e19a865a2cfa4fe904006 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Melero=20Fern=C3=A1ndez?= Date: Tue, 16 Jan 2018 20:39:59 +0000 Subject: [PATCH 2/4] Updated NEWS.d with the bpo-32574 --- .../next/Library/2018-01-16-20-37-28.bpo-32574.ru8eZ9.rst | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 Misc/NEWS.d/next/Library/2018-01-16-20-37-28.bpo-32574.ru8eZ9.rst diff --git a/Misc/NEWS.d/next/Library/2018-01-16-20-37-28.bpo-32574.ru8eZ9.rst b/Misc/NEWS.d/next/Library/2018-01-16-20-37-28.bpo-32574.ru8eZ9.rst new file mode 100644 index 000000000000000..00650d4d02e07e2 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2018-01-16-20-37-28.bpo-32574.ru8eZ9.rst @@ -0,0 +1,3 @@ +Fix memory leak in asyncio.Queue, when the queue has limited size and it is +full, the cancelation of queue.put() can cause a memory leak. Patch by: José +Melero. From 8cc96e7487b429717a2549c4f5904bcadd7d7523 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Melero=20Fern=C3=A1ndez?= Date: Sun, 21 Jan 2018 00:55:48 +0000 Subject: [PATCH 3/4] Added comments and tests refactor of bpo-32574 - Added comments explaining the problem to the test and the bugfix. - Added one test to ensure that the ValueError is silenced. - Refactor of the other test. --- Lib/asyncio/queues.py | 4 ++ Lib/test/test_asyncio/test_queues.py | 56 +++++++++++++++++++++------- 2 files changed, 46 insertions(+), 14 deletions(-) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index e578ac09b6b8942..d95f957dfa7d419 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -122,8 +122,10 @@ async def put(self, item): except: putter.cancel() # Just in case putter is not done yet. try: + # Clean self._putters from canceled putters. self._putters.remove(putter) except ValueError: + # The putter could be removed from self._putters by a previous get_nowait call. pass if not self.full() and not putter.cancelled(): # We were woken up by get_nowait(), but can't take @@ -157,8 +159,10 @@ async def get(self): except: getter.cancel() # Just in case getter is not done yet. try: + # Clean self._getters from canceled getters. self._getters.remove(getter) except ValueError: + # The getter could be removed from self._getters by a previous put_nowait call. pass if not self.empty() and not getter.cancelled(): # We were woken up by put_nowait(), but can't take diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index a05da513985e25d..82e41efd8903d1e 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -520,26 +520,54 @@ async def getter(): self.loop.run_until_complete( asyncio.gather(getter(), t0, t1, t2, t3, loop=self.loop)) - def test_cancelled_putters_not_being_held_in_self_putters(self): + def test_cancelled_puts_not_being_held_in_self_putters(self): def a_generator(): + yield 0.01 yield 0.1 - yield 0.2 - yield 0.3 - self.loop = self.new_test_loop(a_generator) + loop = self.new_test_loop(a_generator) - async def produce(queue, num_items): - for i in range(num_items): - try: - await asyncio.wait_for(queue.put(i), 0.1, loop=self.loop) - except asyncio.TimeoutError: - pass - - queue = asyncio.Queue(loop=self.loop, maxsize=1) - coro = produce(queue, num_items=2) - self.loop.run_until_complete(self.loop.create_task(coro)) + # Full queue. + queue = asyncio.Queue(loop=loop, maxsize=1) + queue.put_nowait(1) + + # Task waiting for space to put an item in the queue. + put_task = loop.create_task(queue.put(1)) + loop.run_until_complete(asyncio.sleep(0.01, loop=loop)) + + # Check that the putter is correctly removed from queue._putters when the task is canceled. + self.assertEqual(len(queue._putters), 1) + put_task.cancel() + with self.assertRaises(asyncio.CancelledError): + loop.run_until_complete(put_task) self.assertEqual(len(queue._putters), 0) + def test_cancelled_put_silence_value_error_exception(self): + def gen(): + yield 0.01 + yield 0.1 + + loop = self.new_test_loop(gen) + + # Full Queue. + queue = asyncio.Queue(1, loop=loop) + queue.put_nowait(1) + + # Task waiting for space to put a item in the queue. + put_task = loop.create_task(queue.put(1)) + loop.run_until_complete(asyncio.sleep(0.01, loop=loop)) + + # get_nowait() remove the future of put_task from queue._putters. + queue.get_nowait() + # When canceled, queue.put is going to remove its future from self._putters + # but it was removed previously by queue.get_nowait(). + put_task.cancel() + + # The ValueError exception triggered by queue._putters.remove(putter) inside queue.put should be silenced. + # If the ValueError is silenced we should catch a CancelledError. + with self.assertRaises(asyncio.CancelledError): + loop.run_until_complete(put_task) + class LifoQueueTests(_QueueTestBase): From ff0185217399ff03f7400f0ae049c8bd07ed19cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Melero=20Fern=C3=A1ndez?= Date: Thu, 25 Jan 2018 23:07:30 +0000 Subject: [PATCH 4/4] Adjust right margin to 80 characters --- Lib/asyncio/queues.py | 6 ++++-- Lib/test/test_asyncio/test_queues.py | 10 ++++++---- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index d95f957dfa7d419..2a8d3e76044109c 100644 --- a/Lib/asyncio/queues.py +++ b/Lib/asyncio/queues.py @@ -125,7 +125,8 @@ async def put(self, item): # Clean self._putters from canceled putters. self._putters.remove(putter) except ValueError: - # The putter could be removed from self._putters by a previous get_nowait call. + # The putter could be removed from self._putters by a + # previous get_nowait call. pass if not self.full() and not putter.cancelled(): # We were woken up by get_nowait(), but can't take @@ -162,7 +163,8 @@ async def get(self): # Clean self._getters from canceled getters. self._getters.remove(getter) except ValueError: - # The getter could be removed from self._getters by a previous put_nowait call. + # The getter could be removed from self._getters by a + # previous put_nowait call. pass if not self.empty() and not getter.cancelled(): # We were woken up by put_nowait(), but can't take diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py index 82e41efd8903d1e..efe719ed39a9897 100644 --- a/Lib/test/test_asyncio/test_queues.py +++ b/Lib/test/test_asyncio/test_queues.py @@ -535,7 +535,8 @@ def a_generator(): put_task = loop.create_task(queue.put(1)) loop.run_until_complete(asyncio.sleep(0.01, loop=loop)) - # Check that the putter is correctly removed from queue._putters when the task is canceled. + # Check that the putter is correctly removed from queue._putters when + # the task is canceled. self.assertEqual(len(queue._putters), 1) put_task.cancel() with self.assertRaises(asyncio.CancelledError): @@ -559,11 +560,12 @@ def gen(): # get_nowait() remove the future of put_task from queue._putters. queue.get_nowait() - # When canceled, queue.put is going to remove its future from self._putters - # but it was removed previously by queue.get_nowait(). + # When canceled, queue.put is going to remove its future from + # self._putters but it was removed previously by queue.get_nowait(). put_task.cancel() - # The ValueError exception triggered by queue._putters.remove(putter) inside queue.put should be silenced. + # The ValueError exception triggered by queue._putters.remove(putter) + # inside queue.put should be silenced. # If the ValueError is silenced we should catch a CancelledError. with self.assertRaises(asyncio.CancelledError): loop.run_until_complete(put_task)