Release C extension mutex when waiting for fds.

Author: aardvark179

lib/truffle/truffle/cext.rb

@@ -1635,14 +1635,24 @@ def rb_iterate(iteration, iterated_object, callback, callback_arg)
   def rb_thread_wait_fd(fd)
     io = IO.for_fd(fd)
     io.autoclose = false
-    IO.select([io])
+    owned = rb_release_cext_mutex
+    begin
+      IO.select([io])
+    ensure
+      rb_acquire_cext_mutex if owned
+    end
     nil
   end
 
   def rb_thread_fd_writable(fd)
     io = IO.for_fd(fd)
     io.autoclose = false
-    _r, w, _e = IO.select(nil, [io])
+    owned = rb_release_cext_mutex
+    begin
+      _r, w, _e = IO.select(nil, [io])
+    ensure
+      rb_acquire_cext_mutex if owned
+    end
     w.size
   end
 
@@ -1661,7 +1671,12 @@ def rb_wait_for_single_fd(fd, events, tv_secs, tv_usecs)
     if tv_secs > 0 || tv_usecs > 0
       timeout = tv_secs + tv_usecs/1.0e6
     end
-    r, w, e = IO.select(read, write, error, *timeout)
+    owned = rb_release_cext_mutex
+    begin
+      r, w, e = IO.select(read, write, error, *timeout)
+    ensure
+      rb_acquire_cext_mutex if owned
+    end
     if r.nil? # timeout
       0
     else

src/main/java/org/truffleruby/cext/CExtNodes.java

@@ -199,6 +199,44 @@ private Object execute(TruffleObject receiver, Object[] args, Node executeNode,
 
     }
 
+    @CoreMethod(names = "rb_acquire_cext_mutex")
+    public abstract static class AcquireCExtMutexNode extends CoreMethodArrayArgumentsNode {
+        @Specialization
+        public boolean acquireMutex(VirtualFrame frame,
+                @Cached("create()") BranchProfile exceptionProfile,
+                @Cached("createBinaryProfile()") ConditionProfile ownedProfile) {
+            if (getContext().getOptions().CEXT_LOCK) {
+                final ReentrantLock lock = getContext().getCExtensionsLock();
+                boolean owned = lock.isHeldByCurrentThread();
+
+                if (ownedProfile.profile(!owned)) {
+                    MutexOperations.lockInternal(getContext(), lock, this);
+                    return true;
+                }
+            }
+            return false;
+        }
+    }
+
+    @CoreMethod(names = "rb_release_cext_mutex")
+    public abstract static class ReleaseCExtMutexNode extends CoreMethodArrayArgumentsNode {
+        @Specialization
+        public boolean releaseMutex(VirtualFrame frame,
+                @Cached("create()") BranchProfile exceptionProfile,
+                @Cached("createBinaryProfile()") ConditionProfile ownedProfile) {
+            if (getContext().getOptions().CEXT_LOCK) {
+                final ReentrantLock lock = getContext().getCExtensionsLock();
+                boolean owned = lock.isHeldByCurrentThread();
+
+                if (ownedProfile.profile(owned)) {
+                    MutexOperations.unlockInternal(lock);
+                    return true;
+                }
+            }
+            return false;
+        }
+    }
+
     @CoreMethod(names = "rb_ulong2num", onSingleton = true, required = 1)
     public abstract static class ULong2NumNode extends CoreMethodArrayArgumentsNode {