[GR-25903] Ensure SafepointManager#poll() is not called from the driving thread.

PullRequest: truffleruby/1940

Author: eregon

lib/truffle/truffle/cext.rb

@@ -1606,16 +1606,9 @@ def rb_thread_call_with_gvl(function, data)
   end
 
   def rb_thread_call_without_gvl(function, data1, unblock, data2)
-    if unblock
-      unblocker = -> {
-        Truffle::Interop.execute_without_conversion(unblock, data2)
-      }
-    end
-
-    Primitive.call_without_c_mutex(
-        -> { Thread.current.unblock(
-            unblocker,
-            -> { function.call(data1) }) }, [])
+    Primitive.call_without_c_mutex(-> {
+      Primitive.call_with_unblocking_function(Thread.current, function, data1, unblock, data2)
+    }, [])
   end
 
   def rb_iterate(iteration, iterated_object, callback, callback_arg)

src/main/java/org/truffleruby/core/thread/ThreadNodes.java

@@ -76,6 +76,8 @@
 import org.truffleruby.core.symbol.RubySymbol;
 import org.truffleruby.core.thread.ThreadManager.UnblockingAction;
 import org.truffleruby.core.thread.ThreadManager.UnblockingActionHolder;
+import org.truffleruby.interop.InteropNodes;
+import org.truffleruby.interop.TranslateInteropExceptionNode;
 import org.truffleruby.language.Nil;
 import org.truffleruby.language.NotProvided;
 import org.truffleruby.language.SafepointAction;
@@ -88,11 +90,14 @@
 import org.truffleruby.language.objects.shared.SharedObjects;
 import org.truffleruby.language.yield.YieldNode;
 
+import com.oracle.truffle.api.CompilerDirectives;
 import com.oracle.truffle.api.CompilerDirectives.CompilationFinal;
 import com.oracle.truffle.api.CompilerDirectives.TruffleBoundary;
 import com.oracle.truffle.api.dsl.Cached;
 import com.oracle.truffle.api.dsl.ImportStatic;
 import com.oracle.truffle.api.dsl.Specialization;
+import com.oracle.truffle.api.interop.InteropException;
+import com.oracle.truffle.api.interop.InteropLibrary;
 import com.oracle.truffle.api.library.CachedLibrary;
 import com.oracle.truffle.api.nodes.Node;
 import com.oracle.truffle.api.object.Shape;
@@ -503,46 +508,49 @@ protected RubyThread wakeup(RubyThread rubyThread) {
     }
 
     @NonStandard
-    @CoreMethod(names = "unblock", required = 2)
-    public abstract static class UnblockNode extends YieldingCoreMethodNode {
+    @Primitive(name = "call_with_unblocking_function")
+    public abstract static class CallWithUnblockingFunctionNode extends PrimitiveArrayArgumentsNode {
 
-        @Specialization
-        protected Object unblock(RubyThread thread, Object unblocker, RubyProc runner,
-                @Cached("createCountingProfile()") LoopConditionProfile loopProfile) {
+        @Specialization(limit = "getCacheLimit()")
+        protected Object call(RubyThread thread, Object function, Object arg, Object unblocker, Object unblockerArg,
+                @CachedLibrary("function") InteropLibrary receivers,
+                @Cached TranslateInteropExceptionNode translateInteropExceptionNode) {
             final ThreadManager threadManager = getContext().getThreadManager();
             final UnblockingAction unblockingAction;
             if (unblocker == nil) {
                 unblockingAction = threadManager.getNativeCallUnblockingAction();
             } else {
-                unblockingAction = makeUnblockingAction((RubyProc) unblocker);
+                unblockingAction = makeUnblockingAction(unblocker, unblockerArg);
             }
-            final UnblockingActionHolder actionHolder = threadManager.getActionHolder(Thread.currentThread());
 
+            final UnblockingActionHolder actionHolder = threadManager.getActionHolder(Thread.currentThread());
             final UnblockingAction oldAction = actionHolder.changeTo(unblockingAction);
+            final ThreadStatus status = thread.status;
+            thread.status = ThreadStatus.SLEEP;
             try {
-                Object result;
-                do {
-                    final ThreadStatus status = thread.status;
-                    thread.status = ThreadStatus.SLEEP;
-
-                    try {
-                        result = yield(runner);
-                    } finally {
-                        thread.status = status;
-                    }
-                } while (loopProfile.profile(result == null));
-
-                return result;
+                return InteropNodes.execute(function, new Object[]{ arg }, receivers, translateInteropExceptionNode);
             } finally {
+                thread.status = status;
                 actionHolder.restore(oldAction);
+
             }
         }
 
         @TruffleBoundary
-        private UnblockingAction makeUnblockingAction(RubyProc unblocker) {
-            return () -> yield(unblocker);
+        private UnblockingAction makeUnblockingAction(Object function, Object argument) {
+            assert InteropLibrary.getUncached().isExecutable(function);
+            return () -> {
+                try {
+                    InteropLibrary.getUncached().execute(function, argument);
+                } catch (InteropException e) {
+                    throw CompilerDirectives.shouldNotReachHere(e);
+                }
+            };
         }
 
+        protected int getCacheLimit() {
+            return getContext().getOptions().DISPATCH_CACHE;
+        }
     }
 
     @CoreMethod(names = "list", onSingleton = true)

src/main/java/org/truffleruby/language/SafepointManager.java

@@ -108,6 +108,10 @@ private void poll(Node currentNode, boolean fromBlockingCall) {
 
     @TruffleBoundary
     private void assumptionInvalidated(Node currentNode, boolean fromBlockingCall) {
+        if (lock.isHeldByCurrentThread()) {
+            throw CompilerDirectives.shouldNotReachHere("poll() should not be called by the driving thread");
+        }
+
         final RubyThread thread = context.getThreadManager().getCurrentThread();
         final InterruptMode interruptMode = thread.interruptMode;
 
@@ -129,6 +133,8 @@ private void assumptionInvalidated(Node currentNode, boolean fromBlockingCall) {
 
     @TruffleBoundary
     private SafepointAction step(Node currentNode, boolean isDrivingThread) {
+        assert isDrivingThread == lock.isHeldByCurrentThread();
+
         final RubyThread thread = context.getThreadManager().getCurrentThread();
 
         // Wait for other threads to reach their safepoint
@@ -314,9 +320,8 @@ private void pauseAllThreadsAndExecute(Node currentNode, SafepointAction action,
         this.action = action;
         this.deferred = deferred;
 
-        /* this is a potential cause for race conditions, but we need to invalidate first so the interrupted threads see
-         * the invalidation in poll() in their catch(InterruptedException) clause and wait on the barrier instead of
-         * retrying their blocking action. */
+        /* We need to invalidate first so the interrupted threads see the invalidation in poll() in their
+         * catch(InterruptedException) clause and wait on the Phaser instead of retrying their blocking action. */
         assumption.invalidate();
         interruptOtherThreads();