git.gag.com Git - debian/openrocket/blob - src/net/sf/openrocket/optimization/general/ParallelExecutorCache.java

   1 package net.sf.openrocket.optimization.general;
   2
   3 import java.util.ArrayList;
   4 import java.util.Collection;
   5 import java.util.HashMap;
   6 import java.util.List;
   7 import java.util.Map;
   8 import java.util.concurrent.Callable;
   9 import java.util.concurrent.ExecutionException;
  10 import java.util.concurrent.ExecutorService;
  11 import java.util.concurrent.Future;
  12 import java.util.concurrent.LinkedBlockingQueue;
  13 import java.util.concurrent.ThreadFactory;
  14 import java.util.concurrent.ThreadPoolExecutor;
  15 import java.util.concurrent.TimeUnit;
  16
  17 /**
  18  * An implementation of a ParallelFunctionCache that evaluates function values
  19  * in parallel and caches them.  This allows pre-calculating possibly required
  20  * function values beforehand.  If values are not required after all, the
  21  * computation can be aborted assuming the function evaluation supports it.
  22  *
  23  * @author Sampo Niskanen <sampo.niskanen@iki.fi>
  24  */
  25 public class ParallelExecutorCache implements ParallelFunctionCache {
  26
  27         private final Map<Point, Double> functionCache = new HashMap<Point, Double>();
  28         private final Map<Point, Future<Double>> futureMap = new HashMap<Point, Future<Double>>();
  29
  30         private ExecutorService executor;
  31
  32         private Function function;
  33
  34
  35
  36         public ParallelExecutorCache() {
  37                 this(Runtime.getRuntime().availableProcessors());
  38         }
  39
  40         public ParallelExecutorCache(int threadCount) {
  41                 executor = new ThreadPoolExecutor(threadCount, threadCount, 60, TimeUnit.SECONDS,
  42                                 new LinkedBlockingQueue<Runnable>(),
  43                                 new ThreadFactory() {
  44                                         @Override
  45                                         public Thread newThread(Runnable r) {
  46                                                 Thread t = new Thread(r);
  47                                                 t.setDaemon(true);
  48                                                 return t;
  49                                         }
  50                                 });
  51         }
  52
  53         public ParallelExecutorCache(ExecutorService executor) {
  54                 this.executor = executor;
  55         }
  56
  57
  58
  59         /**
  60          * Queue a list of function evaluations at the specified points.
  61          *
  62          * @param points        the points at which to evaluate the function.
  63          */
  64         public void compute(Collection<Point> points) {
  65                 for (Point p : points) {
  66                         compute(p);
  67                 }
  68         }
  69
  70
  71         /**
  72          * Queue function evaluation for the specified point.
  73          *
  74          * @param point         the point at which to evaluate the function.
  75          */
  76         public void compute(Point point) {
  77                 if (functionCache.containsKey(point)) {
  78                         // Function has already been evaluated at the point
  79                         return;
  80                 }
  81
  82                 if (futureMap.containsKey(point)) {
  83                         // Function is being evaluated at the point
  84                         return;
  85                 }
  86
  87                 double value = function.preComputed(point);
  88                 if (!Double.isNaN(value)) {
  89                         // Function value was in function cache
  90                         functionCache.put(point, value);
  91                         return;
  92                 }
  93
  94                 // Submit point for evaluation
  95                 FunctionCallable callable = new FunctionCallable(function, point);
  96                 Future<Double> future = executor.submit(callable);
  97                 futureMap.put(point, future);
  98         }
  99
 100
 101         /**
 102          * Wait for a collection of points to be computed.  After calling this method
 103          * the function values are available by calling XXX
 104          *
 105          * @param points        the points to wait for.
 106          * @throws InterruptedException         if this thread was interrupted while waiting.
 107          */
 108         public void waitFor(Collection<Point> points) throws InterruptedException {
 109                 for (Point p : points) {
 110                         waitFor(p);
 111                 }
 112         }
 113
 114         /**
 115          * Wait for a point to be computed.  After calling this method
 116          * the function values are available by calling XXX
 117          *
 118          * @param point         the point to wait for.
 119          * @throws InterruptedException         if this thread was interrupted while waiting.
 120          */
 121         public void waitFor(Point point) throws InterruptedException {
 122                 if (functionCache.containsKey(point)) {
 123                         return;
 124                 }
 125
 126                 Future<Double> future = futureMap.get(point);
 127                 if (future == null) {
 128                         throw new IllegalStateException("waitFor called for " + point + " but it is not being computed");
 129                 }
 130
 131                 try {
 132                         double value = future.get();
 133                         functionCache.put(point, value);
 134                 } catch (ExecutionException e) {
 135                         throw new IllegalStateException("Function threw exception while processing", e.getCause());
 136                 }
 137         }
 138
 139
 140         /**
 141          * Abort the computation of the specified point.  If computation has ended,
 142          * the result is stored in the function cache anyway.
 143          *
 144          * @param points        the points to abort.
 145          * @return                      a list of the points that have been computed anyway
 146          */
 147         public List<Point> abort(Collection<Point> points) {
 148                 List<Point> computed = new ArrayList<Point>(Math.min(points.size(), 10));
 149
 150                 for (Point p : points) {
 151                         if (abort(p)) {
 152                                 computed.add(p);
 153                         }
 154                 }
 155
 156                 return computed;
 157         }
 158
 159
 160         /**
 161          * Abort the computation of the specified point.  If computation has ended,
 162          * the result is stored in the function cache anyway.
 163          *
 164          * @param point         the point to abort.
 165          * @return                      <code>true</code> if the point has been computed anyway, <code>false</code> if not.
 166          */
 167         public boolean abort(Point point) {
 168                 if (functionCache.containsKey(point)) {
 169                         return true;
 170                 }
 171
 172                 Future<Double> future = futureMap.remove(point);
 173                 if (future == null) {
 174                         throw new IllegalStateException("abort called for " + point + " but it is not being computed");
 175                 }
 176
 177                 if (future.isDone()) {
 178                         // Evaluation has been completed, store value in cache
 179                         try {
 180                                 double value = future.get();
 181                                 functionCache.put(point, value);
 182                                 return true;
 183                         } catch (Exception e) {
 184                                 return false;
 185                         }
 186                 } else {
 187                         // Cancel the evaluation
 188                         future.cancel(true);
 189                         return false;
 190                 }
 191         }
 192
 193
 194         public double getValue(Point point) {
 195                 Double d = functionCache.get(point);
 196                 if (d == null) {
 197                         throw new IllegalStateException(point.toString() + " is not in function cache.  " +
 198                                         "functionCache=" + functionCache + "  futureMap=" + futureMap);
 199                 }
 200                 return d;
 201         }
 202
 203
 204
 205         @Override
 206         public Function getFunction() {
 207                 return function;
 208         }
 209
 210         @Override
 211         public void setFunction(Function function) {
 212                 this.function = function;
 213                 clearCache();
 214         }
 215
 216         @Override
 217         public void clearCache() {
 218                 List<Point> list = new ArrayList<Point>(futureMap.keySet());
 219                 abort(list);
 220                 functionCache.clear();
 221         }
 222
 223         public ExecutorService getExecutor() {
 224                 return executor;
 225         }
 226
 227
 228
 229         /**
 230          * A Callable that evaluates a function at a specific point and returns the result.
 231          */
 232         private class FunctionCallable implements Callable<Double> {
 233                 private final Function calledFunction;
 234                 private final Point point;
 235
 236                 public FunctionCallable(Function function, Point point) {
 237                         this.calledFunction = function;
 238                         this.point = point;
 239                 }
 240
 241                 @Override
 242                 public Double call() throws InterruptedException {
 243                         return calledFunction.evaluate(point);
 244                 }
 245         }
 246
 247 }