Skip to content

bpo-46752: Introduce task groups in asyncio #31270

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gvanrossum merged 24 commits into from
Feb 15, 2022
Merged

bpo-46752: Introduce task groups in asyncio #31270

Changes from 1 commit
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
7edee0b
Integrate task groups from EdgeDb
gvanrossum Feb 10, 2022
f495375
Make test_taskgroups.py run and pass
gvanrossum Feb 10, 2022
a87275a
Rename taskgroup to taskgroups in the test code
gvanrossum Feb 10, 2022
4df0acc
Export TaskGroup from asyncio; remove __future__ import
gvanrossum Feb 10, 2022
56db921
Only keep the newest _is_base_error() and _task_cancel()
gvanrossum Feb 10, 2022
500581e
Get rid of MultiError in favor of ExceptionGroup
gvanrossum Feb 11, 2022
4843e94
Add TaskGroupError to __all__
gvanrossum Feb 11, 2022
63e712d
Avoid DeprecationWarning: There is no current event loop
gvanrossum Feb 11, 2022
d233dd1
Prevent warning "test altered the execution environment"
gvanrossum Feb 11, 2022
af574d5
Get rid of custom TaskGroupError
gvanrossum Feb 11, 2022
299f366
Update comments explaining why test 21 doesn't work
gvanrossum Feb 12, 2022
9de3c87
Add tests showing that 'plain' BaseExceptions work
gvanrossum Feb 12, 2022
0e1355d
Allow creating new tasks while __aexit__ is waiting
gvanrossum Feb 12, 2022
77ec0e4
Add an API to Task to manage 'cancel_requested' flag
gvanrossum Feb 14, 2022
17b64b5
Add tests for .cancelling() and .uncancel()
gvanrossum Feb 14, 2022
5e3f4b9
Merge remote-tracking branch 'origin/main' into taskgroups
gvanrossum Feb 14, 2022
0b9bccd
📜🤖 Added by blurb_it.
blurb-it[bot] Feb 14, 2022
137ebe6
Replace EdgeDb copyright with a simpler attribution
gvanrossum Feb 15, 2022
f693c1c
Use task.cancelling() in task repr instead of access to private attri…
asvetlov Feb 15, 2022
b83734c
Change the internal imports
gvanrossum Feb 15, 2022
de3d820
Avoid needing self.loop in test
gvanrossum Feb 15, 2022
9712241
Make test 14 more robust
gvanrossum Feb 15, 2022
b3d4d18
Update Lib/asyncio/taskgroups.py
1st1 Feb 15, 2022
c1e5d64
Update Lib/test/test_asyncio/test_taskgroups.py
1st1 Feb 15, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Rename taskgroup to taskgroups in the test code
  • Loading branch information
@gvanrossum
gvanrossum committed Feb 10, 2022
commit a87275a694fd4e40077472b7df92070584ea4e94
72 changes: 36 additions & 36 deletions Lib/test/test_asyncio/test_taskgroups.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@

import asyncio

from asyncio import taskgroups as taskgroup
from asyncio import taskgroups
import unittest


Expand All @@ -42,7 +42,7 @@ async def foo2():
await asyncio.sleep(0.2)
return 11

async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
t1 = g.create_task(foo1())
t2 = g.create_task(foo2())

Expand All @@ -59,7 +59,7 @@ async def foo2():
await asyncio.sleep(0.2)
return 11

async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
t1 = g.create_task(foo1())
await asyncio.sleep(0.15)
t2 = g.create_task(foo2())
Expand All @@ -77,7 +77,7 @@ async def foo2():
await asyncio.sleep(0.2)
return 11

async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
t1 = g.create_task(foo1())
await asyncio.sleep(0.15)
# cancel t1 explicitly, i.e. everything should continue
Expand Down Expand Up @@ -111,13 +111,13 @@ async def foo2():
async def runner():
nonlocal NUM, t2

async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
g.create_task(foo1())
t2 = g.create_task(foo2())

NUM += 10

with self.assertRaisesRegex(taskgroup.TaskGroupError,
with self.assertRaisesRegex(taskgroups.TaskGroupError,
r'1 sub errors: \(ZeroDivisionError\)'):
await self.loop.create_task(runner())
Comment thread
gvanrossum marked this conversation as resolved.
Outdated

Expand Down Expand Up @@ -147,7 +147,7 @@ async def foo2():
async def runner():
nonlocal NUM, runner_cancel

async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
g.create_task(foo1())
g.create_task(foo1())
g.create_task(foo1())
Expand All @@ -163,7 +163,7 @@ async def runner():
# The 3 foo1 sub tasks can be racy when the host is busy - if the
# cancellation happens in the middle, we'll see partial sub errors here
with self.assertRaisesRegex(
taskgroup.TaskGroupError,
taskgroups.TaskGroupError,
r'(1|2|3) sub errors: \(ZeroDivisionError\)',
):
await self.loop.create_task(runner())
Comment thread
gvanrossum marked this conversation as resolved.
Outdated
Expand All @@ -185,7 +185,7 @@ async def foo():
raise

async def runner():
async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
for _ in range(5):
g.create_task(foo())

Expand Down Expand Up @@ -213,7 +213,7 @@ async def foo():

async def runner():
nonlocal NUM
async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
for _ in range(5):
g.create_task(foo())

Expand All @@ -240,7 +240,7 @@ async def foo():
1 / 0

async def runner():
async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
for _ in range(5):
g.create_task(foo())

Expand Down Expand Up @@ -271,15 +271,15 @@ async def foo2():

async def runner():
nonlocal t1, t2
async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
t1 = g.create_task(foo1())
t2 = g.create_task(foo2())
await asyncio.sleep(0.1)
1 / 0

try:
await runner()
except taskgroup.TaskGroupError as t:
except taskgroups.TaskGroupError as t:
self.assertEqual(t.get_error_types(), {ZeroDivisionError})
else:
self.fail('TaskGroupError was not raised')
Expand All @@ -301,14 +301,14 @@ async def foo2():

async def runner():
nonlocal t1, t2
async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
t1 = g.create_task(foo1())
t2 = g.create_task(foo2())
1 / 0

try:
await runner()
except taskgroup.TaskGroupError as t:
except taskgroups.TaskGroupError as t:
self.assertEqual(t.get_error_types(), {ZeroDivisionError})
else:
self.fail('TaskGroupError was not raised')
Expand All @@ -323,8 +323,8 @@ async def foo():
1 / 0

async def runner():
async with taskgroup.TaskGroup():
async with taskgroup.TaskGroup() as g2:
async with taskgroups.TaskGroup():
async with taskgroups.TaskGroup() as g2:
for _ in range(5):
g2.create_task(foo())

Expand All @@ -348,10 +348,10 @@ async def foo():
1 / 0

async def runner():
async with taskgroup.TaskGroup() as g1:
async with taskgroups.TaskGroup() as g1:
g1.create_task(asyncio.sleep(10))

async with taskgroup.TaskGroup() as g2:
async with taskgroups.TaskGroup() as g2:
for _ in range(5):
g2.create_task(foo())

Expand All @@ -375,14 +375,14 @@ async def crash_after(t):
raise ValueError(t)

async def runner():
async with taskgroup.TaskGroup(name='g1') as g1:
async with taskgroups.TaskGroup(name='g1') as g1:
g1.create_task(crash_after(0.1))

async with taskgroup.TaskGroup(name='g2') as g2:
async with taskgroups.TaskGroup(name='g2') as g2:
g2.create_task(crash_after(0.2))

r = self.loop.create_task(runner())
Comment thread
gvanrossum marked this conversation as resolved.
Outdated
with self.assertRaisesRegex(taskgroup.TaskGroupError, r'1 sub errors'):
with self.assertRaisesRegex(taskgroups.TaskGroupError, r'1 sub errors'):
await r

async def test_taskgroup_14(self):
Expand All @@ -392,14 +392,14 @@ async def crash_after(t):
raise ValueError(t)

async def runner():
async with taskgroup.TaskGroup(name='g1') as g1:
async with taskgroups.TaskGroup(name='g1') as g1:
g1.create_task(crash_after(0.2))

async with taskgroup.TaskGroup(name='g2') as g2:
async with taskgroups.TaskGroup(name='g2') as g2:
g2.create_task(crash_after(0.1))
Comment on lines +392 to +394
Copy link
Copy Markdown
Contributor

@asvetlov asvetlov Feb 15, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This code seems not robust; it can fail on a slow machine: https://github.com/python/cpython/runs/5201217606?check_suite_focus=true.
Perhaps waiting for asyncio.Event (or future) is better than constant small sleeps (but sleeps are better readable, sure).

Copy link
Copy Markdown
Member Author

@gvanrossum gvanrossum Feb 15, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let me fix that by making the longer sleep longer -- that sleep will be interrupted so it doesn't matter how long it is. Let me know if you think that's not the right way to fix this test.


r = self.loop.create_task(runner())
Comment thread
gvanrossum marked this conversation as resolved.
Outdated
with self.assertRaisesRegex(taskgroup.TaskGroupError, r'1 sub errors'):
with self.assertRaisesRegex(taskgroups.TaskGroupError, r'1 sub errors'):
await r

async def test_taskgroup_15(self):
Expand All @@ -409,7 +409,7 @@ async def crash_soon():
1 / 0

async def runner():
async with taskgroup.TaskGroup(name='g1') as g1:
async with taskgroups.TaskGroup(name='g1') as g1:
g1.create_task(crash_soon())
try:
await asyncio.sleep(10)
Expand All @@ -432,7 +432,7 @@ async def crash_soon():
1 / 0

async def nested_runner():
async with taskgroup.TaskGroup(name='g1') as g1:
async with taskgroups.TaskGroup(name='g1') as g1:
g1.create_task(crash_soon())
try:
await asyncio.sleep(10)
Expand All @@ -457,7 +457,7 @@ async def test_taskgroup_17(self):

async def runner():
nonlocal NUM
async with taskgroup.TaskGroup():
async with taskgroups.TaskGroup():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
Expand All @@ -479,7 +479,7 @@ async def test_taskgroup_18(self):

async def runner():
nonlocal NUM
async with taskgroup.TaskGroup():
async with taskgroups.TaskGroup():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
Expand All @@ -496,7 +496,7 @@ async def runner():

try:
await r
except taskgroup.TaskGroupError as t:
except taskgroups.TaskGroupError as t:
self.assertEqual(t.get_error_types(), {MyExc})
else:
self.fail('TaskGroupError was not raised')
Expand All @@ -515,14 +515,14 @@ async def nested():
raise MyExc

async def runner():
async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
g.create_task(crash_soon())
await nested()

r = self.loop.create_task(runner())
Comment thread
gvanrossum marked this conversation as resolved.
Outdated
try:
await r
except taskgroup.TaskGroupError as t:
except taskgroups.TaskGroupError as t:
self.assertEqual(t.get_error_types(), {MyExc, ZeroDivisionError})
else:
self.fail('TasgGroupError was not raised')
Expand All @@ -539,7 +539,7 @@ async def nested():
raise KeyboardInterrupt

async def runner():
async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
g.create_task(crash_soon())
await nested()

Expand All @@ -561,7 +561,7 @@ async def nested():
raise TypeError

async def runner():
async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
g.create_task(crash_soon())
await nested()

Expand All @@ -579,7 +579,7 @@ async def foo2():
return 11

async def runner():
async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
g.create_task(foo1())
g.create_task(foo2())

Expand All @@ -595,7 +595,7 @@ async def test_taskgroup_23(self):
async def do_job(delay):
await asyncio.sleep(delay)

async with taskgroup.TaskGroup() as g:
async with taskgroups.TaskGroup() as g:
for count in range(10):
await asyncio.sleep(0.1)
g.create_task(do_job(0.3))
Expand Down