Improve handling of urgent notification pipe.

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

test/fiber/scheduler.rb

@@ -15,11 +15,9 @@ class Scheduler
     @writable = {}
     @waiting = {}
 
-    @urgent = nil
-
     @lock = Mutex.new
     @locking = 0
-    @ready = []
+    @ready = Array.new
   end
 
   attr :readable
@@ -51,11 +49,17 @@ class Scheduler
       # puts "writable: #{writable}" if writable&.any?
 
       readable&.each do |io|
-        @readable[io]&.resume
+        if fiber = @readable.delete(io)
+          fiber.resume
+        elsif io == @urgent.first
+          @urgent.first.read_nonblock(1024)
+        end
       end
 
       writable&.each do |io|
-        @writable[io]&.resume
+        if fiber = @writable.delete(io)
+          fiber.resume
+        end
       end
 
       if @waiting.any?
@@ -73,9 +77,6 @@ class Scheduler
       end
 
       if @ready.any?
-        # Clear out the urgent notification pipe.
-        @urgent.first.read_nonblock(1024)
-
         ready = nil
 
         @lock.synchronize do
@@ -114,9 +115,6 @@ class Scheduler
 
     Fiber.yield
 
-    @readable.delete(io)
-    @writable.delete(io)
-
     return true
   end
 
@@ -130,10 +128,10 @@ class Scheduler
   def mutex_unlock(mutex, fiber)
     @lock.synchronize do
       @ready << fiber
-
-      if @urgent
-        @urgent.last.write('.')
     end
+
+    if io = @urgent&.last
+      @urgent.last.write_nonblock('.')
     end
   end
 

thread_sync.c

@@ -268,8 +268,8 @@ do_mutex_lock(VALUE self, int interruptible_p)
             rb_raise(rb_eThreadError, "deadlock; recursive locking");
         }
 
-        VALUE scheduler = rb_thread_current_scheduler();
         while (mutex->fiber != fiber) {
+            VALUE scheduler = rb_thread_current_scheduler();
             if (scheduler != Qnil) {
                 list_add_tail(&mutex->waitq, &w.node);
 
@@ -279,13 +279,8 @@ do_mutex_lock(VALUE self, int interruptible_p)
 
                 if (!mutex->fiber) {
                     mutex->fiber = fiber;
-                    break;
+                }
             } else {
-                    // Try again...
-                    continue;
-                }
-            }
-            
                 enum rb_thread_status prev_status = th->status;
                 rb_hrtime_t *timeout = 0;
                 rb_hrtime_t rel = rb_msec2hrtime(100);
@@ -325,6 +320,7 @@ do_mutex_lock(VALUE self, int interruptible_p)
                     th->status = prev_status;
                 }
                 rb_ractor_sleeper_threads_dec(th->ractor);
+            }
 
             if (interruptible_p) {
                 /* release mutex before checking for interrupts...as interrupt checking