From 5f94aa11ebcdeb7be7e95133cd50148c2775e1c4 Mon Sep 17 00:00:00 2001 From: Kumar Aditya Date: Sat, 1 Aug 2026 16:08:01 +0530 Subject: [PATCH 1/3] fix thread safety of concurrently iterating over async generators --- Include/internal/pycore_interpframe_structs.h | 6 +- .../test_async_generators.py | 248 ++++++++++++++ Objects/genobject.c | 304 +++++++++++------- 3 files changed, 436 insertions(+), 122 deletions(-) create mode 100644 Lib/test/test_free_threading/test_async_generators.py diff --git a/Include/internal/pycore_interpframe_structs.h b/Include/internal/pycore_interpframe_structs.h index 4d267e35504b94d..560fb66a97fc89d 100644 --- a/Include/internal/pycore_interpframe_structs.h +++ b/Include/internal/pycore_interpframe_structs.h @@ -66,9 +66,9 @@ struct _PyInterpreterFrame { PyObject *prefix##_qualname; \ _PyErr_StackItem prefix##_exc_state; \ PyObject *prefix##_origin_or_finalizer; \ - char prefix##_hooks_inited; \ - char prefix##_closed; \ - char prefix##_running_async; \ + int8_t prefix##_hooks_inited; \ + int8_t prefix##_closed; \ + int8_t prefix##_running_async; \ /* The frame */ \ int8_t prefix##_frame_state; \ _PyInterpreterFrame prefix##_iframe; \ diff --git a/Lib/test/test_free_threading/test_async_generators.py b/Lib/test/test_free_threading/test_async_generators.py new file mode 100644 index 000000000000000..7e474533ede732f --- /dev/null +++ b/Lib/test/test_free_threading/test_async_generators.py @@ -0,0 +1,248 @@ +import sys +import unittest + +from test.support import threading_helper + +threading_helper.requires_working_threading(module=True) + + +class TestFTAsyncGenerators(unittest.TestCase): + NUM_THREADS = 4 + + def test_concurrent_anext(self): + # Each yielded value must be delivered to exactly one thread. + async def agen(): + for i in range(100): + yield i + + ag = agen() + values = [] + + def drive(): + while True: + try: + ag.asend(None).send(None) + except StopIteration as e: + values.append(e.value) + except StopAsyncIteration: + break + except (RuntimeError, ValueError): + # Another thread is currently driving the generator. + continue + + threading_helper.run_concurrently(drive, self.NUM_THREADS) + self.assertEqual(sorted(values), list(range(100))) + + def test_concurrent_athrow(self): + # Each thrown exception must be delivered to the generator + # exactly once. + received = [] + + async def agen(): + while True: + try: + yield 1 + except ValueError: + received.append(1) + + ag = agen() + with self.assertRaises(StopIteration): + ag.asend(None).send(None) # advance to the first yield + + delivered = [] + + def worker(): + for _ in range(50): + try: + ag.athrow(ValueError).send(None) + except StopIteration as e: + # The generator received the exception and yielded again. + delivered.append(e.value) + except (RuntimeError, ValueError): + # Another thread is currently driving the generator. + pass + + threading_helper.run_concurrently(worker, self.NUM_THREADS) + self.assertEqual(len(received), len(delivered)) + + def test_concurrent_aclose(self): + # The generator must be cleaned up exactly once. + cleanups = [] + + async def agen(): + try: + while True: + yield 1 + finally: + cleanups.append(1) + + ag = agen() + with self.assertRaises(StopIteration): + ag.asend(None).send(None) # advance to the first yield + + def worker(): + try: + ag.aclose().send(None) + except StopIteration: + # aclose() completed. + pass + except StopAsyncIteration: + # The generator was already closed. + pass + except (RuntimeError, ValueError): + # Another thread is currently driving the generator. + pass + + threading_helper.run_concurrently(worker, self.NUM_THREADS) + self.assertEqual(cleanups, [1]) + self.assertRaises(StopAsyncIteration, ag.asend(None).send, None) + + def test_concurrent_shared_asend(self): + # Multiple threads racing on a single asend awaitable: the value + # must be delivered exactly once. + async def agen(): + yield 1 + + ag = agen() + aw = ag.asend(None) + results = [] + + def worker(): + try: + aw.send(None) + except StopIteration as e: + results.append(e.value) + except (RuntimeError, ValueError, StopAsyncIteration): + pass + + threading_helper.run_concurrently(worker, self.NUM_THREADS) + self.assertEqual(results, [1]) + # The awaitable is closed after the operation completed. + self.assertRaises(RuntimeError, aw.send, None) + + def test_concurrent_shared_athrow(self): + # Multiple threads racing on a single athrow awaitable. + async def agen(): + while True: + yield 1 + + ag = agen() + with self.assertRaises(StopIteration): + ag.asend(None).send(None) # advance to the first yield + aw = ag.athrow(ValueError) + + def worker(): + try: + aw.send(None) + except (RuntimeError, ValueError, + StopIteration, StopAsyncIteration): + pass + + threading_helper.run_concurrently(worker, self.NUM_THREADS) + self.assertRaises(StopAsyncIteration, ag.asend(None).send, None) + # The awaitable is closed after the operation completed. + self.assertRaises(RuntimeError, aw.send, None) + + def test_concurrent_shared_aclose(self): + # Multiple threads racing on a single aclose awaitable: the + # generator must be cleaned up exactly once. + cleanups = [] + + async def agen(): + try: + while True: + yield 1 + finally: + cleanups.append(1) + + ag = agen() + with self.assertRaises(StopIteration): + ag.asend(None).send(None) # advance to the first yield + aw = ag.aclose() + + def worker(): + try: + aw.send(None) + except (RuntimeError, ValueError, + StopIteration, StopAsyncIteration): + pass + + threading_helper.run_concurrently(worker, self.NUM_THREADS) + self.assertEqual(cleanups, [1]) + self.assertRaises(StopAsyncIteration, ag.asend(None).send, None) + # The awaitable is closed after the operation completed. + self.assertRaises(RuntimeError, aw.send, None) + + def test_concurrent_anext_athrow(self): + async def agen(): + while True: + try: + yield 1 + except ValueError: + pass + + ag = agen() + + def worker(): + for i in range(100): + try: + if i % 2: + ag.asend(None).send(None) + else: + ag.athrow(ValueError).send(None) + except (RuntimeError, ValueError, + StopIteration, StopAsyncIteration): + pass + + threading_helper.run_concurrently(worker, self.NUM_THREADS) + + def test_concurrent_anext_aclose(self): + async def agen(): + for i in range(100): + yield i + + ag = agen() + + def anext_worker(): + for _ in range(100): + try: + ag.asend(None).send(None) + except (RuntimeError, ValueError, + StopIteration, StopAsyncIteration): + pass + + def aclose_worker(): + for _ in range(100): + try: + ag.aclose().send(None) + except (RuntimeError, ValueError, + StopIteration, StopAsyncIteration): + pass + + threading_helper.run_concurrently( + [anext_worker, aclose_worker, anext_worker, aclose_worker]) + + def test_firstiter_hook_called_once(self): + # Racing the first iteration must invoke the firstiter hook + # exactly once. + async def agen(): + yield 1 + + ag = agen() + calls = [] + + def worker(): + # Async generator hooks are per-thread state. + sys.set_asyncgen_hooks(firstiter=calls.append) + try: + ag.asend(None).send(None) + except (RuntimeError, ValueError, + StopIteration, StopAsyncIteration): + pass + + threading_helper.run_concurrently(worker, self.NUM_THREADS) + self.assertEqual(calls, [ag]) + + +if __name__ == "__main__": + unittest.main() diff --git a/Objects/genobject.c b/Objects/genobject.c index 3cdc06733363d3e..e57cd938827c1b6 100644 --- a/Objects/genobject.c +++ b/Objects/genobject.c @@ -1604,6 +1604,38 @@ 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. +static bool +async_gen_try_claim_running(PyAsyncGenObject *agen) +{ +#ifdef Py_GIL_DISABLED + int8_t expected = 0; + return _Py_atomic_compare_exchange_int8(&agen->ag_running_async, + &expected, 1); +#else + if (agen->ag_running_async) { + return false; + } + agen->ag_running_async = 1; + return true; +#endif +} + typedef struct PyAsyncGenASend { PyObject_HEAD @@ -1613,7 +1645,7 @@ typedef struct PyAsyncGenASend { (equivalent of "asend(None)") */ PyObject *ags_sendval; - AwaitableState ags_state; + int8_t ags_state; } PyAsyncGenASend; #define _PyAsyncGenASend_CAST(op) \ @@ -1630,7 +1662,7 @@ typedef struct PyAsyncGenAThrow { PyObject *agt_tb; PyObject *agt_val; - AwaitableState agt_state; + int8_t agt_state; } PyAsyncGenAThrow; @@ -1672,11 +1704,17 @@ async_gen_init_hooks(PyAsyncGenObject *o) PyObject *finalizer; PyObject *firstiter; +#ifdef Py_GIL_DISABLED + if (_Py_atomic_exchange_int8(&o->ag_hooks_inited, 1)) { + return 0; + } +#else if (o->ag_hooks_inited) { return 0; } o->ag_hooks_inited = 1; +#endif tstate = _PyThreadState_GET(); @@ -1919,10 +1957,10 @@ async_gen_unwrap_value(PyAsyncGenObject *gen, PyObject *result) if (PyErr_ExceptionMatches(PyExc_StopAsyncIteration) || PyErr_ExceptionMatches(PyExc_GeneratorExit) ) { - gen->ag_closed = 1; + FT_ATOMIC_STORE_INT8_RELAXED(gen->ag_closed, 1); } - gen->ag_running_async = 0; + FT_ATOMIC_STORE_INT8_RELAXED(gen->ag_running_async, 0); return NULL; } @@ -1930,7 +1968,7 @@ async_gen_unwrap_value(PyAsyncGenObject *gen, PyObject *result) /* async yield */ _PyGen_SetStopIterationValue(((_PyAsyncGenWrappedValue*)result)->agw_val); Py_DECREF(result); - gen->ag_running_async = 0; + FT_ATOMIC_STORE_INT8_RELAXED(gen->ag_running_async, 0); return NULL; } @@ -1974,34 +2012,41 @@ static PyObject * async_gen_asend_send(PyObject *self, PyObject *arg) { PyAsyncGenASend *o = _PyAsyncGenASend_CAST(self); - if (o->ags_state == AWAITABLE_STATE_CLOSED) { - PyErr_SetString( - PyExc_RuntimeError, - "cannot reuse already awaited __anext__()/asend()"); - return NULL; - } - if (o->ags_state == AWAITABLE_STATE_INIT) { - if (o->ags_gen->ag_running_async) { - o->ags_state = AWAITABLE_STATE_CLOSED; + int8_t state = FT_ATOMIC_LOAD_INT8_RELAXED(o->ags_state); + do { + if (state == AWAITABLE_STATE_CLOSED) { PyErr_SetString( PyExc_RuntimeError, - "anext(): asynchronous generator is already running"); + "cannot reuse already awaited __anext__()/asend()"); return NULL; } - - if (arg == NULL || arg == Py_None) { - arg = o->ags_sendval; + if (state == AWAITABLE_STATE_ITER) { + goto do_send; } - o->ags_state = AWAITABLE_STATE_ITER; + assert(state == AWAITABLE_STATE_INIT); + } while (!_Py_ASYNC_GEN_TRY_SET_STATE(o->ags_state, state, + AWAITABLE_STATE_ITER)); + + // INIT -> ITER transition succeeded, this is the first send. + if (!async_gen_try_claim_running(o->ags_gen)) { + FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED); + PyErr_SetString( + PyExc_RuntimeError, + "anext(): asynchronous generator is already running"); + return NULL; + } + + if (arg == NULL || arg == Py_None) { + arg = o->ags_sendval; } - o->ags_gen->ag_running_async = 1; +do_send:; PyObject *result = gen_send((PyObject*)o->ags_gen, arg); result = async_gen_unwrap_value(o->ags_gen, result); if (result == NULL) { - o->ags_state = AWAITABLE_STATE_CLOSED; + FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED); } return result; @@ -2033,32 +2078,36 @@ async_gen_asend_throw(PyObject *self, PyObject *const *args, Py_ssize_t nargs) { PyAsyncGenASend *o = _PyAsyncGenASend_CAST(self); - if (o->ags_state == AWAITABLE_STATE_CLOSED) { - PyErr_SetString( - PyExc_RuntimeError, - "cannot reuse already awaited __anext__()/asend()"); - return NULL; - } - - if (o->ags_state == AWAITABLE_STATE_INIT) { - if (o->ags_gen->ag_running_async) { - o->ags_state = AWAITABLE_STATE_CLOSED; + int8_t state = FT_ATOMIC_LOAD_INT8_RELAXED(o->ags_state); + do { + if (state == AWAITABLE_STATE_CLOSED) { PyErr_SetString( PyExc_RuntimeError, - "anext(): asynchronous generator is already running"); + "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)); - o->ags_state = AWAITABLE_STATE_ITER; - o->ags_gen->ag_running_async = 1; + if (!async_gen_try_claim_running(o->ags_gen)) { + FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED); + PyErr_SetString( + PyExc_RuntimeError, + "anext(): asynchronous generator is already running"); + return NULL; } +do_throw:; PyObject *result = gen_throw((PyObject*)o->ags_gen, args, nargs); result = async_gen_unwrap_value(o->ags_gen, result); if (result == NULL) { - o->ags_gen->ag_running_async = 0; - o->ags_state = AWAITABLE_STATE_CLOSED; + FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED); + FT_ATOMIC_STORE_INT8_RELAXED(o->ags_gen->ag_running_async, 0); } return result; @@ -2069,7 +2118,7 @@ static PyObject * async_gen_asend_close(PyObject *self, PyObject *args) { PyAsyncGenASend *o = _PyAsyncGenASend_CAST(self); - if (o->ags_state == AWAITABLE_STATE_CLOSED) { + if (FT_ATOMIC_LOAD_INT8_RELAXED(o->ags_state) == AWAITABLE_STATE_CLOSED) { Py_RETURN_NONE; } @@ -2304,77 +2353,90 @@ async_gen_athrow_send(PyObject *self, PyObject *arg) PyGenObject *gen = _PyGen_CAST(o->agt_gen); PyObject *retval; - if (o->agt_state == AWAITABLE_STATE_CLOSED) { + int8_t state = FT_ATOMIC_LOAD_INT8_RELAXED(o->agt_state); + if (state == AWAITABLE_STATE_CLOSED) { PyErr_SetString( PyExc_RuntimeError, "cannot reuse already awaited aclose()/athrow()"); return NULL; } - if (FRAME_STATE_FINISHED(gen->gi_frame_state)) { - o->agt_state = AWAITABLE_STATE_CLOSED; + if (FRAME_STATE_FINISHED(FT_ATOMIC_LOAD_INT8_RELAXED(gen->gi_frame_state))) { + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); PyErr_SetNone(PyExc_StopIteration); return NULL; } - if (o->agt_state == AWAITABLE_STATE_INIT) { - if (o->agt_gen->ag_running_async) { - o->agt_state = AWAITABLE_STATE_CLOSED; - if (o->agt_typ == NULL) { - PyErr_SetString( - PyExc_RuntimeError, - "aclose(): asynchronous generator is already running"); - } - else { - PyErr_SetString( - PyExc_RuntimeError, - "athrow(): asynchronous generator is already running"); - } + do { + if (state == AWAITABLE_STATE_CLOSED) { + PyErr_SetString( + PyExc_RuntimeError, + "cannot reuse already awaited aclose()/athrow()"); return NULL; } - - if (o->agt_gen->ag_closed) { - o->agt_state = AWAITABLE_STATE_CLOSED; - PyErr_SetNone(PyExc_StopAsyncIteration); - return NULL; + if (state == AWAITABLE_STATE_ITER) { + goto do_send; } + assert(state == AWAITABLE_STATE_INIT); + } while (!_Py_ASYNC_GEN_TRY_SET_STATE(o->agt_state, state, + AWAITABLE_STATE_ITER)); - if (arg != Py_None) { - PyErr_SetString(PyExc_RuntimeError, NON_INIT_CORO_MSG); - return NULL; + // INIT -> ITER transition succeeded, this is the first send. + 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) { + PyErr_SetString( + PyExc_RuntimeError, + "aclose(): asynchronous generator is already running"); + } + else { + PyErr_SetString( + PyExc_RuntimeError, + "athrow(): asynchronous generator is already running"); } + return NULL; + } - o->agt_state = AWAITABLE_STATE_ITER; - o->agt_gen->ag_running_async = 1; + if (FT_ATOMIC_LOAD_INT8_RELAXED(o->agt_gen->ag_closed)) { + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); + PyErr_SetNone(PyExc_StopAsyncIteration); + return NULL; + } - if (o->agt_typ == NULL) { - /* aclose() mode */ - o->agt_gen->ag_closed = 1; + if (arg != Py_None) { + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_INIT); + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); + PyErr_SetString(PyExc_RuntimeError, NON_INIT_CORO_MSG); + return NULL; + } - retval = _gen_throw((PyGenObject *)gen, - 0, /* Do not close generator when - PyExc_GeneratorExit is passed */ - PyExc_GeneratorExit, NULL, NULL); + if (o->agt_typ == NULL) { + /* aclose() mode */ + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_closed, 1); - if (retval && _PyAsyncGenWrappedValue_CheckExact(retval)) { - Py_DECREF(retval); - goto yield_close; - } - } else { - retval = _gen_throw((PyGenObject *)gen, - 0, /* Do not close generator when - PyExc_GeneratorExit is passed */ - o->agt_typ, o->agt_val, o->agt_tb); - retval = async_gen_unwrap_value(o->agt_gen, retval); - } - if (retval == NULL) { - goto check_error; + retval = _gen_throw((PyGenObject *)gen, + 0, /* Do not close generator when + PyExc_GeneratorExit is passed */ + PyExc_GeneratorExit, NULL, NULL); + + if (retval && _PyAsyncGenWrappedValue_CheckExact(retval)) { + Py_DECREF(retval); + goto yield_close; } - return retval; + } else { + retval = _gen_throw((PyGenObject *)gen, + 0, /* Do not close generator when + PyExc_GeneratorExit is passed */ + o->agt_typ, o->agt_val, o->agt_tb); + retval = async_gen_unwrap_value(o->agt_gen, retval); } + if (retval == NULL) { + goto check_error; + } + return retval; - assert(o->agt_state == AWAITABLE_STATE_ITER); - +do_send: retval = gen_send((PyObject *)gen, arg); if (o->agt_typ) { return async_gen_unwrap_value(o->agt_gen, retval); @@ -2395,15 +2457,15 @@ async_gen_athrow_send(PyObject *self, PyObject *arg) } yield_close: - o->agt_gen->ag_running_async = 0; - o->agt_state = AWAITABLE_STATE_CLOSED; +FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); +FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); PyErr_SetString( PyExc_RuntimeError, ASYNC_GEN_IGNORED_EXIT_MSG); return NULL; check_error: - o->agt_gen->ag_running_async = 0; - o->agt_state = AWAITABLE_STATE_CLOSED; + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); if (PyErr_ExceptionMatches(PyExc_StopAsyncIteration) || PyErr_ExceptionMatches(PyExc_GeneratorExit)) { @@ -2426,54 +2488,58 @@ async_gen_athrow_throw(PyObject *self, PyObject *const *args, Py_ssize_t nargs) { PyAsyncGenAThrow *o = _PyAsyncGenAThrow_CAST(self); - if (o->agt_state == AWAITABLE_STATE_CLOSED) { - PyErr_SetString( - PyExc_RuntimeError, - "cannot reuse already awaited aclose()/athrow()"); - return NULL; - } - - if (o->agt_state == AWAITABLE_STATE_INIT) { - if (o->agt_gen->ag_running_async) { - o->agt_state = AWAITABLE_STATE_CLOSED; - if (o->agt_typ == NULL) { - PyErr_SetString( - PyExc_RuntimeError, - "aclose(): asynchronous generator is already running"); - } - else { - PyErr_SetString( - PyExc_RuntimeError, - "athrow(): asynchronous generator is already running"); - } + 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)); - o->agt_state = AWAITABLE_STATE_ITER; - o->agt_gen->ag_running_async = 1; + 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) { + PyErr_SetString( + PyExc_RuntimeError, + "aclose(): asynchronous generator is already running"); + } + else { + PyErr_SetString( + PyExc_RuntimeError, + "athrow(): asynchronous generator is already running"); + } + return NULL; } +do_throw:; PyObject *retval = gen_throw((PyObject*)o->agt_gen, args, nargs); if (o->agt_typ) { retval = async_gen_unwrap_value(o->agt_gen, retval); if (retval == NULL) { - o->agt_gen->ag_running_async = 0; - o->agt_state = AWAITABLE_STATE_CLOSED; + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); } return retval; } else { /* aclose() mode */ if (retval && _PyAsyncGenWrappedValue_CheckExact(retval)) { - o->agt_gen->ag_running_async = 0; - o->agt_state = AWAITABLE_STATE_CLOSED; + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); Py_DECREF(retval); PyErr_SetString(PyExc_RuntimeError, ASYNC_GEN_IGNORED_EXIT_MSG); return NULL; } if (retval == NULL) { - o->agt_gen->ag_running_async = 0; - o->agt_state = AWAITABLE_STATE_CLOSED; + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); } if (PyErr_ExceptionMatches(PyExc_StopAsyncIteration) || PyErr_ExceptionMatches(PyExc_GeneratorExit)) @@ -2502,7 +2568,7 @@ static PyObject * async_gen_athrow_close(PyObject *self, PyObject *args) { PyAsyncGenAThrow *agt = _PyAsyncGenAThrow_CAST(self); - if (agt->agt_state == AWAITABLE_STATE_CLOSED) { + if (FT_ATOMIC_LOAD_INT8_RELAXED(agt->agt_state) == AWAITABLE_STATE_CLOSED) { Py_RETURN_NONE; } PyObject *result = async_gen_athrow_throw((PyObject*)agt, From 3f54668e4d04ea705f9b676fcff5b86be2081a8a Mon Sep 17 00:00:00 2001 From: Kumar Aditya Date: Sat, 1 Aug 2026 17:51:58 +0530 Subject: [PATCH 2/3] add more tests --- .../test_async_generators.py | 6 ++--- Objects/genobject.c | 27 ++++++++++--------- 2 files changed, 18 insertions(+), 15 deletions(-) diff --git a/Lib/test/test_free_threading/test_async_generators.py b/Lib/test/test_free_threading/test_async_generators.py index 7e474533ede732f..7987ff6ed2fafec 100644 --- a/Lib/test/test_free_threading/test_async_generators.py +++ b/Lib/test/test_free_threading/test_async_generators.py @@ -26,7 +26,7 @@ def drive(): values.append(e.value) except StopAsyncIteration: break - except (RuntimeError, ValueError): + except RuntimeError: # Another thread is currently driving the generator. continue @@ -58,7 +58,7 @@ def worker(): except StopIteration as e: # The generator received the exception and yielded again. delivered.append(e.value) - except (RuntimeError, ValueError): + except RuntimeError: # Another thread is currently driving the generator. pass @@ -89,7 +89,7 @@ def worker(): except StopAsyncIteration: # The generator was already closed. pass - except (RuntimeError, ValueError): + except RuntimeError: # Another thread is currently driving the generator. pass diff --git a/Objects/genobject.c b/Objects/genobject.c index e57cd938827c1b6..d2a6cea7e2033ee 100644 --- a/Objects/genobject.c +++ b/Objects/genobject.c @@ -1960,7 +1960,6 @@ async_gen_unwrap_value(PyAsyncGenObject *gen, PyObject *result) FT_ATOMIC_STORE_INT8_RELAXED(gen->ag_closed, 1); } - FT_ATOMIC_STORE_INT8_RELAXED(gen->ag_running_async, 0); return NULL; } @@ -1968,7 +1967,6 @@ async_gen_unwrap_value(PyAsyncGenObject *gen, PyObject *result) /* async yield */ _PyGen_SetStopIterationValue(((_PyAsyncGenWrappedValue*)result)->agw_val); Py_DECREF(result); - FT_ATOMIC_STORE_INT8_RELAXED(gen->ag_running_async, 0); return NULL; } @@ -2047,6 +2045,7 @@ do_send:; if (result == NULL) { FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED); + FT_ATOMIC_STORE_INT8_RELEASE(o->ags_gen->ag_running_async, 0); } return result; @@ -2107,7 +2106,7 @@ do_throw:; if (result == NULL) { FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED); - FT_ATOMIC_STORE_INT8_RELAXED(o->ags_gen->ag_running_async, 0); + FT_ATOMIC_STORE_INT8_RELEASE(o->ags_gen->ag_running_async, 0); } return result; @@ -2399,14 +2398,14 @@ async_gen_athrow_send(PyObject *self, PyObject *arg) if (FT_ATOMIC_LOAD_INT8_RELAXED(o->agt_gen->ag_closed)) { FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); - FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); + FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0); PyErr_SetNone(PyExc_StopAsyncIteration); return NULL; } if (arg != Py_None) { FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_INIT); - FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); + FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0); PyErr_SetString(PyExc_RuntimeError, NON_INIT_CORO_MSG); return NULL; } @@ -2439,7 +2438,11 @@ async_gen_athrow_send(PyObject *self, PyObject *arg) do_send: retval = gen_send((PyObject *)gen, arg); if (o->agt_typ) { - return async_gen_unwrap_value(o->agt_gen, retval); + retval = async_gen_unwrap_value(o->agt_gen, retval); + if (retval == NULL) { + FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0); + } + return retval; } else { /* aclose() mode */ if (retval) { @@ -2457,15 +2460,15 @@ async_gen_athrow_send(PyObject *self, PyObject *arg) } yield_close: -FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); -FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); + FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0); PyErr_SetString( PyExc_RuntimeError, ASYNC_GEN_IGNORED_EXIT_MSG); return NULL; check_error: - FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); + FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0); if (PyErr_ExceptionMatches(PyExc_StopAsyncIteration) || PyErr_ExceptionMatches(PyExc_GeneratorExit)) { @@ -2524,7 +2527,7 @@ do_throw:; retval = async_gen_unwrap_value(o->agt_gen, retval); if (retval == NULL) { FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); - FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); + FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0); } return retval; } @@ -2532,14 +2535,14 @@ do_throw:; /* aclose() mode */ if (retval && _PyAsyncGenWrappedValue_CheckExact(retval)) { FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); - FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); + FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0); Py_DECREF(retval); PyErr_SetString(PyExc_RuntimeError, ASYNC_GEN_IGNORED_EXIT_MSG); return NULL; } if (retval == NULL) { FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); - FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_running_async, 0); + FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0); } if (PyErr_ExceptionMatches(PyExc_StopAsyncIteration) || PyErr_ExceptionMatches(PyExc_GeneratorExit)) From 8e8902ae84cdfb9f509078c8fb345d6d1ecdae3e Mon Sep 17 00:00:00 2001 From: Kumar Aditya Date: Sat, 1 Aug 2026 18:02:54 +0530 Subject: [PATCH 3/3] add news --- .../2026-08-01-18-04-11.gh-issue-120321.k3XvQb.rst | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 Misc/NEWS.d/next/Core_and_Builtins/2026-08-01-18-04-11.gh-issue-120321.k3XvQb.rst diff --git a/Misc/NEWS.d/next/Core_and_Builtins/2026-08-01-18-04-11.gh-issue-120321.k3XvQb.rst b/Misc/NEWS.d/next/Core_and_Builtins/2026-08-01-18-04-11.gh-issue-120321.k3XvQb.rst new file mode 100644 index 000000000000000..0c65897d9834716 --- /dev/null +++ b/Misc/NEWS.d/next/Core_and_Builtins/2026-08-01-18-04-11.gh-issue-120321.k3XvQb.rst @@ -0,0 +1,3 @@ +Fix thread safety of :term:`asynchronous generators ` when iterated, closed or thrown into concurrently from multiple +threads on the :term:`free threading` build.