git.gag.com Git - debian/openrocket/blob - core/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.Iterator;
   7 import java.util.List;
   8 import java.util.Map;
   9 import java.util.concurrent.Callable;
  10 import java.util.concurrent.ExecutionException;
  11 import java.util.concurrent.ExecutorService;
  12 import java.util.concurrent.Future;
  13 import java.util.concurrent.LinkedBlockingQueue;
  14 import java.util.concurrent.ThreadFactory;
  15 import java.util.concurrent.ThreadPoolExecutor;
  16 import java.util.concurrent.TimeUnit;
  17
  18 import net.sf.openrocket.util.BugException;
  19
  20 /**
  21  * An implementation of a ParallelFunctionCache that evaluates function values
  22  * in parallel and caches them.  This allows pre-calculating possibly required
  23  * function values beforehand.  If values are not required after all, the
  24  * computation can be aborted assuming the function evaluation supports it.
  25  * <p>
  26  * Note that while this class handles threads and abstracts background execution,
  27  * the public methods themselves are NOT thread-safe and should be called from
  28  * only one thread at a time.
  29  *
  30  * @author Sampo Niskanen <sampo.niskanen@iki.fi>
  31  */
  32 public class ParallelExecutorCache implements ParallelFunctionCache {
  33
  34         private final Map<Point, Double> functionCache = new HashMap<Point, Double>();
  35         private final Map<Point, Future<Double>> futureMap = new HashMap<Point, Future<Double>>();
  36
  37         private ExecutorService executor;
  38
  39         private Function function;
  40
  41
  42         /**
  43          * Construct a cache that uses the same number of computational threads as there are
  44          * processors available.
  45          */
  46         public ParallelExecutorCache() {
  47                 this(Runtime.getRuntime().availableProcessors());
  48         }
  49
  50         /**
  51          * Construct a cache that uses the specified number of computational threads for background
  52          * computation.  The threads that are created are marked as daemon threads.
  53          *
  54          * @param threadCount   the number of threads to use in the executor.
  55          */
  56         public ParallelExecutorCache(int threadCount) {
  57                 this(new ThreadPoolExecutor(threadCount, threadCount, 60, TimeUnit.SECONDS,
  58                                 new LinkedBlockingQueue<Runnable>(),
  59                                 new ThreadFactory() {
  60                                         @Override
  61                                         public Thread newThread(Runnable r) {
  62                                                 Thread t = new Thread(r);
  63                                                 t.setDaemon(true);
  64                                                 return t;
  65                                         }
  66                                 }));
  67         }
  68
  69         /**
  70          * Construct a cache that uses the specified ExecutorService for managing
  71          * computational threads.
  72          *
  73          * @param executor      the executor to use for function evaluations.
  74          */
  75         public ParallelExecutorCache(ExecutorService executor) {
  76                 this.executor = executor;
  77         }
  78
  79
  80
  81         @Override
  82         public void compute(Collection<Point> points) {
  83                 for (Point p : points) {
  84                         compute(p);
  85                 }
  86         }
  87
  88
  89         @Override
  90         public void compute(Point point) {
  91
  92                 if (isOutsideRange(point)) {
  93                         // Point is outside of range
  94                         return;
  95                 }
  96
  97                 if (functionCache.containsKey(point)) {
  98                         // Function has already been evaluated at the point
  99                         return;
 100                 }
 101
 102                 if (futureMap.containsKey(point)) {
 103                         // Function is being evaluated at the point
 104                         return;
 105                 }
 106
 107                 // Submit point for evaluation
 108                 FunctionCallable callable = new FunctionCallable(function, point);
 109                 Future<Double> future = executor.submit(callable);
 110                 futureMap.put(point, future);
 111         }
 112
 113
 114         @Override
 115         public void waitFor(Collection<Point> points) throws InterruptedException, OptimizationException {
 116                 for (Point p : points) {
 117                         waitFor(p);
 118                 }
 119         }
 120
 121
 122         @Override
 123         public void waitFor(Point point) throws InterruptedException, OptimizationException {
 124                 if (isOutsideRange(point)) {
 125                         return;
 126                 }
 127
 128                 if (functionCache.containsKey(point)) {
 129                         return;
 130                 }
 131
 132                 Future<Double> future = futureMap.get(point);
 133                 if (future == null) {
 134                         throw new IllegalStateException("waitFor called for " + point + " but it is not being computed");
 135                 }
 136
 137                 try {
 138                         double value = future.get();
 139                         functionCache.put(point, value);
 140                 } catch (ExecutionException e) {
 141                         Throwable cause = e.getCause();
 142                         if (cause instanceof InterruptedException) {
 143                                 throw (InterruptedException) cause;
 144                         }
 145                         if (cause instanceof OptimizationException) {
 146                                 throw (OptimizationException) cause;
 147                         }
 148                         if (cause instanceof RuntimeException) {
 149                                 throw (RuntimeException) cause;
 150                         }
 151
 152                         throw new BugException("Function threw unknown exception while processing", e);
 153                 }
 154         }
 155
 156
 157
 158         @Override
 159         public List<Point> abort(Collection<Point> points) {
 160                 List<Point> computed = new ArrayList<Point>(Math.min(points.size(), 10));
 161
 162                 for (Point p : points) {
 163                         if (abort(p)) {
 164                                 computed.add(p);
 165                         }
 166                 }
 167
 168                 return computed;
 169         }
 170
 171
 172
 173         @Override
 174         public boolean abort(Point point) {
 175                 if (isOutsideRange(point)) {
 176                         return false;
 177                 }
 178
 179                 if (functionCache.containsKey(point)) {
 180                         return true;
 181                 }
 182
 183                 Future<Double> future = futureMap.remove(point);
 184                 if (future == null) {
 185                         throw new IllegalStateException("abort called for " + point + " but it is not being computed");
 186                 }
 187
 188                 if (future.isDone()) {
 189                         // Evaluation has been completed, store value in cache
 190                         try {
 191                                 double value = future.get();
 192                                 functionCache.put(point, value);
 193                                 return true;
 194                         } catch (Exception e) {
 195                                 return false;
 196                         }
 197                 } else {
 198                         // Cancel the evaluation
 199                         future.cancel(true);
 200                         return false;
 201                 }
 202         }
 203
 204
 205         @Override
 206         public void abortAll() {
 207                 Iterator<Point> iterator = futureMap.keySet().iterator();
 208                 while (iterator.hasNext()) {
 209                         Point point = iterator.next();
 210                         Future<Double> future = futureMap.get(point);
 211                         iterator.remove();
 212
 213                         if (future.isDone()) {
 214                                 // Evaluation has been completed, store value in cache
 215                                 try {
 216                                         double value = future.get();
 217                                         functionCache.put(point, value);
 218                                 } catch (Exception e) {
 219                                         // Ignore
 220                                 }
 221                         } else {
 222                                 // Cancel the evaluation
 223                                 future.cancel(true);
 224                         }
 225                 }
 226         }
 227
 228
 229         @Override
 230         public double getValue(Point point) {
 231                 if (isOutsideRange(point)) {
 232                         return Double.MAX_VALUE;
 233                 }
 234
 235                 Double d = functionCache.get(point);
 236                 if (d == null) {
 237                         throw new IllegalStateException(point + " is not in function cache.  " +
 238                                         "functionCache=" + functionCache + "  futureMap=" + futureMap);
 239                 }
 240                 return d;
 241         }
 242
 243
 244         @Override
 245         public Function getFunction() {
 246                 return function;
 247         }
 248
 249         @Override
 250         public void setFunction(Function function) {
 251                 this.function = function;
 252                 clearCache();
 253         }
 254
 255         @Override
 256         public void clearCache() {
 257                 List<Point> list = new ArrayList<Point>(futureMap.keySet());
 258                 abort(list);
 259                 functionCache.clear();
 260         }
 261
 262
 263         public ExecutorService getExecutor() {
 264                 return executor;
 265         }
 266
 267
 268         /**
 269          * Check whether a point is outside of the valid optimization range.
 270          */
 271         private boolean isOutsideRange(Point p) {
 272                 int n = p.dim();
 273                 for (int i = 0; i < n; i++) {
 274                         double d = p.get(i);
 275                         // Include NaN in disallowed range
 276                         if (!(d >= 0.0 && d <= 1.0)) {
 277                                 return true;
 278                         }
 279                 }
 280                 return false;
 281         }
 282
 283
 284         /**
 285          * A Callable that evaluates a function at a specific point and returns the result.
 286          */
 287         private class FunctionCallable implements Callable<Double> {
 288                 private final Function calledFunction;
 289                 private final Point point;
 290
 291                 public FunctionCallable(Function function, Point point) {
 292                         this.calledFunction = function;
 293                         this.point = point;
 294                 }
 295
 296                 @Override
 297                 public Double call() throws InterruptedException, OptimizationException {
 298                         return calledFunction.evaluate(point);
 299                 }
 300         }
 301
 302
 303 }