# Regression test for the TYPE_LOCK / stop-the-world deadlock.
#
# In the proposed fix these methods are added to class TestType in
# Lib/test/test_free_threading/test_type.py (see fix-<branch>.diff, which also
# contains them).  This standalone copy runs on its own:
#     ./python -m unittest -v test_type_stw_deadlock
# Before the fix, each test hangs (a deadlock; the regrtest --timeout or the
# caller's timeout reports it).  After the fix each finishes in a few seconds.
# On the default (GIL) build the tests pass trivially.
import threading
import unittest
from threading import Thread
from unittest import TestCase

from test.support import threading_helper


@threading_helper.requires_working_threading()
class TestType(TestCase):
    def check_stop_the_world_under_type_lock(self, setter):
        # gh-NNNNNN: Python code that runs with the type lock held and stops
        # the world could deadlock with a thread that holds the type lock
        # pinned while it waits for the stop-the-world mutex (__bases__
        # assignment).  The first thread released the type lock while it was
        # blocked on the stop-the-world mutex, was handed the mutex, and then
        # waited for the type lock while owning the mutex.
        done = threading.Event()

        def run_setter():
            barrier.wait()
            setter()

        def basesetter():
            class A:
                pass
            class B:
                pass
            class C(A):
                pass
            barrier.wait()
            while not done.is_set():
                C.__bases__ = (B,)
                C.__bases__ = (A,)

        def stopper():
            def g():
                pass
            barrier.wait()
            while not done.is_set():
                g.__qualname__ = "g"  # stops the world

        # The stopper thread keeps the stop-the-world mutex contended, which
        # makes the setters block on it.
        setters = [Thread(target=run_setter) for _ in range(2)]
        others = [Thread(target=basesetter), Thread(target=stopper)]
        barrier = threading.Barrier(len(setters) + len(others))
        with threading_helper.start_threads(setters + others):
            for t in setters:
                t.join()
            done.set()

    def test_stop_the_world_in_mro_deadlock(self):
        # type_ready() calls a metaclass mro() with the type lock held.
        N = 20
        def f():
            pass

        class Meta(type):
            def mro(cls):
                f.__qualname__ = "f"  # stops the world
                return type.mro(cls)

        def setter():
            for _ in range(N):
                Meta("K", (), {})

        self.check_stop_the_world_under_type_lock(setter)

    def test_stop_the_world_in_setattr_key_eq_deadlock(self):
        # type_setattro() compares the attribute name with a str subclass key
        # of the type's __dict__ (calling its __eq__) with the type lock held.
        N = 20
        def f():
            pass

        class S(str):
            __hash__ = str.__hash__
            def __eq__(self, other):
                f.__qualname__ = "f"  # stops the world
                return str.__eq__(self, other)

        C = type("C", (), {S("x"): 0})

        def setter():
            for i in range(N):
                C.x = i

        self.check_stop_the_world_under_type_lock(setter)


if __name__ == "__main__":
    unittest.main()