[GR-44211] Use Env#newTruffleThreadBuilder for Ruby Fibers

PullRequest: truffleruby/3681

Author: andrykonchin

spec/ruby/core/fiber/inspect_spec.rb

@@ -0,0 +1,38 @@
+require_relative '../../spec_helper'
+require 'fiber'
+
+describe "Fiber#inspect" do
+  describe "status" do
+    it "is resumed for the root Fiber of a Thread" do
+      inspected = Thread.new { Fiber.current.inspect }.value
+      inspected.should =~ /\A#<Fiber:0x\h+ .*\(resumed\)>\z/
+    end
+
+    it "is created for a Fiber which did not run yet" do
+      inspected = Fiber.new {}.inspect
+      inspected.should =~ /\A#<Fiber:0x\h+ .+ \(created\)>\z/
+    end
+
+    it "is resumed for a Fiber which was resumed" do
+      inspected = Fiber.new { Fiber.current.inspect }.resume
+      inspected.should =~ /\A#<Fiber:0x\h+ .+ \(resumed\)>\z/
+    end
+
+    ruby_version_is "3.0" do
+      it "is resumed for a Fiber which was transferred" do
+        inspected = Fiber.new { Fiber.current.inspect }.transfer
+        inspected.should =~ /\A#<Fiber:0x\h+ .+ \(resumed\)>\z/
+      end
+    end
+
+    it "is suspended for a Fiber which was resumed and yielded" do
+      inspected = Fiber.new { Fiber.yield }.tap(&:resume).inspect
+      inspected.should =~ /\A#<Fiber:0x\h+ .+ \(suspended\)>\z/
+    end
+
+    it "is terminated for a Fiber which has terminated" do
+      inspected = Fiber.new {}.tap(&:resume).inspect
+      inspected.should =~ /\A#<Fiber:0x\h+ .+ \(terminated\)>\z/
+    end
+  end
+end

spec/tags/core/exception/top_level_tags.txt

@@ -3,4 +3,3 @@ slow:An Exception reaching the top level with a custom backtrace is printed on S
 slow:An Exception reaching the top level the Exception#cause is printed to STDERR with backtraces
 slow:An Exception reaching the top level kills all threads and fibers, ensure clauses are only run for threads current fibers, not for suspended fibers with ensure on the root fiber
 slow:An Exception reaching the top level kills all threads and fibers, ensure clauses are only run for threads current fibers, not for suspended fibers with ensure on non-root fiber
-fails(GR-44211):An Exception reaching the top level kills all threads and fibers, ensure clauses are only run for threads current fibers, not for suspended fibers with ensure on non-root fiber

src/main/java/org/truffleruby/RubyLanguage.java

@@ -406,6 +406,17 @@ public void invalidateTracingAssumption() {
         tracingAssumption = tracingCyclicAssumption.getAssumption();
     }
 
+    private boolean multiThreading = false;
+
+    public boolean isMultiThreaded() {
+        return multiThreading;
+    }
+
+    @Override
+    protected void initializeMultiThreading(RubyContext context) {
+        this.multiThreading = true;
+    }
+
     @Override
     protected void initializeMultipleContexts() {
         LOGGER.fine("initializeMultipleContexts()");
@@ -619,8 +630,9 @@ public void initializeThread(RubyContext context, Thread thread) {
                             .shouldNotReachHere("Ruby threads should be initialized on their Java thread");
                 }
                 context.getThreadManager().start(rubyThread, thread);
-            } else {
-                // Fiber
+            } else { // (non-root) Fiber
+                var fiber = this.rubyFiber.get(thread);
+                rubyThread.setCurrentFiber(fiber);
             }
             return;
         }
@@ -657,7 +669,8 @@ public void disposeThread(RubyContext context, Thread thread) {
                 }
                 context.getThreadManager().cleanupThreadState(rubyThread, thread);
             } else { // (non-root) Fiber
-                // Fibers are always cleaned up by their thread's cleanup with FiberManager#killOtherFibers()
+                var fiber = this.rubyFiber.get(thread);
+                context.fiberManager.cleanup(fiber, thread);
             }
             return;
         }

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

@@ -9,8 +9,10 @@
  */
 package org.truffleruby.core.fiber;
 
+import java.util.Objects;
 import java.util.concurrent.CountDownLatch;
 
+import com.oracle.truffle.api.TruffleSafepoint.Interrupter;
 import org.truffleruby.core.fiber.RubyFiber.FiberStatus;
 
 import com.oracle.truffle.api.TruffleContext;
@@ -60,27 +62,56 @@ public FiberManager(RubyLanguage language, RubyContext context) {
     public void initialize(RubyFiber fiber, boolean blocking, RubyProc block, Node currentNode) {
         final SourceSection sourceSection = block.getSharedMethodInfo().getSourceSection();
         fiber.sourceLocation = context.fileLine(sourceSection);
+        fiber.body = block;
+        fiber.initializeNode = currentNode;
         fiber.blocking = blocking;
+    }
+
+    private void createThreadToReceiveFirstMessage(RubyFiber fiber, Node currentNode) {
+        assert fiber.thread == null;
+        RubyProc block = Objects.requireNonNull(fiber.body);
+        fiber.body = null;
+        Node initializeNode = Objects.requireNonNull(fiber.initializeNode);
+        fiber.initializeNode = null;
 
+        var sourceSection = block.getSharedMethodInfo().getSourceSection();
         final TruffleContext truffleContext = context.getEnv().getContext();
 
-        context.getThreadManager().leaveAndEnter(truffleContext, currentNode, () -> {
-            context.getThreadManager().spawnFiber(fiber, sourceSection,
-                    () -> fiberMain(context, fiber, block, currentNode));
-            waitForInitialization(context, fiber, currentNode);
+        truffleContext.leaveAndEnter(currentNode, Interrupter.THREAD_INTERRUPT, (unused) -> {
+            Thread thread = context.getThreadManager().createFiberJavaThread(fiber, sourceSection,
+                    () -> beforeEnter(fiber, initializeNode),
+                    () -> fiberMain(context, fiber, block, initializeNode),
+                    () -> afterLeave(fiber), currentNode);
+            fiber.thread = thread;
+            thread.start();
+            waitForInitializationUnentered(context, fiber, currentNode);
             return BlockingAction.SUCCESS;
-        });
+        }, null);
     }
 
     /** Wait for full initialization of the new fiber */
-    public static void waitForInitialization(RubyContext context, RubyFiber fiber, Node currentNode) {
+    public static void waitForInitializationEntered(RubyContext context, RubyFiber fiber, Node currentNode) {
+        assert context.getEnv().getContext().isEntered();
         final CountDownLatch initializedLatch = fiber.initializedLatch;
 
-        if (context.getEnv().getContext().isEntered()) {
-            context.getThreadManager().runUntilResultKeepStatus(currentNode, CountDownLatch::await, initializedLatch);
-        } else {
-            context.getThreadManager().retryWhileInterrupted(currentNode, CountDownLatch::await, initializedLatch);
+        context.getThreadManager().runUntilResultKeepStatus(currentNode, latch -> {
+            latch.await();
+            return BlockingAction.SUCCESS;
+        }, initializedLatch);
+
+        final Throwable uncaughtException = fiber.uncaughtException;
+        if (uncaughtException != null) {
+            ExceptionOperations.rethrow(uncaughtException);
         }
+    }
+
+    /** Wait for full initialization of the new fiber */
+    public static void waitForInitializationUnentered(RubyContext context, RubyFiber fiber, Node currentNode)
+            throws InterruptedException {
+        assert !context.getEnv().getContext().isEntered();
+        final CountDownLatch initializedLatch = fiber.initializedLatch;
+
+        initializedLatch.await();
 
         final Throwable uncaughtException = fiber.uncaughtException;
         if (uncaughtException != null) {
@@ -90,23 +121,26 @@ public static void waitForInitialization(RubyContext context, RubyFiber fiber, N
 
     private static final BranchProfile UNPROFILED = BranchProfile.getUncached();
 
-    private void fiberMain(RubyContext context, RubyFiber fiber, RubyProc block, Node currentNode) {
+    private void beforeEnter(RubyFiber fiber, Node currentNode) {
         assert !fiber.isRootFiber() : "Root Fibers execute threadMain() and not fiberMain()";
         assertNotEntered("Fibers should start unentered to avoid triggering multithreading");
 
         final Thread thread = Thread.currentThread();
-        final TruffleContext truffleContext = context.getEnv().getContext();
-
         start(fiber, thread);
 
         // fully initialized
         fiber.initializedLatch.countDown();
 
-        final FiberMessage message = waitMessage(fiber, currentNode);
-        fiber.rubyThread.setCurrentFiber(fiber);
+        try {
+            fiber.firstMessage = waitMessage(fiber, currentNode);
+        } catch (InterruptedException e) {
+            throw CompilerDirectives.shouldNotReachHere("unexpected interrupt in Fiber beforeEnter()");
+        }
+    }
 
-        // enter() polls so we need the current Fiber to be set before enter()
-        final Object prev = truffleContext.enter(currentNode);
+    private void fiberMain(RubyContext context, RubyFiber fiber, RubyProc block, Node currentNode) {
+        final FiberMessage message = Objects.requireNonNull(fiber.firstMessage);
+        fiber.firstMessage = null;
 
         FiberMessage lastMessage = null;
         try {
@@ -134,18 +168,21 @@ private void fiberMain(RubyContext context, RubyFiber fiber, RubyProc block, Nod
             final RuntimeException exception = ThreadManager.printInternalError(e);
             lastMessage = new FiberExceptionMessage(exception);
         } finally {
-            final RubyFiber returnFiber = lastMessage == null ? null : getReturnFiber(fiber, currentNode, UNPROFILED);
+            fiber.lastMessage = lastMessage;
+            fiber.returnFiber = lastMessage == null ? null : getReturnFiber(fiber, currentNode, UNPROFILED);
 
             // Perform all cleanup before resuming the parent Fiber
             // Make sure that other fibers notice we are dead before they gain control back
             fiber.status = FiberStatus.TERMINATED;
             // Leave context before addToMessageQueue() -> parent Fiber starts executing
-            truffleContext.leave(currentNode, prev);
-            cleanup(fiber, thread);
+        }
+    }
 
-            if (lastMessage != null) {
-                addToMessageQueue(returnFiber, lastMessage);
-            }
+    private void afterLeave(RubyFiber fiber) {
+        if (fiber.lastMessage != null) {
+            addToMessageQueue(fiber.returnFiber, fiber.lastMessage);
+            fiber.returnFiber = null;
+            fiber.lastMessage = null;
         }
     }
 
@@ -161,7 +198,7 @@ public RubyFiber getReturnFiber(RubyFiber currentFiber, Node currentNode, Branch
             return previousFiber;
         } else {
 
-            if (currentFiber == rootFiber) {
+            if (currentFiber == rootFiber) { // Note: this is always false for the fiberMain() caller
                 errorProfile.enter();
                 throw new RaiseException(context, context.getCoreExceptions().yieldFromRootFiberError(currentNode));
             }
@@ -184,24 +221,9 @@ private void addToMessageQueue(RubyFiber fiber, FiberMessage message) {
 
     /** Send the Java thread that represents this fiber to sleep until it receives a message. */
     @TruffleBoundary
-    private FiberMessage waitMessage(RubyFiber fiber, Node currentNode) {
+    private FiberMessage waitMessage(RubyFiber fiber, Node currentNode) throws InterruptedException {
         assertNotEntered("should have left context while waiting fiber message");
-
-        class State {
-            final RubyFiber fiber;
-            FiberMessage message;
-
-            State(RubyFiber fiber) {
-                this.fiber = fiber;
-            }
-        }
-
-        final State state = new State(fiber);
-        context.getThreadManager().retryWhileInterrupted(
-                currentNode,
-                s -> s.message = s.fiber.messageQueue.take(),
-                state);
-        return state.message;
+        return fiber.messageQueue.take();
     }
 
     private void assertNotEntered(String reason) {
@@ -287,26 +309,29 @@ public DescriptorAndArgs transferControlTo(RubyFiber fromFiber, RubyFiber toFibe
     @TruffleBoundary
     private FiberMessage resumeAndWait(RubyFiber fromFiber, RubyFiber toFiber, FiberOperation operation,
             ArgumentsDescriptor descriptor, Object[] args, Node currentNode) {
+
+        if (toFiber.body != null) {
+            context.fiberManager.createThreadToReceiveFirstMessage(toFiber, currentNode);
+        }
+
         final TruffleContext truffleContext = context.getEnv().getContext();
-        final FiberMessage message = context
-                .getThreadManager()
-                .leaveAndEnter(truffleContext, currentNode, () -> {
+        final FiberMessage message = truffleContext.leaveAndEnter(currentNode, Interrupter.THREAD_INTERRUPT,
+                (unused) -> {
                     resume(fromFiber, toFiber, operation, descriptor, args);
                     return waitMessage(fromFiber, currentNode);
-                });
+                }, null);
         fromFiber.rubyThread.setCurrentFiber(fromFiber);
         return message;
     }
 
     @TruffleBoundary
     public void safepoint(RubyFiber fromFiber, RubyFiber fiber, SafepointAction action, Node currentNode) {
         final TruffleContext truffleContext = context.getEnv().getContext();
-        final FiberResumeMessage returnMessage = (FiberResumeMessage) context
-                .getThreadManager()
-                .leaveAndEnter(truffleContext, currentNode, () -> {
+        final FiberResumeMessage returnMessage = (FiberResumeMessage) truffleContext.leaveAndEnter(currentNode,
+                Interrupter.THREAD_INTERRUPT, (unused) -> {
                     addToMessageQueue(fiber, new FiberSafepointMessage(fromFiber, action));
                     return waitMessage(fromFiber, currentNode);
-                });
+                }, null);
         fromFiber.rubyThread.setCurrentFiber(fromFiber);
 
         if (returnMessage.getArgs() != SAFEPOINT_ARGS) {
@@ -315,7 +340,11 @@ public void safepoint(RubyFiber fromFiber, RubyFiber fiber, SafepointAction acti
     }
 
     public void start(RubyFiber fiber, Thread javaThread) {
-        fiber.thread = javaThread;
+        if (fiber.isRootFiber()) {
+            fiber.thread = javaThread;
+        } else {
+            // fiber.thread set by createThreadToReceiveFirstMessage()
+        }
 
         final RubyThread rubyThread = fiber.rubyThread;
 
@@ -354,26 +383,23 @@ public void killOtherFibers(RubyThread thread) {
         boolean allowSideEffects = safepoint.setAllowSideEffects(false);
         try {
             final TruffleContext truffleContext = context.getEnv().getContext();
-            context.getThreadManager().leaveAndEnter(truffleContext, DummyNode.INSTANCE, () -> {
+            truffleContext.leaveAndEnter(DummyNode.INSTANCE, Interrupter.THREAD_INTERRUPT, (unused) -> {
                 doKillOtherFibers(thread);
                 return BlockingAction.SUCCESS;
-            });
+            }, null);
         } finally {
             safepoint.setAllowSideEffects(allowSideEffects);
         }
     }
 
-    private void doKillOtherFibers(RubyThread thread) {
+    private void doKillOtherFibers(RubyThread thread) throws InterruptedException {
         for (RubyFiber fiber : thread.runningFibers) {
             if (!fiber.isRootFiber()) {
                 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().retryWhileInterrupted(
-                        DummyNode.INSTANCE,
-                        CountDownLatch::await,
-                        finishedLatch);
+                finishedLatch.await();
 
                 final Throwable uncaughtException = fiber.uncaughtException;
                 if (uncaughtException != null) {

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

@@ -109,6 +109,7 @@ protected RubyFiber allocate(RubyClass rubyClass) {
                     getContext(),
                     getLanguage(),
                     thread,
+                    FiberStatus.CREATED,
                     "<uninitialized>");
             AllocationTracing.trace(fiber, this);
             return fiber;
@@ -121,6 +122,13 @@ public abstract static class InitializeNode extends PrimitiveArrayArgumentsNode
         @TruffleBoundary
         @Specialization
         protected Object initialize(RubyFiber fiber, boolean blocking, RubyProc block) {
+            if (!getContext().getEnv().isCreateThreadAllowed()) {
+                // Because TruffleThreadBuilder#build denies it already, before the thread is even started.
+                // The permission is called allowCreateThread, so it kind of makes sense.
+                throw new RaiseException(getContext(),
+                        coreExceptions().securityError("fibers not allowed with allowCreateThread(false)", this));
+            }
+
             getContext().fiberManager.initialize(fiber, blocking, block, this);
             return nil;
         }