diff --git a/array.c b/array.c
index a96c5a0b9727b0..d1453c19fe7f95 100644
--- a/array.c
+++ b/array.c
@@ -6837,6 +6837,22 @@ flatten(VALUE ary, int level)
     return result;
 }
 
+static inline VALUE
+single_nested_array(VALUE ary)
+{
+    // Fast path for the common variadic argument pattern:
+    // def foo(*args)
+    //   args.flatten!
+    //   ...
+    if (RARRAY_LEN(ary) == 1) {
+        VALUE first = RARRAY_AREF(ary, 0);
+        if (RB_TYPE_P(first, T_ARRAY) && CLASS_OF(first) == rb_cArray) {
+            return first;
+        }
+    }
+    return 0;
+}
+
 /*
  *  call-seq:
  *    flatten!(depth = nil) -> self or nil
@@ -6883,11 +6899,24 @@ rb_ary_flatten_bang(int argc, VALUE *argv, VALUE ary)
     if (!NIL_P(lv)) level = NUM2INT(lv);
     if (level == 0) return Qnil;
 
-    result = flatten(ary, level);
-    if (result == ary) {
-        return Qnil;
+    VALUE child = single_nested_array(ary);
+    if (child) {
+        if (level == 1) {
+            result = child;
+        }
+        else {
+            if (level > 1) level--;
+            result = flatten(child, level);
+        }
+    }
+    else {
+        result = flatten(ary, level);
+        if (result == ary) {
+            return Qnil;
+        }
     }
-    if (!(mod = ARY_EMBED_P(result))) rb_ary_freeze(result);
+
+    if (!(mod = ARY_EMBED_P(result) && result != child)) rb_ary_freeze(result);
     rb_ary_replace(ary, result);
     if (mod) ARY_SET_EMBED_LEN(result, 0);
 
@@ -6940,9 +6969,22 @@ rb_ary_flatten(int argc, VALUE *argv, VALUE ary)
         if (level == 0) return ary_make_shared_copy(ary);
     }
 
-    result = flatten(ary, level);
-    if (result == ary) {
-        result = ary_make_shared_copy(ary);
+    VALUE child = single_nested_array(ary);
+    if (child) {
+        if (level == 1) {
+            result = child;
+        }
+        else {
+            level--;
+            result = flatten(child, level);
+        }
+    }
+    else {
+        result = flatten(ary, level);
+    }
+
+    if (result == ary || result == child) {
+        return ary_make_shared_copy(result);
     }
 
     return result;
diff --git a/benchmark/array_flatten.yml b/benchmark/array_flatten.yml
index 88ef544ba05cd1..5ef656dab4a263 100644
--- a/benchmark/array_flatten.yml
+++ b/benchmark/array_flatten.yml
@@ -4,16 +4,26 @@ prelude: |
   small_pairs_ary = [[1, 2]] * 5
   large_pairs_ary = [[1, 2]] * 100
   mostly_flat_ary = 100.times.to_a.push([101, 102])
+  small_nested_ary = [small_flat_ary]
+  large_nested_ary = [large_flat_ary]
 
 benchmark:
   small_flat_ary.flatten: small_flat_ary.flatten
-  small_flat_ary.flatten!: small_flat_ary.flatten!
+  small_flat_ary.flatten!: small_flat_ary.dup.flatten!
   large_flat_ary.flatten: large_flat_ary.flatten
-  large_flat_ary.flatten!: large_flat_ary.flatten!
+  large_flat_ary.flatten!: large_flat_ary.dup.flatten!
   small_pairs_ary.flatten: small_pairs_ary.flatten
   small_pairs_ary.flatten!: small_pairs_ary.dup.flatten!
   large_pairs_ary.flatten: large_pairs_ary.flatten
   large_pairs_ary.flatten!: large_pairs_ary.dup.flatten!
   mostly_flat_ary.flatten: mostly_flat_ary.flatten
   mostly_flat_ary.flatten!: mostly_flat_ary.dup.flatten!
+  small_nested_ary.flatten: small_nested_ary.flatten
+  small_nested_ary.flatten!: small_nested_ary.dup.flatten!
+  large_nested_ary.flatten: large_nested_ary.flatten
+  large_nested_ary.flatten!: large_nested_ary.dup.flatten!
+  small_nested_ary.flatten(1): small_nested_ary.flatten(1)
+  small_nested_ary.flatten!(1): small_nested_ary.dup.flatten!(1)
+  large_nested_ary.flatten(1): large_nested_ary.flatten(1)
+  large_nested_ary.flatten!(1): large_nested_ary.dup.flatten!(1)
 loop_count: 10000
diff --git a/gc.c b/gc.c
index f72d2c09af374c..993ccdd5446224 100644
--- a/gc.c
+++ b/gc.c
@@ -4183,9 +4183,12 @@ rb_gc_obj_foreign_p(VALUE obj)
 bool
 rb_gc_single_objspace_p(void)
 {
-    if (!rb_gc_impl_multi_objspace_p() || ruby_single_main_ractor) return true;
+    if (!rb_gc_impl_multi_objspace_p()) return true;
     rb_vm_t *vm = GET_VM();
-    return vm->ractor.cnt == 1 && vm->gc.zombie_objspaces_count == 0 && gc_absorbing_zombie == 0 &&
+    /* One Ractor is not one objspace: a forked child re-enters single-Ractor mode while
+     * the pre-fork Ractors' objspaces are still parked in zombie_objspaces. */
+    return (ruby_single_main_ractor != NULL || vm->ractor.cnt == 1) &&
+           vm->gc.zombie_objspaces_count == 0 && gc_absorbing_zombie == 0 &&
            !gc_absorbed_since_global_gc &&
            (vm->ractor.main_ractor == NULL ||
             vm->ractor.main_ractor->creating_child_objspace == NULL);
diff --git a/io_buffer.c b/io_buffer.c
index 0da1bf5cd8017e..6b8d834f9b28d6 100644
--- a/io_buffer.c
+++ b/io_buffer.c
@@ -191,6 +191,7 @@ io_buffer_zero(struct rb_io_buffer *buffer)
 {
     buffer->base = NULL;
     buffer->size = 0;
+    buffer->flags = 0;
     buffer->lock_count = 0;
 #if defined(_WIN32)
     buffer->mapping = NULL;
@@ -259,13 +260,6 @@ io_buffer_free(struct rb_io_buffer *buffer)
         // if (RB_TYPE_P(buffer->source, T_STRING)) {
         //     rb_str_unlocktmp(buffer->source);
         // }
-
-        buffer->base = NULL;
-
-        buffer->size = 0;
-        buffer->flags = 0;
-        buffer->lock_count = 0;
-        buffer->source = Qnil;
     }
 
 #if defined(_WIN32)
@@ -277,6 +271,8 @@ io_buffer_free(struct rb_io_buffer *buffer)
         buffer->mapping = NULL;
     }
 #endif
+
+    io_buffer_zero(buffer);
 }
 
 static void
diff --git a/test/ruby/test_io_buffer.rb b/test/ruby/test_io_buffer.rb
index 2c0135b970a7bd..53c4fd7a2ba6fa 100644
--- a/test/ruby/test_io_buffer.rb
+++ b/test/ruby/test_io_buffer.rb
@@ -453,6 +453,16 @@ def test_transfer
     transferred = buffer.transfer
     assert_equal "Hello World", transferred.get_string
     assert_predicate buffer, :null?
+    assert_predicate buffer, :empty?
+    assert_predicate buffer, :valid?
+    refute_predicate buffer, :external?
+    refute_predicate buffer, :internal?
+    refute_predicate buffer, :mapped?
+    refute_predicate buffer, :shared?
+    refute_predicate buffer, :private?
+    refute_predicate buffer, :readonly?
+    assert_equal "", buffer.get_string
+    assert_equal 0, buffer.set_string("")
     assert_raise IO::Buffer::AccessError do
       transferred.set_string("Goodbye")
     end
diff --git a/test/ruby/test_ractor.rb b/test/ruby/test_ractor.rb
index 0901757201740d..e435d0856c72bf 100644
--- a/test/ruby/test_ractor.rb
+++ b/test/ruby/test_ractor.rb
@@ -272,6 +272,21 @@ def test_create_many_ports_with_gc_stress
     RUBY
   end
 
+  def test_fork_child_gc_pins_shareable_objects
+    # A forked child re-enters single-Ractor mode while the Ractors it had before the
+    # fork leave their objspaces behind, so its local GC still has to pin shareable
+    # objects instead of collecting them.
+    assert_ractor(<<~'RUBY')
+      port = Ractor::Port.new
+      Ractor.new(port) { |p| p << Ractor::Port.new; Ractor.receive }
+      foreign_port = port.receive # a Port owned by, and allocated in, the other Ractor
+      pid = fork { 100_000.times { +"x" }; exit!(0) }
+      _, status = Process.waitpid2(pid)
+      assert_predicate status, :success?
+      assert_instance_of Ractor::Port, foreign_port
+    RUBY
+  end if Process.respond_to?(:fork)
+
   def test_fork_raise_isolation_error
     assert_ractor(<<~'RUBY')
       ractor = Ractor.new do
diff --git a/thread.c b/thread.c
index 78c91de8e7df90..85dfaddb44c51e 100644
--- a/thread.c
+++ b/thread.c
@@ -836,6 +836,9 @@ thread_start_func_2(rb_thread_t *th, VALUE *stack_start)
 
 #if defined(USE_MN_THREADS) && USE_MN_THREADS
     if (th_has_coroutine(th)) {
+        // wait out any pending wake while th and its Ractor are still alive
+        rb_thread_wake_fence(th);
+
         // Run the coroutine thread's epilogue here, while th is still valid;
         // co_start then only makes the final transfer (see
         // coroutine_thread_terminated in thread_pthread_mn.c).
@@ -1506,26 +1509,12 @@ hrtime_update_expire(rb_hrtime_t *timeout, const rb_hrtime_t end)
 }
 COMPILER_WARNING_POP
 
+static int sleep_hrtime_until(rb_thread_t *th, rb_hrtime_t end, unsigned int fl);
+
 static int
 sleep_hrtime(rb_thread_t *th, rb_hrtime_t rel, unsigned int fl)
 {
-    enum rb_thread_status prev_status = th->status;
-    int woke;
-    rb_hrtime_t end = rb_hrtime_add(rb_hrtime_now(), rel);
-
-    th->status = THREAD_STOPPED;
-    RUBY_VM_CHECK_INTS_BLOCKING(th->ec);
-    while (th->status == THREAD_STOPPED) {
-        native_sleep(th, &rel);
-        woke = vm_check_ints_blocking(th->ec);
-        if (woke && !(fl & SLEEP_SPURIOUS_CHECK))
-            break;
-        if (hrtime_update_expire(&rel, end))
-            break;
-        woke = 1;
-    }
-    th->status = prev_status;
-    return woke;
+    return sleep_hrtime_until(th, rb_hrtime_add(rb_hrtime_now(), rel), fl);
 }
 
 static int
diff --git a/thread_pthread.c b/thread_pthread.c
index 4c5fdf74c9c8aa..44b02032a6fa69 100644
--- a/thread_pthread.c
+++ b/thread_pthread.c
@@ -334,6 +334,7 @@ static void ractor_sched_enq(rb_vm_t *vm, rb_ractor_t *r);
 static void timer_thread_wakeup(void);
 static void timer_thread_wakeup_locked(rb_vm_t *vm);
 static void timer_thread_wakeup_force(void);
+static void timer_thread_wake_fence(struct rb_thread_struct *th);
 static bool ractor_sched_timeout_arm(rb_thread_t *th, const rb_hrtime_t *rel);
 static bool ractor_sched_timeout_disarm(rb_thread_t *th);
 static void thread_sched_switch(rb_thread_t *cth, rb_thread_t *next_th);
@@ -807,7 +808,7 @@ thread_sched_wakeup_running_thread(struct rb_thread_sched *sched, rb_thread_t *n
         if (next_th->nt) {
             if (th_has_dedicated_nt(next_th)) {
                 RUBY_DEBUG_LOG("pinning th:%u", next_th->serial);
-                rb_native_cond_signal(&next_th->nt->cond.readyq);
+                rb_native_cond_signal(&next_th->nt->readyq);
             }
             else {
                 // TODO
@@ -874,6 +875,8 @@ thread_sched_wait_running_turn(struct rb_thread_sched *sched, rb_thread_t *th, b
     ASSERT_thread_sched_locked(sched, th);
     VM_ASSERT(th == rb_ec_thread_ptr(rb_current_ec_noinline()));
 
+    bool timedout = false;
+
     if (th != sched->running) {
         // TODO: This optimization should also be made to work for MN_THREADS
         if (th->has_dedicated_nt && th == sched->runnable_hot_th && (sched->running == NULL || sched->running->has_dedicated_nt)) {
@@ -923,10 +926,16 @@ thread_sched_wait_running_turn(struct rb_thread_sched *sched, rb_thread_t *th, b
 
                 thread_sched_set_unlocked(sched, th);
                 {
-                    RUBY_DEBUG_LOG("nt:%d cond:%p", th->nt->serial, &th->nt->cond.readyq);
-                    rb_nativethread_cond_t *cond = &th->nt->cond.readyq;
-
-                    if (end) {
+                    RUBY_DEBUG_LOG("nt:%d cond:%p", th->nt->serial, &th->nt->readyq);
+                    rb_nativethread_cond_t *cond = &th->nt->readyq;
+
+                    // Once someone has queued this thread the deadline is spent: it
+                    // is waiting for a turn, not for the time, and arming a kernel
+                    // timer for every round of that costs more than the wait.
+                    // Once someone has queued this thread the deadline is spent: it
+                    // is waiting for a turn, not for the time, and arming a kernel
+                    // timer for every round of that costs more than the wait.
+                    if (end && !th->sched.node.is_ready) {
                         rb_hrtime_t abs = *end;
 
                         if (!condattr_monotonic) {
@@ -934,7 +943,7 @@ thread_sched_wait_running_turn(struct rb_thread_sched *sched, rb_thread_t *th, b
                             rb_hrtime_t now = rb_hrtime_now();
                             abs = native_cond_timeout(cond, *end > now ? *end - now : 0);
                         }
-                        native_cond_timedwait(cond, &sched->lock_, &abs);
+                        timedout = native_cond_timedwait(cond, &sched->lock_, &abs) == ETIMEDOUT;
                     }
                     else {
                         rb_native_cond_wait(cond, &sched->lock_);
@@ -942,7 +951,7 @@ thread_sched_wait_running_turn(struct rb_thread_sched *sched, rb_thread_t *th, b
                 }
                 thread_sched_set_locked(sched, th);
 
-                if (end && rb_hrtime_now() >= *end &&
+                if (timedout &&
                     sched->running != th && !th->sched.node.is_ready) {
                     // the deadline passed and nobody woke this thread: get back in
                     // line for the running turn, then wait for it without a deadline
@@ -1112,6 +1121,9 @@ thread_sched_to_dead_common(struct rb_thread_sched *sched, rb_thread_t *th)
 static void
 thread_sched_to_dead(struct rb_thread_sched *sched, rb_thread_t *th)
 {
+    // wait out any pending wake here, while th's Ractor is still alive
+    timer_thread_wake_fence(th);
+
     thread_sched_lock(sched, th);
     {
         thread_sched_to_dead_common(sched, th);
@@ -1211,11 +1223,21 @@ ubf_waiting(void *ptr)
 
     thread_sched_lock(sched, th);
     {
-        if (sched->running == th) {
-            // not sleeping yet.
+        if (sched->running == th || th->sched.node.is_ready) {
+            // not sleeping yet, or a deadline already put it back in line
         }
         else {
             thread_sched_to_ready_common(sched, th, true, false);
+
+            // If the turn is taken, th stays parked until the running thread yields.
+            // For a timed wait, wake it early anyway: it re-parks at once, but its
+            // wakeup then runs on another core in parallel with the running thread,
+            // off the handoff path.  An untimed wait has no post-wake bookkeeping
+            // worth pipelining, so it skips the extra futex round.
+            if (sched->running != th && th->sched.waiting_timed &&
+                th->nt != NULL && th_has_dedicated_nt(th)) {
+                rb_native_cond_signal(&th->nt->readyq);
+            }
         }
     }
     thread_sched_unlock(sched, th);
@@ -1223,12 +1245,16 @@ ubf_waiting(void *ptr)
 
 // running -> waiting
 //
-// This thread will sleep until other thread wakeup the thread.
+// This thread will sleep until other thread wakeup the thread.  `end` is an
+// absolute deadline, NULL to sleep until woken; only a dedicated native thread,
+// which parks on its own condvar, can take one.
 static void
-thread_sched_to_waiting_until_wakeup(struct rb_thread_sched *sched, rb_thread_t *th)
+thread_sched_to_waiting_until_wakeup(struct rb_thread_sched *sched, rb_thread_t *th, const rb_hrtime_t *end)
 {
     RUBY_DEBUG_LOG("th:%u", rb_th_serial(th));
 
+    VM_ASSERT(end == NULL || th_has_dedicated_nt(th));
+
     RB_VM_SAVE_MACHINE_CONTEXT(th);
 
 
@@ -1242,9 +1268,11 @@ thread_sched_to_waiting_until_wakeup(struct rb_thread_sched *sched, rb_thread_t
         }
         else {
             bool can_direct_transfer = !th_has_dedicated_nt(th);
+            th->sched.waiting_timed = (end != NULL); // never true here for M:N (end is NULL)
             // NOTE: th->status is set before and after this sleep outside of this function in `sleep_forever`
             thread_sched_wakeup_next_thread(sched, th, can_direct_transfer);
-            thread_sched_wait_running_turn(sched, th, can_direct_transfer, NULL);
+            thread_sched_wait_running_turn(sched, th, can_direct_transfer, end);
+            th->sched.waiting_timed = false;
         }
     }
     thread_sched_unlock(sched, th);
@@ -1589,10 +1617,12 @@ rb_ractor_sched_wait(rb_execution_context_t *ec, rb_ractor_t *cr, rb_unblock_fun
             bool can_direct_transfer = !dedicated;
             RB_VM_SAVE_MACHINE_CONTEXT(th);
             th->status = THREAD_STOPPED_FOREVER;
+            th->sched.waiting_timed = (end_p != NULL); // never true here for M:N (end_p is NULL)
             RB_INTERNAL_THREAD_HOOK(RUBY_INTERNAL_THREAD_EVENT_SUSPENDED, th);
             thread_sched_wakeup_next_thread(sched, th, can_direct_transfer);
             // sleep
             thread_sched_wait_running_turn(sched, th, can_direct_transfer, end_p);
+            th->sched.waiting_timed = false;
             th->status = THREAD_RUNNABLE;
 
             // whoever woke this thread took the timeout back first
@@ -2137,11 +2167,7 @@ static void
 native_thread_destroy(struct rb_native_thread *nt)
 {
     if (nt) {
-        rb_native_cond_destroy(&nt->cond.readyq);
-
-        if (&nt->cond.readyq != &nt->cond.intr) {
-            rb_native_cond_destroy(&nt->cond.intr);
-        }
+        rb_native_cond_destroy(&nt->readyq);
 
         native_thread_destroy_atfork(nt);
     }
@@ -2447,11 +2473,7 @@ static void
 native_thread_setup(struct rb_native_thread *nt)
 {
     // init cond
-    rb_native_cond_initialize(&nt->cond.readyq);
-
-    if (&nt->cond.readyq != &nt->cond.intr) {
-        rb_native_cond_initialize(&nt->cond.intr);
-    }
+    rb_native_cond_initialize(&nt->readyq);
 }
 
 static void
@@ -2666,9 +2688,16 @@ thread_sched_reclaim(struct coroutine_context *dead_co)
 }
 #endif
 
+void
+rb_thread_wake_fence(rb_thread_t *th)
+{
+    timer_thread_wake_fence(th);
+}
+
 void
 rb_threadptr_sched_free(rb_thread_t *th)
 {
+    timer_thread_wake_fence(th);
 #if USE_MN_THREADS
     if (th->sched.malloc_stack) {
         // has dedicated
@@ -2749,64 +2778,6 @@ native_fd_select(int n, rb_fdset_t *readfds, rb_fdset_t *writefds, rb_fdset_t *e
     return rb_fd_select(n, readfds, writefds, exceptfds, timeout);
 }
 
-static void
-ubf_pthread_cond_signal(void *ptr)
-{
-    rb_thread_t *th = (rb_thread_t *)ptr;
-    RUBY_DEBUG_LOG("th:%u on nt:%d", rb_th_serial(th), (int)th->nt->serial);
-    rb_native_cond_signal(&th->nt->cond.intr);
-}
-
-static void
-native_cond_sleep(rb_thread_t *th, rb_hrtime_t *rel)
-{
-    rb_nativethread_lock_t *lock = &th->interrupt_lock;
-    rb_nativethread_cond_t *cond = &th->nt->cond.intr;
-
-    /* Solaris cond_timedwait() return EINVAL if an argument is greater than
-     * current_time + 100,000,000.  So cut up to 100,000,000.  This is
-     * considered as a kind of spurious wakeup.  The caller to native_sleep
-     * should care about spurious wakeup.
-     *
-     * See also [Bug #1341] [ruby-core:29702]
-     * http://download.oracle.com/docs/cd/E19683-01/816-0216/6m6ngupgv/index.html
-     */
-    const rb_hrtime_t max = (rb_hrtime_t)100000000 * RB_HRTIME_PER_SEC;
-
-    THREAD_BLOCKING_BEGIN(th);
-    {
-        rb_native_mutex_lock(lock);
-        th->unblock.func = ubf_pthread_cond_signal;
-        th->unblock.arg = th;
-
-        if (RUBY_VM_INTERRUPTED(th->ec)) {
-            /* interrupted.  return immediate */
-            RUBY_DEBUG_LOG("interrupted before sleep th:%u", rb_th_serial(th));
-        }
-        else {
-            if (!rel) {
-                rb_native_cond_wait(cond, lock);
-            }
-            else {
-                rb_hrtime_t end;
-
-                if (*rel > max) {
-                    *rel = max;
-                }
-
-                end = native_cond_timeout(cond, *rel);
-                native_cond_timedwait(cond, lock, &end);
-            }
-        }
-        th->unblock.func = 0;
-
-        rb_native_mutex_unlock(lock);
-    }
-    THREAD_BLOCKING_END(th);
-
-    RUBY_DEBUG_LOG("done th:%u", rb_th_serial(th));
-}
-
 #ifdef USE_UBF_LIST
 static CCAN_LIST_HEAD(ubf_list_head);
 static rb_nativethread_lock_t ubf_list_lock = RB_NATIVETHREAD_LOCK_INIT;
@@ -3181,6 +3152,9 @@ static struct {
     rb_hrtime_t next_expiry;   // never later than the earliest deadline
     struct ccan_list_head waiting_untimed;
     pthread_mutex_t waiting_lock;
+
+    // signaled when wake_pending clears on a thread; see timer_thread_wake_fence
+    rb_nativethread_cond_t wake_pending_cond;
 #endif
 
 #if (HAVE_SYS_EPOLL_H || HAVE_SYS_EVENT_H) && USE_MN_THREADS
@@ -3405,6 +3379,7 @@ rb_thread_create_timer_thread(void)
         timer_th.next_expiry = TIMER_WHEEL_NO_EXPIRY;
         ccan_list_head_init(&timer_th.waiting_untimed);
         rb_native_mutex_initialize(&timer_th.waiting_lock);
+        rb_native_cond_initialize(&timer_th.wake_pending_cond);
 #endif
 
         // open communication channel
@@ -3556,16 +3531,28 @@ native_sleep(rb_thread_t *th, rb_hrtime_t *rel)
     struct rb_thread_sched *sched = TH_SCHED(th);
 
     RUBY_DEBUG_LOG("rel:%d", rel ? (int)*rel : 0);
-    if (rel) {
-        if (th_has_dedicated_nt(th)) {
-            native_cond_sleep(th, rel);
-        }
-        else {
-            thread_sched_wait_events(sched, th, -1, thread_sched_waiting_timeout, rel);
-        }
+
+    if (rel && !th_has_dedicated_nt(th)) {
+        // an M:N thread has no condvar of its own: the timer thread wakes it
+        thread_sched_wait_events(sched, th, -1, thread_sched_waiting_timeout, rel);
+    }
+    else if (rel) {
+        /* Solaris cond_timedwait() returns EINVAL if an argument is greater than
+         * current_time + 100,000,000.  So cut up to 100,000,000.  This is
+         * considered as a kind of spurious wakeup.  The caller to native_sleep
+         * should care about spurious wakeup.
+         *
+         * See also [Bug #1341] [ruby-core:29702]
+         * http://download.oracle.com/docs/cd/E19683-01/816-0216/6m6ngupgv/index.html
+         */
+        const rb_hrtime_t max = (rb_hrtime_t)100000000 * RB_HRTIME_PER_SEC;
+        if (*rel > max) *rel = max;
+
+        rb_hrtime_t end = rb_hrtime_add(rb_hrtime_now(), *rel);
+        thread_sched_to_waiting_until_wakeup(sched, th, &end);
     }
     else {
-        thread_sched_to_waiting_until_wakeup(sched, th);
+        thread_sched_to_waiting_until_wakeup(sched, th, NULL);
     }
 
     RUBY_DEBUG_LOG("wakeup");
diff --git a/thread_pthread.h b/thread_pthread.h
index 170c9309fa32f1..04e3de8845278d 100644
--- a/thread_pthread.h
+++ b/thread_pthread.h
@@ -103,6 +103,14 @@ struct rb_thread_sched_item {
     struct rb_thread_sched_waiting waiting_reason;
     uint32_t event_serial;
 
+    // the timer thread has a wake pending for this thread; under waiting_lock
+    bool wake_pending;
+
+    // parked on its own condvar with a deadline; under the sched lock (see
+    // ubf_waiting).  Always false for an M:N thread: its deadline lives on the
+    // timer wheel, and its early wake comes from the timer thread instead.
+    bool waiting_timed;
+
     bool malloc_stack;
     void *context_stack;
     size_t context_stack_size;
@@ -121,20 +129,8 @@ struct rb_native_thread {
 
     struct rb_thread_struct *running_thread;
 
-    // to control native thread
-#if defined(__GLIBC__) || defined(__FreeBSD__)
-    union
-#else
-    /*
-     * assume the platform condvars are badly implemented and have a
-     * "memory" of which mutex they're associated with
-     */
-    struct
-#endif
-      {
-        rb_nativethread_cond_t intr; /* th->interrupt_lock */
-        rb_nativethread_cond_t readyq; /* use sched->lock */
-    } cond;
+    // to control native thread; use sched->lock
+    rb_nativethread_cond_t readyq;
 
 #ifdef USE_SIGALTSTACK
     void *altstack;
@@ -230,5 +226,6 @@ RUBY_EXTERN native_tls_key_t ruby_current_ec_key;
 struct rb_ractor_struct;
 void rb_ractor_sched_wait(struct rb_execution_context_struct *ec, struct rb_ractor_struct *cr, rb_unblock_function_t *ubf, void *ptr);
 void rb_ractor_sched_wakeup(struct rb_ractor_struct *r, struct rb_thread_struct *th);
+void rb_thread_wake_fence(struct rb_thread_struct *th);
 
 #endif /* RUBY_THREAD_PTHREAD_H */
diff --git a/thread_pthread_mn.c b/thread_pthread_mn.c
index d1bbd362c58039..97674d466d51f2 100644
--- a/thread_pthread_mn.c
+++ b/thread_pthread_mn.c
@@ -260,6 +260,44 @@ timer_thread_wakeup_thread(rb_thread_t *th, uint32_t event_serial)
 
 #define TIMEOUT_WAKE_BATCH 16
 
+// One thread the timer thread is about to wake, with the serial it was armed at.
+struct timer_wake { rb_thread_t *th; uint32_t serial; };
+
+// Mark each thread while a wake is pending for it, so a dying thread can wait
+// (timer_thread_wake_fence).  Set under waiting_lock before the lock is dropped.
+static void
+timer_wake_pending_set(struct timer_wake *batch, int n)
+{
+    for (int i = 0; i < n; i++) {
+        batch[i].th->sched.wake_pending = true;
+    }
+}
+
+static void
+timer_wake_pending_clear(struct timer_wake *batch, int n)
+{
+    rb_native_mutex_lock(&timer_th.waiting_lock);
+    for (int i = 0; i < n; i++) {
+        batch[i].th->sched.wake_pending = false;
+    }
+    rb_native_cond_broadcast(&timer_th.wake_pending_cond);
+    rb_native_mutex_unlock(&timer_th.waiting_lock);
+}
+
+// Wait out a pending wake before a thread is freed: it would touch freed memory,
+// or wake a reused thread whose first serial matches the stale entry.
+static void
+timer_thread_wake_fence(rb_thread_t *th)
+{
+    if (!TIMER_THREAD_CREATED_P()) return;
+
+    rb_native_mutex_lock(&timer_th.waiting_lock);
+    while (th->sched.wake_pending) {
+        rb_native_cond_wait(&timer_th.wake_pending_cond, &timer_th.waiting_lock);
+    }
+    rb_native_mutex_unlock(&timer_th.waiting_lock);
+}
+
 static void
 timer_thread_check_timeout(rb_vm_t *vm)
 {
@@ -269,7 +307,7 @@ timer_thread_check_timeout(rb_vm_t *vm)
 
     ccan_list_head_init(&expired);
 
-    struct timeout_wake { rb_thread_t *th; uint32_t serial; } batch[TIMEOUT_WAKE_BATCH];
+    struct timer_wake batch[TIMEOUT_WAKE_BATCH];
     bool more = true;
 
     while (more) {
@@ -294,12 +332,14 @@ timer_thread_check_timeout(rb_vm_t *vm)
                 n++;
             }
             more = !ccan_list_empty(&expired);
+            timer_wake_pending_set(batch, n);
         }
         rb_native_mutex_unlock(&timer_th.waiting_lock);
 
         for (int i = 0; i < n; i++) {
             timer_thread_wakeup_thread(batch[i].th, batch[i].serial);
         }
+        timer_wake_pending_clear(batch, n);
     }
 }
 
@@ -1452,7 +1492,7 @@ event_wait(rb_vm_t *vm)
 static void
 timer_thread_wake_fd_waiters(int fd, uint32_t generation, uint32_t wake_flags, int result)
 {
-    struct { rb_thread_t *th; uint32_t serial; } batch[FD_WAKE_BATCH];
+    struct timer_wake batch[FD_WAKE_BATCH];
 
     if (wake_flags == 0) return;
 
@@ -1493,12 +1533,15 @@ timer_thread_wake_fd_waiters(int fd, uint32_t generation, uint32_t wake_flags, i
                 // they all just woke up).
                 fd_waiters_arm(fd, e, fd_waiters_union(e));
             }
+
+            timer_wake_pending_set(batch, n);
         }
         rb_native_mutex_unlock(&timer_th.waiting_lock);
 
         for (int i = 0; i < n; i++) {
             timer_thread_wakeup_thread(batch[i].th, batch[i].serial);
         }
+        timer_wake_pending_clear(batch, n);
 
         if (!more) break;
     }
@@ -1698,6 +1741,12 @@ timer_wheel_timeout(int timeout)
     return timeout; // no M:N threads, no timed waiters
 }
 
+static void
+timer_thread_wake_fence(rb_thread_t *th)
+{
+    // no timer wheel, no wake batches
+}
+
 static void
 timer_thread_check_timeout(rb_vm_t *vm)
 {