Make Mutex per-Fiber instead of per-Thread

* Enables Mutex to be used as synchronization between multiple Fibers
  of the same Thread.
* With a Fiber scheduler we can yield to another Fiber on contended
  Mutex#lock instead of blocking the entire thread.
* This also makes the behavior of Mutex consistent across CRuby, JRuby and TruffleRuby.
* [Feature #16792]

2 files changed, 77 insertions, 7 deletions

diff --git a/test/fiber/scheduler.rb b/test/fiber/scheduler.rb
index 1f690b4c08..fa05daf886 100644
--- a/test/fiber/scheduler.rb
+++ b/test/fiber/scheduler.rb
@@ -14,6 +14,12 @@ class Scheduler
     @readable = {}
     @writable = {}
     @waiting = {}
+
+    @urgent = nil
+
+    @lock = Mutex.new
+    @locking = 0
+    @ready = []
   end
 
   attr :readable
@@ -35,9 +41,11 @@ class Scheduler
   end
 
   def run
-    while @readable.any? or @writable.any? or @waiting.any?
+    @urgent = IO.pipe
+
+    while @readable.any? or @writable.any? or @waiting.any? or @locking.positive?
       # Can only handle file descriptors up to 1024...
-      readable, writable = IO.select(@readable.keys, @writable.keys, [], next_timeout)
+      readable, writable = IO.select(@readable.keys + [@urgent.first], @writable.keys, [], next_timeout)
 
       # puts "readable: #{readable}" if readable&.any?
       # puts "writable: #{writable}" if writable&.any?
@@ -63,7 +71,24 @@ class Scheduler
           end
         end
       end
+
+      if @ready.any?
+        # Clear out the urgent notification pipe.
+        @urgent.first.read_nonblock(1024)
+
+        ready = nil
+
+        @lock.synchronize do
+          ready, @ready = @ready, Array.new
+        end
+
+        ready.each do |fiber|
+          fiber.resume
+        end
+      end
     end
+  ensure
+    @urgent.each(&:close)
   end
 
   def current_time
@@ -95,6 +120,23 @@ class Scheduler
     return true
   end
 
+  def mutex_lock(mutex)
+    @locking += 1
+    Fiber.yield
+  ensure
+    @locking -= 1
+  end
+
+  def mutex_unlock(mutex, fiber)
+    @lock.synchronize do
+      @ready << fiber
+
+      if @urgent
+        @urgent.last.write('.')
+      end
+    end
+  end
+
   def fiber(&block)
     fiber = Fiber.new(blocking: false, &block)
 
diff --git a/test/fiber/test_mutex.rb b/test/fiber/test_mutex.rb
index 5179959a6a..393a44fc2f 100644
--- a/test/fiber/test_mutex.rb
+++ b/test/fiber/test_mutex.rb
@@ -14,7 +14,7 @@ class TestFiberMutex < Test::Unit::TestCase
         assert_equal Thread.scheduler, scheduler
 
         mutex.synchronize do
-          assert_nil Thread.scheduler
+          assert Thread.scheduler
         end
       end
     end
@@ -22,7 +22,35 @@ class TestFiberMutex < Test::Unit::TestCase
     thread.join
   end
 
+  def test_mutex_interleaved_locking
+    mutex = Mutex.new
+
+    thread = Thread.new do
+      scheduler = Scheduler.new
+      Thread.current.scheduler = scheduler
+
+      Fiber.schedule do
+        mutex.lock
+        sleep 0.1
+        mutex.unlock
+      end
+
+      Fiber.schedule do
+        mutex.lock
+        sleep 0.1
+        mutex.unlock
+      end
+
+      scheduler.run
+    end
+
+    thread.join
+  end
+
   def test_mutex_deadlock
+    err = /No live threads left. Deadlock\?/
+    assert_in_out_err %W[-I#{__dir__} -], <<-RUBY, ['in synchronize'], err, success: false
+    require 'scheduler'
     mutex = Mutex.new
 
     thread = Thread.new do
@@ -30,18 +58,18 @@ class TestFiberMutex < Test::Unit::TestCase
       Thread.current.scheduler = scheduler
 
       Fiber.schedule do
-        assert_equal Thread.scheduler, scheduler
+        raise unless Thread.scheduler == scheduler
 
         mutex.synchronize do
+          puts 'in synchronize'
           Fiber.yield
         end
       end
 
-      assert_raise ThreadError do
-        mutex.lock
-      end
+      mutex.lock
     end
 
     thread.join
+    RUBY
   end
 end