Add support for Queue & SizedQueue.

Merged: https://github.com/ruby/ruby/pull/3434

test/fiber/test_mutex.rb

@@ -84,6 +84,37 @@ class TestFiberMutex < Test::Unit::TestCase
     assert signalled > 1
   end
 
+  def test_queue
+    queue = Queue.new
+    processed = 0
+
+    thread = Thread.new do
+      scheduler = Scheduler.new
+      Thread.current.scheduler = scheduler
+
+      Fiber.schedule do
+        3.times do |i|
+          queue << i
+          sleep 0.1
+        end
+
+        queue.close
+      end
+
+      Fiber.schedule do
+        while item = queue.pop
+          processed += 1
+        end
+      end
+
+      scheduler.run
+    end
+
+    thread.join
+
+    assert processed == 3
+  end
+
   def test_mutex_deadlock
     err = /No live threads left. Deadlock\?/
     assert_in_out_err %W[-I#{__dir__} -], <<-RUBY, ['in synchronize'], err, success: false

thread.c

@@ -1481,9 +1481,14 @@ rb_thread_sleep_interruptible(void)
 static void
 rb_thread_sleep_deadly_allow_spurious_wakeup(void)
 {
+    VALUE scheduler = rb_thread_current_scheduler();
+    if (scheduler != Qnil) {
+        rb_scheduler_kernel_sleepv(scheduler, 0, NULL);
+    } else {
         thread_debug("rb_thread_sleep_deadly_allow_spurious_wakeup\n");
         sleep_forever(GET_THREAD(), SLEEP_DEADLOCKABLE);
     }
+}
 
 void
 rb_thread_wait_for(struct timeval time)

thread_sync.c

@@ -953,12 +953,16 @@ queue_do_pop(VALUE self, struct rb_queue *q, int should_block)
             return queue_closed_result(self, q);
         }
         else {
+            rb_execution_context_t *ec = GET_EC();
             struct queue_waiter qw;
 
             assert(RARRAY_LEN(q->que) == 0);
             assert(queue_closed_p(self) == 0);
 
-	    qw.w.th = GET_THREAD();
+            qw.w.self = self;
+            qw.w.th = ec->thread_ptr;
+            qw.w.fiber = ec->fiber_ptr;
+
             qw.as.q = q;
             list_add_tail(queue_waitq(qw.as.q), &qw.w.node);
             qw.as.q->num_waiting++;
@@ -1195,10 +1199,14 @@ rb_szqueue_push(int argc, VALUE *argv, VALUE self)
             break;
         }
         else {
+            rb_execution_context_t *ec = GET_EC();
             struct queue_waiter qw;
             struct list_head *pushq = szqueue_pushq(sq);
 
-	    qw.w.th = GET_THREAD();
+            qw.w.self = self;
+            qw.w.th = ec->thread_ptr;
+            qw.w.fiber = ec->fiber_ptr;
+
             qw.as.sq = sq;
             list_add_tail(pushq, &qw.w.node);
             sq->num_waiting_push++;