+
    Bi                     .   ^ RI t ^ RIt^ RIHt ^ RIHtHtHt ^ RIH	t	 ^ RI
Ht ]P                  ! R4      t]]! 4       ]P                  ! ]P                  R4       ! R R	]P                   4      4       4       4       t]R
8X  d   ]P&                  ! 4        R# R# )    N)support)cpython_onlyimport_helperrequires_specialization_ft)assert_python_ok)requires_working_threading_testinternalcapizonly in free-threaded buildsc                      a  ] tR t^t o ]R 4       t]R 4       tR t]P                  ! RRR7      R 4       t
R tR	 tR
tV tR# )	TLBCTestsc                N    \         P                  ! R 4      p\        RRRV4       R# )a  
        import dis
        import queue
        import threading

        from _testinternalcapi import get_tlbc

        def all_opnames(bc):
            return {i.opname for i in dis._get_instructions_bytes(bc)}

        def f(a, b, q=None):
            if q is not None:
                q.put(get_tlbc(f))
            return a + b

        for _ in range(100):
            # specialize
            f(1, 2)

        q = queue.Queue()
        t = threading.Thread(target=f, args=('a', 'b', q))
        t.start()
        t.join()

        assert "BINARY_OP_ADD_INT" in all_opnames(get_tlbc(f))
        assert "BINARY_OP_ADD_INT" not in all_opnames(q.get())
        -Xtlbc=1-cNtextwrapdedentr   selfcodes   & 6/usr/lib/python3.14/test/test_thread_local_bytecode.py.test_new_threads_start_with_unspecialized_code8TLBCTests.test_new_threads_start_with_unspecialized_code   s%       6 	xt4    c                N    \         P                  ! R 4      p\        RRRV4       R# )aJ  
        import dis
        import queue
        import threading

        from _testinternalcapi import get_tlbc

        def all_opnames(bc):
            return {i.opname for i in dis._get_instructions_bytes(bc)}

        def f(a, b):
            return a + b

        def g(a, b, q=None):
            for _ in range(100):
                f(a, b)
            if q is not None:
                q.put(get_tlbc(f))

        # specialize in main thread
        g(1, 2)

        # specialize in other thread
        q = queue.Queue()
        t = threading.Thread(target=g, args=('a', 'b', q))
        t.start()
        t.join()

        assert "BINARY_OP_ADD_INT" in all_opnames(get_tlbc(f))
        t_opnames = all_opnames(q.get())
        assert "BINARY_OP_ADD_INT" not in t_opnames
        assert "BINARY_OP_ADD_UNICODE" in t_opnames
        r   r   r   Nr   r   s   & r   %test_threads_specialize_independently/TLBCTests.test_threads_specialize_independently1   s&         B 	xt4r   c                N    \         P                  ! R 4      p\        RRRV4       R# )a  
        import queue
        import threading

        from _testinternalcapi import get_tlbc_id

        def f(a, b, q=None):
            if q is not None:
                q.put(get_tlbc_id(f))
            return a + b

        q = queue.Queue()
        tlbc_ids = []
        for _ in range(3):
            t = threading.Thread(target=f, args=('a', 'b', q))
            t.start()
            t.join()
            tlbc_ids.append(q.get())

        assert tlbc_ids[0] == tlbc_ids[1]
        assert tlbc_ids[1] == tlbc_ids[2]
        r   r   r   Nr   r   s   & r   2test_reuse_tlbc_across_threads_different_lifetimes<TLBCTests.test_reuse_tlbc_across_threads_different_lifetimesV   s%       , 	xt4r   z(gh-129752: data race on adaptive counterT)threadc                N    \         P                  ! R 4      p\        RRRV4       R# )a  
        import queue
        import threading

        from _testinternalcapi import get_tlbc_id

        def f(a, b, q=None):
            if q is not None:
                q.put(get_tlbc_id(f))
            return a + b

        q = queue.Queue()
        threads = []
        for _ in range(3):
            t = threading.Thread(target=f, args=('a', 'b', q))
            t.start()
            threads.append(t)

        tlbc_ids = []
        for t in threads:
            t.join()
            tlbc_ids.append(q.get())

        main_tlbc_id = get_tlbc_id(f)
        assert main_tlbc_id is not None
        assert tlbc_ids[0] == main_tlbc_id
        assert tlbc_ids[1] == main_tlbc_id
        assert tlbc_ids[2] == main_tlbc_id
        r   tlbc=0r   Nr   r   s   & r   test_no_copies_if_tlbc_disabled)TLBCTests.test_no_copies_if_tlbc_disabledo   s%       : 	xt4r   c                N    \         P                  ! R 4      p\        RRRV4       R# )a  
        import dis
        import queue
        import threading

        from _testinternalcapi import get_tlbc

        def all_opnames(f):
            bc = get_tlbc(f)
            return {i.opname for i in dis._get_instructions_bytes(bc)}

        def f(a, b):
            return a + b

        for _ in range(100):
            f(1, 2)

        assert "BINARY_OP_ADD_INT" not in all_opnames(f)
        r   r"   r   Nr   r   s   & r   'test_no_specialization_if_tlbc_disabled1TLBCTests.test_no_specialization_if_tlbc_disabled   s%       & 	xt4r   c                N    \         P                  ! R 4      p\        RRRV4       R# )a#  
        import queue
        import threading

        from _testinternalcapi import get_tlbc_id

        def g():
            try:
                yield
            except:
                yield get_tlbc_id(g)

        def f(q):
            gen = g()
            next(gen)
            q.put(gen.throw(ValueError))

        q = queue.Queue()
        t = threading.Thread(target=f, args=(q,))
        t.start()
        t.join()

        gen = g()
        next(gen)
        main_id = gen.throw(ValueError)
        assert main_id != q.get()
        r   r   r   Nr   r   s   & r   test_generator_throwTLBCTests.test_generator_throw   s%       6 	xt4r    N)__name__
__module____qualname____firstlineno__r   r   r   r   r   skip_if_sanitizerr#   r&   r)   __static_attributes____classdictcell__)__classdict__s   @r   r   r      sg       5  5<  "5  "5H52 IRVW5 X5@5,5 5r   r   __main__)r   unittesttestr   test.supportr   r   r   test.support.script_helperr   test.support.threading_helperr   import_moduler	   
skipUnlessPy_GIL_DISABLEDTestCaser   r,   mainr+   r   r   <module>r?      s       P P 7 D "//0CD  	W,,.LMq5!! q5 N  q5h zMMO r   