Wake timer to create new SNT when needed for dedicated task (ruby#16009)

When removing a thread from `running_threads`, if we're on a shared
native thread and we're running a dedicated task, we need to wake
the timer thread so it can create a new SNT if necessary. We only
do this if it's waiting forever without the 10ms quantum timeout
for now, because max 10ms of wait is considered "good enough".
In the future, perhaps we can force the timer thread to wake if this
becomes an issue (`timer_thread_wakeup_force`).

Fixes [Bug #21504]

Author: luke-gruber

test/ruby/test_ractor.rb

@@ -201,6 +201,45 @@ def test_max_cpu_1
     RUBY
   end
 
+  def test_timer_thread_create_snt_for_dedicated_task
+    omit "timer thread works differently" if windows?
+    omit "test relies on this as a best-effort safety mechanism" unless defined?(Process::WNOHANG)
+    assert_separately([{ "RUBY_MAX_CPU" => "2" }], <<~'RUBY', timeout: 30)
+      $VERBOSE = nil
+      CHILD_PID = 0
+
+      rs = []
+      2.times do |i|
+        rs << Ractor.new(i) do |j|
+          if j == 0
+            pid = spawn("sleep 60", close_others: true)
+            Object.const_set(:CHILD_PID, pid)
+            Process.waitpid(pid) # block forever (dedicated task)
+          else
+            while CHILD_PID == 0
+              sleep 1 # make sure first ractor blocks forever first (this is what we're testing)
+            end
+            1_000.times do
+              [nil] * 100
+            end
+          end
+        end
+      end
+
+      rs.last.join
+      begin
+        result = Process.waitpid(CHILD_PID, Process::WNOHANG)
+      rescue Errno::ECHILD, Errno::ESRCH
+        # If it's somehow not a child (not running?), don't send it a signal
+      else
+        if result.nil?
+          Process.kill('KILL', CHILD_PID) rescue nil
+        end
+      end
+      rs.first.join # reap
+    RUBY
+  end
+
   def test_symbol_proc_is_shareable
     pr = :symbol.to_proc
     assert_make_shareable(pr)

thread_pthread.c

@@ -90,6 +90,11 @@ static const void *const condattr_monotonic = NULL;
   #endif
 #endif
 
+#ifndef MINIMUM_SNT
+// make at least MINIMUM_SNT snts for debug.
+#define MINIMUM_SNT 0
+#endif
+
 #ifdef HAVE_SCHED_YIELD
 #define native_thread_yield() (void)sched_yield()
 #else
@@ -554,6 +559,20 @@ ractor_sched_timeslice_threads_contain_p(rb_vm_t *vm, rb_thread_t *th)
 
 static void ractor_sched_barrier_join_signal_locked(rb_vm_t *vm);
 
+static bool
+need_more_shared_native_threads_p(rb_vm_t *vm)
+{
+    ASSERT_ractor_sched_locked(vm, NULL);
+
+    unsigned int schedulable_ractor_cnt = vm->ractor.cnt;
+    unsigned int snt_cnt = vm->ractor.sched.snt_cnt;
+    RUBY_ASSERT(schedulable_ractor_cnt >= 1);
+    if (!vm->ractor.main_ractor->threads.sched.enable_mn_threads) {
+        schedulable_ractor_cnt--; // do not need snt for main ractor
+    }
+    return snt_cnt < MINIMUM_SNT || (snt_cnt < schedulable_ractor_cnt && snt_cnt < vm->ractor.sched.max_cpu);
+}
+
 // 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,
@@ -564,6 +583,7 @@ thread_sched_setup_running_threads(struct rb_thread_sched *sched, rb_ractor_t *c
 #endif
 
     rb_thread_t *del_timeslice_th;
+    bool wakeup_timer_thread = false;
 
     if (del_th && sched->is_running_timeslice) {
         del_timeslice_th = del_th;
@@ -592,6 +612,12 @@ thread_sched_setup_running_threads(struct rb_thread_sched *sched, rb_ractor_t *c
                 ractor_sched_barrier_join_signal_locked(vm);
             }
             sched->is_running = false;
+
+            // If we need more SNTs, the timer thread should be awake and monitoring the situation so it can correct it.
+            if (!del_th->has_dedicated_nt && del_th->nt->dedicated > 0 && (sched->running != NULL || vm->ractor.sched.grq_cnt > 0) &&
+                need_more_shared_native_threads_p(vm)) {
+                wakeup_timer_thread = true;
+            }
         }
 
         if (add_th) {
@@ -616,7 +642,7 @@ thread_sched_setup_running_threads(struct rb_thread_sched *sched, rb_ractor_t *c
             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);
+                wakeup_timer_thread = true;
             }
         }
 
@@ -625,6 +651,10 @@ thread_sched_setup_running_threads(struct rb_thread_sched *sched, rb_ractor_t *c
             ccan_list_del_init(&del_timeslice_th->sched.node.timeslice_threads);
         }
 
+        if (wakeup_timer_thread) {
+            timer_thread_wakeup_locked(vm);
+        }
+
         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);
     }
@@ -1264,11 +1294,6 @@ ractor_sched_enq(rb_vm_t *vm, rb_ractor_t *r)
 #define SNT_KEEP_SECONDS 0
 #endif
 
-#ifndef MINIMUM_SNT
-// make at least MINIMUM_SNT snts for debug.
-#define MINIMUM_SNT 0
-#endif
-
 static rb_ractor_t *
 ractor_sched_deq(rb_vm_t *vm, rb_ractor_t *cr)
 {

thread_pthread_mn.c

@@ -394,25 +394,14 @@ nt_free_stack(void *mstack)
     rb_native_mutex_unlock(&nt_machine_stack_lock);
 }
 
-
 static int
 native_thread_check_and_create_shared(rb_vm_t *vm)
 {
     bool need_to_make = false;
 
-    rb_native_mutex_lock(&vm->ractor.sched.lock);
+    ractor_sched_lock(vm, NULL);
     {
-        unsigned int schedulable_ractor_cnt = vm->ractor.cnt;
-        RUBY_ASSERT(schedulable_ractor_cnt >= 1);
-
-        if (!vm->ractor.main_ractor->threads.sched.enable_mn_threads)
-            schedulable_ractor_cnt--; // do not need snt for main ractor
-
-        unsigned int snt_cnt = vm->ractor.sched.snt_cnt;
-        if (((int)snt_cnt < MINIMUM_SNT) ||
-            (snt_cnt < schedulable_ractor_cnt  &&
-             snt_cnt < vm->ractor.sched.max_cpu)) {
-
+        if (need_more_shared_native_threads_p(vm)) {
             RUBY_DEBUG_LOG("added snt:%u dnt:%u ractor_cnt:%u grq_cnt:%u",
                            vm->ractor.sched.snt_cnt,
                            vm->ractor.sched.dnt_cnt,
@@ -426,7 +415,7 @@ native_thread_check_and_create_shared(rb_vm_t *vm)
             RUBY_DEBUG_LOG("snt:%d ractor_cnt:%d", (int)vm->ractor.sched.snt_cnt, (int)vm->ractor.cnt);
         }
     }
-    rb_native_mutex_unlock(&vm->ractor.sched.lock);
+    ractor_sched_unlock(vm, NULL);
 
     if (need_to_make) {
         struct rb_native_thread *nt = native_thread_alloc();