diff --git a/Lib/_threading_local.py b/Lib/_threading_local.py
index 0b9e5d3bb..2af388545 100644
--- a/Lib/_threading_local.py
+++ b/Lib/_threading_local.py
@@ -57,7 +57,7 @@ class _localimpl:
             # as soon as the OS-level thread ends instead.
             local = wrlocal()
             if local is not None:
-                dct = local.dicts.pop(idt)
+                local.dicts.pop(idt)
         wrlocal = ref(self, local_deleted)
         wrthread = ref(thread, thread_deleted)
         thread.__dict__[key] = wrlocal
diff --git a/Lib/test/lock_tests.py b/Lib/test/lock_tests.py
index fb11f4828..691029a1a 100644
--- a/Lib/test/lock_tests.py
+++ b/Lib/test/lock_tests.py
@@ -124,6 +124,11 @@ class BaseLockTests(BaseTestCase):
         lock = self.locktype()
         del lock
 
+    def test_constructor_noargs(self):
+        self.assertRaises(TypeError, self.locktype, 1)
+        self.assertRaises(TypeError, self.locktype, x=1)
+        self.assertRaises(TypeError, self.locktype, 1, x=2)
+
     def test_repr(self):
         lock = self.locktype()
         self.assertRegex(repr(lock), "<unlocked .* object (.*)?at .*>")
diff --git a/Lib/test/test_thread_local_bytecode.py b/Lib/test/test_thread_local_bytecode.py
index d5c56db8d..6f76f1bc3 100644
--- a/Lib/test/test_thread_local_bytecode.py
+++ b/Lib/test/test_thread_local_bytecode.py
@@ -3,7 +3,7 @@ import textwrap
 import unittest
 
 from test import support
-from test.support import cpython_only, import_helper, requires_specialization_ft
+from test.support import cpython_only, import_helper, requires_specialization
 from test.support.script_helper import assert_python_ok
 from test.support.threading_helper import requires_working_threading
 
@@ -15,7 +15,7 @@ _testinternalcapi = import_helper.import_module("_testinternalcapi")
 @requires_working_threading()
 @unittest.skipUnless(support.Py_GIL_DISABLED, "only in free-threaded builds")
 class TLBCTests(unittest.TestCase):
-    @requires_specialization_ft
+    @requires_specialization
     def test_new_threads_start_with_unspecialized_code(self):
         code = textwrap.dedent("""
         import dis
@@ -46,7 +46,7 @@ class TLBCTests(unittest.TestCase):
         """)
         assert_python_ok("-X", "tlbc=1", "-c", code)
 
-    @requires_specialization_ft
+    @requires_specialization
     def test_threads_specialize_independently(self):
         code = textwrap.dedent("""
         import dis
@@ -108,7 +108,6 @@ class TLBCTests(unittest.TestCase):
         """)
         assert_python_ok("-X", "tlbc=1", "-c", code)
 
-    @support.skip_if_sanitizer("gh-129752: data race on adaptive counter", thread=True)
     def test_no_copies_if_tlbc_disabled(self):
         code = textwrap.dedent("""
         import queue
diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py
index 605dffa52..b85fddb30 100644
--- a/Lib/test/test_threading.py
+++ b/Lib/test/test_threading.py
@@ -6,7 +6,7 @@ import test.support
 from test.support import threading_helper, requires_subprocess, requires_gil_enabled
 from test.support import verbose, cpython_only, os_helper
 from test.support.import_helper import ensure_lazy_imports, import_module
-from test.support.script_helper import assert_python_ok, assert_python_failure
+from test.support.script_helper import assert_python_ok, assert_python_failure, spawn_python
 from test.support import force_not_colorized
 
 import random
@@ -324,6 +324,7 @@ class ThreadTests(BaseTestCase):
     # PyThreadState_SetAsyncExc() is a CPython-only gimmick, not (currently)
     # exposed at the Python level.  This test relies on ctypes to get at it.
     @unittest.expectedFailure  # TODO: RUSTPYTHON
+    @cpython_only
     def test_PyThreadState_SetAsyncExc(self):
         ctypes = import_module("ctypes")
 
@@ -412,6 +413,53 @@ class ThreadTests(BaseTestCase):
             t.join()
         # else the thread is still running, and we have no way to kill it
 
+    @cpython_only
+    @unittest.skipUnless(hasattr(signal, "pthread_kill"), "need pthread_kill")
+    @unittest.skipUnless(hasattr(signal, "SIGUSR1"), "need SIGUSR1")
+    def test_PyThreadState_SetAsyncExc_interrupts_sleep(self):
+        _testcapi = import_module("_testlimitedcapi")
+
+        worker_started = threading.Event()
+
+        class InjectedException(Exception):
+            """Custom exception for testing"""
+
+        caught_exception = None
+
+        def catch_exception():
+            nonlocal caught_exception
+            day_as_seconds = 60 * 60 * 24
+            try:
+                worker_started.set()
+                time.sleep(day_as_seconds)
+            except InjectedException as exc:
+                caught_exception = exc
+
+        thread = threading.Thread(target=catch_exception)
+        thread.start()
+        worker_started.wait()
+
+        signal.signal(signal.SIGUSR1, lambda sig, frame: None)
+
+        result = _testcapi.threadstate_set_async_exc(
+            thread.ident, InjectedException)
+        self.assertEqual(result, 1)
+
+        for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+            if not thread.is_alive():
+                break
+            try:
+                signal.pthread_kill(thread.ident, signal.SIGUSR1)
+            except OSError:
+                # The thread might have terminated between the is_alive check
+                # and the pthread_kill
+                break
+
+        thread.join()
+        signal.signal(signal.SIGUSR1, signal.SIG_DFL)
+
+        self.assertIsInstance(caught_exception, InjectedException)
+
     @unittest.skip("TODO: RUSTPYTHON; threading._start_new_thread not exposed")
     def test_limbo_cleanup(self):
         # Issue 7481: Failure to start thread should cleanup the limbo map.
@@ -1250,13 +1298,68 @@ class ThreadTests(BaseTestCase):
         self.assertEqual(err, b"")
         self.assertIn(b"all clear", out)
 
+    @support.subTests('lock_class_name', ['Lock', 'RLock'])
+    def test_acquire_daemon_thread_lock_in_finalization(self, lock_class_name):
+        # gh-123940: Py_Finalize() prevents other threads from running Python
+        # code (and so, releasing locks), so acquiring a locked lock can not
+        # succeed.
+        # We raise an exception rather than hang.
+        code = textwrap.dedent(f"""
+            import threading
+            import time
+
+            thread_started_event = threading.Event()
+
+            lock = threading.{lock_class_name}()
+            def loop():
+                if {lock_class_name!r} == 'RLock':
+                    lock.acquire()
+                with lock:
+                    thread_started_event.set()
+                    while True:
+                        time.sleep(1)
+
+            uncontested_lock = threading.{lock_class_name}()
+
+            class Cycle:
+                def __init__(self):
+                    self.self_ref = self
+                    self.thr = threading.Thread(
+                        target=loop, daemon=True)
+                    self.thr.start()
+                    thread_started_event.wait()
+
+                def __del__(self):
+                    assert self.thr.is_alive()
+
+                    # We *can* acquire an unlocked lock
+                    uncontested_lock.acquire()
+                    if {lock_class_name!r} == 'RLock':
+                        uncontested_lock.acquire()
+
+                    # Acquiring a locked one fails
+                    try:
+                        lock.acquire()
+                    except PythonFinalizationError:
+                        assert self.thr.is_alive()
+                        print('got the correct exception!')
+
+            # Cycle holds a reference to itself, which ensures it is
+            # cleaned up during the GC that runs after daemon threads
+            # have been forced to exit during finalization.
+            Cycle()
+        """)
+        rc, out, err = assert_python_ok("-c", code)
+        self.assertEqual(err, b"")
+        self.assertIn(b"got the correct exception", out)
+
     def test_start_new_thread_failed(self):
         # gh-109746: if Python fails to start newly created thread
         # due to failure of underlying PyThread_start_new_thread() call,
         # its state should be removed from interpreter' thread states list
         # to avoid its double cleanup
         try:
-            from resource import setrlimit, RLIMIT_NPROC
+            from resource import setrlimit, RLIMIT_NPROC  # noqa: F401
         except ImportError as err:
             self.skipTest(err)  # RLIMIT_NPROC is specific to Linux and BSD
         code = """if 1:
@@ -1287,12 +1390,6 @@ class ThreadTests(BaseTestCase):
 
     @cpython_only
     def test_finalize_daemon_thread_hang(self):
-        if support.check_sanitizer(thread=True, memory=True):
-            # the thread running `time.sleep(100)` below will still be alive
-            # at process exit
-            self.skipTest(
-                    "https://github.com/python/cpython/issues/124878 - Known"
-                    " race condition that TSAN identifies.")
         # gh-87135: tests that daemon threads hang during finalization
         script = textwrap.dedent('''
             import os
@@ -1411,10 +1508,6 @@ class ThreadTests(BaseTestCase):
         self.assertEqual(err, b"")
         self.assertEqual(out.strip(), b"Exiting...")
 
-    @unittest.skip("TODO: RUSTPYTHON; Thread._tstate_lock not implemented")
-    def test_tstate_lock(self):
-        return super().test_tstate_lock()
-
 class ThreadJoinOnShutdown(BaseTestCase):
 
     def _run_and_join(self, script):
@@ -2044,6 +2137,32 @@ class ThreadingExceptionTests(BaseTestCase):
         self.assertEqual(out, b"")
         self.assertEqual(err, b"")
 
+    @requires_subprocess()
+    @unittest.skipIf(os.name == 'nt', "signals don't work well on windows")
+    def test_keyboard_interrupt_during_threading_shutdown(self):
+        import subprocess
+        source = f"""
+        from threading import Thread
+        import time
+        import os
+
+
+        def test():
+            print('a', flush=True, end='')
+            time.sleep(10)
+
+
+        for _ in range(3):
+            Thread(target=test).start()
+        """
+
+        with spawn_python("-c", source, stderr=subprocess.PIPE) as proc:
+            self.assertEqual(proc.stdout.read(3), b'aaa')
+            proc.send_signal(signal.SIGINT)
+            proc.stderr.flush()
+            error = proc.stderr.read()
+            self.assertIn(b"KeyboardInterrupt", error)
+
 
 class ThreadRunFail(threading.Thread):
     def run(self):
@@ -2217,8 +2336,7 @@ class CRLockTests(lock_tests.RLockTests):
         ]
         for args, kwargs in arg_types:
             with self.subTest(args=args, kwargs=kwargs):
-                with self.assertWarns(DeprecationWarning):
-                    threading.RLock(*args, **kwargs)
+                self.assertRaises(TypeError, threading.RLock, *args, **kwargs)
 
         # Subtypes with custom `__init__` are allowed (but, not recommended):
         class CustomRLock(self.locktype):
@@ -2236,6 +2354,9 @@ class ConditionAsRLockTests(lock_tests.RLockTests):
     # Condition uses an RLock by default and exports its API.
     locktype = staticmethod(threading.Condition)
 
+    def test_constructor_noargs(self):
+        self.skipTest("Condition allows positional arguments")
+
     def test_recursion_count(self):
         self.skipTest("Condition does not expose _recursion_count()")
 
@@ -2252,6 +2373,231 @@ class BarrierTests(lock_tests.BarrierTests):
     barriertype = staticmethod(threading.Barrier)
 
 
+## Test Synchronization tools for iterators ################
+
+class ThreadingIteratorToolsTests(BaseTestCase):
+    def test_serialize_serializes_concurrent_iteration(self):
+        limit = 10_000
+        workers_count = 10
+        result = 0
+        result_lock = threading.Lock()
+        start = threading.Event()
+
+        def producer(limit):
+            for x in range(limit):
+                yield x
+
+        def consumer(iterator):
+            nonlocal result
+            start.wait()
+            total = 0
+            for x in iterator:
+                total += x
+            with result_lock:
+                result += total
+
+        iterator = threading.serialize_iterator(producer(limit))
+        workers = [
+            threading.Thread(target=consumer, args=(iterator,))
+            for _ in range(workers_count)
+        ]
+        with threading_helper.wait_threads_exit():
+            for worker in workers:
+                worker.start()
+            for worker in workers:
+                # Wait for the worker thread to actually start.
+                while worker.ident is None:
+                    time.sleep(0.1)
+            start.set()
+            for worker in workers:
+                worker.join()
+
+        self.assertEqual(result, limit * (limit - 1) // 2)
+
+    def test_serialize_generator_methods(self):
+        # A generator that yields and receives
+        def echo():
+            try:
+                while True:
+                    val = yield "ready"
+                    yield f"received {val}"
+            except ValueError:
+                yield "caught"
+
+        it = threading.serialize_iterator(echo())
+
+        # Test __next__
+        self.assertEqual(next(it), "ready")
+
+        # Test send()
+        self.assertEqual(it.send("hello"), "received hello")
+        self.assertEqual(next(it), "ready")
+
+        # Test throw()
+        self.assertEqual(it.throw(ValueError), "caught")
+
+        # Test close()
+        it.close()
+        with self.assertRaises(StopIteration):
+            next(it)
+
+    def test_serialize_methods_attribute_error(self):
+        # A standard iterator that does not have send/throw/close
+        # should raise AttributeError when called.
+        standard_it = threading.serialize_iterator([1, 2, 3])
+
+        with self.assertRaises(AttributeError):
+            standard_it.send("foo")
+
+        with self.assertRaises(AttributeError):
+            standard_it.throw(ValueError)
+
+        with self.assertRaises(AttributeError):
+            standard_it.close()
+
+    def test_serialize_generator_methods_locking(self):
+        # Verifies that generator methods also acquire the lock.
+        # We can test this by checking if the lock is held during the call.
+
+        class LockCheckingGenerator:
+            def __init__(self, lock):
+                self.lock = lock
+            def __iter__(self):
+                return self
+            def send(self, value):
+                if not self.lock.locked():
+                    raise RuntimeError("Lock not held during send()")
+                return value
+            def throw(self, *args):
+                if not self.lock.locked():
+                    raise RuntimeError("Lock not held during throw()")
+            def close(self):
+                if not self.lock.locked():
+                    raise RuntimeError("Lock not held during close()")
+
+        # Manually create the serialize object to inspect the lock
+        it = threading.serialize_iterator([])
+        mock_gen = LockCheckingGenerator(it._lock)
+        it._iterator = mock_gen
+
+        # These should not raise RuntimeError
+        it.send(1)
+        it.throw(ValueError)
+        it.close()
+
+    def test_serialize_next_exception(self):
+        # Verify exception pass through for calls to next()
+
+        def f():
+            raise RuntimeError
+            yield None
+
+        g = threading.serialize_iterator(f())
+        with self.assertRaises(RuntimeError):
+            next(g)
+
+    def test_synchronized_serializes_generator_instances(self):
+        unique = 10
+        repetitions = 5
+        limit = 100
+        start = threading.Event()
+
+        @threading.synchronized_iterator
+        def atomic_counter():
+            # The sleep widens the race window that would exist without
+            # synchronization between yielding a value and advancing state.
+            i = 0
+            while True:
+                yield i
+                time.sleep(0.0005)
+                i += 1
+
+        def consumer(counter):
+            start.wait()
+            for _ in range(limit):
+                next(counter)
+
+        unique_counters = [atomic_counter() for _ in range(unique)]
+        counters = unique_counters * repetitions
+        workers = [
+            threading.Thread(target=consumer, args=(counter,))
+            for counter in counters
+        ]
+        with threading_helper.wait_threads_exit():
+            for worker in workers:
+                worker.start()
+            start.set()
+            for worker in workers:
+                worker.join()
+
+        self.assertEqual(
+            {next(counter) for counter in unique_counters},
+            {limit * repetitions},
+        )
+
+    def test_synchronized_preserves_wrapped_metadata(self):
+        def gen():
+            yield 1
+
+        wrapped = threading.synchronized_iterator(gen)
+
+        self.assertEqual(wrapped.__name__, gen.__name__)
+        self.assertIs(wrapped.__wrapped__, gen)
+        self.assertEqual(list(wrapped()), [1])
+
+    def test_concurrent_tee_supports_concurrent_consumers(self):
+        limit = 5_000
+        num_threads = 25
+        successes = 0
+        failures = []
+        result_lock = threading.Lock()
+        start = threading.Event()
+        expected = list(range(limit))
+
+        def producer(limit):
+            for x in range(limit):
+                yield x
+
+        def consumer(iterator):
+            nonlocal successes
+            start.wait()
+            items = list(iterator)
+            with result_lock:
+                if items == expected:
+                    successes += 1
+                else:
+                    failures.append(items[:20])
+
+        tees = threading.concurrent_tee(producer(limit), n=num_threads)
+        workers = [
+            threading.Thread(target=consumer, args=(iterator,))
+            for iterator in tees
+        ]
+        with threading_helper.wait_threads_exit():
+            for worker in workers:
+                worker.start()
+            start.set()
+            for worker in workers:
+                worker.join()
+
+        self.assertEqual(failures, [])
+        self.assertEqual(successes, len(tees))
+
+        # Verify that locks are shared
+        self.assertEqual(len({id(t_obj.lock) for t_obj in tees}), 1)
+
+    def test_concurrent_tee_zero_iterators(self):
+        self.assertEqual(threading.concurrent_tee(range(10), n=0), ())
+
+    def test_concurrent_tee_negative_n(self):
+        with self.assertRaises(ValueError):
+            threading.concurrent_tee(range(10), n=-1)
+
+
+#################
+
+
+
 class MiscTestCase(unittest.TestCase):
     def test__all__(self):
         restore_default_excepthook(self)
@@ -2485,6 +2831,7 @@ class AtexitTests(unittest.TestCase):
 
         self.assertFalse(err)
 
+    @force_not_colorized
     def test_atexit_after_shutdown(self):
         # The only way to do this is by registering an atexit within
         # an atexit, which is intended to raise an exception.
diff --git a/Lib/threading.py b/Lib/threading.py
index c03b0b537..abac31e25 100644
--- a/Lib/threading.py
+++ b/Lib/threading.py
@@ -29,6 +29,7 @@ __all__ = ['get_ident', 'active_count', 'Condition', 'current_thread',
            'Barrier', 'BrokenBarrierError', 'Timer', 'ThreadError',
            'setprofile', 'settrace', 'local', 'stack_size',
            'excepthook', 'ExceptHookArgs', 'gettrace', 'getprofile',
+           'serialize_iterator', 'synchronized_iterator', 'concurrent_tee',
            'setprofile_all_threads','settrace_all_threads']
 
 # Rename some stuff so "from threading import *" is safe
@@ -123,7 +124,7 @@ def gettrace():
 
 Lock = _LockType
 
-def RLock(*args, **kwargs):
+def RLock():
     """Factory function that returns a new reentrant lock.
 
     A reentrant lock must be released by the thread that acquired it. Once a
@@ -132,16 +133,9 @@ def RLock(*args, **kwargs):
     acquired it.
 
     """
-    if args or kwargs:
-        import warnings
-        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)
+        return _PyRLock()
+    return _CRLock()
 
 class _RLock:
     """This class implements reentrant lock objects.
@@ -849,6 +843,148 @@ class BrokenBarrierError(RuntimeError):
     pass
 
 
+## Synchronization tools for iterators #####################
+
+class serialize_iterator:
+    """Wrap a non-concurrent iterator with a lock to enforce sequential access.
+
+    Applies a non-reentrant lock around calls to __next__.  If the
+    wrapped iterator also defines send(), throw(), or close(), those
+    calls are serialized as well.
+
+    Allows iterator and generator instances to be shared by multiple consumer
+    threads.
+
+    For example, itertools.count does not make thread-safe instances,
+    but that is easily fixed with:
+
+        atomic_counter = serialize_iterator(itertools.count())
+
+    """
+
+    __slots__ = ('_iterator', '_lock')
+
+    def __init__(self, iterable):
+        self._iterator = iter(iterable)
+        self._lock = Lock()
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        with self._lock:
+            return next(self._iterator)
+
+    def send(self, value, /):
+        """Send a value to a generator.
+
+        Raises AttributeError if not a generator.
+        """
+        with self._lock:
+            return self._iterator.send(value)
+
+    def throw(self, *args):
+        """Call throw() on a generator.
+
+        Raises AttributeError if not a generator.
+        """
+        with self._lock:
+            return self._iterator.throw(*args)
+
+    def close(self):
+        """Call close() on a generator.
+
+        Raises AttributeError if not a generator.
+        """
+        with self._lock:
+            return self._iterator.close()
+
+
+def synchronized_iterator(func):
+    """Wrap an iterator-returning callable to make its iterators thread-safe.
+
+    Existing itertools and more-itertools can be wrapped so that their
+    iterator instances are serialized.
+
+    For example, itertools.count does not make thread-safe instances,
+    but that is easily fixed with:
+
+        atomic_counter = synchronized_iterator(itertools.count)
+
+    Can also be used as a decorator for generator function definitions
+    so that the generator instances are serialized::
+
+        import time
+
+        @synchronized_iterator
+        def enumerate_and_timestamp(iterable):
+            for count, value in enumerate(iterable):
+                yield count, time.time_ns(), value
+
+    """
+
+    from functools import wraps
+
+    @wraps(func)
+    def inner(*args, **kwargs):
+        iterator = func(*args, **kwargs)
+        return serialize_iterator(iterator)
+
+    return inner
+
+
+def concurrent_tee(iterable, n=2):
+    """Variant of itertools.tee() but with guaranteed threading semantics.
+
+    Takes a non-threadsafe iterator as an input and creates concurrent
+    tee objects for other threads to have reliable independent copies of
+    the data stream.
+
+    The new iterators are only thread-safe if consumed within a single thread.
+    To share just one of the new iterators across multiple threads, wrap it
+    with threading.serialize_iterator().
+    """
+
+    if n < 0:
+        raise ValueError("n must be a non-negative integer")
+    if n == 0:
+        return ()
+    iterator = _concurrent_tee(iterable)
+    result = [iterator]
+    for _ in range(n - 1):
+        result.append(_concurrent_tee(iterator))
+    return tuple(result)
+
+
+class _concurrent_tee:
+    __slots__ = ('iterator', 'link', 'lock')
+
+    def __init__(self, iterable):
+        if isinstance(iterable, _concurrent_tee):
+            self.iterator = iterable.iterator
+            self.link = iterable.link
+            self.lock = iterable.lock
+        else:
+            self.iterator = iter(iterable)
+            self.link = [None, None]
+            self.lock = Lock()
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        link = self.link
+        if link[1] is None:
+            with self.lock:
+                if link[1] is None:
+                    link[0] = next(self.iterator)
+                    link[1] = [None, None]
+        value, self.link = link
+        return value
+
+############################################################
+
+
 # Helper to generate new thread names
 _counter = _count(1).__next__
 def _newname(name_template):
@@ -1565,8 +1701,9 @@ def _shutdown():
     # normally - that won't happen until the interpreter is nearly dead. So
     # mark it done here.
     if _main_thread._os_thread_handle.is_done() and _is_main_interpreter():
-        # _shutdown() was already called
-        return
+        # _shutdown() was already called, but threads might have started
+        # in the meantime.
+        return _thread_shutdown()
 
     global _SHUTTING_DOWN
     _SHUTTING_DOWN = True
diff --git a/crates/vm/src/stdlib/_thread.rs b/crates/vm/src/stdlib/_thread.rs
index a2bab628a..44dddedc6 100644
--- a/crates/vm/src/stdlib/_thread.rs
+++ b/crates/vm/src/stdlib/_thread.rs
@@ -124,15 +124,62 @@ pub(crate) mod _thread {
     }
 
     macro_rules! acquire_lock_impl {
-        ($mu:expr, $blocking:expr, $timeout:expr, $vm:expr) => {{
+        ($mu:expr, $blocking:expr, $timeout:expr, $vm:expr) => {{ acquire_lock_impl!($mu, $blocking, $timeout, $vm, false) }};
+        ($mu:expr, $blocking:expr, $timeout:expr, $vm:expr, $owned:expr) => {{
             let (mu, blocking, timeout, vm) = ($mu, $blocking, $timeout, $vm);
-            match blocking {
-                true if timeout == -1.0 => {
-                    vm.allow_threads(|| mu.lock());
-                    Ok(true)
+            let started = std::time::Instant::now();
+            // CPython rounds positive timeouts up to the next nanosecond.
+            let timeout = (timeout != -1.0)
+                .then(|| Duration::from_nanos((timeout * 1_000_000_000.0).ceil() as u64));
+            let mut remaining = timeout;
+            loop {
+                // Ownership gained in a signal handler does not make an
+                // already pending acquisition recursive.
+                let owned = $owned;
+                if !owned && mu.try_lock() {
+                    break Ok(true);
                 }
-                true => Ok(vm.allow_threads(|| mu.try_lock_for(Duration::from_secs_f64(timeout)))),
-                false => Ok(mu.try_lock()),
+                if !blocking || remaining.is_some_and(|remaining| remaining.is_zero()) {
+                    break Ok(false);
+                }
+                if vm
+                    .state
+                    .finalizing
+                    .load(core::sync::atomic::Ordering::Acquire)
+                {
+                    // Other Python threads cannot release the lock once
+                    // finalization begins. Uncontested calls still work.
+                    break Err(vm.new_python_finalization_error(
+                        "cannot acquire lock at interpreter finalization",
+                    ));
+                }
+                let interval = Duration::from_millis(50);
+                let wait = match timeout {
+                    Some(timeout) => match timeout.checked_sub(started.elapsed()) {
+                        Some(remaining) if !remaining.is_zero() => remaining.min(interval),
+                        _ => break Ok(false),
+                    },
+                    None => interval,
+                };
+                // parking_lot waits are not interrupted by OS signals. Poll
+                // without restarting the deadline or acquiring a lock before
+                // running callbacks. A handler may itself retain an RLock.
+                let acquired = vm.allow_threads(|| {
+                    if owned {
+                        thread::sleep(wait);
+                        false
+                    } else {
+                        mu.try_lock_for(wait)
+                    }
+                });
+                if acquired {
+                    break Ok(true);
+                }
+                if timeout.is_some_and(|timeout| started.elapsed() >= timeout) {
+                    break Ok(false);
+                }
+                vm.check_signals()?;
+                remaining = timeout.map(|timeout| timeout.saturating_sub(started.elapsed()));
             }
         }};
     }
@@ -247,7 +294,18 @@ pub(crate) mod _thread {
     #[pyclass(with(Representable), flags(BASETYPE, HAS_WEAKREF))]
     impl Py<RLock> {
         #[pyslot]
-        fn slot_new(cls: PyTypeRef, _args: FuncArgs, vm: &VirtualMachine) -> PyResult {
+        fn slot_new(cls: PyTypeRef, args: FuncArgs, vm: &VirtualMachine) -> PyResult {
+            let rlock_type = RLock::class(&vm.ctx);
+            let cls_init = cls.slots.init.load().map(crate::types::fn_addr);
+            let rlock_init = rlock_type.slots.init.load().map(crate::types::fn_addr);
+            if cls.is(rlock_type) || cls_init == rlock_init {
+                if !args.args.is_empty() {
+                    return Err(vm.new_type_error("RLock() takes no positional arguments"));
+                }
+                if !args.kwargs.is_empty() {
+                    return Err(vm.new_type_error("RLock() takes no keyword arguments"));
+                }
+            }
             RLock {
                 mu: RawRMutex::INIT,
                 count: core::sync::atomic::AtomicUsize::new(0),
@@ -267,7 +325,13 @@ pub(crate) mod _thread {
                     .fetch_add(1, core::sync::atomic::Ordering::Relaxed);
                 return Ok(true);
             }
-            let result = acquire_lock_impl!(&self.mu, blocking, timeout, vm)?;
+            let result = acquire_lock_impl!(
+                &self.mu,
+                blocking,
+                timeout,
+                vm,
+                self.mu.is_owned_by_current_thread()
+            )?;
             if result {
                 self.count.store(1, core::sync::atomic::Ordering::Relaxed);
             }
@@ -514,8 +578,13 @@ pub(crate) mod _thread {
         }
         #[cfg(any(target_os = "linux", target_os = "macos"))]
         {
-            let os_name = vm.fsencode(&name)?;
-            host_thread::set_current_thread_name_bytes(os_name.as_encoded_bytes());
+            let os_name = vm.state.codec_registry.encode_text(
+                name,
+                "utf-8",
+                Some(vm.ctx.new_utf8_str("replace")),
+                vm,
+            )?;
+            host_thread::set_current_thread_name_bytes(os_name.as_bytes());
         }
         Ok(())
     }