Ractor: stop waiting a second for its own threads to finish

A Ractor whose block ends while one of its own threads is still running took
a full second to terminate: rb_thread_terminate_all() sleeps for a second per
round and relies on the last sub-thread to wake it, but that wakeup was sent
only when the Ractor's main thread already had THREAD_KILLED, which
thread_start_func_2 sets after rb_thread_terminate_all() returns.  For the
main Ractor rb_ec_cleanup() sets it beforehand, which is why only Ractors
were affected.

Setting THREAD_KILLED earlier is not an option here: thread_sched_switch()
reads it as to_dead, promising the M:N scheduler that the coroutine is never
resumed, and this thread still parks and resumes inside the wait.  So mark
the wait itself instead: rb_thread_terminate_all() sets threads.terminating
before it waits, and the exiting thread wakes `main` on that.  Nothing clears
it afterwards, because no thread of that Ractor runs again; the Ractor that
survives a fork gets it reset along with the rest of the thread set.

ractor.c

@@ -771,6 +771,7 @@ rb_ractor_living_threads_init(rb_ractor_t *r)
     ccan_list_head_init(&r->threads.set);
     r->threads.cnt = 0;
     r->threads.blocking_cnt = 0;
+    r->threads.terminating = false;
 }
 
 static void

ractor_core.h

@@ -100,6 +100,9 @@ struct rb_ractor_struct {
         struct rb_thread_sched sched;
         rb_execution_context_t *running_ec;
         rb_thread_t *main;
+
+        // `main` is in rb_thread_terminate_all(), waiting for the others to go
+        bool terminating;
     } threads;
 
     /* Postponed jobs targeted at this Ractor

test/ruby/test_ractor.rb

@@ -166,6 +166,29 @@ class TestRactor < Test::Unit::TestCase
       refute_equal main_ractor_id, ractor_id
     end;
   end
+  def test_ractor_with_live_threads_terminates_without_waiting
+    assert_separately([], __FILE__, __LINE__, <<-'RUBY')
+      Warning[:experimental] = false
+      # A Ractor that ends while a thread of its own is still running used to sit out
+      # the one second poll in rb_thread_terminate_all(), once per Ractor.  Measure
+      # against the same Ractors without a live thread, so that a busy machine, which
+      # makes both of them slow, does not decide this.
+      n = 5
+      elapsed = ->(&blk) {
+        t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        n.times { blk.call }
+        Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0
+      }
+
+      base = elapsed.call { assert_equal :done, Ractor.new { :done }.value }
+      live = elapsed.call { assert_equal :done, Ractor.new { Thread.new { sleep 10 }; :done }.value }
+
+      # the bug costs a second per Ractor, so #{n} seconds here
+      assert_operator live, :<, base + 2.0,
+                      "#{n} Ractors with a live thread took #{live}s, without one #{base}s"
+    RUBY
+  end
+
 
   def test_class_instance_variables
     assert_ractor(<<~'RUBY')

thread.c

@@ -482,6 +482,10 @@ rb_thread_terminate_all(rb_thread_t *th)
     /* unlock all locking mutexes */
     rb_threadptr_unlock_all_locking_mutexes(th);
 
+    // tells the last sub-thread to wake this one out of the sleep below.  Nothing
+    // clears it: no thread of this Ractor can run again once this returns.
+    cr->threads.terminating = true;
+
     EC_PUSH_TAG(ec);
     if (EC_EXEC_TAG() == TAG_NONE) {
       retry:
@@ -810,7 +814,7 @@ thread_start_func_2(rb_thread_t *th, VALUE *stack_start)
                (void *)th, th->locking_mutex);
     }
 
-    if (ractor_main_th->status == THREAD_KILLED &&
+    if (th->ractor->threads.terminating &&
         th->ractor->threads.cnt <= 2 /* main thread and this thread */) {
         /* I'm last thread. wake up main thread from rb_thread_terminate_all */
         rb_threadptr_interrupt(ractor_main_th);