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     private boolean disposed = false;
  32
  33     private Runnable scheduleSyncExec = new Runnable() {
  34         @Override
  35         public void run() {
  36             beginSyncExec.release();
  37             try {
  38                 endSyncExec.acquire();
  39             } catch (InterruptedException e) {
  40             }
  41         }
  42     };
  43
  44     protected StandardRealm(Engine engine, String id) {
  45         this.engine = engine;
  46         this.id = id;
  47         this.nodeManager = createManager();
  48     }
  49
  50     protected StandardNodeManager<Node, Engine> createManager(Node root) {
  51         return new StandardNodeManager<Node, Engine>(this, root);
  52     }
  53
  54     protected StandardNodeManager<Node, Engine> createManager() {
  55         return createManager(createRootNode());
  56     }
  57
  58     /*
  59      * For compatibility reasons. Existing implementations implement createManager() directly and in that case this is not needed.
  60      * New implementations should not implement createManager() but rather implement this.
  61      */
  62     protected Node createRootNode() {
  63         throw new UnsupportedOperationException();
  64     }
  65
  66     protected String getSCLContextKey() {
  67         return getClass().getSimpleName();
  68     }
  69
  70     public String getId() {
  71         return id;
  72     }
  73
  74     public Engine getEngine() {
  75         return engine;
  76     }
  77
  78     public Thread getThread() {
  79         return executorThread;
  80     }
  81
  82     @SuppressWarnings({ "rawtypes", "unchecked" })
  83     public Object syncExec(Function fun) throws InterruptedException {
  84         executor.execute(scheduleSyncExec);
  85         SCLContext context = SCLContext.getCurrent();
  86         Engine oldConnection = (Engine)context.put(getSCLContextKey(), engine);
  87
  88         try {
  89             beginSyncExec.acquire();
  90             Thread oldThread = executorThread;
  91             executorThread = Thread.currentThread();
  92             try {
  93                 return fun.apply(Tuple0.INSTANCE);
  94             } finally {
  95                 executorThread = oldThread;
  96                 endSyncExec.release();
  97             }
  98         } finally {
  99             context.put(getSCLContextKey(), oldConnection);
 100         }
 101     }
 102
 103     @SuppressWarnings("rawtypes")
 104     public void asyncExec(final Function fun) {
 105         executor.execute(new Runnable() {
 106             @SuppressWarnings("unchecked")
 107             @Override
 108             public void run() {
 109                 SCLContext context = SCLContext.getCurrent();
 110                 context.put(getSCLContextKey(), engine);
 111                 fun.apply(Tuple0.INSTANCE);
 112             }
 113         });
 114     }
 115
 116     @Override
 117     public void syncExec(Runnable runnable) throws InterruptedException {
 118         if(executorThread == Thread.currentThread()) {
 119             try {
 120                 runnable.run();
 121             } catch (Throwable t) {
 122                 LOGGER.error("Error executing runnable in realm", t);
 123             } finally {
 124             }
 125             return;
 126         }
 127
 128         executor.execute(scheduleSyncExec);
 129
 130         beginSyncExec.acquire();
 131         Thread oldThread = executorThread;
 132         executorThread = Thread.currentThread();
 133         try {
 134             runnable.run();
 135         } catch (Throwable t) {
 136             LOGGER.error("Error executing runnable in realm", t);
 137         } finally {
 138             executorThread = oldThread;
 139             endSyncExec.release();
 140         }
 141     }
 142
 143     @Override
 144     public void asyncExec(Runnable runnable) {
 145         if(executorThread == Thread.currentThread()) {
 146             try {
 147                 runnable.run();
 148             } catch (Throwable t) {
 149                 LOGGER.error("Error executing runnable in realm", t);
 150             } finally {
 151             }
 152             return;
 153         }
 154
 155         executor.execute(runnable);
 156     }
 157
 158     public void close() {
 159         executor.shutdown();
 160         try {
 161             if (!executor.awaitTermination(500L, TimeUnit.MILLISECONDS)) {
 162                 List<Runnable> runnablesLeft = executor.shutdownNow();
 163                 if (!runnablesLeft.isEmpty()) {
 164                     getLogger().info("Runnables left for realm " + this + " after executor shutdown! " + runnablesLeft);
 165                 }
 166             }
 167         } catch (InterruptedException e) {
 168             getLogger().info("Could not shutdown executor " + executor + " for realm " + this, e);
 169         }
 170
 171         factory.clear();
 172         factory = null;
 173         // Should never be true
 174         if (!executorThread.isAlive())
 175             executorThread.interrupt();
 176         executorThread = null;
 177         executor = null;
 178
 179         // Clear nodeManager
 180         nodeManager.clear();
 181         nodeManager = null;
 182
 183         disposed = true;
 184
 185     }
 186
 187     public boolean isDisposed() {
 188         return disposed;
 189     }
 190
 191     public StandardNodeManager<Node, Engine> getNodeManager() {
 192         return nodeManager;
 193     }
 194
 195     public abstract org.slf4j.Logger getLogger();
 196
 197     private void setExecutorThread(Thread t) {
 198         executorThread = t;
 199     }
 200
 201     private static class StandardRealmThreadFactory implements ThreadFactory {
 202
 203         private StandardRealm<?, ?> realm;
 204
 205         public StandardRealmThreadFactory(StandardRealm<?, ?> realm) {
 206             this.realm = realm;
 207         }
 208
 209         @Override
 210         public Thread newThread(Runnable r) {
 211             Thread t = new Thread(r);
 212             realm.setExecutorThread(t);
 213             return t;
 214         }
 215
 216         void clear() {
 217             realm = null;
 218         }
 219     }
 220
 221 }