From 204abce70e976b68097cc704200db5f62de71250 Mon Sep 17 00:00:00 2001 From: BREW <81044362+b-re-w@users.noreply.github.com> Date: Sun, 25 Oct 2026 16:42:24 +0900 Subject: [PATCH] Test: ThreadSanitizer over parallel regions, with a detached-worker mutant (#277) --- .../src/test/python/test_parallel_loops.py | 203 ++++++++++++++++++ 1 file changed, 203 insertions(+) diff --git a/python-multiplatform-ksp/src/test/python/test_parallel_loops.py b/python-multiplatform-ksp/src/test/python/test_parallel_loops.py index b7591fe5..2764efe8 100644 --- a/python-multiplatform-ksp/src/test/python/test_parallel_loops.py +++ b/python-multiplatform-ksp/src/test/python/test_parallel_loops.py @@ -1777,3 +1777,206 @@ def test_bitwise_equals_interpreter_for_every_thread_count(bbuilt, threads): assert r["regions"] == 0 and r["reruns"] == 0 elif (os.cpu_count() or 1) > 1: assert r["regions"] > 0 and r["reruns"] >= 1 + + +# ===================================================================================================== +# ThreadSanitizer over the parallel regions (SPEC N-15, #277: "race tests and TSan pass on 3.15t"). +# The interpreter is not instrumented; TSan sees the generated C and the runtime header, which is where +# the workers, the partial slots and the join live. A report counts when the first non-sanitizer frame +# of one of its two accesses is in the extension (same rule as test_free_threaded.py). +# ===================================================================================================== + +TSOURCE = '''\ +# typedpython: compiled + +# typedpython: parallel +def t_scale(xs: list[float], a: float, b: float) -> None: + for i in range(len(xs)): + xs[i] = xs[i] * a + b + + +# typedpython: parallel +def t_sum(xs: list[int]) -> int: + s: int = 0 + for i in range(len(xs)): + s += xs[i] + return s + + +# typedpython: parallel +def t_prod(xs: list[int], c: int) -> int: + s: int = 1 + for i in range(len(xs)): + s *= xs[i] * c + return s + + +# typedpython: parallel +def t_min(xs: list[int]) -> int: + m = xs[0] + for i in range(len(xs)): + if xs[i] < m: + m = xs[i] + return m + + +# typedpython: parallel +def t_max(xs: list[float]) -> float: + m = xs[0] + for i in range(len(xs)): + if xs[i] > m: + m = xs[i] + return m + + +# typedpython: parallel +def t_and(xs: list[int]) -> int: + s: int = -1 + for i in range(len(xs)): + s &= xs[i] + return s + + +# typedpython: parallel +def t_or(xs: list[int]) -> int: + s: int = 0 + for i in range(len(xs)): + s |= xs[i] + return s + + +# typedpython: parallel +def t_xor(xs: list[int]) -> int: + s: int = 0 + for i in range(len(xs)): + s ^= xs[i] + return s +''' + +TSAN_PAR_BODY = """ +I = m.__typedpython_interpreted__ +n = 131072 +ints = [((k * 40503) % 65521) - 30000 for k in range(n)] +small = [1 + (k % 2) for k in range(n)] +flts = [((k * 40503) % 65521) / 13.0 for k in range(n)] +bad = [] +def work(seed): + def go(): + for _ in range(3): + xs = list(flts); m.t_scale(xs, 1.5, -0.5) + ys = list(flts); I["t_scale"](ys, 1.5, -0.5) + if xs != ys: bad.append("t_scale") + for nm, a in (("t_sum", ints), ("t_min", ints), ("t_max", flts), ("t_and", ints), + ("t_or", ints), ("t_xor", ints)): + if getattr(m, nm)(list(a)) != I[nm](list(a)): bad.append(nm) + if m.t_prod(list(small[:30]), 1) != I["t_prod"](list(small[:30]), 1): bad.append("t_prod") + return go +errors = run_threads([work(k) for k in range(4)]) +print("TSAN-RUN-DONE", json.dumps({"bad": bad, "errors": errors, "regions": m.__typedpython_parallel__()[0]})) +""" + + +def _tsan_launcher_par(tmp: Path) -> Path | None: + if sys.platform != "darwin" and not sys.platform.startswith("linux"): + return None + cc = (sysconfig.get_config_var("CC") or "cc").split() + lc, launcher = tmp / "tsan_launcher.c", tmp / "tsan_python" + lc.write_text("#include \nint main(int argc, char **argv) { return Py_BytesMain(argc, argv); }\n") + libdir = sysconfig.get_config_var("LIBDIR") + try: + link = subprocess.run(cc + ["-fsanitize=thread", "-g", f"-I{sysconfig.get_paths()['include']}", str(lc), + f"-L{libdir}", f"-lpython{sysconfig.get_config_var('LDVERSION')}", + f"-Wl,-rpath,{libdir}", "-o", str(launcher)], + capture_output=True, text=True) + except OSError: + return None + return launcher if link.returncode == 0 else None + + +def _par_access_is_ours(block: str, name: str) -> bool: + for part in block.split("\n\n")[:2]: + frames = [ln for ln in part.splitlines() if ln.strip().startswith("#")] + first = next((f for f in frames if "libtsan" not in f and "sanitizer_common" not in f), "") + if name in first: + return True + return False + + +def _par_tsan_reports(tmp: Path, launcher: Path, runtime_dir: Path | None, tag: str) -> tuple[int, bool, str]: + d = tmp / tag + d.mkdir(parents=True, exist_ok=True) + path = d / "tsanpar.py" + path.write_text(TSOURCE) + proved, _ = verify.verify(frontend.lower(path)) + flags = list(cbuild.DEFAULT_FLAGS) + ["-O1", "-g", "-fno-omit-frame-pointer", "-fsanitize=thread"] + so = cbuild.build(cgen.generate(proved, path), "tsanpar", d / "out", runtime_dir=runtime_dir, flags=flags) + loader = CHILD + "import time\n" + body = loader + textwrap.dedent(""" + def run_threads(targets): + errors = [] + def wrap(fn): + def go(): + try: + fn() + except BaseException as e: + errors.append(repr(e)) + return go + ts = [threading.Thread(target=wrap(t)) for t in targets] + for t in ts: t.start() + for t in ts: t.join() + return errors + """) + textwrap.dedent(TSAN_PAR_BODY) + env = dict(os.environ, PYTHONHOME=sys.base_prefix, TYPEDPYTHON_THREADS="4", + TSAN_OPTIONS="halt_on_error=0:report_signal_unsafe=0:history_size=4") + env.pop("PYTHON_GIL", None) + env.pop("DYLD_INSERT_LIBRARIES", None) + run = subprocess.run([str(launcher), "-c", body, str(so)], capture_output=True, text=True, env=env, timeout=900) + out = run.stdout + run.stderr + blocks = out.split("WARNING: ThreadSanitizer")[1:] + ours = [b for b in blocks if _par_access_is_ours(b, "tsanpar")] + first = ("WARNING: ThreadSanitizer" + ours[0])[:4000] if ours else "" + done = "TSAN-RUN-DONE" in out and '"bad": []' in out and '"errors": []' in out \ + and '"regions": 0}' not in out # regions really started workers (not a vacuous pass) + return len(ours), done, first + ("" if done else "\n--- output tail ---\n" + out[-1500:]) + + +@pytest.fixture(scope="module") +def par_tsan_launcher(tmp_path_factory): + d = tmp_path_factory.mktemp("partsan") + launcher = _tsan_launcher_par(d) + if launcher is None: + pytest.skip("cannot link a ThreadSanitizer launcher on this host") + return d, launcher + + +@pytest.mark.skipif(not FREE_THREADED, reason="needs a free-threaded CPython") +@pytest.mark.skipif((os.cpu_count() or 1) < 2, reason="needs a second core to start a region") +def test_tsan_reports_nothing_on_parallel_regions(par_tsan_launcher): + d, launcher = par_tsan_launcher + reports, done, first = _par_tsan_reports(d, launcher, None, "real") + assert done and reports == 0, f"{reports} ThreadSanitizer report(s) (done={done}); first:\n{first}" + + +def _racy_par_runtime(tmp: Path) -> Path: + """tp_runtime.h with the join removed: workers are detached, so the caller combines the partials + while the workers are still writing them.""" + h = (cbuild.RUNTIME_DIR / "tp_runtime.h").read_text() + old = " pthread_join(th[k], NULL);\n" + assert old in h + g = h.replace(old, " pthread_detach(th[k]);\n") + d = tmp / "rt" + d.mkdir(parents=True, exist_ok=True) + (d / "tp_runtime.h").write_text(g) + return d + + +@pytest.mark.race_mutant +@pytest.mark.skipif(not FREE_THREADED, reason="needs a free-threaded CPython") +@pytest.mark.skipif((os.cpu_count() or 1) < 2, reason="needs a second core to start a region") +def test_tsan_sees_a_region_combined_before_the_join(par_tsan_launcher): + """The check can fail: with the join replaced by a detach TSan reports the partials and the loop + output being read while workers still write them.""" + d, launcher = par_tsan_launcher + rt = _racy_par_runtime(d / "racy") + reports, _done, _first = _par_tsan_reports(d, launcher, rt, "mutant") + assert reports > 0