From 645f3a50963eeca2ade1ce18a001ad420a4ce268 Mon Sep 17 00:00:00 2001 From: umtong Date: Tue, 27 Oct 2026 02:43:50 +0900 Subject: [PATCH] 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. --- Lib/test/test_asyncgen.py | 109 +++++++ .../test_async_generators.py | 76 +++++ ...-10-26-00-00-00.gh-issue-160347.7Ztq4P.rst | 4 + Objects/genobject.c | 275 +++++++++++------- 4 files changed, 363 insertions(+), 101 deletions(-) create mode 100644 Misc/NEWS.d/next/Core_and_Builtins/2026-10-26-00-00-00.gh-issue-160347.7Ztq4P.rst diff --git a/Lib/test/test_asyncgen.py b/Lib/test/test_asyncgen.py index b5e0891feb794e..3900a4d5c04a4e 100644 --- a/Lib/test/test_asyncgen.py +++ b/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): diff --git a/Lib/test/test_free_threading/test_async_generators.py b/Lib/test/test_free_threading/test_async_generators.py index 7987ff6ed2fafe..88ba3bfb81094e 100644 --- a/Lib/test/test_free_threading/test_async_generators.py +++ b/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: diff --git a/Misc/NEWS.d/next/Core_and_Builtins/2026-10-26-00-00-00.gh-issue-160347.7Ztq4P.rst b/Misc/NEWS.d/next/Core_and_Builtins/2026-10-26-00-00-00.gh-issue-160347.7Ztq4P.rst new file mode 100644 index 00000000000000..aa03c39281ccf9 --- /dev/null +++ b/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. diff --git a/Objects/genobject.c b/Objects/genobject.c index 1c95a569af8b9b..f479128646b8f3 100644 --- a/Objects/genobject.c +++ b/Objects/genobject.c @@ -1617,28 +1617,12 @@ typedef enum { AWAITABLE_STATE_CLOSED, /* closed */ } AwaitableState; -#ifdef Py_GIL_DISABLED -static bool -async_gen_try_set_state(int8_t *state, int8_t *expected, int8_t new_state) -{ - return _Py_atomic_compare_exchange_int8(state, expected, new_state); -} - -# define _Py_ASYNC_GEN_TRY_SET_STATE(state, expected, new_state) \ - async_gen_try_set_state(&(state), &(expected), (new_state)) -#else -# define _Py_ASYNC_GEN_TRY_SET_STATE(state, expected, new_state) \ - ((state) = (new_state), true) -#endif - // Try to transition the async generator to the running state. // Returns false if it is already running. // -// There are two ways to concurrently iterate an async generator: by -// sharing a single asend()/athrow() object across threads, or with -// multiple asend()/athrow() objects sending to the same generator. -// The CAS on ags_state/agt_state handles the first case; the CAS on -// ag_running_async here handles the second. +// This excludes other awaitables until the current awaitable completes, +// including while it is suspended. Individual calls on that awaitable are +// guarded separately so concurrent calls cannot change its state. static bool async_gen_try_claim_running(PyAsyncGenObject *agen) { @@ -1656,6 +1640,34 @@ async_gen_try_claim_running(PyAsyncGenObject *agen) } +// Protect one send(), throw(), or close() call on an awaitable, including its +// state transitions. Unlike ag_running_async, this flag is cleared whenever +// the call returns, including when it yields a value from an inner await. +static bool +async_gen_awaitable_try_enter(PyAsyncGenObject *gen, int8_t *running) +{ +#ifdef Py_GIL_DISABLED + int8_t expected = 0; + if (_Py_atomic_compare_exchange_int8(running, &expected, 1)) { + return true; + } +#else + if (!*running) { + *running = 1; + return true; + } +#endif + gen_raise_already_executing_error(_PyGen_CAST(gen)); + return false; +} + +static void +async_gen_awaitable_exit(int8_t *running) +{ + FT_ATOMIC_STORE_INT8_RELEASE(*running, 0); +} + + typedef struct PyAsyncGenASend { PyObject_HEAD PyAsyncGenObject *ags_gen; @@ -1665,6 +1677,7 @@ typedef struct PyAsyncGenASend { PyObject *ags_sendval; int8_t ags_state; + int8_t ags_running; } PyAsyncGenASend; #define _PyAsyncGenASend_CAST(op) \ @@ -1682,6 +1695,7 @@ typedef struct PyAsyncGenAThrow { PyObject *agt_val; int8_t agt_state; + int8_t agt_running; } PyAsyncGenAThrow; @@ -2044,30 +2058,27 @@ async_gen_asend_traverse(PyObject *self, visitproc visit, void *arg) } -PySendResult -_PyAsyncGenASend_Send(PyObject *self, PyObject *arg, PyObject **presult) +static PySendResult +async_gen_asend_send_impl(PyObject *self, PyObject *arg, PyObject **presult) { PyAsyncGenASend *o = _PyAsyncGenASend_CAST(self); *presult = NULL; int8_t state = FT_ATOMIC_LOAD_INT8_RELAXED(o->ags_state); - do { - if (state == AWAITABLE_STATE_CLOSED) { - PyErr_SetString( - PyExc_RuntimeError, - "cannot reuse already awaited __anext__()/asend()"); - return PYGEN_ERROR; - } - if (state == AWAITABLE_STATE_ITER) { - goto do_send; - } - assert(state == AWAITABLE_STATE_INIT); - } while (!_Py_ASYNC_GEN_TRY_SET_STATE(o->ags_state, state, - AWAITABLE_STATE_ITER)); + if (state == AWAITABLE_STATE_CLOSED) { + PyErr_SetString( + PyExc_RuntimeError, + "cannot reuse already awaited __anext__()/asend()"); + return PYGEN_ERROR; + } + if (state == AWAITABLE_STATE_ITER) { + goto do_send; + } + assert(state == AWAITABLE_STATE_INIT); + FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_ITER); - // The transition above only guards this object, the generator may - // still be running through another asend()/athrow() object so - // try to claim it before running. + // The awaitable is protected against concurrent calls, but another + // awaitable may own the generator while suspended. Claim it before running. if (!async_gen_try_claim_running(o->ags_gen)) { FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED); PyErr_SetString( @@ -2094,6 +2105,19 @@ _PyAsyncGenASend_Send(PyObject *self, PyObject *arg, PyObject **presult) return res; } +PySendResult +_PyAsyncGenASend_Send(PyObject *self, PyObject *arg, PyObject **presult) +{ + PyAsyncGenASend *o = _PyAsyncGenASend_CAST(self); + if (!async_gen_awaitable_try_enter(o->ags_gen, &o->ags_running)) { + *presult = NULL; + return PYGEN_ERROR; + } + PySendResult result = async_gen_asend_send_impl(self, arg, presult); + async_gen_awaitable_exit(&o->ags_running); + return result; +} + static PyObject * async_gen_asend_send(PyObject *self, PyObject *arg) { @@ -2115,28 +2139,26 @@ async_gen_asend_iternext(PyObject *ags) static PyObject * -async_gen_asend_throw(PyObject *self, PyObject *const *args, Py_ssize_t nargs) +async_gen_asend_throw_impl(PyObject *self, PyObject *const *args, + Py_ssize_t nargs) { PyAsyncGenASend *o = _PyAsyncGenASend_CAST(self); int8_t state = FT_ATOMIC_LOAD_INT8_RELAXED(o->ags_state); - do { - if (state == AWAITABLE_STATE_CLOSED) { - PyErr_SetString( - PyExc_RuntimeError, - "cannot reuse already awaited __anext__()/asend()"); - return NULL; - } - if (state == AWAITABLE_STATE_ITER) { - goto do_throw; - } - assert(state == AWAITABLE_STATE_INIT); - } while (!_Py_ASYNC_GEN_TRY_SET_STATE(o->ags_state, state, - AWAITABLE_STATE_ITER)); + if (state == AWAITABLE_STATE_CLOSED) { + PyErr_SetString( + PyExc_RuntimeError, + "cannot reuse already awaited __anext__()/asend()"); + return NULL; + } + if (state == AWAITABLE_STATE_ITER) { + goto do_throw; + } + assert(state == AWAITABLE_STATE_INIT); + FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_ITER); - // The transition above only guards this object, the generator may - // still be running through another asend()/athrow() object so - // try to claim it before running. + // The awaitable is protected against concurrent calls, but another + // awaitable may own the generator while suspended. Claim it before running. if (!async_gen_try_claim_running(o->ags_gen)) { FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED); PyErr_SetString( @@ -2160,14 +2182,27 @@ async_gen_asend_throw(PyObject *self, PyObject *const *args, Py_ssize_t nargs) static PyObject * -async_gen_asend_close(PyObject *self, PyObject *args) +async_gen_asend_throw(PyObject *self, PyObject *const *args, Py_ssize_t nargs) +{ + PyAsyncGenASend *o = _PyAsyncGenASend_CAST(self); + if (!async_gen_awaitable_try_enter(o->ags_gen, &o->ags_running)) { + return NULL; + } + PyObject *result = async_gen_asend_throw_impl(self, args, nargs); + async_gen_awaitable_exit(&o->ags_running); + return result; +} + + +static PyObject * +async_gen_asend_close_impl(PyObject *self, PyObject *args) { PyAsyncGenASend *o = _PyAsyncGenASend_CAST(self); if (FT_ATOMIC_LOAD_INT8_RELAXED(o->ags_state) == AWAITABLE_STATE_CLOSED) { Py_RETURN_NONE; } - PyObject *result = async_gen_asend_throw(self, &PyExc_GeneratorExit, 1); + PyObject *result = async_gen_asend_throw_impl(self, &PyExc_GeneratorExit, 1); if (result == NULL) { if (PyErr_ExceptionMatches(PyExc_StopIteration) || PyErr_ExceptionMatches(PyExc_StopAsyncIteration) || @@ -2184,6 +2219,18 @@ async_gen_asend_close(PyObject *self, PyObject *args) return NULL; } +static PyObject * +async_gen_asend_close(PyObject *self, PyObject *args) +{ + PyAsyncGenASend *o = _PyAsyncGenASend_CAST(self); + if (!async_gen_awaitable_try_enter(o->ags_gen, &o->ags_running)) { + return NULL; + } + PyObject *result = async_gen_asend_close_impl(self, args); + async_gen_awaitable_exit(&o->ags_running); + return result; +} + static void async_gen_asend_finalize(PyObject *self) { @@ -2266,6 +2313,7 @@ async_gen_asend_new(PyAsyncGenObject *gen, PyObject *sendval) ags->ags_gen = (PyAsyncGenObject*)Py_NewRef(gen); ags->ags_sendval = Py_XNewRef(sendval); ags->ags_state = AWAITABLE_STATE_INIT; + ags->ags_running = 0; _PyObject_GC_TRACK((PyObject*)ags); return (PyObject*)ags; @@ -2393,7 +2441,7 @@ async_gen_athrow_traverse(PyObject *self, visitproc visit, void *arg) static PyObject * -async_gen_athrow_send(PyObject *self, PyObject *arg) +async_gen_athrow_send_impl(PyObject *self, PyObject *arg) { PyAsyncGenAThrow *o = _PyAsyncGenAThrow_CAST(self); PyGenObject *gen = _PyGen_CAST(o->agt_gen); @@ -2408,35 +2456,23 @@ async_gen_athrow_send(PyObject *self, PyObject *arg) } if (FRAME_STATE_FINISHED(FT_ATOMIC_LOAD_INT8_RELAXED(gen->gi_frame_state))) { - // Close the awaitable, unless another thread transitioned it - // to a different state in the meantime. - (void)_Py_ASYNC_GEN_TRY_SET_STATE(o->agt_state, state, - AWAITABLE_STATE_CLOSED); + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); PyErr_SetNone(PyExc_StopIteration); return NULL; } - do { - if (state == AWAITABLE_STATE_CLOSED) { - PyErr_SetString( - PyExc_RuntimeError, - "cannot reuse already awaited aclose()/athrow()"); - return NULL; - } - if (state == AWAITABLE_STATE_ITER) { - goto do_send; - } - assert(state == AWAITABLE_STATE_INIT); - if (arg != Py_None) { - PyErr_SetString(PyExc_RuntimeError, NON_INIT_CORO_MSG); - return NULL; - } - } while (!_Py_ASYNC_GEN_TRY_SET_STATE(o->agt_state, state, - AWAITABLE_STATE_ITER)); + if (state == AWAITABLE_STATE_ITER) { + goto do_send; + } + assert(state == AWAITABLE_STATE_INIT); + if (arg != Py_None) { + PyErr_SetString(PyExc_RuntimeError, NON_INIT_CORO_MSG); + return NULL; + } + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_ITER); - // The transition above only guards this object, the generator may - // still be running through another asend()/athrow() object so - // try to claim it before running. + // The awaitable is protected against concurrent calls, but another + // awaitable may own the generator while suspended. Claim it before running. if (!async_gen_try_claim_running(o->agt_gen)) { FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); if (o->agt_typ == NULL) { @@ -2537,28 +2573,39 @@ async_gen_athrow_send(PyObject *self, PyObject *arg) static PyObject * -async_gen_athrow_throw(PyObject *self, PyObject *const *args, Py_ssize_t nargs) +async_gen_athrow_send(PyObject *self, PyObject *arg) +{ + PyAsyncGenAThrow *o = _PyAsyncGenAThrow_CAST(self); + if (!async_gen_awaitable_try_enter(o->agt_gen, &o->agt_running)) { + return NULL; + } + PyObject *result = async_gen_athrow_send_impl(self, arg); + async_gen_awaitable_exit(&o->agt_running); + return result; +} + + +static PyObject * +async_gen_athrow_throw_impl(PyObject *self, PyObject *const *args, + Py_ssize_t nargs) { PyAsyncGenAThrow *o = _PyAsyncGenAThrow_CAST(self); int8_t state = FT_ATOMIC_LOAD_INT8_RELAXED(o->agt_state); - do { - if (state == AWAITABLE_STATE_CLOSED) { - PyErr_SetString( - PyExc_RuntimeError, - "cannot reuse already awaited aclose()/athrow()"); - return NULL; - } - if (state == AWAITABLE_STATE_ITER) { - goto do_throw; - } - assert(state == AWAITABLE_STATE_INIT); - } while (!_Py_ASYNC_GEN_TRY_SET_STATE(o->agt_state, state, - AWAITABLE_STATE_ITER)); + if (state == AWAITABLE_STATE_CLOSED) { + PyErr_SetString( + PyExc_RuntimeError, + "cannot reuse already awaited aclose()/athrow()"); + return NULL; + } + if (state == AWAITABLE_STATE_ITER) { + goto do_throw; + } + assert(state == AWAITABLE_STATE_INIT); + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_ITER); - // The transition above only guards this object, the generator may - // still be running through another asend()/athrow() object so - // try to claim it before running. + // The awaitable is protected against concurrent calls, but another + // awaitable may own the generator while suspended. Claim it before running. if (!async_gen_try_claim_running(o->agt_gen)) { FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); if (o->agt_typ == NULL) { @@ -2614,6 +2661,19 @@ async_gen_athrow_throw(PyObject *self, PyObject *const *args, Py_ssize_t nargs) } +static PyObject * +async_gen_athrow_throw(PyObject *self, PyObject *const *args, Py_ssize_t nargs) +{ + PyAsyncGenAThrow *o = _PyAsyncGenAThrow_CAST(self); + if (!async_gen_awaitable_try_enter(o->agt_gen, &o->agt_running)) { + return NULL; + } + PyObject *result = async_gen_athrow_throw_impl(self, args, nargs); + async_gen_awaitable_exit(&o->agt_running); + return result; +} + + static PyObject * async_gen_athrow_iternext(PyObject *agt) { @@ -2622,14 +2682,14 @@ async_gen_athrow_iternext(PyObject *agt) static PyObject * -async_gen_athrow_close(PyObject *self, PyObject *args) +async_gen_athrow_close_impl(PyObject *self, PyObject *args) { PyAsyncGenAThrow *agt = _PyAsyncGenAThrow_CAST(self); if (FT_ATOMIC_LOAD_INT8_RELAXED(agt->agt_state) == AWAITABLE_STATE_CLOSED) { Py_RETURN_NONE; } - PyObject *result = async_gen_athrow_throw((PyObject*)agt, - &PyExc_GeneratorExit, 1); + PyObject *result = async_gen_athrow_throw_impl((PyObject*)agt, + &PyExc_GeneratorExit, 1); if (result == NULL) { if (PyErr_ExceptionMatches(PyExc_StopIteration) || PyErr_ExceptionMatches(PyExc_StopAsyncIteration) || @@ -2647,6 +2707,18 @@ async_gen_athrow_close(PyObject *self, PyObject *args) } +static PyObject * +async_gen_athrow_close(PyObject *self, PyObject *args) +{ + PyAsyncGenAThrow *o = _PyAsyncGenAThrow_CAST(self); + if (!async_gen_awaitable_try_enter(o->agt_gen, &o->agt_running)) { + return NULL; + } + PyObject *result = async_gen_athrow_close_impl(self, args); + async_gen_awaitable_exit(&o->agt_running); + return result; +} + static void async_gen_athrow_finalize(PyObject *op) { @@ -2739,6 +2811,7 @@ async_gen_athrow_new(PyAsyncGenObject *gen, PyObject *args) o->agt_val = Py_XNewRef(val); o->agt_state = AWAITABLE_STATE_INIT; + o->agt_running = 0; _PyObject_GC_TRACK((PyObject*)o); return (PyObject*)o; }