mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-26 10:19:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

gh-86802: Fix asyncio memory leak; shielded task exceptions log once through the exception handler (gh-134331)

Browse source 
Co-authored-by: Łukasz Langa <lukasz@langa.pl>
This commit is contained in:
Christian Harries 2025-05-20 16:14:27 +01:00 • committed by GitHub
parent f3acbb72ff
commit f695eca60c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 72 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

40
Lib/test/test_asyncio/test_tasks.py
View file

@ -2116,6 +2116,46 @@ class BaseTaskTests:
self.assertTrue(outer.cancelled())
self.assertEqual(0, 0 if outer._callbacks is None else len(outer._callbacks))
def test_shield_cancel_outer_result(self):
mock_handler = mock.Mock()
self.loop.set_exception_handler(mock_handler)
inner = self.new_future(self.loop)
outer = asyncio.shield(inner)
test_utils.run_briefly(self.loop)
outer.cancel()
test_utils.run_briefly(self.loop)
inner.set_result(1)
test_utils.run_briefly(self.loop)
mock_handler.assert_not_called()
def test_shield_cancel_outer_exception(self):
mock_handler = mock.Mock()
self.loop.set_exception_handler(mock_handler)
inner = self.new_future(self.loop)
outer = asyncio.shield(inner)
test_utils.run_briefly(self.loop)
outer.cancel()
test_utils.run_briefly(self.loop)
inner.set_exception(Exception('foo'))
test_utils.run_briefly(self.loop)
mock_handler.assert_called_once()
def test_shield_duplicate_log_once(self):
mock_handler = mock.Mock()
self.loop.set_exception_handler(mock_handler)
inner = self.new_future(self.loop)
outer = asyncio.shield(inner)
test_utils.run_briefly(self.loop)
outer.cancel()
test_utils.run_briefly(self.loop)
outer = asyncio.shield(inner)
test_utils.run_briefly(self.loop)
outer.cancel()
test_utils.run_briefly(self.loop)
inner.set_exception(Exception('foo'))
test_utils.run_briefly(self.loop)
mock_handler.assert_called_once()
def test_shield_shortcut(self):
fut = self.new_future(self.loop)
fut.set_result(42)