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/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/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/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) 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/tool/sync_default_gems.rb b/tool/sync_default_gems.rb index 4e87e12bc8b19d..35246e56c84c4c 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)} @@ -78,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"], @@ -114,8 +117,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 +228,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", 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); } }