Context's shared finalizer

Author: qunaibit

graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/Python3Core.java

@@ -531,6 +531,7 @@ public void postInitialize() {
                 builtin.postInitialize(this);
             }
 
+            getContext().getSharedFinalizer().registerAsyncAction();
             initialized = true;
         }
     }

graalpython/com.oracle.graal.python/src/com/oracle/graal/python/builtins/objects/cext/hpy/GraalHPyContext.java

@@ -865,11 +865,11 @@ static final class GraalHPyHandleReference extends PhantomReference<Object> {
         private final Object nativeSpace;
         private final Object destroyFunc;
 
-        public GraalHPyHandleReference(int id, Object referent, ReferenceQueue<Object> q, Object nativeSpace, Object desctroyFunc) {
+        public GraalHPyHandleReference(int id, Object referent, ReferenceQueue<Object> q, Object nativeSpace, Object destroyFunc) {
             super(referent, q);
             this.id = id;
             this.nativeSpace = nativeSpace;
-            this.destroyFunc = desctroyFunc;
+            this.destroyFunc = destroyFunc;
         }
 
         public Object getNativeSpace() {

graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/AsyncHandler.java

@@ -40,9 +40,14 @@
  */
 package com.oracle.graal.python.runtime;
 
+import java.lang.ref.PhantomReference;
+import java.lang.ref.Reference;
+import java.lang.ref.ReferenceQueue;
 import java.lang.ref.WeakReference;
 import java.util.Arrays;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
@@ -64,8 +69,10 @@
 import com.oracle.graal.python.util.Supplier;
 import com.oracle.truffle.api.CompilerAsserts;
 import com.oracle.truffle.api.CompilerDirectives;
+import com.oracle.truffle.api.CompilerDirectives.TruffleBoundary;
 import com.oracle.truffle.api.RootCallTarget;
 import com.oracle.truffle.api.TruffleLanguage;
+import com.oracle.truffle.api.TruffleLogger;
 import com.oracle.truffle.api.frame.VirtualFrame;
 import com.oracle.truffle.api.profiles.BranchProfile;
 
@@ -127,6 +134,7 @@ public final void execute(PythonContext context) {
     }
 
     private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(4, new ThreadFactory() {
+        @Override
         public Thread newThread(Runnable r) {
             Thread t = Executors.defaultThreadFactory().newThread(r);
             t.setDaemon(true);
@@ -147,6 +155,7 @@ public AsyncRunnable(Supplier<AsyncAction> actionSupplier) {
             this.actionSupplier = actionSupplier;
         }
 
+        @Override
         public void run() {
             AsyncAction asyncAction = actionSupplier.get();
             if (asyncAction != null) {
@@ -292,4 +301,118 @@ private void processAsyncActions() {
     public void shutdown() {
         executorService.shutdownNow();
     }
+
+    public static class SharedFinalizer {
+        private static final TruffleLogger LOGGER = PythonLanguage.getLogger(SharedFinalizer.class);
+
+        private final PythonContext pythonContext;
+        private final ReferenceQueue<Object> queue = new ReferenceQueue<>();
+
+        /**
+         * This is a Set of references to keep them alive after their gc collected referents.
+         */
+        private final ConcurrentMap<FinalizableReference, FinalizableReference> liveReferencesSet = new ConcurrentHashMap<>();
+
+        public SharedFinalizer(PythonContext context) {
+            this.pythonContext = context;
+        }
+
+        /**
+         * Finalizable references is a utility class for freeing resources that {@link Runtime#gc()}
+         * is unaware of, such as of heap allocation through native interface. Resources that can be
+         * freed with {@link Runtime#gc()} should not extend this class.
+         */
+        public abstract static class FinalizableReference extends PhantomReference<Object> {
+            private final Object reference;
+            private boolean released;
+
+            public FinalizableReference(Object referent, Object reference, SharedFinalizer sharedFinalizer) {
+                super(referent, sharedFinalizer.queue);
+                assert reference != null;
+                this.reference = reference;
+                addLiveReference(sharedFinalizer, this);
+            }
+
+            /**
+             * We'll keep a reference for the FinalizableReference object until the async handler
+             * schedule the collect process.
+             */
+            @TruffleBoundary
+            private static void addLiveReference(SharedFinalizer sharedFinalizer, FinalizableReference ref) {
+                sharedFinalizer.liveReferencesSet.put(ref, ref);
+            }
+
+            /**
+             * 
+             * @return the undelying reference which is usually a native pointer.
+             */
+            public final Object getReference() {
+                return reference;
+            }
+
+            public final boolean isReleased() {
+                return released;
+            }
+
+            /**
+             * Mark the FinalizableReference as freed in case it has been freed elsewhare. This will
+             * avoid double-freeing the reference.
+             */
+            public final void markReleased() {
+                this.released = true;
+            }
+
+            /**
+             * This implements the proper way to free the allocated resources associated with the
+             * reference.
+             */
+            public abstract AsyncHandler.AsyncAction release();
+        }
+
+        static class SharedFinalizerErrorCallback implements AsyncHandler.AsyncAction {
+
+            private final Exception exception;
+            private final FinalizableReference referece; // problematic reference
+
+            SharedFinalizerErrorCallback(FinalizableReference referece, Exception e) {
+                this.exception = e;
+                this.referece = referece;
+            }
+
+            @Override
+            public void execute(PythonContext context) {
+                LOGGER.severe(String.format("Error during async action for %s caused by %s", referece.getClass().getSimpleName(), exception.getMessage()));
+            }
+        }
+
+        /**
+         * We register the Async action once on the first encounter of a creation of
+         * {@link FinalizableReference}. This will reduce unnecessary Async thread load when there
+         * isn't any enqueued references.
+         */
+        public void registerAsyncAction() {
+            pythonContext.registerAsyncAction(() -> {
+                Reference<? extends Object> reference = null;
+                try {
+                    reference = queue.remove();
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+                if (reference instanceof FinalizableReference) {
+                    FinalizableReference object = (FinalizableReference) reference;
+                    try {
+                        liveReferencesSet.remove(object);
+                        if (object.isReleased()) {
+                            return null;
+                        }
+                        return object.release();
+                    } catch (Exception e) {
+                        return new SharedFinalizerErrorCallback(object, e);
+                    }
+                }
+                return null;
+            });
+
+        }
+    }
 }

graalpython/com.oracle.graal.python/src/com/oracle/graal/python/runtime/PythonContext.java

@@ -256,6 +256,7 @@ private static final class AtExitHook {
     // The context-local resources
     private final PosixResources resources;
     private final AsyncHandler handler;
+    private final AsyncHandler.SharedFinalizer sharedFinalizer;
 
     // decides if we run the async weakref callbacks and destructors
     private boolean gcEnabled = true;
@@ -272,6 +273,7 @@ public PythonContext(PythonLanguage language, TruffleLanguage.Env env, PythonCor
         this.env = env;
         this.resources = new PosixResources();
         this.handler = new AsyncHandler(this);
+        this.sharedFinalizer = new AsyncHandler.SharedFinalizer(this);
         this.optionValues = PythonOptions.createOptionValuesStorage(env);
         this.resources.setEnv(env);
         this.in = env.in();
@@ -1089,4 +1091,8 @@ public boolean isGcEnabled() {
     public void setGcEnabled(boolean flag) {
         gcEnabled = flag;
     }
+
+    public AsyncHandler.SharedFinalizer getSharedFinalizer() {
+        return sharedFinalizer;
+    }
 }