From dee015e7b10a7383b5ee933b27d85a27010524dc Mon Sep 17 00:00:00 2001 From: Florian Angerer Date: Thu, 27 Aug 2026 15:24:45 +0200 Subject: [PATCH 1/6] Flush standard streams before C API teardown --- .../src/com/oracle/graal/python/runtime/PythonContext.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/PythonContext.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/PythonContext.java index edae82b108..f3cce41c0a 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/PythonContext.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/PythonContext.java @@ -1769,12 +1769,14 @@ public void finalizeContext() { // shut down async actions threads handler.shutdown(); finalizing = true; + // Flushing may execute arbitrary guest code, including C extension code. Do it before + // finalizeCApi tears down native wrappers and other C API state. + stdioFlushFailed = flushStdFiles(); if (cApiContext != null) { cApiContext.finalizeCApi(cancelling); } // interrupt and join or kill python threads joinPythonThreads(); - stdioFlushFailed = flushStdFiles(); if (nativeContext != null) { nativeContext.close(); } From 978b3b9db8febdd2760f27c1bbebaa4bc31c06f5 Mon Sep 17 00:00:00 2001 From: Florian Angerer Date: Thu, 27 Aug 2026 15:49:06 +0200 Subject: [PATCH 2/6] Mark native C API finalizing before teardown --- .../python/builtins/objects/cext/capi/CApiContext.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java index a42887eb29..2b07fef10e 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java @@ -1174,6 +1174,14 @@ public void finalizeCApi(boolean cancelling) { CompilerAsserts.neverPartOfCompilation(); PythonContext context = getContext(); HandleContext handleContext = context.handleContext; + /* + * Cancellation skips the context's atexit hooks, so mark the C API as finalizing here as + * well. This must happen before any native wrappers are freed while other threads may still + * be running native code. + */ + if (nativeFinalizerRunnable != null) { + nativeFinalizerRunnable.run(); + } if (backgroundGCTaskThread != null && backgroundGCTaskThread.isAlive()) { context.killSystemThread(backgroundGCTaskThread); try { @@ -1230,7 +1238,6 @@ public void finalizeCApi(boolean cancelling) { if (nativeFinalizerShutdownHook != null) { try { Runtime.getRuntime().removeShutdownHook(nativeFinalizerShutdownHook); - nativeFinalizerRunnable.run(); } catch (IllegalStateException e) { // Shutdown already in progress, let it do the finalization then } From a4da902fbb4d3fde51d392d337b2b1c9f4db1556 Mon Sep 17 00:00:00 2001 From: Florian Angerer Date: Thu, 27 Aug 2026 18:31:32 +0200 Subject: [PATCH 3/6] Log NativeLibrary unloading --- .../graal/python/runtime/nativeaccess/NativeContext.java | 7 +++++-- .../graal/python/runtime/nativeaccess/NativeLibrary.java | 9 ++++++++- 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/nativeaccess/NativeContext.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/nativeaccess/NativeContext.java index 63206f5df1..bcb2fdee5a 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/nativeaccess/NativeContext.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/nativeaccess/NativeContext.java @@ -53,9 +53,11 @@ import com.oracle.truffle.api.CompilerAsserts; import com.oracle.truffle.api.CompilerDirectives; import com.oracle.truffle.api.CompilerDirectives.TruffleBoundary; +import com.oracle.truffle.api.TruffleLogger; public final class NativeContext { public static final String UNAVAILABLE = "JEP 454 is not included on this JDK, this prevents loading native extensions modules."; + static final TruffleLogger LOGGER = PythonLanguage.getLogger(NativeContext.class); private static final int LOAD_LIBRARY_SEARCH_DLL_LOAD_DIR = 0x00000100; private static final int LOAD_LIBRARY_SEARCH_APPLICATION_DIR = 0x00000200; @@ -81,13 +83,14 @@ public static NativeContext create() { @TruffleBoundary NativeContext() { arena = NativeAccessSupport.createArena(); - defaultLibrary = isWindows() ? null : new NativeLibrary(this, getPosixDefaultLibraryHandle()); + defaultLibrary = isWindows() ? null : new NativeLibrary(this, "DEFAULT", getPosixDefaultLibraryHandle()); callState = ThreadLocal.withInitial(() -> NativeAccessSupport.createCapturedCallState(arena)); } public void close() { CompilerAsserts.neverPartOfCompilation(); for (NativeLibrary library : libraries) { + LOGGER.fine(() -> "Closing " + library); int result; try { result = isWindows() ? (int) FREE_LIBRARY.invokeExact(freeLibraryPtr, library.ptr) : (int) DLCLOSE.invokeExact(dlclosePtr, library.ptr); @@ -129,7 +132,7 @@ public NativeLibrary loadLibrary(String name, int flags) throws NativeLibraryLoa if (lib == 0) { throw createLoadLibraryException(isWindows() ? getLastError() : 0); } - NativeLibrary library = new NativeLibrary(this, lib); + NativeLibrary library = new NativeLibrary(this, name, lib); libraries.add(library); return library; } diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/nativeaccess/NativeLibrary.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/nativeaccess/NativeLibrary.java index 996ea67ad3..8f5938869c 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/nativeaccess/NativeLibrary.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/nativeaccess/NativeLibrary.java @@ -45,10 +45,12 @@ public final class NativeLibrary { private final NativeContext context; + private final String name; // for debugging final long ptr; - NativeLibrary(NativeContext context, long ptr) { + NativeLibrary(NativeContext context, String name, long ptr) { this.context = context; + this.name = name; this.ptr = ptr; } @@ -63,4 +65,9 @@ public long lookupSymbol(String name) { public long lookupOptionalSymbol(String name) { return context.lookupOptionalSymbol(ptr, name); } + + @Override + public String toString() { + return String.format("NativeLibrary(%s, 0x%x)", name, ptr); + } } From 0c413f527704686e08db82d6843653760f272f8e Mon Sep 17 00:00:00 2001 From: Florian Angerer Date: Thu, 27 Aug 2026 19:44:37 +0200 Subject: [PATCH 4/6] Implement refq watcher thread --- .../cext/ReferenceQueueCoordinatorTests.java | 195 +++++++++ .../oracle/graal/python/PythonLanguage.java | 8 + .../builtins/modules/GcModuleBuiltins.java | 2 +- .../modules/cext/PythonCextBuiltins.java | 4 +- .../objects/cext/capi/CApiContext.java | 63 ++- .../builtins/objects/cext/capi/CExtNodes.java | 2 +- .../capi/transitions/CApiTransitions.java | 384 +++++++++++------- .../ReferenceQueueCoordinator.java | 349 ++++++++++++++++ 8 files changed, 859 insertions(+), 148 deletions(-) create mode 100644 graalpython/com.oracle.graal.python.test/src/com/oracle/graal/python/test/builtin/objects/cext/ReferenceQueueCoordinatorTests.java create mode 100644 graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/ReferenceQueueCoordinator.java diff --git a/graalpython/com.oracle.graal.python.test/src/com/oracle/graal/python/test/builtin/objects/cext/ReferenceQueueCoordinatorTests.java b/graalpython/com.oracle.graal.python.test/src/com/oracle/graal/python/test/builtin/objects/cext/ReferenceQueueCoordinatorTests.java new file mode 100644 index 0000000000..cd97fcd213 --- /dev/null +++ b/graalpython/com.oracle.graal.python.test/src/com/oracle/graal/python/test/builtin/objects/cext/ReferenceQueueCoordinatorTests.java @@ -0,0 +1,195 @@ +/* + * Copyright (c) 2026, Oracle and/or its affiliates. All rights reserved. + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. + * + * The Universal Permissive License (UPL), Version 1.0 + * + * Subject to the condition set forth below, permission is hereby granted to any + * person obtaining a copy of this software, associated documentation and/or + * data (collectively the "Software"), free of charge and under any and all + * copyright rights in the Software, and any and all patent rights owned or + * freely licensable by each licensor hereunder covering either (i) the + * unmodified Software as contributed to or provided by such licensor, or (ii) + * the Larger Works (as defined below), to deal in both + * + * (a) the Software, and + * + * (b) any piece of software and/or hardware listed in the lrgrwrks.txt file if + * one is included with the Software each a "Larger Work" to which the Software + * is contributed by such licensors), + * + * without restriction, including without limitation the rights to copy, create + * derivative works of, display, perform, and distribute the Software and make, + * use, sell, offer for sale, import, export, have made, and have sold the + * Software and the Larger Work(s), and to sublicense the foregoing rights on + * either these or other terms. + * + * This license is subject to the following condition: + * + * The above copyright notice and either this complete permission notice or at a + * minimum a reference to the UPL must be included in all copies or substantial + * portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ +package com.oracle.graal.python.test.builtin.objects.cext; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import java.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Test; + +import com.oracle.graal.python.builtins.objects.cext.capi.transitions.ReferenceQueueCoordinator; + +public class ReferenceQueueCoordinatorTests { + + @Test + public void watcherHandsOffExactlyOneReference() { + ReferenceQueueCoordinator coordinator = startedCoordinator(); + Reference reference = new WeakReference<>(new Object()); + + assertTrue(coordinator.beginWatcherRemove()); + Reference handoff = coordinator.endWatcherRemove(reference); + + assertSame(reference, coordinator.beginWatcherDrain(handoff)); + coordinator.finishDrain(); + + assertTrue(coordinator.beginWatcherRemove()); + assertNull(coordinator.endWatcherRemove(null)); + stop(coordinator); + } + + @Test + public void deferredWatcherDrainRetainsHandoff() { + ReferenceQueueCoordinator coordinator = startedCoordinator(); + Reference reference = new WeakReference<>(new Object()); + + assertTrue(coordinator.beginWatcherRemove()); + Reference handoff = coordinator.endWatcherRemove(reference); + + // A caller that cannot drain yet must be able to leave the handoff pending and retry later. + assertSame(reference, coordinator.getPendingReference()); + assertSame(reference, coordinator.beginWatcherDrain(handoff)); + coordinator.finishDrain(); + stop(coordinator); + } + + @Test + public void forcedDrainSupersedesQueuedHandoff() { + ReferenceQueueCoordinator coordinator = startedCoordinator(); + Reference reference = new WeakReference<>(new Object()); + + assertTrue(coordinator.beginWatcherRemove()); + Reference handoff = coordinator.endWatcherRemove(reference); + + assertSame(reference, coordinator.beginForcedDrain(null)); + assertNull(coordinator.beginWatcherDrain(handoff)); + coordinator.finishDrain(); + stop(coordinator); + } + + @Test + public void forcedDrainWithoutHandoffIsStillAcquired() { + ReferenceQueueCoordinator coordinator = startedCoordinator(); + + assertNotNull(coordinator.beginForcedDrain(null)); + coordinator.finishDrain(); + stop(coordinator); + } + + @Test + public void disabledCoordinatorPreservesHandoffUntilReenabled() { + ReferenceQueueCoordinator coordinator = startedCoordinator(); + Reference reference = new WeakReference<>(new Object()); + + assertTrue(coordinator.beginWatcherRemove()); + Reference handoff = coordinator.endWatcherRemove(reference); + + assertTrue(coordinator.disable(null)); + assertNull(coordinator.beginWatcherDrain(handoff)); + assertNull(coordinator.beginForcedDrain(null)); + + coordinator.enable(); + assertSame(reference, coordinator.beginWatcherDrain(handoff)); + coordinator.finishDrain(); + stop(coordinator); + } + + @Test + public void disabledCoordinatorResumesWatcherWhenReenabled() { + ReferenceQueueCoordinator coordinator = startedCoordinator(); + + assertTrue(coordinator.disable(null)); + coordinator.enable(); + assertTrue(coordinator.beginWatcherRemove()); + assertNull(coordinator.endWatcherRemove(null)); + stop(coordinator); + } + + @Test + public void stopInvalidatesPendingHandoff() { + ReferenceQueueCoordinator coordinator = startedCoordinator(); + Reference reference = new WeakReference<>(new Object()); + + assertTrue(coordinator.beginWatcherRemove()); + Reference handoff = coordinator.endWatcherRemove(reference); + coordinator.stop(); + + assertFalse(coordinator.isActive()); + assertNull(coordinator.beginWatcherDrain(handoff)); + } + + @Test + public void blockingRemoveIsInterruptibleForStop() throws InterruptedException { + ReferenceQueueCoordinator coordinator = new ReferenceQueueCoordinator(); + ReferenceQueue queue = new ReferenceQueue<>(); + CountDownLatch removing = new CountDownLatch(1); + AtomicReference> result = new AtomicReference<>(); + Thread watcher = new Thread(() -> { + if (coordinator.beginWatcherRemove()) { + removing.countDown(); + Reference reference = coordinator.blockingRemove(queue); + result.set(reference); + coordinator.endWatcherRemove(reference); + } + coordinator.stopped(); + }); + coordinator.start(watcher); + + assertTrue(removing.await(1, TimeUnit.SECONDS)); + coordinator.stop(); + + assertFalse(watcher.isAlive()); + assertNull(result.get()); + assertFalse(coordinator.isActive()); + } + + private static ReferenceQueueCoordinator startedCoordinator() { + ReferenceQueueCoordinator coordinator = new ReferenceQueueCoordinator(); + coordinator.start(new Thread(() -> { + })); + assertTrue(coordinator.isActive()); + return coordinator; + } + + private static void stop(ReferenceQueueCoordinator coordinator) { + coordinator.stop(); + assertFalse(coordinator.isActive()); + } +} diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/PythonLanguage.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/PythonLanguage.java index 8dd6298e85..2dc6d8242c 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/PythonLanguage.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/PythonLanguage.java @@ -433,6 +433,14 @@ protected void finalizeContext(PythonContext context) { boundaryCallDataMap.size(); } + @Override + protected void disposeContext(PythonContext context) { + if (context.getCApiContext() != null) { + context.getCApiContext().dispose(); + } + super.disposeContext(context); + } + @Override protected boolean areOptionsCompatible(OptionValues firstOptions, OptionValues newOptions) { return PythonOptions.areOptionsCompatible(firstOptions, newOptions); diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/modules/GcModuleBuiltins.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/modules/GcModuleBuiltins.java index 72c68d5540..cf994eb8b2 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/modules/GcModuleBuiltins.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/modules/GcModuleBuiltins.java @@ -211,7 +211,7 @@ static long javaCollect(Node inliningTarget, GilNode gil) { } // collect some weak references now PythonContext.triggerAsyncActions(inliningTarget); - CApiTransitions.pollReferenceQueue(); + CApiTransitions.pollReferenceQueueIfPending(PythonContext.get(null)); /* * CPython's GC returns the number of collected cycles. This is not something we can * determine, but to return some useful info to the Python program, we return the amount diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/modules/cext/PythonCextBuiltins.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/modules/cext/PythonCextBuiltins.java index 9fddad7abf..4bd4af4d7e 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/modules/cext/PythonCextBuiltins.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/modules/cext/PythonCextBuiltins.java @@ -698,7 +698,7 @@ public Object execute(Object[] arguments) { if (retNode != null) { result = retNode.execute(result); } - CApiTransitions.maybeGCALot(); + CApiTransitions.maybeGCALot(this); return result; } catch (Throwable t) { throw checkThrowableBeforeNative(t, "CApiBuiltin", self.name); @@ -1044,7 +1044,7 @@ static void GraalPyPrivate_TriggerGC(long delay) { // Restore interrupt status Thread.currentThread().interrupt(); } - CApiTransitions.pollReferenceQueue(); + CApiTransitions.pollReferenceQueueIfPending(PythonContext.get(null)); PythonContext.triggerAsyncActions(EncapsulatingNodeReference.getCurrent().get()); } diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java index 2b07fef10e..76b6b10c12 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java @@ -42,7 +42,7 @@ import static com.oracle.graal.python.PythonLanguage.CONTEXT_INSENSITIVE_SINGLETONS; import static com.oracle.graal.python.builtins.objects.PythonAbstractObject.UNINITIALIZED; -import static com.oracle.graal.python.builtins.objects.cext.capi.transitions.CApiTransitions.pollReferenceQueue; +import static com.oracle.graal.python.builtins.objects.cext.capi.transitions.CApiTransitions.drainReferenceQueueNow; import static com.oracle.graal.python.builtins.objects.cext.structs.CStructAccess.readIntField; import static com.oracle.graal.python.builtins.objects.cext.structs.CStructAccess.readLongField; import static com.oracle.graal.python.builtins.objects.object.PythonObject.IMMORTAL_REFCNT; @@ -59,6 +59,7 @@ import java.lang.invoke.MethodHandle; import java.lang.invoke.MethodHandles; import java.lang.invoke.MethodType; +import java.lang.ref.Reference; import java.lang.ref.WeakReference; import java.util.ArrayList; import java.util.Arrays; @@ -311,6 +312,7 @@ public TruffleString getInitFunctionName() { public final BackgroundGCTask gcTask; private Thread backgroundGCTaskThread; + private final ReferenceQueueWatcherTask referenceQueueWatcherTask; public static TruffleLogger getLogger(Class clazz) { return PythonLanguage.getLogger(LOGGER_CAPI_NAME + "." + clazz.getSimpleName()); @@ -326,6 +328,7 @@ public CApiContext(PythonContext context, NativeLibrary library, NativeLibraryLo Arrays.fill(singletonNativePtrs, UNINITIALIZED); this.gcTask = new BackgroundGCTask(context); + this.referenceQueueWatcherTask = new ReferenceQueueWatcherTask(context); } @TruffleBoundary @@ -695,6 +698,31 @@ private void perform() { } } + private static final class ReferenceQueueWatcherTask implements Runnable { + private final WeakReference contextRef; + + private ReferenceQueueWatcherTask(PythonContext context) { + contextRef = new WeakReference<>(context); + } + + @Override + public void run() { + PythonContext context = contextRef.get(); + if (context == null) { + return; + } + HandleContext handleContext = context.handleContext; + try { + while (handleContext.referenceQueueCoordinator.beginWatcherRemove()) { + Reference reference = handleContext.referenceQueueCoordinator.blockingRemove(handleContext.referenceQueue); + handleContext.referenceQueueCoordinator.endWatcherRemove(reference); + } + } finally { + handleContext.referenceQueueCoordinator.stopped(); + } + } + } + @TruffleBoundary public long getCurrentRSS() { if (backgroundGCTaskThread != null && backgroundGCTaskThread.isAlive()) { @@ -726,6 +754,28 @@ void runBackgroundGCTask(PythonContext context) { backgroundGCTaskThread.start(); } + void runReferenceQueueWatcher(PythonContext context) { + CompilerAsserts.neverPartOfCompilation(); + if (context.getEnv().isPreInitialization() || context.getOption(PythonOptions.NoAsyncActions) || !PythonOptions.AUTOMATIC_ASYNC_ACTIONS) { + return; + } + Thread thread = context.getEnv().createSystemThread(referenceQueueWatcherTask, context.getThreadGroup()); + thread.setName("C API reference queue watcher"); + context.handleContext.referenceQueueCoordinator.start(thread); + } + + private void stopReferenceQueueWatcher(PythonContext context) { + context.handleContext.referenceQueueCoordinator.stop(); + } + + /** + * Stops system threads that must not outlive the polyglot context. This must not run guest + * code: context disposal is still performed if cancellation interrupts finalization. + */ + public void dispose() { + stopReferenceQueueWatcher(getContext()); + } + /** * This represents whether the current process has already loaded an instance of the native CAPI * extensions - this can only be loaded globally once per process or in isolation multiple @@ -854,10 +904,11 @@ public static CApiContext ensureCapiWasLoaded(Node node, PythonContext context, context.runCApiHooks(); context.setCApiState(PythonContext.CApiState.INITIALIZED); // volatile write try { + cApiContext.runReferenceQueueWatcher(context); cApiContext.runBackgroundGCTask(context); } catch (RuntimeException e) { - // This can happen when other languages restrict multithreading - LOGGER.warning(() -> "didn't start the background GC task due to: " + e.getMessage()); + // This can happen when other languages restrict multithreading. + LOGGER.warning(() -> "didn't start a C API background task due to: " + e.getMessage()); } } catch (ImportException e) { context.setCApiState(PythonContext.CApiState.CANNOT_IMPORT); @@ -1159,13 +1210,14 @@ public void exitCApiContext() { * ensure that the GIL is held. */ try (GilNode.UncachedAcquire ignored = GilNode.uncachedAcquire()) { + PythonContext context = getContext(); /* * Polling the native reference queue is the only task we can do here because * deallocating objects may run arbitrary guest code that can again call into the * interpreter. */ - pollReferenceQueue(); - CApiTransitions.deallocateNativeWeakRefs(getContext()); + drainReferenceQueueNow(context); + CApiTransitions.deallocateNativeWeakRefs(context); } } @@ -1182,6 +1234,7 @@ public void finalizeCApi(boolean cancelling) { if (nativeFinalizerRunnable != null) { nativeFinalizerRunnable.run(); } + stopReferenceQueueWatcher(context); if (backgroundGCTaskThread != null && backgroundGCTaskThread.isAlive()) { context.killSystemThread(backgroundGCTaskThread); try { diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CExtNodes.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CExtNodes.java index d08a5a6422..e106a8dc70 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CExtNodes.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CExtNodes.java @@ -283,7 +283,7 @@ public static T allocateNativePart(Node inliningTarget, TransformExceptionFromNativeNode.executeUncached(context.getThreadState(context.getLanguage()), NativeCAPISymbol.FUN_PY_TYPE_GENERIC_NEW_RAW.getTsName(), nativeObject == NULLPTR, true); CApiTransitions.writeNativeRefCount(nativeObject, MANAGED_REFCNT); - CApiTransitions.createReference(managedSide, nativeObject); + CApiTransitions.createReference(context, managedSide, nativeObject); assert managedSide.isNative(); return managedSide; } diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/CApiTransitions.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/CApiTransitions.java index b7843c1e73..afe09b08f6 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/CApiTransitions.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/CApiTransitions.java @@ -228,8 +228,14 @@ public HandleContext(boolean useShadowTable) { public final ReferenceQueue referenceQueue = new ReferenceQueue<>(); + public final ReferenceQueueCoordinator referenceQueueCoordinator = new ReferenceQueueCoordinator(); + volatile PollingState referenceQueuePollingState = RQ_UNINITIALIZED; + long referenceQueueWatcherDrains; + long referenceQueueForcedDrains; + long referenceQueueReferencesDrained; + @TruffleBoundary public static T putShadowTable(HashMap table, long pointer, T ref) { return table.put(pointer, ref); @@ -472,136 +478,207 @@ public static PyCapsuleReference registerPyCapsuleDestructor(PyCapsule capsule) return ref; } + /** + * Synchronously drains the reference queue on the current Python thread while it owns the + * GIL. If there is no watcher, the queue is polled directly. Otherwise, the coordinator + * interrupts a watcher blocked in {@link ReferenceQueue#remove()}, waits until it has left the + * remove operation, and transfers exclusive ownership to this thread by changing its state to + * {@code DRAINING}. If the watcher removed a reference before observing the interrupt, that + * reference is preserved in the coordinator state and processed before the remaining queue is + * polled. Finally, ownership is returned and the watcher is resumed. Draining is still subject + * to the reference-queue polling state and is skipped while polling is disabled or unsafe. + */ + @SuppressWarnings("try") + public static int drainReferenceQueueNow(PythonContext context) { + if (!PythonOptions.AUTOMATIC_ASYNC_ACTIONS || context.handleContext.referenceQueueCoordinator.getState() == ReferenceQueueCoordinator.State.INACTIVE) { + return pollReferenceQueue(context, null, false); + } + return pollReferenceQueue(context, null, true); + } + @TruffleBoundary @SuppressWarnings("try") - public static int pollReferenceQueue() { - PythonContext context = PythonContext.get(null); + private static int pollReferenceQueue(PythonContext context, Reference handoff, boolean force) { HandleContext handleContext = context.handleContext; - int manuallyCollected = 0; - if (handleContext.referenceQueuePollingState != RQ_READY) { - return manuallyCollected; - } - /* - * Polling the reference queue may deallocate native GC objects and therefore re-enter - * native code paths that use '_PyThreadState_GET()' to obtain the current thread's GC - * state. So, we may only poll once the current thread has installed its native - * 'tstate_current' pointer. - */ - if (!context.getThreadState(context.getLanguage()).isNativeThreadStateInitialized()) { - return manuallyCollected; - } - try (GilNode.UncachedAcquire ignored = GilNode.uncachedAcquire()) { + boolean coordinated = false; + boolean watcherHandoff = false; + try { if (handleContext.referenceQueuePollingState != RQ_READY) { - return manuallyCollected; + return 0; } + /* + * Polling the reference queue may deallocate native GC objects and therefore re-enter + * native code paths that use '_PyThreadState_GET()' to obtain the current thread's GC + * state. So, we may only poll once the current thread has installed its native + * 'tstate_current' pointer. + */ if (!context.getThreadState(context.getLanguage()).isNativeThreadStateInitialized()) { - return manuallyCollected; + return 0; } - ReferenceQueue queue = handleContext.referenceQueue; - int count = 0; - long start = 0; - boolean polling = false; - ArrayList referencesToBeFreed = handleContext.referencesToBeFreed; - try { - while (true) { - Object entry = queue.poll(); - if (entry == null) { - if (count > 0) { - assert handleContext.referenceQueuePollingState == RQ_POLLING || handleContext.referenceQueuePollingState == RQ_DISABLED_PERMANENT; - releaseNativeObjects(context, referencesToBeFreed); - LOGGER.fine("collected " + count + " references from native reference queue in " + ((System.nanoTime() - start) / 1000000) + "ms"); - } - return manuallyCollected; + try (GilNode.UncachedAcquire ignored = GilNode.uncachedAcquire()) { + if (handleContext.referenceQueuePollingState != RQ_READY) { + return 0; + } + if (!context.getThreadState(context.getLanguage()).isNativeThreadStateInitialized()) { + return 0; + } + /* + * Do not claim a watcher handoff until all conditions that can defer polling have + * been checked. The watcher has already removed this reference from the queue, so + * restoring the coordinator state after an unused claim would otherwise lose the + * reference permanently. + */ + Reference firstReference = null; + if (handoff != null) { + firstReference = handleContext.referenceQueueCoordinator.beginWatcherDrain(handoff); + if (firstReference == null) { + return 0; } - if (count == 0) { - assert handleContext.referenceQueuePollingState == RQ_READY; - handleContext.referenceQueuePollingState = RQ_POLLING; - polling = true; - start = System.nanoTime(); - } else { - assert handleContext.referenceQueuePollingState == RQ_POLLING; + coordinated = true; + watcherHandoff = true; + } else if (force) { + Object previousState = handleContext.referenceQueueCoordinator.beginForcedDrain(context.getLanguage().unavailableSafepointLocation); + if (previousState == null) { + return 0; } - count++; - LOGGER.fine(() -> PythonUtils.formatJString("releasing %s, no remaining managed references", entry)); - if (entry instanceof PythonObjectReference reference) { - if (HandlePointerConverter.pointsToPyHandleSpace(reference.pointer)) { - assert !HandlePointerConverter.pointsToPyIntHandle(reference.pointer); - assert !HandlePointerConverter.pointsToPyFloatHandle(reference.pointer); - assert nativeStubLookupGet(handleContext, reference.pointer, reference.handleTableIndex) != null : Long.toHexString(reference.pointer); - LOGGER.finer(() -> PythonUtils.formatJString("releasing native stub lookup for managed object %x => %s", reference.pointer, reference)); - nativeStubLookupRemove(handleContext, reference); - /* - * We may only free native object stubs if their reference count is - * zero. We cannot free other structs (e.g. PyDateTime_CAPI) because we - * don't know if they are still used from native code. Those must be - * free'd at context finalization. - */ - long stubPointer = HandlePointerConverter.pointerToStub(reference.pointer); - long newRefCount = subNativeRefCount(stubPointer, MANAGED_REFCNT); - if (newRefCount == 0) { - LOGGER.finer(() -> PythonUtils.formatJString("No more references for %s (refcount->0): freeing native stub", reference)); - freeNativeStub(reference); - } else { - LOGGER.finer(() -> PythonUtils.formatJString("Some native references to %s remain (refcount=%d): not freeing native stub yet", reference, newRefCount)); - /* - * In this case, the object is no longer referenced from managed but - * still from native code (since the reference count is greater 0). - * This case is possible if there are reference cycles that include - * managed objects. We overwrite field 'CFields.GraalPyObject__id' - * to avoid incorrect reuse of the ID which could resolve to another - * object. - */ - writeIntField(stubPointer, CFields.GraalPyObject__handle_table_index, 0); - // this can only happen if the object is a GC object - assert reference.gc; - /* - * Since the managed object is already dead (only the native object - * stub is still alive), we need to remove the object from its - * current GC list. Otherwise, the Python GC would try to traverse - * the object on the next collection which would lead to a crash. - */ - GCListRemoveNode.executeUncached(stubPointer); + coordinated = true; + if (previousState instanceof Reference reference) { + firstReference = reference; + } + } + if (watcherHandoff) { + handleContext.referenceQueueWatcherDrains++; + } else { + handleContext.referenceQueueForcedDrains++; + } + ReferenceQueue queue = handleContext.referenceQueue; + int count = 0; + long start = 0; + boolean polling = false; + ArrayList referencesToBeFreed = handleContext.referencesToBeFreed; + try { + Reference entry = firstReference; + while (true) { + if (entry == null) { + entry = queue.poll(); + } + if (entry == null) { + if (count > 0) { + assert handleContext.referenceQueuePollingState == RQ_POLLING || handleContext.referenceQueuePollingState == RQ_DISABLED_PERMANENT; + releaseNativeObjects(context, referencesToBeFreed); } + handleContext.referenceQueueReferencesDrained += count; + if (count > 0) { + logReferenceQueueDrain(handleContext, watcherHandoff, count, start); + } + return 0; + } + if (count == 0) { + assert handleContext.referenceQueuePollingState == RQ_READY; + handleContext.referenceQueuePollingState = RQ_POLLING; + polling = true; + start = System.nanoTime(); } else { - assert nativeLookupGet(handleContext, reference.pointer) != null : Long.toHexString(reference.pointer); - LOGGER.finer(() -> PythonUtils.formatJString("releasing native stub lookup for managed object with replacement %x => %s", reference.pointer, reference)); - if (nativeLookupRemove(handleContext, reference.pointer) != null) { - // The reference was still in our lookup table, it was not otherwise - // freed and we can process it now. - if (reference.isAllocatedFromJava()) { - LOGGER.finer(() -> PythonUtils.formatJString("freeing managed object %s replacement", reference)); - freeNativeStruct(reference); + assert handleContext.referenceQueuePollingState == RQ_POLLING; + } + count++; + Reference currentEntry = entry; + entry = null; + LOGGER.fine(() -> PythonUtils.formatJString("releasing %s, no remaining managed references", currentEntry)); + if (currentEntry instanceof PythonObjectReference reference) { + if (HandlePointerConverter.pointsToPyHandleSpace(reference.pointer)) { + assert !HandlePointerConverter.pointsToPyIntHandle(reference.pointer); + assert !HandlePointerConverter.pointsToPyFloatHandle(reference.pointer); + assert nativeStubLookupGet(handleContext, reference.pointer, reference.handleTableIndex) != null : Long.toHexString(reference.pointer); + LOGGER.finer(() -> PythonUtils.formatJString("releasing native stub lookup for managed object %x => %s", reference.pointer, reference)); + nativeStubLookupRemove(handleContext, reference); + /* + * We may only free native object stubs if their reference count is + * zero. We cannot free other structs (e.g. PyDateTime_CAPI) because we + * don't know if they are still used from native code. Those must be + * free'd at context finalization. + */ + long stubPointer = HandlePointerConverter.pointerToStub(reference.pointer); + long newRefCount = subNativeRefCount(stubPointer, MANAGED_REFCNT); + if (newRefCount == 0) { + LOGGER.finer(() -> PythonUtils.formatJString("No more references for %s (refcount->0): freeing native stub", reference)); + freeNativeStub(reference); } else { - referencesToBeFreed.add(reference.pointer); + LOGGER.finer(() -> PythonUtils.formatJString("Some native references to %s remain (refcount=%d): not freeing native stub yet", reference, newRefCount)); + /* + * In this case, the object is no longer referenced from managed but + * still from native code (since the reference count is greater 0). + * This case is possible if there are reference cycles that include + * managed objects. We overwrite field 'CFields.GraalPyObject__id' + * to avoid incorrect reuse of the ID which could resolve to another + * object. + */ + writeIntField(stubPointer, CFields.GraalPyObject__handle_table_index, 0); + // this can only happen if the object is a GC object + assert reference.gc; + /* + * Since the managed object is already dead (only the native object + * stub is still alive), we need to remove the object from its + * current GC list. Otherwise, the Python GC would try to traverse + * the object on the next collection which would lead to a crash. + */ + GCListRemoveNode.executeUncached(stubPointer); } } else { - // This handle was removed from the native lookup table before, - // probably during an explicit collection. This can happen during - // shutdown when tp_dealloc is called for some objects and that - // causes upcalls and reference queue polling on references that - // were already removed and had their memory freed + assert nativeLookupGet(handleContext, reference.pointer) != null : Long.toHexString(reference.pointer); + LOGGER.finer(() -> PythonUtils.formatJString("releasing native stub lookup for managed object with replacement %x => %s", reference.pointer, reference)); + if (nativeLookupRemove(handleContext, reference.pointer) != null) { + // The reference was still in our lookup table, it was not otherwise + // freed and we can process it now. + if (reference.isAllocatedFromJava()) { + LOGGER.finer(() -> PythonUtils.formatJString("freeing managed object %s replacement", reference)); + freeNativeStruct(reference); + } else { + referencesToBeFreed.add(reference.pointer); + } + } else { + // This handle was removed from the native lookup table before, + // probably during an explicit collection. This can happen during + // shutdown when tp_dealloc is called for some objects and that + // causes upcalls and reference queue polling on references that + // were already removed and had their memory freed + } } + } else if (currentEntry instanceof NativeObjectReference reference) { + if (nativeLookupRemove(handleContext, reference.pointer) == reference) { + // The reference was still in our lookup table, it was not otherwise + // freed and we can process it now + LOGGER.finer(() -> PythonUtils.formatJString("releasing native lookup for native object %x => %s", reference.pointer, reference)); + processNativeObjectReference(reference, referencesToBeFreed); + } + } else if (currentEntry instanceof NativeStorageReference reference) { + handleContext.nativeStorageReferences.remove(reference); + processNativeStorageReference(reference); + } else if (currentEntry instanceof PyCapsuleReference reference) { + handleContext.pyCapsuleReferences.remove(reference); + processPyCapsuleReference(reference); } - } else if (entry instanceof NativeObjectReference reference) { - if (nativeLookupRemove(handleContext, reference.pointer) == reference) { - // The reference was still in our lookup table, it was not otherwise - // freed and we can process it now - LOGGER.finer(() -> PythonUtils.formatJString("releasing native lookup for native object %x => %s", reference.pointer, reference)); - processNativeObjectReference(reference, referencesToBeFreed); - } - } else if (entry instanceof NativeStorageReference reference) { - handleContext.nativeStorageReferences.remove(reference); - processNativeStorageReference(reference); - } else if (entry instanceof PyCapsuleReference reference) { - handleContext.pyCapsuleReferences.remove(reference); - processPyCapsuleReference(reference); } - } - } finally { - if (polling && handleContext.referenceQueuePollingState == RQ_POLLING) { - handleContext.referenceQueuePollingState = RQ_READY; + } finally { + if (polling && handleContext.referenceQueuePollingState == RQ_POLLING) { + handleContext.referenceQueuePollingState = RQ_READY; + } } } + } finally { + if (coordinated) { + handleContext.referenceQueueCoordinator.finishDrain(); + } + } + } + + private static void logReferenceQueueDrain(HandleContext handleContext, boolean watcherHandoff, int count, long start) { + if (LOGGER.isLoggable(Level.FINE)) { + LOGGER.fine(PythonUtils.formatJString( + "native reference queue drain reason=%s drained=%d time=%dms totals={watcher-drains=%d, forced-drains=%d, references-drained=%d}", + watcherHandoff ? "watcher" : "forced", count, (System.nanoTime() - start) / 1000000, + handleContext.referenceQueueWatcherDrains, handleContext.referenceQueueForcedDrains, + handleContext.referenceQueueReferencesDrained)); } } @@ -772,7 +849,7 @@ public static void deallocNativeReplacements(PythonContext context, HandleContex } } releaseNativeObjects(context, referencesToBeFreed); - pollReferenceQueue(); + pollReferenceQueueIfPending(context); } public static void freeNativeReplacementStructs(PythonContext context, HandleContext handleContext) { @@ -797,6 +874,10 @@ public static void freeNativeReplacementStructs(PythonContext context, HandleCon public static boolean disableReferenceQueuePolling(HandleContext handleContext) { if (handleContext.referenceQueuePollingState == RQ_READY) { + PythonContext context = PythonContext.get(null); + if (!handleContext.referenceQueueCoordinator.disable(context.getLanguage().unavailableSafepointLocation)) { + return true; + } handleContext.referenceQueuePollingState = RQ_DISABLED_TEMP; return false; } @@ -806,6 +887,7 @@ public static boolean disableReferenceQueuePolling(HandleContext handleContext) public static void enableReferenceQueuePolling(HandleContext handleContext) { if (handleContext.referenceQueuePollingState == RQ_DISABLED_TEMP) { handleContext.referenceQueuePollingState = RQ_READY; + handleContext.referenceQueueCoordinator.enable(); } } @@ -818,6 +900,30 @@ public static void disableReferenceQueuePollingPermanently(HandleContext handleC handleContext.referenceQueuePollingState = RQ_DISABLED_PERMANENT; } + /** + * Processes a lock-free handoff from the reference-queue watcher. The watcher removes the + * first reference and publishes it by replacing {@code WATCHING} in the coordinator's atomic + * state. A Python thread reads that state once; a {@link Reference} means that work is pending + * and is passed on as the expected value used to claim exclusive draining ownership. In the + * watcher steady state, a non-reference value returns immediately without taking a lock or + * inspecting the queue. If no watcher exists, reference-queue polling falls back to the direct, + * uncoordinated path. + */ + public static void pollReferenceQueueIfPending(PythonContext context) { + HandleContext handleContext = context.handleContext; + if (!PythonOptions.AUTOMATIC_ASYNC_ACTIONS) { + pollReferenceQueue(context, null, false); + return; + } + Object state = handleContext.referenceQueueCoordinator.getState(); + if (state instanceof Reference reference) { + pollReferenceQueue(context, reference, false); + } else if (state == ReferenceQueueCoordinator.State.INACTIVE) { + // Preserve polling when the watcher is disabled by configuration. + pollReferenceQueue(context, null, false); + } + } + private static void freeNativeStub(PythonObjectReference ref) { freeNativeStub(ref.pointer, ref.gc); } @@ -1000,14 +1106,14 @@ public static void deallocateNativeWeakRefs(PythonContext pythonContext) { assert context.nativeWeakRef.isEmpty(); } - public static void maybeGCALot() { + public static void maybeGCALot(Node inliningTarget) { if (GCALot != 0) { - maybeGC(); + maybeGC(PythonContext.get(inliningTarget)); } } @TruffleBoundary - private static void maybeGC() { + private static void maybeGC(PythonContext context) { GCALotTotalCounter++; if (GCALotTotalCounter < GCALotWait) { // skip @@ -1015,7 +1121,7 @@ private static void maybeGC() { LOGGER.info("GC A Lot - calling System.gc (opportunities=" + GCALotTotalCounter + ")"); GCALotCounter = 0; PythonUtils.forceFullGC(); - pollReferenceQueue(); + pollReferenceQueueIfPending(context); } } @@ -1310,7 +1416,7 @@ static long doMemoryView(@SuppressWarnings("unused") Node inliningTarget, PMemor assert !mv.isNative(); assert initialRefCount == IMMORTAL_REFCNT; long ptr = PyMemoryViewWrapper.allocate(mv); - CApiTransitions.createReference(mv, ptr); + CApiTransitions.createReference(PythonContext.get(null), mv, ptr); return ptr; } @@ -1416,7 +1522,8 @@ static long doGeneric(Node inliningTarget, PythonAbstractObject object, Object t @Cached PyObjectGCTrackNode gcTrackNode) { log(object); - pollReferenceQueue(); + PythonContext pythonContext = PythonContext.get(inliningTarget); + pollReferenceQueueIfPending(pythonContext); /* * Allocate a native stub object (C type: GraalPy*Object). For types that participate in @@ -1427,7 +1534,6 @@ static long doGeneric(Node inliningTarget, PythonAbstractObject object, Object t long stubPointer = NativeMemory.malloc(allocationSize); NativeMemory.memset(stubPointer, (byte) 0, ctype.size() + presize); - PythonContext pythonContext = PythonContext.get(inliningTarget); HandleContext handleContext = pythonContext.handleContext; long taggedPointer = HandlePointerConverter.stubToPointer(stubPointer); @@ -1511,7 +1617,7 @@ static long doGeneric(Node inliningTarget, PythonAbstractObject object, Object t */ @TruffleBoundary @SuppressWarnings("try") - public static void createReference(PythonObject obj, long ptr) { + public static void createReference(PythonContext pythonContext, PythonObject obj, long ptr) { try (GilNode.UncachedAcquire ignored = GilNode.uncachedAcquire()) { /* * The first test if '!obj.isNative()' in the caller is done on a fast-path but not @@ -1520,8 +1626,8 @@ public static void createReference(PythonObject obj, long ptr) { if (!obj.isNative()) { logVoid(obj, ptr); obj.setNativePointer(ptr); - pollReferenceQueue(); - HandleContext context = getContext(); + pollReferenceQueueIfPending(pythonContext); + HandleContext context = pythonContext.handleContext; nativeLookupPut(context, ptr, PythonObjectReference.createReplacement(context, obj, ptr, false)); } } @@ -1826,13 +1932,13 @@ static long doPythonObject(Node inliningTarget, PythonObject pythonObject, boole @Exclusive @Cached FirstToNativeNode firstToNativeNode, @Exclusive @Cached UpdateStrongRefNode updateRefNode) { CompilerAsserts.partialEvaluationConstant(needsTransfer); - assert PythonContext.get(inliningTarget).ownsGil(); - pollReferenceQueue(); + PythonContext context = PythonContext.get(inliningTarget); + assert context.ownsGil(); + pollReferenceQueueIfPending(context); long pointer; if (!pythonObject.isNative()) { assert !CApiContext.isSpecialSingleton(pythonObject); - PythonContext context = PythonContext.get(inliningTarget); boolean immortal = isImmortalPythonObject(context, pythonObject); pointer = firstToNativeNode.execute(inliningTarget, pythonObject, FirstToNativeNode.getInitialRefcnt(needsTransfer, immortal)); pythonObject.setNativePointer(pointer); @@ -1863,8 +1969,9 @@ static long doGeneric(Node inliningTarget, Object obj, boolean needsTransfer, @Exclusive @Cached FirstToNativeNode firstToNativeNode, @Exclusive @Cached UpdateStrongRefNode updateRefNode) { CompilerAsserts.partialEvaluationConstant(needsTransfer); - assert PythonContext.get(inliningTarget).ownsGil(); - pollReferenceQueue(); + PythonContext context = PythonContext.get(inliningTarget); + assert context.ownsGil(); + pollReferenceQueueIfPending(context); Object profiled = classProfile.profile(inliningTarget, obj); @@ -1889,8 +1996,6 @@ static long doGeneric(Node inliningTarget, Object obj, boolean needsTransfer, return doNative(inliningTarget, pythonAbstractNativeObject, needsTransfer, hasReplicatedNativeReferences, updateRefNode); } - PythonContext context = PythonContext.get(inliningTarget); - /* * Step 4: Special singletons (e.g. PNone) are context-independent. Their native * companions are stored in a special cache. @@ -2119,7 +2224,8 @@ static Object doGeneric(Node inliningTarget, long pointer, boolean needsTransfer /* * Here we are encountering a weakref object that has died in the managed side, * e.g. PReferenceType, but we kept alive in the native side, see - * pollReferenceQueue(). Though, if this happens to an object that shouldn't + * pollReferenceQueueIfPending(). Though, if this happens to an object that + * shouldn't * have died in the managed side, the native side should catch it with a null * pointer check. */ @@ -2145,7 +2251,7 @@ static Object doGeneric(Node inliningTarget, long pointer, boolean needsTransfer if (LOGGER.isLoggable(Level.FINE)) { LOGGER.fine(() -> "re-creating collected PythonAbstractNativeObject reference" + Long.toHexString(pointer)); } - return createAbstractNativeObject(threadState, handleContext, needsTransfer, pointer); + return createAbstractNativeObject(pythonContext, threadState, handleContext, needsTransfer, pointer); } if (isNativeObjectProfile.profile(inliningTarget, ref instanceof PythonAbstractNativeObject)) { if (needsTransfer) { @@ -2157,7 +2263,7 @@ static Object doGeneric(Node inliningTarget, long pointer, boolean needsTransfer result = (PythonAbstractObject) ref; } } else { - return createAbstractNativeObject(threadState, handleContext, needsTransfer, pointer); + return createAbstractNativeObject(pythonContext, threadState, handleContext, needsTransfer, pointer); } } return updateRef(inliningTarget, wrapperProfile, updateRefNode, needsTransfer, release, result); @@ -2328,7 +2434,7 @@ static Object doNonWrapper(long pointer, boolean stealing, Object ref = lookup.get(); if (createNativeProfile.profile(inliningTarget, ref == null)) { LOGGER.fine(() -> "re-creating collected PythonAbstractNativeObject reference" + Long.toHexString(pointer)); - return createAbstractNativeObject(threadState, handleContext, stealing, pointer); + return createAbstractNativeObject(pythonContext, threadState, handleContext, stealing, pointer); } if (isNativeObjectProfile.profile(inliningTarget, ref instanceof PythonAbstractNativeObject)) { if (stealing) { @@ -2340,7 +2446,7 @@ static Object doNonWrapper(long pointer, boolean stealing, pythonAbstractObject = (PythonAbstractObject) ref; } } else { - return createAbstractNativeObject(threadState, handleContext, stealing, pointer); + return createAbstractNativeObject(pythonContext, threadState, handleContext, stealing, pointer); } } return NativeToPythonInternalNode.updateRef(inliningTarget, wrapperProfile, updateRefNode, stealing, false, pythonAbstractObject); @@ -2464,7 +2570,7 @@ static Object doGeneric(Node inliningTarget, long pointer, if (LOGGER.isLoggable(Level.FINE)) { LOGGER.fine(PythonUtils.formatJString("re-creating collected PythonNativeClass reference 0x%x", pointer)); } - return recreatePythonNativeClass(handleContext, pointer); + return recreatePythonNativeClass(pythonContext, handleContext, pointer); } assert clazz instanceof PythonAbstractClass; return clazz; @@ -2603,8 +2709,8 @@ public static void writeNativeRefCount(long pointer, long newValue) { UNSAFE.putLong(pointer + TP_REFCNT_OFFSET, newValue); } - private static PythonAbstractNativeObject createAbstractNativeObject(PythonThreadState threadState, HandleContext handleContext, boolean transfer, long pointer) { - pollReferenceQueue(); + private static PythonAbstractNativeObject createAbstractNativeObject(PythonContext pythonContext, PythonThreadState threadState, HandleContext handleContext, boolean transfer, long pointer) { + pollReferenceQueueIfPending(pythonContext); PythonAbstractNativeObject result = new PythonAbstractNativeObject(pointer); long refCntDelta = MANAGED_REFCNT - (transfer ? 1 : 0); /* @@ -2632,8 +2738,8 @@ private static PythonAbstractNativeObject createAbstractNativeObject(PythonThrea * index from the native {@code PyTypeObject}. */ @TruffleBoundary - private static PythonNativeClass recreatePythonNativeClass(HandleContext handleContext, long pointer) { - pollReferenceQueue(); + private static PythonNativeClass recreatePythonNativeClass(PythonContext pythonContext, HandleContext handleContext, long pointer) { + pollReferenceQueueIfPending(pythonContext); PythonAbstractNativeObject result = new PythonAbstractNativeObject(pointer); /* * Some APIs might be called from tp_dealloc/tp_del/tp_finalize where the refcount is 0. In diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/ReferenceQueueCoordinator.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/ReferenceQueueCoordinator.java new file mode 100644 index 0000000000..de00bb0ab2 --- /dev/null +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/ReferenceQueueCoordinator.java @@ -0,0 +1,349 @@ +/* + * Copyright (c) 2026, 2026, Oracle and/or its affiliates. All rights reserved. + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. + * + * The Universal Permissive License (UPL), Version 1.0 + * + * Subject to the condition set forth below, permission is hereby granted to any + * person obtaining a copy of this software, associated documentation and/or + * data (collectively the "Software"), free of charge and under any and all + * copyright rights in the Software, and any and all patent rights owned or + * freely licensable by each licensor hereunder covering either (i) the + * unmodified Software as contributed to or provided by such licensor, or (ii) + * the Larger Works (as defined below), to deal in both + * + * (a) the Software, and + * + * (b) any piece of software and/or hardware listed in the lrgrwrks.txt file if + * one is included with the Software each a "Larger Work" to which the Software + * is contributed by such licensors), + * + * without restriction, including without limitation the rights to copy, create + * derivative works of, display, perform, and distribute the Software and make, + * use, sell, offer for sale, import, export, have made, and have sold the + * Software and the Larger Work(s), and to sublicense the foregoing rights on + * either these or other terms. + * + * This license is subject to the following condition: + * + * The above copyright notice and either this complete permission notice or at a + * minimum a reference to the UPL must be included in all copies or substantial + * portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.oracle.graal.python.builtins.objects.cext.capi.transitions; + +import java.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; + +import com.oracle.truffle.api.TruffleSafepoint; +import com.oracle.truffle.api.nodes.Node; + +/** + * Coordinates exclusive ownership of the reference queue between the watcher and Python + * execution threads. The watcher only removes the first reference. All processing and the + * remaining queue drain happen on a Python thread while it owns the GIL. + */ +public final class ReferenceQueueCoordinator { + enum State { + /** + * Initial state and steady state when no watcher was started, for example during + * preinitialization, with async actions disabled, or when system-thread creation + * failed. Python threads poll the reference queue directly in this state. Shutdown + * uses {@link #STOPPING} and {@link #STOPPED} instead. + */ + INACTIVE, + WATCHING, + DRAINING, + DISABLED, + STOPPING, + STOPPED + } + + private static final class DisabledReference { + final Reference reference; + + DisabledReference(Reference reference) { + this.reference = reference; + } + } + + private final ReentrantLock lock = new ReentrantLock(); + private final Condition stateChanged = lock.newCondition(); + private final AtomicReference state = new AtomicReference<>(State.INACTIVE); + private Thread watcherThread; + private volatile boolean watcherInRemove; + private volatile boolean pauseRequested; + + /** + * Takes ownership of the watcher thread, publishes the watching state, and starts the thread. + */ + public void start(Thread thread) { + lock.lock(); + try { + assert state.get() == State.INACTIVE : state.get(); + watcherThread = thread; + state.set(State.WATCHING); + } finally { + lock.unlock(); + } + try { + thread.start(); + } catch (RuntimeException e) { + stopped(); + throw e; + } + } + + public boolean isActive() { + Object current = state.get(); + return current != State.INACTIVE && current != State.STOPPING && current != State.STOPPED; + } + + /** + * Called by the watcher before entering {@link ReferenceQueue#remove()}. + */ + public boolean beginWatcherRemove() { + lock.lock(); + try { + while (state.get() != State.WATCHING || pauseRequested) { + Object current = state.get(); + if (current == State.STOPPING || current == State.STOPPED) { + return false; + } + try { + stateChanged.await(); + } catch (InterruptedException e) { + // Lifecycle transitions set the state before interrupting the watcher. + } + } + watcherInRemove = true; + return true; + } finally { + lock.unlock(); + } + } + + public Reference blockingRemove(ReferenceQueue queue) { + try { + return queue.remove(); + } catch (InterruptedException e) { + return null; + } + } + + /** + * Completes one watcher remove operation and returns its handoff, or {@code null} + * when no action should be submitted. + */ + public Reference endWatcherRemove(Reference reference) { + assert watcherInRemove; + Reference handoff = null; + if (reference != null) { + if (state.compareAndSet(State.WATCHING, reference)) { + handoff = pauseRequested ? null : reference; + } + } + watcherInRemove = false; + signalStateChanged(); + return handoff; + } + + /** + * Claims a reference handed off by the watcher. Returns the reference if the claim + * succeeded, or {@code null} if another thread or a lifecycle transition won the race. + */ + public Reference beginWatcherDrain(Reference reference) { + if (!state.compareAndSet(reference, State.DRAINING)) { + return null; + } + return reference; + } + + public Reference getPendingReference() { + Object current = state.get(); + return current instanceof Reference reference ? reference : null; + } + + Object getState() { + return state.get(); + } + + /** + * Claims the queue for a forced drain. The returned value is the previous coordinator + * state: a {@link Reference} is the first entry to process and {@link State#WATCHING} + * means that the claim succeeded without a watcher handoff. {@code null} means that the + * queue could not be claimed. + */ + public Object beginForcedDrain(Node location) { + lock.lock(); + try { + Object current = state.get(); + if (current == State.INACTIVE) { + return null; + } + if (current == State.STOPPING || current == State.STOPPED || current == State.DRAINING || current == State.DISABLED || current instanceof DisabledReference) { + return null; + } + pauseRequested = true; + if (watcherInRemove) { + watcherThread.interrupt(); + while (watcherInRemove) { + TruffleSafepoint.setBlockedThreadInterruptible(location, Condition::await, stateChanged); + } + } + current = state.get(); + if (current == State.STOPPING || current == State.STOPPED || current == State.DRAINING || current == State.DISABLED || current instanceof DisabledReference || + !state.compareAndSet(current, State.DRAINING)) { + pauseRequested = false; + stateChanged.signalAll(); + return null; + } + pauseRequested = false; + return current; + } finally { + lock.unlock(); + } + } + + public void finishDrain() { + lock.lock(); + try { + if (state.compareAndSet(State.DRAINING, State.WATCHING)) { + stateChanged.signalAll(); + } + } finally { + lock.unlock(); + } + } + + public boolean disable(Node location) { + lock.lock(); + try { + Object current = state.get(); + if (current == State.INACTIVE) { + return true; + } + if (current == State.DISABLED || current instanceof DisabledReference || current == State.DRAINING || current == State.STOPPING || current == State.STOPPED) { + return false; + } + pauseRequested = true; + if (watcherInRemove) { + watcherThread.interrupt(); + while (watcherInRemove) { + TruffleSafepoint.setBlockedThreadInterruptible(location, Condition::await, stateChanged); + } + } + current = state.get(); + if (current == State.DISABLED || current instanceof DisabledReference || current == State.DRAINING || current == State.STOPPING || current == State.STOPPED) { + pauseRequested = false; + stateChanged.signalAll(); + return false; + } + Object disabledState = current instanceof Reference reference ? new DisabledReference(reference) : State.DISABLED; + if (!state.compareAndSet(current, disabledState)) { + pauseRequested = false; + stateChanged.signalAll(); + return false; + } + pauseRequested = false; + return true; + } finally { + lock.unlock(); + } + } + + /** + * Re-enables the watcher without processing references on the native GC call stack. + */ + public void enable() { + lock.lock(); + try { + Object current = state.get(); + Object enabledState; + if (current == State.DISABLED) { + enabledState = State.WATCHING; + } else if (current instanceof DisabledReference disabled) { + enabledState = disabled.reference; + } else { + // INACTIVE is expected when automatic async actions are disabled. + return; + } + if (state.compareAndSet(current, enabledState)) { + stateChanged.signalAll(); + } + } finally { + lock.unlock(); + } + } + + /** + * Stops the owned watcher thread. The coordinator first publishes {@link State#STOPPING}, then + * interrupts the thread so that it leaves {@link ReferenceQueue#remove()}, and joins it before + * clearing the lifecycle state. If joining is interrupted, shutdown still completes and the + * caller's interrupt status is restored afterward. + */ + public void stop() { + Thread thread; + lock.lock(); + try { + Object current = state.get(); + if (current == State.INACTIVE || current == State.STOPPED) { + return; + } + state.set(State.STOPPING); + stateChanged.signalAll(); + thread = watcherThread; + } finally { + lock.unlock(); + } + if (thread != null && thread.isAlive()) { + thread.interrupt(); + boolean interrupted = false; + while (thread.isAlive()) { + try { + thread.join(); + } catch (InterruptedException e) { + interrupted = true; + } + } + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + stopped(); + } + + public void stopped() { + lock.lock(); + try { + state.set(State.STOPPED); + watcherThread = null; + watcherInRemove = false; + pauseRequested = false; + stateChanged.signalAll(); + } finally { + lock.unlock(); + } + } + + private void signalStateChanged() { + lock.lock(); + try { + stateChanged.signalAll(); + } finally { + lock.unlock(); + } + } +} From d02eae71f214fbd28b9acc390a0a26fefeb2efa9 Mon Sep 17 00:00:00 2001 From: Florian Angerer Date: Thu, 27 Aug 2026 19:48:59 +0200 Subject: [PATCH 5/6] Fix finalization ordering issue --- .../oracle/graal/python/runtime/PythonContext.java | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/PythonContext.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/PythonContext.java index f3cce41c0a..df0b4c4c53 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/PythonContext.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/PythonContext.java @@ -1777,12 +1777,18 @@ public void finalizeContext() { } // interrupt and join or kill python threads joinPythonThreads(); - if (nativeContext != null) { - nativeContext.close(); - } freeContextMemory(); // destroy thread state data, if anything is still running, it will crash now disposeThreadStates(); + /* + * Closing NativeContext must be the last operation because it may unload native libraries + * with their global state. If any code still has the address to some variable and writes + * to it, this can cause severe memory corruptions. E.g. the address of `tstate_current` is + * stored in `PythonThreadState.nativeThreadLocalVarPointer` and will be nulled on disposal. + */ + if (nativeContext != null) { + nativeContext.close(); + } } // interrupt and join or kill system threads joinSystemThreads(); From 95976664930c998f2dc064756be5db787026af11 Mon Sep 17 00:00:00 2001 From: Florian Angerer Date: Thu, 3 Sep 2026 13:48:38 +0200 Subject: [PATCH 6/6] Add state logging for watcher thread --- .../objects/cext/capi/CApiContext.java | 9 ++- .../ReferenceQueueCoordinator.java | 58 ++++++++++++++++--- 2 files changed, 57 insertions(+), 10 deletions(-) diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java index 76b6b10c12..ff972999d4 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/CApiContext.java @@ -95,6 +95,7 @@ import com.oracle.graal.python.builtins.objects.cext.capi.transitions.CApiTransitions.HandlePointerConverter; import com.oracle.graal.python.builtins.objects.cext.capi.transitions.CApiTransitions.NativeToPythonInternalNode; import com.oracle.graal.python.builtins.objects.cext.capi.transitions.CApiTransitions.PythonToNativeInternalNode; +import com.oracle.graal.python.builtins.objects.cext.capi.transitions.ReferenceQueueCoordinator; import com.oracle.graal.python.builtins.objects.cext.common.CExtContext; import com.oracle.graal.python.builtins.objects.cext.common.LoadCExtException.ApiInitException; import com.oracle.graal.python.builtins.objects.cext.common.LoadCExtException.ImportException; @@ -761,7 +762,13 @@ void runReferenceQueueWatcher(PythonContext context) { } Thread thread = context.getEnv().createSystemThread(referenceQueueWatcherTask, context.getThreadGroup()); thread.setName("C API reference queue watcher"); - context.handleContext.referenceQueueCoordinator.start(thread); + /* + * The watcher runs without an entered context, so it cannot use a logger obtained through + * TruffleLogger.getLogger. Env.getLogger returns a context-bound logger that is safe to use + * from such a system thread. + */ + TruffleLogger logger = context.getEnv().getLogger(LOGGER_CAPI_NAME + "." + ReferenceQueueCoordinator.class.getSimpleName()); + context.handleContext.referenceQueueCoordinator.start(thread, logger); } private void stopReferenceQueueWatcher(PythonContext context) { diff --git a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/ReferenceQueueCoordinator.java b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/ReferenceQueueCoordinator.java index de00bb0ab2..3b93e2f979 100644 --- a/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/ReferenceQueueCoordinator.java +++ b/graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/capi/transitions/ReferenceQueueCoordinator.java @@ -47,6 +47,7 @@ import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; +import com.oracle.truffle.api.TruffleLogger; import com.oracle.truffle.api.TruffleSafepoint; import com.oracle.truffle.api.nodes.Node; @@ -82,6 +83,7 @@ private static final class DisabledReference { private final ReentrantLock lock = new ReentrantLock(); private final Condition stateChanged = lock.newCondition(); private final AtomicReference state = new AtomicReference<>(State.INACTIVE); + private TruffleLogger logger; private Thread watcherThread; private volatile boolean watcherInRemove; private volatile boolean pauseRequested; @@ -90,11 +92,16 @@ private static final class DisabledReference { * Takes ownership of the watcher thread, publishes the watching state, and starts the thread. */ public void start(Thread thread) { + start(thread, null); + } + + public void start(Thread thread, TruffleLogger contextBoundLogger) { lock.lock(); try { assert state.get() == State.INACTIVE : state.get(); + logger = contextBoundLogger; watcherThread = thread; - state.set(State.WATCHING); + setState(State.WATCHING); } finally { lock.unlock(); } @@ -151,7 +158,7 @@ public Reference endWatcherRemove(Reference reference) { assert watcherInRemove; Reference handoff = null; if (reference != null) { - if (state.compareAndSet(State.WATCHING, reference)) { + if (compareAndSetState(State.WATCHING, reference)) { handoff = pauseRequested ? null : reference; } } @@ -165,7 +172,7 @@ public Reference endWatcherRemove(Reference reference) { * succeeded, or {@code null} if another thread or a lifecycle transition won the race. */ public Reference beginWatcherDrain(Reference reference) { - if (!state.compareAndSet(reference, State.DRAINING)) { + if (!compareAndSetState(reference, State.DRAINING)) { return null; } return reference; @@ -205,7 +212,7 @@ public Object beginForcedDrain(Node location) { } current = state.get(); if (current == State.STOPPING || current == State.STOPPED || current == State.DRAINING || current == State.DISABLED || current instanceof DisabledReference || - !state.compareAndSet(current, State.DRAINING)) { + !compareAndSetState(current, State.DRAINING)) { pauseRequested = false; stateChanged.signalAll(); return null; @@ -220,7 +227,7 @@ public Object beginForcedDrain(Node location) { public void finishDrain() { lock.lock(); try { - if (state.compareAndSet(State.DRAINING, State.WATCHING)) { + if (compareAndSetState(State.DRAINING, State.WATCHING)) { stateChanged.signalAll(); } } finally { @@ -252,7 +259,7 @@ public boolean disable(Node location) { return false; } Object disabledState = current instanceof Reference reference ? new DisabledReference(reference) : State.DISABLED; - if (!state.compareAndSet(current, disabledState)) { + if (!compareAndSetState(current, disabledState)) { pauseRequested = false; stateChanged.signalAll(); return false; @@ -280,7 +287,7 @@ public void enable() { // INACTIVE is expected when automatic async actions are disabled. return; } - if (state.compareAndSet(current, enabledState)) { + if (compareAndSetState(current, enabledState)) { stateChanged.signalAll(); } } finally { @@ -302,7 +309,7 @@ public void stop() { if (current == State.INACTIVE || current == State.STOPPED) { return; } - state.set(State.STOPPING); + setState(State.STOPPING); stateChanged.signalAll(); thread = watcherThread; } finally { @@ -328,7 +335,7 @@ public void stop() { public void stopped() { lock.lock(); try { - state.set(State.STOPPED); + setState(State.STOPPED); watcherThread = null; watcherInRemove = false; pauseRequested = false; @@ -346,4 +353,37 @@ private void signalStateChanged() { lock.unlock(); } } + + private boolean compareAndSetState(Object expectedState, Object newState) { + if (!state.compareAndSet(expectedState, newState)) { + return false; + } + logStateChange(expectedState, newState); + return true; + } + + private void setState(Object newState) { + Object previousState = state.getAndSet(newState); + if (previousState != newState) { + logStateChange(previousState, newState); + } + } + + private void logStateChange(Object previousState, Object newState) { + TruffleLogger currentLogger = logger; + if (currentLogger != null) { + currentLogger.fine(() -> String.format("Reference queue coordinator %x state changed on thread '%s': %s -> %s", System.identityHashCode(this), Thread.currentThread().getName(), + describeState(previousState), describeState(newState))); + } + } + + private static String describeState(Object value) { + if (value instanceof DisabledReference disabled) { + return String.format("DISABLED[%s]", describeState(disabled.reference)); + } + if (value instanceof Reference reference) { + return String.format("PENDING[%s@%x]", reference.getClass().getSimpleName(), System.identityHashCode(reference)); + } + return String.valueOf(value); + } }