Changeset 9351 in josm

Timestamp:

2016-01-09T15:41:47+01:00 (9 years ago)

Author:

simon04

Message:

Refactoring: use Fork/Join Tasks for parallel execution

Location:

trunk/src/org/openstreetmap/josm

Files:

• data/osm/MultipolygonBuilder.java (modified) (5 diffs)
• data/osm/visitor/paint/StyledMapRenderer.java (modified) (8 diffs)
• tools/Utils.java (modified) (2 diffs)

trunk/src/org/openstreetmap/josm/data/osm/MultipolygonBuilder.java

@@ -14 +14 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinTask;
 import java.util.concurrent.Future;
+import java.util.concurrent.RecursiveTask;
 
 import org.openstreetmap.josm.tools.Geometry;
@@ -31 +33 @@
 public class MultipolygonBuilder {
 
-    private static final Pair<Integer, ExecutorService> THREAD_POOL =
-            Utils.newThreadPool("multipolygon_creation.numberOfThreads", "multipolygon-builder-%d", Thread.NORM_PRIORITY);
+    private static final ForkJoinPool THREAD_POOL =
+            Utils.newForkJoinPool("multipolygon_creation.numberOfThreads", "multipolygon-builder-%d", Thread.NORM_PRIORITY);
 
     /**
@@ -303 +305 @@
      */
     private static List<PolygonLevel> findOuterWaysMultiThread(List<JoinedPolygon> boundaryWays) {
-        final List<PolygonLevel> result = new ArrayList<>();
-        final List<Worker> tasks = new ArrayList<>();
-        final int bucketsize = Math.max(32, boundaryWays.size()/THREAD_POOL.a/3);
-        final int noBuckets = (boundaryWays.size() + bucketsize - 1) / bucketsize;
-        final boolean singleThread = THREAD_POOL.a == 1 || noBuckets == 1;
-        for (int i = 0; i < noBuckets; i++) {
-            int from = i*bucketsize;
-            int to = Math.min((i+1)*bucketsize, boundaryWays.size());
-            List<PolygonLevel> target = singleThread ? result : new ArrayList<PolygonLevel>(to - from);
-            tasks.add(new Worker(boundaryWays, from, to, target));
-        }
-        if (singleThread) {
-            try {
-                for (Worker task : tasks) {
-                    if (task.call() == null) {
-                        return null;
-                    }
-                }
-            } catch (Exception ex) {
-                throw new RuntimeException(ex);
-            }
-        } else if (!tasks.isEmpty()) {
-            try {
-                for (Future<List<PolygonLevel>> future : THREAD_POOL.b.invokeAll(tasks)) {
-                    List<PolygonLevel> res = future.get();
-                    if (res == null) {
-                        return null;
-                    }
-                    result.addAll(res);
-                }
-            } catch (InterruptedException | ExecutionException ex) {
-                throw new RuntimeException(ex);
-            }
-        }
-        return result;
-    }
-
-    private static class Worker implements Callable<List<PolygonLevel>> {
+        return THREAD_POOL.invoke(new Worker(boundaryWays, 0, boundaryWays.size(), new ArrayList<PolygonLevel>(),
+                Math.max(32, boundaryWays.size() / THREAD_POOL.getParallelism() / 3)));
+    }
+
+    private static class Worker extends RecursiveTask<List<PolygonLevel>> {
 
         private final List<JoinedPolygon> input;
@@ -346 +315 @@
         private final int to;
         private final List<PolygonLevel> output;
-
-        Worker(List<JoinedPolygon> input, int from, int to, List<PolygonLevel> output) {
+        private final int directExecutionTaskSize;
+
+        Worker(List<JoinedPolygon> input, int from, int to, List<PolygonLevel> output, int directExecutionTaskSize) {
             this.input = input;
             this.from = from;
             this.to = to;
             this.output = output;
+            this.directExecutionTaskSize = directExecutionTaskSize;
         }
 
@@ -407 +378 @@
 
         @Override
-        public List<PolygonLevel> call() throws Exception {
+        protected List<PolygonLevel> compute() {
+            if (to - from < directExecutionTaskSize) {
+                return computeDirectly();
+            } else {
+                final Collection<ForkJoinTask<List<PolygonLevel>>> tasks = new ArrayList<>();
+                for (int fromIndex = from; fromIndex < to; fromIndex += directExecutionTaskSize) {
+                    final List<PolygonLevel> output = new ArrayList<>();
+                    tasks.add(new Worker(input, fromIndex, Math.min(fromIndex + directExecutionTaskSize, to), output, directExecutionTaskSize));
+                }
+                for (ForkJoinTask<List<PolygonLevel>> task : tasks) {
+                    output.addAll(task.join());
+                }
+                return output;
+            }
+        }
+
+        List<PolygonLevel> computeDirectly() {
             for (int i = from; i < to; i++) {
                 if (processOuterWay(0, input, output, input.get(i)) == null) {

trunk/src/org/openstreetmap/josm/data/osm/visitor/paint/StyledMapRenderer.java

@@ -33 +33 @@
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinTask;
+import java.util.concurrent.RecursiveTask;
 
 import javax.swing.AbstractButton;
@@ -78 +77 @@
 import org.openstreetmap.josm.tools.Geometry.AreaAndPerimeter;
 import org.openstreetmap.josm.tools.ImageProvider;
-import org.openstreetmap.josm.tools.Pair;
 import org.openstreetmap.josm.tools.Utils;
 
@@ -87 +85 @@
 public class StyledMapRenderer extends AbstractMapRenderer {
 
-    private static final Pair<Integer, ExecutorService> THREAD_POOL =
-            Utils.newThreadPool("mappaint.StyledMapRenderer.style_creation.numberOfThreads", "styled-map-renderer-%d", Thread.NORM_PRIORITY);
+    private static final ForkJoinPool THREAD_POOL =
+            Utils.newForkJoinPool("mappaint.StyledMapRenderer.style_creation.numberOfThreads", "styled-map-renderer-%d", Thread.NORM_PRIORITY);
 
     /**
@@ -1761 +1759 @@
     }
 
-    private class ComputeStyleListWorker implements Callable<List<StyleRecord>>, Visitor {
+    private class ComputeStyleListWorker extends RecursiveTask<List<StyleRecord>> implements Visitor {
         private final List<? extends OsmPrimitive> input;
-        private final int from;
-        private final int to;
         private final List<StyleRecord> output;
 
         private final ElemStyles styles = MapPaintStyles.getStyles();
+        private final int directExecutionTaskSize;
 
         private final boolean drawArea = circum <= Main.pref.getInteger("mappaint.fillareas", 10000000);
@@ -1776 +1773 @@
          * Constructs a new {@code ComputeStyleListWorker}.
          * @param input the primitives to process
-         * @param from first index of <code>input</code> to use
-         * @param to last index + 1
          * @param output the list of styles to which styles will be added
+         * @param directExecutionTaskSize the threshold deciding whether to subdivide the tasks
          */
-        ComputeStyleListWorker(final List<? extends OsmPrimitive> input, int from, int to, List<StyleRecord> output) {
+        ComputeStyleListWorker(final List<? extends OsmPrimitive> input, List<StyleRecord> output, int directExecutionTaskSize) {
             this.input = input;
-            this.from = from;
-            this.to = to;
             this.output = output;
+            this.directExecutionTaskSize = directExecutionTaskSize;
             this.styles.setDrawMultipolygon(drawMultipolygon);
         }
 
         @Override
-        public List<StyleRecord> call() throws Exception {
+        protected List<StyleRecord> compute() {
+            if (input.size() <= directExecutionTaskSize) {
+                return computeDirectly();
+            } else {
+                final Collection<ForkJoinTask<List<StyleRecord>>> tasks = new ArrayList<>();
+                for (int fromIndex = 0; fromIndex < input.size(); fromIndex += directExecutionTaskSize) {
+                    final int toIndex = Math.min(fromIndex + directExecutionTaskSize, input.size());
+                    final List<StyleRecord> output = new ArrayList<>(directExecutionTaskSize);
+                    tasks.add(new ComputeStyleListWorker(input.subList(fromIndex, toIndex), output, directExecutionTaskSize).fork());
+                }
+                for (ForkJoinTask<List<StyleRecord>> task : tasks) {
+                    output.addAll(task.join());
+                }
+                return output;
+            }
+        }
+
+        public List<StyleRecord> computeDirectly() {
             MapCSSStyleSource.STYLE_SOURCE_LOCK.readLock().lock();
             try {
-                for (int i = from; i < to; i++) {
-                    OsmPrimitive osm = input.get(i);
+                for (final OsmPrimitive osm : input) {
                     if (osm.isDrawable()) {
                         osm.accept(this);
@@ -1853 +1864 @@
     }
 
-    private class ConcurrentTasksHelper {
-
-        private final List<StyleRecord> allStyleElems;
-
-        ConcurrentTasksHelper(List<StyleRecord> allStyleElems) {
-            this.allStyleElems = allStyleElems;
-        }
-
-        void process(List<? extends OsmPrimitive> prims) {
-            final List<ComputeStyleListWorker> tasks = new ArrayList<>();
-            final int bucketsize = Math.max(100, prims.size()/THREAD_POOL.a/3);
-            final int noBuckets = (prims.size() + bucketsize - 1) / bucketsize;
-            final boolean singleThread = THREAD_POOL.a == 1 || noBuckets == 1;
-            for (int i = 0; i < noBuckets; i++) {
-                int from = i*bucketsize;
-                int to = Math.min((i+1)*bucketsize, prims.size());
-                List<StyleRecord> target = singleThread ? allStyleElems : new ArrayList<StyleRecord>(to - from);
-                tasks.add(new ComputeStyleListWorker(prims, from, to, target));
-            }
-            if (singleThread) {
-                try {
-                    for (ComputeStyleListWorker task : tasks) {
-                        task.call();
-                    }
-                } catch (Exception ex) {
-                    throw new RuntimeException(ex);
-                }
-            } else if (!tasks.isEmpty()) {
-                try {
-                    for (Future<List<StyleRecord>> future : THREAD_POOL.b.invokeAll(tasks)) {
-                        allStyleElems.addAll(future.get());
-                    }
-                } catch (InterruptedException | ExecutionException ex) {
-                    throw new RuntimeException(ex);
-                }
-            }
-        }
-    }
-
     @Override
     public void render(final DataSet data, boolean renderVirtualNodes, Bounds bounds) {
@@ -1914 +1886 @@
             final List<StyleRecord> allStyleElems = new ArrayList<>(nodes.size()+ways.size()+relations.size());
 
-            ConcurrentTasksHelper helper = new ConcurrentTasksHelper(allStyleElems);
-
             // Need to process all relations first.
             // Reason: Make sure, ElemStyles.getStyleCacheWithRange is
@@ -1921 +1891 @@
             // (Could be synchronized, but try to avoid this for
             // performance reasons.)
-            helper.process(relations);
-            helper.process(new CompositeList<>(nodes, ways));
+            THREAD_POOL.invoke(new ComputeStyleListWorker(relations, allStyleElems,
+                    Math.max(20, relations.size() / THREAD_POOL.getParallelism() / 3)));
+            THREAD_POOL.invoke(new ComputeStyleListWorker(new CompositeList<>(nodes, ways), allStyleElems,
+                    Math.max(100, (nodes.size() + ways.size()) / THREAD_POOL.getParallelism() / 3)));
 
             if (benchmark) {

trunk/src/org/openstreetmap/josm/tools/Utils.java

@@ -51 +51 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinWorkerThread;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicLong;
@@ -1411 +1413 @@
 
     /**
-     * Returns a pair containing the number of threads (n), and a thread pool (if n > 1) to perform
-     * multi-thread computation in the context of the given preference key.
-     * @param pref The preference key
+     * Returns a {@link ForkJoinPool} with the parallelism given by the preference key.
+     * @param pref The preference key to determine parallelism
      * @param nameFormat see {@link #newThreadFactory(String, int)}
      * @param threadPriority see {@link #newThreadFactory(String, int)}
-     * @return a pair containing the number of threads (n), and a thread pool (if n > 1, null otherwise)
-     * @since 7423
-     */
-    public static Pair<Integer, ExecutorService> newThreadPool(String pref, String nameFormat, int threadPriority) {
+     * @return a {@link ForkJoinPool}
+     */
+    public static ForkJoinPool newForkJoinPool(String pref, final String nameFormat, final int threadPriority) {
         int noThreads = Main.pref.getInteger(pref, Runtime.getRuntime().availableProcessors());
-        ExecutorService pool = noThreads <= 1 ? null : Executors.newFixedThreadPool(noThreads, newThreadFactory(nameFormat, threadPriority));
-        return new Pair<>(noThreads, pool);
+        return new ForkJoinPool(noThreads, new ForkJoinPool.ForkJoinWorkerThreadFactory() {
+            final AtomicLong count = new AtomicLong(0);
+            @Override
+            public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
+                final ForkJoinWorkerThread thread = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
+                thread.setName(String.format(Locale.ENGLISH, nameFormat, count.getAndIncrement()));
+                thread.setPriority(threadPriority);
+                return thread;
+            }
+        }, null, true);
     }