From 3b523a844402e23c180c46de2c932ab99d0fbe49 Mon Sep 17 00:00:00 2001 From: losewayy Date: Sat, 14 Nov 2026 19:49:46 +0800 Subject: [PATCH] gh-160347: Fix concurrent driving race on async generator awaitables Add AWAITABLE_STATE_EXECUTING to guard each step of PyAsyncGenASend and PyAsyncGenAThrow while a call is running, instead of publishing ITER before the initial send or throw executes. Concurrent or reentrant calls on the same awaitable now raise ValueError without corrupting ag_running_async or closing an active operation. --- Lib/test/test_asyncgen.py | 78 ++++ .../test_async_generators.py | 166 +++++++ ...-11-14-11-49-16.gh-issue-160347.JirvQ7.rst | 4 + Objects/genobject.c | 404 ++++++++---------- 4 files changed, 421 insertions(+), 231 deletions(-) create mode 100644 Misc/NEWS.d/next/Core_and_Builtins/2026-11-14-11-49-16.gh-issue-160347.JirvQ7.rst diff --git a/Lib/test/test_asyncgen.py b/Lib/test/test_asyncgen.py index 47a091cb6e47a7e..57af4892ab56e60 100644 --- a/Lib/test/test_asyncgen.py +++ b/Lib/test/test_asyncgen.py @@ -665,6 +665,84 @@ async def agenfn(): gen.send(None) self.assertEqual(cm.exception.value, 2) + def test_async_gen_awaitable_reentrant_send(self): + class YieldOnce: + def __await__(self): + yield "inner" + + aw = None + seen_errors = [] + + async def reenter_on_asend(): + try: + aw.send(None) + except ValueError as exc: + seen_errors.append(str(exc)) + await YieldOnce() + yield 42 + + ag = reenter_on_asend() + aw = ag.asend(None) + self.assertEqual(aw.send(None), "inner") + self.assertEqual(seen_errors, ["async generator already executing"]) + self.assertTrue(ag.ag_running) + with self.assertRaises(StopIteration) as cm: + aw.send(None) + self.assertEqual(cm.exception.value, 42) + self.assertFalse(ag.ag_running) + with self.assertRaisesRegex(RuntimeError, "cannot reuse"): + aw.send(None) + + seen_errors.clear() + + async def reenter_on_aclose(): + try: + yield 1 + finally: + try: + aw.send(None) + except ValueError as exc: + seen_errors.append(("send", str(exc))) + try: + aw.throw(RuntimeError("boom")) + except ValueError as exc: + seen_errors.append(("throw", str(exc))) + await YieldOnce() + + ag = reenter_on_aclose() + with self.assertRaises(StopIteration): + ag.asend(None).send(None) + aw = ag.aclose() + self.assertEqual(aw.send(None), "inner") + self.assertEqual(seen_errors, [ + ("send", "async generator already executing"), + ("throw", "async generator already executing"), + ]) + self.assertTrue(ag.ag_running) + with self.assertRaises(StopIteration): + aw.send(None) + self.assertFalse(ag.ag_running) + with self.assertRaisesRegex(RuntimeError, "cannot reuse"): + aw.send(None) + + def test_async_gen_finished_aclose_athrow_non_none_send(self): + async def empty(): + if False: + yield + + for make_aw in ( + lambda ag: ag.aclose(), + lambda ag: ag.athrow(ValueError), + ): + ag = empty() + with self.assertRaises(StopAsyncIteration): + ag.asend(None).send(None) + aw = make_aw(ag) + with self.assertRaises(StopIteration): + aw.send(1) + with self.assertRaises(RuntimeError): + aw.send(None) + 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 7987ff6ed2fafec..f4a0db0568e237f 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) @@ -243,6 +245,170 @@ def worker(): threading_helper.run_concurrently(worker, self.NUM_THREADS) self.assertEqual(calls, [ag]) + def test_concurrent_shared_aclose_does_not_resume(self): + # gh-160347: concurrent calls on a shared aclose() awaitable must + # never resume the async generator past its suspended yield. + rounds = 500 + start_barrier = threading.Barrier(self.NUM_THREADS + 1) + done_barrier = threading.Barrier(self.NUM_THREADS + 1) + state = {"aw": None, "stop": False} + errors = [] + + def worker(): + try: + while True: + start_barrier.wait(timeout=support.SHORT_TIMEOUT) + if state["stop"]: + return + try: + state["aw"].send(None) + except (RuntimeError, ValueError, + StopIteration, StopAsyncIteration): + pass + done_barrier.wait(timeout=support.SHORT_TIMEOUT) + except threading.BrokenBarrierError: + return + except BaseException as exc: + errors.append(exc) + start_barrier.abort() + done_barrier.abort() + + threads = [threading.Thread(target=worker) + for _ in range(self.NUM_THREADS)] + started = [] + try: + for t in threads: + t.start() + started.append(t) + + for _ in range(rounds): + cleanups = [] + resumed = [] + + async def agen(): + try: + yield 1 + finally: + cleanups.append(1) + resumed.append(1) + yield 2 + + ag = agen() + with self.assertRaises(StopIteration): + ag.asend(None).send(None) + + aw = ag.aclose() + state["aw"] = aw + start_barrier.wait(timeout=support.SHORT_TIMEOUT) + done_barrier.wait(timeout=support.SHORT_TIMEOUT) + + self.assertEqual(cleanups, [1]) + self.assertEqual(resumed, []) + self.assertFalse(ag.ag_running) + self.assertRaises(RuntimeError, aw.send, None) + self.assertRaises( + StopAsyncIteration, ag.asend(None).send, None) + finally: + state["stop"] = True + try: + if not errors and len(started) == len(threads): + start_barrier.wait(timeout=support.SHORT_TIMEOUT) + else: + start_barrier.abort() + done_barrier.abort() + except threading.BrokenBarrierError: + start_barrier.abort() + done_barrier.abort() + for t in started: + t.join(timeout=support.SHORT_TIMEOUT) + self.assertFalse(t.is_alive()) + self.assertEqual(errors, []) + + def test_concurrent_shared_suspended_awaitable(self): + # Concurrent calls on a shared awaitable that is suspended at + # AWAITABLE_STATE_ITER must fail the racing caller without corrupting + # ag_running or closing the awaitable while the active step finishes. + for kind in ("asend", "athrow", "aclose"): + driver_in_step = threading.Event() + racer_done = threading.Event() + racer_errors = [] + + class SyncYieldTwice: + def __await__(self): + yield "first" + driver_in_step.set() + self.wait_ok = racer_done.wait( + timeout=support.SHORT_TIMEOUT) + yield "second" + + sync = SyncYieldTwice() + + async def agen_asend(): + await sync + yield 2 + + async def agen_athrow(): + try: + yield 1 + except ValueError: + await sync + yield 2 + + async def agen_aclose(): + try: + yield 1 + except GeneratorExit: + await sync + raise + + if kind == "asend": + ag = agen_asend() + aw = ag.asend(None) + elif kind == "athrow": + ag = agen_athrow() + with self.assertRaises(StopIteration): + ag.asend(None).send(None) + aw = ag.athrow(ValueError) + else: + ag = agen_aclose() + with self.assertRaises(StopIteration): + ag.asend(None).send(None) + aw = ag.aclose() + + # Advance to "first" so aw is suspended in AWAITABLE_STATE_ITER. + self.assertEqual(aw.send(None), "first") + self.assertTrue(ag.ag_running) + + driver_results = [] + + def driver(): + driver_results.append(aw.send(None)) + + def racer(): + if not driver_in_step.wait(timeout=support.SHORT_TIMEOUT): + racer_errors.append( + TimeoutError("driver did not enter step")) + return + try: + aw.send(None) + except ValueError as exc: + racer_errors.append(exc) + finally: + racer_done.set() + + threading_helper.run_concurrently([driver, racer]) + self.assertTrue(sync.wait_ok) + self.assertEqual(driver_results, ["second"]) + self.assertEqual(len(racer_errors), 1) + self.assertIsInstance(racer_errors[0], ValueError) + self.assertIn("already executing", str(racer_errors[0])) + self.assertTrue(ag.ag_running) + + with self.assertRaises(StopIteration): + aw.send(None) + self.assertFalse(ag.ag_running) + self.assertRaises(RuntimeError, aw.send, None) + if __name__ == "__main__": unittest.main() diff --git a/Misc/NEWS.d/next/Core_and_Builtins/2026-11-14-11-49-16.gh-issue-160347.JirvQ7.rst b/Misc/NEWS.d/next/Core_and_Builtins/2026-11-14-11-49-16.gh-issue-160347.JirvQ7.rst new file mode 100644 index 000000000000000..bdef48bc26aba60 --- /dev/null +++ b/Misc/NEWS.d/next/Core_and_Builtins/2026-11-14-11-49-16.gh-issue-160347.JirvQ7.rst @@ -0,0 +1,4 @@ +Fix state corruption caused by concurrent or reentrant calls on the same async +generator awaitable (``asend()``, ``athrow()``, or ``aclose()``). This could +prevent an ``aclose()`` operation from closing the generator or make an active +awaitable unusable. diff --git a/Objects/genobject.c b/Objects/genobject.c index 82a1bb1ff420682..724068b82921995 100644 --- a/Objects/genobject.c +++ b/Objects/genobject.c @@ -1612,9 +1612,10 @@ PyCoro_New(PyFrameObject *f, PyObject *name, PyObject *qualname) typedef enum { - AWAITABLE_STATE_INIT, /* new awaitable, has not yet been iterated */ - AWAITABLE_STATE_ITER, /* being iterated */ - AWAITABLE_STATE_CLOSED, /* closed */ + AWAITABLE_STATE_INIT, /* new awaitable, has not yet been iterated */ + AWAITABLE_STATE_ITER, /* suspended mid-iteration */ + AWAITABLE_STATE_EXECUTING, /* currently executing a step */ + AWAITABLE_STATE_CLOSED, /* closed */ } AwaitableState; #ifdef Py_GIL_DISABLED @@ -1655,6 +1656,31 @@ async_gen_try_claim_running(PyAsyncGenObject *agen) #endif } +// Try to transition an async generator awaitable from INIT or ITER to +// EXECUTING. Returns the previous state on success, or -1 with an exception +// set. Callers must transition the state back to AWAITABLE_STATE_ITER on +// yield, or to AWAITABLE_STATE_CLOSED (and release ag_running_async if +// claimed) on exit. +static int +async_gen_awaitable_try_enter(int8_t *state_ptr, PyAsyncGenObject *gen, + const char *closed_msg) +{ + int8_t state = FT_ATOMIC_LOAD_INT8_RELAXED(*state_ptr); + do { + if (state == AWAITABLE_STATE_CLOSED) { + PyErr_SetString(PyExc_RuntimeError, closed_msg); + return -1; + } + if (state == AWAITABLE_STATE_EXECUTING) { + gen_raise_already_executing_error((PyGenObject *)gen); + return -1; + } + assert(state == AWAITABLE_STATE_INIT || state == AWAITABLE_STATE_ITER); + } while (!_Py_ASYNC_GEN_TRY_SET_STATE(*state_ptr, state, + AWAITABLE_STATE_EXECUTING)); + return state; +} + typedef struct PyAsyncGenASend { PyObject_HEAD @@ -2050,43 +2076,37 @@ _PyAsyncGenASend_Send(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) { + int prev_state = async_gen_awaitable_try_enter( + &o->ags_state, o->ags_gen, + "cannot reuse already awaited __anext__()/asend()"); + if (prev_state < 0) { + return PYGEN_ERROR; + } + + if (prev_state == AWAITABLE_STATE_INIT) { + // 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. + if (!async_gen_try_claim_running(o->ags_gen)) { + FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED); PyErr_SetString( PyExc_RuntimeError, - "cannot reuse already awaited __anext__()/asend()"); + "anext(): asynchronous generator is already running"); 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)); - - // 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. - 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 PYGEN_ERROR; - } - if (arg == NULL || arg == Py_None) { - arg = o->ags_sendval; + if (arg == NULL || arg == Py_None) { + arg = o->ags_sendval; + } } - PyObject *result; - PySendResult res; -do_send: - result = gen_send((PyObject*)o->ags_gen, arg); - res = async_gen_unwrap_send(o->ags_gen, result, presult); + PyObject *result = gen_send((PyObject*)o->ags_gen, arg); + PySendResult res = async_gen_unwrap_send(o->ags_gen, result, presult); - if (res != PYGEN_NEXT) { + if (res == PYGEN_NEXT) { + FT_ATOMIC_STORE_INT8_RELEASE(o->ags_state, AWAITABLE_STATE_ITER); + } + else { FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED); FT_ATOMIC_STORE_INT8_RELEASE(o->ags_gen->ag_running_async, 0); } @@ -2119,38 +2139,33 @@ async_gen_asend_throw(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) { + int prev_state = async_gen_awaitable_try_enter( + &o->ags_state, o->ags_gen, + "cannot reuse already awaited __anext__()/asend()"); + if (prev_state < 0) { + return NULL; + } + + if (prev_state == AWAITABLE_STATE_INIT) { + // 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. + if (!async_gen_try_claim_running(o->ags_gen)) { + FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED); PyErr_SetString( PyExc_RuntimeError, - "cannot reuse already awaited __anext__()/asend()"); + "anext(): asynchronous generator is already running"); 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)); - - // 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. - 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; } - PyObject *result; -do_throw: - result = gen_throw((PyObject*)o->ags_gen, args, nargs); + PyObject *result = gen_throw((PyObject*)o->ags_gen, args, nargs); result = async_gen_unwrap_value(o->ags_gen, result); - if (result == NULL) { + if (result != NULL) { + FT_ATOMIC_STORE_INT8_RELEASE(o->ags_state, AWAITABLE_STATE_ITER); + } + else { FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED); FT_ATOMIC_STORE_INT8_RELEASE(o->ags_gen->ag_running_async, 0); } @@ -2398,146 +2413,119 @@ async_gen_athrow_traverse(PyObject *self, visitproc visit, void *arg) static PyObject * -async_gen_athrow_send(PyObject *self, PyObject *arg) +async_gen_athrow_finish(PyAsyncGenAThrow *o, PyObject *retval) { - PyAsyncGenAThrow *o = _PyAsyncGenAThrow_CAST(self); - PyGenObject *gen = _PyGen_CAST(o->agt_gen); - PyObject *retval; + if (o->agt_typ) { + retval = async_gen_unwrap_value(o->agt_gen, retval); + if (retval != NULL) { + FT_ATOMIC_STORE_INT8_RELEASE(o->agt_state, AWAITABLE_STATE_ITER); + } + else { + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); + FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0); + } + return retval; + } - 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()"); + /* aclose() mode */ + if (retval && _PyAsyncGenWrappedValue_CheckExact(retval)) { + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); + 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_RELEASE(o->agt_state, AWAITABLE_STATE_ITER); + return retval; + } - 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); + FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0); + if (PyErr_ExceptionMatches(PyExc_StopAsyncIteration) || + PyErr_ExceptionMatches(PyExc_GeneratorExit)) + { + /* when aclose() is called we don't want to propagate + StopAsyncIteration or GeneratorExit; just raise + StopIteration, signalling that this 'aclose()' await + is done. + */ + PyErr_Clear(); PyErr_SetNone(PyExc_StopIteration); + } + return NULL; +} + +static PyObject * +async_gen_athrow_send(PyObject *self, PyObject *arg) +{ + PyAsyncGenAThrow *o = _PyAsyncGenAThrow_CAST(self); + PyGenObject *gen = _PyGen_CAST(o->agt_gen); + + int prev_state = async_gen_awaitable_try_enter( + &o->agt_state, o->agt_gen, + "cannot reuse already awaited aclose()/athrow()"); + if (prev_state < 0) { return NULL; } - do { - if (state == AWAITABLE_STATE_CLOSED) { - PyErr_SetString( - PyExc_RuntimeError, - "cannot reuse already awaited aclose()/athrow()"); + PyObject *retval; + if (prev_state == AWAITABLE_STATE_INIT) { + 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 (state == AWAITABLE_STATE_ITER) { - goto do_send; - } - assert(state == AWAITABLE_STATE_INIT); + if (arg != Py_None) { + FT_ATOMIC_STORE_INT8_RELEASE(o->agt_state, AWAITABLE_STATE_INIT); PyErr_SetString(PyExc_RuntimeError, NON_INIT_CORO_MSG); return NULL; } - } while (!_Py_ASYNC_GEN_TRY_SET_STATE(o->agt_state, 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. - 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 { + // 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. + if (!async_gen_try_claim_running(o->agt_gen)) { + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); PyErr_SetString( PyExc_RuntimeError, - "athrow(): asynchronous generator is already running"); - } - return NULL; - } - - 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_RELEASE(o->agt_gen->ag_running_async, 0); - PyErr_SetNone(PyExc_StopAsyncIteration); - return NULL; - } - - if (o->agt_typ == NULL) { - /* aclose() mode */ - FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_closed, 1); - - 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; + o->agt_typ == NULL + ? "aclose(): asynchronous generator is already running" + : "athrow(): asynchronous generator is already running"); + return NULL; } - } 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; -do_send: - retval = gen_send((PyObject *)gen, arg); - if (o->agt_typ) { - retval = async_gen_unwrap_value(o->agt_gen, retval); - if (retval == NULL) { + 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_RELEASE(o->agt_gen->ag_running_async, 0); + PyErr_SetNone(PyExc_StopAsyncIteration); + return NULL; } - return retval; - } else { - /* aclose() mode */ - if (retval) { - if (_PyAsyncGenWrappedValue_CheckExact(retval)) { - Py_DECREF(retval); - goto yield_close; - } - else { - return retval; - } + + if (o->agt_typ == NULL) { + /* aclose() mode */ + FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_closed, 1); + + retval = _gen_throw((PyGenObject *)gen, + 0, /* Do not close generator when + PyExc_GeneratorExit is passed */ + PyExc_GeneratorExit, NULL, NULL); } else { - goto check_error; + retval = _gen_throw((PyGenObject *)gen, + 0, /* Do not close generator when + PyExc_GeneratorExit is passed */ + o->agt_typ, o->agt_val, o->agt_tb); } } - -yield_close: - 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_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)) - { - if (o->agt_typ == NULL) { - /* when aclose() is called we don't want to propagate - StopAsyncIteration or GeneratorExit; just raise - StopIteration, signalling that this 'aclose()' await - is done. - */ - PyErr_Clear(); - PyErr_SetNone(PyExc_StopIteration); - } + else { + retval = gen_send((PyObject *)gen, arg); } - return NULL; + + return async_gen_athrow_finish(o, retval); } @@ -2546,76 +2534,30 @@ async_gen_athrow_throw(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)); - - // 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. - 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"); - } + int prev_state = async_gen_awaitable_try_enter( + &o->agt_state, o->agt_gen, + "cannot reuse already awaited aclose()/athrow()"); + if (prev_state < 0) { return NULL; } - PyObject *retval; -do_throw: - 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) { + if (prev_state == AWAITABLE_STATE_INIT) { + // 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. + if (!async_gen_try_claim_running(o->agt_gen)) { FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); - FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0); - } - return retval; - } - else { - /* aclose() mode */ - if (retval && _PyAsyncGenWrappedValue_CheckExact(retval)) { - FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED); - FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0); - Py_DECREF(retval); - PyErr_SetString(PyExc_RuntimeError, ASYNC_GEN_IGNORED_EXIT_MSG); + PyErr_SetString( + PyExc_RuntimeError, + o->agt_typ == NULL + ? "aclose(): asynchronous generator is already running" + : "athrow(): asynchronous generator is already running"); return NULL; } - if (retval == NULL) { - 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)) - { - /* when aclose() is called we don't want to propagate - StopAsyncIteration or GeneratorExit; just raise - StopIteration, signalling that this 'aclose()' await - is done. - */ - PyErr_Clear(); - PyErr_SetNone(PyExc_StopIteration); - } - return retval; } + + PyObject *retval = gen_throw((PyObject*)o->agt_gen, args, nargs); + return async_gen_athrow_finish(o, retval); }