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, 56 insertions, 16 deletions

diff --git a/ractor_core.h b/ractor_core.h
index 968c12d291..1b9003b6d8 100644
--- a/ractor_core.h
+++ b/ractor_core.h
@@ -9,22 +9,66 @@
 #endif
 
 enum rb_ractor_basket_type {
+    // basket is empty
     basket_type_none,
+
+    // value is available
     basket_type_ref,
     basket_type_copy,
     basket_type_move,
     basket_type_will,
+
+    // basket should be deleted
     basket_type_deleted,
+
+    // basket is reserved
     basket_type_reserved,
+
+    // take_basket is available
+    basket_type_take_basket,
+
+    // basket is keeping by yielding ractor
+    basket_type_yielding,
+};
+
+// per ractor taking configuration
+struct rb_ractor_selector_take_config {
+    bool closed;
+    bool oneshot;
 };
 
 struct rb_ractor_basket {
-    bool exception;
-    enum rb_ractor_basket_type type;
-    VALUE v;
+    union {
+        enum rb_ractor_basket_type e;
+        rb_atomic_t atomic;
+    } type;
     VALUE sender;
+
+    union {
+        struct {
+            VALUE v;
+            bool exception;
+        } send;
+
+        struct {
+            struct rb_ractor_basket *basket;
+            struct rb_ractor_selector_take_config *config;
+        } take;
+    } p; // payload
 };
 
+static inline bool
+basket_type_p(struct rb_ractor_basket *b, enum rb_ractor_basket_type type)
+{
+    return b->type.e == type;
+}
+
+static inline bool
+basket_none_p(struct rb_ractor_basket *b)
+{
+    return basket_type_p(b, basket_type_none);
+}
+
 struct rb_ractor_queue {
     struct rb_ractor_basket *baskets;
     int start;
@@ -34,12 +78,6 @@ struct rb_ractor_queue {
     unsigned int reserved_cnt;
 };
 
-struct rb_ractor_waiting_list {
-    int cnt;
-    int size;
-    rb_ractor_t **ractors;
-};
-
 enum rb_ractor_wait_status {
     wait_none      = 0x00,
     wait_receiving = 0x01,
@@ -66,18 +104,21 @@ struct rb_ractor_sync {
 #endif
     rb_nativethread_cond_t cond;
 
-    // communication
-    struct rb_ractor_queue  incoming_queue;
-    struct rb_ractor_waiting_list taking_ractors;
-
     bool incoming_port_closed;
     bool outgoing_port_closed;
 
+    // All sent messages will be pushed into recv_queue
+    struct rb_ractor_queue recv_queue;
+
+    // The following ractors waiting for the yielding by this ractor
+    struct rb_ractor_queue takers_queue;
+
+    // Enabled if the ractor already terminated and not taken yet.
+    struct rb_ractor_basket will_basket;
+
     struct ractor_wait {
         enum rb_ractor_wait_status status;
         enum rb_ractor_wakeup_status wakeup_status;
-        struct rb_ractor_basket yielded_basket;
-        struct rb_ractor_basket taken_basket;
     } wait;
 };
 
@@ -107,7 +148,6 @@ struct rb_ractor_struct {
 
     struct rb_ractor_sync sync;
     VALUE receiving_mutex;
-    bool yield_atexit;
 
     // vm wide barrier synchronization
     rb_nativethread_cond_t barrier_wait_cond;