thread: wake the timer thread when an enqueued ractor has no server

ractor_sched_enq only signalled ractor.sched.cond.  The signal reaches
a parked shared native thread, and a running one revisits the queue in
ractor_sched_deq (under the same lock) before it can wait, so as long
as any shared native thread exists the entry is served.  With none --
each one dedicated to a blocking region or retired -- the signal is
lost, and if the timer thread is also in its untimed sleep
(timeslice_wait_inf), nothing serves the enqueued ractor: its runnable
thread never runs.  native_thread_check_and_create_shared, which would
repair the shortage, runs on the timer thread's timeout branch, which
is never reached again.

Wake the timer thread under the same lock its timeout decision is made
under, from the two transitions that can strand an entry: an enqueue
that finds snt_cnt == 0, and the blocking-region entry that dedicates
the last shared native thread away while the queue is non-empty.
Both are off the ordinary path, and timer_thread_wakeup_locked writes
to the communication pipe only when the timer thread sleeps untimed,
so the common case costs nothing.  [Bug #21504]

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

bootstraptest/test_thread.rb

@@ -571,3 +571,44 @@ assert_equal 'ok', %q{
     got == 'x' ? 'ok' : got.inspect
   end.value
 }
+
+# A ractor made runnable while every shared native thread is dedicated to a
+# blocking region must still be served: ractor_sched_enq has to wake the timer
+# thread, whose untimed sleep otherwise never ends.  [Bug #21504]
+assert_equal 'ok', %q{
+  lockpath = "mn_enq_wake_#{$$}.lock"
+  flagpath = "mn_enq_wake_#{$$}.flag"
+  begin
+    File.write(lockpath, "")
+    lock = File.open(lockpath, "r+")
+    lock.flock(File::LOCK_EX)
+    r = Ractor.new(lockpath, flagpath) do |lockpath, flagpath|
+      f = File.open(lockpath, "r+")
+      t = Thread.new do
+        sleep 0.2   # let the Ractor.receive below park first
+        # Stay off the scheduler until the timer thread is in its untimed sleep;
+        # blocking right away would be repaired by the pending 10ms timeout.
+        t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        nil while Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0 < 0.05
+        f.flock(File::LOCK_EX)   # the last shared native thread goes dedicated
+      end
+      msg = Ractor.receive
+      File.write(flagpath, "")
+      t.join
+      msg
+    end
+    sleep 1   # r is parked, its flock thread is dedicated, the timer sleeps untimed
+    r.send(:ok)
+    served = false
+    deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 5
+    until served || Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline
+      served = File.exist?(flagpath)
+      sleep 0.05
+    end
+    lock.flock(File::LOCK_UN)
+    served ? r.value.to_s : 'the enqueued ractor was never served'
+  ensure
+    File.unlink(lockpath) rescue nil
+    File.unlink(flagpath) rescue nil
+  end
+}

thread_pthread.c

@@ -1402,6 +1402,15 @@ ractor_sched_enq(rb_vm_t *vm, rb_ractor_t *r)
 
         rb_native_cond_signal(&vm->ractor.sched.cond);
 
+        // The signal reaches a parked snt, and a running one revisits the
+        // queue in ractor_sched_deq before it can wait (same lock as here).
+        // With every snt dedicated or retired, only the timer thread's
+        // timeout branch can serve the entry or widen the pool: wake it
+        // (a no-op unless it sleeps untimed).
+        if (vm->ractor.sched.snt_cnt == 0) {
+            timer_thread_wakeup_locked(vm);
+        }
+
         // ractor_sched_dump(vm);
     }
     ractor_sched_unlock(vm, cr);
@@ -1940,6 +1949,12 @@ native_thread_dedicated_inc(rb_vm_t *vm, rb_ractor_t *cr, struct rb_native_threa
         {
             vm->ractor.sched.snt_cnt--;
             vm->ractor.sched.dnt_cnt++;
+
+            // This may have dedicated the last snt away from a pending
+            // entry whose enqueue saw snt_cnt > 0 (see ractor_sched_enq).
+            if (vm->ractor.sched.snt_cnt == 0 && vm->ractor.sched.grq_cnt > 0) {
+                timer_thread_wakeup_locked(vm);
+            }
         }
         ractor_sched_unlock(vm, cr);
     }