git.gag.com Git - debian/openrocket/blob - core/src/net/sf/openrocket/startup/ConcurrentLoadingThrustCurveMotorSetDatabase.java

   1 package net.sf.openrocket.startup;
   2
   3 import java.io.File;
   4 import java.io.IOException;
   5 import java.io.InputStream;
   6 import java.util.List;
   7 import java.util.concurrent.ExecutorService;
   8 import java.util.concurrent.LinkedBlockingQueue;
   9 import java.util.concurrent.ThreadFactory;
  10 import java.util.concurrent.ThreadPoolExecutor;
  11 import java.util.concurrent.TimeUnit;
  12 import java.util.concurrent.atomic.AtomicInteger;
  13
  14 import net.sf.openrocket.database.ThrustCurveMotorSet;
  15 import net.sf.openrocket.database.ThrustCurveMotorSetDatabase;
  16 import net.sf.openrocket.file.iterator.DirectoryIterator;
  17 import net.sf.openrocket.file.iterator.FileIterator;
  18 import net.sf.openrocket.file.motor.MotorLoaderHelper;
  19 import net.sf.openrocket.gui.util.SimpleFileFilter;
  20 import net.sf.openrocket.gui.util.SwingPreferences;
  21 import net.sf.openrocket.logging.LogHelper;
  22 import net.sf.openrocket.motor.Motor;
  23 import net.sf.openrocket.motor.ThrustCurveMotor;
  24 import net.sf.openrocket.util.BugException;
  25 import net.sf.openrocket.util.Pair;
  26
  27 /**
  28  * Load motors in parallel using a three stage pipeline.
  29  *
  30  * Stage 1: single thread managed by the ThrustCurveMotorSetDatabase.  This thread generates
  31  *          one object for each thrust curve motor file and puts it in the second stage.
  32  *
  33  * Stage 2: multiple threads which process individual files.  Each process takes
  34  *          a single motor file and parses out the list of motors it contains.
  35  *          The list of motors is queued up for the third stage to process.
  36  *
  37  * Stage 3: single thread which processes the list of motors generated in stage 2.
  38  *          This thread puts all the motors from the list in the motor set database.
  39  *
  40  * It is important that stage 3 be done with a single thread because ThrustCurveMotorSetDatabase
  41  * is not thread safe.  Even if synchronization were to be done, it is unlikely that parallelizing
  42  * this process would improve anything.
  43  *
  44  *
  45  */
  46 public class ConcurrentLoadingThrustCurveMotorSetDatabase extends ThrustCurveMotorSetDatabase {
  47
  48         private static final LogHelper log = Application.getLogger();
  49         private final String thrustCurveDirectory;
  50
  51         /** Block motor loading for this many milliseconds */
  52         // Block motor loading for 1.5 seconds to allow window painting to be faster
  53         private static AtomicInteger blockLoading = new AtomicInteger(1500);
  54
  55         public ConcurrentLoadingThrustCurveMotorSetDatabase(String thrustCurveDirectory) {
  56                 // configure ThrustCurveMotorSetDatabase as true so we get our own thread in
  57                 // loadMotors.
  58                 super(true);
  59                 this.thrustCurveDirectory = thrustCurveDirectory;
  60         }
  61
  62         @Override
  63         protected void loadMotors() {
  64
  65                 // Block loading until timeout occurs or database is taken into use
  66                 log.info("Blocking motor loading while starting up");
  67                 /*
  68                 while (!inUse && blockLoading.addAndGet(-100) > 0) {
  69                         try {
  70                                 Thread.sleep(100);
  71                         } catch (InterruptedException e) {
  72                         }
  73                 }
  74                 */
  75                 log.info("Blocking ended, inUse=" + inUse + " blockLoading=" + blockLoading.get());
  76
  77                 BookKeeping keeper = new BookKeeping();
  78                 keeper.start();
  79
  80                 try {
  81                         keeper.waitForFinish();
  82                 }
  83                 catch ( InterruptedException iex ) {
  84                         throw new BugException(iex);
  85                 }
  86
  87                 keeper = null;
  88
  89         }
  90
  91         private void addAll( List<Motor> motors ) {
  92                 for (Motor m : motors) {
  93                         addMotor( (ThrustCurveMotor) m);
  94                 }
  95         }
  96
  97         /**
  98          * A class which holds all the threading data.
  99          * Implemented as an inner class so we can easily jettison the references when
 100          * the processing is terminated.
 101          *
 102          */
 103         private class BookKeeping {
 104
 105                 /*
 106                  * Executor for Stage 3.
 107                  */
 108                 private final ExecutorService writerThread;
 109
 110                 /*
 111                  * Executor for Stage 2.
 112                  */
 113                 private final ExecutorService loaderPool;
 114
 115                 /*
 116                  * Runnable used for Stage 1.
 117                  */
 118                 private final WorkGenerator workGenerator;
 119
 120                 private long startTime;
 121
 122                 /*
 123                  * Number of thrust curves loaded
 124                  */
 125                 private int thrustCurveCount = 0;
 126
 127                 /*
 128                  * Number of files processed.
 129                  */
 130                 private int fileCount = 0;
 131
 132                 /*
 133                  * We have to hold on to the zip file iterator which is used to load
 134                  * the system motor files until all processing is done.  This is because
 135                  * closing the iterator prematurely causes all the InputStreams opened
 136                  * with it to close.
 137                  */
 138                 private FileIterator iterator;
 139
 140                 private BookKeeping() {
 141
 142                         writerThread = new ThreadPoolExecutor(1,1,200, TimeUnit.SECONDS,
 143                                         new LinkedBlockingQueue<Runnable>(),
 144                                         new ThreadFactory() {
 145                                 @Override
 146                                 public Thread newThread(Runnable r) {
 147                                         Thread t = new Thread(r,"MotorWriterThread");
 148                                         return t;
 149                                 }
 150                         });
 151
 152                         loaderPool = new ThreadPoolExecutor(10,10, 2, TimeUnit.SECONDS,
 153                                         new LinkedBlockingQueue<Runnable>(),
 154                                         new ThreadFactory() {
 155                                 int threadCount = 0;
 156                                 @Override
 157                                 public Thread newThread(Runnable r) {
 158                                         Thread t = new Thread(r,"MotorLoaderPool-" + threadCount++);
 159                                         return t;
 160                                 }
 161                         });
 162
 163                         workGenerator = new WorkGenerator();
 164
 165                 }
 166
 167                 private void start() {
 168
 169                         startTime = System.currentTimeMillis();
 170
 171                         log.info("Starting motor loading from " + thrustCurveDirectory + " in background thread.");
 172
 173                         // Run the work generator - in this thread.
 174                         workGenerator.run();
 175
 176                 }
 177
 178                 private void waitForFinish() throws InterruptedException {
 179                         try {
 180                                 loaderPool.shutdown();
 181                                 loaderPool.awaitTermination(90, TimeUnit.SECONDS);
 182                                 writerThread.shutdown();
 183                                 writerThread.awaitTermination(90, TimeUnit.SECONDS);
 184                         }
 185                         finally {
 186                                 iterator.close();
 187                         }
 188
 189                         long endTime = System.currentTimeMillis();
 190
 191                         int distinctMotorCount = 0;
 192                         int distinctThrustCurveCount = 0;
 193                         distinctMotorCount = motorSets.size();
 194                         for (ThrustCurveMotorSet set : motorSets) {
 195                                 distinctThrustCurveCount += set.getMotorCount();
 196                         }
 197
 198                         log.info("Motor loading done, took " + (endTime - startTime) + " ms to load "
 199                                         + fileCount + " files/directories containing "
 200                                         + thrustCurveCount + " thrust curves which contained "
 201                                         + distinctMotorCount + " distinct motors with "
 202                                         + distinctThrustCurveCount + " distinct thrust curves.");
 203
 204                 }
 205
 206
 207                 private class WorkGenerator implements Runnable {
 208
 209                         @Override
 210                         public void run() {
 211                                 // Start loading
 212                                 log.info("Loading motors from " + thrustCurveDirectory);
 213
 214                                 iterator = DirectoryIterator.findDirectory(thrustCurveDirectory,
 215                                                 new SimpleFileFilter("", false, "eng", "rse"));
 216
 217                                 // Load the packaged thrust curves
 218                                 if (iterator == null) {
 219                                         throw new IllegalStateException("Thrust curve directory " + thrustCurveDirectory +
 220                                                         "not found, distribution built wrong");
 221                                 }
 222
 223                                 while( iterator.hasNext() ) {
 224                                         Pair<String,InputStream> f = iterator.next();
 225                                         MotorLoader loader = new MotorLoader( f.getV(), f.getU() );
 226                                         loaderPool.execute(loader);
 227                                         fileCount ++;
 228                                 }
 229
 230                                 // Load the user-defined thrust curves
 231                                 for (File file : ((SwingPreferences) Application.getPreferences()).getUserThrustCurveFiles()) {
 232                                         log.info("Loading motors from " + file);
 233                                         MotorLoader loader = new MotorLoader( file );
 234                                         loaderPool.execute(loader);
 235                                         fileCount++;
 236                                 }
 237                         }
 238                 }
 239
 240                 private class MotorLoader implements Runnable {
 241
 242                         private final InputStream is;
 243                         private final String fileName;
 244
 245                         private final File file;
 246
 247                         public MotorLoader( File file ) {
 248                                 super();
 249                                 this.file = file;
 250                                 this.is = null;
 251                                 this.fileName = null;
 252                         }
 253
 254                         public MotorLoader(InputStream is, String fileName) {
 255                                 super();
 256                                 this.file = null;
 257                                 this.is = is;
 258                                 this.fileName = fileName;
 259                         }
 260
 261                         @Override
 262                         public void run() {
 263                                 log.debug("Loading motor from " + fileName);
 264
 265                                 try {
 266                                         List<Motor> motors;
 267                                         if ( file == null ) {
 268                                                 motors = MotorLoaderHelper.load(is, fileName);
 269                                         } else {
 270                                                 motors = MotorLoaderHelper.load(file);
 271                                         }
 272                                         writerThread.submit( new MotorInserter(motors));
 273                                 }
 274                                 finally {
 275                                         if ( is != null ) {
 276                                                 try {
 277                                                         is.close();
 278                                                 } catch ( IOException iex ) {
 279                                                 }
 280                                         }
 281                                 }
 282                         }
 283                 }
 284
 285                 private class MotorInserter implements Runnable {
 286
 287                         private final List<Motor> motors;
 288
 289                         MotorInserter( List<Motor> motors ) {
 290                                 this.motors = motors;
 291                         }
 292
 293                         @Override
 294                         public void run() {
 295                                 thrustCurveCount += motors.size();
 296                                 ConcurrentLoadingThrustCurveMotorSetDatabase.this.addAll(motors);
 297                         }
 298
 299                 }
 300         }
 301
 302 }