gh-160347: Guard async generator awaitables against concurrent entry

Protect send(), throw(), and close() on ASend and AThrow awaitables so
concurrent or reentrant callers cannot corrupt the active operation.
Cover same-thread reentry, direct await, and coordinated thread races.

Author: umtong

Lib/test/test_asyncgen.py

@@ -648,6 +648,115 @@ async def agenfn():
             gen.send(None)
         self.assertEqual(cm.exception.value, 2)
 
+    def test_async_gen_awaitable_reentrant(self):
+        # Rejecting reentry must not close the awaitable or release its
+        # claim on the generator while the original call is still running.
+        class MyError(Exception):
+            pass
+
+        cases = (
+            ('asend', (None,), StopAsyncIteration),
+            ('athrow', (MyError,), MyError),
+            ('aclose', (), StopIteration),
+        )
+        for kind, args, completion in cases:
+            for method in ('send', 'throw', 'close'):
+                with self.subTest(kind=kind, method=method):
+                    errors = []
+                    running = []
+                    cleanups = []
+
+                    def drive():
+                        if method == 'send':
+                            return aw.send(None)
+                        if method == 'throw':
+                            return aw.throw(MyError)
+                        return aw.close()
+
+                    async def agen():
+                        try:
+                            yield 1
+                        finally:
+                            try:
+                                drive()
+                            except ValueError as exc:
+                                errors.append(str(exc))
+                            running.append(ag.ag_running)
+                            if method != 'close':
+                                await awaitable()
+                            cleanups.append(1)
+
+                    ag = agen()
+                    self.addCleanup(ag.aclose().close)
+                    with self.assertRaises(StopIteration):
+                        ag.__anext__().send(None)
+                    aw = getattr(ag, kind)(*args)
+                    result = drive()
+
+                    self.assertEqual(
+                        errors, ['async generator already executing'])
+                    self.assertEqual(running, [True])
+                    if method == 'close':
+                        self.assertIsNone(result)
+                    else:
+                        self.assertEqual(result, ('result',))
+                        self.assertTrue(ag.ag_running)
+                        error = MyError if method == 'throw' else completion
+                        with self.assertRaises(error):
+                            aw.send(None)
+                    self.assertEqual(cleanups, [1])
+                    self.assertIsNone(ag.ag_frame)
+                    with self.assertRaisesRegex(
+                        RuntimeError, 'cannot reuse already awaited'
+                    ):
+                        aw.send(None)
+
+    def test_async_gen_awaitable_reentrant_await(self):
+        # Awaiting asend() uses its am_send slot rather than its send method.
+        class MyError(Exception):
+            pass
+
+        cases = (
+            ('asend', (None,), StopAsyncIteration),
+            ('athrow', (MyError,), MyError),
+            ('aclose', (), StopIteration),
+        )
+        for kind, args, completion in cases:
+            with self.subTest(kind=kind):
+                cleanups = []
+
+                async def drive():
+                    await aw
+
+                async def agen():
+                    try:
+                        yield 1
+                    finally:
+                        reentrant = drive()
+                        with self.assertRaisesRegex(
+                            ValueError, 'async generator already executing'
+                        ):
+                            reentrant.send(None)
+                        await awaitable()
+                        cleanups.append(1)
+
+                ag = agen()
+                self.addCleanup(ag.aclose().close)
+                with self.assertRaises(StopIteration):
+                    ag.__anext__().send(None)
+                aw = getattr(ag, kind)(*args)
+                owner = drive()
+                try:
+                    self.assertEqual(owner.send(None), ('result',))
+                    self.assertTrue(ag.ag_running)
+                    with self.assertRaises(completion):
+                        owner.send(None)
+                    self.assertEqual(cleanups, [1])
+                    self.assertIsNone(ag.ag_frame)
+                finally:
+                    with contextlib.suppress(RuntimeError):
+                        owner.close()
+
 
 class AsyncGenAsyncioTest(unittest.TestCase):
 

Lib/test/test_free_threading/test_async_generators.py

@@ -1,6 +1,8 @@
 import sys
+import threading
 import unittest
 
+from test import support
 from test.support import threading_helper
 
 threading_helper.requires_working_threading(module=True)
@@ -173,6 +175,80 @@ def worker():
         # The awaitable is closed after the operation completed.
         self.assertRaises(RuntimeError, aw.send, None)
 
+    def test_concurrent_shared_awaitable_resume(self):
+        class MyError(Exception):
+            pass
+
+        class YieldOnce:
+            def __await__(self):
+                yield 'suspended'
+
+        cases = (
+            ('asend', (None,), StopAsyncIteration),
+            ('athrow', (MyError,), MyError),
+            ('aclose', (), StopIteration),
+        )
+        for kind, args, completion in cases:
+            for method in ('send', 'throw', 'close'):
+                with self.subTest(kind=kind, method=method):
+                    entered = threading.Event()
+                    release = threading.Event()
+                    cleanups = []
+
+                    async def agen():
+                        try:
+                            yield 1
+                        finally:
+                            entered.set()
+                            self.assertTrue(
+                                release.wait(support.SHORT_TIMEOUT))
+                            await YieldOnce()
+                            cleanups.append(1)
+
+                    ag = agen()
+                    self.addCleanup(ag.aclose().close)
+                    with self.assertRaises(StopIteration):
+                        ag.__anext__().send(None)
+                    aw = getattr(ag, kind)(*args)
+
+                    def owner():
+                        self.assertEqual(aw.send(None), 'suspended')
+
+                    def contender():
+                        try:
+                            self.assertTrue(
+                                entered.wait(support.SHORT_TIMEOUT))
+                            with self.assertRaisesRegex(
+                                ValueError, 'async generator already executing'
+                            ):
+                                if method == 'send':
+                                    aw.send(None)
+                                elif method == 'throw':
+                                    aw.throw(MyError)
+                                else:
+                                    aw.close()
+                            self.assertTrue(ag.ag_running)
+                        finally:
+                            release.set()
+
+                    threading_helper.run_concurrently([owner, contender])
+                    # Rejecting the contender must leave the owner's pending
+                    # await intact, including its claim on the generator.
+                    self.assertTrue(ag.ag_running)
+                    with self.assertRaisesRegex(
+                        RuntimeError,
+                        'asynchronous generator is already running',
+                    ):
+                        ag.__anext__().send(None)
+                    with self.assertRaises(completion):
+                        aw.send(None)
+                    self.assertEqual(cleanups, [1])
+                    self.assertIsNone(ag.ag_frame)
+                    with self.assertRaisesRegex(
+                        RuntimeError, 'cannot reuse already awaited'
+                    ):
+                        aw.send(None)
+
     def test_concurrent_anext_athrow(self):
         async def agen():
             while True:

Misc/NEWS.d/next/Core_and_Builtins/2026-10-26-00-00-00.gh-issue-160347.7Ztq4P.rst

@@ -0,0 +1,4 @@
+Prevent concurrent or reentrant calls to the same asynchronous generator
+awaitable from corrupting its state. This fixes races in :meth:`~agen.aclose`,
+:meth:`~agen.athrow`, and :meth:`~agen.asend` that could delay cleanup or make an
+active awaitable unusable.