Ractor#monitor: name the ractor in the exit token

A monitoring port received a bare :exited or :aborted, so a port could only
ever watch one ractor: with two, the token that arrived said what happened
but not to which.  Watching a group meant a port per ractor and a table
from the port back to the ractor -- which is what Ractor.select did, and
what anyone writing a supervisor had to do as well.

The token is now [ractor, :exited] or [ractor, :aborted], so one port can
watch a whole group and a supervisor needs nothing but monitor:

    workers.each { |r| r.monitor port }

    until workers.empty?
      r, status = port.receive
      workers.delete(r)
      workers << restart(r) if status == :aborted
    end

The pair is built by the receiver, from a basket type of its own, rather
than sent as a shareable array.  A shareable array would be simpler -- four
lines against thirty -- but it is left behind for every ractor that ever
terminates, and shareable objects are only reclaimed by a global
collection.  Creating 20,000 ractors and joining them:

                          wall       RSS held afterwards
    built on receipt     309 ms                    98 MB
    shareable array      570 ms                   533 MB

It cannot be sent as an ordinary copy either: the tokens go out from a
thread that has no VM stack left, which is what ractor_basket_new_ref
exists to avoid, and a copy pushes a tag.  Both halves are shareable
already, so the basket holds them as they are and the array is allocated
where it is read.

The array costs join, which received one symbol per call, 0.309us to
0.327us on a terminated ractor.  Ractor.select pays it back many times over
in the commit that follows.

This is an incompatible change to an experimental API, so NEWS says so.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

NEWS.md

@@ -388,6 +388,28 @@ A lot of work has gone into making Ractors more stable, performant, and usable.
   * `ObjectSpace.define_finalizer` on another Ractor's object raises
     `Ractor::IsolationError`.
 
+* `Ractor#monitor` now sends an Array naming the Ractor and what happened to
+  it, `[ractor, :exited]` or `[ractor, :aborted]`, where it used to send the
+  bare Symbol `:exited` or `:aborted`.  Several Ractors can then report to one
+  port and the receiver still knows which one finished.  The Array is built for
+  the receiving Ractor, so watching many Ractors leaves no shareable objects
+  behind.
+
+      r = Ractor.new { :ok }
+      r.monitor(port = Ractor::Port.new)
+      port.receive #=> [r, :exited]
+
+  One port can therefore watch a whole group, which is all a supervisor
+  needs:
+
+      workers.each { |r| r.monitor port }
+
+      until workers.empty?
+        r, status = port.receive
+        workers.delete(r)
+        workers << restart(r) if status == :aborted
+      end
+
 ### M:N thread scheduler
 
 * The scheduler scales with the number of waiters and of Ractors, where it

bootstraptest/test_ractor.rb

@@ -2710,7 +2710,7 @@ assert_equal 'ok', %q{
 
 ## Ractor#monitor
 
-# monitor port returns `:exited` when the monitering Ractor terminated.
+# monitor port returns [ractor, :exited] when the monitering Ractor terminated.
 assert_equal 'true', %q{
   r = Ractor.new do
     Ractor.main << :ok1
@@ -2719,10 +2719,10 @@ assert_equal 'true', %q{
 
   r.monitor port = Ractor::Port.new
   Ractor.receive # :ok1
-  port.receive == :exited
+  port.receive == [r, :exited]
 }
 
-# monitor port returns `:exited` even if the monitoring Ractor was terminated.
+# monitor port returns [ractor, :exited] even if the monitoring Ractor was terminated.
 assert_equal 'true', %q{
   r = Ractor.new do
     :ok
@@ -2731,7 +2731,7 @@ assert_equal 'true', %q{
   r.join # wait for r's terminateion
 
   r.monitor port = Ractor::Port.new
-  port.receive == :exited
+  port.receive == [r, :exited]
 }
 
 # monitor returns false if the monitoring Ractor was terminated.
@@ -2745,7 +2745,7 @@ assert_equal 'false', %q{
   r.monitor Ractor::Port.new
 }
 
-# monitor port returns `:aborted` when the monitering Ractor is aborted.
+# monitor port returns [ractor, :aborted] when the monitering Ractor is aborted.
 assert_equal 'true', %q{
   r = Ractor.new do
     Ractor.main << :ok1
@@ -2754,10 +2754,10 @@ assert_equal 'true', %q{
 
   r.monitor port = Ractor::Port.new
   Ractor.receive # :ok1
-  port.receive == :aborted
+  port.receive == [r, :aborted]
 }
 
-# monitor port returns `:aborted` even if the monitoring Ractor was aborted.
+# monitor port returns [ractor, :aborted] even if the monitoring Ractor was aborted.
 assert_equal 'true', %q{
   r = Ractor.new do
     raise 'ok'
@@ -2770,7 +2770,7 @@ assert_equal 'true', %q{
   end
 
   r.monitor port = Ractor::Port.new
-  port.receive == :aborted
+  port.receive == [r, :aborted]
 }
 
 assert_equal 'ok', %q{

ractor.rb

@@ -599,7 +599,7 @@ class Ractor
     port = Port.new
 
     self.monitor port
-    if port.receive == :aborted
+    if port.receive[1] == :aborted
       __builtin_ractor_value
     end
 
@@ -632,23 +632,28 @@ class Ractor
   # call-seq:
   #    ractor.monitor(port) -> true or false
   #
-  # Registers the port as a monitoring port for this ractor. When the ractor terminates,
-  # the port receives a Symbol object.
+  # Registers the port as a monitoring port for this ractor. When the ractor
+  # terminates, the port receives an Array naming the ractor and what happened to
+  # it, so that several ractors can report to one port.
   #
-  # * +:exited+ is sent if the ractor terminates without an unhandled exception.
-  # * +:aborted+ is sent if the ractor terminates by an unhandled exception.
+  # * <tt>[ractor, :exited]</tt> if the ractor terminated without an unhandled
+  #   exception.
+  # * <tt>[ractor, :aborted]</tt> if it terminated by one.
+  #
+  # The Array is built for the receiving ractor, so watching many ractors does not
+  # leave shareable objects behind.
   #
   # Returns +true+ if the monitor was registered (the ractor is still running).
   # Returns +false+ if the ractor had already terminated; in that case the
-  # termination message (+:exited+ or +:aborted+) is sent to the port immediately.
+  # termination message is sent to the port immediately.
   #
   #     r = Ractor.new{ some_task() }
   #     r.monitor(port = Ractor::Port.new)
-  #     port.receive #=> :exited and r is terminated
+  #     port.receive #=> [r, :exited]
   #
   #     r = Ractor.new{ raise "foo" }
   #     r.monitor(port = Ractor::Port.new)
-  #     port.receive #=> :aborted and r is terminated by the RuntimeError "foo"
+  #     port.receive #=> [r, :aborted]
   #
   def monitor port
     __builtin_ractor_monitor(port)

ractor_sync.c

@@ -16,6 +16,7 @@ static VALUE rb_cRactorPort;
 static VALUE ractor_receive(rb_execution_context_t *ec, const struct ractor_port *rp, const rb_hrtime_t *end);
 static VALUE ractor_send(rb_execution_context_t *ec, const struct ractor_port *rp, VALUE obj, VALUE move);
 static struct ractor_basket *ractor_basket_new_ref(VALUE shareable);
+static struct ractor_basket *ractor_basket_new_exit(VALUE sender, VALUE token);
 static void ractor_send_basket(rb_execution_context_t *ec, const struct ractor_port *rp, struct ractor_basket *b, bool raise_on_error);
 static void ractor_add_port(rb_ractor_t *r, st_data_t id);
 
@@ -235,6 +236,10 @@ enum ractor_basket_type {
     basket_type_ref,
     basket_type_copy,
     basket_type_move,
+    /* A ractor's exit token.  The pair it becomes is built on the receiving side,
+     * so a terminating ractor adds no shareable object to the vm, and building it
+     * needs neither a tag nor a courier -- which the sender has no stack for. */
+    basket_type_exit,
 };
 
 struct ractor_basket {
@@ -291,6 +296,9 @@ ractor_basket_mark(const struct ractor_basket *b)
         /* Marshaled bytes are off-heap and hold nothing to mark. */
         rb_gc_mark(b->p.v);
     }
+
+    /* An exit token names a ractor that may be reachable from nothing else. */
+    rb_gc_mark(b->sender);
 }
 
 static void
@@ -770,10 +778,12 @@ ractor_mark_monitors(rb_ractor_t *r)
     }
 }
 
+/* Paired with the ractor it is about when it is received, so that many ractors
+ * can report to one port. */
 static VALUE
-ractor_exit_token(bool exc)
+ractor_exit_token(const rb_ractor_t *r)
 {
-    if (exc) {
+    if (r->sync.legacy_exc) {
         RUBY_DEBUG_LOG("aborted");
         return ID2SYM(idAborted);
     }
@@ -807,7 +817,7 @@ ractor_monitor(rb_execution_context_t *ec, VALUE self, VALUE port)
 
     if (terminated) {
         SIZED_FREE(rm);
-        ractor_port_send(ec, port, ractor_exit_token(r->sync.legacy_exc), Qfalse);
+        ractor_send_basket(ec, rp, ractor_basket_new_exit(self, ractor_exit_token(r)), false);
 
         return Qfalse;
     }
@@ -867,14 +877,14 @@ ractor_notify_exit(rb_execution_context_t *ec, rb_ractor_t *cr, VALUE legacy, bo
 static void
 ractor_send_exit_tokens(rb_execution_context_t *ec, rb_ractor_t *cr)
 {
-    VALUE token = ractor_exit_token(cr->sync.legacy_exc);
+    VALUE token = ractor_exit_token(cr);
     struct ractor_monitor *rm, *nxt;
 
     ccan_list_for_each_safe(&cr->sync.monitors, rm, nxt, node)
     {
         RUBY_DEBUG_LOG("port:%u@r%u", (unsigned int)ractor_port_id(&rm->port), (unsigned int)rb_ractor_id(rm->port.r));
 
-        ractor_send_basket(ec, &rm->port, ractor_basket_new_ref(token), false);
+        ractor_send_basket(ec, &rm->port, ractor_basket_new_exit(cr->pub.self, token), false);
 
         ccan_list_del(&rm->node);
         SIZED_FREE(rm);
@@ -1208,6 +1218,9 @@ ractor_basket_value(struct ractor_basket *b)
     switch (b->type) {
       case basket_type_ref:
         break;
+      case basket_type_exit:
+        /* Allocated here, in the receiving ractor: a copy, not a shared object. */
+        return rb_ary_new_from_args(2, b->sender, b->p.v);
       case basket_type_copy: {
         /* An off-heap copy courier rebuilds exactly like a move one; only the sources
          * differ (still alive here, already shells there). */
@@ -1347,6 +1360,7 @@ basket_type_name(enum ractor_basket_type type)
       case basket_type_ref: return "ref";
       case basket_type_copy: return "copy";
       case basket_type_move: return "move";
+      case basket_type_exit: return "exit";
     }
     VM_ASSERT(0);
     return NULL;
@@ -1676,6 +1690,20 @@ ractor_basket_new_ref(VALUE shareable)
     return b;
 }
 
+/* sender is the ractor the token is about; both it and the token are shareable,
+ * so nothing is copied until the receiver builds the pair. */
+static struct ractor_basket *
+ractor_basket_new_exit(VALUE sender, VALUE token)
+{
+    struct ractor_basket *b = ractor_basket_alloc();
+
+    b->type = basket_type_exit;
+    b->sender = sender;
+    b->p.v = token;
+
+    return b;
+}
+
 static VALUE
 ractor_send0(rb_execution_context_t *ec, const struct ractor_port *rp, VALUE obj, VALUE move, bool raise_on_error)
 {

test/ruby/test_ractor.rb

@@ -751,7 +751,7 @@ class TestRactor < Test::Unit::TestCase
       b = Ractor.new(target) do |t|
         t.monitor(p = Ractor::Port.new)
         Ractor.main << :ready
-        p.receive
+        p.receive == [t, :exited]
       end
 
       Ractor.receive  # b's monitor is registered
@@ -765,7 +765,7 @@ class TestRactor < Test::Unit::TestCase
       end
 
       assert_equal :ok, a.value
-      assert_equal :exited, b.value
+      assert_equal true, b.value
     RUBY
   end