From 93a521bf3a16bc48eee17717e867d7364f4a1ad2 Mon Sep 17 00:00:00 2001 From: Jean Boussier Date: Sun, 23 Aug 2026 13:34:21 +0200 Subject: [PATCH 1/4] hash.c: split allocation and st_init `st_replace` assumes that the `st_table` hasn't been initialized, so when duping a hash, we shouldn't pre-allocate the `st_table` otherwise we'll leak memory. --- hash.c | 44 +++++++++++++++++++++++++----------------- imemo.c | 3 +++ st.c | 16 +++++++++++---- test/ruby/test_hash.rb | 30 ++++++++++++++++++++++++++++ 4 files changed, 71 insertions(+), 22 deletions(-) diff --git a/hash.c b/hash.c index e8e97a80f80e75..417e7a5023e127 100644 --- a/hash.c +++ b/hash.c @@ -1501,6 +1501,20 @@ hash_alloc_capa(VALUE klass, VALUE flags, VALUE ifnone, size_t size, bool frozen { VALUE hash = rb_newobj_of(klass, T_HASH | flags, hash_slot_size(size, frozen)); rb_hash_set_ifnone(hash, ifnone); + +#ifdef RUBY_DEBUG + if (hash_slot_size(size, frozen) >= sizeof(struct RHash) + sizeof(st_table)) { + RHASH_ST_TABLE(hash)->num_entries = 0; + RHASH_ST_TABLE(hash)->entries = NULL; + } +#endif + + return hash; +} + +static VALUE +hash_init_capa(VALUE hash, size_t size) +{ if (size > RHASH_AR_TABLE_MAX_SIZE) { hash_st_table_init(hash, &objhash, size); } @@ -1510,7 +1524,7 @@ hash_alloc_capa(VALUE klass, VALUE flags, VALUE ifnone, size_t size, bool frozen static VALUE hash_hidden_new(size_t size) { - return hash_alloc_capa(0, 0, Qnil, size, false); + return hash_init_capa(hash_alloc_capa(0, 0, Qnil, size, false), size); } VALUE @@ -1554,7 +1568,7 @@ copy_compare_by_id(VALUE hash, VALUE basis) VALUE rb_hash_new_capa(long capa) { - return hash_alloc_capa(rb_cHash, 0, Qnil, capa, false); + return hash_init_capa(hash_alloc_capa(rb_cHash, 0, Qnil, capa, false), capa); } VALUE @@ -1566,12 +1580,14 @@ rb_hash_new(void) VALUE rb_hash_alloc_fixed_size(VALUE klass, st_index_t size) { - return hash_alloc_capa(klass, 0, Qnil, size, true); + return hash_init_capa(hash_alloc_capa(klass, 0, Qnil, size, true), size); } static VALUE hash_copy(VALUE ret, VALUE hash) { + RUBY_ASSERT(RHASH_SIZE(ret) == 0); + if (rb_hash_compare_by_id_p(hash)) { rb_gc_register_pinning_obj(ret); } @@ -1582,6 +1598,12 @@ hash_copy(VALUE ret, VALUE hash) } else { st_table *tab = RHASH_ST_TABLE(ret); + + // If `hash` is an ar_table it can't be `compare_by_identity?`. + RUBY_ASSERT(!rb_hash_compare_by_id_p(hash)); + RUBY_ASSERT(RHASH_ST_TABLE(ret)->entries == NULL); + st_init_existing_table_with_size(RHASH_ST_TABLE(ret), &objhash, RHASH_SIZE(hash)); + int bound = RHASH_AR_TABLE_BOUND(hash); for (int i = 0; i < bound; i++) { if (ar_cleared_entry(hash, i)) continue; @@ -1594,11 +1616,8 @@ hash_copy(VALUE ret, VALUE hash) } } else { - HASH_ASSERT(sizeof(st_table) <= sizeof(ar_table)); - RHASH_SET_ST_FLAG(ret); st_replace(RHASH_ST_TABLE(ret), RHASH_ST_TABLE(hash)); - rb_gc_writebarrier_remember(ret); } return ret; @@ -1611,9 +1630,6 @@ hash_dup_with_compare_by_id(VALUE hash) if (RHASH_ST_TABLE_P(hash)) { RHASH_SET_ST_FLAG(dup); } - else { - RHASH_UNSET_ST_FLAG(dup); - } return hash_copy(dup, hash); } @@ -1639,8 +1655,7 @@ rb_hash_dup(VALUE hash) VALUE rb_hash_resurrect(VALUE hash) { - VALUE ret = hash_dup(hash, rb_cHash, 0); - return ret; + return hash_dup(hash, rb_cHash, 0); } #if USE_ZJIT @@ -3082,18 +3097,11 @@ rb_hash_replace(VALUE hash, VALUE hash2) if (RHASH_AR_TABLE_P(hash)) { hash_ar_free_and_clear_table(hash); - if (RHASH_SIZE(hash2) > RHASH_AR_TABLE_MAX_SIZE) { - RHASH_SET_ST_FLAG(hash); - } } else { hash_st_free_and_clear_table(hash); } - if (RHASH_ST_TABLE_P(hash)) { - st_init_existing_table_with_size(RHASH_ST_TABLE(hash), &objhash, RHASH_SIZE(hash2)); - } - hash_copy(hash, hash2); return hash; diff --git a/imemo.c b/imemo.c index 44b8bd67073b50..890b4ee9447a4a 100644 --- a/imemo.c +++ b/imemo.c @@ -215,6 +215,9 @@ rb_imemo_fields_clone(VALUE fields_obj) // to mark an uninitialized table. clone = imemo_fields_new(owner, ROOT_SHAPE_ID, sizeof(struct rb_fields), false /* TODO: check */); st_table *dest_table = rb_imemo_fields_complex_tbl(clone); +#ifdef RUBY_DEBUG + dest_table->entries = NULL; +#endif st_replace(dest_table, src_table); st_foreach(dest_table, imemo_fields_complex_wb_i, (st_data_t)clone); RBASIC_SET_FULL_SHAPE_ID(clone, shape_id); diff --git a/st.c b/st.c index fb89c21329d90e..f7b870fece49d0 100644 --- a/st.c +++ b/st.c @@ -1352,9 +1352,8 @@ st_insert2(st_table *tab, st_data_t key, st_data_t value, return 1; } -/* Create a copy of old_tab into new_tab. */ -st_table * -st_replace(st_table *new_tab, st_table *old_tab) +static st_table * +st_replace_no_check(st_table *new_tab, st_table *old_tab) { *new_tab = *old_tab; size_t memsize = get_allocated_entries(old_tab) * sizeof(st_table_entry); @@ -1370,6 +1369,15 @@ st_replace(st_table *new_tab, st_table *old_tab) return new_tab; } + +/* Create a copy of old_tab into new_tab. */ +st_table * +st_replace(st_table *new_tab, st_table *old_tab) +{ + RUBY_ASSERT(new_tab->entries == NULL); + return st_replace_no_check(new_tab, old_tab); +} + /* Create and return a copy of table OLD_TAB. */ st_table * st_copy(st_table *old_tab) @@ -1382,7 +1390,7 @@ st_copy(st_table *old_tab) return NULL; #endif - if (st_replace(new_tab, old_tab) == NULL) { + if (st_replace_no_check(new_tab, old_tab) == NULL) { st_free_table(new_tab); return NULL; } diff --git a/test/ruby/test_hash.rb b/test/ruby/test_hash.rb index a6b7f26dd6e6cd..6b4c72db7fb771 100644 --- a/test/ruby/test_hash.rb +++ b/test/ruby/test_hash.rb @@ -2107,6 +2107,36 @@ def test_replace_st_with_ar assert_equal(h2, h1) end + def test_replace_ar_with_st + # AR hash + h1 = { a: 1, b: 2, c: 3, d: 4, e: 5, f: 6, g: 7 } + # ST hash + h2 = { a: 1, b: 2, c: 3, d: 4, e: 5, f: 6, g: 7, h: 8, i: 9 } + # Replace AR hash with ST hash + h1.replace(h2) + assert_equal(h2, h1) + end + + def test_replace_ar_with_ar + # AR hash + h1 = { a: 1, b: 2 } + # AR hash + h2 = { a: 1 } + # Replace AR hash with AR hash + h1.replace(h2) + assert_equal(h2, h1) + end + + def test_replace_st_with_st + # ST hash + h1 = { a: 1, b: 2, c: 3, d: 4, e: 5, f: 6, g: 7, h: 8, i: 9 } + # ST hash + h2 = { a: 9, b: 2, c: 3, d: 4, e: 5, f: 6, g: 7, h: 8, i: 9 } + # Replace ST hash with AR hash + h1.replace(h2) + assert_equal(h2, h1) + end + def test_nil_to_h h = nil.to_h assert_equal({}, h) From e896324f3b84bc9be138810ff380fab449e84f2b Mon Sep 17 00:00:00 2001 From: Nobuyoshi Nakada Date: Sun, 23 Aug 2026 23:17:39 +0900 Subject: [PATCH 2/4] sync_default_gems.rb: Exclude well known binary file suffixes --- tool/sync_default_gems.rb | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tool/sync_default_gems.rb b/tool/sync_default_gems.rb index 4e87e12bc8b19d..618b6b82df8ce2 100755 --- a/tool/sync_default_gems.rb +++ b/tool/sync_default_gems.rb @@ -21,6 +21,7 @@ module SyncDefaultGems # exclude: [ "fnmatch_pattern_after_mapping", ... ] Repository = Data.define(:upstream, :branch, :mappings, :exclude) do def excluded?(newpath) + return true if newpath.end_with?(*%w".a .bundle .dll .dylib .so .o .obj") p = newpath until p == "." return true if exclude.any? {|pat| File.fnmatch?(pat, p, File::FNM_PATHNAME|File::FNM_EXTGLOB)} @@ -114,8 +115,6 @@ def lib((upstream, branch), gemspec_in_subdir: false) ["lib", "ext/date/lib"], ["test/date", "test/date"], ["date.gemspec", "ext/date/date.gemspec"], - ], exclude: [ - "ext/date/lib/date_core.bundle", ]), delegate: lib("ruby/delegate"), did_you_mean: repo("ruby/did_you_mean", [ @@ -227,7 +226,6 @@ def lib((upstream, branch), gemspec_in_subdir: false) "ext/psych/lib/org", "ext/psych/lib/psych.jar", "ext/psych/lib/psych_jars.rb", - "ext/psych/lib/psych.{bundle,so}", "ext/psych/lib/2.*", "ext/psych/yaml/LICENSE", "ext/psych/.gitignore", From fecb62c147369b6bde57354cff9e5e7536c4b640 Mon Sep 17 00:00:00 2001 From: Nobuyoshi Nakada Date: Sun, 23 Aug 2026 23:22:58 +0900 Subject: [PATCH 3/4] encoding.c does not benlong to Onigmo --- tool/sync_default_gems.rb | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tool/sync_default_gems.rb b/tool/sync_default_gems.rb index 618b6b82df8ce2..35246e56c84c4c 100755 --- a/tool/sync_default_gems.rb +++ b/tool/sync_default_gems.rb @@ -79,6 +79,8 @@ def lib((upstream, branch), gemspec_in_subdir: false) ["regsyntax.c", "regsyntax.c"], ["onigmo.h", "include/ruby/onigmo.h"], ["enc", "enc"], + ], exclude: [ + "encoding.c", ]), "io-console": repo("ruby/io-console", [ ["ext/io/console", "ext/io/console"], From 8d46d62fb08e0cf6520fbe172841b80501e42524 Mon Sep 17 00:00:00 2001 From: Koichi Sasada Date: Sun, 23 Aug 2026 09:20:02 +0000 Subject: [PATCH 4/4] thread: keep context switches off the scheduler lock thread_sched_setup_running_threads runs on every context switch (park, resume, and both ends of every blocking region) and took the global ractor.sched.lock to maintain the running_threads and timeslice lists and running_cnt. With many ractors switching concurrently this lock is the scaling ceiling. Replace all three with ntlist, a per-location registration of who runs where, and no shared counter: - a thread on a shared nt registers in that nt's running_th, under the nt's own lock -- uncontended, no cross-core traffic - a dedicated nt goes on ntlist.running_dnts while its thread runs (the switch rate of dedicated threads is what it always was; only the lock shrank), so the fast path writes no shared cache line at all The consumers walk the registrations: the barrier's interrupt walk also counts them into a snapshot (barrier_running_cnt) and stamps each nt with the barrier serial; a stamped nt deregistering during that barrier decrements the snapshot under sched.lock, so the completion check stays O(1) however many runners there are. The timeslice list becomes timeslice.scheds, the scheds whose readyq holds waiters. The enq that fills an empty readyq (and the add path taking a turn with waiters already queued) links the sched; the timer thread interrupts each listed sched's running thread and prunes entries whose readyq drained. A steadily contended sched stays linked, so the switch path never touches the list, and idle scheds are not walked at all. The timer takes each sched's lock by trylock, because the switchers nest sched.lock -> timeslice.lock; pruning is lazy, so ractor_free delists the dying ractor's sched. Barrier pairing needs no fences: the barrier sets barrier_is_waiting and then walks the registrations under their locks; a switch moves its registration under the same lock and then reads the flag. A walk that missed a registration ran before that registration's critical section, so its flag store is visible to that switcher. Nothing may start running while the flag stands -- an interrupt flag alone would let a resuming thread run, and allocate, inside the stop-the-world section -- so the add path takes its registration back and waits for rb_ractor_sched_barrier_end's broadcast. While a scan holds a running_th lock the registered thread cannot finish parking, so it cannot die under the scan. An snt turned dedicated under rb_thread_lock_native_thread has no creation-time running_thread, so the running_dnts registration writes nt->running_thread itself. R ractor pairs ping-ponging via ports (msgs/sec) and R ractors doing pipe write/read round-trips (rt/sec), 16-HT machine (Ryzen 9 5900HX), mean of 2 alternating same-tree runs: before after port pairs 8 503k 815k +62% port pairs 32 453k 799k +76% pipe self 4R 837k 940k +12% pipe self 16R 820k 898k +10% Co-Authored-By: Claude Fable 5 --- debug.c | 4 +- ractor.c | 9 +- thread_pthread.c | 509 +++++++++++++++++++++++++++++++---------------- thread_pthread.h | 22 +- vm_core.h | 28 ++- vm_sync.c | 6 +- 6 files changed, 388 insertions(+), 190 deletions(-) diff --git a/debug.c b/debug.c index 0ced0df9c7341c..d0cf306a523d9c 100644 --- a/debug.c +++ b/debug.c @@ -620,8 +620,8 @@ ruby_debug_log(const char *file, int line, const char *func_name, const char *fm rb_vm_t *vm = GET_VM(); if (r && len < MAX_DEBUG_LOG_MESSAGE_LEN) { - r = snprintf(buff + len, MAX_DEBUG_LOG_MESSAGE_LEN - len, "\tr:#%d/%u (%u)", - cr ? (int)rb_ractor_id(cr) : -1, vm->ractor.cnt, vm->ractor.sched.running_cnt); + r = snprintf(buff + len, MAX_DEBUG_LOG_MESSAGE_LEN - len, "\tr:#%d/%u", + cr ? (int)rb_ractor_id(cr) : -1, vm->ractor.cnt); if (r < 0) rb_bug("ruby_debug_log returns %d", r); len += r; diff --git a/ractor.c b/ractor.c index 1d9659e4aa1c85..5bf4e41b748b63 100644 --- a/ractor.c +++ b/ractor.c @@ -399,6 +399,10 @@ free_targeted_hooks(st_table *hooks_tbl) st_foreach(hooks_tbl, free_targeted_hook_lists, 0); } +#ifdef RUBY_THREAD_PTHREAD_H +void rb_thread_sched_destroy(struct rb_thread_sched *); +#endif + static void ractor_free(void *ptr) { @@ -406,6 +410,9 @@ ractor_free(void *ptr) RUBY_DEBUG_LOG("free r:%d", rb_ractor_id(r)); free_targeted_hooks(&r->pub.targeted_hooks); +#ifdef RUBY_THREAD_PTHREAD_H + rb_thread_sched_destroy(&r->threads.sched); +#endif rb_native_mutex_destroy(&r->sync.lock); #ifdef RUBY_THREAD_WIN32_H rb_native_cond_destroy(&r->sync.wakeup_cond); @@ -1196,7 +1203,7 @@ rb_ractor_terminate_all(void) rb_del_running_thread(rb_ec_thread_ptr(cr->threads.running_ec)); rb_vm_cond_timedwait(vm, &vm->ractor.sync.terminate_cond, 1000 /* ms */); #ifdef RUBY_THREAD_PTHREAD_H - while (vm->ractor.sched.barrier_waiting) { + while (vm->ractor.sched.barrier_is_waiting) { // A barrier is waiting. Threads relinquish the VM lock before joining the barrier and // since we just acquired the VM lock back, we're blocking other threads from joining it. // We loop until the barrier is over. We can't join this barrier because our thread isn't added to diff --git a/thread_pthread.c b/thread_pthread.c index b17bbbc3f3ea06..e624555c682e52 100644 --- a/thread_pthread.c +++ b/thread_pthread.c @@ -535,134 +535,191 @@ ASSERT_ractor_sched_locked(rb_vm_t *vm, rb_ractor_t *cr) VM_ASSERT(cr == NULL || vm->ractor.sched.lock_owner == cr); } -RBIMPL_ATTR_MAYBE_UNUSED() -static bool -ractor_sched_running_threads_contain_p(rb_vm_t *vm, rb_thread_t *th) +static void ractor_sched_barrier_join_signal_locked(rb_vm_t *vm); + +/* ntlist registration: a thread that executes Ruby code is always registered, + * in its snt's nt->running_th or on running_dnts via its dedicated nt. The + * only unregistered execution is scheduler glue (parking, resuming), which + * touches no Ruby heap, and the barrier wait below. */ +static void +ntlist_add_running(rb_vm_t *vm, rb_thread_t *th) { - rb_thread_t *rth; - ccan_list_for_each(&vm->ractor.sched.running_threads, rth, sched.node.running_threads) { - if (rth == th) return true; + struct rb_native_thread *nt = th->nt; + + // a dedicated nt is not on the snts list the scans walk: running_dnts instead + if (nt != NULL && nt->dedicated == 0) { + rb_native_mutex_lock(&nt->running_th_lock); + { + VM_ASSERT(nt->running_th == NULL); + nt->running_th = th; + } + rb_native_mutex_unlock(&nt->running_th_lock); + } + else { + rb_native_mutex_lock(&vm->ractor.sched.ntlist.lock); + { + // an snt gone dedicated (rb_thread_lock_native_thread) has no + // creation-time running_thread: the registration supplies it + nt->running_thread = th; + ccan_list_add(&vm->ractor.sched.ntlist.running_dnts, &nt->running_dnts_node); + } + rb_native_mutex_unlock(&vm->ractor.sched.ntlist.lock); } - return false; } -RBIMPL_ATTR_MAYBE_UNUSED() -static unsigned int -ractor_sched_running_threads_size(rb_vm_t *vm) +// Returns whether the active barrier's walk had counted this registration: +// such a deregistration owes the snapshot count a decrement. Read and +// cleared under the registration's own lock, so it pairs with the walk. +static bool +ntlist_del_running(rb_vm_t *vm, rb_thread_t *th) { - rb_thread_t *th; - unsigned int i = 0; - ccan_list_for_each(&vm->ractor.sched.running_threads, th, sched.node.running_threads) { - i++; + struct rb_native_thread *nt = th->nt; + uint32_t serial; + bool counted; + bool in_running_th; + + // The registration itself says where it is: nt->running_th holds th, or + // th's nt hangs on running_dnts. barrier_serial is read inside the + // registration's lock, ordered with the walk that stamped there. + rb_native_mutex_lock(&nt->running_th_lock); + { + in_running_th = (nt->running_th == th); + if (in_running_th) { + nt->running_th = NULL; + serial = vm->ractor.sched.barrier_serial; + counted = (nt->barrier_counted_serial == serial); + nt->barrier_counted_serial = serial - 1; // only once per barrier + } } - return i; -} + rb_native_mutex_unlock(&nt->running_th_lock); -RBIMPL_ATTR_MAYBE_UNUSED() -static unsigned int -ractor_sched_timeslice_threads_size(rb_vm_t *vm) -{ - rb_thread_t *th; - unsigned int i = 0; - ccan_list_for_each(&vm->ractor.sched.timeslice_threads, th, sched.node.timeslice_threads) { - i++; + if (!in_running_th) { + rb_native_mutex_lock(&vm->ractor.sched.ntlist.lock); + { + ccan_list_del_init(&nt->running_dnts_node); + serial = vm->ractor.sched.barrier_serial; + counted = (nt->barrier_counted_serial == serial); + nt->barrier_counted_serial = serial - 1; + } + rb_native_mutex_unlock(&vm->ractor.sched.ntlist.lock); } - return i; + return counted; } -RBIMPL_ATTR_MAYBE_UNUSED() +// Stamp a registration into the active barrier's snapshot unless the walk +// already counted it; returns whether it stamped. Called under sched.lock, +// so it is serialized with the walk: the stamp says exactly whether the +// registration came first. static bool -ractor_sched_timeslice_threads_contain_p(rb_vm_t *vm, rb_thread_t *th) +ntlist_stamp_if_uncounted(rb_vm_t *vm, rb_thread_t *th) { - rb_thread_t *rth; - ccan_list_for_each(&vm->ractor.sched.timeslice_threads, rth, sched.node.timeslice_threads) { - if (rth == th) return true; + struct rb_native_thread *nt = th->nt; + uint32_t serial = vm->ractor.sched.barrier_serial; // sched.lock is held + bool stamped; + bool in_running_th; + + rb_native_mutex_lock(&nt->running_th_lock); + { + in_running_th = (nt->running_th == th); + if (in_running_th) { + stamped = (nt->barrier_counted_serial != serial); + nt->barrier_counted_serial = serial; + } } - return false; + rb_native_mutex_unlock(&nt->running_th_lock); + + if (!in_running_th) { + rb_native_mutex_lock(&vm->ractor.sched.ntlist.lock); + { + stamped = (nt->barrier_counted_serial != serial); + nt->barrier_counted_serial = serial; + } + rb_native_mutex_unlock(&vm->ractor.sched.ntlist.lock); + } + return stamped; } -static void ractor_sched_barrier_join_signal_locked(rb_vm_t *vm); +// Record a thread entering/leaving the running set, with no global lock and +// no count: the records themselves are what the barrier counts. Pairing: +// the barrier sets barrier_is_waiting and then walks the records under their +// locks; we move a record and then read the flag, so one side sees the other. +// List sched for the timer's timeslice ticks. The caller holds sched->lock_ +// with the readyq non-empty, so the timer cannot prune the entry meanwhile. +static void +timeslice_sched_link(rb_vm_t *vm, struct rb_thread_sched *sched) +{ + rb_native_mutex_lock(&vm->ractor.sched.timeslice.lock); + { + if (sched->timeslice_node.next == &sched->timeslice_node) { + ccan_list_add_tail(&vm->ractor.sched.timeslice.scheds, &sched->timeslice_node); + } + } + rb_native_mutex_unlock(&vm->ractor.sched.timeslice.lock); +} -// setup timeslice signals by the timer thread. static void thread_sched_setup_running_threads(struct rb_thread_sched *sched, rb_ractor_t *cr, rb_vm_t *vm, - rb_thread_t *add_th, rb_thread_t *del_th, rb_thread_t *add_timeslice_th) + rb_thread_t *add_th, rb_thread_t *del_th) { -#if USE_RUBY_DEBUG_LOG - unsigned int prev_running_cnt = vm->ractor.sched.running_cnt; -#endif + RUBY_DEBUG_LOG("+:%u -:%u", rb_th_serial(add_th), rb_th_serial(del_th)); - rb_thread_t *del_timeslice_th; + if (del_th) { + bool counted = ntlist_del_running(vm, del_th); + sched->is_running = false; - if (del_th && sched->is_running_timeslice) { - del_timeslice_th = del_th; - sched->is_running_timeslice = false; - } - else { - del_timeslice_th = NULL; + // The first load is only a filter; the one under sched.lock decides. + // A missed flag means this deregistration preceded the barrier's walk. + if (UNLIKELY(RUBY_ATOMIC_LOAD(vm->ractor.sched.barrier_is_waiting))) { + ractor_sched_lock(vm, cr); + { + if (RUBY_ATOMIC_LOAD(vm->ractor.sched.barrier_is_waiting)) { + if (counted) { + VM_ASSERT(vm->ractor.sched.barrier_running_cnt > 0); + vm->ractor.sched.barrier_running_cnt--; + } + ractor_sched_barrier_join_signal_locked(vm); + } + } + ractor_sched_unlock(vm, cr); + } } - RUBY_DEBUG_LOG("+:%u -:%u +ts:%u -ts:%u", - rb_th_serial(add_th), rb_th_serial(del_th), - rb_th_serial(add_timeslice_th), rb_th_serial(del_timeslice_th)); + if (add_th) { + ntlist_add_running(vm, add_th); - ractor_sched_lock(vm, cr); - { - // update running_threads - if (del_th) { - VM_ASSERT(ractor_sched_running_threads_contain_p(vm, del_th)); - VM_ASSERT(del_timeslice_th != NULL || - !ractor_sched_timeslice_threads_contain_p(vm, del_th)); - - ccan_list_del_init(&del_th->sched.node.running_threads); - vm->ractor.sched.running_cnt--; - - if (UNLIKELY(vm->ractor.sched.barrier_waiting)) { - ractor_sched_barrier_join_signal_locked(vm); + if (UNLIKELY(RUBY_ATOMIC_LOAD(vm->ractor.sched.barrier_is_waiting))) { + // A stop-the-world section. In its waiting phase sched.lock is + // takable: join the snapshot count and take the interrupt (this + // thread joins at its next check, like any walked runner). In + // the GC phase the barrier holds sched.lock to its end, so this + // blocks here, as the old global-lock design did. + ractor_sched_lock(vm, cr); + { + if (RUBY_ATOMIC_LOAD(vm->ractor.sched.barrier_is_waiting) && + ntlist_stamp_if_uncounted(vm, add_th)) { + // the walk ran before this registration; count it in + RUBY_DEBUG_LOG("barrier_is_waiting"); + vm->ractor.sched.barrier_running_cnt++; + RUBY_VM_SET_VM_BARRIER_INTERRUPT(add_th->ec); + } } - sched->is_running = false; + ractor_sched_unlock(vm, cr); } - if (add_th) { - if (vm->ractor.sched.barrier_waiting) { - // TODO: GC barrier check? - RUBY_DEBUG_LOG("barrier_waiting"); - RUBY_VM_SET_VM_BARRIER_INTERRUPT(add_th->ec); - } - - VM_ASSERT(!ractor_sched_running_threads_contain_p(vm, add_th)); - VM_ASSERT(!ractor_sched_timeslice_threads_contain_p(vm, add_th)); + sched->is_running = true; - ccan_list_add(&vm->ractor.sched.running_threads, &add_th->sched.node.running_threads); - vm->ractor.sched.running_cnt++; - sched->is_running = true; - } - - if (add_timeslice_th) { - // update timeslice threads - int was_empty = ccan_list_empty(&vm->ractor.sched.timeslice_threads); - VM_ASSERT(!ractor_sched_timeslice_threads_contain_p(vm, add_timeslice_th)); - ccan_list_add(&vm->ractor.sched.timeslice_threads, &add_timeslice_th->sched.node.timeslice_threads); - sched->is_running_timeslice = true; - if (was_empty) { - timer_thread_wakeup_locked(vm); + // taking a turn with waiters already queued needs the timeslice ticks + if (!ccan_list_empty(&sched->readyq)) { + timeslice_sched_link(vm, sched); + ractor_sched_lock(vm, cr); + { + if (vm->ractor.sched.timeslice_wait_inf) { + timer_thread_wakeup_locked(vm); + } } + ractor_sched_unlock(vm, cr); } - - if (del_timeslice_th) { - VM_ASSERT(ractor_sched_timeslice_threads_contain_p(vm, del_timeslice_th)); - ccan_list_del_init(&del_timeslice_th->sched.node.timeslice_threads); - } - - VM_ASSERT(ractor_sched_running_threads_size(vm) == vm->ractor.sched.running_cnt); - VM_ASSERT(ractor_sched_timeslice_threads_size(vm) <= vm->ractor.sched.running_cnt); } - ractor_sched_unlock(vm, cr); - - //RUBY_DEBUG_LOG("+:%u -:%u +ts:%u -ts:%u run:%u->%u", - // rb_th_serial(add_th), rb_th_serial(del_th), - // rb_th_serial(add_timeslice_th), rb_th_serial(del_timeslice_th), - RUBY_DEBUG_LOG("run:%u->%u", prev_running_cnt, vm->ractor.sched.running_cnt); } static void @@ -672,7 +729,7 @@ thread_sched_add_running_thread(struct rb_thread_sched *sched, rb_thread_t *th) VM_ASSERT(sched->running == th); rb_vm_t *vm = th->vm; - thread_sched_setup_running_threads(sched, th->ractor, vm, th, NULL, ccan_list_empty(&sched->readyq) ? NULL : th); + thread_sched_setup_running_threads(sched, th->ractor, vm, th, NULL); } static void @@ -681,7 +738,7 @@ thread_sched_del_running_thread(struct rb_thread_sched *sched, rb_thread_t *th) ASSERT_thread_sched_locked(sched, th); rb_vm_t *vm = th->vm; - thread_sched_setup_running_threads(sched, th->ractor, vm, NULL, th, NULL); + thread_sched_setup_running_threads(sched, th->ractor, vm, NULL, th); } void @@ -780,20 +837,26 @@ thread_sched_enq(struct rb_thread_sched *sched, rb_thread_t *ready_th) VM_ASSERT(sched->running != NULL); VM_ASSERT(!thread_sched_readyq_contain_p(sched, ready_th)); - if (sched->is_running) { - if (ccan_list_empty(&sched->readyq)) { - // add sched->running to timeslice - thread_sched_setup_running_threads(sched, ready_th->ractor, ready_th->vm, NULL, NULL, sched->running); - } - } - else { - // ractor_sched lock is needed - // VM_ASSERT(!ractor_sched_timeslice_threads_contain_p(ready_th->vm, sched->running)); - } + bool timeslice_onset = sched->is_running && ccan_list_empty(&sched->readyq); ccan_list_add_tail(&sched->readyq, &ready_th->sched.node.readyq); ready_th->sched.node.is_ready = true; sched->readyq_cnt++; + + if (timeslice_onset) { + // The running thread needs the timeslice ticks now. Linked before + // the check under sched.lock: either the timer's scan (same lock) + // sees the sched, or this sees timeslice_wait_inf. + rb_vm_t *vm = ready_th->vm; + timeslice_sched_link(vm, sched); + ractor_sched_lock(vm, NULL); + { + if (vm->ractor.sched.timeslice_wait_inf) { + timer_thread_wakeup_locked(vm); + } + } + ractor_sched_unlock(vm, NULL); + } } // DNT: kick condvar @@ -916,7 +979,7 @@ thread_sched_wait_running_turn(struct rb_thread_sched *sched, rb_thread_t *th, b } // already deleted from running threads - // VM_ASSERT(!ractor_sched_running_threads_contain_p(th->vm, th)); // need locking + // wait for execution right rb_thread_t *next_th; @@ -1021,7 +1084,7 @@ thread_sched_wait_running_turn(struct rb_thread_sched *sched, rb_thread_t *th, b sched->runnable_hot_th = NULL; sched->runnable_hot_th_waiting = 0; - // VM_ASSERT(ractor_sched_running_threads_contain_p(th->vm, th)); need locking + RB_INTERNAL_THREAD_HOOK(RUBY_INTERNAL_THREAD_EVENT_RESUMED, th); } @@ -1316,6 +1379,7 @@ rb_thread_sched_init(struct rb_thread_sched *sched, bool atfork) ccan_list_head_init(&sched->readyq); sched->readyq_cnt = 0; ccan_list_node_init(&sched->grq_node); // self-linked = not enqueued + ccan_list_node_init(&sched->timeslice_node); #if USE_MN_THREADS if (!atfork) sched->enable_mn_threads = true; // MN is enabled on Ractors @@ -1671,10 +1735,12 @@ rb_ractor_sched_wakeup(rb_ractor_t *r, rb_thread_t *r_th) static bool ractor_sched_barrier_completed_p(rb_vm_t *vm) { - RUBY_DEBUG_LOG("run:%u wait:%u", vm->ractor.sched.running_cnt, vm->ractor.sched.barrier_waiting_cnt); - VM_ASSERT(vm->ractor.sched.running_cnt - 1 >= vm->ractor.sched.barrier_waiting_cnt); + // The snapshot barrier_running_cnt is taken by the barrier's walk and + // decremented by counted deregistrations; no rescan is needed here. + RUBY_DEBUG_LOG("run:%u wait:%u", vm->ractor.sched.barrier_running_cnt, vm->ractor.sched.barrier_joined_cnt); + VM_ASSERT(vm->ractor.sched.barrier_running_cnt - 1 >= vm->ractor.sched.barrier_joined_cnt); - return (vm->ractor.sched.running_cnt - vm->ractor.sched.barrier_waiting_cnt) == 1; + return (vm->ractor.sched.barrier_running_cnt - vm->ractor.sched.barrier_joined_cnt) == 1; } void @@ -1682,8 +1748,8 @@ rb_ractor_sched_barrier_start(rb_vm_t *vm, rb_ractor_t *cr) { VM_ASSERT(cr == GET_RACTOR()); VM_ASSERT(vm->ractor.sync.lock_owner == cr); // VM is locked - VM_ASSERT(!vm->ractor.sched.barrier_waiting); - VM_ASSERT(vm->ractor.sched.barrier_waiting_cnt == 0); + VM_ASSERT(!vm->ractor.sched.barrier_is_waiting); + VM_ASSERT(vm->ractor.sched.barrier_joined_cnt == 0); VM_ASSERT(vm->ractor.sched.barrier_ractor == NULL); VM_ASSERT(vm->ractor.sched.barrier_lock_rec == 0); @@ -1693,7 +1759,7 @@ rb_ractor_sched_barrier_start(rb_vm_t *vm, rb_ractor_t *cr) ractor_sched_lock(vm, cr); { - vm->ractor.sched.barrier_waiting = true; + RUBY_ATOMIC_SET(vm->ractor.sched.barrier_is_waiting, 1); vm->ractor.sched.barrier_ractor = cr; vm->ractor.sched.barrier_lock_rec = vm->ractor.sync.lock_rec; @@ -1703,14 +1769,48 @@ rb_ractor_sched_barrier_start(rb_vm_t *vm, rb_ractor_t *cr) vm->ractor.sync.lock_owner = NULL; rb_native_mutex_unlock(&vm->ractor.sync.lock); - // interrupts all running threads + // Interrupt all running threads: running_dnts plus each snt's running_th. + // A switch before this scan is visible to it; one after it sees + // barrier_is_waiting (set above) and waits. + // Interrupt and count every registered runner, stamping each nt so a + // deregistration during this barrier knows it was counted. rb_thread_t *ith; - ccan_list_for_each(&vm->ractor.sched.running_threads, ith, sched.node.running_threads) { - if (ith->ractor != cr) { - RUBY_DEBUG_LOG("barrier request to th:%u", rb_th_serial(ith)); - RUBY_VM_SET_VM_BARRIER_INTERRUPT(ith->ec); + unsigned int running_cnt = 0; + uint32_t serial = vm->ractor.sched.barrier_serial; + + rb_native_mutex_lock(&vm->ractor.sched.ntlist.lock); + { + struct rb_native_thread *dnt; + ccan_list_for_each(&vm->ractor.sched.ntlist.running_dnts, dnt, running_dnts_node) { + ith = dnt->running_thread; + dnt->barrier_counted_serial = serial; + running_cnt++; + if (ith->ractor != cr) { + RUBY_DEBUG_LOG("barrier request to th:%u", rb_th_serial(ith)); + RUBY_VM_SET_VM_BARRIER_INTERRUPT(ith->ec); + } + } + + struct rb_native_thread *nt; + ccan_list_for_each(&vm->ractor.sched.ntlist.snts, nt, snts_node) { + rb_native_mutex_lock(&nt->running_th_lock); + { + ith = nt->running_th; + if (ith != NULL) { + nt->barrier_counted_serial = serial; + running_cnt++; + if (ith->ractor != cr) { + RUBY_DEBUG_LOG("barrier request to th:%u", rb_th_serial(ith)); + RUBY_VM_SET_VM_BARRIER_INTERRUPT(ith->ec); + } + } + } + rb_native_mutex_unlock(&nt->running_th_lock); } } + rb_native_mutex_unlock(&vm->ractor.sched.ntlist.lock); + + vm->ractor.sched.barrier_running_cnt = running_cnt; // wait for other ractors while (!ractor_sched_barrier_completed_p(vm)) { @@ -1723,7 +1823,7 @@ rb_ractor_sched_barrier_start(rb_vm_t *vm, rb_ractor_t *cr) // no other ractors are there vm->ractor.sched.barrier_serial++; - vm->ractor.sched.barrier_waiting_cnt = 0; + vm->ractor.sched.barrier_joined_cnt = 0; rb_native_cond_broadcast(&vm->ractor.sched.barrier_release_cond); // acquire VM lock @@ -1741,11 +1841,11 @@ void rb_ractor_sched_barrier_end(rb_vm_t *vm, rb_ractor_t *cr) { RUBY_DEBUG_LOG("serial:%u", (unsigned int)vm->ractor.sched.barrier_serial - 1); - VM_ASSERT(vm->ractor.sched.barrier_waiting); + VM_ASSERT(vm->ractor.sched.barrier_is_waiting); VM_ASSERT(vm->ractor.sched.barrier_ractor); VM_ASSERT(vm->ractor.sched.barrier_lock_rec > 0); - vm->ractor.sched.barrier_waiting = false; + RUBY_ATOMIC_SET(vm->ractor.sched.barrier_is_waiting, 0); vm->ractor.sched.barrier_ractor = NULL; vm->ractor.sched.barrier_lock_rec = 0; ractor_sched_unlock(vm, cr); @@ -1762,7 +1862,7 @@ ractor_sched_barrier_join_signal_locked(rb_vm_t *vm) static void ractor_sched_barrier_join_wait_locked(rb_vm_t *vm, rb_thread_t *th) { - VM_ASSERT(vm->ractor.sched.barrier_waiting); + VM_ASSERT(vm->ractor.sched.barrier_is_waiting); unsigned int barrier_serial = vm->ractor.sched.barrier_serial; @@ -1785,7 +1885,7 @@ rb_ractor_sched_barrier_join(rb_vm_t *vm, rb_ractor_t *cr) VM_ASSERT(cr->threads.sched.running != NULL); // running ractor VM_ASSERT(cr == GET_RACTOR()); VM_ASSERT(vm->ractor.sync.lock_owner == NULL); // VM is locked, but owner == NULL - VM_ASSERT(vm->ractor.sched.barrier_waiting); // VM needs barrier sync + VM_ASSERT(vm->ractor.sched.barrier_is_waiting); // VM needs barrier sync #if USE_RUBY_DEBUG_LOG || VM_CHECK_MODE > 0 unsigned int barrier_serial = vm->ractor.sched.barrier_serial; @@ -1795,17 +1895,16 @@ rb_ractor_sched_barrier_join(rb_vm_t *vm, rb_ractor_t *cr) rb_native_mutex_unlock(&vm->ractor.sync.lock); { - VM_ASSERT(vm->ractor.sched.barrier_waiting); // VM needs barrier sync + VM_ASSERT(vm->ractor.sched.barrier_is_waiting); // VM needs barrier sync VM_ASSERT(vm->ractor.sched.barrier_serial == barrier_serial); ractor_sched_lock(vm, cr); { // running_cnt - /* Every joiner is a member of the running set: a dying thread now + /* Every joiner is a member of the running set: a dying thread * leaves the living set before handing over its scheduler slot. */ - VM_ASSERT(ractor_sched_running_threads_contain_p(vm, GET_THREAD())); - vm->ractor.sched.barrier_waiting_cnt++; - RUBY_DEBUG_LOG("waiting_cnt:%u serial:%u", vm->ractor.sched.barrier_waiting_cnt, barrier_serial); + vm->ractor.sched.barrier_joined_cnt++; + RUBY_DEBUG_LOG("waiting_cnt:%u serial:%u", vm->ractor.sched.barrier_joined_cnt, barrier_serial); ractor_sched_barrier_join_signal_locked(vm); ractor_sched_barrier_join_wait_locked(vm, cr->threads.sched.running); @@ -1817,25 +1916,20 @@ rb_ractor_sched_barrier_join(rb_vm_t *vm, rb_ractor_t *cr) // VM locked here } -#if 0 -// TODO - -static void clear_thread_cache_altstack(void); - -static void +// Called when the ractor holding this sched is freed. A drained sched can +// still be on timeslice.scheds (pruning is lazy); an unlisted node is +// self-linked (fork re-inits them all), making this del a no-op. +void rb_thread_sched_destroy(struct rb_thread_sched *sched) { - /* - * only called once at VM shutdown (not atfork), another thread - * may still grab vm->gvl.lock when calling gvl_release at - * the end of thread_start_func_2 - */ - if (0) { - rb_native_mutex_destroy(&sched->lock); + rb_vm_t *vm = GET_VM(); + + rb_native_mutex_lock(&vm->ractor.sched.timeslice.lock); + { + ccan_list_del_init(&sched->timeslice_node); } - clear_thread_cache_altstack(); + rb_native_mutex_unlock(&vm->ractor.sched.timeslice.lock); } -#endif #ifdef RB_THREAD_T_HAS_NATIVE_ID static int @@ -1872,7 +1966,6 @@ thread_sched_atfork(struct rb_thread_sched *sched) vm->ractor.sched.dnt_cnt = 0; #endif } - vm->ractor.sched.running_cnt = 0; rb_native_mutex_initialize(&vm->ractor.sched.lock); #if VM_CHECK_MODE > 0 @@ -1889,16 +1982,35 @@ thread_sched_atfork(struct rb_thread_sched *sched) vm->ractor.sched.grq_cnt = 0; // the list was just emptied; reset the count with it // A fork during a VM barrier leaves the child with barrier state that can // never complete (the other ractors are gone); reset it like the rest. - vm->ractor.sched.barrier_waiting = false; - vm->ractor.sched.barrier_waiting_cnt = 0; + vm->ractor.sched.barrier_is_waiting = 0; // single-threaded child + vm->ractor.sched.barrier_joined_cnt = 0; vm->ractor.sched.barrier_ractor = NULL; vm->ractor.sched.barrier_lock_rec = 0; // Threads that were winding down in the parent do not exist in the child; // without this reset the child's ruby_vm_destruct would wait for their // reclaim (which never comes) forever. vm->ractor.sched.winding_cnt = 0; - ccan_list_head_init(&vm->ractor.sched.timeslice_threads); - ccan_list_head_init(&vm->ractor.sched.running_threads); + rb_native_mutex_initialize(&vm->ractor.sched.ntlist.lock); + ccan_list_head_init(&vm->ractor.sched.ntlist.running_dnts); + ccan_list_head_init(&vm->ractor.sched.ntlist.snts); // those nts are gone + rb_native_mutex_initialize(&vm->ractor.sched.timeslice.lock); + ccan_list_head_init(&vm->ractor.sched.timeslice.scheds); + rb_native_mutex_initialize(&th->nt->running_th_lock); // a scan could hold it at fork + // Fork can copy nodes linked (or torn mid-link); re-init every sched's + // node so rb_thread_sched_destroy's del_init stays a no-op for them. + rb_ractor_t *r; + ccan_list_for_each(&vm->ractor.set, r, vmlr_node) { + ccan_list_node_init(&r->threads.sched.timeslice_node); + } + ccan_list_for_each(&vm->ractor.terminated_set, r, vmlr_node) { + ccan_list_node_init(&r->threads.sched.timeslice_node); + } + // th re-records itself below; the parent's record did not survive the lists + if (th->nt && th->nt->dedicated == 0) { + // surviving on an snt: put that nt back on the (just emptied) snts + // list, or the scans could not see this thread's record + ccan_list_add(&vm->ractor.sched.ntlist.snts, &th->nt->snts_node); + } #if USE_MN_THREADS nt_machine_stack_atfork(); @@ -1906,13 +2018,12 @@ thread_sched_atfork(struct rb_thread_sched *sched) rb_internal_thread_event_hooks_rw_lock_atfork(); VM_ASSERT(sched->is_running); - sched->is_running_timeslice = false; if (sched->running != th) { thread_sched_to_running(sched, th); } else { - thread_sched_setup_running_threads(sched, th->ractor, vm, th, NULL, NULL); + thread_sched_setup_running_threads(sched, th->ractor, vm, th, NULL); } #ifdef RB_THREAD_T_HAS_NATIVE_ID @@ -2010,8 +2121,11 @@ Init_native_thread(rb_thread_t *main_th) rb_native_cond_initialize(&vm->ractor.sched.barrier_release_cond); ccan_list_head_init(&vm->ractor.sched.grq); - ccan_list_head_init(&vm->ractor.sched.timeslice_threads); - ccan_list_head_init(&vm->ractor.sched.running_threads); + rb_native_mutex_initialize(&vm->ractor.sched.ntlist.lock); + ccan_list_head_init(&vm->ractor.sched.ntlist.running_dnts); + ccan_list_head_init(&vm->ractor.sched.ntlist.snts); + rb_native_mutex_initialize(&vm->ractor.sched.timeslice.lock); + ccan_list_head_init(&vm->ractor.sched.timeslice.scheds); // setup main thread main_th->nt->thread_id = pthread_self(); @@ -2026,12 +2140,13 @@ Init_native_thread(rb_thread_t *main_th) TH_SCHED(main_th)->running = main_th; main_th->has_dedicated_nt = 1; - thread_sched_setup_running_threads(TH_SCHED(main_th), main_th->ractor, vm, main_th, NULL, NULL); - - // setup main NT + // setup main NT (before the record below: its kind decides where it goes) main_th->nt->dedicated = 1; + main_th->nt->running_thread = main_th; main_th->nt->vm = vm; + thread_sched_setup_running_threads(TH_SCHED(main_th), main_th->ractor, vm, main_th, NULL); + // setup mn #if USE_RUBY_DEBUG_LOG vm->ractor.sched.dnt_cnt = 1; @@ -2177,6 +2292,7 @@ native_thread_destroy(struct rb_native_thread *nt) { if (nt) { rb_native_cond_destroy(&nt->readyq); + rb_native_mutex_destroy(&nt->running_th_lock); native_thread_destroy_atfork(nt); } @@ -2483,6 +2599,8 @@ native_thread_setup(struct rb_native_thread *nt) { // init cond rb_native_cond_initialize(&nt->readyq); + // also for the main thread's nt: a zero-filled mutex is not usable on macOS + rb_native_mutex_initialize(&nt->running_th_lock); } static void @@ -2573,8 +2691,18 @@ nt_start(void *ptr) RUBY_DEBUG_LOG("nt:%u", nt->serial); + bool in_snts = false; + if (!nt->dedicated) { coroutine_initialize_main(nt->nt_context); + + // join the snt list that the barrier/timeslice scans walk + rb_native_mutex_lock(&vm->ractor.sched.ntlist.lock); + { + ccan_list_add(&vm->ractor.sched.ntlist.snts, &nt->snts_node); + } + rb_native_mutex_unlock(&vm->ractor.sched.ntlist.lock); + in_snts = true; } bool retired = false; @@ -2659,6 +2787,18 @@ nt_start(void *ptr) } } + if (in_snts) { + // Leaving the shared loop: every path back here deregistered first + // (park and death both precede the transfer), so only the snts entry + // is left to remove. + VM_ASSERT(nt->running_th == NULL); + rb_native_mutex_lock(&vm->ractor.sched.ntlist.lock); + { + ccan_list_del_init(&nt->snts_node); + } + rb_native_mutex_unlock(&vm->ractor.sched.ntlist.lock); + } + if (retired) { // The counts dropped this nt already; nothing can reference it now. RUBY_DEBUG_LOG("retired nt:%u", nt->serial); @@ -3179,6 +3319,7 @@ static struct { #define TIMER_THREAD_CREATED_P() (timer_th.created_fork_gen == current_fork_gen) static void timer_thread_check_timeslice(rb_vm_t *vm); +static bool timeslice_scan(rb_vm_t *vm, bool interrupt); static int timer_thread_set_timeout(rb_vm_t *vm); #include "thread_pthread_mn.c" @@ -3193,13 +3334,12 @@ timer_thread_set_timeout(rb_vm_t *vm) ractor_sched_lock(vm, NULL); { - if ( !ccan_list_empty(&vm->ractor.sched.timeslice_threads) // (1-1) Provide time slice for active NTs + if ( timeslice_scan(vm, false) // (1-1) Provide time slice for active NTs || !ubf_threads_empty() // (1-3) Periodic UBF || vm->ractor.sched.grq_cnt > 0 // (1-4) Lazy GRQ deq start ) { - RUBY_DEBUG_LOG("timeslice:%d ubf:%d grq:%d", - !ccan_list_empty(&vm->ractor.sched.timeslice_threads), + RUBY_DEBUG_LOG("ubf:%d grq:%d", !ubf_threads_empty(), (vm->ractor.sched.grq_cnt > 0)); @@ -3233,15 +3373,50 @@ timer_thread_check_signal(rb_vm_t *vm) } } +// Tick (with `interrupt`) each listed sched's running thread and prune scheds +// whose readyq drained; returns whether any sched still needs ticks. +static bool +timeslice_scan(rb_vm_t *vm, bool interrupt) +{ + bool found = false; + struct rb_thread_sched *sched, *next; + + rb_native_mutex_lock(&vm->ractor.sched.timeslice.lock); + { + ccan_list_for_each_safe(&vm->ractor.sched.timeslice.scheds, sched, next, timeslice_node) { + // trylock: timeslice_sched_link nests sched.lock -> timeslice.lock, + // this scan holds the locks the other way around + if (rb_native_mutex_trylock(&sched->lock_) == 0) { + if (ccan_list_empty(&sched->readyq)) { + ccan_list_del_init(&sched->timeslice_node); // a later enq relinks it + } + else if (sched->is_running) { + VM_ASSERT(sched->running != NULL); + found = true; + if (interrupt) { + RUBY_DEBUG_LOG("timeslice th:%u", rb_th_serial(sched->running)); + RUBY_VM_SET_TIMER_INTERRUPT(sched->running->ec); + } + } + // else: waiters behind a blocked runner need no ticks; the + // add path wakes the timer when the sched runs again + rb_native_mutex_unlock(&sched->lock_); + } + else { + found = true; // busy switching; tick it on the next round + } + } + } + rb_native_mutex_unlock(&vm->ractor.sched.timeslice.lock); + + return found; +} + static void timer_thread_check_timeslice(rb_vm_t *vm) { // TODO: check time - rb_thread_t *th; - ccan_list_for_each(&vm->ractor.sched.timeslice_threads, th, sched.node.timeslice_threads) { - RUBY_DEBUG_LOG("timeslice th:%u", rb_th_serial(th)); - RUBY_VM_SET_TIMER_INTERRUPT(th->ec); - } + timeslice_scan(vm, true); } void diff --git a/thread_pthread.h b/thread_pthread.h index 04e3de8845278d..4d34941647dd13 100644 --- a/thread_pthread.h +++ b/thread_pthread.h @@ -90,14 +90,6 @@ struct rb_thread_sched_item { // There is no clear relationship between this and th->status. bool is_ready; - // connected to vm->ractor.sched.timeslice_threads - // locked by vm->ractor.sched.lock - struct ccan_list_node timeslice_threads; - - // connected to vm->ractor.sched.running_threads - // locked by vm->ractor.sched.lock - struct ccan_list_node running_threads; - } node; struct rb_thread_sched_waiting waiting_reason; @@ -129,6 +121,17 @@ struct rb_native_thread { struct rb_thread_struct *running_thread; + // The running thread on this shared nt, for the barrier/timeslice scans. + // While a scan holds running_th_lock the thread cannot finish parking. + rb_nativethread_lock_t running_th_lock; + struct rb_thread_struct *running_th; + struct ccan_list_node snts_node; // in vm->ractor.sched.ntlist.snts + // in vm->ractor.sched.ntlist.running_dnts while running_thread runs + struct ccan_list_node running_dnts_node; + // barrier_serial stamped by the barrier's counting walk; this nt's + // deregistration during that barrier decrements the snapshot count + uint32_t barrier_counted_serial; + // to control native thread; use sched->lock rb_nativethread_cond_t readyq; @@ -167,7 +170,7 @@ struct rb_thread_sched { struct rb_thread_struct *runnable_hot_th; int runnable_hot_th_waiting; bool is_running; - bool is_running_timeslice; + bool enable_mn_threads; struct ccan_list_head readyq; @@ -178,6 +181,7 @@ struct rb_thread_sched { // node itself: enqueuers assert it, and direct transfers cancel an // outstanding entry (see ractor_sched_cancel_enq). struct ccan_list_node grq_node; + struct ccan_list_node timeslice_node; // self-linked = not on timeslice.scheds }; struct rb_thread_context; diff --git a/vm_core.h b/vm_core.h index 35c2d69512317f..e867230f66d475 100644 --- a/vm_core.h +++ b/vm_core.h @@ -752,18 +752,26 @@ typedef struct rb_vm_struct { rb_atomic_t snt_cnt; // count of shared NTs; lock-free (see native_thread_dedicated_inc) unsigned int dnt_cnt; // count of dedicated NTs; logging only (USE_RUBY_DEBUG_LOG), not atomic - unsigned int running_cnt; unsigned int max_cpu; struct ccan_list_head grq; // // Global Ready Queue rb_atomic_t winding_cnt; // native threads between a coroutine epilogue and its reclaim; ruby_vm_destruct waits for 0 unsigned int grq_cnt; - // running threads - struct ccan_list_head running_threads; - - // threads which switch context by timeslice - struct ccan_list_head timeslice_threads; + // What the barrier walk visits: threads running on dedicated + // nts, and the shared nts (whose running_th fields hold the rest). + struct { + rb_nativethread_lock_t lock; + struct ccan_list_head running_dnts; + struct ccan_list_head snts; + } ntlist; + + // scheds whose readyq holds waiters: the timer ticks their + // running thread (timeslice_scan) and prunes drained entries. + struct { + rb_nativethread_lock_t lock; + struct ccan_list_head scheds; + } timeslice; // true if timeslice timer is not enable bool timeslice_wait_inf; @@ -771,8 +779,12 @@ typedef struct rb_vm_struct { // barrier rb_nativethread_cond_t barrier_complete_cond; rb_nativethread_cond_t barrier_release_cond; - bool barrier_waiting; - unsigned int barrier_waiting_cnt; + // bool; nonzero while a stop-the-world section is active. Set + // before the barrier walks the running records; a record moved + // after the walk sees it (thread_sched_setup_running_threads). + rb_atomic_t barrier_is_waiting; + unsigned int barrier_joined_cnt; // threads joined so far; under sched.lock + unsigned int barrier_running_cnt; // runners counted by the barrier's walk; under sched.lock unsigned int barrier_serial; struct rb_ractor_struct *barrier_ractor; unsigned int barrier_lock_rec; diff --git a/vm_sync.c b/vm_sync.c index 6b988c1596f3a3..40adafe09290e1 100644 --- a/vm_sync.c +++ b/vm_sync.c @@ -41,7 +41,7 @@ RUBY_ASSERT_vm_locking_with_barrier(void) if (vm->ractor.cnt > 1) { /* Written to only when holding both ractor.sync and ractor.sched lock */ - VM_ASSERT(vm->ractor.sched.barrier_waiting); + VM_ASSERT(vm->ractor.sched.barrier_is_waiting); } } } @@ -66,7 +66,7 @@ static bool vm_need_barrier_waiting(const rb_vm_t *vm) { #ifdef RUBY_THREAD_PTHREAD_H - return vm->ractor.sched.barrier_waiting; + return vm->ractor.sched.barrier_is_waiting; #else return vm->ractor.sync.barrier_waiting; #endif @@ -287,7 +287,7 @@ rb_vm_barrier(void) return; } else { - VM_ASSERT(!vm->ractor.sched.barrier_waiting); + VM_ASSERT(!vm->ractor.sched.barrier_is_waiting); rb_ractor_sched_barrier_start(vm, cr); } }