From 4056015608e1b16f1dcc4faf1491de9853df0601 Mon Sep 17 00:00:00 2001 From: Benoit Daloze Date: Sun, 16 Aug 2026 15:49:36 +0200 Subject: [PATCH 1/3] [ruby/prism] Remove rubyvm: keyword argument from Prism.find (https://github.com/ruby/prism/pull/4202) * This does not scale well with more find implementations. * RubyVM is CRuby-specific and experimental so not something usually exposed in public APIs. * Tests can use the specific class directly, which is more reliable. https://github.com/ruby/prism/commit/411f741593 --- lib/prism.rb | 6 +++--- lib/prism/node_find.rb | 10 +++++----- test/prism/ruby/find_test.rb | 16 ++++++++-------- 3 files changed, 16 insertions(+), 16 deletions(-) diff --git a/lib/prism.rb b/lib/prism.rb index 8dc8050cc4d827..0ddf668735d81e 100644 --- a/lib/prism.rb +++ b/lib/prism.rb @@ -90,9 +90,9 @@ def self.load(source, serialized, freeze = false) # an exact match. On other implementations, it falls back to best-effort # matching by source location line number. #-- - #: (Method | UnboundMethod | Proc | Thread::Backtrace::Location callable, ?rubyvm: bool) -> Node? - def self.find(callable, rubyvm: !!defined?(RubyVM)) - NodeFind.find(callable, rubyvm) + #: (Method | UnboundMethod | Proc | Thread::Backtrace::Location callable) -> Node? + def self.find(callable) + NodeFind.find(callable) end # @rbs! diff --git a/lib/prism/node_find.rb b/lib/prism/node_find.rb index 697ee430e8b90e..85576f83e77e77 100644 --- a/lib/prism/node_find.rb +++ b/lib/prism/node_find.rb @@ -14,11 +14,11 @@ module Prism module NodeFind # :nodoc: # Find the node for the given callable or backtrace location. #-- - #: (Method | UnboundMethod | Proc | Thread::Backtrace::Location callable, bool rubyvm) -> Node? - def self.find(callable, rubyvm) + #: (Method | UnboundMethod | Proc | Thread::Backtrace::Location callable) -> Node? + def self.find(callable) case callable when Proc - if rubyvm + if defined?(::RubyVM) RubyVMCallableFind.new.find(callable) elsif callable.lambda? LineLambdaFind.new.find(callable) @@ -26,13 +26,13 @@ def self.find(callable, rubyvm) LineProcFind.new.find(callable) end when Method, UnboundMethod - if rubyvm + if defined?(::RubyVM) RubyVMCallableFind.new.find(callable) else LineMethodFind.new.find(callable) end when Thread::Backtrace::Location - if rubyvm + if defined?(::RubyVM) RubyVMBacktraceLocationFind.new.find(callable) else LineBacktraceLocationFind.new.find(callable) diff --git a/test/prism/ruby/find_test.rb b/test/prism/ruby/find_test.rb index 5b59113d30db69..a2553d0de0db8b 100644 --- a/test/prism/ruby/find_test.rb +++ b/test/prism/ruby/find_test.rb @@ -143,42 +143,42 @@ def test_multiple_methods_on_same_line assert_def_node Prism.find(Fixtures::MultipleOnLine.method(:second)), :second end - # === Fallback (line-based) tests via rubyvm: false === + # === Fallback (line-based) tests === def test_fallback_simple_method - assert_def_node Prism.find(Fixtures::Methods.instance_method(:simple_method), rubyvm: false), :simple_method + assert_def_node NodeFind::LineMethodFind.new.find(Fixtures::Methods.instance_method(:simple_method)), :simple_method end def test_fallback_singleton_method - assert_def_node Prism.find(Fixtures::Methods.method(:singleton_method_fixture), rubyvm: false), :singleton_method_fixture + assert_def_node NodeFind::LineMethodFind.new.find(Fixtures::Methods.method(:singleton_method_fixture)), :singleton_method_fixture end def test_fallback_lambda - node = Prism.find(Fixtures::Procs::SIMPLE_LAMBDA, rubyvm: false) + node = NodeFind::LineLambdaFind.new.find(Fixtures::Procs::SIMPLE_LAMBDA) assert_instance_of LambdaNode, node end def test_fallback_proc - node = Prism.find(Fixtures::Procs::SIMPLE_PROC, rubyvm: false) + node = NodeFind::LineProcFind.new.find(Fixtures::Procs::SIMPLE_PROC) assert_instance_of CallNode, node assert node.block.is_a?(BlockNode) end def test_fallback_define_method - node = Prism.find(Fixtures::DefineMethod.instance_method(:dynamic), rubyvm: false) + node = NodeFind::LineMethodFind.new.find(Fixtures::DefineMethod.instance_method(:dynamic)) assert_instance_of CallNode, node assert node.block.is_a?(BlockNode) end def test_fallback_for_loop - node = Prism.find(Fixtures::ForLoop::FOR_PROC, rubyvm: false) + node = NodeFind::LineProcFind.new.find(Fixtures::ForLoop::FOR_PROC) assert_instance_of ForNode, node end def test_fallback_backtrace_location location = zero_division_location assert_not_nil location - node = Prism.find(location, rubyvm: false) + node = NodeFind::LineBacktraceLocationFind.new.find(location) assert_not_nil node assert_equal location.lineno, node.location.start_line end From 168eada4353f3059e5f7c37f9f7b0786b9899745 Mon Sep 17 00:00:00 2001 From: Koichi Sasada Date: Fri, 14 Aug 2026 18:53:14 +0000 Subject: [PATCH 2/3] thread: replace the timed-waiter list with a hierarchical timer wheel timer_thread_register_waiting kept the timed waiters on a single list sorted by deadline, inserting with a linear scan. A server whose connections share one timeout value -- every keep-alive loop -- makes each insert walk the whole list: with 8000 waiting connections the scan burns half the CPU (43.71% + 8.30% in thread_sched_wait_events) and M:N falls behind 1:1. Bucket timed waiters into a hierarchical timing wheel instead (Varghese & Lauck, "Hashed and Hierarchical Timing Wheels: Data Structures for the Efficient Implementation of a Timer Facility", SOSP '87): 4 levels of 64 slots, 1ms slots at the finest level and 64x coarser per level, ~4.6h of direct range; farther deadlines park on the coarsest level's farthest slot and re-sort on arrival. A not-yet-due waiter found while draining a slot is re-bucketed onto a finer level, so each waiter moves at most once per level. Deadlines are floored to a slot, so the wheel never fires late, and the re-bucketing keeps sleep precision. Insertion becomes O(1) arithmetic. Removal -- the most frequent operation, since most timed fd waits are cancelled by their fd becoming ready -- stays O(1): the cached next-expiry is a lower bound that only insertion has to maintain (waking early is harmless, so removal does no bookkeeping at all). The timer thread is now woken only when an insertion shortens that bound, instead of on every insertion. Untimed (fd-only) waiters move to their own plain list. The wheel and its operations are compiled only with USE_MN_THREADS; the slot width can be widened with -DTIMER_WHEEL_TICK_MS=n for experiments. Co-Authored-By: Claude Fable 5 --- thread_pthread.c | 145 +++------------- thread_pthread.h | 7 +- thread_pthread_mn.c | 408 ++++++++++++++++++++++++++++++++++++++------ 3 files changed, 388 insertions(+), 172 deletions(-) diff --git a/thread_pthread.c b/thread_pthread.c index 4480d2ff90f131..b5de62c37db882 100644 --- a/thread_pthread.c +++ b/thread_pthread.c @@ -3091,9 +3091,22 @@ static struct { struct kevent finished_events[KQUEUE_EVENTS_MAX]; #endif - // waiting threads list - struct ccan_list_head waiting; // waiting threads in ractors +#if USE_MN_THREADS + /* Timed waiters, bucketed by deadline into a hierarchical timer wheel; + * untimed (fd-only) waiters keep a plain list. Both under waiting_lock. + * The wheel operations all live in thread_pthread_mn.c (timer_wheel_*). */ +#define TIMER_WHEEL_LEVELS 4 +#define TIMER_WHEEL_SLOT_BITS 6 +#define TIMER_WHEEL_SLOTS (1 << TIMER_WHEEL_SLOT_BITS) + struct timer_wheel_level { + uint64_t occupied; // bit s: slots[s] is non-empty + struct ccan_list_head slots[TIMER_WHEEL_SLOTS]; + } wheel[TIMER_WHEEL_LEVELS]; + uint64_t wheel_cursor_tick; // slots for ticks <= this are drained + rb_hrtime_t next_expiry; // never later than the earliest deadline + struct ccan_list_head waiting_untimed; pthread_mutex_t waiting_lock; +#endif #if (HAVE_SYS_EPOLL_H || HAVE_SYS_EVENT_H) && USE_MN_THREADS // fd -> struct rb_fd_waiters, in chunks so entries never move. @@ -3109,22 +3122,9 @@ static struct { static void timer_thread_check_timeslice(rb_vm_t *vm); static int timer_thread_set_timeout(rb_vm_t *vm); -static void timer_thread_wakeup_thread(rb_thread_t *th, uint32_t event_serial); -static rb_thread_t *thread_sched_waiting_thread(struct rb_thread_sched_waiting *w); #include "thread_pthread_mn.c" -static rb_thread_t * -thread_sched_waiting_thread(struct rb_thread_sched_waiting *w) -{ - if (w) { - return (rb_thread_t *)((size_t)w - offsetof(rb_thread_t, sched.waiting_reason)); - } - else { - return NULL; - } -} - static int timer_thread_set_timeout(rb_vm_t *vm) { @@ -3154,31 +3154,7 @@ timer_thread_set_timeout(rb_vm_t *vm) } ractor_sched_unlock(vm, NULL); - // Always check waiting threads to find minimum timeout - // even when scheduler has work (grq_cnt > 0) - rb_native_mutex_lock(&timer_th.waiting_lock); - { - struct rb_thread_sched_waiting *w = ccan_list_top(&timer_th.waiting, struct rb_thread_sched_waiting, node); - rb_thread_t *th = thread_sched_waiting_thread(w); - - if (th && (th->sched.waiting_reason.flags & thread_sched_waiting_timeout)) { - rb_hrtime_t now = rb_hrtime_now(); - rb_hrtime_t hrrel = rb_hrtime_sub(th->sched.waiting_reason.data.timeout, now); - - RUBY_DEBUG_LOG("th:%u now:%lu rel:%lu", rb_th_serial(th), (unsigned long)now, (unsigned long)hrrel); - - rb_hrtime_t msec = (hrrel + RB_HRTIME_PER_MSEC - 1) / RB_HRTIME_PER_MSEC; - // A deadline further away than INT_MAX ms must clamp, not truncate: - // a negative timeout would be an untimed epoll_wait. - int thread_timeout = msec > INT_MAX ? INT_MAX : (int)msec; // ms - - // Use minimum of scheduler timeout and thread sleep timeout - if (timeout < 0 || thread_timeout < timeout) { - timeout = thread_timeout; - } - } - } - rb_native_mutex_unlock(&timer_th.waiting_lock); + timeout = timer_wheel_timeout(timeout); RUBY_DEBUG_LOG("timeout:%d inf:%d", timeout, (int)vm->ractor.sched.timeslice_wait_inf); @@ -3199,83 +3175,6 @@ timer_thread_check_signal(rb_vm_t *vm) } } -static bool -timer_thread_check_exceed(rb_hrtime_t abs, rb_hrtime_t now) -{ - return abs <= now; -} - -static rb_thread_t * -timer_thread_deq_wakeup(rb_vm_t *vm, rb_hrtime_t now, uint32_t *event_serial) -{ - struct rb_thread_sched_waiting *w = ccan_list_top(&timer_th.waiting, struct rb_thread_sched_waiting, node); - - if (w != NULL && - (w->flags & thread_sched_waiting_timeout) && - timer_thread_check_exceed(w->data.timeout, now)) { - - RUBY_DEBUG_LOG("wakeup th:%u", rb_th_serial(thread_sched_waiting_thread(w))); - - // delete from waiting list - ccan_list_del_init(&w->node); - - rb_thread_t *th = thread_sched_waiting_thread(w); - -#if (HAVE_SYS_EPOLL_H || HAVE_SYS_EVENT_H) && USE_MN_THREADS - // An fd+timeout waiter is also on its fd's waiter list; leave it there too. - timer_thread_unregister_waiting(th, w->data.fd, w->flags); -#endif - - // setup result - w->flags = thread_sched_waiting_none; - w->data.result = 0; - - *event_serial = w->data.event_serial; - return th; - } - - return NULL; -} - -static void -timer_thread_wakeup_thread_locked(struct rb_thread_sched *sched, rb_thread_t *th, uint32_t event_serial) -{ - if (sched->running != th && th->sched.event_serial == event_serial) { - thread_sched_to_ready_common(sched, th, true, false); - } -} - -static void -timer_thread_wakeup_thread(rb_thread_t *th, uint32_t event_serial) -{ - RUBY_DEBUG_LOG("th:%u", rb_th_serial(th)); - struct rb_thread_sched *sched = TH_SCHED(th); - - thread_sched_lock(sched, th); - { - timer_thread_wakeup_thread_locked(sched, th, event_serial); - } - thread_sched_unlock(sched, th); -} - -static void -timer_thread_check_timeout(rb_vm_t *vm) -{ - rb_hrtime_t now = rb_hrtime_now(); - rb_thread_t *th; - uint32_t event_serial; - - rb_native_mutex_lock(&timer_th.waiting_lock); - { - while ((th = timer_thread_deq_wakeup(vm, now, &event_serial)) != NULL) { - rb_native_mutex_unlock(&timer_th.waiting_lock); - timer_thread_wakeup_thread(th, event_serial); - rb_native_mutex_lock(&timer_th.waiting_lock); - } - } - rb_native_mutex_unlock(&timer_th.waiting_lock); -} - static void timer_thread_check_timeslice(rb_vm_t *vm) { @@ -3420,8 +3319,18 @@ rb_thread_create_timer_thread(void) // starts over. } - ccan_list_head_init(&timer_th.waiting); +#if USE_MN_THREADS + for (int lvl = 0; lvl < TIMER_WHEEL_LEVELS; lvl++) { + timer_th.wheel[lvl].occupied = 0; + for (int slot = 0; slot < TIMER_WHEEL_SLOTS; slot++) { + ccan_list_head_init(&timer_th.wheel[lvl].slots[slot]); + } + } + timer_th.wheel_cursor_tick = timer_wheel_tick(rb_hrtime_now()); + timer_th.next_expiry = TIMER_WHEEL_NO_EXPIRY; + ccan_list_head_init(&timer_th.waiting_untimed); rb_native_mutex_initialize(&timer_th.waiting_lock); +#endif // open communication channel setup_communication_pipe_internal(timer_th.comm_fds); diff --git a/thread_pthread.h b/thread_pthread.h index a8821e1aa9ad38..170c9309fa32f1 100644 --- a/thread_pthread.h +++ b/thread_pthread.h @@ -52,9 +52,14 @@ struct rb_thread_sched_waiting { int result; } data; - // connected to timer_th.waiting (ordered by timeout) + // connected to a timer_th wheel slot (timed) or timer_th.waiting_untimed struct ccan_list_node node; + /* which wheel slot `node` is on; meaningful only while flags has + * thread_sched_waiting_timeout */ + uint8_t wheel_lvl; + uint8_t wheel_slot; + // connected to rb_fd_waiters.waiters of data.fd struct ccan_list_node fd_node; }; diff --git a/thread_pthread_mn.c b/thread_pthread_mn.c index d8e8fb37caf5b0..938ac1001f4b38 100644 --- a/thread_pthread_mn.c +++ b/thread_pthread_mn.c @@ -2,8 +2,306 @@ #if USE_MN_THREADS +#if HAVE_SYS_EPOLL_H || HAVE_SYS_EVENT_H static void timer_thread_unregister_waiting(rb_thread_t *th, int fd, enum thread_sched_waiting_flag flags); -static void timer_thread_wakeup_thread_locked(struct rb_thread_sched *sched, rb_thread_t *th, uint32_t event_serial); +#endif + +static bool +timer_thread_check_exceed(rb_hrtime_t abs, rb_hrtime_t now) +{ + return abs <= now; +} + +static rb_thread_t * +thread_sched_waiting_thread(struct rb_thread_sched_waiting *w) +{ + if (w) { + return (rb_thread_t *)((size_t)w - offsetof(rb_thread_t, sched.waiting_reason)); + } + else { + return NULL; + } +} + +#define TIMER_WHEEL_NO_EXPIRY RB_HRTIME_MAX +#ifndef TIMER_WHEEL_TICK_MS +#define TIMER_WHEEL_TICK_MS 1 // L0 slot width; coarser trades sleep accuracy for fewer drains +#endif + +/* Timer wheel over the timed waiters. Every operation runs under + * timer_th.waiting_lock. + * + * Level L buckets deadlines into 64 slots of 64^L ms each; a deadline is + * placed by its distance from the drain cursor, so its slot is always + * strictly ahead of the cursor at that level. The drain refines a + * not-yet-due waiter onto a finer level instead of cascading whole slots, + * so one waiter moves at most once per level over its lifetime. + * + * This is the hierarchical timing wheel of Varghese & Lauck, "Hashed and + * Hierarchical Timing Wheels" (SOSP '87). */ + +static uint64_t +timer_wheel_tick(rb_hrtime_t hrt) +{ + return hrt / (RB_HRTIME_PER_MSEC * TIMER_WHEEL_TICK_MS); +} + +static inline int +timer_wheel_ctz64(uint64_t v) +{ +#if defined(__GNUC__) || defined(__clang__) + return __builtin_ctzll(v); +#else + int n = 0; + while (!(v & 1)) { v >>= 1; n++; } + return n; +#endif +} + +static int +timer_wheel_level(uint64_t dist_ms) +{ + if (dist_ms < ((uint64_t)1 << TIMER_WHEEL_SLOT_BITS)) return 0; + if (dist_ms < ((uint64_t)1 << 2 * TIMER_WHEEL_SLOT_BITS)) return 1; + if (dist_ms < ((uint64_t)1 << 3 * TIMER_WHEEL_SLOT_BITS)) return 2; + return TIMER_WHEEL_LEVELS - 1; +} + +static void +timer_wheel_insert(struct rb_thread_sched_waiting *w) +{ + uint64_t dl_tick = timer_wheel_tick(w->data.timeout); + uint64_t cur = timer_th.wheel_cursor_tick; + + /* A deadline at or behind the cursor parks one tick ahead: its own slot + * is already drained and would otherwise wait out a full wheel turn. */ + uint64_t target = dl_tick > cur ? dl_tick : cur + 1; + int lvl = timer_wheel_level(target - cur); + int shift = lvl * TIMER_WHEEL_SLOT_BITS; + uint64_t slot_tick = target >> shift; + + if (lvl == TIMER_WHEEL_LEVELS - 1) { + /* Beyond the wheel range: park on the farthest slot; the drain + * re-inserts it with the then-smaller distance. */ + uint64_t far = (cur >> shift) + TIMER_WHEEL_SLOTS - 1; + if (slot_tick > far) slot_tick = far; + } + + int slot = (int)(slot_tick & (TIMER_WHEEL_SLOTS - 1)); + + ccan_list_add_tail(&timer_th.wheel[lvl].slots[slot], &w->node); + timer_th.wheel[lvl].occupied |= UINT64_C(1) << slot; + w->wheel_lvl = (uint8_t)lvl; + w->wheel_slot = (uint8_t)slot; + + if (w->data.timeout < timer_th.next_expiry) { + timer_th.next_expiry = w->data.timeout; + } +} + +// Unlink a waiter from the wheel slot or the untimed list it is on. +static void +timer_wheel_del(struct rb_thread_sched_waiting *w) +{ + ccan_list_del_init(&w->node); + + if (w->flags & thread_sched_waiting_timeout) { + struct timer_wheel_level *lv = &timer_th.wheel[w->wheel_lvl]; + if (ccan_list_empty(&lv->slots[w->wheel_slot])) { + lv->occupied &= ~(UINT64_C(1) << w->wheel_slot); + } + } +} + +/* A lower bound on the earliest deadline: the start time of the nearest + * occupied slot per level. Never later than any real deadline, so waking + * by it is at worst early, which is harmless. */ +static rb_hrtime_t +timer_wheel_next_expiry(void) +{ + rb_hrtime_t best = TIMER_WHEEL_NO_EXPIRY; + + for (int lvl = 0; lvl < TIMER_WHEEL_LEVELS; lvl++) { + uint64_t occ = timer_th.wheel[lvl].occupied; + if (!occ) continue; + + int shift = lvl * TIMER_WHEEL_SLOT_BITS; + uint64_t cur_tick = timer_th.wheel_cursor_tick >> shift; + unsigned base = (unsigned)((cur_tick + 1) & (TIMER_WHEEL_SLOTS - 1)); + // rotate so bit k = slot for tick cur_tick+1+k + uint64_t rot = (occ >> base) | (base ? (occ << (TIMER_WHEEL_SLOTS - base)) : 0); + uint64_t tick = cur_tick + 1 + timer_wheel_ctz64(rot); + rb_hrtime_t start = (rb_hrtime_t)(tick << shift) * RB_HRTIME_PER_MSEC * TIMER_WHEEL_TICK_MS; + + if (start < best) best = start; + } + + return best; +} + +/* Drain every slot whose tick moved behind `now`, collecting due waiters + * onto `expired` and re-bucketing not-yet-due ones onto a finer level. + * timer_th.waiting_lock must be held. */ +static void +timer_wheel_drain(rb_hrtime_t now, uint64_t now_tick, struct ccan_list_head *expired) +{ + uint64_t prev_tick = timer_th.wheel_cursor_tick; + timer_th.wheel_cursor_tick = now_tick; // re-inserts below map against the new cursor + /* A deadline inside the current tick parks one tick ahead, so its slot + * start is later than the deadline. Keep the exact value: the recompute + * below only knows slot starts, and next_expiry must not exceed a deadline. */ + rb_hrtime_t reinserted = TIMER_WHEEL_NO_EXPIRY; + + for (int lvl = 0; lvl < TIMER_WHEEL_LEVELS; lvl++) { + int shift = lvl * TIMER_WHEEL_SLOT_BITS; + uint64_t from = prev_tick >> shift; + uint64_t to = now_tick >> shift; + + if (to == from) break; // no boundary crossed; coarser levels crossed none either + + uint64_t steps = to - from; + if (steps > TIMER_WHEEL_SLOTS) steps = TIMER_WHEEL_SLOTS; + struct timer_wheel_level *lv = &timer_th.wheel[lvl]; + + for (uint64_t tick = to - steps + 1; tick <= to; tick++) { + int slot = (int)(tick & (TIMER_WHEEL_SLOTS - 1)); + + if (!(lv->occupied & (UINT64_C(1) << slot))) continue; + lv->occupied &= ~(UINT64_C(1) << slot); + + /* Detach first: a far-future waiter re-clamped by the insert below + * can land back on this very slot and must not be drained again. */ + struct ccan_list_head pending; + ccan_list_head_init(&pending); + ccan_list_append_list(&pending, &lv->slots[slot]); + + struct rb_thread_sched_waiting *w; + while ((w = ccan_list_pop(&pending, struct rb_thread_sched_waiting, node)) != NULL) { + if (timer_thread_check_exceed(w->data.timeout, now)) { + rb_thread_t *th = thread_sched_waiting_thread(w); + + RUBY_DEBUG_LOG("wakeup th:%u", rb_th_serial(th)); + +#if HAVE_SYS_EPOLL_H || HAVE_SYS_EVENT_H + // An fd+timeout waiter is also on its fd's waiter list. + timer_thread_unregister_waiting(th, w->data.fd, w->flags); +#endif + /* flags stay set until the wakeup below takes them under + * the lock: a waiter whose flags are already cleared may + * run and re-register through this same `w`, which would + * relink the node we are still holding on `expired`. */ + w->data.result = 0; + + ccan_list_add_tail(expired, &w->node); + } + else { + /* arrived early: the distance shrank, so this lands on a + * finer level (or a farther slot of the coarsest one) */ + if (w->data.timeout < reinserted) reinserted = w->data.timeout; + timer_wheel_insert(w); + } + } + } + } + + timer_th.next_expiry = timer_wheel_next_expiry(); + if (reinserted < timer_th.next_expiry) timer_th.next_expiry = reinserted; +} + +/* Merge the wheel's next expiry into the poll timeout (ms; -1 = none). + * Checked even when the scheduler has other work (grq_cnt > 0). */ +static int +timer_wheel_timeout(int timeout) +{ + rb_native_mutex_lock(&timer_th.waiting_lock); + { + if (timer_th.next_expiry != TIMER_WHEEL_NO_EXPIRY) { + rb_hrtime_t now = rb_hrtime_now(); + rb_hrtime_t hrrel = rb_hrtime_sub(timer_th.next_expiry, now); + + RUBY_DEBUG_LOG("now:%lu rel:%lu", (unsigned long)now, (unsigned long)hrrel); + + rb_hrtime_t msec = (hrrel + RB_HRTIME_PER_MSEC - 1) / RB_HRTIME_PER_MSEC; + // A deadline further away than INT_MAX ms must clamp, not truncate: + // a negative timeout would be an untimed epoll_wait. + int thread_timeout = msec > INT_MAX ? INT_MAX : (int)msec; // ms + + // Use minimum of scheduler timeout and thread sleep timeout + if (timeout < 0 || thread_timeout < timeout) { + timeout = thread_timeout; + } + } + } + rb_native_mutex_unlock(&timer_th.waiting_lock); + + return timeout; +} + +static void +timer_thread_wakeup_thread_locked(struct rb_thread_sched *sched, rb_thread_t *th, uint32_t event_serial) +{ + if (sched->running != th && th->sched.event_serial == event_serial) { + thread_sched_to_ready_common(sched, th, true, false); + } +} + +static void +timer_thread_wakeup_thread(rb_thread_t *th, uint32_t event_serial) +{ + RUBY_DEBUG_LOG("th:%u", rb_th_serial(th)); + struct rb_thread_sched *sched = TH_SCHED(th); + + thread_sched_lock(sched, th); + { + timer_thread_wakeup_thread_locked(sched, th, event_serial); + } + thread_sched_unlock(sched, th); +} + +#define TIMEOUT_WAKE_BATCH 16 + +static void +timer_thread_check_timeout(rb_vm_t *vm) +{ + rb_hrtime_t now = rb_hrtime_now(); + uint64_t now_tick = timer_wheel_tick(now); + struct ccan_list_head expired; + + ccan_list_head_init(&expired); + + struct timeout_wake { rb_thread_t *th; uint32_t serial; } batch[TIMEOUT_WAKE_BATCH]; + bool more = true; + + while (more) { + int n = 0; + + rb_native_mutex_lock(&timer_th.waiting_lock); + { + // A second pass finds the cursor already at now_tick and drains nothing. + if (now_tick > timer_th.wheel_cursor_tick) { + timer_wheel_drain(now, now_tick, &expired); + } + + struct rb_thread_sched_waiting *w; + while (n < TIMEOUT_WAKE_BATCH && + (w = ccan_list_pop(&expired, struct rb_thread_sched_waiting, node)) != NULL) { + // Name the thread and its serial here, then release it: once the + // flags are clear the thread may run and re-register through `w`, + // and a serial read after that would match the new registration. + batch[n].th = thread_sched_waiting_thread(w); + batch[n].serial = w->data.event_serial; + w->flags = thread_sched_waiting_none; + n++; + } + more = !ccan_list_empty(&expired); + } + 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); + } + } +} static bool timer_thread_cancel_waiting(rb_thread_t *th) @@ -14,7 +312,7 @@ timer_thread_cancel_waiting(rb_thread_t *th) { if (th->sched.waiting_reason.flags) { canceled = true; - ccan_list_del_init(&th->sched.waiting_reason.node); + timer_wheel_del(&th->sched.waiting_reason); timer_thread_unregister_waiting(th, th->sched.waiting_reason.data.fd, th->sched.waiting_reason.flags); th->sched.waiting_reason.flags = thread_sched_waiting_none; } @@ -672,24 +970,8 @@ native_thread_create_shared(rb_thread_t *th) return 0; } -#else // USE_MN_THREADS - -static int -native_thread_create_shared(rb_thread_t *th) -{ - rb_bug("unreachable"); -} - -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) -{ - rb_bug("unreachable"); -} - -#endif // USE_MN_THREADS - /// EPOLL/KQUEUE specific code -#if (HAVE_SYS_EPOLL_H || HAVE_SYS_EVENT_H) && USE_MN_THREADS +#if HAVE_SYS_EPOLL_H || HAVE_SYS_EVENT_H /// Per-fd waiter table (struct rb_fd_waiters). One fd may have several waiters /// -- a reader and a writer on one socket -- so the backend is armed with their @@ -891,18 +1173,27 @@ static void verify_waiting_list(void) { #if VM_CHECK_MODE > 0 - struct rb_thread_sched_waiting *w, *prev_w = NULL; + struct rb_thread_sched_waiting *w; + + for (int lvl = 0; lvl < TIMER_WHEEL_LEVELS; lvl++) { + const struct timer_wheel_level *lv = &timer_th.wheel[lvl]; - // waiting list's timeout order should be [1, 2, 3, ..., 0, 0, 0] + for (int slot = 0; slot < TIMER_WHEEL_SLOTS; slot++) { + bool occupied = (lv->occupied >> slot) & 1; + VM_ASSERT(occupied == !ccan_list_empty(&lv->slots[slot])); - ccan_list_for_each(&timer_th.waiting, w, node) { - // fprintf(stderr, "verify_waiting_list th:%u abs:%lu\n", rb_th_serial(wth), (unsigned long)wth->sched.waiting_reason.data.timeout); - if (prev_w) { - rb_hrtime_t timeout = w->data.timeout; - rb_hrtime_t prev_timeout = prev_w->data.timeout; - VM_ASSERT(timeout == 0 || prev_timeout <= timeout); + ccan_list_for_each(&lv->slots[slot], w, node) { + VM_ASSERT(w->flags & thread_sched_waiting_timeout); + VM_ASSERT(w->data.timeout != 0); + VM_ASSERT(w->wheel_lvl == lvl); + VM_ASSERT(w->wheel_slot == slot); + } } - prev_w = w; + } + + ccan_list_for_each(&timer_th.waiting_untimed, w, node) { + VM_ASSERT(!(w->flags & thread_sched_waiting_timeout)); + VM_ASSERT(w->data.timeout == 0); } #endif } @@ -1036,36 +1327,21 @@ timer_thread_register_waiting(rb_thread_t *th, int fd, enum thread_sched_waiting if (abs == 0) { // no timeout VM_ASSERT(!(flags & thread_sched_waiting_timeout)); - ccan_list_add_tail(&timer_th.waiting, &th->sched.waiting_reason.node); + ccan_list_add_tail(&timer_th.waiting_untimed, &th->sched.waiting_reason.node); } else { RUBY_DEBUG_LOG("abs:%lu", (unsigned long)abs); VM_ASSERT(flags & thread_sched_waiting_timeout); - // insert th to sorted list (TODO: O(n)) - struct rb_thread_sched_waiting *w, *prev_w = NULL; - - ccan_list_for_each(&timer_th.waiting, w, node) { - if ((w->flags & thread_sched_waiting_timeout) && - w->data.timeout < abs) { - prev_w = w; - } - else { - break; - } - } - - if (prev_w) { - ccan_list_add_after(&timer_th.waiting, &prev_w->node, &th->sched.waiting_reason.node); - } - else { - ccan_list_add(&timer_th.waiting, &th->sched.waiting_reason.node); - } + rb_hrtime_t prev_expiry = timer_th.next_expiry; + timer_wheel_insert(&th->sched.waiting_reason); verify_waiting_list(); - // update timeout seconds; force wake so timer thread notices short deadlines - timer_thread_wakeup_force(); + if (timer_th.next_expiry < prev_expiry) { + // an earlier deadline than the timer thread is armed for + timer_thread_wakeup_force(); + } } } else { @@ -1184,7 +1460,7 @@ timer_thread_wake_fd_waiters(int fd, uint32_t generation, uint32_t wake_flags, i } ccan_list_del_init(&w->fd_node); - ccan_list_del_init(&w->node); // also leaves the timeout list + timer_wheel_del(w); // also leaves the timer wheel w->flags = thread_sched_waiting_none; w->data.fd = -1; @@ -1320,7 +1596,9 @@ timer_thread_polling(rb_vm_t *vm) } } -#else // HAVE_SYS_EPOLL_H || HAVE_SYS_EVENT_H +#endif // HAVE_SYS_EPOLL_H || HAVE_SYS_EVENT_H + +#else // USE_MN_THREADS static void timer_thread_setup_mn(void) @@ -1370,4 +1648,28 @@ timer_thread_polling(rb_vm_t *vm) } } -#endif // HAVE_SYS_EPOLL_H || HAVE_SYS_EVENT_H +static int +native_thread_create_shared(rb_thread_t *th) +{ + rb_bug("unreachable"); +} + +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) +{ + rb_bug("unreachable"); +} + +static int +timer_wheel_timeout(int timeout) +{ + return timeout; // no M:N threads, no timed waiters +} + +static void +timer_thread_check_timeout(rb_vm_t *vm) +{ + // no M:N threads, no timed waiters +} + +#endif // USE_MN_THREADS From d8f0457c5e4b323ffbe00a49b3473e5f48af544d Mon Sep 17 00:00:00 2001 From: Koichi Sasada Date: Sun, 16 Aug 2026 05:11:33 +0000 Subject: [PATCH 3/3] Load did_you_mean before the GC compact stress block TestException#test_detailed_message_under_gc_compact_stress calls Exception#detailed_message inside EnvUtil.under_gc_compact_stress. The first error display of a process lazily requires the decoration gems (error_highlight, did_you_mean, syntax_suggest -- error.c's require_decoration_gems), so the whole library load happened inside the block, where GC.stress makes every allocation a full mark+compact. The cost of each of those collections is proportional to the live heap, so the load does not just take a while, it degrades with the age of the worker. Measured here (x86_64, idle, gems enabled): live slots block 14k 13.7s 114k 61.8s 414k 222.9s A test-all worker that has already run other files sits far above that, and the parallel runner kills a worker that has not responded for 1200s (worker_timeout) with SIGSEGV -- which is what CI reported: worker 63824=ruby/test_exception does not respond; SEGV is sent 1) Timeout: TestException#test_detailed_message_under_gc_compact_stress rb_crash_63824.txt: lib/did_you_mean/spell_checker.rb: [BUG] Segmentation fault at 0x000003e90000f94a test/ruby/test_exception.rb:1484:in 'detailed_message' lib/did_you_mean.rb:4:in '' lib/did_you_mean.rb:4:in 'require_relative' (The reported "address" is not a fault address: for a signal sent by another process siginfo's si_addr aliases si_pid/si_uid, so it reads back as (uid << 32) | sender_pid -- 0x3e9 is uid 1001, the runner user.) Whether this test took milliseconds or ran past the timeout depended on whether an earlier file in the same worker had already loaded did_you_mean, which made it look like a random hang. It only bites where the decoration gems are enabled: `make test-all` passes --disable-gems through RUN_OPTS, but the ZJIT/YJIT jobs override RUN_OPTS, so gems are on there. Do the load before the block; this test is about detailed_message, not about require. With gems enabled, the block goes from 28.7s to 0.22s; the --disable-gems path is unchanged (0.17s). CI: https://github.com/ruby/ruby/actions/runs/31860169748/job/94952042861 Co-Authored-By: Claude Opus 5 (1M context) --- test/ruby/test_exception.rb | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/test/ruby/test_exception.rb b/test/ruby/test_exception.rb index 0dc91f9abc4f43..9790f9ec9c46fd 100644 --- a/test/ruby/test_exception.rb +++ b/test/ruby/test_exception.rb @@ -1479,6 +1479,12 @@ def test_detailed_message def test_detailed_message_under_gc_compact_stress omit "compaction doesn't work well on s390x" if RUBY_PLATFORM =~ /s390x/ # https://github.com/ruby/ruby/pull/5077 + + # The first error display lazily requires did_you_mean and friends; inside the + # block that library load costs one full mark+compact per allocation, enough to + # trip the parallel runner's no-response timeout. Load it here instead. + RuntimeError.new("").detailed_message + EnvUtil.under_gc_compact_stress do e = RuntimeError.new("foo\nbar\nbaz") assert_equal("foo (RuntimeError)\nbar\nbaz", e.detailed_message)