[GR-54695] Fix setPendingFlagForVirtualThread() and cleanup TruffleSafepointTest

PullRequest: graal/18049

Author: eregon

compiler/src/jdk.graal.compiler/src/jdk/graal/compiler/debug/ScopeImpl.java

@@ -150,7 +150,16 @@ private boolean isEmptyScope() {
     private boolean interceptDisabled;
 
     ScopeImpl(DebugContext owner, Thread thread, boolean interceptDisabled) {
-        this(owner, thread.getName(), null, false, interceptDisabled);
+        this(owner, getThreadName(thread), null, false, interceptDisabled);
+    }
+
+    private static String getThreadName(Thread thread) {
+        String name = thread.getName();
+        if (name.isEmpty()) { // the default for virtual threads
+            return thread.toString();
+        } else {
+            return name;
+        }
     }
 
     private ScopeImpl(DebugContext owner, String unqualifiedName, ScopeImpl parent, boolean sandbox, boolean interceptDisabled, Object... context) {

truffle/src/com.oracle.truffle.api.test/src/com/oracle/truffle/api/test/TruffleSafepointTest.java

@@ -67,6 +67,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -170,7 +171,15 @@ public static void afterClass() throws InterruptedException {
     @Before
     public void before() throws ExecutionException, InterruptedException {
         Assume.assumeFalse(vthreads && !canCreateVirtualThreads());
-        this.service = vthreads ? Executors.newVirtualThreadPerTaskExecutor() : cachedThreadPool;
+        if (vthreads) {
+            ThreadFactory factory = Thread.ofVirtual().uncaughtExceptionHandler((thread, exception) -> {
+                System.err.println("Uncaught exception in " + thread);
+                exception.printStackTrace(System.err);
+            }).factory();
+            this.service = Executors.newThreadPerTaskExecutor(factory);
+        } else {
+            this.service = cachedThreadPool;
+        }
 
         ProxyLanguage.setDelegate(new ProxyLanguage() {
             @Override
@@ -679,9 +688,20 @@ protected void perform(Access access) {
 
     @TruffleBoundary
     private static void sleepNanosBoundary(int nanos) {
-        try {
-            Thread.sleep(0, nanos);
-        } catch (InterruptedException ie) {
+        long deadline = System.nanoTime() + nanos;
+        long now;
+        boolean interrupted = false;
+        while ((now = System.nanoTime()) < deadline) {
+            try {
+                Thread.sleep(0, (int) (deadline - now));
+                break;
+            } catch (InterruptedException e) {
+                // cancellation uses Thread#interrupt()
+                interrupted = true;
+            }
+        }
+        if (interrupted) {
+            Thread.currentThread().interrupt();
         }
     }
 
@@ -1230,6 +1250,11 @@ public void testContextSafepoint() {
                     contextSafepoint();
                     reschedule();
                 }
+                /*
+                 * There is a poll() after this while loop, on the return from the CallTarget. That
+                 * ensures all ThreadLocalActions are processed without needing to wait for their
+                 * futures. This also helped to find some bugs.
+                 */
                 return true;
             })) {
                 AtomicInteger eventCounter = new AtomicInteger();
@@ -1697,12 +1722,16 @@ void forEachConfig(TestRunner run) {
         }
 
         // asynchronous execution of all configs
-        if (RERUN_THREAD_CONFIG_ASYNC) {
+        /*
+         * this is not done for virtual threads because it would cause to run more virtual threads
+         * concurrently than the number of processors (see above).
+         */
+        if (RERUN_THREAD_CONFIG_ASYNC && !vthreads) {
             List<Future<?>> futures = new ArrayList<>();
             for (int threads : threadConfigs) {
                 for (int events : ITERATION_CONFIGS) {
-                    if (threads * events <= MAX_THREAD_ITERATIONS_PRODUCT && !(vthreads && threads > processors)) {
-                        futures.add(service.submit(() -> run.run(threads, events)));
+                    if (threads * events <= MAX_THREAD_ITERATIONS_PRODUCT) {
+                        futures.add(cachedThreadPool.submit(() -> run.run(threads, events)));
                     }
                 }
             }
@@ -1931,13 +1960,20 @@ protected Context createTestContext() {
     @TruffleBoundary
     private static void waitForLatch(CountDownLatch latch) throws AssertionError {
         latch.countDown();
-        try {
-            latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
+        boolean interrupted = false;
+        while (true) {
+            try {
+                if (!latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+                    throw failTimeout("waitForLatch()", null);
+                }
+                break;
+            } catch (InterruptedException e) {
+                // cancellation uses Thread#interrupt()
+                interrupted = true;
+            }
         }
-        try {
-            latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
+        if (interrupted) {
+            Thread.currentThread().interrupt();
         }
     }
 

truffle/src/com.oracle.truffle.runtime/src/com/oracle/truffle/runtime/hotspot/HotSpotThreadLocalHandshake.java

@@ -92,11 +92,6 @@ static void doHandshake(Object node) {
         SINGLETON.processHandshake((Node) node);
     }
 
-    @Override
-    protected void setFastPending(Thread t) {
-        setVolatile(t, 1);
-    }
-
     @Override
     @TruffleBoundary
     public TruffleSafepointImpl getCurrent() {
@@ -110,21 +105,24 @@ public TruffleSafepointImpl getCurrent() {
 
     @Override
     protected void clearFastPending() {
-        setVolatile(Thread.currentThread(), 0);
+        Thread carrierThread = JAVA_LANG_ACCESS.currentCarrierThread();
+        long eetop = UNSAFE.getLongVolatile(carrierThread, THREAD_EETOP_OFFSET);
+        UNSAFE.putIntVolatile(null, eetop + PENDING_OFFSET, 0);
     }
 
-    private static void setVolatile(Thread thread, int value) {
+    @Override
+    protected void setFastPending(Thread thread) {
         /*
          * The thread will not go away here because the Truffle implementation ensures that this
          * method is no longer used if the thread is no longer active. It only sets this state for
          * contexts that are currently entered on a thread. Being entered implies that the thread is
          * active.
          */
-        assert thread.isAlive() : "thread must remain alive while setting fast pending";
+        assert thread.isAlive() : "thread must remain alive while setting the pending flag";
 
         long eetop = UNSAFE.getLongVolatile(thread, THREAD_EETOP_OFFSET);
         if (eetop != 0) {
-            UNSAFE.putIntVolatile(null, eetop + PENDING_OFFSET, value);
+            UNSAFE.putIntVolatile(null, eetop + PENDING_OFFSET, 1);
         } else { // only the case for VirtualThreads
             Object carrierThread = UNSAFE.getObjectVolatile(thread, THREAD_CARRIER_THREAD_OFFSET);
             if (carrierThread == null) {
@@ -137,7 +135,7 @@ private static void setVolatile(Thread thread, int value) {
                 return;
             }
             eetop = UNSAFE.getLongVolatile(carrierThread, THREAD_EETOP_OFFSET);
-            UNSAFE.putIntVolatile(null, eetop + PENDING_OFFSET, value);
+            UNSAFE.putIntVolatile(null, eetop + PENDING_OFFSET, 1);
             /*
              * If the VirtualThread moves to another carrier thread after this method returns, the
              * pending flag will still be set correctly thanks to setPendingFlagForVirtualThread()
@@ -150,15 +148,44 @@ private static void setVolatile(Thread thread, int value) {
     static void setPendingFlagForVirtualThread() {
         TruffleSafepointImpl safepoint = STATE.get();
         if (safepoint != null) {
-            boolean pending = safepoint.isFastPendingSet();
-
-            // VirtualThread#carrierThread is not set yet, it set after this hook is called.
-            // However, Thread.currentCarrierThread() is already set so we can use that.
-            // We could also get the carrier thread from the hook arguments but that seems more
-            // expensive.
             Thread carrierThread = JAVA_LANG_ACCESS.currentCarrierThread();
             long eetop = UNSAFE.getLongVolatile(carrierThread, THREAD_EETOP_OFFSET);
-            UNSAFE.putIntVolatile(null, eetop + PENDING_OFFSET, pending ? 1 : 0);
+
+            /*
+             * When this method and setFastPending() are run concurrently, there is a possibility
+             * that we set the carrier pending flag to 0 here based on pendingBefore==false, after
+             * setFastPending() sets the flag to 1 (which would lose the flag):
+             *
+             * @formatter:off
+             * with VT=VirtualThread MT=the thread that submits the ThreadLocalAction:
+             * VT: boolean pending = this.fastPendingSet; // false
+             * MT: fastPendingSet = true;
+             * MT: carrierThread.pending = true;
+             * VT: carrierThread.pending = pending; // false
+             * @formatter:on
+             *
+             * So we check for that case with pendingAfter and fix it. This is correct because
+             * either we wrote the 0 before setFastPending() wrote the 1 (harmless), or we wrote it
+             * after and then we must see pendingAfter==true, because (the caller of)
+             * setFastPending() sets fastPendingSet before writing to the carrier pending flag.
+             *
+             * If this method and setFastPending() are not run concurrently, then it is as if both
+             * methods were executed one after another (since all accesses are volatile and so
+             * sequentially consistent) and it works trivially.
+             *
+             * This solution is better than `if (isFastPendingSet()) { carrierThread.pending = 1; }`
+             * because that can leave the carrier flag as true after an unmount of a virtual thread
+             * which did not process safepoints yet and the newly-mounted virtual thread would not
+             * clear it, causing extra stub calls (clearing is only done if fastPendingSet is true,
+             * necessary for DefaultThreadLocalHandshake where clearFastPending() is not
+             * idempotent).
+             */
+            boolean pendingBefore = safepoint.isFastPendingSet();
+            UNSAFE.putIntVolatile(null, eetop + PENDING_OFFSET, pendingBefore ? 1 : 0);
+            boolean pendingAfter = safepoint.isFastPendingSet();
+            if (!pendingBefore && pendingAfter) {
+                UNSAFE.putIntVolatile(null, eetop + PENDING_OFFSET, 1);
+            }
         }
     }