Ractor: let a wait for a message take a timeout

Ractor::Port#receive, Ractor.receive and Ractor.select gain a `timeout:`
keyword and return nil when it passes.

The wait itself stays where it is: rb_ractor_sched_wait() keeps parking the
thread in the thread scheduler, so an M:N thread still hands its native
thread back instead of becoming dedicated.  How the deadline is taken
depends on which kind of thread waits:

A dedicated native thread parks on its own condvar, so it takes the deadline
there, the way native_cond_sleep() does.  Nothing else is involved: the
condvar it waits on is the one a send already signals.

An M:N thread has no condvar of its own, so its deadline is armed on the
timer thread as a timeout-only wheel entry, and the timer thread wakes it
through thread_sched_to_ready_common(), which is exactly how
rb_ractor_sched_wakeup() wakes it for a send.  That gives such a waiter two
wakers, so rb_ractor_sched_wakeup() now takes an armed timeout back before
waking, skips a thread that a fired timeout already made runnable, and bumps
the scheduler event serial so a timeout that has not fired cannot wake it a
second time.

`timeout: 0` is a poll: it never blocks, converts nothing and reads no clock,
so it costs what a receive of a waiting message costs.  It still delivers the
incoming queue first, so a message another thread just sent is not missed.

A timeout that expires leaves the waiter on the Ractor's waiter list, as a
spurious wakeup does, and ractor_wait_receive() still delivers pending
messages before it reports the timeout, so a message that arrived just before
the deadline is not lost.

Both kinds of wait exist on every pthread platform, including builds without
the timer wheel (USE_MN_THREADS == 0), where every thread is dedicated.  On
win32 the wait is a condvar wait, which takes the timeout directly.

ractor.c

@@ -894,7 +894,7 @@ void
 rb_ractor_receive_parameters(rb_execution_context_t *ec, rb_ractor_t *r, int len, VALUE *ptr)
 {
     for (int i=0; i<len; i++) {
-        ptr[i] = ractor_receive(ec, ractor_default_port(r));
+        ptr[i] = ractor_receive(ec, ractor_default_port(r), NULL);
     }
 }
 
@@ -3950,7 +3950,7 @@ rb_ractor_require(VALUE feature, bool silent)
     rb_ractor_interrupt_exec(main_r, ractor_require_func, (void *)crr_obj, rb_interrupt_exec_flag_value_data);
 
     // wait for require done
-    VALUE results = ractor_port_receive(ec, crr->port);
+    VALUE results = ractor_port_receive(ec, crr->port, Qnil);
     ractor_port_close(ec, crr->port);
 
     VALUE exc = rb_ary_pop(results);
@@ -4001,7 +4001,7 @@ rb_ractor_autoload_load(VALUE module, ID name)
     rb_ractor_interrupt_exec(main_r, ractor_autoload_load_func, (void *)crr_obj, rb_interrupt_exec_flag_value_data);
 
     // wait for require done
-    VALUE results = ractor_port_receive(ec, crr->port);
+    VALUE results = ractor_port_receive(ec, crr->port, Qnil);
     ractor_port_close(ec, crr->port);
 
     VALUE exc = rb_ary_pop(results);

ractor.rb

@@ -265,10 +265,11 @@ class Ractor
 
   #
   # call-seq:
-  #    Ractor.select(*ractors_or_ports) -> [ractor or port, obj]
+  #    Ractor.select(*ractors_or_ports, timeout: nil) -> [ractor or port, obj] or nil
   #
   # Blocks the current Thread until one of the given ports has received a message. Returns an
   # array of two elements where the first element is the Port and the second is the received object.
+  # With +timeout+ (in seconds) it returns +nil+ instead once the timeout passes.
   # This method can also accept Ractor objects themselves, and in that case will wait until one
   # has terminated and return a two-element array where the first element is the ractor and the
   # second is its termination value.
@@ -307,7 +308,7 @@ class Ractor
   #      values << val
   #    end
   #
-  def self.select(*ports)
+  def self.select(*ports, timeout: nil)
     raise ArgumentError, 'specify at least one Ractor::Port or Ractor' if ports.empty?
 
     monitors = {} # Ractor::Port => Ractor
@@ -327,7 +328,10 @@ class Ractor
     end
 
     begin
-      result_port, obj = __builtin_ractor_select_internal(ports)
+      result = __builtin_ractor_select_internal(ports, timeout)
+      return nil if result.nil? # timed out
+
+      result_port, obj = result
 
       if r = monitors[result_port]
         [r, r.value]
@@ -345,11 +349,11 @@ class Ractor
 
   #
   # call-seq:
-  #    Ractor.receive -> obj
+  #    Ractor.receive(timeout: nil) -> obj or nil
   #
   # Receives a message from the current ractor's default port.
-  def self.receive
-    Ractor.current.default_port.receive
+  def self.receive(timeout: nil)
+    Ractor.current.default_port.receive(timeout: timeout)
   end
 
   class << self
@@ -357,8 +361,8 @@ class Ractor
   end
 
   # same as Ractor.receive
-  private def receive
-    default_port.receive
+  private def receive(timeout: nil)
+    default_port.receive(timeout: timeout)
   end
   alias recv receive
 
@@ -702,7 +706,7 @@ class Ractor
   class Port
     #
     # call-seq:
-    #    port.receive -> msg
+    #    port.receive(timeout: nil) -> msg or nil
     #
     # Receives a message from the port (which was sent there by Port#send). Only the ractor
     # that created the port can receive messages this way.
@@ -743,6 +747,17 @@ class Ractor
     #     Still received only one
     #     Received: message2
     #
+    # With +timeout+ (in seconds) the method gives up waiting and returns +nil+
+    # once it passes. A message that arrives just as the timeout expires is still
+    # returned; the timeout bounds the wait, it does not cut delivery off.
+    #
+    #     port = Ractor::Port.new
+    #     port.receive(timeout: 0.1) #=> nil
+    #     port.receive(timeout: 0)   #=> nil
+    #
+    # A +timeout+ of 0 never blocks and reads no clock: it takes a message if one
+    # is already there and returns +nil+ otherwise.
+    #
     # If the port is closed and there are no more messages in the message queue,
     # the method raises Ractor::ClosedError.
     #
@@ -750,9 +765,9 @@ class Ractor
     #     port.close
     #     port.receive #=> raise Ractor::ClosedError
     #
-    def receive
+    def receive(timeout: nil)
       __builtin_cexpr! %q{
-        ractor_port_receive(ec, self)
+        ractor_port_receive(ec, self, timeout)
       }
     end
 

ractor_core.h

@@ -4,6 +4,7 @@
 #include "vm_core.h"
 #include "id_table.h"
 #include "vm_debug.h"
+#include "hrtime.h"
 
 #ifndef RACTOR_CHECK_MODE
 #define RACTOR_CHECK_MODE (VM_CHECK_MODE || RUBY_DEBUG) && (SIZEOF_UINT64_T == SIZEOF_VALUE)
@@ -181,6 +182,9 @@ struct ractor_waiter {
     rb_thread_t *th;
     struct ccan_list_node node;
     rb_atomic_t event_serial;
+
+    // absolute deadline for this wait, NULL when there is no timeout
+    const rb_hrtime_t *end;
 };
 
 static inline VALUE

ractor_sync.c

@@ -13,7 +13,7 @@ ractor_port_id(const struct ractor_port *rp)
 
 static VALUE rb_cRactorPort;
 
-static VALUE ractor_receive(rb_execution_context_t *ec, const struct ractor_port *rp);
+static VALUE ractor_receive(rb_execution_context_t *ec, const struct ractor_port *rp, const rb_hrtime_t *end);
 static VALUE ractor_send(rb_execution_context_t *ec, const struct ractor_port *rp, VALUE obj, VALUE move);
 static struct ractor_basket *ractor_basket_new_ref(VALUE shareable);
 static void ractor_send_basket(rb_execution_context_t *ec, const struct ractor_port *rp, struct ractor_basket *b, bool raise_on_error);
@@ -149,8 +149,10 @@ ractor_port_p(VALUE self)
     return rb_typeddata_is_kind_of(self, &ractor_port_data_type);
 }
 
+static const rb_hrtime_t *ractor_timeout_deadline(VALUE timeout, rb_hrtime_t *storage);
+
 static VALUE
-ractor_port_receive(rb_execution_context_t *ec, VALUE self)
+ractor_port_receive(rb_execution_context_t *ec, VALUE self, VALUE timeout)
 {
     const struct ractor_port *rp = ractor_port_ptr_check(self);
 
@@ -158,9 +160,14 @@ ractor_port_receive(rb_execution_context_t *ec, VALUE self)
         rb_raise(rb_eRactorError, "only allowed from the creator Ractor of this port");
     }
 
-    VALUE v = ractor_receive(ec, rp);
+    rb_hrtime_t deadline;
+    const rb_hrtime_t *end = ractor_timeout_deadline(timeout, &deadline);
+
+    VALUE v = ractor_receive(ec, rp, end);
     RB_GC_GUARD(self);
-    return v;
+
+    // no message before the timeout
+    return UNDEF_P(v) ? Qnil : v;
 }
 
 static VALUE
@@ -1294,13 +1301,22 @@ basket_type_name(enum ractor_basket_type type)
 #else // win32
 
 static void
-ractor_cond_wait(rb_ractor_t *r)
+ractor_cond_wait(rb_ractor_t *r, const rb_hrtime_t *end)
 {
 #if RACTOR_CHECK_MODE > 0
     VALUE locked_by = r->sync.locked_by;
     r->sync.locked_by = Qnil;
 #endif
-    rb_native_cond_wait(&r->sync.wakeup_cond, &r->sync.lock);
+    if (end) {
+        rb_hrtime_t now = rb_hrtime_now();
+        rb_hrtime_t rel = *end > now ? *end - now : 0;
+        // the condvar takes msec: never round a live timeout down to 0
+        unsigned long msec = (unsigned long)(rel / RB_HRTIME_PER_MSEC);
+        rb_native_cond_timedwait(&r->sync.wakeup_cond, &r->sync.lock, msec > 0 ? msec : 1);
+    }
+    else {
+        rb_native_cond_wait(&r->sync.wakeup_cond, &r->sync.lock);
+    }
 
 #if RACTOR_CHECK_MODE > 0
     r->sync.locked_by = locked_by;
@@ -1316,7 +1332,7 @@ ractor_wait_no_gvl(void *ptr)
     RACTOR_LOCK_SELF(cr);
     {
         if (waiter->wakeup_status == wakeup_none) {
-            ractor_cond_wait(cr);
+            ractor_cond_wait(cr, waiter->end);
         }
     }
     RACTOR_UNLOCK_SELF(cr);
@@ -1406,14 +1422,16 @@ ubf_ractor_wait(void *ptr)
     rb_native_mutex_lock(&th->interrupt_lock);
 }
 
+// Waits for an event on cr.  `end` is an absolute deadline, NULL to wait forever.
 static enum ractor_wakeup_status
-ractor_wait(rb_execution_context_t *ec, rb_ractor_t *cr)
+ractor_wait(rb_execution_context_t *ec, rb_ractor_t *cr, const rb_hrtime_t *end)
 {
     rb_thread_t *th = rb_ec_thread_ptr(ec);
 
     struct ractor_waiter waiter = {
         .wakeup_status = wakeup_none,
         .th = th,
+        .end = end,
     };
 
     RUBY_DEBUG_LOG("wait%s", "");
@@ -1481,19 +1499,30 @@ ractor_check_received(rb_ractor_t *cr, struct ractor_queue *messages)
     return received;
 }
 
-static void
-ractor_wait_receive(rb_execution_context_t *ec, rb_ractor_t *cr)
+// Returns false if the deadline `end` passed with nothing to deliver.  Incoming
+// messages are delivered even then, so the caller retries its queue once more.
+static bool
+ractor_wait_receive(rb_execution_context_t *ec, rb_ractor_t *cr, const rb_hrtime_t *end)
 {
     struct ractor_queue messages;
     bool deliverred = false;
+    bool timedout = false;
 
     RACTOR_LOCK_SELF(cr);
     {
         if (ractor_check_received(cr, &messages)) {
             deliverred = true;
         }
+        else if (!end) {
+            ractor_wait(ec, cr, NULL); // no timeout: wait until a message arrives
+        }
+        else if (*end == 0) {
+            timedout = true; // `timeout: 0`: over without reading any clock
+        }
         else {
-            ractor_wait(ec, cr);
+            // only a wakeup nobody claimed can be the deadline, so only then look at
+            // the clock: a send or an interrupt says what woke this thread by itself
+            timedout = ractor_wait(ec, cr, end) == wakeup_none && rb_hrtime_now() >= *end;
         }
     }
     RACTOR_UNLOCK_SELF(cr);
@@ -1506,6 +1535,8 @@ ractor_wait_receive(rb_execution_context_t *ec, rb_ractor_t *cr)
             ractor_queue_enq(cr, ractor_get_queue(cr, b->port_id, false), b);
         }
     }
+
+    return !timedout;
 }
 
 static VALUE
@@ -1533,8 +1564,12 @@ ractor_try_receive(rb_execution_context_t *ec, rb_ractor_t *cr, const struct rac
     }
 }
 
+// Returns Qundef if the deadline passed first.  It bounds how long this blocks; it
+// does not cut delivery off.  A message that lands while the timeout is being
+// reported is still returned, as Thread::Queue#pop(timeout:) does.  Either way
+// nothing is lost: a basket only leaves the queue when it is returned.
 static VALUE
-ractor_receive(rb_execution_context_t *ec, const struct ractor_port *rp)
+ractor_receive(rb_execution_context_t *ec, const struct ractor_port *rp, const rb_hrtime_t *end)
 {
     rb_ractor_t *cr = rb_ec_ractor_ptr(ec);
     VM_ASSERT(cr == rp->r);
@@ -1547,12 +1582,33 @@ ractor_receive(rb_execution_context_t *ec, const struct ractor_port *rp)
         if (v != Qundef) {
             return v;
         }
-        else {
-            ractor_wait_receive(ec, cr);
+        else if (!ractor_wait_receive(ec, cr, end)) {
+            return Qundef;
         }
     }
 }
 
+// A timeout argument becomes an absolute deadline, or 0 for `timeout: 0`, which
+// every wait reads as "do not wait".  Returns NULL when there is no timeout.
+static const rb_hrtime_t *
+ractor_timeout_deadline(VALUE timeout, rb_hrtime_t *storage)
+{
+    if (NIL_P(timeout)) return NULL;
+
+    if (!(FIXNUM_P(timeout) && FIX2LONG(timeout) == 0)) {
+        struct timeval tv = rb_time_interval(timeout); // raises on a negative timeout
+        rb_hrtime_t rel = rb_timeval2hrtime(&tv);
+
+        if (rel > 0) {
+            *storage = rb_hrtime_add(rb_hrtime_now(), rel);
+            return storage;
+        }
+    }
+
+    *storage = 0;
+    return storage;
+}
+
 // Ractor#send
 
 static void
@@ -1816,7 +1872,7 @@ ractor_selector_wait_i(st_data_t key, st_data_t val, st_data_t data)
 }
 
 static VALUE
-ractor_selector__wait(rb_execution_context_t *ec, VALUE selector)
+ractor_selector__wait(rb_execution_context_t *ec, VALUE selector, const rb_hrtime_t *end)
 {
     rb_ractor_t *cr = rb_ec_ractor_ptr(ec);
     struct ractor_selector *s = RACTOR_SELECTOR_PTR(selector);
@@ -1833,8 +1889,9 @@ ractor_selector__wait(rb_execution_context_t *ec, VALUE selector)
         if (data.found) {
             return rb_ary_new_from_args(2, data.rpv, data.v);
         }
-
-        ractor_wait_receive(ec, cr);
+        else if (!ractor_wait_receive(ec, cr, end)) {
+            return Qnil;
+        }
     }
 }
 
@@ -1847,7 +1904,7 @@ ractor_selector__wait(rb_execution_context_t *ec, VALUE selector)
 static VALUE
 ractor_selector_wait(VALUE selector)
 {
-    return ractor_selector__wait(GET_EC(), selector);
+    return ractor_selector__wait(GET_EC(), selector, NULL);
 }
 
 static VALUE
@@ -1863,10 +1920,13 @@ ractor_selector_new(int argc, VALUE *ractors, VALUE klass)
 }
 
 static VALUE
-ractor_select_internal(rb_execution_context_t *ec, VALUE self, VALUE ports)
+ractor_select_internal(rb_execution_context_t *ec, VALUE self, VALUE ports, VALUE timeout)
 {
+    rb_hrtime_t deadline;
+    const rb_hrtime_t *end = ractor_timeout_deadline(timeout, &deadline);
+
     VALUE selector = ractor_selector_new(RARRAY_LENINT(ports), (VALUE *)RARRAY_CONST_PTR(ports), rb_cRactorSelector);
-    VALUE result = ractor_selector__wait(ec, selector);
+    VALUE result = ractor_selector__wait(ec, selector, end);
 
     RB_GC_GUARD(selector);
     RB_GC_GUARD(ports);

test/ruby/test_ractor.rb

@@ -579,6 +579,66 @@ class TestRactor < Test::Unit::TestCase
       assert_equal 4, ports.size
     RUBY
   end
+  def test_port_receive_timeout
+    assert_separately([], __FILE__, __LINE__, <<-'RUBY')
+      Warning[:experimental] = false
+      port = Ractor::Port.new
+
+      t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      assert_nil port.receive(timeout: 0.1)
+      assert_operator Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0, :>=, 0.1
+
+      # a message that is already there wins over the timeout
+      port << :a
+      assert_equal :a, port.receive(timeout: 10)
+
+      # timeout: 0 polls
+      assert_nil port.receive(timeout: 0)
+      port << :b
+      assert_equal :b, port.receive(timeout: 0)
+    RUBY
+  end
+
+  def test_receive_timeout_on_mn_thread
+    assert_separately([], __FILE__, __LINE__, <<-'RUBY')
+      Warning[:experimental] = false
+      # a Ractor's thread is an M:N thread: the timeout must not need a native thread
+      r = Ractor.new do
+        t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        [Ractor.receive(timeout: 0.1), Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0]
+      end
+      v, elapsed = r.value
+      assert_nil v
+      assert_operator elapsed, :>=, 0.1
+    RUBY
+  end
+
+  def test_select_timeout
+    assert_separately([], __FILE__, __LINE__, <<-'RUBY')
+      Warning[:experimental] = false
+      p1, p2 = Ractor::Port.new, Ractor::Port.new
+      assert_nil Ractor.select(p1, p2, timeout: 0.1)
+
+      p2 << :b
+      assert_equal [p2, :b], Ractor.select(p1, p2, timeout: 10)
+    RUBY
+  end
+
+  def test_receive_timeout_racing_with_send
+    assert_separately([], __FILE__, __LINE__, <<-'RUBY')
+      Warning[:experimental] = false
+      # the timeout and a send aim at the same instant: both wake the waiter
+      results = []
+      300.times do
+        port = Ractor::Port.new
+        th = Thread.new(port) {|p| sleep 0.001; p << :msg }
+        results << port.receive(timeout: 0.001)
+        th.join
+      end
+      assert_empty results.uniq - [:msg, nil]
+    RUBY
+  end
+
 
   # Moving a Hash that has Hash keys must not lose entries (regression guard for inserting a
   # key before its contents are filled in, which corrupts its hash value).

thread_pthread.c

@@ -334,6 +334,8 @@ 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 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);
 static void ractor_sched_cancel_enq(rb_vm_t *vm, struct rb_thread_sched *sched);
 #if USE_MN_THREADS
@@ -863,9 +865,9 @@ thread_sched_to_ready(struct rb_thread_sched *sched, rb_thread_t *th)
     thread_sched_unlock(sched, th);
 }
 
-// wait until sched->running is `th`.
+// wait until sched->running is `th`.  `end` is an absolute deadline for a dedicated
 static void
-thread_sched_wait_running_turn(struct rb_thread_sched *sched, rb_thread_t *th, bool can_direct_transfer)
+thread_sched_wait_running_turn(struct rb_thread_sched *sched, rb_thread_t *th, bool can_direct_transfer, const rb_hrtime_t *end)
 {
     RUBY_DEBUG_LOG("th:%u", rb_th_serial(th));
 
@@ -922,10 +924,32 @@ 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_native_cond_wait(&th->nt->cond.readyq, &sched->lock_);
+                    rb_nativethread_cond_t *cond = &th->nt->cond.readyq;
+
+                    if (end) {
+                        rb_hrtime_t abs = *end;
+
+                        if (!condattr_monotonic) {
+                            // the condvar counts in another clock: restate it there
+                            rb_hrtime_t now = rb_hrtime_now();
+                            abs = native_cond_timeout(cond, *end > now ? *end - now : 0);
+                        }
+                        native_cond_timedwait(cond, &sched->lock_, &abs);
+                    }
+                    else {
+                        rb_native_cond_wait(cond, &sched->lock_);
+                    }
                 }
                 thread_sched_set_locked(sched, th);
 
+                if (end && rb_hrtime_now() >= *end &&
+                    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
+                    thread_sched_to_ready_common(sched, th, false, false);
+                    end = NULL;
+                }
+
                 if (sched->runnable_hot_th != NULL && sched->runnable_hot_th_waiting) {
                     VM_ASSERT(sched->runnable_hot_th != th);
                     // Give the hot thread a chance to preempt, if it's actively spinning.
@@ -1012,7 +1036,7 @@ thread_sched_to_running_common(struct rb_thread_sched *sched, rb_thread_t *th)
     }
 
     // TODO: check SNT number
-    thread_sched_wait_running_turn(sched, th, false);
+    thread_sched_wait_running_turn(sched, th, false, NULL);
 }
 
 // waiting -> ready -> running
@@ -1220,7 +1244,7 @@ thread_sched_to_waiting_until_wakeup(struct rb_thread_sched *sched, rb_thread_t
             bool can_direct_transfer = !th_has_dedicated_nt(th);
             // 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);
+            thread_sched_wait_running_turn(sched, th, can_direct_transfer, NULL);
         }
     }
     thread_sched_unlock(sched, th);
@@ -1242,7 +1266,7 @@ thread_sched_yield(struct rb_thread_sched *sched, rb_thread_t *th)
             thread_sched_wakeup_next_thread(sched, th, !th_has_dedicated_nt(th));
             bool can_direct_transfer = !th_has_dedicated_nt(th);
             thread_sched_to_ready_common(sched, th, false, can_direct_transfer);
-            thread_sched_wait_running_turn(sched, th, can_direct_transfer);
+            thread_sched_wait_running_turn(sched, th, can_direct_transfer, NULL);
             th->status = THREAD_RUNNABLE;
         }
         else {
@@ -1531,15 +1555,49 @@ rb_ractor_sched_wait(rb_execution_context_t *ec, rb_ractor_t *cr, rb_unblock_fun
     thread_sched_lock(sched, th);
     rb_ractor_unlock_self(cr);
     {
-        // setup sleep
-        bool can_direct_transfer = !th_has_dedicated_nt(th);
-        RB_VM_SAVE_MACHINE_CONTEXT(th);
-        th->status = THREAD_STOPPED_FOREVER;
-        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);
-        th->status = THREAD_RUNNABLE;
+        // A dedicated native thread takes the deadline on the very condvar a wakeup
+        // signals.  An M:N thread has no condvar of its own, so its deadline goes to
+        // the timer thread, which then wakes it the way rb_ractor_sched_wakeup() does.
+        bool dedicated = th_has_dedicated_nt(th);
+        const rb_hrtime_t *end_p = NULL;
+        bool armed = false, expired = false;
+
+        if (waiter->end) {
+            if (dedicated) {
+                end_p = waiter->end;
+            }
+            else {
+                // the timer wheel takes a relative timeout
+                rb_hrtime_t now = rb_hrtime_now();
+                rb_hrtime_t rel = *waiter->end > now ? *waiter->end - now : 0;
+
+                armed = ractor_sched_timeout_arm(th, &rel);
+                expired = !armed;
+            }
+        }
+
+        if (expired) {
+            RUBY_DEBUG_LOG("expired before sleep%s", "");
+        }
+        else if (armed && th->sched.waiting_reason.flags == thread_sched_waiting_none) {
+            // the timer thread already took this thread out of the wheel; bump the
+            // serial so that it does not try to wake a thread that never slept
+            th->sched.event_serial++;
+        }
+        else {
+            // setup sleep
+            bool can_direct_transfer = !dedicated;
+            RB_VM_SAVE_MACHINE_CONTEXT(th);
+            th->status = THREAD_STOPPED_FOREVER;
+            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->status = THREAD_RUNNABLE;
+
+            // whoever woke this thread took the timeout back first
+            VM_ASSERT(th->sched.waiting_reason.flags == thread_sched_waiting_none);
+        }
     }
     thread_sched_unlock(sched, th);
     rb_ractor_lock_self(cr);
@@ -1561,7 +1619,20 @@ rb_ractor_sched_wakeup(rb_ractor_t *r, rb_thread_t *r_th)
     {
         if (r_th->status == THREAD_STOPPED_FOREVER) {
             RUBY_ATOMIC_ADD(r_th->unblock.event_serial, 1);
-            thread_sched_to_ready_common(sched, r_th, true, false);
+
+            // r_th must not resume with a wheel entry left behind: take its timeout
+            // back, as ubf_event_waiting() does.  Only r_th arms it, and it is
+            // parked here, so reading the flags without the timer lock is safe.
+            if (r_th->sched.waiting_reason.flags != thread_sched_waiting_none) {
+                ractor_sched_timeout_disarm(r_th);
+            }
+
+            // a timeout that fired first may have made r_th runnable already: waking
+            // it twice would put it on the readyq twice
+            if (sched->running != r_th && !r_th->sched.node.is_ready) {
+                r_th->sched.event_serial++; // a timeout still armed must not wake it again
+                thread_sched_to_ready_common(sched, r_th, true, false);
+            }
         }
     }
     thread_sched_unlock(sched, r_th);
@@ -2491,7 +2562,7 @@ nt_start(void *ptr)
                 if (sched->running == th) {
                     thread_sched_add_running_thread(sched, th);
                 }
-                thread_sched_wait_running_turn(sched, th, false);
+                thread_sched_wait_running_turn(sched, th, false, NULL);
             }
             thread_sched_unlock(sched, th);
 

thread_pthread_mn.c

@@ -364,6 +364,24 @@ enum timer_thread_register_result {
 static enum timer_thread_register_result
 timer_thread_register_waiting(rb_thread_t *th, int fd, enum thread_sched_waiting_flag flags, rb_hrtime_t *rel, uint32_t event_serial);
 
+// Arm a timeout-only wake on the timer thread for a Ractor wait.  Returns false
+// if the deadline has already passed, in which case nothing was registered.
+static bool
+ractor_sched_timeout_arm(rb_thread_t *th, const rb_hrtime_t *rel)
+{
+    rb_hrtime_t rel_copy = *rel;
+
+    return timer_thread_register_waiting(th, -1, thread_sched_waiting_timeout, &rel_copy,
+                                         ++th->sched.event_serial) == timer_thread_registered;
+}
+
+// Returns true if the timeout was still armed, i.e. it did not fire.
+static bool
+ractor_sched_timeout_disarm(rb_thread_t *th)
+{
+    return timer_thread_cancel_waiting(th);
+}
+
 // return how the wait ended; see enum thread_sched_wait_result
 static enum thread_sched_wait_result
 thread_sched_wait_events(struct rb_thread_sched *sched, rb_thread_t *th, int fd, enum thread_sched_waiting_flag events, rb_hrtime_t *rel)
@@ -414,7 +432,7 @@ thread_sched_wait_events(struct rb_thread_sched *sched, rb_thread_t *th, int fd,
                 enum rb_thread_status prev_status = th->status;
                 if (prev_status == THREAD_RUNNABLE) th->status = THREAD_STOPPED_FOREVER;
                 thread_sched_wakeup_next_thread(sched, th, true);
-                thread_sched_wait_running_turn(sched, th, true);
+                thread_sched_wait_running_turn(sched, th, true, NULL);
                 if (prev_status == THREAD_RUNNABLE) th->status = THREAD_RUNNABLE;
 
                 RUBY_DEBUG_LOG("wakeup");
@@ -1660,6 +1678,20 @@ thread_sched_wait_events(struct rb_thread_sched *sched, rb_thread_t *th, int fd,
     rb_bug("unreachable");
 }
 
+// Without the wheel every thread is dedicated, so a Ractor wait takes its
+// deadline on its own condvar and never reaches these.
+static bool
+ractor_sched_timeout_arm(rb_thread_t *th, const rb_hrtime_t *rel)
+{
+    rb_bug("unreachable");
+}
+
+static bool
+ractor_sched_timeout_disarm(rb_thread_t *th)
+{
+    rb_bug("unreachable");
+}
+
 static int
 timer_wheel_timeout(int timeout)
 {