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 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) 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