thread: do not publish an M:N thread before its native thread exists

native_thread_create_shared called thread_sched_to_ready before trying
to widen the shared pool, so when pthread_create failed the creator's
failure path (marking the thread killed, removing it from the living
set, rb_ractor_cancel_creation) ran against a thread that an already
parked shared native thread could pick up and run to its end
concurrently -- including rb_ractor_postmortem_free of the very
rb_thread_t the creator is still writing to.  Observed under
RLIMIT_NPROC exhaustion as
  ractor.c: rb_ractor_cancel_creation: r->status_ == ractor_blocking
  ractor.c: ractor_check_blocking: cr->threads.cnt >= cr->threads.blocking_cnt + 1
i.e. the living-set removal ran twice.

Widen the pool first and make the thread ready only after that
succeeded; the failure path then handles a thread that never became
runnable, which is what it assumes.  A never-started coroutine's stack
and context are already freed by rb_threadptr_sched_free.

Also roll snt_cnt back when native_thread_create0 fails, and free the
allocated rb_native_thread.  The count stayed raised before, so
native_thread_check_and_create_shared concluded forever that the pool
was wide enough and never tried again: permanent starvation instead of
a retry on the next timer tick.

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

bootstraptest/test_thread.rb

@@ -612,3 +612,51 @@ assert_equal 'ok', %q{
     File.unlink(flagpath) rescue nil
   end
 }
+
+# Creating a thread when no native thread can be spawned must fail cleanly:
+# the thread must not be published to the scheduler before its native thread
+# exists, or an existing shared thread runs it to death concurrently with the
+# creator's failure path (living-set removal races its own).
+assert_equal 'ok', %q{
+  can_limit = begin
+    Process.setrlimit(:NPROC, Process.getrlimit(:NPROC)[0])
+    true
+  rescue StandardError, NotImplementedError
+    false
+  end
+  if !can_limit
+    'ok'   # cannot make thread creation fail on this platform; nothing to test
+  else
+    warm = 2.times.map { Ractor.new { nil until Ractor.receive == :quit } }
+    sleep 0.3   # the pool now has shared native threads parked for the warm ractors
+    Process.setrlimit(:NPROC, 1)
+    # RLIMIT_NPROC binds neither root (CI containers) nor macOS threads;
+    # probe that thread creation actually fails before asserting on it.
+    limited = begin
+      Thread.new {}.join
+      false
+    rescue ThreadError
+      true
+    end
+    result =
+      if !limited
+        'ok'
+      else
+        errs = 0
+        20.times do
+          begin
+            Ractor.new { :born }
+          rescue ThreadError
+            errs += 1
+          end
+        end
+        sleep 0.5   # a wrongly-published thread would be served and die about now
+        errs == 20 ? 'ok' : "#{errs} of 20 raised"
+      end
+    warm.each { |r| r.send(:quit) }
+    warm.each(&:value)
+    GC.start
+    result
+  end
+}
+

thread_pthread_mn.c

@@ -458,7 +458,16 @@ native_thread_check_and_create_shared(rb_vm_t *vm)
     if (need_to_make) {
         struct rb_native_thread *nt = native_thread_alloc();
         nt->vm = vm;
-        return native_thread_create0(nt);
+        int err = native_thread_create0(nt);
+        if (err) {
+            // Roll back, or this function would conclude forever that the
+            // pool is wide enough and never try again.
+            rb_native_mutex_lock(&vm->ractor.sched.lock);
+            vm->ractor.sched.snt_cnt--;
+            rb_native_mutex_unlock(&vm->ractor.sched.lock);
+            native_thread_destroy(nt);
+        }
+        return err;
     }
     else {
         return 0;
@@ -619,11 +628,15 @@ native_thread_create_shared(rb_thread_t *th)
     tctx->co.argument = th;
 
     RUBY_DEBUG_LOG("th:%u vm_stack:%p machine_stack:%p", rb_th_serial(th), vm_stack, machine_stack);
-    thread_sched_to_ready(TH_SCHED(th), th);
 
-    // setup nt.  th is runnable now and a Ractor's thread that runs to its end frees
-    // its own rb_thread_t (rb_ractor_postmortem_free), so th must not be read again.
-    return native_thread_check_and_create_shared(vm);
+    // Widen the pool before publishing th.  Once ready, a Ractor's thread that
+    // runs to its end frees its own rb_thread_t (rb_ractor_postmortem_free),
+    // and the caller's create-failure path assumes th never became runnable.
+    int create_err = native_thread_check_and_create_shared(vm);
+    if (create_err) return create_err;
+
+    thread_sched_to_ready(TH_SCHED(th), th);
+    return 0;
 }
 
 #else // USE_MN_THREADS