Free-threading: concurrent send() on a shared aclose() awaitable can resume the async generator instead of closing it

Start in Objects/genobject.c at async_gen_athrow_send(), especially the INIT-to-ITER transition, ag_running_async claim, and aclose _gen_throw() path described in the report. Compare the shared aclose and athrow paths, then run test_free_threading.test_async_generators.test_concurrent_shared_aclose and the related shared athrow test on a free-threaded debug build. Done when concurrent sends reliably close or throw into the generator and its cleanup runs.

Bug report

Bug description:

On the free-threaded build (--disable-gil --with-pydebug), test_free_threading.test_async_generators.test_concurrent_shared_aclose fails intermittently:

  File "Lib/test/test_free_threading/test_async_generators.py", line 171, in test_concurrent_shared_aclose
    self.assertEqual(cleanups, [1])
AssertionError: Lists differ: [] != [1]

Both send() calls on the shared aclose() awaitable raise (the test ignores RuntimeError/ValueError), and the generator's finally block never runs.

I think the cause is this window in async_gen_athrow_send():

1. Thread T1 moves agt_state from INIT to ITER (L2434-L2435), claims ag_running_async (L2440) and sets ag_closed (L2464), but has not yet called _gen_throw(GeneratorExit) (L2466-L2469).
2. Thread T2 sees ITER and takes the do_send path (L2427, L2488), which calls gen_send(gen, None) without claiming ag_running_async. The generator is resumed with None instead of GeneratorExit, runs on and yields again, and T2 gets RuntimeError: async generator ignored GeneratorExit.
3. Meanwhile, T1's _gen_throw() finds the frame executing and fails with ValueError: async generator already executing.

With the GIL there is no switch point between the state change and _gen_throw(), so this cannot happen there. The agt_state CAS and the separate ag_running_async claim were added in gh-120321 (61818b6087); the ITER → gen_send() path itself is older.

Deterministic reproducer: widen the window with a temporary delay before the _gen_throw() call in aclose mode (debug only, not a fix):

--- a/Objects/genobject.c
+++ b/Objects/genobject.c
@@ -3,6 +3,7 @@
 #define _PY_INTERPRETER
 
 #include "Python.h"
+#include <unistd.h>                // TEMPORARY: usleep()
 #include "pycore_call.h"          // _PyObject_CallNoArgs()
 #include "pycore_ceval.h"         // _PyEval_EvalFrame()
 #include "pycore_frame.h"         // _PyInterpreterFrame
@@ -2462,6 +2463,9 @@
     if (o->agt_typ == NULL) {
         /* aclose() mode */
         FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_closed, 1);
+        if (getenv("PY_ACLOSE_DELAY_US")) {  /* TEMPORARY: widen race window */
+            usleep((useconds_t)atoi(getenv("PY_ACLOSE_DELAY_US")));
+        }
 
         retval = _gen_throw((PyGenObject *)gen,
                             0,  /* Do not close generator when

then run the following. The time.sleep(0.3) keeps T2 inside the generator while T1 calls _gen_throw():

# Two threads call send(None) on ONE shared aclose() awaitable, 20 ms apart.
import threading, time

def make(cleanups):
    async def agen():
        try:
            while True:
                yield 1
                time.sleep(0.3)  # reached only if resumed by send(None), not by aclose()
        finally:
            cleanups.append(1)
    return agen()

for trial in range(5):
    cleanups = []
    ag = make(cleanups)
    try:
        ag.asend(None).send(None)          # advance to the first yield
    except StopIteration:
        pass
    aw = ag.aclose()
    res = {}
    def worker(name):
        try:
            aw.send(None)
            res[name] = "returned"
        except BaseException as e:
            res[name] = f"{type(e).__name__}: {e}"
    t1 = threading.Thread(target=worker, args=("T1",))
    t2 = threading.Thread(target=worker, args=("T2",))
    t1.start(); time.sleep(0.02); t2.start(); t1.join(); t2.join()
    print(trial, res, "cleanups:", cleanups)

With PY_ACLOSE_DELAY_US=100000 every trial fails:

0 {'T1': 'ValueError: async generator already executing', 'T2': 'RuntimeError: async generator ignored GeneratorExit'} cleanups: []

Without the variable all trials pass (cleanups: [1]).

Without any patch, running test_concurrent_shared_aclose in a loop in 8 parallel processes failed 12 times in 800,000 runs on main (1705e512b30), and 25 times in 800,000 on 5d887642d12 (just before gh-158304, which also touched async generators), so it is not a new regression. In the same setup test_concurrent_shared_athrow failed 17 times in 400,000 runs with StopIteration: 1 from the asend() check at line 142, so the generator was still alive: it was resumed instead of receiving the thrown ValueError. It goes through the same function, so I expect it is the same window. test_concurrent_shared_asend did not fail in 400,000 runs.

CPython versions tested on:

CPython main branch

Operating systems tested on:

macOS