[GR-33075] Inline the state of FiberManager in RubyThread to remove one indirection

PullRequest: truffleruby/2869

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,22 +9,16 @@
  */
 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;
 import com.oracle.truffle.api.TruffleSafepoint;
 import org.truffleruby.RubyContext;
 import org.truffleruby.RubyLanguage;
 import org.truffleruby.core.DummyNode;
-import org.truffleruby.core.array.ArrayHelpers;
-import org.truffleruby.core.array.RubyArray;
 import org.truffleruby.core.basicobject.BasicObjectNodes.ObjectIDNode;
-import org.truffleruby.core.basicobject.RubyBasicObject;
 import org.truffleruby.core.exception.ExceptionOperations;
 import org.truffleruby.core.exception.RubyException;
-import org.truffleruby.core.klass.RubyClass;
 import org.truffleruby.core.proc.ProcOperations;
 import org.truffleruby.core.thread.RubyThread;
 import org.truffleruby.core.proc.RubyProc;
@@ -37,79 +31,24 @@
 import org.truffleruby.language.control.RaiseException;
 import org.truffleruby.language.control.TerminationException;
 
-import com.oracle.truffle.api.CompilerAsserts;
 import com.oracle.truffle.api.CompilerDirectives;
 import com.oracle.truffle.api.CompilerDirectives.TruffleBoundary;
 import com.oracle.truffle.api.nodes.Node;
-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) {
-        return createFiber(
-                language,
-                context,
-                thread,
-                context.getCoreLibrary().fiberClass,
-                language.fiberShape,
-                "root");
-    }
-
-    public 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,
-                language.basicObjectShape);
-        final RubyArray catchTags = ArrayHelpers.createEmptyArray(context, language);
-
-        return new RubyFiber(rubyClass, shape, fiberLocals, catchTags, thread, sourceLocation);
     }
 
     public void initialize(RubyFiber fiber, RubyProc block, Node currentNode) {
@@ -149,7 +88,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 +159,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 +210,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);
@@ -318,9 +258,8 @@ public void start(RubyFiber fiber, Thread javaThread) {
 
         if (Thread.currentThread() == javaThread) {
             context.getThreadManager().rubyFiber.set(fiber);
-        } else {
-            context.getThreadManager().javaThreadToRubyFiber.put(javaThread, fiber);
         }
+        context.getThreadManager().javaThreadToRubyFiber.put(javaThread, fiber);
 
         fiber.thread = javaThread;
 
@@ -329,7 +268,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 +278,7 @@ public void cleanup(RubyFiber fiber, Thread javaThread) {
 
         threadManager.cleanupValuesForJavaThread(javaThread);
 
-        runningFibers.remove(fiber);
+        fiber.rubyThread.runningFibers.remove(fiber);
 
         fiber.thread = null;
 
@@ -352,7 +291,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 +305,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 +333,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 +347,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 = new RubyFiber(
+                    rubyClass,
+                    getLanguage().fiberShape,
+                    getContext(),
+                    getLanguage(),
+                    thread,
+                    "<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;
         }
     }