Avoid missed wakeup with fiber scheduler and Fiber.blocking. (#6588)

* Ensure that blocked fibers don't prevent valid wakeups.

1 files changed, 17 insertions, 8 deletions

diff --git a/thread_sync.c b/thread_sync.c
index 3888534468..2bcf59137e 100644
--- a/thread_sync.c
+++ b/thread_sync.c
@@ -20,6 +20,16 @@ struct sync_waiter {
     struct ccan_list_node node;
 };
 
+static inline rb_fiber_t*
+nonblocking_fiber(rb_fiber_t *fiber)
+{
+    if (rb_fiberptr_blocking(fiber)) {
+        return NULL;
+    }
+
+    return fiber;
+}
+
 struct queue_sleep_arg {
     VALUE self;
     VALUE timeout;
@@ -37,8 +47,7 @@ sync_wakeup(struct ccan_list_head *head, long max)
         ccan_list_del_init(&cur->node);
 
         if (cur->th->status != THREAD_KILLED) {
-
-            if (cur->th->scheduler != Qnil && rb_fiberptr_blocking(cur->fiber) == 0) {
+            if (cur->th->scheduler != Qnil && cur->fiber) {
                 rb_fiber_scheduler_unblock(cur->th->scheduler, cur->self, rb_fiberptr_self(cur->fiber));
             }
             else {
@@ -306,7 +315,7 @@ do_mutex_lock(VALUE self, int interruptible_p)
                 struct sync_waiter sync_waiter = {
                     .self = self,
                     .th = th,
-                    .fiber = fiber
+                    .fiber = nonblocking_fiber(fiber)
                 };
 
                 ccan_list_add_tail(&mutex->waitq, &sync_waiter.node);
@@ -339,7 +348,7 @@ do_mutex_lock(VALUE self, int interruptible_p)
                 struct sync_waiter sync_waiter = {
                     .self = self,
                     .th = th,
-                    .fiber = fiber
+                    .fiber = nonblocking_fiber(fiber)
                 };
 
                 ccan_list_add_tail(&mutex->waitq, &sync_waiter.node);
@@ -437,7 +446,7 @@ rb_mutex_unlock_th(rb_mutex_t *mutex, rb_thread_t *th, rb_fiber_t *fiber)
         ccan_list_for_each_safe(&mutex->waitq, cur, next, node) {
             ccan_list_del_init(&cur->node);
 
-            if (cur->th->scheduler != Qnil && rb_fiberptr_blocking(cur->fiber) == 0) {
+            if (cur->th->scheduler != Qnil && cur->fiber) {
                 rb_fiber_scheduler_unblock(cur->th->scheduler, cur->self, rb_fiberptr_self(cur->fiber));
                 goto found;
             }
@@ -1051,7 +1060,7 @@ queue_do_pop(VALUE self, struct rb_queue *q, int should_block, VALUE timeout)
             assert(queue_closed_p(self) == 0);
 
             struct queue_waiter queue_waiter = {
-                .w = {.self = self, .th = ec->thread_ptr, .fiber = ec->fiber_ptr},
+                .w = {.self = self, .th = ec->thread_ptr, .fiber = nonblocking_fiber(ec->fiber_ptr)},
                 .as = {.q = q}
             };
 
@@ -1258,7 +1267,7 @@ rb_szqueue_push(rb_execution_context_t *ec, VALUE self, VALUE object, VALUE non_
         else {
             rb_execution_context_t *ec = GET_EC();
             struct queue_waiter queue_waiter = {
-                .w = {.self = self, .th = ec->thread_ptr, .fiber = ec->fiber_ptr},
+                .w = {.self = self, .th = ec->thread_ptr, .fiber = nonblocking_fiber(ec->fiber_ptr)},
                 .as = {.sq = sq}
             };
 
@@ -1491,7 +1500,7 @@ rb_condvar_wait(int argc, VALUE *argv, VALUE self)
     struct sync_waiter sync_waiter = {
         .self = args.mutex,
         .th = ec->thread_ptr,
-        .fiber = ec->fiber_ptr
+        .fiber = nonblocking_fiber(ec->fiber_ptr)
     };
 
     ccan_list_add_tail(&cv->waitq, &sync_waiter.node);