Upgrade threading to 3.13.11; sys.setprofile & impl more threading (#6691)

* Upgrade threading,test_threading from 3.13.11

* increase ci limit to 45mins

* impl more threading

* no static for asyncgen

.github/workflows/ci.yaml

@@ -106,9 +106,7 @@ env:
     test_weakref
     test_yield_from
   ENV_POLLUTING_TESTS_COMMON: >-
-    test_threading
   ENV_POLLUTING_TESTS_LINUX: >-
-    test.test_concurrent_futures.test_thread_pool
     test.test_multiprocessing_fork.test_processes
     test.test_multiprocessing_fork.test_threads
     test.test_multiprocessing_forkserver.test_processes
@@ -116,7 +114,6 @@ env:
     test.test_multiprocessing_spawn.test_processes
     test.test_multiprocessing_spawn.test_threads
   ENV_POLLUTING_TESTS_MACOS: >-
-    test.test_concurrent_futures.test_thread_pool
     test.test_multiprocessing_forkserver.test_processes
     test.test_multiprocessing_forkserver.test_threads
     test.test_multiprocessing_spawn.test_processes
@@ -297,21 +294,21 @@ jobs:
         env:
           RUSTPYTHON_SKIP_ENV_POLLUTERS: true
         run: target/release/rustpython -m test -j 1 -u all --slowest --fail-env-changed --timeout 600 -v ${{ env.PLATFORM_INDEPENDENT_TESTS }}
-        timeout-minutes: 35
+        timeout-minutes: 45
 
       - if: runner.os == 'Linux'
         name: run cpython platform-dependent tests (Linux)
         env:
           RUSTPYTHON_SKIP_ENV_POLLUTERS: true
         run: target/release/rustpython -m test -j 1 -u all --slowest --fail-env-changed --timeout 600 -v -x ${{ env.PLATFORM_INDEPENDENT_TESTS }}
-        timeout-minutes: 35
+        timeout-minutes: 45
 
       - if: runner.os == 'macOS'
         name: run cpython platform-dependent tests (MacOS)
         env:
           RUSTPYTHON_SKIP_ENV_POLLUTERS: true
         run: target/release/rustpython -m test -j 1 --slowest --fail-env-changed --timeout 600 -v -x ${{ env.PLATFORM_INDEPENDENT_TESTS }}
-        timeout-minutes: 35
+        timeout-minutes: 45
 
       - if: runner.os == 'Windows'
         name: run cpython platform-dependent tests (windows partial - fixme)

Lib/_dummy_thread.py

@@ -11,15 +11,35 @@ Suggested usage is::
         import _dummy_thread as _thread
 
 """
+
 # Exports only things specified by thread documentation;
 # skipping obsolete synonyms allocate(), start_new(), exit_thread().
-__all__ = ['error', 'start_new_thread', 'exit', 'get_ident', 'allocate_lock',
-           'interrupt_main', 'LockType', 'RLock',
-           '_count']
+__all__ = [
+    "error",
+    "start_new_thread",
+    "exit",
+    "get_ident",
+    "allocate_lock",
+    "interrupt_main",
+    "LockType",
+    "RLock",
+    "_count",
+    "start_joinable_thread",
+    "daemon_threads_allowed",
+    "_shutdown",
+    "_make_thread_handle",
+    "_ThreadHandle",
+    "_get_main_thread_ident",
+    "_is_main_interpreter",
+    "_local",
+]
 
 # A dummy value
 TIMEOUT_MAX = 2**31
 
+# Main thread ident for dummy implementation
+_MAIN_THREAD_IDENT = -1
+
 # NOTE: this module can be imported early in the extension building process,
 # and so top level imports of other modules should be avoided.  Instead, all
 # imports are done when needed on a function-by-function basis.  Since threads
@@ -27,6 +47,7 @@ TIMEOUT_MAX = 2**31
 
 error = RuntimeError
 
+
 def start_new_thread(function, args, kwargs={}):
     """Dummy implementation of _thread.start_new_thread().
 
@@ -52,6 +73,7 @@ def start_new_thread(function, args, kwargs={}):
         pass
     except:
         import traceback
+
         traceback.print_exc()
     _main = True
     global _interrupt
@@ -59,10 +81,58 @@ def start_new_thread(function, args, kwargs={}):
         _interrupt = False
         raise KeyboardInterrupt
 
+
+def start_joinable_thread(function, handle=None, daemon=True):
+    """Dummy implementation of _thread.start_joinable_thread().
+
+    In dummy thread, we just run the function synchronously.
+    """
+    if handle is None:
+        handle = _ThreadHandle()
+    try:
+        function()
+    except SystemExit:
+        pass
+    except:
+        import traceback
+
+        traceback.print_exc()
+    handle._set_done()
+    return handle
+
+
+def daemon_threads_allowed():
+    """Dummy implementation of _thread.daemon_threads_allowed()."""
+    return True
+
+
+def _shutdown():
+    """Dummy implementation of _thread._shutdown()."""
+    pass
+
+
+def _make_thread_handle(ident):
+    """Dummy implementation of _thread._make_thread_handle()."""
+    handle = _ThreadHandle()
+    handle._ident = ident
+    return handle
+
+
+def _get_main_thread_ident():
+    """Dummy implementation of _thread._get_main_thread_ident()."""
+    return _MAIN_THREAD_IDENT
+
+
+def _is_main_interpreter():
+    """Dummy implementation of _thread._is_main_interpreter()."""
+    return True
+
+
 def exit():
     """Dummy implementation of _thread.exit()."""
     raise SystemExit
 
+
 def get_ident():
     """Dummy implementation of _thread.get_ident().
 
@@ -70,26 +140,31 @@ def get_ident():
     available, it is safe to assume that the current process is the
     only thread.  Thus a constant can be safely returned.
     """
-    return -1
+    return _MAIN_THREAD_IDENT
+
 
 def allocate_lock():
     """Dummy implementation of _thread.allocate_lock()."""
     return LockType()
 
+
 def stack_size(size=None):
     """Dummy implementation of _thread.stack_size()."""
     if size is not None:
         raise error("setting thread stack size not supported")
     return 0
 
+
 def _set_sentinel():
     """Dummy implementation of _thread._set_sentinel()."""
     return LockType()
 
+
 def _count():
     """Dummy implementation of _thread._count()."""
     return 0
 
+
 class LockType(object):
     """Class implementing dummy implementation of _thread.LockType.
 
@@ -125,6 +200,7 @@ class LockType(object):
             else:
                 if timeout > 0:
                     import time
+
                     time.sleep(timeout)
                 return False
 
@@ -153,14 +229,41 @@ class LockType(object):
             "locked" if self.locked_status else "unlocked",
             self.__class__.__module__,
             self.__class__.__qualname__,
-            hex(id(self))
+            hex(id(self)),
         )
 
+
+class _ThreadHandle:
+    """Dummy implementation of _thread._ThreadHandle."""
+
+    def __init__(self):
+        self._ident = _MAIN_THREAD_IDENT
+        self._done = False
+
+    @property
+    def ident(self):
+        return self._ident
+
+    def _set_done(self):
+        self._done = True
+
+    def is_done(self):
+        return self._done
+
+    def join(self, timeout=None):
+        # In dummy thread, thread is always done
+        return
+
+    def __repr__(self):
+        return f"<_ThreadHandle ident={self._ident}>"
+
+
 # Used to signal that interrupt_main was called in a "thread"
 _interrupt = False
 # True when not executing in a "thread"
 _main = True
 
+
 def interrupt_main():
     """Set _interrupt flag to True to have start_new_thread raise
     KeyboardInterrupt upon exiting."""
@@ -170,6 +273,7 @@ def interrupt_main():
         global _interrupt
         _interrupt = True
 
+
 class RLock:
     def __init__(self):
         self.locked_count = 0
@@ -190,7 +294,7 @@ class RLock:
         return True
 
     def locked(self):
-        return self.locked_status != 0
+        return self.locked_count != 0
 
     def __repr__(self):
         return "<%s %s.%s object owner=%s count=%s at %s>" % (
@@ -199,5 +303,36 @@ class RLock:
             self.__class__.__qualname__,
             get_ident() if self.locked_count else 0,
             self.locked_count,
-            hex(id(self))
+            hex(id(self)),
         )
+
+
+class _local:
+    """Dummy implementation of _thread._local (thread-local storage)."""
+
+    def __init__(self):
+        object.__setattr__(self, "_local__impl", {})
+
+    def __getattribute__(self, name):
+        if name.startswith("_local__"):
+            return object.__getattribute__(self, name)
+        impl = object.__getattribute__(self, "_local__impl")
+        try:
+            return impl[name]
+        except KeyError:
+            raise AttributeError(name)
+
+    def __setattr__(self, name, value):
+        if name.startswith("_local__"):
+            return object.__setattr__(self, name, value)
+        impl = object.__getattribute__(self, "_local__impl")
+        impl[name] = value
+
+    def __delattr__(self, name):
+        if name.startswith("_local__"):
+            return object.__delattr__(self, name)
+        impl = object.__getattribute__(self, "_local__impl")
+        try:
+            del impl[name]
+        except KeyError:
+            raise AttributeError(name)

Lib/dummy_threading.py

@@ -6,6 +6,7 @@ directly imported it would have made all subsequent imports succeed
 regardless of whether ``_thread`` was available which is not desired.
 
 """
+
 from sys import modules as sys_modules
 
 import _dummy_thread
@@ -19,35 +20,38 @@ try:
     # Could have checked if ``_thread`` was not in sys.modules and gone
     # a different route, but decided to mirror technique used with
     # ``threading`` below.
-    if '_thread' in sys_modules:
-        held_thread = sys_modules['_thread']
+    if "_thread" in sys_modules:
+        held_thread = sys_modules["_thread"]
         holding_thread = True
     # Must have some module named ``_thread`` that implements its API
     # in order to initially import ``threading``.
-    sys_modules['_thread'] = sys_modules['_dummy_thread']
+    sys_modules["_thread"] = sys_modules["_dummy_thread"]
 
-    if 'threading' in sys_modules:
+    if "threading" in sys_modules:
         # If ``threading`` is already imported, might as well prevent
         # trying to import it more than needed by saving it if it is
         # already imported before deleting it.
-        held_threading = sys_modules['threading']
+        held_threading = sys_modules["threading"]
         holding_threading = True
-        del sys_modules['threading']
+        del sys_modules["threading"]
 
-    if '_threading_local' in sys_modules:
+    if "_threading_local" in sys_modules:
         # If ``_threading_local`` is already imported, might as well prevent
         # trying to import it more than needed by saving it if it is
         # already imported before deleting it.
-        held__threading_local = sys_modules['_threading_local']
+        held__threading_local = sys_modules["_threading_local"]
         holding__threading_local = True
-        del sys_modules['_threading_local']
+        del sys_modules["_threading_local"]
 
     import threading
+
     # Need a copy of the code kept somewhere...
-    sys_modules['_dummy_threading'] = sys_modules['threading']
-    del sys_modules['threading']
-    sys_modules['_dummy__threading_local'] = sys_modules['_threading_local']
-    del sys_modules['_threading_local']
+    sys_modules["_dummy_threading"] = sys_modules["threading"]
+    del sys_modules["threading"]
+    # _threading_local may not be imported if _thread._local is available
+    if "_threading_local" in sys_modules:
+        sys_modules["_dummy__threading_local"] = sys_modules["_threading_local"]
+        del sys_modules["_threading_local"]
     from _dummy_threading import *
     from _dummy_threading import __all__
 
@@ -55,23 +59,23 @@ finally:
     # Put back ``threading`` if we overwrote earlier
 
     if holding_threading:
-        sys_modules['threading'] = held_threading
+        sys_modules["threading"] = held_threading
         del held_threading
     del holding_threading
 
     # Put back ``_threading_local`` if we overwrote earlier
 
     if holding__threading_local:
-        sys_modules['_threading_local'] = held__threading_local
+        sys_modules["_threading_local"] = held__threading_local
         del held__threading_local
     del holding__threading_local
 
     # Put back ``thread`` if we overwrote, else del the entry we made
     if holding_thread:
-        sys_modules['_thread'] = held_thread
+        sys_modules["_thread"] = held_thread
         del held_thread
     else:
-        del sys_modules['_thread']
+        del sys_modules["_thread"]
     del holding_thread
 
     del _dummy_thread

Lib/test/lock_tests.py

@@ -2,6 +2,7 @@
 Various tests for synchronization primitives.
 """
 
+import gc
 import sys
 import time
 from _thread import start_new_thread, TIMEOUT_MAX
@@ -13,54 +14,79 @@ from test import support
 from test.support import threading_helper
 
 
-def _wait():
-    # A crude wait/yield function not relying on synchronization primitives.
-    time.sleep(0.01)
+requires_fork = unittest.skipUnless(support.has_fork_support,
+                                    "platform doesn't support fork "
+                                     "(no _at_fork_reinit method)")
+
+
+def wait_threads_blocked(nthread):
+    # Arbitrary sleep to wait until N threads are blocked,
+    # like waiting for a lock.
+    time.sleep(0.010 * nthread)
+
 
 class Bunch(object):
     """
     A bunch of threads.
     """
-    def __init__(self, f, n, wait_before_exit=False):
+    def __init__(self, func, nthread, wait_before_exit=False):
         """
-        Construct a bunch of `n` threads running the same function `f`.
+        Construct a bunch of `nthread` threads running the same function `func`.
         If `wait_before_exit` is True, the threads won't terminate until
         do_finish() is called.
         """
-        self.f = f
-        self.n = n
+        self.func = func
+        self.nthread = nthread
         self.started = []
         self.finished = []
+        self.exceptions = []
         self._can_exit = not wait_before_exit
-        self.wait_thread = threading_helper.wait_threads_exit()
-        self.wait_thread.__enter__()
+        self._wait_thread = None
 
-        def task():
-            tid = threading.get_ident()
-            self.started.append(tid)
-            try:
-                f()
-            finally:
-                self.finished.append(tid)
-                while not self._can_exit:
-                    _wait()
+    def task(self):
+        tid = threading.get_ident()
+        self.started.append(tid)
+        try:
+            self.func()
+        except BaseException as exc:
+            self.exceptions.append(exc)
+        finally:
+            self.finished.append(tid)
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                if self._can_exit:
+                    break
+
+    def __enter__(self):
+        self._wait_thread = threading_helper.wait_threads_exit(support.SHORT_TIMEOUT)
+        self._wait_thread.__enter__()
 
         try:
-            for i in range(n):
-                start_new_thread(task, ())
+            for _ in range(self.nthread):
+                start_new_thread(self.task, ())
         except:
             self._can_exit = True
             raise
 
-    def wait_for_started(self):
-        while len(self.started) < self.n:
-            _wait()
+        for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+            if len(self.started) >= self.nthread:
+                break
 
-    def wait_for_finished(self):
-        while len(self.finished) < self.n:
-            _wait()
-        # Wait for threads exit
-        self.wait_thread.__exit__(None, None, None)
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+            if len(self.finished) >= self.nthread:
+                break
+
+        # Wait until threads completely exit according to _thread._count()
+        self._wait_thread.__exit__(None, None, None)
+
+        # Break reference cycle
+        exceptions = self.exceptions
+        self.exceptions = None
+        if exceptions:
+            raise ExceptionGroup(f"{self.func} threads raised exceptions",
+                                 exceptions)
 
     def do_finish(self):
         self._can_exit = True
@@ -88,6 +114,12 @@ class BaseLockTests(BaseTestCase):
     Tests for both recursive and non-recursive locks.
     """
 
+    def wait_phase(self, phase, expected):
+        for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+            if len(phase) >= expected:
+                break
+        self.assertEqual(len(phase), expected)
+
     def test_constructor(self):
         lock = self.locktype()
         del lock
@@ -125,44 +157,60 @@ class BaseLockTests(BaseTestCase):
         result = []
         def f():
             result.append(lock.acquire(False))
-        Bunch(f, 1).wait_for_finished()
+        with Bunch(f, 1):
+            pass
         self.assertFalse(result[0])
         lock.release()
 
-    @unittest.skip("TODO: RUSTPYTHON, sometimes hangs")
+    @unittest.skip('TODO: RUSTPYTHON; sometimes hangs')
     def test_acquire_contended(self):
         lock = self.locktype()
         lock.acquire()
-        N = 5
         def f():
             lock.acquire()
             lock.release()
 
-        b = Bunch(f, N)
-        b.wait_for_started()
-        _wait()
-        self.assertEqual(len(b.finished), 0)
-        lock.release()
-        b.wait_for_finished()
-        self.assertEqual(len(b.finished), N)
+        N = 5
+        with Bunch(f, N) as bunch:
+            # Threads block on lock.acquire()
+            wait_threads_blocked(N)
+            self.assertEqual(len(bunch.finished), 0)
+
+            # Threads unblocked
+            lock.release()
+
+        self.assertEqual(len(bunch.finished), N)
 
     def test_with(self):
         lock = self.locktype()
         def f():
             lock.acquire()
             lock.release()
-        def _with(err=None):
+
+        def with_lock(err=None):
             with lock:
                 if err is not None:
                     raise err
-        _with()
-        # Check the lock is unacquired
-        Bunch(f, 1).wait_for_finished()
-        self.assertRaises(TypeError, _with, TypeError)
-        # Check the lock is unacquired
-        Bunch(f, 1).wait_for_finished()
 
-    @unittest.skip("TODO: RUSTPYTHON, sometimes hangs")
+        # Acquire the lock, do nothing, with releases the lock
+        with lock:
+            pass
+
+        # Check that the lock is unacquired
+        with Bunch(f, 1):
+            pass
+
+        # Acquire the lock, raise an exception, with releases the lock
+        with self.assertRaises(TypeError):
+            with lock:
+                raise TypeError
+
+        # Check that the lock is unacquired even if after an exception
+        # was raised in the previous "with lock:" block
+        with Bunch(f, 1):
+            pass
+
+    @unittest.skip('TODO: RUSTPYTHON; sometimes hangs')
     def test_thread_leak(self):
         # The lock shouldn't leak a Thread instance when used from a foreign
         # (non-threading) thread.
@@ -170,22 +218,16 @@ class BaseLockTests(BaseTestCase):
         def f():
             lock.acquire()
             lock.release()
-        n = len(threading.enumerate())
+
         # We run many threads in the hope that existing threads ids won't
         # be recycled.
-        Bunch(f, 15).wait_for_finished()
-        if len(threading.enumerate()) != n:
-            # There is a small window during which a Thread instance's
-            # target function has finished running, but the Thread is still
-            # alive and registered.  Avoid spurious failures by waiting a
-            # bit more (seen on a buildbot).
-            time.sleep(0.4)
-            self.assertEqual(n, len(threading.enumerate()))
+        with Bunch(f, 15):
+            pass
 
     def test_timeout(self):
         lock = self.locktype()
         # Can't set timeout if not blocking
-        self.assertRaises(ValueError, lock.acquire, 0, 1)
+        self.assertRaises(ValueError, lock.acquire, False, 1)
         # Invalid timeout values
         self.assertRaises(ValueError, lock.acquire, timeout=-100)
         self.assertRaises(OverflowError, lock.acquire, timeout=1e100)
@@ -204,7 +246,8 @@ class BaseLockTests(BaseTestCase):
             results.append(lock.acquire(timeout=0.5))
             t2 = time.monotonic()
             results.append(t2 - t1)
-        Bunch(f, 1).wait_for_finished()
+        with Bunch(f, 1):
+            pass
         self.assertFalse(results[0])
         self.assertTimeout(results[1], 0.5)
 
@@ -217,6 +260,7 @@ class BaseLockTests(BaseTestCase):
         lock = self.locktype()
         ref = weakref.ref(lock)
         del lock
+        gc.collect()  # For PyPy or other GCs.
         self.assertIsNone(ref())
 
 
@@ -237,15 +281,13 @@ class LockTests(BaseLockTests):
             phase.append(None)
 
         with threading_helper.wait_threads_exit():
+            # Thread blocked on lock.acquire()
             start_new_thread(f, ())
-            while len(phase) == 0:
-                _wait()
-            _wait()
-            self.assertEqual(len(phase), 1)
+            self.wait_phase(phase, 1)
+
+            # Thread unblocked
             lock.release()
-            while len(phase) == 1:
-                _wait()
-            self.assertEqual(len(phase), 2)
+            self.wait_phase(phase, 2)
 
     def test_different_thread(self):
         # Lock can be released from a different thread.
@@ -253,8 +295,8 @@ class LockTests(BaseLockTests):
         lock.acquire()
         def f():
             lock.release()
-        b = Bunch(f, 1)
-        b.wait_for_finished()
+        with Bunch(f, 1):
+            pass
         lock.acquire()
         lock.release()
 
@@ -268,6 +310,25 @@ class LockTests(BaseLockTests):
         self.assertFalse(lock.locked())
         self.assertTrue(lock.acquire(blocking=False))
 
+    @requires_fork
+    def test_at_fork_reinit(self):
+        def use_lock(lock):
+            # make sure that the lock still works normally
+            # after _at_fork_reinit()
+            lock.acquire()
+            lock.release()
+
+        # unlocked
+        lock = self.locktype()
+        lock._at_fork_reinit()
+        use_lock(lock)
+
+        # locked: _at_fork_reinit() resets the lock to the unlocked state
+        lock2 = self.locktype()
+        lock2.acquire()
+        lock2._at_fork_reinit()
+        use_lock(lock2)
+
 
 class RLockTests(BaseLockTests):
     """
@@ -306,17 +367,52 @@ class RLockTests(BaseLockTests):
         lock.release()
         self.assertRaises(RuntimeError, lock._release_save)
 
+    def test_recursion_count(self):
+        lock = self.locktype()
+        self.assertEqual(0, lock._recursion_count())
+        lock.acquire()
+        self.assertEqual(1, lock._recursion_count())
+        lock.acquire()
+        lock.acquire()
+        self.assertEqual(3, lock._recursion_count())
+        lock.release()
+        self.assertEqual(2, lock._recursion_count())
+        lock.release()
+        lock.release()
+        self.assertEqual(0, lock._recursion_count())
+
+        phase = []
+
+        def f():
+            lock.acquire()
+            phase.append(None)
+
+            self.wait_phase(phase, 2)
+            lock.release()
+            phase.append(None)
+
+        with threading_helper.wait_threads_exit():
+            # Thread blocked on lock.acquire()
+            start_new_thread(f, ())
+            self.wait_phase(phase, 1)
+            self.assertEqual(0, lock._recursion_count())
+
+            # Thread unblocked
+            phase.append(None)
+            self.wait_phase(phase, 3)
+            self.assertEqual(0, lock._recursion_count())
+
     def test_different_thread(self):
         # Cannot release from a different thread
         lock = self.locktype()
         def f():
             lock.acquire()
-        b = Bunch(f, 1, True)
-        try:
-            self.assertRaises(RuntimeError, lock.release)
-        finally:
-            b.do_finish()
-        b.wait_for_finished()
+
+        with Bunch(f, 1, True) as bunch:
+            try:
+                self.assertRaises(RuntimeError, lock.release)
+            finally:
+                bunch.do_finish()
 
     def test__is_owned(self):
         lock = self.locktype()
@@ -328,7 +424,8 @@ class RLockTests(BaseLockTests):
         result = []
         def f():
             result.append(lock._is_owned())
-        Bunch(f, 1).wait_for_finished()
+        with Bunch(f, 1):
+            pass
         self.assertFalse(result[0])
         lock.release()
         self.assertTrue(lock._is_owned())
@@ -361,12 +458,15 @@ class EventTests(BaseTestCase):
         def f():
             results1.append(evt.wait())
             results2.append(evt.wait())
-        b = Bunch(f, N)
-        b.wait_for_started()
-        _wait()
-        self.assertEqual(len(results1), 0)
-        evt.set()
-        b.wait_for_finished()
+
+        with Bunch(f, N):
+            # Threads blocked on first evt.wait()
+            wait_threads_blocked(N)
+            self.assertEqual(len(results1), 0)
+
+            # Threads unblocked
+            evt.set()
+
         self.assertEqual(results1, [True] * N)
         self.assertEqual(results2, [True] * N)
 
@@ -389,46 +489,61 @@ class EventTests(BaseTestCase):
             r = evt.wait(0.5)
             t2 = time.monotonic()
             results2.append((r, t2 - t1))
-        Bunch(f, N).wait_for_finished()
+
+        with Bunch(f, N):
+            pass
+
         self.assertEqual(results1, [False] * N)
         for r, dt in results2:
             self.assertFalse(r)
             self.assertTimeout(dt, 0.5)
+
         # The event is set
         results1 = []
         results2 = []
         evt.set()
-        Bunch(f, N).wait_for_finished()
+        with Bunch(f, N):
+            pass
+
         self.assertEqual(results1, [True] * N)
         for r, dt in results2:
             self.assertTrue(r)
 
     def test_set_and_clear(self):
-        # Issue #13502: check that wait() returns true even when the event is
+        # gh-57711: check that wait() returns true even when the event is
         # cleared before the waiting thread is woken up.
-        evt = self.eventtype()
+        event = self.eventtype()
         results = []
-        timeout = 0.250
-        N = 5
         def f():
-            results.append(evt.wait(timeout * 4))
-        b = Bunch(f, N)
-        b.wait_for_started()
-        time.sleep(timeout)
-        evt.set()
-        evt.clear()
-        b.wait_for_finished()
+            results.append(event.wait(support.LONG_TIMEOUT))
+
+        N = 5
+        with Bunch(f, N):
+            # Threads blocked on event.wait()
+            wait_threads_blocked(N)
+
+            # Threads unblocked
+            event.set()
+            event.clear()
+
         self.assertEqual(results, [True] * N)
 
-    def test_reset_internal_locks(self):
+    @requires_fork
+    def test_at_fork_reinit(self):
         # ensure that condition is still using a Lock after reset
         evt = self.eventtype()
         with evt._cond:
             self.assertFalse(evt._cond.acquire(False))
-        evt._reset_internal_locks()
+        evt._at_fork_reinit()
         with evt._cond:
             self.assertFalse(evt._cond.acquire(False))
 
+    def test_repr(self):
+        evt = self.eventtype()
+        self.assertRegex(repr(evt), r"<\w+\.Event at .*: unset>")
+        evt.set()
+        self.assertRegex(repr(evt), r"<\w+\.Event at .*: set>")
+
 
 class ConditionTests(BaseTestCase):
     """
@@ -466,15 +581,14 @@ class ConditionTests(BaseTestCase):
         # Note that this test is sensitive to timing.  If the worker threads
         # don't execute in a timely fashion, the main thread may think they
         # are further along then they are.  The main thread therefore issues
-        # _wait() statements to try to make sure that it doesn't race ahead
-        # of the workers.
+        # wait_threads_blocked() statements to try to make sure that it doesn't
+        # race ahead of the workers.
         # Secondly, this test assumes that condition variables are not subject
         # to spurious wakeups.  The absence of spurious wakeups is an implementation
         # detail of Condition Variables in current CPython, but in general, not
         # a guaranteed property of condition variables as a programming
         # construct.  In particular, it is possible that this can no longer
         # be conveniently guaranteed should their implementation ever change.
-        N = 5
         ready = []
         results1 = []
         results2 = []
@@ -483,58 +597,83 @@ class ConditionTests(BaseTestCase):
             cond.acquire()
             ready.append(phase_num)
             result = cond.wait()
+
             cond.release()
             results1.append((result, phase_num))
+
             cond.acquire()
             ready.append(phase_num)
+
             result = cond.wait()
             cond.release()
             results2.append((result, phase_num))
-        b = Bunch(f, N)
-        b.wait_for_started()
-        # first wait, to ensure all workers settle into cond.wait() before
-        # we continue. See issues #8799 and #30727.
-        while len(ready) < 5:
-            _wait()
-        ready.clear()
-        self.assertEqual(results1, [])
-        # Notify 3 threads at first
-        cond.acquire()
-        cond.notify(3)
-        _wait()
-        phase_num = 1
-        cond.release()
-        while len(results1) < 3:
-            _wait()
-        self.assertEqual(results1, [(True, 1)] * 3)
-        self.assertEqual(results2, [])
-        # make sure all awaken workers settle into cond.wait()
-        while len(ready) < 3:
-            _wait()
-        # Notify 5 threads: they might be in their first or second wait
-        cond.acquire()
-        cond.notify(5)
-        _wait()
-        phase_num = 2
-        cond.release()
-        while len(results1) + len(results2) < 8:
-            _wait()
-        self.assertEqual(results1, [(True, 1)] * 3 + [(True, 2)] * 2)
-        self.assertEqual(results2, [(True, 2)] * 3)
-        # make sure all workers settle into cond.wait()
-        while len(ready) < 5:
-            _wait()
-        # Notify all threads: they are all in their second wait
-        cond.acquire()
-        cond.notify_all()
-        _wait()
-        phase_num = 3
-        cond.release()
-        while len(results2) < 5:
-            _wait()
-        self.assertEqual(results1, [(True, 1)] * 3 + [(True,2)] * 2)
-        self.assertEqual(results2, [(True, 2)] * 3 + [(True, 3)] * 2)
-        b.wait_for_finished()
+
+        N = 5
+        with Bunch(f, N):
+            # first wait, to ensure all workers settle into cond.wait() before
+            # we continue. See issues #8799 and #30727.
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                if len(ready) >= N:
+                    break
+
+            ready.clear()
+            self.assertEqual(results1, [])
+
+            # Notify 3 threads at first
+            count1 = 3
+            cond.acquire()
+            cond.notify(count1)
+            wait_threads_blocked(count1)
+
+            # Phase 1
+            phase_num = 1
+            cond.release()
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                if len(results1) >= count1:
+                    break
+
+            self.assertEqual(results1, [(True, 1)] * count1)
+            self.assertEqual(results2, [])
+
+            # Wait until awaken workers are blocked on cond.wait()
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                if len(ready) >= count1 :
+                    break
+
+            # Notify 5 threads: they might be in their first or second wait
+            cond.acquire()
+            cond.notify(5)
+            wait_threads_blocked(N)
+
+            # Phase 2
+            phase_num = 2
+            cond.release()
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                if len(results1) + len(results2) >= (N + count1):
+                    break
+
+            count2 = N - count1
+            self.assertEqual(results1, [(True, 1)] * count1 + [(True, 2)] * count2)
+            self.assertEqual(results2, [(True, 2)] * count1)
+
+            # Make sure all workers settle into cond.wait()
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                if len(ready) >= N:
+                    break
+
+            # Notify all threads: they are all in their second wait
+            cond.acquire()
+            cond.notify_all()
+            wait_threads_blocked(N)
+
+            # Phase 3
+            phase_num = 3
+            cond.release()
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                if len(results2) >= N:
+                    break
+            self.assertEqual(results1, [(True, 1)] * count1 + [(True, 2)] * count2)
+            self.assertEqual(results2, [(True, 2)] * count1 + [(True, 3)] * count2)
 
     def test_notify(self):
         cond = self.condtype()
@@ -544,19 +683,23 @@ class ConditionTests(BaseTestCase):
 
     def test_timeout(self):
         cond = self.condtype()
+        timeout = 0.5
         results = []
-        N = 5
         def f():
             cond.acquire()
             t1 = time.monotonic()
-            result = cond.wait(0.5)
+            result = cond.wait(timeout)
             t2 = time.monotonic()
             cond.release()
             results.append((t2 - t1, result))
-        Bunch(f, N).wait_for_finished()
+
+        N = 5
+        with Bunch(f, N):
+            pass
         self.assertEqual(len(results), N)
+
         for dt, result in results:
-            self.assertTimeout(dt, 0.5)
+            self.assertTimeout(dt, timeout)
             # Note that conceptually (that"s the condition variable protocol)
             # a wait() may succeed even if no one notifies us and before any
             # timeout occurs.  Spurious wakeups can occur.
@@ -569,17 +712,16 @@ class ConditionTests(BaseTestCase):
         state = 0
         def f():
             with cond:
-                result = cond.wait_for(lambda : state==4)
+                result = cond.wait_for(lambda: state == 4)
                 self.assertTrue(result)
                 self.assertEqual(state, 4)
-        b = Bunch(f, 1)
-        b.wait_for_started()
-        for i in range(4):
-            time.sleep(0.01)
-            with cond:
-                state += 1
-                cond.notify()
-        b.wait_for_finished()
+
+        with Bunch(f, 1):
+            for i in range(4):
+                time.sleep(0.010)
+                with cond:
+                    state += 1
+                    cond.notify()
 
     def test_waitfor_timeout(self):
         cond = self.condtype()
@@ -593,15 +735,15 @@ class ConditionTests(BaseTestCase):
                 self.assertFalse(result)
                 self.assertTimeout(dt, 0.1)
                 success.append(None)
-        b = Bunch(f, 1)
-        b.wait_for_started()
-        # Only increment 3 times, so state == 4 is never reached.
-        for i in range(3):
-            time.sleep(0.01)
-            with cond:
-                state += 1
-                cond.notify()
-        b.wait_for_finished()
+
+        with Bunch(f, 1):
+            # Only increment 3 times, so state == 4 is never reached.
+            for i in range(3):
+                time.sleep(0.010)
+                with cond:
+                    state += 1
+                    cond.notify()
+
         self.assertEqual(len(success), 1)
 
 
@@ -630,41 +772,107 @@ class BaseSemaphoreTests(BaseTestCase):
         del sem
 
     def test_acquire_contended(self):
-        sem = self.semtype(7)
+        sem_value = 7
+        sem = self.semtype(sem_value)
         sem.acquire()
-        N = 10
+
         sem_results = []
         results1 = []
         results2 = []
         phase_num = 0
-        def f():
+
+        def func():
             sem_results.append(sem.acquire())
             results1.append(phase_num)
+
             sem_results.append(sem.acquire())
             results2.append(phase_num)
-        b = Bunch(f, 10)
-        b.wait_for_started()
-        while len(results1) + len(results2) < 6:
-            _wait()
-        self.assertEqual(results1 + results2, [0] * 6)
-        phase_num = 1
-        for i in range(7):
+
+        def wait_count(count):
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                if len(results1) + len(results2) >= count:
+                    break
+
+        N = 10
+        with Bunch(func, N):
+            # Phase 0
+            count1 = sem_value - 1
+            wait_count(count1)
+            self.assertEqual(results1 + results2, [0] * count1)
+
+            # Phase 1
+            phase_num = 1
+            for i in range(sem_value):
+                sem.release()
+            count2 = sem_value
+            wait_count(count1 + count2)
+            self.assertEqual(sorted(results1 + results2),
+                             [0] * count1 + [1] * count2)
+
+            # Phase 2
+            phase_num = 2
+            count3 = (sem_value - 1)
+            for i in range(count3):
+                sem.release()
+            wait_count(count1 + count2 + count3)
+            self.assertEqual(sorted(results1 + results2),
+                             [0] * count1 + [1] * count2 + [2] * count3)
+            # The semaphore is still locked
+            self.assertFalse(sem.acquire(False))
+
+            # Final release, to let the last thread finish
+            count4 = 1
             sem.release()
-        while len(results1) + len(results2) < 13:
-            _wait()
-        self.assertEqual(sorted(results1 + results2), [0] * 6 + [1] * 7)
-        phase_num = 2
-        for i in range(6):
+
+        self.assertEqual(sem_results,
+                         [True] * (count1 + count2 + count3 + count4))
+
+    def test_multirelease(self):
+        sem_value = 7
+        sem = self.semtype(sem_value)
+        sem.acquire()
+
+        results1 = []
+        results2 = []
+        phase_num = 0
+        def func():
+            sem.acquire()
+            results1.append(phase_num)
+
+            sem.acquire()
+            results2.append(phase_num)
+
+        def wait_count(count):
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                if len(results1) + len(results2) >= count:
+                    break
+
+        with Bunch(func, 10):
+            # Phase 0
+            count1 = sem_value - 1
+            wait_count(count1)
+            self.assertEqual(results1 + results2, [0] * count1)
+
+            # Phase 1
+            phase_num = 1
+            count2 = sem_value
+            sem.release(count2)
+            wait_count(count1 + count2)
+            self.assertEqual(sorted(results1 + results2),
+                             [0] * count1 + [1] * count2)
+
+            # Phase 2
+            phase_num = 2
+            count3 = sem_value - 1
+            sem.release(count3)
+            wait_count(count1 + count2 + count3)
+            self.assertEqual(sorted(results1 + results2),
+                             [0] * count1 + [1] * count2 + [2] * count3)
+            # The semaphore is still locked
+            self.assertFalse(sem.acquire(False))
+
+            # Final release, to let the last thread finish
             sem.release()
-        while len(results1) + len(results2) < 19:
-            _wait()
-        self.assertEqual(sorted(results1 + results2), [0] * 6 + [1] * 7 + [2] * 6)
-        # The semaphore is still locked
-        self.assertFalse(sem.acquire(False))
-        # Final release, to let the last thread finish
-        sem.release()
-        b.wait_for_finished()
-        self.assertEqual(sem_results, [True] * (6 + 7 + 6 + 1))
 
     def test_try_acquire(self):
         sem = self.semtype(2)
@@ -681,7 +889,8 @@ class BaseSemaphoreTests(BaseTestCase):
         def f():
             results.append(sem.acquire(False))
             results.append(sem.acquire(False))
-        Bunch(f, 5).wait_for_finished()
+        with Bunch(f, 5):
+            pass
         # There can be a thread switch between acquiring the semaphore and
         # appending the result, therefore results will not necessarily be
         # ordered.
@@ -707,12 +916,14 @@ class BaseSemaphoreTests(BaseTestCase):
         def f():
             sem.acquire()
             sem.release()
-        b = Bunch(f, 1)
-        b.wait_for_started()
-        _wait()
-        self.assertFalse(b.finished)
-        sem.release()
-        b.wait_for_finished()
+
+        with Bunch(f, 1) as bunch:
+            # Thread blocked on sem.acquire()
+            wait_threads_blocked(1)
+            self.assertFalse(bunch.finished)
+
+            # Thread unblocked
+            sem.release()
 
     def test_with(self):
         sem = self.semtype(2)
@@ -744,6 +955,15 @@ class SemaphoreTests(BaseSemaphoreTests):
         sem.acquire()
         sem.release()
 
+    def test_repr(self):
+        sem = self.semtype(3)
+        self.assertRegex(repr(sem), r"<\w+\.Semaphore at .*: value=3>")
+        sem.acquire()
+        self.assertRegex(repr(sem), r"<\w+\.Semaphore at .*: value=2>")
+        sem.release()
+        sem.release()
+        self.assertRegex(repr(sem), r"<\w+\.Semaphore at .*: value=4>")
+
 
 class BoundedSemaphoreTests(BaseSemaphoreTests):
     """
@@ -758,6 +978,12 @@ class BoundedSemaphoreTests(BaseSemaphoreTests):
         sem.release()
         self.assertRaises(ValueError, sem.release)
 
+    def test_repr(self):
+        sem = self.semtype(3)
+        self.assertRegex(repr(sem), r"<\w+\.BoundedSemaphore at .*: value=3/3>")
+        sem.acquire()
+        self.assertRegex(repr(sem), r"<\w+\.BoundedSemaphore at .*: value=2/3>")
+
 
 class BarrierTests(BaseTestCase):
     """
@@ -768,13 +994,13 @@ class BarrierTests(BaseTestCase):
 
     def setUp(self):
         self.barrier = self.barriertype(self.N, timeout=self.defaultTimeout)
+
     def tearDown(self):
         self.barrier.abort()
 
     def run_threads(self, f):
-        b = Bunch(f, self.N-1)
-        f()
-        b.wait_for_finished()
+        with Bunch(f, self.N):
+            pass
 
     def multipass(self, results, n):
         m = self.barrier.parties
@@ -789,6 +1015,10 @@ class BarrierTests(BaseTestCase):
         self.assertEqual(self.barrier.n_waiting, 0)
         self.assertFalse(self.barrier.broken)
 
+    def test_constructor(self):
+        self.assertRaises(ValueError, self.barriertype, parties=0)
+        self.assertRaises(ValueError, self.barriertype, parties=-1)
+
     def test_barrier(self, passes=1):
         """
         Test that a barrier is passed in lockstep
@@ -865,8 +1095,9 @@ class BarrierTests(BaseTestCase):
             i = self.barrier.wait()
             if i == self.N//2:
                 # Wait until the other threads are all in the barrier.
-                while self.barrier.n_waiting < self.N-1:
-                    time.sleep(0.001)
+                for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                    if self.barrier.n_waiting >= (self.N - 1):
+                        break
                 self.barrier.reset()
             else:
                 try:
@@ -926,27 +1157,56 @@ class BarrierTests(BaseTestCase):
             i = self.barrier.wait()
             if i == self.N // 2:
                 # One thread is late!
-                time.sleep(1.0)
+                time.sleep(self.defaultTimeout / 2)
             # Default timeout is 2.0, so this is shorter.
             self.assertRaises(threading.BrokenBarrierError,
-                              self.barrier.wait, 0.5)
+                              self.barrier.wait, self.defaultTimeout / 4)
         self.run_threads(f)
 
     def test_default_timeout(self):
         """
         Test the barrier's default timeout
         """
-        # create a barrier with a low default timeout
-        barrier = self.barriertype(self.N, timeout=0.3)
+        timeout = 0.100
+        barrier = self.barriertype(2, timeout=timeout)
         def f():
-            i = barrier.wait()
-            if i == self.N // 2:
-                # One thread is later than the default timeout of 0.3s.
-                time.sleep(1.0)
-            self.assertRaises(threading.BrokenBarrierError, barrier.wait)
-        self.run_threads(f)
+            self.assertRaises(threading.BrokenBarrierError,
+                              barrier.wait)
+
+        start_time = time.monotonic()
+        with Bunch(f, 1):
+            pass
+        dt = time.monotonic() - start_time
+        self.assertGreaterEqual(dt, timeout)
 
     def test_single_thread(self):
         b = self.barriertype(1)
         b.wait()
         b.wait()
+
+    def test_repr(self):
+        barrier = self.barriertype(3)
+        timeout = support.LONG_TIMEOUT
+        self.assertRegex(repr(barrier), r"<\w+\.Barrier at .*: waiters=0/3>")
+        def f():
+            barrier.wait(timeout)
+
+        N = 2
+        with Bunch(f, N):
+            # Threads blocked on barrier.wait()
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                if barrier.n_waiting >= N:
+                    break
+            self.assertRegex(repr(barrier),
+                             r"<\w+\.Barrier at .*: waiters=2/3>")
+
+            # Threads unblocked
+            barrier.wait(timeout)
+
+        self.assertRegex(repr(barrier),
+                         r"<\w+\.Barrier at .*: waiters=0/3>")
+
+        # Abort the barrier
+        barrier.abort()
+        self.assertRegex(repr(barrier),
+                         r"<\w+\.Barrier at .*: broken>")

Lib/test/test_concurrent_futures/test_process_pool.py

@@ -190,7 +190,6 @@ class ProcessPoolExecutorTest(ExecutorTest):
         for i, future in enumerate(futures):
             self.assertEqual(future.result(), mul(i, i))
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON AttributeError: module 'threading' has no attribute '_start_joinable_thread'. Did you mean: '_start_new_thread'?
     def test_python_finalization_error(self):
         # gh-109047: Catch RuntimeError on thread creation
         # during Python finalization.

Lib/test/test_concurrent_futures/test_shutdown.py

@@ -108,7 +108,6 @@ class ExecutorShutdownTest:
         # one finished.
         self.assertGreater(len(others), 0)
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON AssertionError: b'' != b'apple'
     def test_hang_gh83386(self):
         """shutdown(wait=False) doesn't hang at exit with running futures.
 
@@ -256,6 +255,11 @@ class ThreadPoolShutdownTest(ThreadPoolMixin, ExecutorShutdownTest, BaseTestCase
 
 
 class ProcessPoolShutdownTest(ExecutorShutdownTest):
+    # TODO: RUSTPYTHON - flaky, dict changed size during iteration race condition
+    @unittest.skip("TODO: RUSTPYTHON - flaky race condition on macOS")
+    def test_cancel_futures(self):
+        return super().test_cancel_futures()
+
     def test_processes_terminate(self):
         def acquire_lock(lock):
             lock.acquire()

Lib/test/test_concurrent_futures/test_thread_pool.py

@@ -66,14 +66,8 @@ class ThreadPoolExecutorTest(ThreadPoolMixin, ExecutorTest, BaseTestCase):
                 with futures.ProcessPoolExecutor(1, mp_context=mp.get_context('fork')) as workers:
                     workers.submit(tuple)
 
-    import sys  # TODO: RUSTPYTHON
-    @unittest.skipIf(
-        'RUSTPYTHON_SKIP_ENV_POLLUTERS' in os.environ,
-        'TODO: RUSTPYTHON environment pollution when running rustpython -m test --fail-env-changed due to unknown reason'
-    )
     @support.requires_fork()
     @unittest.skipUnless(hasattr(os, 'register_at_fork'), 'need os.register_at_fork')
-    @unittest.expectedFailure # TODO: RUSTPYTHON AssertionError: DeprecationWarning not triggered
     def test_process_fork_from_a_threadpool(self):
         # bpo-43944: clear concurrent.futures.thread._threads_queues after fork,
         # otherwise child process will try to join parent thread

Lib/test/test_importlib/test_locks.py

@@ -29,6 +29,8 @@ class ModuleLockAsRLockTests:
     test_timeout = None
     # _release_save() unsupported
     test_release_save_unacquired = None
+    # _recursion_count() unsupported
+    test_recursion_count = None
     # lock status in repr unsupported
     test_repr = None
     test_locked_repr = None
@@ -92,7 +94,8 @@ class DeadlockAvoidanceTests:
                 b.release()
             if ra:
                 a.release()
-        lock_tests.Bunch(f, NTHREADS).wait_for_finished()
+        with lock_tests.Bunch(f, NTHREADS):
+            pass
         self.assertEqual(len(results), NTHREADS)
         return results
 

Lib/test/test_sys.py

@@ -920,7 +920,7 @@ class SysModuleTest(unittest.TestCase):
     def test_clear_type_cache(self):
         sys._clear_type_cache()
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON
+    @unittest.skip("TODO: RUSTPYTHON; cp424 encoding not supported, causes panic")
     @force_not_colorized
     @support.requires_subprocess()
     def test_ioencoding(self):

Lib/threading.py

@@ -3,11 +3,11 @@
 import os as _os
 import sys as _sys
 import _thread
-import functools
+import warnings
 
 from time import monotonic as _time
 from _weakrefset import WeakSet
-from itertools import islice as _islice, count as _count
+from itertools import count as _count
 try:
     from _collections import deque as _deque
 except ImportError:
@@ -28,13 +28,20 @@ __all__ = ['get_ident', 'active_count', 'Condition', 'current_thread',
            'Event', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Thread',
            'Barrier', 'BrokenBarrierError', 'Timer', 'ThreadError',
            'setprofile', 'settrace', 'local', 'stack_size',
-           'excepthook', 'ExceptHookArgs', 'gettrace', 'getprofile']
+           'excepthook', 'ExceptHookArgs', 'gettrace', 'getprofile',
+           'setprofile_all_threads','settrace_all_threads']
 
 # Rename some stuff so "from threading import *" is safe
-_start_new_thread = _thread.start_new_thread
+_start_joinable_thread = _thread.start_joinable_thread
+_daemon_threads_allowed = _thread.daemon_threads_allowed
 _allocate_lock = _thread.allocate_lock
-_set_sentinel = _thread._set_sentinel
+_LockType = _thread.LockType
+_thread_shutdown = _thread._shutdown
+_make_thread_handle = _thread._make_thread_handle
+_ThreadHandle = _thread._ThreadHandle
 get_ident = _thread.get_ident
+_get_main_thread_ident = _thread._get_main_thread_ident
+_is_main_interpreter = _thread._is_main_interpreter
 try:
     get_native_id = _thread.get_native_id
     _HAVE_THREAD_NATIVE_ID = True
@@ -49,6 +56,13 @@ except AttributeError:
 TIMEOUT_MAX = _thread.TIMEOUT_MAX
 del _thread
 
+# get thread-local implementation, either from the thread
+# module, or from the python fallback
+
+try:
+    from _thread import _local as local
+except ImportError:
+    from _threading_local import local
 
 # Support for profile and trace hooks
 
@@ -60,11 +74,20 @@ def setprofile(func):
 
     The func will be passed to sys.setprofile() for each thread, before its
     run() method is called.
-
     """
     global _profile_hook
     _profile_hook = func
 
+def setprofile_all_threads(func):
+    """Set a profile function for all threads started from the threading module
+    and all Python threads that are currently executing.
+
+    The func will be passed to sys.setprofile() for each thread, before its
+    run() method is called.
+    """
+    setprofile(func)
+    _sys._setprofileallthreads(func)
+
 def getprofile():
     """Get the profiler function as set by threading.setprofile()."""
     return _profile_hook
@@ -74,12 +97,10 @@ def settrace(func):
 
     The func will be passed to sys.settrace() for each thread, before its run()
     method is called.
-
     """
     global _trace_hook
     _trace_hook = func
 
-# TODO: RUSTPYTHON
 def settrace_all_threads(func):
     """Set a trace function for all threads started from the threading module
     and all Python threads that are currently executing.
@@ -96,7 +117,7 @@ def gettrace():
 
 # Synchronization classes
 
-Lock = _allocate_lock
+Lock = _LockType
 
 def RLock(*args, **kwargs):
     """Factory function that returns a new reentrant lock.
@@ -107,6 +128,12 @@ def RLock(*args, **kwargs):
     acquired it.
 
     """
+    if args or kwargs:
+        warnings.warn(
+            'Passing arguments to RLock is deprecated and will be removed in 3.15',
+            DeprecationWarning,
+            stacklevel=2,
+        )
     if _CRLock is None:
         return _PyRLock(*args, **kwargs)
     return _CRLock(*args, **kwargs)
@@ -229,6 +256,13 @@ class _RLock:
     def _is_owned(self):
         return self._owner == get_ident()
 
+    # Internal method used for reentrancy checks
+
+    def _recursion_count(self):
+        if self._owner != get_ident():
+            return 0
+        return self._count
+
 _PyRLock = _RLock
 
 
@@ -254,18 +288,12 @@ class Condition:
         # If the lock defines _release_save() and/or _acquire_restore(),
         # these override the default implementations (which just call
         # release() and acquire() on the lock).  Ditto for _is_owned().
-        try:
+        if hasattr(lock, '_release_save'):
             self._release_save = lock._release_save
-        except AttributeError:
-            pass
-        try:
+        if hasattr(lock, '_acquire_restore'):
             self._acquire_restore = lock._acquire_restore
-        except AttributeError:
-            pass
-        try:
+        if hasattr(lock, '_is_owned'):
             self._is_owned = lock._is_owned
-        except AttributeError:
-            pass
         self._waiters = _deque()
 
     def _at_fork_reinit(self):
@@ -308,7 +336,7 @@ class Condition:
         awakened or timed out, it re-acquires the lock and returns.
 
         When the timeout argument is present and not None, it should be a
-        floating point number specifying a timeout for the operation in seconds
+        floating-point number specifying a timeout for the operation in seconds
         (or fractions thereof).
 
         When the underlying lock is an RLock, it is not released using its
@@ -436,6 +464,11 @@ class Semaphore:
         self._cond = Condition(Lock())
         self._value = value
 
+    def __repr__(self):
+        cls = self.__class__
+        return (f"<{cls.__module__}.{cls.__qualname__} at {id(self):#x}:"
+                f" value={self._value}>")
+
     def acquire(self, blocking=True, timeout=None):
         """Acquire a semaphore, decrementing the internal counter by one.
 
@@ -494,8 +527,7 @@ class Semaphore:
             raise ValueError('n must be one or more')
         with self._cond:
             self._value += n
-            for i in range(n):
-                self._cond.notify()
+            self._cond.notify(n)
 
     def __exit__(self, t, v, tb):
         self.release()
@@ -519,9 +551,14 @@ class BoundedSemaphore(Semaphore):
     """
 
     def __init__(self, value=1):
-        Semaphore.__init__(self, value)
+        super().__init__(value)
         self._initial_value = value
 
+    def __repr__(self):
+        cls = self.__class__
+        return (f"<{cls.__module__}.{cls.__qualname__} at {id(self):#x}:"
+                f" value={self._value}/{self._initial_value}>")
+
     def release(self, n=1):
         """Release a semaphore, incrementing the internal counter by one or more.
 
@@ -538,8 +575,7 @@ class BoundedSemaphore(Semaphore):
             if self._value + n > self._initial_value:
                 raise ValueError("Semaphore released too many times")
             self._value += n
-            for i in range(n):
-                self._cond.notify()
+            self._cond.notify(n)
 
 
 class Event:
@@ -557,8 +593,13 @@ class Event:
         self._cond = Condition(Lock())
         self._flag = False
 
+    def __repr__(self):
+        cls = self.__class__
+        status = 'set' if self._flag else 'unset'
+        return f"<{cls.__module__}.{cls.__qualname__} at {id(self):#x}: {status}>"
+
     def _at_fork_reinit(self):
-        # Private method called by Thread._reset_internal_locks()
+        # Private method called by Thread._after_fork()
         self._cond._at_fork_reinit()
 
     def is_set(self):
@@ -568,7 +609,7 @@ class Event:
     def isSet(self):
         """Return true if and only if the internal flag is true.
 
-        This method is deprecated, use notify_all() instead.
+        This method is deprecated, use is_set() instead.
 
         """
         import warnings
@@ -605,11 +646,12 @@ class Event:
         the optional timeout occurs.
 
         When the timeout argument is present and not None, it should be a
-        floating point number specifying a timeout for the operation in seconds
+        floating-point number specifying a timeout for the operation in seconds
         (or fractions thereof).
 
         This method returns the internal flag on exit, so it will always return
-        True except if a timeout is given and the operation times out.
+        ``True`` except if a timeout is given and the operation times out, when
+        it will return ``False``.
 
         """
         with self._cond:
@@ -648,6 +690,8 @@ class Barrier:
         default for all subsequent 'wait()' calls.
 
         """
+        if parties < 1:
+            raise ValueError("parties must be >= 1")
         self._cond = Condition(Lock())
         self._action = action
         self._timeout = timeout
@@ -655,6 +699,13 @@ class Barrier:
         self._state = 0  # 0 filling, 1 draining, -1 resetting, -2 broken
         self._count = 0
 
+    def __repr__(self):
+        cls = self.__class__
+        if self.broken:
+            return f"<{cls.__module__}.{cls.__qualname__} at {id(self):#x}: broken>"
+        return (f"<{cls.__module__}.{cls.__qualname__} at {id(self):#x}:"
+                f" waiters={self.n_waiting}/{self.parties}>")
+
     def wait(self, timeout=None):
         """Wait for the barrier.
 
@@ -802,25 +853,6 @@ _active = {}    # maps thread id to Thread object
 _limbo = {}
 _dangling = WeakSet()
 
-# Set of Thread._tstate_lock locks of non-daemon threads used by _shutdown()
-# to wait until all Python thread states get deleted:
-# see Thread._set_tstate_lock().
-_shutdown_locks_lock = _allocate_lock()
-_shutdown_locks = set()
-
-def _maintain_shutdown_locks():
-    """
-    Drop any shutdown locks that don't correspond to running threads anymore.
-
-    Calling this from time to time avoids an ever-growing _shutdown_locks
-    set when Thread objects are not joined explicitly. See bpo-37788.
-
-    This must be called with _shutdown_locks_lock acquired.
-    """
-    # If a lock was released, the corresponding thread has exited
-    to_remove = [lock for lock in _shutdown_locks if not lock.locked()]
-    _shutdown_locks.difference_update(to_remove)
-
 
 # Main class for threads
 
@@ -848,7 +880,7 @@ class Thread:
         *name* is the thread name. By default, a unique name is constructed of
         the form "Thread-N" where N is a small decimal number.
 
-        *args* is the argument tuple for the target invocation. Defaults to ().
+        *args* is a list or tuple of arguments for the target invocation. Defaults to ().
 
         *kwargs* is a dictionary of keyword arguments for the target
         invocation. Defaults to {}.
@@ -877,15 +909,16 @@ class Thread:
         self._args = args
         self._kwargs = kwargs
         if daemon is not None:
+            if daemon and not _daemon_threads_allowed():
+                raise RuntimeError('daemon threads are disabled in this (sub)interpreter')
             self._daemonic = daemon
         else:
             self._daemonic = current_thread().daemon
         self._ident = None
         if _HAVE_THREAD_NATIVE_ID:
             self._native_id = None
-        self._tstate_lock = None
+        self._handle = _ThreadHandle()
         self._started = Event()
-        self._is_stopped = False
         self._initialized = True
         # Copy of sys.stderr used by self._invoke_excepthook()
         self._stderr = _sys.stderr
@@ -893,30 +926,26 @@ class Thread:
         # For debugging and _after_fork()
         _dangling.add(self)
 
-    def _reset_internal_locks(self, is_alive):
-        # private!  Called by _after_fork() to reset our internal locks as
-        # they may be in an invalid state leading to a deadlock or crash.
+    def _after_fork(self, new_ident=None):
+        # Private!  Called by threading._after_fork().
         self._started._at_fork_reinit()
-        if is_alive:
-            # bpo-42350: If the fork happens when the thread is already stopped
-            # (ex: after threading._shutdown() has been called), _tstate_lock
-            # is None. Do nothing in this case.
-            if self._tstate_lock is not None:
-                self._tstate_lock._at_fork_reinit()
-                self._tstate_lock.acquire()
+        if new_ident is not None:
+            # This thread is alive.
+            self._ident = new_ident
+            assert self._handle.ident == new_ident
+            if _HAVE_THREAD_NATIVE_ID:
+                self._set_native_id()
         else:
-            # The thread isn't alive after fork: it doesn't have a tstate
-            # anymore.
-            self._is_stopped = True
-            self._tstate_lock = None
+            # Otherwise, the thread is dead, Jim.  _PyThread_AfterFork()
+            # already marked our handle done.
+            pass
 
     def __repr__(self):
         assert self._initialized, "Thread.__init__() was not called"
         status = "initial"
         if self._started.is_set():
             status = "started"
-        self.is_alive() # easy way to get ._is_stopped set when appropriate
-        if self._is_stopped:
+        if self._handle.is_done():
             status = "stopped"
         if self._daemonic:
             status += " daemon"
@@ -943,12 +972,14 @@ class Thread:
         with _active_limbo_lock:
             _limbo[self] = self
         try:
-            _start_new_thread(self._bootstrap, ())
+            # Start joinable thread
+            _start_joinable_thread(self._bootstrap, handle=self._handle,
+                                   daemon=self.daemon)
         except Exception:
             with _active_limbo_lock:
                 del _limbo[self]
             raise
-        self._started.wait()
+        self._started.wait()  # Will set ident and native_id
 
     def run(self):
         """Method representing the thread's activity.
@@ -994,23 +1025,9 @@ class Thread:
         def _set_native_id(self):
             self._native_id = get_native_id()
 
-    def _set_tstate_lock(self):
-        """
-        Set a lock object which will be released by the interpreter when
-        the underlying thread state (see pystate.h) gets deleted.
-        """
-        self._tstate_lock = _set_sentinel()
-        self._tstate_lock.acquire()
-
-        if not self.daemon:
-            with _shutdown_locks_lock:
-                _maintain_shutdown_locks()
-                _shutdown_locks.add(self._tstate_lock)
-
     def _bootstrap_inner(self):
         try:
             self._set_ident()
-            self._set_tstate_lock()
             if _HAVE_THREAD_NATIVE_ID:
                 self._set_native_id()
             self._started.set()
@@ -1028,40 +1045,7 @@ class Thread:
             except:
                 self._invoke_excepthook(self)
         finally:
-            with _active_limbo_lock:
-                try:
-                    # We don't call self._delete() because it also
-                    # grabs _active_limbo_lock.
-                    del _active[get_ident()]
-                except:
-                    pass
-
-    def _stop(self):
-        # After calling ._stop(), .is_alive() returns False and .join() returns
-        # immediately.  ._tstate_lock must be released before calling ._stop().
-        #
-        # Normal case:  C code at the end of the thread's life
-        # (release_sentinel in _threadmodule.c) releases ._tstate_lock, and
-        # that's detected by our ._wait_for_tstate_lock(), called by .join()
-        # and .is_alive().  Any number of threads _may_ call ._stop()
-        # simultaneously (for example, if multiple threads are blocked in
-        # .join() calls), and they're not serialized.  That's harmless -
-        # they'll just make redundant rebindings of ._is_stopped and
-        # ._tstate_lock.  Obscure:  we rebind ._tstate_lock last so that the
-        # "assert self._is_stopped" in ._wait_for_tstate_lock() always works
-        # (the assert is executed only if ._tstate_lock is None).
-        #
-        # Special case:  _main_thread releases ._tstate_lock via this
-        # module's _shutdown() function.
-        lock = self._tstate_lock
-        if lock is not None:
-            assert not lock.locked()
-        self._is_stopped = True
-        self._tstate_lock = None
-        if not self.daemon:
-            with _shutdown_locks_lock:
-                # Remove our lock and other released locks from _shutdown_locks
-                _maintain_shutdown_locks()
+            self._delete()
 
     def _delete(self):
         "Remove current thread from the dict of currently running threads."
@@ -1080,7 +1064,7 @@ class Thread:
         or until the optional timeout occurs.
 
         When the timeout argument is present and not None, it should be a
-        floating point number specifying a timeout for the operation in seconds
+        floating-point number specifying a timeout for the operation in seconds
         (or fractions thereof). As join() always returns None, you must call
         is_alive() after join() to decide whether a timeout happened -- if the
         thread is still alive, the join() call timed out.
@@ -1103,39 +1087,12 @@ class Thread:
         if self is current_thread():
             raise RuntimeError("cannot join current thread")
 
-        if timeout is None:
-            self._wait_for_tstate_lock()
-        else:
-            # the behavior of a negative timeout isn't documented, but
-            # historically .join(timeout=x) for x<0 has acted as if timeout=0
-            self._wait_for_tstate_lock(timeout=max(timeout, 0))
+        # the behavior of a negative timeout isn't documented, but
+        # historically .join(timeout=x) for x<0 has acted as if timeout=0
+        if timeout is not None:
+            timeout = max(timeout, 0)
 
-    def _wait_for_tstate_lock(self, block=True, timeout=-1):
-        # Issue #18808: wait for the thread state to be gone.
-        # At the end of the thread's life, after all knowledge of the thread
-        # is removed from C data structures, C code releases our _tstate_lock.
-        # This method passes its arguments to _tstate_lock.acquire().
-        # If the lock is acquired, the C code is done, and self._stop() is
-        # called.  That sets ._is_stopped to True, and ._tstate_lock to None.
-        lock = self._tstate_lock
-        if lock is None:
-            # already determined that the C code is done
-            assert self._is_stopped
-            return
-
-        try:
-            if lock.acquire(block, timeout):
-                lock.release()
-                self._stop()
-        except:
-            if lock.locked():
-                # bpo-45274: lock.acquire() acquired the lock, but the function
-                # was interrupted with an exception before reaching the
-                # lock.release(). It can happen if a signal handler raises an
-                # exception, like CTRL+C which raises KeyboardInterrupt.
-                lock.release()
-                self._stop()
-            raise
+        self._handle.join(timeout)
 
     @property
     def name(self):
@@ -1186,10 +1143,7 @@ class Thread:
 
         """
         assert self._initialized, "Thread.__init__() not called"
-        if self._is_stopped or not self._started.is_set():
-            return False
-        self._wait_for_tstate_lock(False)
-        return not self._is_stopped
+        return self._started.is_set() and not self._handle.is_done()
 
     @property
     def daemon(self):
@@ -1210,6 +1164,8 @@ class Thread:
     def daemon(self, daemonic):
         if not self._initialized:
             raise RuntimeError("Thread.__init__() not called")
+        if daemonic and not _daemon_threads_allowed():
+            raise RuntimeError('daemon threads are disabled in this interpreter')
         if self._started.is_set():
             raise RuntimeError("cannot set daemon status of active thread")
         self._daemonic = daemonic
@@ -1396,19 +1352,45 @@ class _MainThread(Thread):
 
     def __init__(self):
         Thread.__init__(self, name="MainThread", daemon=False)
-        self._set_tstate_lock()
         self._started.set()
-        self._set_ident()
+        self._ident = _get_main_thread_ident()
+        self._handle = _make_thread_handle(self._ident)
         if _HAVE_THREAD_NATIVE_ID:
             self._set_native_id()
         with _active_limbo_lock:
             _active[self._ident] = self
 
 
+# Helper thread-local instance to detect when a _DummyThread
+# is collected. Not a part of the public API.
+_thread_local_info = local()
+
+
+class _DeleteDummyThreadOnDel:
+    '''
+    Helper class to remove a dummy thread from threading._active on __del__.
+    '''
+
+    def __init__(self, dummy_thread):
+        self._dummy_thread = dummy_thread
+        self._tident = dummy_thread.ident
+        # Put the thread on a thread local variable so that when
+        # the related thread finishes this instance is collected.
+        #
+        # Note: no other references to this instance may be created.
+        # If any client code creates a reference to this instance,
+        # the related _DummyThread will be kept forever!
+        _thread_local_info._track_dummy_thread_ref = self
+
+    def __del__(self, _active_limbo_lock=_active_limbo_lock, _active=_active):
+        with _active_limbo_lock:
+            if _active.get(self._tident) is self._dummy_thread:
+                _active.pop(self._tident, None)
+
+
 # Dummy thread class to represent threads not started here.
-# These aren't garbage collected when they die, nor can they be waited for.
-# If they invoke anything in threading.py that calls current_thread(), they
-# leave an entry in the _active dict forever after.
+# These should be added to `_active` and removed automatically
+# when they die, although they can't be waited for.
 # Their purpose is to return *something* from current_thread().
 # They are marked as daemon threads so we won't wait for them
 # when we exit (conform previous semantics).
@@ -1416,24 +1398,31 @@ class _MainThread(Thread):
 class _DummyThread(Thread):
 
     def __init__(self):
-        Thread.__init__(self, name=_newname("Dummy-%d"), daemon=True)
-
+        Thread.__init__(self, name=_newname("Dummy-%d"),
+                        daemon=_daemon_threads_allowed())
         self._started.set()
         self._set_ident()
+        self._handle = _make_thread_handle(self._ident)
         if _HAVE_THREAD_NATIVE_ID:
             self._set_native_id()
         with _active_limbo_lock:
             _active[self._ident] = self
-
-    def _stop(self):
-        pass
+        _DeleteDummyThreadOnDel(self)
 
     def is_alive(self):
-        assert not self._is_stopped and self._started.is_set()
-        return True
+        if not self._handle.is_done() and self._started.is_set():
+            return True
+        raise RuntimeError("thread is not alive")
 
     def join(self, timeout=None):
-        assert False, "cannot join a dummy thread"
+        raise RuntimeError("cannot join a dummy thread")
+
+    def _after_fork(self, new_ident=None):
+        if new_ident is not None:
+            self.__class__ = _MainThread
+            self._name = 'MainThread'
+            self._daemonic = False
+        Thread._after_fork(self, new_ident=new_ident)
 
 
 # Global API functions
@@ -1468,6 +1457,8 @@ def active_count():
     enumerate().
 
     """
+    # NOTE: if the logic in here ever changes, update Modules/posixmodule.c
+    # warn_about_fork_with_threads() to match.
     with _active_limbo_lock:
         return len(_active) + len(_limbo)
 
@@ -1514,8 +1505,7 @@ def _register_atexit(func, *arg, **kwargs):
     if _SHUTTING_DOWN:
         raise RuntimeError("can't register atexit after shutdown")
 
-    call = functools.partial(func, *arg, **kwargs)
-    _threading_atexits.append(call)
+    _threading_atexits.append(lambda: func(*arg, **kwargs))
 
 
 from _thread import stack_size
@@ -1530,12 +1520,11 @@ def _shutdown():
     """
     Wait until the Python thread state of all non-daemon threads get deleted.
     """
-    # Obscure:  other threads may be waiting to join _main_thread.  That's
-    # dubious, but some code does it.  We can't wait for C code to release
-    # the main thread's tstate_lock - that won't happen until the interpreter
-    # is nearly dead.  So we release it here.  Note that just calling _stop()
-    # isn't enough:  other threads may already be waiting on _tstate_lock.
-    if _main_thread._is_stopped:
+    # Obscure: other threads may be waiting to join _main_thread.  That's
+    # dubious, but some code does it. We can't wait for it to be marked as done
+    # normally - that won't happen until the interpreter is nearly dead. So
+    # mark it done here.
+    if _main_thread._handle.is_done() and _is_main_interpreter():
         # _shutdown() was already called
         return
 
@@ -1547,39 +1536,11 @@ def _shutdown():
     for atexit_call in reversed(_threading_atexits):
         atexit_call()
 
-    # Main thread
-    if _main_thread.ident == get_ident():
-        tlock = _main_thread._tstate_lock
-        # The main thread isn't finished yet, so its thread state lock can't
-        # have been released.
-        assert tlock is not None
-        assert tlock.locked()
-        tlock.release()
-        _main_thread._stop()
-    else:
-        # bpo-1596321: _shutdown() must be called in the main thread.
-        # If the threading module was not imported by the main thread,
-        # _main_thread is the thread which imported the threading module.
-        # In this case, ignore _main_thread, similar behavior than for threads
-        # spawned by C libraries or using _thread.start_new_thread().
-        pass
+    if _is_main_interpreter():
+        _main_thread._handle._set_done()
 
-    # Join all non-deamon threads
-    while True:
-        with _shutdown_locks_lock:
-            locks = list(_shutdown_locks)
-            _shutdown_locks.clear()
-
-        if not locks:
-            break
-
-        for lock in locks:
-            # mimic Thread.join()
-            lock.acquire()
-            lock.release()
-
-        # new threads can be spawned while we were waiting for the other
-        # threads to complete
+    # Wait for all non-daemon threads to exit.
+    _thread_shutdown()
 
 
 def main_thread():
@@ -1588,16 +1549,9 @@ def main_thread():
     In normal conditions, the main thread is the thread from which the
     Python interpreter was started.
     """
+    # XXX Figure this out for subinterpreters.  (See gh-75698.)
     return _main_thread
 
-# get thread-local implementation, either from the thread
-# module, or from the python fallback
-
-try:
-    from _thread import _local as local
-except ImportError:
-    from _threading_local import local
-
 
 def _after_fork():
     """
@@ -1606,7 +1560,6 @@ def _after_fork():
     # Reset _active_limbo_lock, in case we forked while the lock was held
     # by another (non-forked) thread.  http://bugs.python.org/issue874900
     global _active_limbo_lock, _main_thread
-    global _shutdown_locks_lock, _shutdown_locks
     _active_limbo_lock = RLock()
 
     # fork() only copied the current thread; clear references to others.
@@ -1622,10 +1575,6 @@ def _after_fork():
 
     _main_thread = current
 
-    # reset _shutdown() locks: threads re-register their _tstate_lock below
-    _shutdown_locks_lock = _allocate_lock()
-    _shutdown_locks = set()
-
     with _active_limbo_lock:
         # Dangling thread instances must still have their locks reset,
         # because someone may join() them.
@@ -1635,16 +1584,13 @@ def _after_fork():
             # Any lock/condition variable may be currently locked or in an
             # invalid state, so we reinitialize them.
             if thread is current:
-                # There is only one active thread. We reset the ident to
-                # its new value since it can have changed.
-                thread._reset_internal_locks(True)
+                # This is the one and only active thread.
                 ident = get_ident()
-                thread._ident = ident
+                thread._after_fork(new_ident=ident)
                 new_active[ident] = thread
             else:
                 # All the others are already stopped.
-                thread._reset_internal_locks(False)
-                thread._stop()
+                thread._after_fork()
 
         _limbo.clear()
         _active.clear()

crates/vm/src/builtins/asyncgenerator.rs

@@ -57,14 +57,14 @@ impl PyAsyncGen {
 
         zelf.ag_hooks_inited.store(true);
 
-        // Get and store finalizer from thread-local storage
-        let finalizer = crate::vm::thread::ASYNC_GEN_FINALIZER.with_borrow(|f| f.as_ref().cloned());
+        // Get and store finalizer from VM
+        let finalizer = vm.async_gen_finalizer.borrow().clone();
         if let Some(finalizer) = finalizer {
             *zelf.ag_finalizer.lock() = Some(finalizer);
         }
 
         // Call firstiter hook
-        let firstiter = crate::vm::thread::ASYNC_GEN_FIRSTITER.with_borrow(|f| f.as_ref().cloned());
+        let firstiter = vm.async_gen_firstiter.borrow().clone();
         if let Some(firstiter) = firstiter {
             let obj: PyObjectRef = zelf.to_owned().into();
             firstiter.call((obj,), vm)?;

crates/vm/src/stdlib/posix.rs

@@ -661,6 +661,11 @@ pub mod module {
     }
 
     fn py_os_after_fork_child(vm: &VirtualMachine) {
+        // Mark all other threads as done before running Python callbacks
+        // See _PyThread_AfterFork behavior
+        #[cfg(feature = "threading")]
+        crate::stdlib::thread::after_fork_child(vm);
+
         let after_forkers_child: Vec<PyObjectRef> = vm.state.after_forkers_child.lock().clone();
         run_at_forkers(after_forkers_child, false, vm);
     }
@@ -670,8 +675,64 @@ pub mod module {
         run_at_forkers(after_forkers_parent, false, vm);
     }
 
+    /// Warn if forking from a multi-threaded process
+    fn warn_if_multi_threaded(name: &str, vm: &VirtualMachine) {
+        // Only check threading if it was already imported
+        // Avoid vm.import() which can execute arbitrary Python code in the fork path
+        let threading = match vm
+            .sys_module
+            .get_attr("modules", vm)
+            .and_then(|m| m.get_item("threading", vm))
+        {
+            Ok(m) => m,
+            Err(_) => return,
+        };
+        let active = threading.get_attr("_active", vm).ok();
+        let limbo = threading.get_attr("_limbo", vm).ok();
+
+        let count_dict = |obj: Option<crate::PyObjectRef>| -> usize {
+            obj.and_then(|o| o.length_opt(vm))
+                .and_then(|r| r.ok())
+                .unwrap_or(0)
+        };
+
+        let num_threads = count_dict(active) + count_dict(limbo);
+        if num_threads > 1 {
+            // Use Python warnings module to ensure filters are applied correctly
+            let Ok(warnings) = vm.import("warnings", 0) else {
+                return;
+            };
+            let Ok(warn_fn) = warnings.get_attr("warn", vm) else {
+                return;
+            };
+
+            let pid = unsafe { libc::getpid() };
+            let msg = format!(
+                "This process (pid={}) is multi-threaded, use of {}() may lead to deadlocks in the child.",
+                pid, name
+            );
+
+            // Call warnings.warn(message, DeprecationWarning, stacklevel=2)
+            // stacklevel=2 to point to the caller of fork()
+            let args = crate::function::FuncArgs::new(
+                vec![
+                    vm.ctx.new_str(msg).into(),
+                    vm.ctx.exceptions.deprecation_warning.as_object().to_owned(),
+                ],
+                crate::function::KwArgs::new(
+                    [("stacklevel".to_owned(), vm.ctx.new_int(2).into())]
+                        .into_iter()
+                        .collect(),
+                ),
+            );
+            let _ = warn_fn.call(args, vm);
+        }
+    }
+
     #[pyfunction]
     fn fork(vm: &VirtualMachine) -> i32 {
+        warn_if_multi_threaded("fork", vm);
+
         let pid: i32;
         py_os_before_fork(vm);
         unsafe {

crates/vm/src/stdlib/sys.rs

@@ -815,6 +815,32 @@ mod sys {
         }
     }
 
+    /// Return a dictionary mapping each thread's identifier to the topmost stack frame
+    /// currently active in that thread at the time the function is called.
+    #[cfg(feature = "threading")]
+    #[pyfunction]
+    fn _current_frames(vm: &VirtualMachine) -> PyResult<PyDictRef> {
+        use crate::AsObject;
+        use crate::stdlib::thread::get_all_current_frames;
+
+        let frames = get_all_current_frames(vm);
+        let dict = vm.ctx.new_dict();
+
+        for (thread_id, frame) in frames {
+            let key = vm.ctx.new_int(thread_id);
+            dict.set_item(key.as_object(), frame.into(), vm)?;
+        }
+
+        Ok(dict)
+    }
+
+    /// Stub for non-threading builds - returns empty dict
+    #[cfg(not(feature = "threading"))]
+    #[pyfunction]
+    fn _current_frames(vm: &VirtualMachine) -> PyResult<PyDictRef> {
+        Ok(vm.ctx.new_dict())
+    }
+
     #[pyfunction]
     fn gettrace(vm: &VirtualMachine) -> PyObjectRef {
         vm.trace_func.borrow().clone()
@@ -1107,6 +1133,22 @@ mod sys {
         update_use_tracing(vm);
     }
 
+    #[pyfunction]
+    fn _settraceallthreads(tracefunc: PyObjectRef, vm: &VirtualMachine) {
+        let func = (!vm.is_none(&tracefunc)).then(|| tracefunc.clone());
+        *vm.state.global_trace_func.lock() = func;
+        vm.trace_func.replace(tracefunc);
+        update_use_tracing(vm);
+    }
+
+    #[pyfunction]
+    fn _setprofileallthreads(profilefunc: PyObjectRef, vm: &VirtualMachine) {
+        let func = (!vm.is_none(&profilefunc)).then(|| profilefunc.clone());
+        *vm.state.global_profile_func.lock() = func;
+        vm.profile_func.replace(profilefunc);
+        update_use_tracing(vm);
+    }
+
     #[cfg(feature = "threading")]
     #[pyattr]
     fn thread_info(vm: &VirtualMachine) -> PyTupleRef {
@@ -1187,10 +1229,10 @@ mod sys {
         }
 
         if let Some(finalizer) = args.finalizer.into_option() {
-            crate::vm::thread::ASYNC_GEN_FINALIZER.set(finalizer);
+            *vm.async_gen_finalizer.borrow_mut() = finalizer;
         }
         if let Some(firstiter) = args.firstiter.into_option() {
-            crate::vm::thread::ASYNC_GEN_FIRSTITER.set(firstiter);
+            *vm.async_gen_firstiter.borrow_mut() = firstiter;
         }
 
         Ok(())
@@ -1212,12 +1254,8 @@ mod sys {
     #[pyfunction]
     fn get_asyncgen_hooks(vm: &VirtualMachine) -> AsyncgenHooksData {
         AsyncgenHooksData {
-            firstiter: crate::vm::thread::ASYNC_GEN_FIRSTITER
-                .with_borrow(Clone::clone)
-                .to_pyobject(vm),
-            finalizer: crate::vm::thread::ASYNC_GEN_FINALIZER
-                .with_borrow(Clone::clone)
-                .to_pyobject(vm),
+            firstiter: vm.async_gen_firstiter.borrow().clone().to_pyobject(vm),
+            finalizer: vm.async_gen_finalizer.borrow().clone().to_pyobject(vm),
         }
     }
 

crates/vm/src/stdlib/thread.rs

@@ -1,13 +1,16 @@
 //! Implementation of the _thread module
 #[cfg_attr(target_arch = "wasm32", allow(unused_imports))]
-pub(crate) use _thread::{RawRMutex, make_module};
+pub(crate) use _thread::{
+    CurrentFrameSlot, HandleEntry, RawRMutex, ShutdownEntry, after_fork_child,
+    get_all_current_frames, get_ident, init_main_thread_ident, make_module,
+};
 
 #[pymodule]
 pub(crate) mod _thread {
     use crate::{
         AsObject, Py, PyPayload, PyRef, PyResult, VirtualMachine,
         builtins::{PyDictRef, PyStr, PyTupleRef, PyType, PyTypeRef},
-        convert::ToPyException,
+        frame::FrameRef,
         function::{ArgCallable, Either, FuncArgs, KwArgs, OptionalArg, PySetterValue},
         types::{Constructor, GetAttr, Representable, SetAttr},
     };
@@ -19,7 +22,6 @@ pub(crate) mod _thread {
         lock_api::{RawMutex as RawMutexT, RawMutexTimed, RawReentrantMutex},
     };
     use std::thread;
-    use thread_local::ThreadLocal;
 
     // PYTHREAD_NAME: show current thread name
     pub const PYTHREAD_NAME: Option<&str> = {
@@ -110,7 +112,8 @@ pub(crate) mod _thread {
     }
 
     #[pyattr(name = "LockType")]
-    #[pyclass(module = "thread", name = "lock")]
+    #[pyattr(name = "lock")]
+    #[pyclass(module = "_thread", name = "lock")]
     #[derive(PyPayload)]
     struct Lock {
         mu: RawMutex,
@@ -172,10 +175,10 @@ pub(crate) mod _thread {
     }
 
     impl Constructor for Lock {
-        type Args = FuncArgs;
+        type Args = ();
 
-        fn py_new(_cls: &Py<PyType>, _args: Self::Args, vm: &VirtualMachine) -> PyResult<Self> {
-            Err(vm.new_type_error("cannot create '_thread.lock' instances"))
+        fn py_new(_cls: &Py<PyType>, _args: Self::Args, _vm: &VirtualMachine) -> PyResult<Self> {
+            Ok(Self { mu: RawMutex::INIT })
         }
     }
 
@@ -188,7 +191,7 @@ pub(crate) mod _thread {
 
     pub type RawRMutex = RawReentrantMutex<RawMutex, RawThreadId>;
     #[pyattr]
-    #[pyclass(module = "thread", name = "RLock")]
+    #[pyclass(module = "_thread", name = "RLock")]
     #[derive(PyPayload)]
     struct RLock {
         mu: RawRMutex,
@@ -201,7 +204,7 @@ pub(crate) mod _thread {
         }
     }
 
-    #[pyclass(with(Representable))]
+    #[pyclass(with(Representable), flags(BASETYPE))]
     impl RLock {
         #[pyslot]
         fn slot_new(cls: PyTypeRef, _args: FuncArgs, vm: &VirtualMachine) -> PyResult {
@@ -249,9 +252,10 @@ pub(crate) mod _thread {
             }
             self.count.store(0, core::sync::atomic::Ordering::Relaxed);
             let new_mut = RawRMutex::INIT;
-
-            let old_mutex: AtomicCell<&RawRMutex> = AtomicCell::new(&self.mu);
-            old_mutex.swap(&new_mut);
+            unsafe {
+                let old_mutex: &AtomicCell<RawRMutex> = core::mem::transmute(&self.mu);
+                old_mutex.swap(new_mut);
+            }
 
             Ok(())
         }
@@ -283,12 +287,28 @@ pub(crate) mod _thread {
         }
     }
 
+    /// Get thread identity - uses pthread_self() on Unix for fork compatibility
     #[pyfunction]
-    fn get_ident() -> u64 {
-        thread_to_id(&thread::current())
+    pub fn get_ident() -> u64 {
+        current_thread_id()
     }
 
-    fn thread_to_id(t: &thread::Thread) -> u64 {
+    /// Get OS-level thread ID (pthread_self on Unix)
+    /// This is important for fork compatibility - the ID must remain stable after fork
+    #[cfg(unix)]
+    fn current_thread_id() -> u64 {
+        // pthread_self() like CPython for fork compatibility
+        unsafe { libc::pthread_self() as u64 }
+    }
+
+    #[cfg(not(unix))]
+    fn current_thread_id() -> u64 {
+        thread_to_rust_id(&thread::current())
+    }
+
+    /// Convert Rust thread to ID (used for non-unix platforms)
+    #[cfg(not(unix))]
+    fn thread_to_rust_id(t: &thread::Thread) -> u64 {
         use core::hash::{Hash, Hasher};
         struct U64Hash {
             v: Option<u64>,
@@ -304,13 +324,25 @@ pub(crate) mod _thread {
                 self.v.expect("should have written a u64")
             }
         }
-        // TODO: use id.as_u64() once it's stable, until then, ThreadId is just a wrapper
-        // around NonZeroU64, so this should work (?)
         let mut h = U64Hash { v: None };
         t.id().hash(&mut h);
         h.finish()
     }
 
+    /// Get thread ID for a given thread handle (used by start_new_thread)
+    fn thread_to_id(handle: &thread::JoinHandle<()>) -> u64 {
+        #[cfg(unix)]
+        {
+            // On Unix, use pthread ID from the handle
+            use std::os::unix::thread::JoinHandleExt;
+            handle.as_pthread_t() as u64
+        }
+        #[cfg(not(unix))]
+        {
+            thread_to_rust_id(handle.thread())
+        }
+    }
+
     #[pyfunction]
     const fn allocate_lock() -> Lock {
         Lock { mu: RawMutex::INIT }
@@ -343,9 +375,9 @@ pub(crate) mod _thread {
             )
             .map(|handle| {
                 vm.state.thread_count.fetch_add(1);
-                thread_to_id(handle.thread())
+                thread_to_id(&handle)
             })
-            .map_err(|err| err.to_pyexception(vm))
+            .map_err(|err| vm.new_runtime_error(format!("can't start new thread: {err}")))
     }
 
     fn run_thread(func: ArgCallable, args: FuncArgs, vm: &VirtualMachine) {
@@ -365,9 +397,24 @@ pub(crate) mod _thread {
                 unsafe { lock.mu.unlock() };
             }
         }
+        // Clean up thread-local storage while VM context is still active
+        // This ensures __del__ methods are called properly
+        cleanup_thread_local_data();
+        // Clean up frame tracking
+        crate::vm::thread::cleanup_current_thread_frames(vm);
         vm.state.thread_count.fetch_sub(1);
     }
 
+    /// Clean up thread-local data for the current thread.
+    /// This triggers __del__ on objects stored in thread-local variables.
+    fn cleanup_thread_local_data() {
+        // Take all guards - this will trigger LocalGuard::drop for each,
+        // which removes the thread's dict from each Local instance
+        LOCAL_GUARDS.with(|guards| {
+            guards.borrow_mut().clear();
+        });
+    }
+
     #[cfg(not(target_arch = "wasm32"))]
     #[pyfunction]
     fn interrupt_main(signum: OptionalArg<i32>, vm: &VirtualMachine) -> PyResult<()> {
@@ -400,6 +447,107 @@ pub(crate) mod _thread {
         vm.state.thread_count.load()
     }
 
+    #[pyfunction]
+    fn daemon_threads_allowed() -> bool {
+        // RustPython always allows daemon threads
+        true
+    }
+
+    // Registry for non-daemon threads that need to be joined at shutdown
+    pub type ShutdownEntry = (
+        std::sync::Weak<parking_lot::Mutex<ThreadHandleInner>>,
+        std::sync::Weak<(parking_lot::Mutex<bool>, parking_lot::Condvar)>,
+    );
+
+    #[pyfunction]
+    fn _shutdown(vm: &VirtualMachine) {
+        // Wait for all non-daemon threads to finish
+        let current_ident = get_ident();
+
+        loop {
+            // Find a thread that's not finished and not the current thread
+            let handle_to_join = {
+                let mut handles = vm.state.shutdown_handles.lock();
+                // Clean up finished entries
+                handles.retain(|(inner_weak, _): &ShutdownEntry| {
+                    inner_weak.upgrade().map_or(false, |inner| {
+                        let guard = inner.lock();
+                        guard.state != ThreadHandleState::Done && guard.ident != current_ident
+                    })
+                });
+
+                // Find first unfinished handle
+                handles
+                    .iter()
+                    .find_map(|(inner_weak, done_event_weak): &ShutdownEntry| {
+                        let inner = inner_weak.upgrade()?;
+                        let done_event = done_event_weak.upgrade()?;
+                        let guard = inner.lock();
+                        if guard.state != ThreadHandleState::Done && guard.ident != current_ident {
+                            Some((inner.clone(), done_event.clone()))
+                        } else {
+                            None
+                        }
+                    })
+            };
+
+            match handle_to_join {
+                Some((_, done_event)) => {
+                    // Wait for this thread to finish (infinite timeout)
+                    // Only check done flag to avoid lock ordering issues
+                    // (done_event lock vs inner lock)
+                    let (lock, cvar) = &*done_event;
+                    let mut done = lock.lock();
+                    while !*done {
+                        cvar.wait(&mut done);
+                    }
+                }
+                None => break, // No more threads to wait on
+            }
+        }
+    }
+
+    /// Add a non-daemon thread handle to the shutdown registry
+    fn add_to_shutdown_handles(
+        vm: &VirtualMachine,
+        inner: &std::sync::Arc<parking_lot::Mutex<ThreadHandleInner>>,
+        done_event: &std::sync::Arc<(parking_lot::Mutex<bool>, parking_lot::Condvar)>,
+    ) {
+        let mut handles = vm.state.shutdown_handles.lock();
+        handles.push((
+            std::sync::Arc::downgrade(inner),
+            std::sync::Arc::downgrade(done_event),
+        ));
+    }
+
+    #[pyfunction]
+    fn _make_thread_handle(ident: u64, vm: &VirtualMachine) -> PyRef<ThreadHandle> {
+        let handle = ThreadHandle::new(vm);
+        {
+            let mut inner = handle.inner.lock();
+            inner.ident = ident;
+            inner.state = ThreadHandleState::Running;
+        }
+        handle.into_ref(&vm.ctx)
+    }
+
+    #[pyfunction]
+    fn _get_main_thread_ident(vm: &VirtualMachine) -> u64 {
+        vm.state.main_thread_ident.load()
+    }
+
+    #[pyfunction]
+    fn _is_main_interpreter() -> bool {
+        // RustPython only has one interpreter
+        true
+    }
+
+    /// Initialize the main thread ident. Should be called once at interpreter startup.
+    pub fn init_main_thread_ident(vm: &VirtualMachine) {
+        let ident = get_ident();
+        vm.state.main_thread_ident.store(ident);
+    }
+
     /// ExceptHookArgs - simple class to hold exception hook arguments
     /// This allows threading.py to import _excepthook and _ExceptHookArgs from _thread
     #[pyattr]
@@ -532,23 +680,92 @@ pub(crate) mod _thread {
         Ok(())
     }
 
+    // Thread-local storage for cleanup guards
+    // When a thread terminates, the guard is dropped, which triggers cleanup
+    thread_local! {
+        static LOCAL_GUARDS: std::cell::RefCell<Vec<LocalGuard>> = const { std::cell::RefCell::new(Vec::new()) };
+    }
+
+    // Guard that removes thread-local data when dropped
+    struct LocalGuard {
+        local: std::sync::Weak<LocalData>,
+        thread_id: std::thread::ThreadId,
+    }
+
+    impl Drop for LocalGuard {
+        fn drop(&mut self) {
+            if let Some(local_data) = self.local.upgrade() {
+                // Remove from map while holding the lock, but drop the value
+                // outside the lock to prevent deadlock if __del__ accesses _local
+                let removed = local_data.data.lock().remove(&self.thread_id);
+                drop(removed);
+            }
+        }
+    }
+
+    // Shared data structure for Local
+    struct LocalData {
+        data: parking_lot::Mutex<std::collections::HashMap<std::thread::ThreadId, PyDictRef>>,
+    }
+
+    impl std::fmt::Debug for LocalData {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            f.debug_struct("LocalData").finish_non_exhaustive()
+        }
+    }
+
     #[pyattr]
-    #[pyclass(module = "thread", name = "_local")]
+    #[pyclass(module = "_thread", name = "_local")]
     #[derive(Debug, PyPayload)]
     struct Local {
-        data: ThreadLocal<PyDictRef>,
+        inner: std::sync::Arc<LocalData>,
     }
 
     #[pyclass(with(GetAttr, SetAttr), flags(BASETYPE))]
     impl Local {
         fn l_dict(&self, vm: &VirtualMachine) -> PyDictRef {
-            self.data.get_or(|| vm.ctx.new_dict()).clone()
+            let thread_id = std::thread::current().id();
+
+            // Fast path: check if dict exists under lock
+            if let Some(dict) = self.inner.data.lock().get(&thread_id).cloned() {
+                return dict;
+            }
+
+            // Slow path: allocate dict outside lock to reduce lock hold time
+            let new_dict = vm.ctx.new_dict();
+
+            // Insert with double-check to handle races
+            let mut data = self.inner.data.lock();
+            use std::collections::hash_map::Entry;
+            let (dict, need_guard) = match data.entry(thread_id) {
+                Entry::Occupied(e) => (e.get().clone(), false),
+                Entry::Vacant(e) => {
+                    e.insert(new_dict.clone());
+                    (new_dict, true)
+                }
+            };
+            drop(data); // Release lock before TLS access
+
+            // Register cleanup guard only if we inserted a new entry
+            if need_guard {
+                let guard = LocalGuard {
+                    local: std::sync::Arc::downgrade(&self.inner),
+                    thread_id,
+                };
+                LOCAL_GUARDS.with(|guards| {
+                    guards.borrow_mut().push(guard);
+                });
+            }
+
+            dict
         }
 
         #[pyslot]
         fn slot_new(cls: PyTypeRef, _args: FuncArgs, vm: &VirtualMachine) -> PyResult {
             Self {
-                data: ThreadLocal::new(),
+                inner: std::sync::Arc::new(LocalData {
+                    data: parking_lot::Mutex::new(std::collections::HashMap::new()),
+                }),
             }
             .into_ref_with_type(vm, cls)
             .map(Into::into)
@@ -597,4 +814,354 @@ pub(crate) mod _thread {
             }
         }
     }
+
+    // Registry of all ThreadHandles for fork cleanup
+    // Stores weak references so handles can be garbage collected normally
+    pub type HandleEntry = (
+        std::sync::Weak<parking_lot::Mutex<ThreadHandleInner>>,
+        std::sync::Weak<(parking_lot::Mutex<bool>, parking_lot::Condvar)>,
+    );
+
+    // Re-export type from vm::thread for PyGlobalState
+    pub use crate::vm::thread::CurrentFrameSlot;
+
+    /// Get all threads' current frames. Used by sys._current_frames().
+    pub fn get_all_current_frames(vm: &VirtualMachine) -> Vec<(u64, FrameRef)> {
+        let registry = vm.state.thread_frames.lock();
+        registry
+            .iter()
+            .filter_map(|(id, slot)| slot.lock().clone().map(|f| (*id, f)))
+            .collect()
+    }
+
+    /// Called after fork() in child process to mark all other threads as done.
+    /// This prevents join() from hanging on threads that don't exist in the child.
+    pub fn after_fork_child(vm: &VirtualMachine) {
+        let current_ident = get_ident();
+
+        // Update main thread ident - after fork, the current thread becomes the main thread
+        vm.state.main_thread_ident.store(current_ident);
+
+        // Reinitialize frame slot for current thread
+        crate::vm::thread::reinit_frame_slot_after_fork(vm);
+
+        // Clean up thread handles if we can acquire the lock.
+        // Use try_lock because the mutex might have been held during fork.
+        // If we can't acquire it, just skip - the child process will work
+        // correctly with new handles it creates.
+        if let Some(mut handles) = vm.state.thread_handles.try_lock() {
+            // Clean up dead weak refs and mark non-current threads as done
+            handles.retain(|(inner_weak, done_event_weak): &HandleEntry| {
+                let Some(inner) = inner_weak.upgrade() else {
+                    return false; // Remove dead entries
+                };
+                let Some(done_event) = done_event_weak.upgrade() else {
+                    return false;
+                };
+
+                // Try to lock the inner state - skip if we can't
+                let Some(mut inner_guard) = inner.try_lock() else {
+                    return false;
+                };
+
+                // Skip current thread and not-started threads
+                if inner_guard.ident == current_ident {
+                    return true;
+                }
+                if inner_guard.state == ThreadHandleState::NotStarted {
+                    return true;
+                }
+
+                // Mark as done and notify waiters
+                inner_guard.state = ThreadHandleState::Done;
+                inner_guard.join_handle = None; // Can't join OS thread from child
+                drop(inner_guard);
+
+                // Try to notify waiters - skip if we can't acquire the lock
+                let (lock, cvar) = &*done_event;
+                if let Some(mut done) = lock.try_lock() {
+                    *done = true;
+                    cvar.notify_all();
+                }
+
+                true
+            });
+        }
+    }
+
+    // Thread handle state enum
+    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+    pub enum ThreadHandleState {
+        NotStarted,
+        Starting,
+        Running,
+        Done,
+    }
+
+    // Internal shared state for thread handle
+    pub struct ThreadHandleInner {
+        pub state: ThreadHandleState,
+        pub ident: u64,
+        pub join_handle: Option<thread::JoinHandle<()>>,
+        pub joining: bool, // True if a thread is currently joining
+        pub joined: bool,  // Track if join has completed
+    }
+
+    impl fmt::Debug for ThreadHandleInner {
+        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+            f.debug_struct("ThreadHandleInner")
+                .field("state", &self.state)
+                .field("ident", &self.ident)
+                .field("join_handle", &self.join_handle.is_some())
+                .field("joining", &self.joining)
+                .field("joined", &self.joined)
+                .finish()
+        }
+    }
+
+    /// _ThreadHandle - handle for joinable threads
+    #[pyattr]
+    #[pyclass(module = "_thread", name = "_ThreadHandle")]
+    #[derive(Debug, PyPayload)]
+    struct ThreadHandle {
+        inner: std::sync::Arc<parking_lot::Mutex<ThreadHandleInner>>,
+        // Event to signal thread completion (for timed join support)
+        done_event: std::sync::Arc<(parking_lot::Mutex<bool>, parking_lot::Condvar)>,
+    }
+
+    #[pyclass]
+    impl ThreadHandle {
+        fn new(vm: &VirtualMachine) -> Self {
+            let inner = std::sync::Arc::new(parking_lot::Mutex::new(ThreadHandleInner {
+                state: ThreadHandleState::NotStarted,
+                ident: 0,
+                join_handle: None,
+                joining: false,
+                joined: false,
+            }));
+            let done_event =
+                std::sync::Arc::new((parking_lot::Mutex::new(false), parking_lot::Condvar::new()));
+
+            // Register in global registry for fork cleanup
+            vm.state.thread_handles.lock().push((
+                std::sync::Arc::downgrade(&inner),
+                std::sync::Arc::downgrade(&done_event),
+            ));
+
+            Self { inner, done_event }
+        }
+
+        #[pygetset]
+        fn ident(&self) -> u64 {
+            self.inner.lock().ident
+        }
+
+        #[pymethod]
+        fn is_done(&self) -> bool {
+            self.inner.lock().state == ThreadHandleState::Done
+        }
+
+        #[pymethod]
+        fn _set_done(&self) {
+            self.inner.lock().state = ThreadHandleState::Done;
+            // Signal waiting threads that this thread is done
+            let (lock, cvar) = &*self.done_event;
+            *lock.lock() = true;
+            cvar.notify_all();
+        }
+
+        #[pymethod]
+        fn join(
+            &self,
+            timeout: OptionalArg<Option<Either<f64, i64>>>,
+            vm: &VirtualMachine,
+        ) -> PyResult<()> {
+            // Convert timeout to Duration (None or negative = infinite wait)
+            let timeout_duration = match timeout.flatten() {
+                Some(Either::A(t)) if t >= 0.0 => Some(Duration::from_secs_f64(t)),
+                Some(Either::B(t)) if t >= 0 => Some(Duration::from_secs(t as u64)),
+                _ => None,
+            };
+
+            // Check for self-join first
+            {
+                let inner = self.inner.lock();
+                let current_ident = get_ident();
+                if inner.ident == current_ident && inner.state == ThreadHandleState::Running {
+                    return Err(vm.new_runtime_error("cannot join current thread".to_owned()));
+                }
+            }
+
+            // Wait for thread completion using Condvar (supports timeout)
+            // Loop to handle spurious wakeups
+            let (lock, cvar) = &*self.done_event;
+            let mut done = lock.lock();
+
+            while !*done {
+                if let Some(timeout) = timeout_duration {
+                    let result = cvar.wait_for(&mut done, timeout);
+                    if result.timed_out() && !*done {
+                        // Timeout occurred and done is still false
+                        return Ok(());
+                    }
+                } else {
+                    // Infinite wait
+                    cvar.wait(&mut done);
+                }
+            }
+            drop(done);
+
+            // Thread is done, now perform cleanup
+            let join_handle = {
+                let mut inner = self.inner.lock();
+
+                // If already joined, return immediately (idempotent)
+                if inner.joined {
+                    return Ok(());
+                }
+
+                // If another thread is already joining, wait for them to finish
+                if inner.joining {
+                    drop(inner);
+                    // Wait on done_event
+                    let (lock, cvar) = &*self.done_event;
+                    let mut done = lock.lock();
+                    while !*done {
+                        cvar.wait(&mut done);
+                    }
+                    return Ok(());
+                }
+
+                // Mark that we're joining
+                inner.joining = true;
+
+                // Take the join handle if available
+                inner.join_handle.take()
+            };
+
+            // Perform the actual join outside the lock
+            if let Some(handle) = join_handle {
+                // Ignore the result - panics in spawned threads are already handled
+                let _ = handle.join();
+            }
+
+            // Mark as joined and clear joining flag
+            {
+                let mut inner = self.inner.lock();
+                inner.joined = true;
+                inner.joining = false;
+            }
+
+            Ok(())
+        }
+
+        #[pyslot]
+        fn slot_new(cls: PyTypeRef, _args: FuncArgs, vm: &VirtualMachine) -> PyResult {
+            ThreadHandle::new(vm)
+                .into_ref_with_type(vm, cls)
+                .map(Into::into)
+        }
+    }
+
+    #[derive(FromArgs)]
+    struct StartJoinableThreadArgs {
+        #[pyarg(positional)]
+        function: ArgCallable,
+        #[pyarg(any, optional)]
+        handle: OptionalArg<PyRef<ThreadHandle>>,
+        #[pyarg(any, default = true)]
+        daemon: bool,
+    }
+
+    #[pyfunction]
+    fn start_joinable_thread(
+        args: StartJoinableThreadArgs,
+        vm: &VirtualMachine,
+    ) -> PyResult<PyRef<ThreadHandle>> {
+        let handle = match args.handle {
+            OptionalArg::Present(h) => h,
+            OptionalArg::Missing => ThreadHandle::new(vm).into_ref(&vm.ctx),
+        };
+
+        // Mark as starting
+        handle.inner.lock().state = ThreadHandleState::Starting;
+
+        // Add non-daemon threads to shutdown registry so _shutdown() will wait for them
+        if !args.daemon {
+            add_to_shutdown_handles(vm, &handle.inner, &handle.done_event);
+        }
+
+        let func = args.function;
+        let handle_clone = handle.clone();
+        let inner_clone = handle.inner.clone();
+        let done_event_clone = handle.done_event.clone();
+
+        let mut thread_builder = thread::Builder::new();
+        let stacksize = vm.state.stacksize.load();
+        if stacksize != 0 {
+            thread_builder = thread_builder.stack_size(stacksize);
+        }
+
+        let join_handle = thread_builder
+            .spawn(vm.new_thread().make_spawn_func(move |vm| {
+                // Set ident and mark as running
+                {
+                    let mut inner = inner_clone.lock();
+                    inner.ident = get_ident();
+                    inner.state = ThreadHandleState::Running;
+                }
+
+                // Ensure cleanup happens even if the function panics
+                let inner_for_cleanup = inner_clone.clone();
+                let done_event_for_cleanup = done_event_clone.clone();
+                let vm_state = vm.state.clone();
+                scopeguard::defer! {
+                    // Mark as done
+                    inner_for_cleanup.lock().state = ThreadHandleState::Done;
+
+                    // Signal waiting threads that this thread is done
+                    {
+                        let (lock, cvar) = &*done_event_for_cleanup;
+                        *lock.lock() = true;
+                        cvar.notify_all();
+                    }
+
+                    // Handle sentinels
+                    for lock in SENTINELS.take() {
+                        if lock.mu.is_locked() {
+                            unsafe { lock.mu.unlock() };
+                        }
+                    }
+
+                    // Clean up thread-local data while VM context is still active
+                    cleanup_thread_local_data();
+
+                    // Clean up frame tracking
+                    crate::vm::thread::cleanup_current_thread_frames(vm);
+
+                    vm_state.thread_count.fetch_sub(1);
+                }
+
+                // Run the function
+                match func.invoke((), vm) {
+                    Ok(_) => {}
+                    Err(e) if e.fast_isinstance(vm.ctx.exceptions.system_exit) => {}
+                    Err(exc) => {
+                        vm.run_unraisable(
+                            exc,
+                            Some("Exception ignored in thread started by".to_owned()),
+                            func.into(),
+                        );
+                    }
+                }
+            }))
+            .map_err(|err| vm.new_runtime_error(format!("can't start new thread: {err}")))?;
+
+        vm.state.thread_count.fetch_add(1);
+
+        // Store the join handle
+        handle.inner.lock().join_handle = Some(join_handle);
+
+        Ok(handle_clone)
+    }
 }

crates/vm/src/vm/interpreter.rs

@@ -110,9 +110,10 @@ impl Interpreter {
 
     /// Finalize vm and turns an exception to exit code.
     ///
-    /// Finalization steps including 4 steps:
+    /// Finalization steps including 5 steps:
     /// 1. Flush stdout and stderr.
     /// 1. Handle exit exception and turn it to exit code.
+    /// 1. Wait for non-daemon threads (threading._shutdown).
     /// 1. Run atexit exit functions.
     /// 1. Mark vm as finalized.
     ///
@@ -128,6 +129,9 @@ impl Interpreter {
                 0
             };
 
+            // Wait for non-daemon threads (wait_for_thread_shutdown)
+            wait_for_thread_shutdown(vm);
+
             atexit::_run_exitfuncs(vm);
 
             vm.state.finalizing.store(true, Ordering::Release);
@@ -139,6 +143,35 @@ impl Interpreter {
     }
 }
 
+/// Wait until threading._shutdown completes, provided
+/// the threading module was imported in the first place.
+/// The shutdown routine will wait until all non-daemon
+/// "threading" threads have completed.
+fn wait_for_thread_shutdown(vm: &VirtualMachine) {
+    // Try to get the threading module if it was already imported
+    // Use sys.modules.get("threading") like PyImport_GetModule
+    let threading = match (|| -> PyResult<_> {
+        let sys_modules = vm.sys_module.get_attr("modules", vm)?;
+        let threading = sys_modules.get_item("threading", vm)?;
+        Ok(threading)
+    })() {
+        Ok(module) => module,
+        Err(_) => {
+            // threading not imported, nothing to do
+            return;
+        }
+    };
+
+    // Call threading._shutdown()
+    if let Err(e) = vm.call_method(&threading, "_shutdown", ()) {
+        vm.run_unraisable(
+            e,
+            Some("Exception ignored on threading shutdown".to_owned()),
+            threading,
+        );
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/vm/src/vm/mod.rs

@@ -82,6 +82,10 @@ pub struct VirtualMachine {
     pub state: PyRc<PyGlobalState>,
     pub initialized: bool,
     recursion_depth: Cell<usize>,
+    /// Async generator firstiter hook (per-thread, set via sys.set_asyncgen_hooks)
+    pub async_gen_firstiter: RefCell<Option<PyObjectRef>>,
+    /// Async generator finalizer hook (per-thread, set via sys.set_asyncgen_hooks)
+    pub async_gen_finalizer: RefCell<Option<PyObjectRef>>,
 }
 
 #[derive(Debug, Default)]
@@ -107,6 +111,22 @@ pub struct PyGlobalState {
     pub after_forkers_parent: PyMutex<Vec<PyObjectRef>>,
     pub int_max_str_digits: AtomicCell<usize>,
     pub switch_interval: AtomicCell<f64>,
+    /// Global trace function for all threads (set by sys._settraceallthreads)
+    pub global_trace_func: PyMutex<Option<PyObjectRef>>,
+    /// Global profile function for all threads (set by sys._setprofileallthreads)
+    pub global_profile_func: PyMutex<Option<PyObjectRef>>,
+    /// Main thread identifier (pthread_self on Unix)
+    #[cfg(feature = "threading")]
+    pub main_thread_ident: AtomicCell<u64>,
+    /// Registry of all threads' current frames for sys._current_frames()
+    #[cfg(feature = "threading")]
+    pub thread_frames: parking_lot::Mutex<HashMap<u64, stdlib::thread::CurrentFrameSlot>>,
+    /// Registry of all ThreadHandles for fork cleanup
+    #[cfg(feature = "threading")]
+    pub thread_handles: parking_lot::Mutex<Vec<stdlib::thread::HandleEntry>>,
+    /// Registry for non-daemon threads that need to be joined at shutdown
+    #[cfg(feature = "threading")]
+    pub shutdown_handles: parking_lot::Mutex<Vec<stdlib::thread::ShutdownEntry>>,
 }
 
 pub fn process_hash_secret_seed() -> u32 {
@@ -191,9 +211,21 @@ impl VirtualMachine {
                 after_forkers_parent: PyMutex::default(),
                 int_max_str_digits,
                 switch_interval: AtomicCell::new(0.005),
+                global_trace_func: PyMutex::default(),
+                global_profile_func: PyMutex::default(),
+                #[cfg(feature = "threading")]
+                main_thread_ident: AtomicCell::new(0),
+                #[cfg(feature = "threading")]
+                thread_frames: parking_lot::Mutex::new(HashMap::new()),
+                #[cfg(feature = "threading")]
+                thread_handles: parking_lot::Mutex::new(Vec::new()),
+                #[cfg(feature = "threading")]
+                shutdown_handles: parking_lot::Mutex::new(Vec::new()),
             }),
             initialized: false,
             recursion_depth: Cell::new(0),
+            async_gen_firstiter: RefCell::new(None),
+            async_gen_finalizer: RefCell::new(None),
         };
 
         if vm.state.hash_secret.hash_str("")
@@ -299,6 +331,10 @@ impl VirtualMachine {
             panic!("Double Initialize Error");
         }
 
+        // Initialize main thread ident before any threading operations
+        #[cfg(feature = "threading")]
+        stdlib::thread::init_main_thread_ident(self);
+
         stdlib::builtins::init_module(self, &self.builtins);
         stdlib::sys::init_module(self, &self.sys_module, &self.builtins);
 
@@ -604,6 +640,9 @@ impl VirtualMachine {
     ) -> PyResult<R> {
         self.with_recursion("", || {
             self.frames.borrow_mut().push(frame.clone());
+            // Update the current frame slot for sys._current_frames()
+            #[cfg(feature = "threading")]
+            crate::vm::thread::update_current_frame(Some(frame.clone()));
             // Push a new exception context for frame isolation
             // Each frame starts with no active exception (None)
             // This prevents exceptions from leaking between function calls
@@ -613,6 +652,9 @@ impl VirtualMachine {
             self.pop_exception();
             // defer dec frame
             let _popped = self.frames.borrow_mut().pop();
+            // Update the frame slot to the new top frame (or None if empty)
+            #[cfg(feature = "threading")]
+            crate::vm::thread::update_current_frame(self.frames.borrow().last().cloned());
             result
         })
     }

crates/vm/src/vm/thread.rs

@@ -1,17 +1,27 @@
-use crate::{AsObject, PyObject, PyObjectRef, VirtualMachine};
+#[cfg(feature = "threading")]
+use crate::frame::FrameRef;
+use crate::{AsObject, PyObject, VirtualMachine};
 use core::{
     cell::{Cell, RefCell},
     ptr::NonNull,
 };
 use itertools::Itertools;
+#[cfg(feature = "threading")]
+use std::sync::Arc;
 use std::thread_local;
 
+/// Type for current frame slot - shared between threads for sys._current_frames()
+#[cfg(feature = "threading")]
+pub type CurrentFrameSlot = Arc<parking_lot::Mutex<Option<FrameRef>>>;
+
 thread_local! {
     pub(super) static VM_STACK: RefCell<Vec<NonNull<VirtualMachine>>> = Vec::with_capacity(1).into();
 
     pub(crate) static COROUTINE_ORIGIN_TRACKING_DEPTH: Cell<u32> = const { Cell::new(0) };
-    pub(crate) static ASYNC_GEN_FINALIZER: RefCell<Option<PyObjectRef>> = const { RefCell::new(None) };
-    pub(crate) static ASYNC_GEN_FIRSTITER: RefCell<Option<PyObjectRef>> = const { RefCell::new(None) };
+
+    /// Current thread's frame slot for sys._current_frames()
+    #[cfg(feature = "threading")]
+    static CURRENT_FRAME_SLOT: RefCell<Option<CurrentFrameSlot>> = const { RefCell::new(None) };
 }
 
 scoped_tls::scoped_thread_local!(static VM_CURRENT: VirtualMachine);
@@ -26,11 +36,74 @@ pub fn with_current_vm<R>(f: impl FnOnce(&VirtualMachine) -> R) -> R {
 pub fn enter_vm<R>(vm: &VirtualMachine, f: impl FnOnce() -> R) -> R {
     VM_STACK.with(|vms| {
         vms.borrow_mut().push(vm.into());
+
+        // Initialize frame slot for this thread if not already done
+        #[cfg(feature = "threading")]
+        init_frame_slot_if_needed(vm);
+
         scopeguard::defer! { vms.borrow_mut().pop(); }
         VM_CURRENT.set(vm, f)
     })
 }
 
+/// Initialize frame slot for current thread if not already initialized.
+/// Called automatically by enter_vm().
+#[cfg(feature = "threading")]
+fn init_frame_slot_if_needed(vm: &VirtualMachine) {
+    CURRENT_FRAME_SLOT.with(|slot| {
+        if slot.borrow().is_none() {
+            let thread_id = crate::stdlib::thread::get_ident();
+            let new_slot = Arc::new(parking_lot::Mutex::new(None));
+            vm.state
+                .thread_frames
+                .lock()
+                .insert(thread_id, new_slot.clone());
+            *slot.borrow_mut() = Some(new_slot);
+        }
+    });
+}
+
+/// Update the current thread's frame. Called when frames are pushed/popped.
+/// This is a hot path - uses only thread-local storage, no locks.
+#[cfg(feature = "threading")]
+pub fn update_current_frame(frame: Option<FrameRef>) {
+    CURRENT_FRAME_SLOT.with(|slot| {
+        if let Some(s) = slot.borrow().as_ref() {
+            *s.lock() = frame;
+        }
+    });
+}
+
+/// Cleanup frame tracking for the current thread. Called at thread exit.
+#[cfg(feature = "threading")]
+pub fn cleanup_current_thread_frames(vm: &VirtualMachine) {
+    let thread_id = crate::stdlib::thread::get_ident();
+    vm.state.thread_frames.lock().remove(&thread_id);
+    CURRENT_FRAME_SLOT.with(|s| {
+        *s.borrow_mut() = None;
+    });
+}
+
+/// Reinitialize frame slot after fork. Called in child process.
+/// Creates a fresh slot and registers it for the current thread.
+#[cfg(feature = "threading")]
+pub fn reinit_frame_slot_after_fork(vm: &VirtualMachine) {
+    let current_ident = crate::stdlib::thread::get_ident();
+    let new_slot = Arc::new(parking_lot::Mutex::new(None));
+
+    // Try to update the global registry. If we can't get the lock
+    // (parent thread might have been holding it during fork), skip.
+    if let Some(mut registry) = vm.state.thread_frames.try_lock() {
+        registry.clear();
+        registry.insert(current_ident, new_slot.clone());
+    }
+
+    // Always update thread-local to point to the new slot
+    CURRENT_FRAME_SLOT.with(|s| {
+        *s.borrow_mut() = Some(new_slot);
+    });
+}
+
 pub fn with_vm<F, R>(obj: &PyObject, f: F) -> Option<R>
 where
     F: Fn(&VirtualMachine) -> R,
@@ -139,6 +212,10 @@ impl VirtualMachine {
     /// specific guaranteed behavior.
     #[cfg(feature = "threading")]
     pub fn new_thread(&self) -> ThreadedVirtualMachine {
+        let global_trace = self.state.global_trace_func.lock().clone();
+        let global_profile = self.state.global_profile_func.lock().clone();
+        let use_tracing = global_trace.is_some() || global_profile.is_some();
+
         let vm = Self {
             builtins: self.builtins.clone(),
             sys_module: self.sys_module.clone(),
@@ -147,9 +224,9 @@ impl VirtualMachine {
             wasm_id: self.wasm_id.clone(),
             exceptions: RefCell::default(),
             import_func: self.import_func.clone(),
-            profile_func: RefCell::new(self.ctx.none()),
-            trace_func: RefCell::new(self.ctx.none()),
-            use_tracing: Cell::new(false),
+            profile_func: RefCell::new(global_profile.unwrap_or_else(|| self.ctx.none())),
+            trace_func: RefCell::new(global_trace.unwrap_or_else(|| self.ctx.none())),
+            use_tracing: Cell::new(use_tracing),
             recursion_limit: self.recursion_limit.clone(),
             signal_handlers: None,
             signal_rx: None,
@@ -157,6 +234,8 @@ impl VirtualMachine {
             state: self.state.clone(),
             initialized: self.initialized,
             recursion_depth: Cell::new(0),
+            async_gen_firstiter: RefCell::new(None),
+            async_gen_finalizer: RefCell::new(None),
         };
         ThreadedVirtualMachine { vm }
     }