gerrit.simantics Code Review - simantics/platform.git/blob

   1 package org.simantics.simulator.toolkit;
   2
   3 import java.math.BigDecimal;
   4 import java.util.ArrayList;
   5 import java.util.concurrent.Callable;
   6 import java.util.concurrent.CopyOnWriteArrayList;
   7 import java.util.concurrent.Semaphore;
   8
   9 import org.simantics.simulator.ExperimentState;
  10 import org.slf4j.Logger;
  11 import org.slf4j.LoggerFactory;
  12
  13 /**
  14  * @author Antti Villberg
  15  * @since 1.34.0
  16  */
  17 abstract public class DynamicExperimentThread extends Thread {
  18
  19         private static final Logger LOGGER = LoggerFactory.getLogger(DynamicExperimentThread.class);
  20
  21         private CopyOnWriteArrayList<DynamicExperimentThreadListener> listeners = new CopyOnWriteArrayList<>();
  22
  23         private ExperimentState state = StandardExperimentStates.CREATED;
  24
  25         private long runStart = 0;
  26
  27         private double desiredRealtimeRatio = 1000.0;
  28         private double obtainedRealtimeRatio = 1.0;
  29
  30         private long runTimeNs = 0;
  31         private long endTimeNs = 0;
  32         protected long simulationStepNs = 0;
  33         protected double stepInSeconds = 1.0;
  34
  35         public DynamicExperimentThread() {
  36         }
  37
  38         private void updateTimes() {
  39                 long time = System.nanoTime();
  40                 long elapsed = time-runStart;
  41
  42                 obtainedRealtimeRatio = longToDoubleDivision(runTimeNs, elapsed);
  43         }
  44
  45         private long rt;
  46         private long rt_l;
  47
  48         protected double longToDoubleDivision(long l1, long l2) {
  49                 rt = l1 / l2;
  50                 rt_l = l1 % l2;
  51                 double d = ((double)rt_l)/((double)l2);
  52                 d += (double)rt;
  53                 return d;
  54         }
  55
  56         protected ArrayList<Runnable> tasks = new ArrayList<>();
  57
  58         abstract public void step(double stepLengthNanoSeconds);
  59
  60         public boolean inState(Class<? extends ExperimentState> state) {
  61                 return state.isInstance(this.state);
  62         }
  63
  64         public void initialize() throws Exception {
  65         }
  66
  67         public void deinitialize() throws Exception {
  68         }
  69
  70         long stepTime = 0;
  71         long taskTime = 0;
  72
  73         @Override
  74         public void run() {
  75
  76                 try {
  77
  78                         try {
  79
  80                                 initialize();
  81
  82                                 try {
  83                                         runReally();
  84                                 } catch (Exception e) {
  85                                         LOGGER.error("Unhandled exception while running simulation thread", e);
  86                                 }
  87
  88                         } catch (Exception e) {
  89                                 LOGGER.error("Unhandled exception while initializing simulation thread", e);
  90                         }
  91
  92                 } finally {
  93
  94                         try {
  95                                 deinitialize();
  96                         } catch (Exception e) {
  97                                 LOGGER.error("Error while deinitializing simulation thread", e);
  98                         }
  99
 100                 }
 101
 102         }
 103
 104         protected boolean inActiveState() {
 105                 return !(
 106                                 inState(StandardExperimentStates.Disposed.class)
 107                                 || inState(StandardExperimentStates.Disposing.class)
 108                                 //|| inState(StandardExperimentStates.Failure.class)
 109                                 || inState(StandardExperimentStates.ToBeDisposed.class)
 110                                 );
 111         }
 112
 113         private void runReally() {
 114
 115                 while(inActiveState()) {
 116
 117                         if(inState(StandardExperimentStates.Running.class)) {
 118
 119                                 long asd = System.nanoTime();
 120                                 step(simulationStepNs);
 121                                 stepTime += System.nanoTime() - asd;
 122                                 runTimeNs += simulationStepNs;
 123                                 updateTimes();
 124                                 long asd2 = System.nanoTime();
 125                                 runTasks();
 126                                 taskTime += System.nanoTime() - asd2;
 127
 128                                 //System.err.println(" st = " + 1e-9*stepTime + " tt = " + 1e-9*taskTime);
 129
 130                                 while(obtainedRealtimeRatio > desiredRealtimeRatio) {
 131                                         int ran = runTasks();
 132                                         if(ran == 0) {
 133                                                 long elapsed = System.nanoTime()-runStart;
 134                                                 long deltaNs = BigDecimal.valueOf(runTimeNs).divide(BigDecimal.valueOf(desiredRealtimeRatio)).longValue() - elapsed;
 135
 136                                                 long deltaMs = deltaNs / 1000000;
 137                                                 int deltaNsRem = (int)(deltaNs % 1000000);
 138
 139                                                 if(deltaNs > 0) {
 140                                                         synchronized(tasks) {
 141                                                                 try {
 142                                                                         tasks.wait(deltaMs, deltaNsRem);
 143                                                                 } catch (InterruptedException e) {
 144                                                                         e.printStackTrace();
 145                                                                 }
 146                                                         }
 147                                                 }
 148                                         }
 149                                         updateTimes();
 150                                 }
 151
 152                         } else {
 153
 154                                 while(!inState(StandardExperimentStates.Running.class) && inActiveState()) {
 155
 156                                         synchronized(tasks) {
 157                                                 int ran = runTasks();
 158                                                 if(ran == 0) {
 159                                                         try {
 160                                                                 tasks.wait(Integer.MAX_VALUE);
 161                                                         } catch (InterruptedException e) {
 162                                                                 e.printStackTrace();
 163                                                         }
 164                                                 }
 165                                         }
 166
 167                                 }
 168
 169                         }
 170
 171                         if(runTimeNs >= endTimeNs && inActiveState())
 172                                 changeState(StandardExperimentStates.STOPPED);
 173
 174                 }
 175
 176         }
 177
 178         Thread executorThread = this;
 179
 180         Semaphore beginSyncExec = new Semaphore(0);
 181         Semaphore endSyncExec = new Semaphore(0);
 182
 183         Runnable scheduleSyncExec = () -> {
 184                 beginSyncExec.release();
 185                 try {
 186                         endSyncExec.acquire();
 187                 } catch (InterruptedException e) {
 188                 }
 189         };
 190
 191         public int runTasks() {
 192                 ArrayList<Runnable> todo = new ArrayList<>();
 193                 synchronized(tasks) {
 194                         todo.addAll(tasks);
 195                         tasks.clear();
 196                 }
 197                 todo.forEach(Runnable::run);
 198                 return todo.size();
 199         }
 200
 201         public void queue(Runnable runnable) {
 202                 synchronized(tasks) {
 203                         tasks.add(runnable);
 204                         tasks.notify();
 205                 }
 206         }
 207
 208         public <T> T syncExec(Callable<T> callable) throws InterruptedException {
 209
 210                 if(executorThread == Thread.currentThread()) {
 211                         try {
 212                                 return callable.call();
 213                         } catch (Throwable t) {
 214                                 LOGGER.error("syncExec in current thread failed", t);
 215                                 return null;
 216                         } finally {
 217                         }
 218                 }
 219
 220                 queue(scheduleSyncExec);
 221
 222                 beginSyncExec.acquire();
 223                 Thread oldThread = executorThread;
 224                 executorThread = Thread.currentThread();
 225                 try {
 226                         return callable.call();
 227                 } catch (Throwable t) {
 228                         LOGGER.error("syncExec failed", t);
 229                         return null;
 230                 } finally {
 231                         executorThread = oldThread;
 232                         endSyncExec.release();
 233                 }
 234
 235         }
 236
 237         public void asyncExec(Runnable runnable) {
 238
 239                 if(executorThread == Thread.currentThread()) {
 240                         try {
 241                                 runnable.run();
 242                         } catch (Throwable t) {
 243                                 LOGGER.error("asyncExec failed", t);
 244                         } finally {
 245                         }
 246                         return;
 247                 }
 248
 249                 queue(runnable);
 250
 251         }
 252
 253         public void setSimulationStepNs(long ns) {
 254                 simulationStepNs = ns;
 255                 stepInSeconds = BigDecimal.valueOf(simulationStepNs).multiply(BigDecimal.valueOf(1e-9)).doubleValue();
 256         }
 257
 258         public void runDuration(long ns) {
 259                 runStart = System.nanoTime();
 260                 runTimeNs = 0;
 261                 endTimeNs = ns;
 262                 synchronized(tasks) {
 263                         changeState(StandardExperimentStates.RUNNING);
 264                         tasks.notify();
 265                 }
 266         }
 267
 268         public ExperimentState getExperimentState() {
 269                 return state;
 270         }
 271
 272         public void changeState(ExperimentState state) {
 273                 this.state = state;
 274                 fireStateChanged(state);
 275         }
 276
 277         public void addListener(DynamicExperimentThreadListener listener) {
 278                 if(!listeners.contains(listener))
 279                         listeners.add(listener);
 280         }
 281
 282         public void removeListener(DynamicExperimentThreadListener listener) {
 283                 listeners.remove(listener);
 284         }
 285
 286         protected void fireAfterStep() {
 287                 listeners.forEach(DynamicExperimentThreadListener::afterStep);
 288         }
 289
 290         protected void fireBeforeStep() {
 291                 listeners.forEach(DynamicExperimentThreadListener::beforeStep);
 292         }
 293
 294         protected void fireStateChanged(ExperimentState newState) {
 295                 listeners.forEach(l -> l.stateChanged(newState));
 296         }
 297
 298 }