# Candidate common-code trigger for gh-158195 (no metaclass mro(), no str-subclass keys). # reprer threads : repr(K.__dict__) -- dict_repr holds K's dict mutex (unpinned # critical section) while calling the values' Python __repr__, # which allocate; an automatic GC can then stop the world. # patcher threads : mock.patch.object() a dunder on K: # apply_type_slot_updates pins TYPE_LOCK and K's dict mutex, then # stops the world. # contender threads: functools.wraps (plain stop-the-world). # usage: python repro_dict_repr.py [gc|stw|plain] [seconds] (default: gc 5); prints "done" on success, # faulthandler timeout (exit 1) on a hang. import faulthandler, functools, sys, threading, time from unittest import mock if sys._is_gil_enabled() is not False: sys.exit("needs a free-threaded build with the GIL disabled") MODE = sys.argv[1] if len(sys.argv) > 1 else "gc" # gc (default) | stw | plain SECS = float(sys.argv[2]) if len(sys.argv) > 2 else 5.0 faulthandler.dump_traceback_later(SECS + 15, exit=True) def f(): pass class Setting: # an ordinary object with a Python __repr__ def __init__(self, name): self.name = name def __repr__(self): parts = [[p] for p in self.name] # allocate container objects if MODE == "gc": # cyclic garbage: automatic GC fires often for _ in range(50): a = []; a.append(a) elif MODE == "stw": # explicit stop-the-world (control) f.__qualname__ = "f" return f"Setting({self.name!r}, {len(parts)})" class K: a = Setting("alpha"); b = Setting("beta"); c = Setting("gamma") def method(self): pass stop = threading.Event(); errors = [] def guarded(fn): def run(): try: fn() except BaseException as e: errors.append(e); raise return run def reprer(): while not stop.is_set(): repr(K.__dict__) def make_patcher(name): # unittest.mock.patch.object() a dunder on K: setting and then removing it # changes the type slot, so apply_type_slot_updates() pins TYPE_LOCK and # K's dict mutex and stops the world. One dunder per thread. def fn(self, other): return NotImplemented def patcher(): while not stop.is_set(): with mock.patch.object(K, name, fn, create=True): pass patcher.__name__ = f"patcher_{name}" return patcher def contender(): def g(): pass while not stop.is_set(): functools.wraps(g)(lambda: None) ts = ([threading.Thread(target=guarded(reprer)) for _ in range(2)] + [threading.Thread(target=guarded(make_patcher(n))) for n in ('__lt__', '__gt__')] + [threading.Thread(target=guarded(contender)) for _ in range(2)]) for t in ts: t.start() time.sleep(SECS); stop.set() for t in ts: t.join() if errors: sys.exit(f"FAIL: {errors[0]!r}") faulthandler.cancel_dump_traceback_later() print("done", MODE)