Added ExecutorService for Landmarks (graphhopper#995)

Author: boldtrn

core/src/main/java/com/graphhopper/routing/lm/LMAlgoFactoryDecorator.java

@@ -41,8 +41,8 @@
 
 import java.io.IOException;
 import java.util.*;
+import java.util.concurrent.ExecutorCompletionService;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
 
 import static com.graphhopper.util.Parameters.Landmark.DISABLE;
 
@@ -264,6 +264,7 @@ public RoutingAlgorithm createAlgo(Graph g, AlgorithmOptions opts) {
      * @see com.graphhopper.routing.ch.CHAlgoFactoryDecorator#prepare(StorableProperties) for a very similar method
      */
     public boolean loadOrDoWork(final StorableProperties properties) {
+        ExecutorCompletionService completionService = new ExecutorCompletionService<>(threadPool);
         int counter = 0;
         boolean prepared = false;
         for (final PrepareLandmarks plm : preparations) {
@@ -274,7 +275,7 @@ public boolean loadOrDoWork(final StorableProperties properties) {
             prepared = true;
             LOGGER.info(counter + "/" + getPreparations().size() + " calling LM prepare.doWork for " + plm.getWeighting() + " ... (" + Helper.getMemInfo() + ")");
             final String name = AbstractWeighting.weightingToFileName(plm.getWeighting());
-            threadPool.execute(new Runnable() {
+            completionService.submit(new Runnable() {
                 @Override
                 public void run() {
                     String errorKey = Landmark.PREPARE + "error." + name;
@@ -289,17 +290,17 @@ public void run() {
                         properties.put(errorKey, ex.getMessage());
                     }
                 }
-            });
+            }, name);
         }
 
         threadPool.shutdown();
         try {
-            if (!threadPool.awaitTermination(Integer.MAX_VALUE, TimeUnit.DAYS))
-                threadPool.shutdownNow();
-
-        } catch (InterruptedException ie) {
+            for(int i = 0; i < getPreparations().size(); i++){
+                completionService.take().get();
+            }
+        } catch (Exception e) {
             threadPool.shutdownNow();
-            throw new RuntimeException(ie);
+            throw new RuntimeException(e);
         }
         return prepared;
     }

core/src/main/java/com/graphhopper/routing/lm/LandmarkStorage.java

@@ -372,6 +372,9 @@ private boolean createLandmarksForSubnetwork(final int startNode, final byte[] s
             // 1b) we have one landmark, now determine the other landmarks
             tmpLandmarkNodeIds[0] = explorer.getLastNode();
             for (int lmIdx = 0; lmIdx < tmpLandmarkNodeIds.length - 1; lmIdx++) {
+                if(Thread.currentThread().isInterrupted()){
+                    throw new RuntimeException("Thread was interrupted");
+                }
                 explorer = new LandmarkExplorer(graph, this, initWeighting, traversalMode);
                 explorer.setFilter(blockedEdges, true, true);
                 // set all current landmarks as start so that the next getLastNode is hopefully a "far away" node
@@ -389,6 +392,9 @@ private boolean createLandmarksForSubnetwork(final int startNode, final byte[] s
 
         // 2) calculate weights for all landmarks -> 'from' and 'to' weight
         for (int lmIdx = 0; lmIdx < tmpLandmarkNodeIds.length; lmIdx++) {
+            if(Thread.currentThread().isInterrupted()){
+                throw new RuntimeException("Thread was interrupted");
+            }
             int lmNodeId = tmpLandmarkNodeIds[lmIdx];
             LandmarkExplorer explorer = new LandmarkExplorer(graph, this, weighting, traversalMode);
             explorer.initFrom(lmNodeId, 0);