[mypyc] Fix free-threading race condition in argument parsing (#21613)

Global argument parser state initialization was missing synchronization.

Don't use a mutex on the hot path as an optimization.

Fixes #21578 (probably). I added a regression test that failed about 30%
of time on master (on macOS).

I used coding agent assist.

Author: JukkaL

mypyc/lib-rt/getargsfast.c

@@ -18,7 +18,22 @@
 #include <Python.h>
 #include "CPy.h"
 
-#define PARSER_INITED(parser) ((parser)->kwtuple != NULL)
+// The kwtuple field doubles as the "parser has been initialized" flag: it is
+// written last (after all other parser fields) and read first. On free-threaded
+// builds the fast paths read it without holding any lock, so the write must be a
+// release store and the reads acquire loads. That way a thread that observes a
+// non-NULL kwtuple is guaranteed to also see the fully-initialized min/max/etc.
+// fields. On GIL builds these are plain accesses with no overhead.
+#ifdef Py_GIL_DISABLED
+#define PARSER_KWTUPLE(parser) _Py_atomic_load_ptr_acquire(&(parser)->kwtuple)
+#define SET_PARSER_KWTUPLE(parser, value) \
+    _Py_atomic_store_ptr_release(&(parser)->kwtuple, (value))
+#else
+#define PARSER_KWTUPLE(parser) ((parser)->kwtuple)
+#define SET_PARSER_KWTUPLE(parser, value) ((parser)->kwtuple = (value))
+#endif
+
+#define PARSER_INITED(parser) (PARSER_KWTUPLE(parser) != NULL)
 
 /* Forward */
 static int
@@ -115,19 +130,21 @@ CPyArg_ParseStackAndKeywordsSimple(PyObject *const *args, Py_ssize_t nargs, PyOb
 /* List of static parsers. */
 static struct CPyArg_Parser *static_arg_parsers = NULL;
 
+#ifdef Py_GIL_DISABLED
+// Serializes one-time initialization of parsers and insertion into the
+// static_arg_parsers list. Only contended the first time a given compiled
+// function is called; once a parser is initialized the fast paths never lock.
+static PyMutex static_arg_parsers_mutex;
+#endif
+
 static int
-parser_init(CPyArg_Parser *parser)
+parser_init_locked(CPyArg_Parser *parser)
 {
     const char * const *keywords;
     const char *format;
     int i, len, min, max, nkw;
     PyObject *kwtuple;
 
-    assert(parser->keywords != NULL);
-    if (PARSER_INITED(parser)) {
-        return 1;
-    }
-
     keywords = parser->keywords;
     /* scan keywords and count the number of positional-only parameters */
     for (i = 0; keywords[i] && !*keywords[i]; i++) {
@@ -244,14 +261,53 @@ parser_init(CPyArg_Parser *parser)
         PyUnicode_InternInPlace(&str);
         PyTuple_SET_ITEM(kwtuple, i, str);
     }
-    parser->kwtuple = kwtuple;
 
     assert(parser->next == NULL);
     parser->next = static_arg_parsers;
     static_arg_parsers = parser;
+
+    // Publish the parser last: storing kwtuple marks it as initialized, so all
+    // other fields (and the list insertion above) must already be in place. On
+    // free-threaded builds this is a release store paired with the acquire loads
+    // in PARSER_INITED/PARSER_KWTUPLE.
+    SET_PARSER_KWTUPLE(parser, kwtuple);
     return 1;
 }
 
+// Cold path of parser_init: perform the one-time initialization. On
+// free-threaded builds this is serialized so that only one thread builds the
+// parser and inserts it into the static_arg_parsers list.
+static CPy_NOINLINE int
+parser_init_slow(CPyArg_Parser *parser)
+{
+#ifdef Py_GIL_DISABLED
+    PyMutex_Lock(&static_arg_parsers_mutex);
+    // Re-check now that we hold the lock: another thread may have initialized
+    // the parser while we were waiting.
+    if (PARSER_INITED(parser)) {
+        PyMutex_Unlock(&static_arg_parsers_mutex);
+        return 1;
+    }
+    int retval = parser_init_locked(parser);
+    PyMutex_Unlock(&static_arg_parsers_mutex);
+    return retval;
+#else
+    return parser_init_locked(parser);
+#endif
+}
+
+// Hot path: a parser is almost always already initialized, so keep the common
+// case inline and branch out to parser_init_slow only on first use.
+static inline int
+parser_init(CPyArg_Parser *parser)
+{
+    assert(parser->keywords != NULL);
+    if (likely(PARSER_INITED(parser))) {
+        return 1;
+    }
+    return parser_init_slow(parser);
+}
+
 static PyObject*
 find_keyword(PyObject *kwnames, PyObject *const *kwstack, PyObject *key)
 {

mypyc/test-data/run-functions.test

@@ -1524,3 +1524,86 @@ def test_multiple_params() -> None:
     # Different parameter name than loop variable
     funcs2 = uses_multiple_params_different_name(["a", "b"], "!")
     assert [f() for f in funcs2] == ["a!", "b!"]
+
+[case testConcurrentFirstCallWithKeywordArgs]
+# Regression test for a free-threading data race in argument parser
+# initialization. The first call to a compiled function with keyword arguments
+# lazily initializes a static CPyArg_Parser and pushes it onto a global list.
+# When many threads race that first call on a free-threaded build, the
+# initialization and list insertion must be synchronized, or the runtime hits
+# a failed assertion (parser->next == NULL) and aborts.
+import sys
+import threading
+
+# A pool of distinct functions, none of which is called before the concurrent
+# test. Each function's first call lazily initializes a static CPyArg_Parser,
+# and the test races that initialization across threads. Keyword arguments
+# force the call through the slow path that runs parser_init. Using many
+# functions gives many independent races per run, so the test reliably triggers
+# the bug on an unfixed free-threaded build instead of depending on a single
+# narrow timing window.
+def g0(a: int, b: int, c: int) -> int: return a + b + c
+def g1(a: int, b: int, c: int) -> int: return a + b + c
+def g2(a: int, b: int, c: int) -> int: return a + b + c
+def g3(a: int, b: int, c: int) -> int: return a + b + c
+def g4(a: int, b: int, c: int) -> int: return a + b + c
+def g5(a: int, b: int, c: int) -> int: return a + b + c
+def g6(a: int, b: int, c: int) -> int: return a + b + c
+def g7(a: int, b: int, c: int) -> int: return a + b + c
+def g8(a: int, b: int, c: int) -> int: return a + b + c
+def g9(a: int, b: int, c: int) -> int: return a + b + c
+def g10(a: int, b: int, c: int) -> int: return a + b + c
+def g11(a: int, b: int, c: int) -> int: return a + b + c
+def g12(a: int, b: int, c: int) -> int: return a + b + c
+def g13(a: int, b: int, c: int) -> int: return a + b + c
+def g14(a: int, b: int, c: int) -> int: return a + b + c
+def g15(a: int, b: int, c: int) -> int: return a + b + c
+def g16(a: int, b: int, c: int) -> int: return a + b + c
+def g17(a: int, b: int, c: int) -> int: return a + b + c
+def g18(a: int, b: int, c: int) -> int: return a + b + c
+def g19(a: int, b: int, c: int) -> int: return a + b + c
+def g20(a: int, b: int, c: int) -> int: return a + b + c
+def g21(a: int, b: int, c: int) -> int: return a + b + c
+def g22(a: int, b: int, c: int) -> int: return a + b + c
+def g23(a: int, b: int, c: int) -> int: return a + b + c
+def g24(a: int, b: int, c: int) -> int: return a + b + c
+def g25(a: int, b: int, c: int) -> int: return a + b + c
+def g26(a: int, b: int, c: int) -> int: return a + b + c
+def g27(a: int, b: int, c: int) -> int: return a + b + c
+def g28(a: int, b: int, c: int) -> int: return a + b + c
+def g29(a: int, b: int, c: int) -> int: return a + b + c
+
+FUNCS = [g0, g1, g2, g3, g4, g5, g6, g7, g8, g9, g10, g11, g12, g13, g14,
+         g15, g16, g17, g18, g19, g20, g21, g22, g23, g24, g25, g26, g27,
+         g28, g29]
+
+def is_gil_disabled() -> bool:
+    return hasattr(sys, "_is_gil_enabled") and not sys._is_gil_enabled()
+
+def test_concurrent_first_call() -> None:
+    if not is_gil_disabled():
+        # The race can only happen without the GIL.
+        return
+
+    num_threads = 16
+    barrier = threading.Barrier(num_threads)
+    errors: list[str] = []
+
+    def run() -> None:
+        # Line up all threads, then let them race freely through the list. The
+        # first call to each function lazily initializes its parser, so every
+        # function is a fresh race under real parallel pressure.
+        barrier.wait()
+        try:
+            for fn in FUNCS:
+                assert fn(a=1, b=2, c=3) == 6
+        except BaseException as e:
+            errors.append(repr(e))
+
+    threads = [threading.Thread(target=run) for _ in range(num_threads)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert not errors, errors