bundles/org.simantics.simulator.toolkit/src/org/simantics/simulator/toolkit/StandardRealm.java

   1 package org.simantics.simulator.toolkit;
   2
   3 import java.util.List;
   4 import java.util.concurrent.LinkedBlockingQueue;
   5 import java.util.concurrent.Semaphore;
   6 import java.util.concurrent.ThreadFactory;
   7 import java.util.concurrent.ThreadPoolExecutor;
   8 import java.util.concurrent.TimeUnit;
   9 import java.util.function.Function;
  10
  11 import org.simantics.scl.runtime.SCLContext;
  12 import org.simantics.scl.runtime.tuple.Tuple0;
  13 import org.simantics.simulator.variable.Realm;
  14 import org.slf4j.Logger;
  15 import org.slf4j.LoggerFactory;
  16
  17 abstract public class StandardRealm<Node, Engine extends StandardNodeManagerSupport<Node>> implements Realm {
  18
  19     private static final Logger LOGGER = LoggerFactory.getLogger(StandardRealm.class);
  20
  21     private String id;
  22     protected Thread executorThread;
  23     private StandardRealmThreadFactory factory = new StandardRealmThreadFactory(this);
  24     private ThreadPoolExecutor executor = new ThreadPoolExecutor(0, 1, 60, TimeUnit.SECONDS,
  25             new LinkedBlockingQueue<>(), factory);
  26     private Semaphore beginSyncExec = new Semaphore(0);
  27     private Semaphore endSyncExec = new Semaphore(0);
  28
  29     private Engine engine;
  30     protected StandardNodeManager<Node, Engine> nodeManager;
  31
  32     private Runnable scheduleSyncExec = new Runnable() {
  33         @Override
  34         public void run() {
  35             beginSyncExec.release();
  36             try {
  37                 endSyncExec.acquire();
  38             } catch (InterruptedException e) {
  39             }
  40         }
  41     };
  42
  43     protected StandardRealm(Engine engine, String id) {
  44         this.engine = engine;
  45         this.id = id;
  46         this.nodeManager = createManager();
  47     }
  48
  49     abstract protected StandardNodeManager<Node, Engine> createManager();
  50
  51     protected String getSCLContextKey() {
  52         return getClass().getSimpleName();
  53     }
  54
  55     public String getId() {
  56         return id;
  57     }
  58
  59     public Engine getEngine() {
  60         return engine;
  61     }
  62
  63     public Thread getThread() {
  64         return executorThread;
  65     }
  66
  67     @SuppressWarnings({ "rawtypes", "unchecked" })
  68     public Object syncExec(Function fun) throws InterruptedException {
  69         executor.execute(scheduleSyncExec);
  70         SCLContext context = SCLContext.getCurrent();
  71         Engine oldConnection = (Engine)context.put(getSCLContextKey(), engine);
  72
  73         try {
  74             beginSyncExec.acquire();
  75             Thread oldThread = executorThread;
  76             executorThread = Thread.currentThread();
  77             try {
  78                 return fun.apply(Tuple0.INSTANCE);
  79             } finally {
  80                 executorThread = oldThread;
  81                 endSyncExec.release();
  82             }
  83         } finally {
  84             context.put(getSCLContextKey(), oldConnection);
  85         }
  86     }
  87
  88     @SuppressWarnings("rawtypes")
  89     public void asyncExec(final Function fun) {
  90         executor.execute(new Runnable() {
  91             @SuppressWarnings("unchecked")
  92             @Override
  93             public void run() {
  94                 SCLContext context = SCLContext.getCurrent();
  95                 context.put(getSCLContextKey(), engine);
  96                 fun.apply(Tuple0.INSTANCE);
  97             }
  98         });
  99     }
 100
 101     @Override
 102     public void syncExec(Runnable runnable) throws InterruptedException {
 103         if(executorThread == Thread.currentThread()) {
 104             try {
 105                 runnable.run();
 106             } catch (Throwable t) {
 107                 LOGGER.error("Error executing runnable in realm", t);
 108             } finally {
 109             }
 110             return;
 111         }
 112
 113         executor.execute(scheduleSyncExec);
 114
 115         beginSyncExec.acquire();
 116         Thread oldThread = executorThread;
 117         executorThread = Thread.currentThread();
 118         try {
 119             runnable.run();
 120         } catch (Throwable t) {
 121             LOGGER.error("Error executing runnable in realm", t);
 122         } finally {
 123             executorThread = oldThread;
 124             endSyncExec.release();
 125         }
 126     }
 127
 128     @Override
 129     public void asyncExec(Runnable runnable) {
 130         if(executorThread == Thread.currentThread()) {
 131             try {
 132                 runnable.run();
 133             } catch (Throwable t) {
 134                 LOGGER.error("Error executing runnable in realm", t);
 135             } finally {
 136             }
 137             return;
 138         }
 139
 140         executor.execute(runnable);
 141     }
 142
 143     public void close() {
 144         executor.shutdown();
 145         try {
 146             if (!executor.awaitTermination(500L, TimeUnit.MILLISECONDS)) {
 147                 List<Runnable> runnablesLeft = executor.shutdownNow();
 148                 if (!runnablesLeft.isEmpty()) {
 149                     getLogger().info("Runnables left for realm " + this + " after executor shutdown! " + runnablesLeft);
 150                 }
 151             }
 152         } catch (InterruptedException e) {
 153             getLogger().info("Could not shutdown executor " + executor + " for realm " + this, e);
 154         }
 155
 156         factory.clear();
 157         factory = null;
 158         // Should never be true
 159         if (!executorThread.isAlive())
 160             executorThread.interrupt();
 161         executorThread = null;
 162         executor = null;
 163
 164         // Clear nodeManager
 165         nodeManager.clear();
 166         nodeManager = null;
 167     }
 168
 169     public StandardNodeManager<Node, Engine> getNodeManager() {
 170         return nodeManager;
 171     }
 172
 173     public abstract org.slf4j.Logger getLogger();
 174
 175     private void setExecutorThread(Thread t) {
 176         executorThread = t;
 177     }
 178
 179     private static class StandardRealmThreadFactory implements ThreadFactory {
 180
 181         private StandardRealm<?, ?> realm;
 182
 183         public StandardRealmThreadFactory(StandardRealm<?, ?> realm) {
 184             this.realm = realm;
 185         }
 186
 187         @Override
 188         public Thread newThread(Runnable r) {
 189             Thread t = new Thread(r);
 190             realm.setExecutorThread(t);
 191             return t;
 192         }
 193
 194         void clear() {
 195             realm = null;
 196         }
 197     }
 198
 199 }