X-Git-Url: https://gerrit.simantics.org/r/gitweb?a=blobdiff_plain;f=bundles%2Forg.simantics.db.layer0%2Fsrc%2Forg%2Fsimantics%2Fdb%2Flayer0%2FStandardRealm.java;h=68d002458b3af80e448f19f2f4314dc6b4794af2;hb=82ed7c74;hp=c0fbb669430f79f6142a71810d9fe3d6ab25b1b9;hpb=969bd23cab98a79ca9101af33334000879fb60c5;p=simantics%2Fplatform.git

diff --git a/bundles/org.simantics.db.layer0/src/org/simantics/db/layer0/StandardRealm.java b/bundles/org.simantics.db.layer0/src/org/simantics/db/layer0/StandardRealm.java
index c0fbb6694..68d002458 100644
--- a/bundles/org.simantics.db.layer0/src/org/simantics/db/layer0/StandardRealm.java
+++ b/bundles/org.simantics.db.layer0/src/org/simantics/db/layer0/StandardRealm.java
@@ -1,164 +1,199 @@
-package org.simantics.db.layer0;
-
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
-
-import org.simantics.db.common.utils.Logger;
-import org.simantics.scl.runtime.SCLContext;
-import org.simantics.scl.runtime.tuple.Tuple0;
-import org.simantics.simulator.variable.Realm;
-
-abstract public class StandardRealm<Node, Engine extends StandardEngine<Node>> implements Realm {
-	
-    String id;
-    Thread executorThread;
-    ExecutorService executor = new ThreadPoolExecutor(0, 1, 60, TimeUnit.SECONDS,
-            new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
-        @Override
-        public Thread newThread(Runnable r) {
-            executorThread = new Thread(r);
-            return executorThread;
-        }
-    });
-    
-    Semaphore beginSyncExec = new Semaphore(0);
-    Semaphore endSyncExec = new Semaphore(0);
-    
-    Engine engine;
-    StandardSessionManager<Node, Engine> sessionManager;
-    StandardNodeManager<Node, Engine> nodeManager;
-    
-    Runnable scheduleSyncExec = new Runnable() {
-        @Override
-        public void run() {
-            beginSyncExec.release();
-            try {
-                endSyncExec.acquire();
-            } catch (InterruptedException e) {
-            }
-        }
-    };
-    
-    protected StandardRealm(StandardSessionManager<Node, Engine> sessionManager, Engine engine, String id) {
-    	this.sessionManager = sessionManager;
-        this.engine = engine;
-        this.id = id;
-        this.nodeManager = createManager();
-    }
-    
-    abstract protected StandardNodeManager<Node, Engine> createManager();
-    
-    protected String getSCLContextKey() {
-    	return getClass().getSimpleName();
-    }
-
-    public String getId() {
-        return id;
-    }
-    
-    public Engine getEngine() {
-        return engine;
-    }
-
-    public Thread getThread() {
-        return executorThread;
-    }
-    
-    @SuppressWarnings({ "rawtypes", "unchecked" })
-    public Object syncExec(Function fun) throws InterruptedException {
-    	
-        executor.execute(scheduleSyncExec);
-        SCLContext context = SCLContext.getCurrent();
-        Engine oldConnection = (Engine)context.put(getSCLContextKey(), engine);
-        
-        try {
-            beginSyncExec.acquire();
-            Thread oldThread = executorThread;
-            executorThread = Thread.currentThread();
-            try {
-                return fun.apply(Tuple0.INSTANCE);
-            } finally {
-                executorThread = oldThread;
-                endSyncExec.release();
-            }
-        } finally {
-            context.put(getSCLContextKey(), oldConnection);
-        }
-    }
-    
-    @SuppressWarnings("rawtypes")
-    public void asyncExec(final Function fun) {
-        executor.execute(new Runnable() {
-            @SuppressWarnings("unchecked")
-            @Override
-            public void run() {
-                SCLContext context = SCLContext.getCurrent();
-                context.put(getSCLContextKey(), engine);
-                fun.apply(Tuple0.INSTANCE);
-            }
-        });
-    }
-
-    @Override
-    public void syncExec(Runnable runnable) throws InterruptedException {
-    	
-        if(executorThread == Thread.currentThread()) {
-            try {
-                runnable.run();
-            } catch (Throwable t) {
-            	Logger.defaultLogError(t);
-            } finally {
-            }
-            return;
-        }
-
-    	executor.execute(scheduleSyncExec);
-        
-        beginSyncExec.acquire();
-        Thread oldThread = executorThread;
-        executorThread = Thread.currentThread();
-        try {
-            runnable.run();
-        } catch (Throwable t) {
-        	Logger.defaultLogError(t);
-        } finally {
-            executorThread = oldThread;
-            endSyncExec.release();
-        }
-    }
-
-    @Override
-    public void asyncExec(Runnable runnable) {
-
-    	if(executorThread == Thread.currentThread()) {
-            try {
-                runnable.run();
-            } catch (Throwable t) {
-            	Logger.defaultLogError(t);
-            } finally {
-            }
-            return;
-        }
-    	
-        executor.execute(runnable);
-    }
-    
-    public void close() {
-        sessionManager.removeRealm(id);
-        executor.shutdown();
-        try {
-            executor.awaitTermination(500L, TimeUnit.MILLISECONDS);
-        } catch (InterruptedException e) {
-        }
-    }
-
-    public StandardNodeManager<Node, Engine> getNodeManager() {
-        return nodeManager;
-    }
-    
-}
+package org.simantics.db.layer0;
+
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import org.simantics.db.common.utils.Logger;
+import org.simantics.scl.runtime.SCLContext;
+import org.simantics.scl.runtime.tuple.Tuple0;
+import org.simantics.simulator.variable.Realm;
+
+abstract public class StandardRealm<Node, Engine extends StandardEngine<Node>> implements Realm {
+
+    private String id;
+    private Thread executorThread;
+    private StandardRealmThreadFactory factory = new StandardRealmThreadFactory(this);
+    private ThreadPoolExecutor executor = new ThreadPoolExecutor(0, 1, 60, TimeUnit.SECONDS,
+            new LinkedBlockingQueue<Runnable>(), factory);
+    private Semaphore beginSyncExec = new Semaphore(0);
+    private Semaphore endSyncExec = new Semaphore(0);
+    
+    private Engine engine;
+    private StandardNodeManager<Node, Engine> nodeManager;
+    
+    private Runnable scheduleSyncExec = new Runnable() {
+        @Override
+        public void run() {
+            beginSyncExec.release();
+            try {
+                endSyncExec.acquire();
+            } catch (InterruptedException e) {
+            }
+        }
+    };
+    
+    protected StandardRealm(Engine engine, String id) {
+        this.engine = engine;
+        this.id = id;
+        this.nodeManager = createManager();
+    }
+
+    abstract protected StandardNodeManager<Node, Engine> createManager();
+    
+    protected String getSCLContextKey() {
+    	return getClass().getSimpleName();
+    }
+
+    public String getId() {
+        return id;
+    }
+    
+    public Engine getEngine() {
+        return engine;
+    }
+
+    public Thread getThread() {
+        return executorThread;
+    }
+    
+    @SuppressWarnings({ "rawtypes", "unchecked" })
+    public Object syncExec(Function fun) throws InterruptedException {
+    	
+        executor.execute(scheduleSyncExec);
+        SCLContext context = SCLContext.getCurrent();
+        Engine oldConnection = (Engine)context.put(getSCLContextKey(), engine);
+        
+        try {
+            beginSyncExec.acquire();
+            Thread oldThread = executorThread;
+            executorThread = Thread.currentThread();
+            try {
+                return fun.apply(Tuple0.INSTANCE);
+            } finally {
+                executorThread = oldThread;
+                endSyncExec.release();
+            }
+        } finally {
+            context.put(getSCLContextKey(), oldConnection);
+        }
+    }
+    
+    @SuppressWarnings("rawtypes")
+    public void asyncExec(final Function fun) {
+        executor.execute(new Runnable() {
+            @SuppressWarnings("unchecked")
+            @Override
+            public void run() {
+                SCLContext context = SCLContext.getCurrent();
+                context.put(getSCLContextKey(), engine);
+                fun.apply(Tuple0.INSTANCE);
+            }
+        });
+    }
+
+    @Override
+    public void syncExec(Runnable runnable) throws InterruptedException {
+    	
+        if(executorThread == Thread.currentThread()) {
+            try {
+                runnable.run();
+            } catch (Throwable t) {
+            	Logger.defaultLogError(t);
+            } finally {
+            }
+            return;
+        }
+
+    	executor.execute(scheduleSyncExec);
+        
+        beginSyncExec.acquire();
+        Thread oldThread = executorThread;
+        executorThread = Thread.currentThread();
+        try {
+            runnable.run();
+        } catch (Throwable t) {
+        	Logger.defaultLogError(t);
+        } finally {
+            executorThread = oldThread;
+            endSyncExec.release();
+        }
+    }
+
+    @Override
+    public void asyncExec(Runnable runnable) {
+
+    	if(executorThread == Thread.currentThread()) {
+            try {
+                runnable.run();
+            } catch (Throwable t) {
+            	Logger.defaultLogError(t);
+            } finally {
+            }
+            return;
+        }
+    	
+        executor.execute(runnable);
+    }
+    
+    public void close() {
+        executor.shutdown();
+        try {
+            if (!executor.awaitTermination(500L, TimeUnit.MILLISECONDS)) {
+                List<Runnable> runnablesLeft = executor.shutdownNow();
+                if (!runnablesLeft.isEmpty()) {
+                    getLogger().info("Runnables left for realm " + this + " after executor shutdown! " + runnablesLeft);
+                }
+            }
+        } catch (InterruptedException e) {
+            getLogger().info("Could not shutdown executor " + executor + " for realm " + this, e);
+        }
+        
+        factory.clear();
+        factory = null;
+        // Should never be true
+        if (!executorThread.isAlive())
+            executorThread.interrupt();
+        executorThread = null;
+        executor = null;
+        
+        // Clear nodeManager
+        nodeManager.clear();
+        nodeManager = null;
+    }
+
+    public StandardNodeManager<Node, Engine> getNodeManager() {
+        return nodeManager;
+    }
+
+    public abstract org.slf4j.Logger getLogger();
+
+    private void setExecutorThread(Thread t) {
+        executorThread = t;
+    }
+
+    private static class StandardRealmThreadFactory implements ThreadFactory {
+        
+        private StandardRealm<?, ?> realm;
+
+        public StandardRealmThreadFactory(StandardRealm<?, ?> realm) {
+            this.realm = realm;
+        }
+        
+        @Override
+        public Thread newThread(Runnable r) {
+            Thread t = new Thread(r);
+            realm.setExecutorThread(t);
+            return t;
+        }
+        
+        void clear() {
+            realm = null;
+        }
+    }
+}