Inline the state of FiberManager in RubyThread to remove one indirection

Author: eregon

src/main/java/org/truffleruby/RubyContext.java

@@ -47,6 +47,7 @@
 import org.truffleruby.core.array.RubyArray;
 import org.truffleruby.core.encoding.EncodingManager;
 import org.truffleruby.core.exception.CoreExceptions;
+import org.truffleruby.core.fiber.FiberManager;
 import org.truffleruby.core.hash.PreInitializationManager;
 import org.truffleruby.core.hash.ReHashable;
 import org.truffleruby.core.inlined.CoreMethods;
@@ -142,6 +143,7 @@ public class RubyContext {
     private final CoreLibrary coreLibrary;
     @CompilationFinal private CoreMethods coreMethods;
     private final ThreadManager threadManager;
+    public final FiberManager fiberManager;
     private final LexicalScope rootLexicalScope;
     private volatile ConsoleHolder consoleHolder;
 
@@ -226,6 +228,7 @@ public RubyContext(RubyLanguage language, TruffleLanguage.Env env) {
 
         Metrics.printTime("before-thread-manager");
         threadManager = new ThreadManager(this, language);
+        fiberManager = new FiberManager(language, this);
         threadManager.initialize();
         threadManager.initializeMainThread(Thread.currentThread());
         Metrics.printTime("after-thread-manager");

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

@@ -9,8 +9,6 @@
  */
 package org.truffleruby.core.fiber;
 
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 
 import com.oracle.truffle.api.TruffleContext;
@@ -44,54 +42,22 @@
 import com.oracle.truffle.api.object.Shape;
 import com.oracle.truffle.api.profiles.BranchProfile;
 import com.oracle.truffle.api.source.SourceSection;
-import org.truffleruby.language.objects.ObjectGraphNode;
 import org.truffleruby.language.objects.shared.SharedObjects;
 
-/** Manages Ruby {@code Fiber} objects for a given Ruby thread. */
-public class FiberManager implements ObjectGraphNode {
+/** Helps managing Ruby {@code Fiber} objects. Only one per {@link RubyContext}. */
+public class FiberManager {
 
     public static final String NAME_PREFIX = "Ruby Fiber";
 
     private final RubyLanguage language;
     private final RubyContext context;
-    private final RubyFiber rootFiber;
-    private RubyFiber currentFiber;
-    private final Set<RubyFiber> runningFibers = newFiberSet();
 
-    public FiberManager(RubyLanguage language, RubyContext context, RubyThread rubyThread) {
+    public FiberManager(RubyLanguage language, RubyContext context) {
         this.language = language;
         this.context = context;
-        this.rootFiber = createRootFiber(language, context, rubyThread);
-        this.currentFiber = rootFiber;
     }
 
-    @TruffleBoundary
-    private static Set<RubyFiber> newFiberSet() {
-        return ConcurrentHashMap.newKeySet();
-    }
-
-    public RubyFiber getRootFiber() {
-        return rootFiber;
-    }
-
-    public RubyFiber getCurrentFiber() {
-        assert language
-                .getCurrentThread().fiberManager == this : "Trying to read the current Fiber of another Thread which is inherently racy";
-        return currentFiber;
-    }
-
-    // If the currentFiber is read from another Ruby Thread,
-    // there is no guarantee that fiber will remain the current one
-    // as it could switch to another Fiber before the actual operation on the returned fiber.
-    public RubyFiber getCurrentFiberRacy() {
-        return currentFiber;
-    }
-
-    private void setCurrentFiber(RubyFiber fiber) {
-        currentFiber = fiber;
-    }
-
-    private RubyFiber createRootFiber(RubyLanguage language, RubyContext context, RubyThread thread) {
+    public static RubyFiber createRootFiber(RubyLanguage language, RubyContext context, RubyThread thread) {
         return createFiber(
                 language,
                 context,
@@ -101,8 +67,8 @@ private RubyFiber createRootFiber(RubyLanguage language, RubyContext context, Ru
                 "root");
     }
 
-    public RubyFiber createFiber(RubyLanguage language, RubyContext context, RubyThread thread, RubyClass rubyClass,
-            Shape shape, String sourceLocation) {
+    public static RubyFiber createFiber(RubyLanguage language, RubyContext context, RubyThread thread,
+            RubyClass rubyClass, Shape shape, String sourceLocation) {
         CompilerAsserts.partialEvaluationConstant(language);
         final RubyBasicObject fiberLocals = new RubyBasicObject(
                 context.getCoreLibrary().objectClass,
@@ -149,7 +115,7 @@ 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) {
-        assert fiber != rootFiber : "Root Fibers execute threadMain() and not fiberMain()";
+        assert !fiber.isRootFiber() : "Root Fibers execute threadMain() and not fiberMain()";
 
         final boolean entered = !context.getOptions().FIBER_LEAVE_CONTEXT;
         assert entered == context.getEnv().getContext().isEntered();
@@ -220,8 +186,9 @@ private void fiberMain(RubyContext context, RubyFiber fiber, RubyProc block, Nod
     }
 
     public RubyFiber getReturnFiber(RubyFiber currentFiber, Node currentNode, BranchProfile errorProfile) {
-        assert currentFiber == this.currentFiber;
+        assert currentFiber == currentFiber.rubyThread.getCurrentFiber();
 
+        final RubyFiber rootFiber = currentFiber.rubyThread.getRootFiber();
         if (currentFiber == rootFiber) {
             errorProfile.enter();
             throw new RaiseException(context, context.getCoreExceptions().yieldFromRootFiberError(currentNode));
@@ -270,7 +237,7 @@ private void assertNotEntered(String reason) {
 
     @TruffleBoundary
     private Object[] handleMessage(RubyFiber fiber, FiberMessage message, Node currentNode) {
-        setCurrentFiber(fiber);
+        fiber.rubyThread.setCurrentFiber(fiber);
 
         if (message instanceof FiberShutdownMessage) {
             throw new FiberShutdownException(currentNode);
@@ -329,7 +296,7 @@ public void start(RubyFiber fiber, Thread javaThread) {
 
         // share RubyFiber as its fiberLocals might be accessed by other threads with Thread#[]
         SharedObjects.propagate(language, rubyThread, fiber);
-        runningFibers.add(fiber);
+        rubyThread.runningFibers.add(fiber);
     }
 
     public void cleanup(RubyFiber fiber, Thread javaThread) {
@@ -339,7 +306,7 @@ public void cleanup(RubyFiber fiber, Thread javaThread) {
 
         threadManager.cleanupValuesForJavaThread(javaThread);
 
-        runningFibers.remove(fiber);
+        fiber.rubyThread.runningFibers.remove(fiber);
 
         fiber.thread = null;
 
@@ -352,7 +319,7 @@ public void cleanup(RubyFiber fiber, Thread javaThread) {
     }
 
     @TruffleBoundary
-    public void killOtherFibers() {
+    public void killOtherFibers(RubyThread thread) {
         // 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.
@@ -366,17 +333,17 @@ public void killOtherFibers() {
         try {
             final TruffleContext truffleContext = context.getEnv().getContext();
             context.getThreadManager().leaveAndEnter(truffleContext, DummyNode.INSTANCE, () -> {
-                doKillOtherFibers();
+                doKillOtherFibers(thread);
                 return BlockingAction.SUCCESS;
             });
         } finally {
             safepoint.setAllowSideEffects(allowSideEffects);
         }
     }
 
-    private void doKillOtherFibers() {
-        for (RubyFiber fiber : runningFibers) {
-            if (fiber != rootFiber) {
+    private void doKillOtherFibers(RubyThread thread) {
+        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
@@ -394,16 +361,10 @@ private void doKillOtherFibers() {
         }
     }
 
-    @Override
-    public void getAdjacentObjects(Set<Object> reachable) {
-        // share fibers of a thread as its fiberLocals might be accessed by other threads with Thread#[]
-        reachable.addAll(runningFibers);
-    }
-
-    public String getFiberDebugInfo() {
+    public String getFiberDebugInfo(RubyThread rubyThread) {
         final StringBuilder builder = new StringBuilder();
 
-        for (RubyFiber fiber : runningFibers) {
+        for (RubyFiber fiber : rubyThread.runningFibers) {
             builder.append("  fiber @");
             builder.append(ObjectIDNode.getUncached().execute(fiber));
 
@@ -414,11 +375,11 @@ public String getFiberDebugInfo() {
                 builder.append(" #").append(thread.getId()).append(' ').append(thread);
             }
 
-            if (fiber == rootFiber) {
+            if (fiber.isRootFiber()) {
                 builder.append(" (root)");
             }
 
-            if (fiber == currentFiber) {
+            if (fiber == fiber.rubyThread.getCurrentFiber()) {
                 builder.append(" (current)");
             }
 

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

@@ -79,8 +79,7 @@ protected Object transfer(
                         coreExceptions().fiberError("fiber called across threads", this));
             }
 
-            final FiberManager fiberManager = currentThread.fiberManager;
-            return singleValue(fiberManager.transferControlTo(currentFiber, fiber, operation, args, this));
+            return singleValue(getContext().fiberManager.transferControlTo(currentFiber, fiber, operation, args, this));
         }
 
     }
@@ -95,14 +94,13 @@ protected RubyFiber allocate(RubyClass rubyClass) {
             }
 
             final RubyThread thread = getLanguage().getCurrentThread();
-            final RubyFiber fiber = thread.fiberManager
-                    .createFiber(
-                            getLanguage(),
-                            getContext(),
-                            thread,
-                            rubyClass,
-                            getLanguage().fiberShape,
-                            "<uninitialized>");
+            final RubyFiber fiber = FiberManager.createFiber(
+                    getLanguage(),
+                    getContext(),
+                    thread,
+                    rubyClass,
+                    getLanguage().fiberShape,
+                    "<uninitialized>");
             AllocationTracing.trace(fiber, this);
             return fiber;
         }
@@ -115,7 +113,7 @@ public abstract static class InitializeNode extends CoreMethodArrayArgumentsNode
         @Specialization
         protected Object initialize(RubyFiber fiber, RubyProc block) {
             final RubyThread thread = getLanguage().getCurrentThread();
-            thread.fiberManager.initialize(fiber, block, this);
+            getContext().fiberManager.initialize(fiber, block, this);
             return nil;
         }
 
@@ -138,8 +136,7 @@ protected Object resume(RubyFiber fiber, Object[] args,
             fiber.transferred = true;
 
             final RubyThread currentThread = getLanguage().getCurrentThread();
-            final FiberManager fiberManager = currentThread.fiberManager;
-            final RubyFiber currentFiber = fiberManager.getCurrentFiber();
+            final RubyFiber currentFiber = currentThread.getCurrentFiber();
 
             if (sameFiberProfile.profile(currentFiber == fiber)) {
                 // A Fiber can transfer to itself
@@ -169,9 +166,8 @@ protected Object resume(FiberOperation operation, RubyFiber fiber, Object[] args
                 @Cached ConditionProfile transferredProfile) {
 
             final RubyFiber parentFiber = fiber.lastResumedByFiber;
-            final FiberManager fiberToResumeManager = fiber.rubyThread.fiberManager;
 
-            if (doubleResumeProfile.profile(parentFiber != null || fiber == fiberToResumeManager.getRootFiber())) {
+            if (doubleResumeProfile.profile(parentFiber != null || fiber.isRootFiber())) {
                 throw new RaiseException(getContext(), coreExceptions().fiberError("double resume", this));
             }
 
@@ -182,8 +178,7 @@ protected Object resume(FiberOperation operation, RubyFiber fiber, Object[] args
             }
 
             final RubyThread currentThread = getLanguage().getCurrentThread();
-            final FiberManager fiberManager = currentThread.fiberManager;
-            final RubyFiber currentFiber = fiberManager.getCurrentFiber();
+            final RubyFiber currentFiber = currentThread.getCurrentFiber();
 
             return fiberTransferNode
                     .executeTransferControlTo(currentThread, currentFiber, fiber, operation, args);
@@ -233,10 +228,10 @@ protected Object fiberYield(Object[] args,
                 @Cached BranchProfile errorProfile) {
 
             final RubyThread currentThread = getLanguage().getCurrentThread();
-            final FiberManager fiberManager = currentThread.fiberManager;
-            final RubyFiber currentFiber = fiberManager.getCurrentFiber();
+            final RubyFiber currentFiber = currentThread.getCurrentFiber();
 
-            final RubyFiber fiberYieldedTo = fiberManager.getReturnFiber(currentFiber, this, errorProfile);
+            final RubyFiber fiberYieldedTo = getContext().fiberManager
+                    .getReturnFiber(currentFiber, this, errorProfile);
 
             return fiberTransferNode.executeTransferControlTo(
                     currentThread,
@@ -263,8 +258,7 @@ public abstract static class CurrentNode extends CoreMethodNode {
 
         @Specialization
         protected RubyFiber current() {
-            final RubyThread currentThread = getLanguage().getCurrentThread();
-            return currentThread.fiberManager.getCurrentFiber();
+            return getLanguage().getCurrentThread().getCurrentFiber();
         }
 
     }
@@ -300,8 +294,7 @@ public abstract static class FiberGetCatchTagsNode extends PrimitiveArrayArgumen
 
         @Specialization
         protected RubyArray getCatchTags() {
-            final RubyThread currentThread = getLanguage().getCurrentThread();
-            final RubyFiber currentFiber = currentThread.fiberManager.getCurrentFiber();
+            final RubyFiber currentFiber = getLanguage().getCurrentThread().getCurrentFiber();
             return currentFiber.catchTags;
         }
     }

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

@@ -24,7 +24,7 @@
 import com.oracle.truffle.api.CompilerDirectives.TruffleBoundary;
 import com.oracle.truffle.api.object.Shape;
 
-public class RubyFiber extends RubyDynamicObject implements ObjectGraphNode {
+public final class RubyFiber extends RubyDynamicObject implements ObjectGraphNode {
 
     public final RubyBasicObject fiberLocals;
     public final RubyArray catchTags;
@@ -55,22 +55,15 @@ public RubyFiber(
         this.sourceLocation = sourceLocation;
     }
 
-    @TruffleBoundary
-    private static LinkedBlockingQueue<FiberManager.FiberMessage> newMessageQueue() {
-        return new LinkedBlockingQueue<>();
-    }
-
-    @Override
-    public void getAdjacentObjects(Set<Object> reachable) {
-        reachable.add(fiberLocals);
-        reachable.add(rubyThread);
+    public boolean isRootFiber() {
+        return rubyThread.getRootFiber() == this;
     }
 
     public String getStatus() {
         if (!resumed) {
             return "created";
         } else if (alive) {
-            if (rubyThread.fiberManager.getCurrentFiberRacy() == this) {
+            if (rubyThread.getCurrentFiberRacy() == this) {
                 return "resumed";
             } else {
                 return "suspended";
@@ -80,4 +73,15 @@ public String getStatus() {
         }
     }
 
+    @TruffleBoundary
+    private static LinkedBlockingQueue<FiberManager.FiberMessage> newMessageQueue() {
+        return new LinkedBlockingQueue<>();
+    }
+
+    @Override
+    public void getAdjacentObjects(Set<Object> reachable) {
+        reachable.add(fiberLocals);
+        reachable.add(rubyThread);
+    }
+
 }