Let an interrupted M:N io wait handle the interrupt and wait again

thread_sched_wait_events() returned thread_sched_wait_event when an
interrupt was already pending at entry, without looking at the fd, and
thread_io_wait() reported that as readiness: under RUBY_MN_THREADS=2,
IO#wait_readable on a pipe nobody wrote to returned truthy when a trap
arrived just before the wait.

A wakeup for a masked interrupt (a trap arriving inside a trap handler)
was reported as a timeout, so wait_readable(0.3) in a trap handler
returned nil at once.

Add thread_sched_wait_interrupted for both "stopped for an interrupt"
cases.  thread_io_wait() now runs the whole wait under one tag, handles
the interrupt and waits again on the M:N scheduler until its deadline,
and treats a wakeup without an event before the deadline the same way.
At the deadline it probes the fd once.  rb_thread_io_blocking_call()
retries func for every M:N outcome but unhandled.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

test/ruby/test_thread.rb

@@ -1756,6 +1756,46 @@ q.pop
     end;
   end
 
+  # RUBY_MN_THREADS=2 puts the main thread, which receives the traps, on the M:N io wait.
+  def test_mn_threads_io_wait_under_signals
+    omit "needs fork and SIGUSR1" unless Process.respond_to?(:fork) && Signal.list.key?("USR1")
+    assert_separately([{'RUBY_MN_THREADS' => '2'}], "#{<<~"begin;"}\n#{<<~'end;'}", timeout: 60)
+    begin;
+      r, w = IO.pipe
+      in_trap = nil
+      trap(:USR1) do
+        if in_trap == :measure
+          # Signals keep coming while TRAP is masked; their wakeups must not end the wait.
+          t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+          ret = r.wait_readable(0.2)
+          in_trap = [ret, Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0]
+        end
+      end
+      ppid = Process.pid
+      # Unthrottled, the traps arrive faster than the main thread can run them.
+      pid = fork { (loop { Process.kill(:USR1, ppid); sleep 0.0001 }) rescue exit!(0) }
+      begin
+        # A trap pending when the wait starts must not be reported as readiness.
+        ready = 0
+        deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 1
+        while Process.clock_gettime(Process::CLOCK_MONOTONIC) < deadline
+          ready += 1 if r.wait_readable(0.0001)
+        end
+        assert_equal 0, ready, "wait_readable reported a pipe nobody wrote to as readable"
+
+        in_trap = :measure
+        sleep 0.01 while in_trap == :measure
+        ret, elapsed = in_trap
+        assert_nil ret
+        assert_operator elapsed, :>=, 0.15, "wait_readable in a trap handler returned before its timeout"
+      ensure
+        Process.kill(:KILL, pid)
+        Process.wait(pid)
+        w.close # held open until here so that r never reaches EOF
+      end
+    end;
+  end
+
   # [Bug #21926]
   def test_thread_join_during_finalizers
     assert_separately([], "#{<<~"begin;"}\n#{<<~'end;'}", timeout: 60)

thread.c

@@ -2134,6 +2134,7 @@ enum io_wait_result {
     io_wait_ready,     // the MN scheduler waited and the fd is ready
     io_wait_timed_out, // the MN scheduler waited until the timeout expired
     io_wait_unhandled, // the MN scheduler did not wait; use the blocking path
+    io_wait_interrupted, // the MN scheduler stopped for an interrupt; handle it and wait again
 };
 
 // Wait for `fd` on the MN scheduler, if it can take this wait at all.
@@ -2169,6 +2170,8 @@ thread_io_wait_events(rb_thread_t *th, int fd, int events, const struct timeval
             // Never waited: reporting "ready" here would fabricate readiness and
             // spin, so hand the wait back to the caller's blocking path.
             return io_wait_unhandled;
+          case thread_sched_wait_interrupted:
+            return io_wait_interrupted;
         }
     }
 #endif // defined(USE_MN_THREADS) && USE_MN_THREADS
@@ -2239,7 +2242,8 @@ rb_thread_io_blocking_call(struct rb_io* io, rb_blocking_function_t *func, void
             RUBY_ASSERT(th == rb_ec_thread_ptr(ec));
             if (events && blocking_call_retryable_p((int)val, saved_errno)) {
                 // `func` just returned EAGAIN, so the fd is known not to be ready.
-                if (thread_io_wait_events(th, fd, events, NULL, true) == io_wait_ready) {
+                // Any M:N outcome but unhandled (ready, interrupted, woken early) retries `func`.
+                if (thread_io_wait_events(th, fd, events, NULL, true) != io_wait_unhandled) {
                     RUBY_VM_CHECK_INTS_BLOCKING(ec);
                     goto retry;
                 }
@@ -4962,43 +4966,50 @@ COMPILER_WARNING_IGNORED(-Wdangling-pointer)
 COMPILER_WARNING_POP
     }
 
-    // A zero timeout is a plain probe; ppoll answers it without parking.
-    bool mn_wait = timeout == NULL || timeout->tv_sec != 0 || timeout->tv_usec != 0;
-
-    enum io_wait_result mn_result = io_wait_unhandled;
     struct timeval tv_rest;
-    if (mn_wait) {
-        rb_hrtime_t started = timeout ? rb_hrtime_now() : 0;
-        mn_result = thread_io_wait_events(th, fd, events, timeout, false);
+    struct timeval *volatile wait_timeout = timeout;
+    const rb_hrtime_t deadline = timeout ? rb_hrtime_add(rb_hrtime_now(), rb_timeval2hrtime(timeout)) : 0;
 
-        // The M:N wait may hand an interrupted wait back to the blocking path
-        // below.  Charge what it already waited against the timeout, so the
-        // deadline is the one the caller asked for.
-        if (mn_result == io_wait_unhandled && timeout) {
-            rb_hrtime_t total = rb_timeval2hrtime(timeout);
-            rb_hrtime_t spent = rb_hrtime_sub(rb_hrtime_now(), started);
-            rb_hrtime_t rest = spent < total ? total - spent : 0;
-            rb_hrtime2timeval(&tv_rest, &rest);
-            timeout = &tv_rest;
+    // The interrupt checks below may raise; the tag lets the exit path unlink blocking_operation.
+    EC_PUSH_TAG(ec);
+    if ((state = EC_EXEC_TAG()) == TAG_NONE) {
+        enum io_wait_result mn_result;
+
+        while (1) {
+            // A zero timeout is a plain probe; ppoll answers it without parking.
+            bool mn_wait = wait_timeout == NULL || wait_timeout->tv_sec != 0 || wait_timeout->tv_usec != 0;
+            mn_result = mn_wait ? thread_io_wait_events(th, fd, events, wait_timeout, false) : io_wait_unhandled;
+            if (mn_result == io_wait_ready) break;
+
+            // Charge what the M:N wait spent against the deadline the caller asked for.
+            rb_hrtime_t rest = 0;
+            if (wait_timeout) {
+                rb_hrtime_t now = rb_hrtime_now();
+                rest = now < deadline ? deadline - now : 0;
+                rb_hrtime2timeval(&tv_rest, &rest);
+                wait_timeout = &tv_rest;
+            }
+            if (mn_result == io_wait_unhandled) break;
+            if (mn_result == io_wait_timed_out && wait_timeout && rest == 0) break;
+
+            // Stopped without an event before the deadline (an interrupt, or a wakeup
+            // for a masked one): handle it and wait again.  At the deadline this probes once.
+            RUBY_VM_CHECK_INTS_BLOCKING(ec);
         }
-    }
 
-    switch (mn_result) {
-      case io_wait_ready:
-        fds[0].revents = events;
-        errno = 0;
-        break;
-      case io_wait_timed_out:
-        // revents stays 0, so the result below becomes 0 as with ppoll's timeout.
-        errno = 0;
-        break;
-      case io_wait_unhandled:
-        EC_PUSH_TAG(ec);
-        struct timeval *volatile blocking_timeout = timeout;
-        if ((state = EC_EXEC_TAG()) == TAG_NONE) {
+        switch (mn_result) {
+          case io_wait_ready:
+            fds[0].revents = events;
+            errno = 0;
+            break;
+          case io_wait_timed_out:
+            // revents stays 0, so the result below becomes 0 as with ppoll's timeout.
+            errno = 0;
+            break;
+          case io_wait_unhandled: {
             rb_hrtime_t *to, rel, end = 0;
             RUBY_VM_CHECK_INTS_BLOCKING(ec);
-            timeout_prepare(&to, &rel, &end, blocking_timeout);
+            timeout_prepare(&to, &rel, &end, wait_timeout);
             do {
                 nfds = numberof(fds);
                 result = wait_for_single_fd_blocking_region(th, fds, nfds, to, &lerrno);
@@ -5007,10 +5018,13 @@ COMPILER_WARNING_POP
             } while (wait_retryable(&result, lerrno, to, end));
 
             RUBY_VM_CHECK_INTS_BLOCKING(ec);
+            break;
+          }
+          case io_wait_interrupted:
+            UNREACHABLE;
         }
-
-        EC_POP_TAG();
     }
+    EC_POP_TAG();
 
     if (io) {
         rb_io_blocking_operation_exit(io, &blocking_operation);

thread_sched.h

@@ -32,6 +32,7 @@ enum thread_sched_wait_result {
     thread_sched_wait_event,       // an event the caller asked for fired
     thread_sched_wait_timeout,     // the timeout expired before any event
     thread_sched_wait_unavailable, // not registered; the caller must fall back
+    thread_sched_wait_interrupted, // stopped for an interrupt; handle it and wait again
 };
 
 // this data should be protected by timer_th.waiting_lock

thread_sched_mn.c

@@ -547,10 +547,9 @@ thread_sched_wait_events(struct rb_thread_sched *sched, rb_thread_t *th, int fd,
     {
         // NOTE: there's a lock ordering inversion here with the ubf call, but it's benign.
         if (ubf_set(th, ubf_event_waiting, (void *)th, NULL)) {
-            // Already interrupted: report an event so the caller retries and then
-            // processes the interrupt, as it always has.
+            // Already interrupted: the fd was never looked at, so this is no event.
             thread_sched_unlock(sched, th);
-            return thread_sched_wait_event;
+            return thread_sched_wait_interrupted;
         }
 
         reg = timer_thread_register_waiting(th, fd, events, rel, event_serial);
@@ -609,9 +608,8 @@ thread_sched_wait_events(struct rb_thread_sched *sched, rb_thread_t *th, int fd,
     if (reg == timer_thread_unavailable) return thread_sched_wait_unavailable;
     // A ready fd never registered, so it never timed out either.
     if (reg == timer_thread_already_ready) return thread_sched_wait_event;
-    // No event fired and we are interrupted: that is an interrupt, not a
-    // timeout.  Hand the wait back so the caller runs it and re-polls the fd.
-    if (timedout && RUBY_VM_INTERRUPTED(th->ec)) return thread_sched_wait_unavailable;
+    // No event fired and we are interrupted: that is an interrupt, not a timeout.
+    if (timedout && RUBY_VM_INTERRUPTED(th->ec)) return thread_sched_wait_interrupted;
     return timedout ? thread_sched_wait_timeout : thread_sched_wait_event;
 }