From 7467d784eba1ff4d5cde403c09aab5d682d405fc Mon Sep 17 00:00:00 2001 From: Koichi Sasada Date: Tue, 20 Oct 2026 19:08:12 +0000 Subject: [PATCH] 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 --- test/ruby/test_thread.rb | 40 ++++++++++++++++++++ thread.c | 82 +++++++++++++++++++++++----------------- thread_sched.h | 1 + thread_sched_mn.c | 10 ++--- 4 files changed, 93 insertions(+), 40 deletions(-) diff --git a/test/ruby/test_thread.rb b/test/ruby/test_thread.rb index ef9aac47dba..e5c01a7f0b3 100644 --- a/test/ruby/test_thread.rb +++ b/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) diff --git a/thread.c b/thread.c index 2e8766eb929..8ecfc7bc0e0 100644 --- a/thread.c +++ b/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); diff --git a/thread_sched.h b/thread_sched.h index 0e48f214857..fe985731f7f 100644 --- a/thread_sched.h +++ b/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 diff --git a/thread_sched_mn.c b/thread_sched_mn.c index 585274155ce..cb0defe5c4c 100644 --- a/thread_sched_mn.c +++ b/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; }