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

[3.12] gh-124309: fix staggered race on eager tasks (GH-124847) (#125340)

Browse source 
gh-124309: fix staggered race on eager tasks (GH-124847)

This patch is entirely by Thomas and Peter

(cherry picked from commit 979c0df7c0)

Co-authored-by: Thomas Grainger <tagrain@gmail.com>
Co-authored-by: Peter Bierma <zintensitydev@gmail.com>
This commit is contained in:
Miss Islington (bot) 2024-10-12 05:12:11 +02:00 • committed by GitHub
parent 2264c097e0
commit affffc7dda
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 88 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

17
Lib/asyncio/staggered.py
View file

@ -69,7 +69,11 @@ async def staggered_race(coro_fns, delay, *, loop=None):
exceptions = []
running_tasks = []
async def run_one_coro(previous_failed) -> None:
async def run_one_coro(ok_to_start, previous_failed) -> None:
# in eager tasks this waits for the calling task to append this task
# to running_tasks, in regular tasks this wait is a no-op that does
# not yield a future. See gh-124309.
await ok_to_start.wait()
# Wait for the previous task to finish, or for delay seconds
if previous_failed is not None:
with contextlib.suppress(exceptions_mod.TimeoutError):
@ -85,8 +89,12 @@ async def staggered_race(coro_fns, delay, *, loop=None):
return
# Start task that will run the next coroutine
this_failed = locks.Event()
next_task = loop.create_task(run_one_coro(this_failed))
next_ok_to_start = locks.Event()
next_task = loop.create_task(run_one_coro(next_ok_to_start, this_failed))
running_tasks.append(next_task)
# next_task has been appended to running_tasks so next_task is ok to
# start.
next_ok_to_start.set()
assert len(running_tasks) == this_index + 2
# Prepare place to put this coroutine's exceptions if not won
exceptions.append(None)
@ -116,8 +124,11 @@ async def staggered_race(coro_fns, delay, *, loop=None):
if i != this_index:
t.cancel()
first_task = loop.create_task(run_one_coro(None))
ok_to_start = locks.Event()
first_task = loop.create_task(run_one_coro(ok_to_start, None))
running_tasks.append(first_task)
# first_task has been appended to running_tasks so first_task is ok to start.
ok_to_start.set()
try:
# Wait for a growing list of tasks to all finish: poor man's version of
# curio's TaskGroup or trio's nursery