Rewrite Ractor synchronization mechanism

This patch rewrites Ractor synchronization mechanism, send/receive
and take/yield.

* API
  * Ractor::Selector is introduced for lightweight waiting
    for many ractors.
* Data structure
  * remove `struct rb_ractor_waiting_list` and use
    `struct rb_ractor_queue takers_queue` to manage takers.
  * remove `rb_ractor_t::yield_atexit` and use
    `rb_ractor_t::sync::will_basket::type` to check the will.
  * add `rb_ractor_basket::p.take` to represent a taking ractor.
* Synchronization protocol
  * For the Ractor local GC, `take` can not make a copy object
    directly so ask to generate the copy from the yielding ractor.
  * The following steps shows what `r1.take` does on `r0`.
    * step1: (r0) register `r0` into `r1`'s takers.
    * step2: (r0) check `r1`'s status and wakeup r0 if `r1` is waiting
             for yielding a value.
    * step3: (r0) sleep until `r1` wakes up `r0`.
  * The following steps shows what `Ractor.yield(v)` on `r1`.
    * step1: (r1) check first takers of `r1` and if there is (`r0`),
             make a copy object of `v` and pass it to `r0` and
             wakes up `r0`.
    * step2: (r1) if there is no taker ractors, sleep until
             another ractor try to take.

1 files changed, 111 insertions, 8 deletions

diff --git a/ractor.rb b/ractor.rb
index fc9462c149..6c57d65f87 100644
--- a/ractor.rb
+++ b/ractor.rb
@@ -358,14 +358,117 @@ class Ractor
   def self.select(*ractors, yield_value: yield_unspecified = true, move: false)
     raise ArgumentError, 'specify at least one ractor or `yield_value`' if yield_unspecified && ractors.empty?
 
-    __builtin_cstmt! %q{
-      const VALUE *rs = RARRAY_CONST_PTR_TRANSIENT(ractors);
-      VALUE rv;
-      VALUE v = ractor_select(ec, rs, RARRAY_LENINT(ractors),
-                              yield_unspecified == Qtrue ? Qundef : yield_value,
-                              (bool)RTEST(move) ? true : false, &rv);
-      return rb_ary_new_from_args(2, rv, v);
-    }
+    begin
+      if ractors.delete Ractor.current
+        do_receive = true
+      else
+        do_receive = false
+      end
+      selector = Ractor::Selector.new(*ractors)
+
+      if yield_unspecified
+        selector.wait receive: do_receive
+      else
+        selector.wait receive: do_receive, yield_value: yield_value, move: move
+      end
+    ensure
+      selector.clear
+    end
+  end
+
+  #
+  # Ractor::Selector provides a functionality to wait multiple Ractor events.
+  # Ractor::Selector#wait is more lightweight than Ractor.select()
+  # because we don't have to specify all target ractors for each wait time.
+  #
+  # Ractor.select() uses Ractor::Selector internally to implement it.
+  #
+  class Selector
+    # call-seq:
+    #   Ractor::Selector.new(*ractors)
+    #
+    # Creates a selector object.
+    #
+    # If a ractors parameter is given, it is same as the following code.
+    #
+    #    selector = Ractor::Selector.new
+    #    ractors.each{|r| selector.add r}
+    #
+    def self.new(*rs)
+      selector = __builtin_cexpr! %q{
+        ractor_selector_create(self);
+      }
+      rs.each{|r| selector.add(r) }
+      selector
+    end
+
+    # call-seq:
+    #   selector.add(ractor)
+    #
+    # Registers a ractor as a taking target by the selector.
+    #
+    def add r
+      __builtin_ractor_selector_add r
+    end
+
+    # call-seq:
+    #   selector.remove(ractor)
+    #
+    # Deregisters a ractor as a taking target by the selector.
+    #
+    def remove r
+      __builtin_ractor_selector_remove r
+    end
+
+    # call-seq:
+    #   selector.clear
+    #
+    # Deregisters all ractors.
+    def clear
+      __builtin_ractor_selector_clear
+    end
+
+    # call-seq:
+    #   selector.wait(receive: false, yield_value: yield_value, move: false) -> [ractor or symbol, value]
+    #
+    # Waits Ractor events. It is lighter than Ractor.select() for many ractors.
+    #
+    # The simplest form is waiting for taking a value from one of
+    # registerred ractors like that.
+    #
+    #   selector = Ractor::Selector.new(r1, r2, r3)
+    #   r, v = selector.wait
+    #
+    # On this case, when r1, r2 or r3 is ready to take (yielding a value),
+    # this method takes the value from the ready (yielded) ractor
+    # and returns [the yielded ractor, the taking value].
+    #
+    # Note that if a take target ractor is closed, the ractor will be removed
+    # automatically.
+    #
+    # If you also want to wait with receiving an object from other ractors,
+    # you can specify receive: true keyword like:
+    #
+    #   r, v = selector.wait receive: true
+    #
+    # On this case, wait for taking from r1, r2 or r3 and waiting for receving
+    # a value from other ractors.
+    # If it successes the receiving, it returns an array object [:receive, the received value].
+    #
+    # If you also want to wait with yielding a value, you can specify
+    # :yield_value like:
+    #
+    #   r, v = selector.wait yield_value: obj
+    #
+    # On this case wait for taking from r1, r2, or r3 and waiting for taking
+    # yielding value (obj) by another ractor.
+    # If antoher ractor takes the value (obj), it returns an array object [:yield, nil].
+    #
+    # You can specify a keyword parameter <tt>move: true</tt> like Ractor.yield(obj, move: true)
+    #
+    def wait receive: false, yield_value: yield_unspecified = true, move: false
+      __builtin_ractor_selector_wait receive, !yield_unspecified, yield_value, move
+    end
   end
 
   #