1 files changed, 621 insertions, 805 deletions

diff --git a/thread_sync.c b/thread_sync.c
index 955e552945..c9e056dfe2 100644
--- a/thread_sync.c
+++ b/thread_sync.c
@@ -2,12 +2,12 @@
 #include "ccan/list/list.h"
 #include "builtin.h"
 
-static VALUE rb_cMutex, rb_cQueue, rb_cSizedQueue, rb_cConditionVariable;
-static VALUE rb_eClosedQueueError;
+static VALUE rb_cMutex, rb_eClosedQueueError;
 
 /* Mutex */
 typedef struct rb_mutex_struct {
-    rb_fiber_t *fiber;
+    rb_serial_t ec_serial;
+    rb_thread_t *th; // even if the fiber is collected, we might need access to the thread in mutex_free
     struct rb_mutex_struct *next_mutex;
     struct ccan_list_head waitq; /* protected by GVL */
 } rb_mutex_t;
@@ -80,33 +80,7 @@ static void rb_mutex_abandon_all(rb_mutex_t *mutexes);
 static void rb_mutex_abandon_keeping_mutexes(rb_thread_t *th);
 static void rb_mutex_abandon_locking_mutex(rb_thread_t *th);
 #endif
-static const char* rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_fiber_t *fiber);
-
-/*
- *  Document-class: Thread::Mutex
- *
- *  Thread::Mutex implements a simple semaphore that can be used to
- *  coordinate access to shared data from multiple concurrent threads.
- *
- *  Example:
- *
- *    semaphore = Thread::Mutex.new
- *
- *    a = Thread.new {
- *      semaphore.synchronize {
- *        # access shared resource
- *      }
- *    }
- *
- *    b = Thread.new {
- *      semaphore.synchronize {
- *        # access shared resource
- *      }
- *    }
- *
- */
-
-#define mutex_mark ((void(*)(void*))0)
+static const char* rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial);
 
 static size_t
 rb_mutex_num_waiting(rb_mutex_t *mutex)
@@ -123,13 +97,18 @@ rb_mutex_num_waiting(rb_mutex_t *mutex)
 
 rb_thread_t* rb_fiber_threadptr(const rb_fiber_t *fiber);
 
+static bool
+mutex_locked_p(rb_mutex_t *mutex)
+{
+    return mutex->ec_serial != 0;
+}
+
 static void
 mutex_free(void *ptr)
 {
     rb_mutex_t *mutex = ptr;
-    if (mutex->fiber) {
-        /* rb_warn("free locked mutex"); */
-        const char *err = rb_mutex_unlock_th(mutex, rb_fiber_threadptr(mutex->fiber), mutex->fiber);
+    if (mutex_locked_p(mutex)) {
+        const char *err = rb_mutex_unlock_th(mutex, mutex->th, 0);
         if (err) rb_bug("%s", err);
     }
     ruby_xfree(ptr);
@@ -143,8 +122,8 @@ mutex_memsize(const void *ptr)
 
 static const rb_data_type_t mutex_data_type = {
     "mutex",
-    {mutex_mark, mutex_free, mutex_memsize,},
-    0, 0, RUBY_TYPED_WB_PROTECTED | RUBY_TYPED_FREE_IMMEDIATELY
+    {NULL, mutex_free, mutex_memsize,},
+    0, 0, RUBY_TYPED_FREE_IMMEDIATELY
 };
 
 static rb_mutex_t *
@@ -175,41 +154,24 @@ mutex_alloc(VALUE klass)
     return obj;
 }
 
-/*
- *  call-seq:
- *     Thread::Mutex.new   -> mutex
- *
- *  Creates a new Mutex
- */
-static VALUE
-mutex_initialize(VALUE self)
-{
-    return self;
-}
-
 VALUE
 rb_mutex_new(void)
 {
     return mutex_alloc(rb_cMutex);
 }
 
-/*
- * call-seq:
- *    mutex.locked?  -> true or false
- *
- * Returns +true+ if this lock is currently held by some thread.
- */
 VALUE
 rb_mutex_locked_p(VALUE self)
 {
     rb_mutex_t *mutex = mutex_ptr(self);
 
-    return RBOOL(mutex->fiber);
+    return RBOOL(mutex_locked_p(mutex));
 }
 
 static void
 thread_mutex_insert(rb_thread_t *thread, rb_mutex_t *mutex)
 {
+    RUBY_ASSERT(!mutex->next_mutex);
     if (thread->keeping_mutexes) {
         mutex->next_mutex = thread->keeping_mutexes;
     }
@@ -234,45 +196,50 @@ thread_mutex_remove(rb_thread_t *thread, rb_mutex_t *mutex)
 }
 
 static void
-mutex_locked(rb_thread_t *th, VALUE self)
+mutex_set_owner(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial)
 {
-    rb_mutex_t *mutex = mutex_ptr(self);
+    mutex->th = th;
+    mutex->ec_serial = ec_serial;
+}
 
+static void
+mutex_locked(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial)
+{
+    mutex_set_owner(mutex, th, ec_serial);
     thread_mutex_insert(th, mutex);
 }
 
-/*
- * call-seq:
- *    mutex.try_lock  -> true or false
- *
- * Attempts to obtain the lock and returns immediately. Returns +true+ if the
- * lock was granted.
- */
-VALUE
-rb_mutex_trylock(VALUE self)
+static inline bool
+do_mutex_trylock(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial)
 {
-    rb_mutex_t *mutex = mutex_ptr(self);
-
-    if (mutex->fiber == 0) {
+    if (mutex->ec_serial == 0) {
         RUBY_DEBUG_LOG("%p ok", mutex);
 
-        rb_fiber_t *fiber = GET_EC()->fiber_ptr;
-        rb_thread_t *th = GET_THREAD();
-        mutex->fiber = fiber;
-
-        mutex_locked(th, self);
-        return Qtrue;
+        mutex_locked(mutex, th, ec_serial);
+        return true;
     }
     else {
         RUBY_DEBUG_LOG("%p ng", mutex);
-        return Qfalse;
+        return false;
     }
 }
 
 static VALUE
-mutex_owned_p(rb_fiber_t *fiber, rb_mutex_t *mutex)
+rb_mut_trylock(rb_execution_context_t *ec, VALUE self)
 {
-    return RBOOL(mutex->fiber == fiber);
+    return RBOOL(do_mutex_trylock(mutex_ptr(self), ec->thread_ptr, rb_ec_serial(ec)));
+}
+
+VALUE
+rb_mutex_trylock(VALUE self)
+{
+    return rb_mut_trylock(GET_EC(), self);
+}
+
+static VALUE
+mutex_owned_p(rb_serial_t ec_serial, rb_mutex_t *mutex)
+{
+    return RBOOL(mutex->ec_serial == ec_serial);
 }
 
 static VALUE
@@ -292,13 +259,29 @@ delete_from_waitq(VALUE value)
 
 static inline rb_atomic_t threadptr_get_interrupts(rb_thread_t *th);
 
+struct mutex_args {
+    VALUE self;
+    rb_mutex_t *mutex;
+    rb_execution_context_t *ec;
+};
+
+static inline void
+mutex_args_init(struct mutex_args *args, VALUE mutex)
+{
+    args->self = mutex;
+    args->mutex = mutex_ptr(mutex);
+    args->ec = GET_EC();
+}
+
 static VALUE
-do_mutex_lock(VALUE self, int interruptible_p)
+do_mutex_lock(struct mutex_args *args, int interruptible_p)
 {
-    rb_execution_context_t *ec = GET_EC();
+    VALUE self = args->self;
+    rb_execution_context_t *ec = args->ec;
     rb_thread_t *th = ec->thread_ptr;
     rb_fiber_t *fiber = ec->fiber_ptr;
-    rb_mutex_t *mutex = mutex_ptr(self);
+    rb_serial_t ec_serial = rb_ec_serial(ec);
+    rb_mutex_t *mutex = args->mutex;
     rb_atomic_t saved_ints = 0;
 
     /* When running trap handler */
@@ -307,13 +290,13 @@ do_mutex_lock(VALUE self, int interruptible_p)
         rb_raise(rb_eThreadError, "can't be called from trap context");
     }
 
-    if (rb_mutex_trylock(self) == Qfalse) {
-        if (mutex->fiber == fiber) {
+    if (!do_mutex_trylock(mutex, th, ec_serial)) {
+        if (mutex->ec_serial == ec_serial) {
             rb_raise(rb_eThreadError, "deadlock; recursive locking");
         }
 
-        while (mutex->fiber != fiber) {
-            VM_ASSERT(mutex->fiber != NULL);
+        while (mutex->ec_serial != ec_serial) {
+            VM_ASSERT(mutex->ec_serial != 0);
 
             VALUE scheduler = rb_fiber_scheduler_current();
             if (scheduler != Qnil) {
@@ -327,12 +310,12 @@ do_mutex_lock(VALUE self, int interruptible_p)
 
                 rb_ensure(call_rb_fiber_scheduler_block, self, delete_from_waitq, (VALUE)&sync_waiter);
 
-                if (!mutex->fiber) {
-                    mutex->fiber = fiber;
+                if (!mutex->ec_serial) {
+                    mutex_set_owner(mutex, th, ec_serial);
                 }
             }
             else {
-                if (!th->vm->thread_ignore_deadlock && rb_fiber_threadptr(mutex->fiber) == th) {
+                if (!th->vm->thread_ignore_deadlock && mutex->th == th) {
                     rb_raise(rb_eThreadError, "deadlock; lock already owned by another fiber belonging to the same thread");
                 }
 
@@ -358,6 +341,7 @@ do_mutex_lock(VALUE self, int interruptible_p)
                 rb_ractor_sleeper_threads_inc(th->ractor);
                 rb_check_deadlock(th->ractor);
 
+                RUBY_ASSERT(!th->locking_mutex);
                 th->locking_mutex = self;
 
                 ccan_list_add_tail(&mutex->waitq, &sync_waiter.node);
@@ -367,14 +351,13 @@ do_mutex_lock(VALUE self, int interruptible_p)
                 ccan_list_del(&sync_waiter.node);
 
                 // unlocked by another thread while sleeping
-                if (!mutex->fiber) {
-                    mutex->fiber = fiber;
+                if (!mutex->ec_serial) {
+                    mutex_set_owner(mutex, th, ec_serial);
                 }
 
                 rb_ractor_sleeper_threads_dec(th->ractor);
                 th->status = prev_status;
                 th->locking_mutex = Qfalse;
-                th->locking_mutex = Qfalse;
 
                 RUBY_DEBUG_LOG("%p wakeup", mutex);
             }
@@ -382,10 +365,13 @@ do_mutex_lock(VALUE self, int interruptible_p)
             if (interruptible_p) {
                 /* release mutex before checking for interrupts...as interrupt checking
                  * code might call rb_raise() */
-                if (mutex->fiber == fiber) mutex->fiber = 0;
+                if (mutex->ec_serial == ec_serial) {
+                    mutex->th = NULL;
+                    mutex->ec_serial = 0;
+                }
                 RUBY_VM_CHECK_INTS_BLOCKING(th->ec); /* may release mutex */
-                if (!mutex->fiber) {
-                    mutex->fiber = fiber;
+                if (!mutex->ec_serial) {
+                    mutex_set_owner(mutex, th, ec_serial);
                 }
             }
             else {
@@ -404,13 +390,13 @@ do_mutex_lock(VALUE self, int interruptible_p)
         }
 
         if (saved_ints) th->ec->interrupt_flag = saved_ints;
-        if (mutex->fiber == fiber) mutex_locked(th, self);
+        if (mutex->ec_serial == ec_serial) mutex_locked(mutex, th, ec_serial);
     }
 
     RUBY_DEBUG_LOG("%p locked", mutex);
 
     // assertion
-    if (mutex_owned_p(fiber, mutex) == Qfalse) rb_bug("do_mutex_lock: mutex is not owned.");
+    if (mutex_owned_p(ec_serial, mutex) == Qfalse) rb_bug("do_mutex_lock: mutex is not owned.");
 
     return self;
 }
@@ -418,52 +404,57 @@ do_mutex_lock(VALUE self, int interruptible_p)
 static VALUE
 mutex_lock_uninterruptible(VALUE self)
 {
-    return do_mutex_lock(self, 0);
+    struct mutex_args args;
+    mutex_args_init(&args, self);
+    return do_mutex_lock(&args, 0);
+}
+
+static VALUE
+rb_mut_lock(rb_execution_context_t *ec, VALUE self)
+{
+    struct mutex_args args = {
+        .self = self,
+        .mutex = mutex_ptr(self),
+        .ec = ec,
+    };
+    return do_mutex_lock(&args, 1);
 }
 
-/*
- * call-seq:
- *    mutex.lock  -> self
- *
- * Attempts to grab the lock and waits if it isn't available.
- * Raises +ThreadError+ if +mutex+ was locked by the current thread.
- */
 VALUE
 rb_mutex_lock(VALUE self)
 {
-    return do_mutex_lock(self, 1);
+    struct mutex_args args;
+    mutex_args_init(&args, self);
+    return do_mutex_lock(&args, 1);
+}
+
+static VALUE
+rb_mut_owned_p(rb_execution_context_t *ec, VALUE self)
+{
+    return mutex_owned_p(rb_ec_serial(ec), mutex_ptr(self));
 }
 
-/*
- * call-seq:
- *    mutex.owned?  -> true or false
- *
- * Returns +true+ if this lock is currently held by current thread.
- */
 VALUE
 rb_mutex_owned_p(VALUE self)
 {
-    rb_fiber_t *fiber = GET_EC()->fiber_ptr;
-    rb_mutex_t *mutex = mutex_ptr(self);
-
-    return mutex_owned_p(fiber, mutex);
+    return rb_mut_owned_p(GET_EC(), self);
 }
 
 static const char *
-rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_fiber_t *fiber)
+rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_serial_t ec_serial)
 {
     RUBY_DEBUG_LOG("%p", mutex);
 
-    if (mutex->fiber == 0) {
+    if (mutex->ec_serial == 0) {
         return "Attempt to unlock a mutex which is not locked";
     }
-    else if (mutex->fiber != fiber) {
+    else if (ec_serial && mutex->ec_serial != ec_serial) {
         return "Attempt to unlock a mutex which is locked by another thread/fiber";
     }
 
     struct sync_waiter *cur = 0, *next;
 
-    mutex->fiber = 0;
+    mutex->ec_serial = 0;
     thread_mutex_remove(th, mutex);
 
     ccan_list_for_each_safe(&mutex->waitq, cur, next, node) {
@@ -494,6 +485,24 @@ rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_fiber_t *fiber)
     return NULL;
 }
 
+static void
+do_mutex_unlock(struct mutex_args *args)
+{
+    const char *err;
+    rb_mutex_t *mutex = args->mutex;
+    rb_thread_t *th = rb_ec_thread_ptr(args->ec);
+
+    err = rb_mutex_unlock_th(mutex, th, rb_ec_serial(args->ec));
+    if (err) rb_raise(rb_eThreadError, "%s", err);
+}
+
+static VALUE
+do_mutex_unlock_safe(VALUE args)
+{
+    do_mutex_unlock((struct mutex_args *)args);
+    return Qnil;
+}
+
 /*
  * call-seq:
  *    mutex.unlock    -> self
@@ -504,13 +513,21 @@ rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_fiber_t *fiber)
 VALUE
 rb_mutex_unlock(VALUE self)
 {
-    const char *err;
-    rb_mutex_t *mutex = mutex_ptr(self);
-    rb_thread_t *th = GET_THREAD();
-
-    err = rb_mutex_unlock_th(mutex, th, GET_EC()->fiber_ptr);
-    if (err) rb_raise(rb_eThreadError, "%s", err);
+    struct mutex_args args;
+    mutex_args_init(&args, self);
+    do_mutex_unlock(&args);
+    return self;
+}
 
+static VALUE
+rb_mut_unlock(rb_execution_context_t *ec, VALUE self)
+{
+    struct mutex_args args = {
+        .self = self,
+        .mutex = mutex_ptr(self),
+        .ec = ec,
+    };
+    do_mutex_unlock(&args);
     return self;
 }
 
@@ -541,7 +558,7 @@ rb_mutex_abandon_all(rb_mutex_t *mutexes)
     while (mutexes) {
         mutex = mutexes;
         mutexes = mutex->next_mutex;
-        mutex->fiber = 0;
+        mutex->ec_serial = 0;
         mutex->next_mutex = 0;
         ccan_list_head_init(&mutex->waitq);
     }
@@ -579,15 +596,15 @@ mutex_sleep_begin(VALUE _arguments)
     return woken;
 }
 
-VALUE
-rb_mutex_sleep(VALUE self, VALUE timeout)
+static VALUE
+rb_mut_sleep(rb_execution_context_t *ec, VALUE self, VALUE timeout)
 {
     if (!NIL_P(timeout)) {
         // Validate the argument:
         rb_time_interval(timeout);
     }
 
-    rb_mutex_unlock(self);
+    rb_mut_unlock(ec, self);
     time_t beg = time(0);
 
     struct rb_mutex_sleep_arguments arguments = {
@@ -595,69 +612,45 @@ rb_mutex_sleep(VALUE self, VALUE timeout)
         .timeout = timeout,
     };
 
-    VALUE woken = rb_ensure(mutex_sleep_begin, (VALUE)&arguments, mutex_lock_uninterruptible, self);
+    VALUE woken = rb_ec_ensure(ec, mutex_sleep_begin, (VALUE)&arguments, mutex_lock_uninterruptible, self);
 
-    RUBY_VM_CHECK_INTS_BLOCKING(GET_EC());
+    RUBY_VM_CHECK_INTS_BLOCKING(ec);
     if (!woken) return Qnil;
     time_t end = time(0) - beg;
     return TIMET2NUM(end);
 }
 
-/*
- * call-seq:
- *    mutex.sleep(timeout = nil)    -> number or nil
- *
- * Releases the lock and sleeps +timeout+ seconds if it is given and
- * non-nil or forever.  Raises +ThreadError+ if +mutex+ wasn't locked by
- * the current thread.
- *
- * When the thread is next woken up, it will attempt to reacquire
- * the lock.
- *
- * Note that this method can wakeup without explicit Thread#wakeup call.
- * For example, receiving signal and so on.
- *
- * Returns the slept time in seconds if woken up, or +nil+ if timed out.
- */
-static VALUE
-mutex_sleep(int argc, VALUE *argv, VALUE self)
+VALUE
+rb_mutex_sleep(VALUE self, VALUE timeout)
 {
-    VALUE timeout;
-
-    timeout = rb_check_arity(argc, 0, 1) ? argv[0] : Qnil;
-    return rb_mutex_sleep(self, timeout);
+    return rb_mut_sleep(GET_EC(), self, timeout);
 }
 
-/*
- * call-seq:
- *    mutex.synchronize { ... }    -> result of the block
- *
- * Obtains a lock, runs the block, and releases the lock when the block
- * completes.  See the example under Thread::Mutex.
- */
-
 VALUE
-rb_mutex_synchronize(VALUE mutex, VALUE (*func)(VALUE arg), VALUE arg)
+rb_mutex_synchronize(VALUE self, VALUE (*func)(VALUE arg), VALUE arg)
 {
-    rb_mutex_lock(mutex);
-    return rb_ensure(func, arg, rb_mutex_unlock, mutex);
+    struct mutex_args args;
+    mutex_args_init(&args, self);
+    do_mutex_lock(&args, 1);
+    return rb_ec_ensure(args.ec, func, arg, do_mutex_unlock_safe, (VALUE)&args);
 }
 
-/*
- * call-seq:
- *    mutex.synchronize { ... }    -> result of the block
- *
- * Obtains a lock, runs the block, and releases the lock when the block
- * completes.  See the example under Thread::Mutex.
- */
 static VALUE
-rb_mutex_synchronize_m(VALUE self)
+do_ec_yield(VALUE _ec)
 {
-    if (!rb_block_given_p()) {
-        rb_raise(rb_eThreadError, "must be called with a block");
-    }
+    return rb_ec_yield((rb_execution_context_t *)_ec, Qundef);
+}
 
-    return rb_mutex_synchronize(self, rb_yield, Qundef);
+VALUE
+rb_mut_synchronize(rb_execution_context_t *ec, VALUE self)
+{
+    struct mutex_args args = {
+        .self = self,
+        .mutex = mutex_ptr(self),
+        .ec = ec,
+    };
+    do_mutex_lock(&args, 1);
+    return rb_ec_ensure(args.ec, do_ec_yield, (VALUE)ec, do_mutex_unlock_safe, (VALUE)&args);
 }
 
 void
@@ -673,46 +666,68 @@ rb_mutex_allow_trap(VALUE self, int val)
 
 /* Queue */
 
-#define queue_waitq(q) UNALIGNED_MEMBER_PTR(q, waitq)
-#define queue_list(q) UNALIGNED_MEMBER_PTR(q, que)
-RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_BEGIN()
 struct rb_queue {
     struct ccan_list_head waitq;
     rb_serial_t fork_gen;
-    const VALUE que;
+    long capa;
+    long len;
+    long offset;
+    VALUE *buffer;
     int num_waiting;
-} RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_END();
+};
+
+#define szqueue_waitq(sq) &sq->q.waitq
+#define szqueue_pushq(sq) &sq->pushq
 
-#define szqueue_waitq(sq) UNALIGNED_MEMBER_PTR(sq, q.waitq)
-#define szqueue_list(sq) UNALIGNED_MEMBER_PTR(sq, q.que)
-#define szqueue_pushq(sq) UNALIGNED_MEMBER_PTR(sq, pushq)
-RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_BEGIN()
 struct rb_szqueue {
     struct rb_queue q;
     int num_waiting_push;
     struct ccan_list_head pushq;
     long max;
-} RBIMPL_ATTR_PACKED_STRUCT_UNALIGNED_END();
+};
 
 static void
-queue_mark(void *ptr)
+queue_mark_and_move(void *ptr)
 {
     struct rb_queue *q = ptr;
-
     /* no need to mark threads in waitq, they are on stack */
-    rb_gc_mark(q->que);
+    for (long index = 0; index < q->len; index++) {
+        rb_gc_mark_and_move(&q->buffer[((q->offset + index) % q->capa)]);
+    }
+}
+
+static inline void
+queue_free_buffer(struct rb_queue *q)
+{
+    if (q->buffer) {
+        SIZED_FREE_N(q->buffer, q->capa);
+    }
+}
+
+static void
+queue_free(void *ptr)
+{
+    struct rb_queue *q = ptr;
+    queue_free_buffer(q);
+    SIZED_FREE(q);
 }
 
 static size_t
 queue_memsize(const void *ptr)
 {
-    return sizeof(struct rb_queue);
+    const struct rb_queue *q = ptr;
+    return sizeof(struct rb_queue) + (q->capa * sizeof(VALUE));
 }
 
 static const rb_data_type_t queue_data_type = {
-    "queue",
-    {queue_mark, RUBY_TYPED_DEFAULT_FREE, queue_memsize,},
-    0, 0, RUBY_TYPED_FREE_IMMEDIATELY|RUBY_TYPED_WB_PROTECTED
+    .wrap_struct_name = "Thread::Queue",
+    .function = {
+        .dmark = queue_mark_and_move,
+        .dfree = queue_free,
+        .dsize = queue_memsize,
+        .dcompact = queue_mark_and_move,
+    },
+    .flags = RUBY_TYPED_FREE_IMMEDIATELY | RUBY_TYPED_WB_PROTECTED,
 };
 
 static VALUE
@@ -722,27 +737,27 @@ queue_alloc(VALUE klass)
     struct rb_queue *q;
 
     obj = TypedData_Make_Struct(klass, struct rb_queue, &queue_data_type, q);
-    ccan_list_head_init(queue_waitq(q));
+    ccan_list_head_init(&q->waitq);
     return obj;
 }
 
-static int
+static inline bool
 queue_fork_check(struct rb_queue *q)
 {
     rb_serial_t fork_gen = GET_VM()->fork_gen;
 
-    if (q->fork_gen == fork_gen) {
-        return 0;
+    if (RB_LIKELY(q->fork_gen == fork_gen)) {
+        return false;
     }
     /* forked children can't reach into parent thread stacks */
     q->fork_gen = fork_gen;
-    ccan_list_head_init(queue_waitq(q));
+    ccan_list_head_init(&q->waitq);
     q->num_waiting = 0;
-    return 1;
+    return true;
 }
 
-static struct rb_queue *
-queue_ptr(VALUE obj)
+static inline struct rb_queue *
+raw_queue_ptr(VALUE obj)
 {
     struct rb_queue *q;
 
@@ -752,6 +767,22 @@ queue_ptr(VALUE obj)
     return q;
 }
 
+static inline void
+check_queue(VALUE obj, struct rb_queue *q)
+{
+    if (RB_UNLIKELY(q->buffer == NULL)) {
+         rb_raise(rb_eTypeError, "%+"PRIsVALUE" not initialized", obj);
+    }
+}
+
+static inline struct rb_queue *
+queue_ptr(VALUE obj)
+{
+    struct rb_queue *q = raw_queue_ptr(obj);
+    check_queue(obj, q);
+    return q;
+}
+
 #define QUEUE_CLOSED          FL_USER5
 
 static rb_hrtime_t
@@ -771,23 +802,38 @@ queue_timeout2hrtime(VALUE timeout)
 }
 
 static void
-szqueue_mark(void *ptr)
+szqueue_mark_and_move(void *ptr)
 {
     struct rb_szqueue *sq = ptr;
 
-    queue_mark(&sq->q);
+    queue_mark_and_move(&sq->q);
+}
+
+static void
+szqueue_free(void *ptr)
+{
+    struct rb_szqueue *sq = ptr;
+    queue_free_buffer(&sq->q);
+    SIZED_FREE(sq);
 }
 
 static size_t
 szqueue_memsize(const void *ptr)
 {
-    return sizeof(struct rb_szqueue);
+    const struct rb_szqueue *sq = ptr;
+    return sizeof(struct rb_szqueue) + (sq->q.capa * sizeof(VALUE));
 }
 
 static const rb_data_type_t szqueue_data_type = {
-    "sized_queue",
-    {szqueue_mark, RUBY_TYPED_DEFAULT_FREE, szqueue_memsize,},
-    0, 0, RUBY_TYPED_FREE_IMMEDIATELY|RUBY_TYPED_WB_PROTECTED
+    .wrap_struct_name = "Thread::SizedQueue",
+    .function = {
+        .dmark = szqueue_mark_and_move,
+        .dfree = szqueue_free,
+        .dsize = szqueue_memsize,
+        .dcompact = szqueue_mark_and_move,
+    },
+    .parent = &queue_data_type,
+    .flags = RUBY_TYPED_FREE_IMMEDIATELY | RUBY_TYPED_WB_PROTECTED,
 };
 
 static VALUE
@@ -801,13 +847,13 @@ szqueue_alloc(VALUE klass)
     return obj;
 }
 
-static struct rb_szqueue *
-szqueue_ptr(VALUE obj)
+static inline struct rb_szqueue *
+raw_szqueue_ptr(VALUE obj)
 {
     struct rb_szqueue *sq;
 
     TypedData_Get_Struct(obj, struct rb_szqueue, &szqueue_data_type, sq);
-    if (queue_fork_check(&sq->q)) {
+    if (RB_UNLIKELY(queue_fork_check(&sq->q))) {
         ccan_list_head_init(szqueue_pushq(sq));
         sq->num_waiting_push = 0;
     }
@@ -815,28 +861,15 @@ szqueue_ptr(VALUE obj)
     return sq;
 }
 
-static VALUE
-ary_buf_new(void)
-{
-    return rb_ary_hidden_new(1);
-}
-
-static VALUE
-check_array(VALUE obj, VALUE ary)
-{
-    if (!RB_TYPE_P(ary, T_ARRAY)) {
-        rb_raise(rb_eTypeError, "%+"PRIsVALUE" not initialized", obj);
-    }
-    return ary;
-}
-
-static long
-queue_length(VALUE self, struct rb_queue *q)
+static inline struct rb_szqueue *
+szqueue_ptr(VALUE obj)
 {
-    return RARRAY_LEN(check_array(self, q->que));
+    struct rb_szqueue *sq = raw_szqueue_ptr(obj);
+    check_queue(obj, &sq->q);
+    return sq;
 }
 
-static int
+static inline bool
 queue_closed_p(VALUE self)
 {
     return FL_TEST_RAW(self, QUEUE_CLOSED) != 0;
@@ -860,171 +893,95 @@ raise_closed_queue_error(VALUE self)
 static VALUE
 queue_closed_result(VALUE self, struct rb_queue *q)
 {
-    RUBY_ASSERT(queue_length(self, q) == 0);
+    RUBY_ASSERT(q->len == 0);
     return Qnil;
 }
 
-/*
- *  Document-class: Thread::Queue
- *
- *  The Thread::Queue class implements multi-producer, multi-consumer
- *  queues.  It is especially useful in threaded programming when
- *  information must be exchanged safely between multiple threads. The
- *  Thread::Queue class implements all the required locking semantics.
- *
- *  The class implements FIFO (first in, first out) type of queue.
- *  In a FIFO queue, the first tasks added are the first retrieved.
- *
- *  Example:
- *
- *	queue = Thread::Queue.new
- *
- *	producer = Thread.new do
- *	  5.times do |i|
- *	    sleep rand(i) # simulate expense
- *	    queue << i
- *	    puts "#{i} produced"
- *	  end
- *	end
- *
- *	consumer = Thread.new do
- *	  5.times do |i|
- *	    value = queue.pop
- *	    sleep rand(i/2) # simulate expense
- *	    puts "consumed #{value}"
- *	  end
- *	end
- *
- *	consumer.join
- *
- */
+#define QUEUE_INITIAL_CAPA 8
 
-/*
- * Document-method: Queue::new
- *
- * call-seq:
- *   Thread::Queue.new -> empty_queue
- *   Thread::Queue.new(enumerable) -> queue
- *
- * Creates a new queue instance, optionally using the contents of an +enumerable+
- * for its initial state.
- *
- * Example:
- *
- *    	q = Thread::Queue.new
- *      #=> #<Thread::Queue:0x00007ff7501110d0>
- *      q.empty?
- *      #=> true
- *
- *    	q = Thread::Queue.new([1, 2, 3])
- *    	#=> #<Thread::Queue:0x00007ff7500ec500>
- *      q.empty?
- *      #=> false
- *      q.pop
- *      #=> 1
- */
+static inline void
+ring_buffer_init(struct rb_queue *q, long initial_capa)
+{
+    q->buffer = ALLOC_N(VALUE, initial_capa);
+    q->capa = initial_capa;
+}
 
-static VALUE
-rb_queue_initialize(int argc, VALUE *argv, VALUE self)
+static inline void
+ring_buffer_expand(struct rb_queue *q)
 {
-    VALUE initial;
-    struct rb_queue *q = queue_ptr(self);
-    if ((argc = rb_scan_args(argc, argv, "01", &initial)) == 1) {
-        initial = rb_to_array(initial);
-    }
-    RB_OBJ_WRITE(self, queue_list(q), ary_buf_new());
-    ccan_list_head_init(queue_waitq(q));
-    if (argc == 1) {
-        rb_ary_concat(q->que, initial);
-    }
-    return self;
+    RUBY_ASSERT(q->capa > 0);
+    VALUE *new_buffer = ALLOC_N(VALUE, q->capa * 2);
+    MEMCPY(new_buffer, q->buffer + q->offset, VALUE, q->capa - q->offset);
+    MEMCPY(new_buffer + (q->capa - q->offset), q->buffer, VALUE, q->offset);
+    VALUE *old_buffer = q->buffer;
+    q->buffer = new_buffer;
+    q->offset = 0;
+    ruby_xfree_sized(old_buffer, q->capa * sizeof(VALUE));
+    q->capa *= 2;
 }
 
-static VALUE
-queue_do_push(VALUE self, struct rb_queue *q, VALUE obj)
+static void
+ring_buffer_push(VALUE self, struct rb_queue *q, VALUE obj)
 {
-    if (queue_closed_p(self)) {
-        raise_closed_queue_error(self);
+    if (RB_UNLIKELY(q->len >= q->capa)) {
+        ring_buffer_expand(q);
     }
-    rb_ary_push(check_array(self, q->que), obj);
-    wakeup_one(queue_waitq(q));
-    return self;
+    RUBY_ASSERT(q->capa > q->len);
+    long index = (q->offset + q->len) % q->capa;
+    q->len++;
+    RB_OBJ_WRITE(self, &q->buffer[index], obj);
 }
 
-/*
- * Document-method: Thread::Queue#close
- * call-seq:
- *   close
- *
- * Closes the queue. A closed queue cannot be re-opened.
- *
- * After the call to close completes, the following are true:
- *
- * - +closed?+ will return true
- *
- * - +close+ will be ignored.
- *
- * - calling enq/push/<< will raise a +ClosedQueueError+.
- *
- * - when +empty?+ is false, calling deq/pop/shift will return an object
- *   from the queue as usual.
- * - when +empty?+ is true, deq(false) will not suspend the thread and will return nil.
- *   deq(true) will raise a +ThreadError+.
- *
- * ClosedQueueError is inherited from StopIteration, so that you can break loop block.
- *
- * Example:
- *
- *    	q = Thread::Queue.new
- *      Thread.new{
- *        while e = q.deq # wait for nil to break loop
- *          # ...
- *        end
- *      }
- *      q.close
- */
-
 static VALUE
-rb_queue_close(VALUE self)
+ring_buffer_shift(struct rb_queue *q)
 {
-    struct rb_queue *q = queue_ptr(self);
-
-    if (!queue_closed_p(self)) {
-        FL_SET(self, QUEUE_CLOSED);
-
-        wakeup_all(queue_waitq(q));
+    if (!q->len) {
+        return Qnil;
     }
 
-    return self;
+    VALUE obj = q->buffer[q->offset];
+    q->len--;
+    if (q->len == 0) {
+        q->offset = 0;
+    }
+    else {
+        q->offset = (q->offset + 1) % q->capa;
+    }
+    return obj;
 }
 
-/*
- * Document-method: Thread::Queue#closed?
- * call-seq: closed?
- *
- * Returns +true+ if the queue is closed.
- */
-
 static VALUE
-rb_queue_closed_p(VALUE self)
+queue_initialize(rb_execution_context_t *ec, VALUE self, VALUE initial)
 {
-    return RBOOL(queue_closed_p(self));
+    struct rb_queue *q = raw_queue_ptr(self);
+    ccan_list_head_init(&q->waitq);
+    if (NIL_P(initial)) {
+        ring_buffer_init(q, QUEUE_INITIAL_CAPA);
+    }
+    else {
+        initial = rb_to_array(initial);
+        long len = RARRAY_LEN(initial);
+        long initial_capa = QUEUE_INITIAL_CAPA;
+        while (initial_capa < len) {
+            initial_capa *= 2;
+        }
+        ring_buffer_init(q, initial_capa);
+        MEMCPY(q->buffer, RARRAY_CONST_PTR(initial), VALUE, len);
+        q->len = len;
+    }
+    return self;
 }
 
-/*
- * Document-method: Thread::Queue#push
- * call-seq:
- *   push(object)
- *   enq(object)
- *   <<(object)
- *
- * Pushes the given +object+ to the queue.
- */
-
 static VALUE
-rb_queue_push(VALUE self, VALUE obj)
+queue_do_push(VALUE self, struct rb_queue *q, VALUE obj)
 {
-    return queue_do_push(self, queue_ptr(self), obj);
+    check_queue(self, q);
+    if (queue_closed_p(self)) {
+        raise_closed_queue_error(self);
+    }
+    ring_buffer_push(self, q, obj);
+    wakeup_one(&q->waitq);
+    return self;
 }
 
 static VALUE
@@ -1065,12 +1022,11 @@ szqueue_sleep_done(VALUE p)
     return Qfalse;
 }
 
-static VALUE
-queue_do_pop(VALUE self, struct rb_queue *q, int should_block, VALUE timeout)
+static inline VALUE
+queue_do_pop(rb_execution_context_t *ec, VALUE self, struct rb_queue *q, VALUE non_block, VALUE timeout)
 {
-    check_array(self, q->que);
-    if (RARRAY_LEN(q->que) == 0) {
-        if (!should_block) {
+    if (q->len == 0) {
+        if (RTEST(non_block)) {
             rb_raise(rb_eThreadError, "queue empty");
         }
 
@@ -1080,14 +1036,12 @@ queue_do_pop(VALUE self, struct rb_queue *q, int should_block, VALUE timeout)
     }
 
     rb_hrtime_t end = queue_timeout2hrtime(timeout);
-    while (RARRAY_LEN(q->que) == 0) {
+    while (q->len == 0) {
         if (queue_closed_p(self)) {
             return queue_closed_result(self, q);
         }
         else {
-            rb_execution_context_t *ec = GET_EC();
-
-            RUBY_ASSERT(RARRAY_LEN(q->que) == 0);
+            RUBY_ASSERT(q->len == 0);
             RUBY_ASSERT(queue_closed_p(self) == 0);
 
             struct queue_waiter queue_waiter = {
@@ -1095,7 +1049,7 @@ queue_do_pop(VALUE self, struct rb_queue *q, int should_block, VALUE timeout)
                 .as = {.q = q}
             };
 
-            struct ccan_list_head *waitq = queue_waitq(q);
+            struct ccan_list_head *waitq = &q->waitq;
 
             ccan_list_add_tail(waitq, &queue_waiter.w.node);
             queue_waiter.as.q->num_waiting++;
@@ -1112,116 +1066,32 @@ queue_do_pop(VALUE self, struct rb_queue *q, int should_block, VALUE timeout)
         }
     }
 
-    return rb_ary_shift(q->que);
+    return ring_buffer_shift(q);
 }
 
 static VALUE
 rb_queue_pop(rb_execution_context_t *ec, VALUE self, VALUE non_block, VALUE timeout)
 {
-    return queue_do_pop(self, queue_ptr(self), !RTEST(non_block), timeout);
-}
-
-/*
- * Document-method: Thread::Queue#empty?
- * call-seq: empty?
- *
- * Returns +true+ if the queue is empty.
- */
-
-static VALUE
-rb_queue_empty_p(VALUE self)
-{
-    return RBOOL(queue_length(self, queue_ptr(self)) == 0);
+    return queue_do_pop(ec, self, queue_ptr(self), non_block, timeout);
 }
 
-/*
- * Document-method: Thread::Queue#clear
- *
- * Removes all objects from the queue.
- */
-
-static VALUE
-rb_queue_clear(VALUE self)
-{
-    struct rb_queue *q = queue_ptr(self);
-
-    rb_ary_clear(check_array(self, q->que));
-    return self;
-}
-
-/*
- * Document-method: Thread::Queue#length
- * call-seq:
- *   length
- *   size
- *
- * Returns the length of the queue.
- */
-
-static VALUE
-rb_queue_length(VALUE self)
-{
-    return LONG2NUM(queue_length(self, queue_ptr(self)));
-}
-
-NORETURN(static VALUE rb_queue_freeze(VALUE self));
-/*
- * call-seq:
- *   freeze
- *
- * The queue can't be frozen, so this method raises an exception:
- *   Thread::Queue.new.freeze # Raises TypeError (cannot freeze #<Thread::Queue:0x...>)
- *
- */
-static VALUE
-rb_queue_freeze(VALUE self)
-{
-    rb_raise(rb_eTypeError, "cannot freeze " "%+"PRIsVALUE, self);
-    UNREACHABLE_RETURN(self);
-}
-
-/*
- * Document-method: Thread::Queue#num_waiting
- *
- * Returns the number of threads waiting on the queue.
- */
-
-static VALUE
-rb_queue_num_waiting(VALUE self)
+static void
+queue_clear(struct rb_queue *q)
 {
-    struct rb_queue *q = queue_ptr(self);
-
-    return INT2NUM(q->num_waiting);
+    q->len = 0;
+    q->offset = 0;
 }
 
-/*
- *  Document-class: Thread::SizedQueue
- *
- * This class represents queues of specified size capacity.  The push operation
- * may be blocked if the capacity is full.
- *
- * See Thread::Queue for an example of how a Thread::SizedQueue works.
- */
-
-/*
- * Document-method: SizedQueue::new
- * call-seq: new(max)
- *
- * Creates a fixed-length queue with a maximum size of +max+.
- */
-
 static VALUE
-rb_szqueue_initialize(VALUE self, VALUE vmax)
+szqueue_initialize(rb_execution_context_t *ec, VALUE self, VALUE vmax)
 {
-    long max;
-    struct rb_szqueue *sq = szqueue_ptr(self);
+    long max = NUM2LONG(vmax);
+    struct rb_szqueue *sq = raw_szqueue_ptr(self);
 
-    max = NUM2LONG(vmax);
     if (max <= 0) {
         rb_raise(rb_eArgError, "queue size must be positive");
     }
-
-    RB_OBJ_WRITE(self, szqueue_list(sq), ary_buf_new());
+    ring_buffer_init(&sq->q, QUEUE_INITIAL_CAPA);
     ccan_list_head_init(szqueue_waitq(sq));
     ccan_list_head_init(szqueue_pushq(sq));
     sq->max = max;
@@ -1229,74 +1099,12 @@ rb_szqueue_initialize(VALUE self, VALUE vmax)
     return self;
 }
 
-/*
- * Document-method: Thread::SizedQueue#close
- * call-seq:
- *   close
- *
- * Similar to Thread::Queue#close.
- *
- * The difference is behavior with waiting enqueuing threads.
- *
- * If there are waiting enqueuing threads, they are interrupted by
- * raising ClosedQueueError('queue closed').
- */
-static VALUE
-rb_szqueue_close(VALUE self)
-{
-    if (!queue_closed_p(self)) {
-        struct rb_szqueue *sq = szqueue_ptr(self);
-
-        FL_SET(self, QUEUE_CLOSED);
-        wakeup_all(szqueue_waitq(sq));
-        wakeup_all(szqueue_pushq(sq));
-    }
-    return self;
-}
-
-/*
- * Document-method: Thread::SizedQueue#max
- *
- * Returns the maximum size of the queue.
- */
-
-static VALUE
-rb_szqueue_max_get(VALUE self)
-{
-    return LONG2NUM(szqueue_ptr(self)->max);
-}
-
-/*
- * Document-method: Thread::SizedQueue#max=
- * call-seq: max=(number)
- *
- * Sets the maximum size of the queue to the given +number+.
- */
-
-static VALUE
-rb_szqueue_max_set(VALUE self, VALUE vmax)
-{
-    long max = NUM2LONG(vmax);
-    long diff = 0;
-    struct rb_szqueue *sq = szqueue_ptr(self);
-
-    if (max <= 0) {
-        rb_raise(rb_eArgError, "queue size must be positive");
-    }
-    if (max > sq->max) {
-        diff = max - sq->max;
-    }
-    sq->max = max;
-    sync_wakeup(szqueue_pushq(sq), diff);
-    return vmax;
-}
-
 static VALUE
 rb_szqueue_push(rb_execution_context_t *ec, VALUE self, VALUE object, VALUE non_block, VALUE timeout)
 {
     struct rb_szqueue *sq = szqueue_ptr(self);
 
-    if (queue_length(self, &sq->q) >= sq->max) {
+    if (sq->q.len >= sq->max) {
         if (RTEST(non_block)) {
             rb_raise(rb_eThreadError, "queue full");
         }
@@ -1307,12 +1115,11 @@ rb_szqueue_push(rb_execution_context_t *ec, VALUE self, VALUE object, VALUE non_
     }
 
     rb_hrtime_t end = queue_timeout2hrtime(timeout);
-    while (queue_length(self, &sq->q) >= sq->max) {
+    while (sq->q.len >= sq->max) {
         if (queue_closed_p(self)) {
             raise_closed_queue_error(self);
         }
         else {
-            rb_execution_context_t *ec = GET_EC();
             struct queue_waiter queue_waiter = {
                 .w = {.self = self, .th = ec->thread_ptr, .fiber = nonblocking_fiber(ec->fiber_ptr)},
                 .as = {.sq = sq}
@@ -1339,85 +1146,17 @@ rb_szqueue_push(rb_execution_context_t *ec, VALUE self, VALUE object, VALUE non_
 }
 
 static VALUE
-szqueue_do_pop(VALUE self, int should_block, VALUE timeout)
+rb_szqueue_pop(rb_execution_context_t *ec, VALUE self, VALUE non_block, VALUE timeout)
 {
     struct rb_szqueue *sq = szqueue_ptr(self);
-    VALUE retval = queue_do_pop(self, &sq->q, should_block, timeout);
+    VALUE retval = queue_do_pop(ec, self, &sq->q, non_block, timeout);
 
-    if (queue_length(self, &sq->q) < sq->max) {
+    if (sq->q.len < sq->max) {
         wakeup_one(szqueue_pushq(sq));
     }
 
     return retval;
 }
-static VALUE
-rb_szqueue_pop(rb_execution_context_t *ec, VALUE self, VALUE non_block, VALUE timeout)
-{
-    return szqueue_do_pop(self, !RTEST(non_block), timeout);
-}
-
-/*
- * Document-method: Thread::SizedQueue#clear
- *
- * Removes all objects from the queue.
- */
-
-static VALUE
-rb_szqueue_clear(VALUE self)
-{
-    struct rb_szqueue *sq = szqueue_ptr(self);
-
-    rb_ary_clear(check_array(self, sq->q.que));
-    wakeup_all(szqueue_pushq(sq));
-    return self;
-}
-
-/*
- * Document-method: Thread::SizedQueue#length
- * call-seq:
- *   length
- *   size
- *
- * Returns the length of the queue.
- */
-
-static VALUE
-rb_szqueue_length(VALUE self)
-{
-    struct rb_szqueue *sq = szqueue_ptr(self);
-
-    return LONG2NUM(queue_length(self, &sq->q));
-}
-
-/*
- * Document-method: Thread::SizedQueue#num_waiting
- *
- * Returns the number of threads waiting on the queue.
- */
-
-static VALUE
-rb_szqueue_num_waiting(VALUE self)
-{
-    struct rb_szqueue *sq = szqueue_ptr(self);
-
-    return INT2NUM(sq->q.num_waiting + sq->num_waiting_push);
-}
-
-/*
- * Document-method: Thread::SizedQueue#empty?
- * call-seq: empty?
- *
- * Returns +true+ if the queue is empty.
- */
-
-static VALUE
-rb_szqueue_empty_p(VALUE self)
-{
-    struct rb_szqueue *sq = szqueue_ptr(self);
-
-    return RBOOL(queue_length(self, &sq->q) == 0);
-}
-
 
 /* ConditionalVariable */
 struct rb_condvar {
@@ -1425,88 +1164,6 @@ struct rb_condvar {
     rb_serial_t fork_gen;
 };
 
-/*
- *  Document-class: Thread::ConditionVariable
- *
- *  ConditionVariable objects augment class Mutex. Using condition variables,
- *  it is possible to suspend while in the middle of a critical section until a
- *  condition is met, such as a resource becomes available.
- *
- *  Due to non-deterministic scheduling and spurious wake-ups, users of
- *  condition variables should always use a separate boolean predicate (such as
- *  reading from a boolean variable) to check if the condition is actually met
- *  before starting to wait, and should wait in a loop, re-checking the
- *  condition every time the ConditionVariable is waken up.  The idiomatic way
- *  of using condition variables is calling the +wait+ method in an +until+
- *  loop with the predicate as the loop condition.
- *
- *    condvar.wait(mutex) until condition_is_met
- *
- *  In the example below, we use the boolean variable +resource_available+
- *  (which is protected by +mutex+) to indicate the availability of the
- *  resource, and use +condvar+ to wait for that variable to become true.  Note
- *  that:
- *
- *  1.  Thread +b+ may be scheduled before thread +a1+ and +a2+, and may run so
- *      fast that it have already made the resource available before either
- *      +a1+ or +a2+ starts. Therefore, +a1+ and +a2+ should check if
- *      +resource_available+ is already true before starting to wait.
- *  2.  The +wait+ method may spuriously wake up without signalling. Therefore,
- *      thread +a1+ and +a2+ should recheck +resource_available+ after the
- *      +wait+ method returns, and go back to wait if the condition is not
- *      actually met.
- *  3.  It is possible that thread +a2+ starts right after thread +a1+ is waken
- *      up by +b+.  Thread +a2+ may have acquired the +mutex+ and consumed the
- *      resource before thread +a1+ acquires the +mutex+.  This necessitates
- *      rechecking after +wait+, too.
- *
- *  Example:
- *
- *    mutex = Thread::Mutex.new
- *
- *    resource_available = false
- *    condvar = Thread::ConditionVariable.new
- *
- *    a1 = Thread.new {
- *      # Thread 'a1' waits for the resource to become available and consumes
- *      # the resource.
- *      mutex.synchronize {
- *        condvar.wait(mutex) until resource_available
- *        # After the loop, 'resource_available' is guaranteed to be true.
- *
- *        resource_available = false
- *        puts "a1 consumed the resource"
- *      }
- *    }
- *
- *    a2 = Thread.new {
- *      # Thread 'a2' behaves like 'a1'.
- *      mutex.synchronize {
- *        condvar.wait(mutex) until resource_available
- *        resource_available = false
- *        puts "a2 consumed the resource"
- *      }
- *    }
- *
- *    b = Thread.new {
- *      # Thread 'b' periodically makes the resource available.
- *      loop {
- *        mutex.synchronize {
- *          resource_available = true
- *
- *          # Notify one waiting thread if any.  It is possible that neither
- *          # 'a1' nor 'a2 is waiting on 'condvar' at this moment.  That's OK.
- *          condvar.signal
- *        }
- *        sleep 1
- *      }
- *    }
- *
- *    # Eventually both 'a1' and 'a2' will have their resources, albeit in an
- *    # unspecified order.
- *    [a1, a2].each {|th| th.join}
- */
-
 static size_t
 condvar_memsize(const void *ptr)
 {
@@ -1548,21 +1205,8 @@ condvar_alloc(VALUE klass)
     return obj;
 }
 
-/*
- * Document-method: ConditionVariable::new
- *
- * Creates a new condition variable instance.
- */
-
-static VALUE
-rb_condvar_initialize(VALUE self)
-{
-    struct rb_condvar *cv = condvar_ptr(self);
-    ccan_list_head_init(&cv->waitq);
-    return self;
-}
-
 struct sleep_call {
+    rb_execution_context_t *ec;
     VALUE mutex;
     VALUE timeout;
 };
@@ -1573,162 +1217,334 @@ static VALUE
 do_sleep(VALUE args)
 {
     struct sleep_call *p = (struct sleep_call *)args;
-    return rb_funcallv(p->mutex, id_sleep, 1, &p->timeout);
+    if (CLASS_OF(p->mutex) == rb_cMutex) {
+        return rb_mut_sleep(p->ec, p->mutex, p->timeout);
+    }
+    else {
+        return rb_funcallv(p->mutex, id_sleep, 1, &p->timeout);
+    }
 }
 
-/*
- * Document-method: Thread::ConditionVariable#wait
- * call-seq: wait(mutex, timeout=nil)
- *
- * Releases the lock held in +mutex+ and waits; reacquires the lock on wakeup.
- *
- * If +timeout+ is given, this method returns after +timeout+ seconds passed,
- * even if no other thread doesn't signal.
- *
- * This method may wake up spuriously due to underlying implementation details.
- *
- * Returns the slept result on +mutex+.
- */
-
 static VALUE
-rb_condvar_wait(int argc, VALUE *argv, VALUE self)
+rb_condvar_wait(rb_execution_context_t *ec, VALUE self, VALUE mutex, VALUE timeout)
 {
-    rb_execution_context_t *ec = GET_EC();
-
     struct rb_condvar *cv = condvar_ptr(self);
-    struct sleep_call args;
-
-    rb_scan_args(argc, argv, "11", &args.mutex, &args.timeout);
+    struct sleep_call args = {
+        .ec = ec,
+        .mutex = mutex,
+        .timeout = timeout,
+    };
 
     struct sync_waiter sync_waiter = {
-        .self = args.mutex,
+        .self = mutex,
         .th = ec->thread_ptr,
         .fiber = nonblocking_fiber(ec->fiber_ptr)
     };
 
     ccan_list_add_tail(&cv->waitq, &sync_waiter.node);
-    return rb_ensure(do_sleep, (VALUE)&args, delete_from_waitq, (VALUE)&sync_waiter);
+    return rb_ec_ensure(ec, do_sleep, (VALUE)&args, delete_from_waitq, (VALUE)&sync_waiter);
 }
 
-/*
- * Document-method: Thread::ConditionVariable#signal
- *
- * Wakes up the first thread in line waiting for this lock.
- */
-
 static VALUE
-rb_condvar_signal(VALUE self)
+rb_condvar_signal(rb_execution_context_t *ec, VALUE self)
 {
     struct rb_condvar *cv = condvar_ptr(self);
     wakeup_one(&cv->waitq);
     return self;
 }
 
-/*
- * Document-method: Thread::ConditionVariable#broadcast
- *
- * Wakes up all threads waiting for this lock.
- */
-
 static VALUE
-rb_condvar_broadcast(VALUE self)
+rb_condvar_broadcast(rb_execution_context_t *ec, VALUE self)
 {
     struct rb_condvar *cv = condvar_ptr(self);
     wakeup_all(&cv->waitq);
     return self;
 }
 
-NORETURN(static VALUE undumpable(VALUE obj));
-/* :nodoc: */
+/* Thread::Monitor */
+
+struct rb_monitor {
+    long count;
+    rb_serial_t ec_serial;
+    VALUE mutex;
+};
+
+static void
+monitor_mark(void *ptr)
+{
+    struct rb_monitor *mc = ptr;
+    rb_gc_mark_movable(mc->mutex);
+}
+
+static void
+monitor_compact(void *ptr)
+{
+    struct rb_monitor *mc = ptr;
+    mc->mutex = rb_gc_location(mc->mutex);
+}
+
+static const rb_data_type_t monitor_data_type = {
+    .wrap_struct_name = "monitor",
+    .function = {
+        .dmark = monitor_mark,
+        .dfree = RUBY_TYPED_DEFAULT_FREE,
+        .dsize = NULL, // Fully embeded
+        .dcompact = monitor_compact,
+    },
+    .flags = RUBY_TYPED_FREE_IMMEDIATELY | RUBY_TYPED_WB_PROTECTED | RUBY_TYPED_EMBEDDABLE,
+};
+
 static VALUE
-undumpable(VALUE obj)
+monitor_alloc(VALUE klass)
+{
+    struct rb_monitor *mc;
+    VALUE obj;
+
+    obj = TypedData_Make_Struct(klass, struct rb_monitor, &monitor_data_type, mc);
+    RB_OBJ_WRITE(obj, &mc->mutex, rb_mutex_new());
+    mc->ec_serial = 0;
+    mc->count = 0;
+
+    return obj;
+}
+
+static struct rb_monitor *
+monitor_ptr(VALUE monitor)
 {
-    rb_raise(rb_eTypeError, "can't dump %"PRIsVALUE, rb_obj_class(obj));
-    UNREACHABLE_RETURN(Qnil);
+    struct rb_monitor *mc;
+    TypedData_Get_Struct(monitor, struct rb_monitor, &monitor_data_type, mc);
+    return mc;
+}
+
+static bool
+mc_owner_p(struct rb_monitor *mc, rb_serial_t current_fiber_serial)
+{
+    return mc->ec_serial == current_fiber_serial;
 }
 
 static VALUE
-define_thread_class(VALUE outer, const ID name, VALUE super)
+rb_monitor_try_enter(rb_execution_context_t *ec, VALUE monitor)
 {
-    VALUE klass = rb_define_class_id_under(outer, name, super);
-    rb_const_set(rb_cObject, name, klass);
-    return klass;
+    struct rb_monitor *mc = monitor_ptr(monitor);
+
+    rb_serial_t current_fiber_serial = rb_ec_serial(ec);
+    if (!mc_owner_p(mc, current_fiber_serial)) {
+        if (!rb_mut_trylock(ec, mc->mutex)) {
+            return Qfalse;
+        }
+        mc->ec_serial = current_fiber_serial;
+        mc->count = 0;
+    }
+    mc->count += 1;
+    return Qtrue;
 }
 
+struct monitor_args {
+    VALUE monitor;
+    struct rb_monitor *mc;
+    rb_serial_t current_fiber_serial;
+    rb_execution_context_t *ec;
+};
+
 static void
-Init_thread_sync(void)
+monitor_enter0(struct monitor_args *args)
+{
+    if (!mc_owner_p(args->mc, args->current_fiber_serial)) {
+        struct mutex_args mut_args = {
+            .self = args->mc->mutex,
+            .mutex = mutex_ptr(args->mc->mutex),
+            .ec= args->ec,
+        };
+        do_mutex_lock(&mut_args, 1);
+        args->mc->ec_serial = args->current_fiber_serial;
+        args->mc->count = 0;
+    }
+    args->mc->count++;
+}
+
+static VALUE
+rb_monitor_enter(rb_execution_context_t *ec, VALUE monitor)
 {
-#undef rb_intern
-#if defined(TEACH_RDOC) && TEACH_RDOC == 42
-    rb_cMutex = rb_define_class_under(rb_cThread, "Mutex", rb_cObject);
-    rb_cConditionVariable = rb_define_class_under(rb_cThread, "ConditionVariable", rb_cObject);
-    rb_cQueue = rb_define_class_under(rb_cThread, "Queue", rb_cObject);
-    rb_cSizedQueue = rb_define_class_under(rb_cThread, "SizedQueue", rb_cObject);
-#endif
+    struct monitor_args args = {
+        .monitor = monitor,
+        .mc = monitor_ptr(monitor),
+        .ec = ec,
+        .current_fiber_serial = rb_ec_serial(ec),
+    };
+    monitor_enter0(&args);
+    return Qnil;
+}
 
-#define DEFINE_CLASS(name, super) \
-    rb_c##name = define_thread_class(rb_cThread, rb_intern(#name), rb_c##super)
+static inline void
+monitor_check_owner0(struct monitor_args *args)
+{
+    if (!mc_owner_p(args->mc, args->current_fiber_serial)) {
+        rb_raise(rb_eThreadError, "current fiber not owner");
+    }
+}
+
+static VALUE
+rb_monitor_check_owner(rb_execution_context_t *ec, VALUE monitor)
+{
+    struct monitor_args args = {
+        .monitor = monitor,
+        .mc = monitor_ptr(monitor),
+        .ec = ec,
+        .current_fiber_serial = rb_ec_serial(ec),
+    };
+    monitor_check_owner0(&args);
+    return Qnil;
+}
+
+static void
+monitor_exit0(struct monitor_args *args)
+{
+    monitor_check_owner0(args);
+
+    if (args->mc->count <= 0) rb_bug("monitor_exit: count:%d", (int)args->mc->count);
+    args->mc->count--;
+
+    if (args->mc->count == 0) {
+        args->mc->ec_serial = 0;
+
+        struct mutex_args mut_args = {
+            .self = args->mc->mutex,
+            .mutex = mutex_ptr(args->mc->mutex),
+            .ec= args->ec,
+        };
+        do_mutex_unlock(&mut_args);
+    }
+}
 
+static VALUE
+rb_monitor_exit(rb_execution_context_t *ec, VALUE monitor)
+{
+    struct monitor_args args = {
+        .monitor = monitor,
+        .mc = monitor_ptr(monitor),
+        .ec = ec,
+        .current_fiber_serial = rb_ec_serial(ec),
+    };
+    monitor_exit0(&args);
+    return Qnil;
+}
+
+static VALUE
+rb_monitor_locked_p(rb_execution_context_t *ec, VALUE monitor)
+{
+    struct rb_monitor *mc = monitor_ptr(monitor);
+    return rb_mutex_locked_p(mc->mutex);
+}
+
+static VALUE
+rb_monitor_owned_p(rb_execution_context_t *ec, VALUE monitor)
+{
+    struct rb_monitor *mc = monitor_ptr(monitor);
+    return RBOOL(rb_mutex_locked_p(mc->mutex) && mc_owner_p(mc, rb_ec_serial(ec)));
+}
+
+static VALUE
+monitor_exit_for_cond(VALUE monitor)
+{
+    struct rb_monitor *mc = monitor_ptr(monitor);
+    long cnt = mc->count;
+    mc->ec_serial = 0;
+    mc->count = 0;
+    return LONG2NUM(cnt);
+}
+
+struct wait_for_cond_data {
+    VALUE monitor;
+    VALUE cond;
+    VALUE timeout;
+    VALUE count;
+};
+
+static VALUE
+monitor_wait_for_cond_body(VALUE v)
+{
+    struct wait_for_cond_data *data = (struct wait_for_cond_data *)v;
+    struct rb_monitor *mc = monitor_ptr(data->monitor);
+    // cond.wait(monitor.mutex, timeout)
+    VALUE signaled = rb_funcall(data->cond, rb_intern("wait"), 2, mc->mutex, data->timeout);
+    return RTEST(signaled) ? Qtrue : Qfalse;
+}
+
+static VALUE
+monitor_enter_for_cond(VALUE v)
+{
+    // assert(rb_mutex_owned_p(mc->mutex) == Qtrue)
+    // but rb_mutex_owned_p is not exported...
+
+    struct wait_for_cond_data *data = (struct wait_for_cond_data *)v;
+    struct rb_monitor *mc = monitor_ptr(data->monitor);
+    mc->ec_serial = rb_ec_serial(GET_EC());
+    mc->count = NUM2LONG(data->count);
+    return Qnil;
+}
+
+static VALUE
+rb_monitor_wait_for_cond(rb_execution_context_t *ec, VALUE monitor, VALUE cond, VALUE timeout)
+{
+    VALUE count = monitor_exit_for_cond(monitor);
+    struct wait_for_cond_data data = {
+        monitor,
+        cond,
+        timeout,
+        count,
+    };
+
+    return rb_ensure(monitor_wait_for_cond_body, (VALUE)&data,
+                     monitor_enter_for_cond, (VALUE)&data);
+}
+
+static VALUE
+monitor_sync_ensure(VALUE v_args)
+{
+    monitor_exit0((struct monitor_args *)v_args);
+    return Qnil;
+}
+
+static VALUE
+rb_monitor_synchronize(rb_execution_context_t *ec, VALUE monitor)
+{
+    struct monitor_args args = {
+        .monitor = monitor,
+        .mc = monitor_ptr(monitor),
+        .ec = ec,
+        .current_fiber_serial = rb_ec_serial(ec),
+    };
+    monitor_enter0(&args);
+    return rb_ec_ensure(ec, do_ec_yield, (VALUE)ec, monitor_sync_ensure, (VALUE)&args);
+}
+
+static void
+Init_thread_sync(void)
+{
     /* Mutex */
-    DEFINE_CLASS(Mutex, Object);
+    rb_cMutex = rb_define_class_id_under(rb_cThread, rb_intern("Mutex"), rb_cObject);
     rb_define_alloc_func(rb_cMutex, mutex_alloc);
-    rb_define_method(rb_cMutex, "initialize", mutex_initialize, 0);
-    rb_define_method(rb_cMutex, "locked?", rb_mutex_locked_p, 0);
-    rb_define_method(rb_cMutex, "try_lock", rb_mutex_trylock, 0);
-    rb_define_method(rb_cMutex, "lock", rb_mutex_lock, 0);
-    rb_define_method(rb_cMutex, "unlock", rb_mutex_unlock, 0);
-    rb_define_method(rb_cMutex, "sleep", mutex_sleep, -1);
-    rb_define_method(rb_cMutex, "synchronize", rb_mutex_synchronize_m, 0);
-    rb_define_method(rb_cMutex, "owned?", rb_mutex_owned_p, 0);
 
     /* Queue */
-    DEFINE_CLASS(Queue, Object);
+    VALUE rb_cQueue = rb_define_class_id_under_no_pin(rb_cThread, rb_intern("Queue"), rb_cObject);
     rb_define_alloc_func(rb_cQueue, queue_alloc);
 
     rb_eClosedQueueError = rb_define_class("ClosedQueueError", rb_eStopIteration);
 
-    rb_define_method(rb_cQueue, "initialize", rb_queue_initialize, -1);
-    rb_undef_method(rb_cQueue, "initialize_copy");
-    rb_define_method(rb_cQueue, "marshal_dump", undumpable, 0);
-    rb_define_method(rb_cQueue, "close", rb_queue_close, 0);
-    rb_define_method(rb_cQueue, "closed?", rb_queue_closed_p, 0);
-    rb_define_method(rb_cQueue, "push", rb_queue_push, 1);
-    rb_define_method(rb_cQueue, "empty?", rb_queue_empty_p, 0);
-    rb_define_method(rb_cQueue, "clear", rb_queue_clear, 0);
-    rb_define_method(rb_cQueue, "length", rb_queue_length, 0);
-    rb_define_method(rb_cQueue, "num_waiting", rb_queue_num_waiting, 0);
-    rb_define_method(rb_cQueue, "freeze", rb_queue_freeze, 0);
-
-    rb_define_alias(rb_cQueue, "enq", "push");
-    rb_define_alias(rb_cQueue, "<<", "push");
-    rb_define_alias(rb_cQueue, "size", "length");
-
-    DEFINE_CLASS(SizedQueue, Queue);
+    VALUE rb_cSizedQueue = rb_define_class_id_under_no_pin(rb_cThread, rb_intern("SizedQueue"), rb_cQueue);
     rb_define_alloc_func(rb_cSizedQueue, szqueue_alloc);
 
-    rb_define_method(rb_cSizedQueue, "initialize", rb_szqueue_initialize, 1);
-    rb_define_method(rb_cSizedQueue, "close", rb_szqueue_close, 0);
-    rb_define_method(rb_cSizedQueue, "max", rb_szqueue_max_get, 0);
-    rb_define_method(rb_cSizedQueue, "max=", rb_szqueue_max_set, 1);
-    rb_define_method(rb_cSizedQueue, "empty?", rb_szqueue_empty_p, 0);
-    rb_define_method(rb_cSizedQueue, "clear", rb_szqueue_clear, 0);
-    rb_define_method(rb_cSizedQueue, "length", rb_szqueue_length, 0);
-    rb_define_method(rb_cSizedQueue, "num_waiting", rb_szqueue_num_waiting, 0);
-    rb_define_alias(rb_cSizedQueue, "size", "length");
-
     /* CVar */
-    DEFINE_CLASS(ConditionVariable, Object);
+    VALUE rb_cConditionVariable = rb_define_class_id_under_no_pin(rb_cThread, rb_intern("ConditionVariable"), rb_cObject);
     rb_define_alloc_func(rb_cConditionVariable, condvar_alloc);
 
     id_sleep = rb_intern("sleep");
 
-    rb_define_method(rb_cConditionVariable, "initialize", rb_condvar_initialize, 0);
-    rb_undef_method(rb_cConditionVariable, "initialize_copy");
-    rb_define_method(rb_cConditionVariable, "marshal_dump", undumpable, 0);
-    rb_define_method(rb_cConditionVariable, "wait", rb_condvar_wait, -1);
-    rb_define_method(rb_cConditionVariable, "signal", rb_condvar_signal, 0);
-    rb_define_method(rb_cConditionVariable, "broadcast", rb_condvar_broadcast, 0);
+    /* Monitor */
+    VALUE rb_cMonitor = rb_define_class_id_under_no_pin(rb_cThread, rb_intern("Monitor"), rb_cObject);
+    rb_define_alloc_func(rb_cMonitor, monitor_alloc);
 
+    rb_provide("monitor.so");
     rb_provide("thread.rb");
 }