Skip to content

Commit 1eec518

Browse files
[3.15] gh-156523: Fix asyncio.as_completed() not recording the awaiting task (GH-156527) (#156552)
Co-authored-by: Timofei Ivankov <128279579+deadlovelll@users.noreply.github.com>
1 parent d122f3e commit 1eec518

3 files changed

Lines changed: 68 additions & 0 deletions

File tree

Lib/asyncio/tasks.py

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -562,9 +562,11 @@ def __init__(self, aws, timeout):
562562
self._timeout_handle = None
563563

564564
loop = events.get_event_loop()
565+
self._cur_task = current_task()
565566
todo = {ensure_future(aw, loop=loop) for aw in set(aws)}
566567
for f in todo:
567568
f.add_done_callback(self._handle_completion)
569+
futures.future_add_to_awaited_by(f, self._cur_task)
568570
if todo and timeout is not None:
569571
self._timeout_handle = (
570572
loop.call_later(timeout, self._handle_timeout)
@@ -595,13 +597,15 @@ def __next__(self):
595597
def _handle_timeout(self):
596598
for f in self._todo:
597599
f.remove_done_callback(self._handle_completion)
600+
futures.future_discard_from_awaited_by(f, self._cur_task)
598601
self._done.put_nowait(None) # Sentinel for _wait_for_one().
599602
self._todo.clear() # Can't do todo.remove(f) in the loop.
600603

601604
def _handle_completion(self, f):
602605
if not self._todo:
603606
return # _handle_timeout() was here first.
604607
self._todo.remove(f)
608+
futures.future_discard_from_awaited_by(f, self._cur_task)
605609
self._done.put_nowait(f)
606610
if not self._todo and self._timeout_handle is not None:
607611
self._timeout_handle.cancel()

Lib/test/test_asyncio/test_graph.py

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -271,6 +271,68 @@ async def main(t1, t2):
271271
]
272272
])
273273

274+
async def test_stack_as_completed(self):
275+
# gh-156523: as_completed() must record the awaiting task
276+
stack_for_inner = None
277+
278+
async def inner():
279+
await asyncio.sleep(0)
280+
nonlocal stack_for_inner
281+
stack_for_inner = capture_test_stack()
282+
283+
async def main(t):
284+
for f in asyncio.as_completed([t]):
285+
await f
286+
287+
t = asyncio.create_task(inner(), name='inner')
288+
await main(t)
289+
self.assertFalse(t._asyncio_awaited_by)
290+
291+
self.assertEqual(stack_for_inner[0], [
292+
'T<inner>',
293+
['s capture_test_stack', 'a inner'],
294+
[
295+
['T<anon>',
296+
['a get', 'a _wait_for_one', 'a main',
297+
'a test_stack_as_completed'],
298+
[]
299+
]
300+
]
301+
])
302+
303+
async def test_stack_as_completed_timeout(self):
304+
# gh-156523: the awaiting task must be dropped when as_completed() times out
305+
stack_for_inner = None
306+
307+
async def inner():
308+
nonlocal stack_for_inner
309+
stack_for_inner = capture_test_stack()
310+
await asyncio.sleep(3600)
311+
312+
async def main(t):
313+
with self.assertRaises(TimeoutError):
314+
for f in asyncio.as_completed([t], timeout=0.01):
315+
await f
316+
317+
t = asyncio.create_task(inner(), name='inner')
318+
await main(t)
319+
self.assertFalse(t._asyncio_awaited_by)
320+
t.cancel()
321+
with self.assertRaises(asyncio.CancelledError):
322+
await t
323+
324+
self.assertEqual(stack_for_inner[0], [
325+
'T<inner>',
326+
['s capture_test_stack', 'a inner'],
327+
[
328+
['T<anon>',
329+
['a get', 'a _wait_for_one', 'a main',
330+
'a test_stack_as_completed_timeout'],
331+
[]
332+
]
333+
]
334+
])
335+
274336
async def test_stack_task(self):
275337

276338
stack_for_inner = None
Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
Fix :func:`asyncio.as_completed` not recording the awaiting task in the call
2+
graph.

0 commit comments

Comments
 (0)