Concurrent mutation of cached slot names can crash default object state collection #160440

#!/usr/bin/env python3
"""Exercise concurrent mutation of CPython's cached default-state slot names.

Run with a free-threaded CPython interpreter and the GIL disabled. A clean
run is finite stress evidence; it does not prove that the race is absent.
"""

import argparse
import faulthandler
import json
import sys
import threading
import time


class Name(str):
    """A fresh, non-interned name whose lifetime depends on owned references."""


class Sample:
    pass


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--mode", choices=("resize", "replace"), default="resize")
    parser.add_argument("--operation", choices=("state", "reduce"), default="reduce")
    parser.add_argument("--protocol", choices=range(2, 6), type=int, default=5)
    parser.add_argument("--calls", type=int, default=100_000)
    args = parser.parse_args()
    if args.calls <= 0:
        parser.error("--calls must be positive")
    gil_enabled = getattr(sys, "_is_gil_enabled", lambda: True)()
    if gil_enabled:
        parser.error("requires a free-threaded CPython interpreter with the GIL disabled")

    faulthandler.enable()
    obj = Sample()
    obj.value = 42
    obj.other = 99
    Sample.__slotnames__ = [Name("value")]
    names = Sample.__slotnames__
    stopped = threading.Event()
    started = threading.Event()
    mutations = 0
    writer_errors = []

    def writer():
        nonlocal mutations
        try:
            while not stopped.is_set():
                names[:] = [Name("value")]
                if args.mode == "resize":
                    names.clear()
                else:
                    names[:] = [Name("other")]
                mutations += 1
                started.set()
        except BaseException as exc:
            writer_errors.append(exc)
            started.set()

    print(json.dumps({"event": "start", "python_version": list(sys.version_info[:3]),
                      "implementation": sys.implementation.name, "gil_enabled": gil_enabled,
                      "arguments": vars(args)}), flush=True)
    thread = threading.Thread(target=writer, daemon=True)
    thread.start()
    successes = size_change_errors = 0
    begin = time.monotonic()
    try:
        if not started.wait(10):
            raise RuntimeError("writer did not start within 10 seconds")
        initial_mutations = mutations
        for _ in range(args.calls):
            try:
                if args.operation == "state":
                    state = obj.__getstate__()
                else:
                    reduced = obj.__reduce_ex__(args.protocol)
                    if reduced[1] != (Sample,) or reduced[3:] != (None, None):
                        raise AssertionError(f"unexpected reduction: {reduced!r}")
                    state = reduced[2]
            except RuntimeError as exc:
                if (args.mode != "resize" or
                    str(exc) != "__slotnames__ changed size during iteration"):
                    raise
                size_change_errors += 1
                continue

            if type(state) is tuple:
                if len(state) != 2:
                    raise AssertionError(f"unexpected state tuple: {state!r}")
                dictionary, slots = state
                if slots not in ({"value": 42}, {"other": 99}):
                    raise AssertionError(f"unexpected slot state: {slots!r}")
            else:
                if args.mode == "replace":
                    raise AssertionError(f"missing slot state: {state!r}")
                dictionary = state
            if dictionary is not obj.__dict__ or dictionary != {"value": 42, "other": 99}:
                raise AssertionError(f"unexpected dictionary state: {dictionary!r}")
            successes += 1
        concurrent_mutations = mutations - initial_mutations
    finally:
        stopped.set()
        thread.join(10)
    if thread.is_alive():
        raise RuntimeError("writer did not stop within 10 seconds")
    if writer_errors:
        raise writer_errors[0]
    if concurrent_mutations <= 0:
        raise RuntimeError("inconclusive: no writer progress observed during reader calls")
    print(json.dumps({"event": "complete", "successes": successes,
                      "size_change_errors": size_change_errors,
                      "concurrent_mutations": concurrent_mutations,
                      "elapsed_seconds": time.monotonic() - begin}), flush=True)


if __name__ == "__main__":
    main()