[GR-6190] Do not trigger initializeMultiThreading() when creating Fibers

* Instead, leave the TruffleContext explicitly before switching to another Fiber.
* Leave the SafepointManager before leaving the TruffleContext, as we
  cannot execute safepoint actions without being in the context.
* Print internal errors escaping fiberMain() and the fiber pool.

Author: eregon

CHANGELOG.md

@@ -5,6 +5,7 @@ New features:
 * Access to local variables of the interactive Binding via language bindings is now supported: `context.getBindings("ruby").putMember("my_var", 42);` (#2030).
 * `VALUE`s in C extensions now expose the Ruby object when viewed in the debugger, as long as they have not been converted to native values.
 * Signal handlers can now be run without triggering multi-threading.
+* Fibers no longer trigger Truffle multi-threading.
 
 Bug fixes:
 

src/main/java/org/truffleruby/core/fiber/FiberManager.java

@@ -14,6 +14,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 
+import com.oracle.truffle.api.TruffleContext;
 import org.truffleruby.RubyContext;
 import org.truffleruby.RubyLanguage;
 import org.truffleruby.core.array.ArrayHelpers;
@@ -103,7 +104,7 @@ public RubyFiber createFiber(RubyLanguage language, RubyContext context, RubyThr
     public void initialize(RubyFiber fiber, RubyProc block, Node currentNode) {
         ThreadManager.FIBER_BEING_SPAWNED.set(fiber);
         try {
-            context.getThreadManager().spawnFiber(() -> fiberMain(context, fiber, block, currentNode));
+            context.getThreadManager().spawnFiber(fiber, () -> fiberMain(context, fiber, block, currentNode));
             waitForInitialization(context, fiber, currentNode);
         } finally {
             ThreadManager.FIBER_BEING_SPAWNED.remove();
@@ -125,7 +126,7 @@ public static void waitForInitialization(RubyContext context, RubyFiber fiber, N
         }
     }
 
-    private static final BranchProfile UNPROFILED = BranchProfile.create();
+    private static final BranchProfile UNPROFILED = BranchProfile.getUncached();
 
     private void fiberMain(RubyContext context, RubyFiber fiber, RubyProc block, Node currentNode) {
         assert fiber != rootFiber : "Root Fibers execute threadMain() and not fiberMain()";
@@ -135,16 +136,25 @@ private void fiberMain(RubyContext context, RubyFiber fiber, RubyProc block, Nod
         final String oldName = thread.getName();
         thread.setName(NAME_PREFIX + " id=" + thread.getId() + " from " + RubyContext.fileLine(sourceSection));
 
-        start(fiber, thread);
-        try {
+        start(fiber, thread, false);
+
+        final TruffleContext truffleContext = context.getEnv().getContext();
+        assert !truffleContext.isEntered();
 
-            final Object[] args = waitForResume(fiber);
+        final FiberMessage message = waitMessage(fiber);
+        final Object prev = truffleContext.enter(currentNode);
+        context.getSafepointManager().enterThread(); // not done in start() above because the context was not entered
+        try {
             final Object result;
             try {
+                final Object[] args = handleMessage(fiber, message);
                 result = ProcOperations.rootCall(block, args);
             } finally {
                 // Make sure that other fibers notice we are dead before they gain control back
                 fiber.alive = false;
+                // Leave before resume/sendExceptionToParentFiber -> addToMessageQueue() -> parent Fiber starts executing
+                context.getSafepointManager().leaveThread();
+                truffleContext.leave(currentNode, prev);
             }
             resume(fiber, getReturnFiber(fiber, currentNode, UNPROFILED), FiberOperation.YIELD, result);
 
@@ -164,8 +174,11 @@ private void fiberMain(RubyContext context, RubyFiber fiber, RubyProc block, Nod
                     fiber,
                     new RaiseException(context, context.getCoreExceptions().unexpectedReturn(currentNode)),
                     currentNode);
+        } catch (Throwable e) {
+            final RuntimeException exception = ThreadManager.printInternalError(e);
+            sendExceptionToParentFiber(fiber, exception, currentNode);
         } finally {
-            cleanup(fiber, thread);
+            cleanup(fiber, thread, false);
             thread.setName(oldName);
         }
     }
@@ -193,16 +206,19 @@ public RubyFiber getReturnFiber(RubyFiber currentFiber, Node currentNode, Branch
 
     @TruffleBoundary
     private void addToMessageQueue(RubyFiber fiber, FiberMessage message) {
+        assert !context.getEnv().getContext().isEntered() : "should have left context when sending message to fiber";
         fiber.messageQueue.add(message);
     }
 
-    /** Send the Java thread that represents this fiber to sleep until it receives a resume or exit message. */
+    /** Send the Java thread that represents this fiber to sleep until it receives a message. */
     @TruffleBoundary
-    private Object[] waitForResume(RubyFiber fiber) {
-        final FiberMessage message = context.getThreadManager().runUntilResultKeepStatus(
-                null,
-                () -> fiber.messageQueue.take());
+    private FiberMessage waitMessage(RubyFiber fiber) {
+        assert !context.getEnv().getContext().isEntered() : "should have left context while waiting fiber message";
+        return ThreadManager.retryWhileInterrupted(fiber.messageQueue::take);
+    }
 
+    @TruffleBoundary
+    private Object[] handleMessage(RubyFiber fiber, FiberMessage message) {
         setCurrentFiber(fiber);
 
         if (message instanceof FiberShutdownMessage) {
@@ -227,12 +243,31 @@ private void resume(RubyFiber fromFiber, RubyFiber fiber, FiberOperation operati
         addToMessageQueue(fiber, new FiberResumeMessage(operation, fromFiber, args));
     }
 
+    @TruffleBoundary
     public Object[] transferControlTo(RubyFiber fromFiber, RubyFiber fiber, FiberOperation operation, Object[] args) {
-        resume(fromFiber, fiber, operation, args);
-        return waitForResume(fromFiber);
+        final TruffleContext truffleContext = context.getEnv().getContext();
+
+        final FiberMessage message;
+        final boolean isRubyManagedThread = context.getThreadManager().isRubyManagedThread(Thread.currentThread());
+        if (isRubyManagedThread) {
+            context.getSafepointManager().leaveThread();
+        }
+        try {
+            message = truffleContext.leaveAndEnter(null, () -> {
+                resume(fromFiber, fiber, operation, args);
+                return waitMessage(fromFiber);
+            });
+        } finally {
+            if (isRubyManagedThread) {
+                context.getSafepointManager().enterThread();
+            }
+        }
+
+        return handleMessage(fromFiber, message);
     }
 
-    public void start(RubyFiber fiber, Thread javaThread) {
+    public void start(RubyFiber fiber, Thread javaThread, boolean entered) {
+        assert entered == context.getEnv().getContext().isEntered();
         final ThreadManager threadManager = context.getThreadManager();
 
         if (Thread.currentThread() == javaThread) {
@@ -249,31 +284,34 @@ public void start(RubyFiber fiber, Thread javaThread) {
 
         runningFibers.add(fiber);
 
-        if (threadManager.isRubyManagedThread(javaThread)) {
+        if (threadManager.isRubyManagedThread(javaThread) && Thread.currentThread() == javaThread && entered) {
             context.getSafepointManager().enterThread();
         }
 
         // fully initialized
         fiber.initializedLatch.countDown();
     }
 
-    public void cleanup(RubyFiber fiber, Thread javaThread) {
+    public void cleanup(RubyFiber fiber, Thread javaThread, boolean entered) {
+        assert entered == context.getEnv().getContext().isEntered();
+        final ThreadManager threadManager = context.getThreadManager();
+
         fiber.alive = false;
 
-        if (context.getThreadManager().isRubyManagedThread(javaThread)) {
+        if (threadManager.isRubyManagedThread(javaThread) && Thread.currentThread() == javaThread && entered) {
             context.getSafepointManager().leaveThread();
         }
 
-        context.getThreadManager().cleanupValuesForJavaThread(javaThread);
+        threadManager.cleanupValuesForJavaThread(javaThread);
 
         runningFibers.remove(fiber);
 
         fiber.thread = null;
 
         if (Thread.currentThread() == javaThread) {
-            context.getThreadManager().rubyFiber.remove();
+            threadManager.rubyFiber.remove();
         }
-        context.getThreadManager().rubyFiberForeignMap.remove(javaThread);
+        threadManager.rubyFiberForeignMap.remove(javaThread);
 
         fiber.finishedLatch.countDown();
     }
@@ -283,13 +321,32 @@ public void killOtherFibers() {
         // All Fibers except the current one are in waitForResume(),
         // so sending a FiberShutdownMessage is enough to finish them.
         // This also avoids the performance cost of a safepoint.
+
+        // This method might not be executed on the rootFiber Java Thread but possibly on another Java Thread.
+
+        final TruffleContext truffleContext = context.getEnv().getContext();
+        assert truffleContext.isEntered();
+        assert context.getThreadManager().isRubyManagedThread(Thread.currentThread());
+
+        context.getSafepointManager().leaveThread();
+        try {
+            truffleContext.leaveAndEnter(null, () -> {
+                doKillOtherFibers();
+                return null;
+            });
+        } finally {
+            context.getSafepointManager().enterThread();
+        }
+    }
+
+    private void doKillOtherFibers() {
         for (RubyFiber fiber : runningFibers) {
             if (fiber != rootFiber) {
                 addToMessageQueue(fiber, new FiberShutdownMessage());
 
                 // Wait for the Fiber to finish so we only run one Fiber at a time
                 final CountDownLatch finishedLatch = fiber.finishedLatch;
-                context.getThreadManager().runUntilResultKeepStatus(null, () -> {
+                ThreadManager.retryWhileInterrupted(() -> {
                     finishedLatch.await();
                     return BlockingAction.SUCCESS;
                 });
@@ -300,7 +357,7 @@ public void killOtherFibers() {
     @TruffleBoundary
     public void shutdown(Thread javaThread) {
         killOtherFibers();
-        cleanup(rootFiber, javaThread);
+        cleanup(rootFiber, javaThread, true);
     }
 
     public String getFiberDebugInfo() {

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

@@ -151,9 +151,20 @@ public void restartMainThread(Thread mainJavaThread) {
     public static final ThreadLocal<RubyFiber> FIBER_BEING_SPAWNED = new ThreadLocal<>();
 
     private Thread createFiberJavaThread(Runnable runnable) {
-        RubyFiber fiber = FIBER_BEING_SPAWNED.get();
+        final RubyFiber fiber = FIBER_BEING_SPAWNED.get();
         assert fiber != null;
-        return createJavaThread(runnable, fiber);
+
+        if (context.isPreInitializing()) {
+            throw new UnsupportedOperationException("fibers should not be created while pre-initializing the context");
+        }
+
+        final Thread thread = new Thread(runnable); // context.getEnv().createUnenteredThread(runnable);
+        thread.setUncaughtExceptionHandler((javaThread, throwable) -> {
+            System.err.println("Throwable escaped Fiber pool thread:");
+            throwable.printStackTrace();
+        });
+        rubyManagedThreads.add(thread);
+        return thread;
     }
 
     private Thread createJavaThread(Runnable runnable, RubyFiber fiber) {
@@ -168,26 +179,39 @@ private Thread createJavaThread(Runnable runnable, RubyFiber fiber) {
         }
 
         final Thread thread = context.getEnv().createThread(runnable);
+        thread.setUncaughtExceptionHandler(uncaughtExceptionHandler(fiber));
+        rubyManagedThreads.add(thread);
+        return thread;
+    }
 
+    private static Thread.UncaughtExceptionHandler uncaughtExceptionHandler(RubyFiber fiber) {
         assert fiber != null;
-        thread.setUncaughtExceptionHandler((javaThread, throwable) -> {
+        return (javaThread, throwable) -> {
+            printInternalError(throwable);
             try {
-                fiber.uncaughtException = throwable;
-                fiber.initializedLatch.countDown();
-            } catch (Throwable t) {
+                // the Fiber is not yet initialized, unblock the caller and rethrow the exception to it
+                if (fiber.initializedLatch.getCount() > 0) {
+                    fiber.uncaughtException = throwable;
+                    fiber.initializedLatch.countDown();
+                }
+            } catch (Throwable t) { // exception inside this UncaughtExceptionHandler
                 t.initCause(throwable);
-                t.printStackTrace();
-                Thread.getDefaultUncaughtExceptionHandler().uncaughtException(javaThread, t);
+                printInternalError(t);
             }
-        });
-
-        rubyManagedThreads.add(thread);
-        return thread;
+        };
     }
 
     @SuppressFBWarnings("RV")
-    public void spawnFiber(Runnable task) {
-        fiberPool.submit(task);
+    public void spawnFiber(RubyFiber fiber, Runnable task) {
+        fiberPool.submit(() -> {
+            try {
+                task.run();
+            } catch (Throwable t) {
+                // Fibers are run on a thread-pool, so make sure any exception escaping
+                // is handled here as the thread pool ignores exceptions.
+                uncaughtExceptionHandler(fiber).uncaughtException(Thread.currentThread(), t);
+            }
+        });
     }
 
     @TruffleBoundary
@@ -280,11 +304,7 @@ private void threadMain(RubyThread thread, Node currentNode, Supplier<Object> ta
             rethrowOnMainThread(currentNode, e);
             setThreadValue(thread, Nil.INSTANCE);
         } catch (Throwable e) {
-            final String message = StringUtils
-                    .format("%s terminated with internal error:", Thread.currentThread().getName());
-            final RuntimeException runtimeException = new RuntimeException(message, e);
-            // Immediately print internal exceptions, in case they would cause a deadlock
-            runtimeException.printStackTrace();
+            final RuntimeException runtimeException = printInternalError(e);
             rethrowOnMainThread(currentNode, runtimeException);
             setThreadValue(thread, Nil.INSTANCE);
         } finally {
@@ -293,6 +313,15 @@ private void threadMain(RubyThread thread, Node currentNode, Supplier<Object> ta
         }
     }
 
+    public static RuntimeException printInternalError(Throwable e) {
+        final String message = StringUtils
+                .format("%s terminated with internal error:", Thread.currentThread().getName());
+        final RuntimeException runtimeException = new RuntimeException(message, e);
+        // Immediately print internal exceptions, in case they would cause a deadlock
+        runtimeException.printStackTrace();
+        return runtimeException;
+    }
+
     private void rethrowOnMainThread(Node currentNode, RuntimeException e) {
         context.getSafepointManager().pauseRubyThreadAndExecute(
                 "rethrow " + e.getClass() + " to main thread",
@@ -361,7 +390,7 @@ private void start(RubyThread thread, Thread javaThread) {
         registerThread(thread);
 
         final FiberManager fiberManager = thread.fiberManager;
-        fiberManager.start(fiberManager.getRootFiber(), javaThread);
+        fiberManager.start(fiberManager.getRootFiber(), javaThread, true);
     }
 
     public void cleanup(RubyThread thread, Thread javaThread) {
@@ -441,6 +470,7 @@ public static <T> T retryWhileInterrupted(BlockingAction<T> action) {
 
     @TruffleBoundary
     public <T> T runUntilResultKeepStatus(Node currentNode, BlockingAction<T> action) {
+        assert context.getEnv().getContext().isEntered() : "Use retryWhileInterrupted() when not entered";
         T result = null;
 
         do {

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

@@ -87,6 +87,7 @@ public void enterThread() {
     @TruffleBoundary
     public void leaveThread() {
         final Thread thread = Thread.currentThread();
+        assert runningThreads.contains(thread);
 
         phaser.arriveAndDeregister();
         if (!runningThreads.remove(thread)) {

src/test/java/org/truffleruby/ContextPermissionsTest.java

@@ -10,9 +10,13 @@
 package org.truffleruby;
 
 import org.graalvm.polyglot.Context;
+import org.graalvm.polyglot.Value;
 import org.junit.Assert;
 import org.junit.Test;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
 public class ContextPermissionsTest {
 
     @Test
@@ -77,4 +81,32 @@ public void testNoThreadsEnforcesSingleThreadedOption() throws Throwable {
         }
     }
 
+    @Test
+    public void testFiberDoesNotTriggerMultiThreading() {
+        try (Context context = Context.newBuilder("ruby").allowCreateThread(false).build()) {
+            final Value array = context.eval(
+                    "ruby",
+                    "a = [1]; f = Fiber.new { a << 3; Fiber.yield; a << 5 }; a << 2; f.resume; a << 4; f.resume");
+            assertTrue(array.hasArrayElements());
+            assertEquals(5, array.getArraySize());
+            for (int i = 0; i < 5; i++) {
+                assertEquals(i + 1, array.getArrayElement(i).asInt());
+            }
+        }
+    }
+
+    @Test
+    public void testNestedFiberAndTerminateFiber() {
+        try (Context context = Context.newBuilder("ruby").allowCreateThread(false).build()) {
+            final Value array = context.eval(
+                    "ruby",
+                    "a = []; Fiber.new { a << 1; Fiber.new { a << 2; Fiber.yield; unreachable }.resume; a << 3 }.resume");
+            assertTrue(array.hasArrayElements());
+            assertEquals(3, array.getArraySize());
+            for (int i = 0; i < 3; i++) {
+                assertEquals(i + 1, array.getArrayElement(i).asInt());
+            }
+        }
+    }
+
 }