Merge pull request #877 from knutwalker/max-ce-concurrency

Mats-SX · web-flow · commit e0e37506eb8a · 2019-04-09T18:05:14.000+02:00
Manage Concurrency Handling
diff --git a/algo/src/main/java/org/neo4j/graphalgo/ArticleRankProc.java b/algo/src/main/java/org/neo4j/graphalgo/ArticleRankProc.java
@@ -20,7 +20,6 @@
 
 import org.neo4j.graphalgo.api.Graph;
 import org.neo4j.graphalgo.api.GraphFactory;
-import org.neo4j.graphalgo.api.HugeGraph;
 import org.neo4j.graphalgo.core.GraphLoader;
 import org.neo4j.graphalgo.core.ProcedureConfiguration;
 import org.neo4j.graphalgo.core.utils.Pools;
@@ -29,7 +28,6 @@
 import org.neo4j.graphalgo.core.utils.paged.AllocationTracker;
 import org.neo4j.graphalgo.impl.Algorithm;
 import org.neo4j.graphalgo.impl.pagerank.PageRankAlgorithm;
-import org.neo4j.graphalgo.impl.pagerank.PageRankResult;
 import org.neo4j.graphalgo.impl.results.CentralityResult;
 import org.neo4j.graphalgo.results.CentralityScore;
 import org.neo4j.graphalgo.results.PageRankScore;
@@ -43,7 +41,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.IntStream;
 import java.util.stream.LongStream;
 import java.util.stream.Stream;
 
@@ -161,7 +158,7 @@ private CentralityResult runAlgorithm(
         double dampingFactor = configuration.get(CONFIG_DAMPING, DEFAULT_DAMPING);
         int iterations = configuration.getIterations(DEFAULT_ITERATIONS);
         final int batchSize = configuration.getBatchSize();
-        final int concurrency = configuration.getConcurrency(Pools.getNoThreadsInDefaultPool());
+        final int concurrency = configuration.getConcurrency();
         log.debug("Computing article rank with damping of " + dampingFactor + " and " + iterations + " iterations.");
 
         List<Node> sourceNodes = configuration.get("sourceNodes", new ArrayList<>());
diff --git a/algo/src/main/java/org/neo4j/graphalgo/CentralityUtils.java b/algo/src/main/java/org/neo4j/graphalgo/CentralityUtils.java
@@ -7,11 +7,9 @@
 import org.neo4j.graphalgo.core.utils.ProgressTimer;
 import org.neo4j.graphalgo.core.utils.TerminationFlag;
 import org.neo4j.graphalgo.core.write.Exporter;
-import org.neo4j.graphalgo.impl.pagerank.PageRankResult;
 import org.neo4j.graphalgo.impl.results.CentralityResult;
 import org.neo4j.graphalgo.results.AbstractWriteBuilder;
 import org.neo4j.graphalgo.results.CentralityScore;
-import org.neo4j.graphalgo.results.PageRankScore;
 import org.neo4j.kernel.internal.GraphDatabaseAPI;
 import org.neo4j.logging.Log;
 
diff --git a/algo/src/main/java/org/neo4j/graphalgo/DegreeCentralityProc.java b/algo/src/main/java/org/neo4j/graphalgo/DegreeCentralityProc.java
@@ -161,7 +161,7 @@ private CentralityResult evaluate(
             CentralityScore.Stats.Builder statsBuilder,
             String weightPropertyKey, Direction direction) {
 
-        final int concurrency = configuration.getConcurrency(Pools.getNoThreadsInDefaultPool());
+        final int concurrency = configuration.getConcurrency();
 
         if (direction == Direction.BOTH) {
             direction = Direction.OUTGOING;
diff --git a/algo/src/main/java/org/neo4j/graphalgo/EigenvectorCentralityProc.java b/algo/src/main/java/org/neo4j/graphalgo/EigenvectorCentralityProc.java
@@ -158,7 +158,7 @@ private CentralityResult runAlgorithm(
         double dampingFactor = 1.0;
         int iterations = configuration.getIterations(DEFAULT_ITERATIONS);
         final int batchSize = configuration.getBatchSize();
-        final int concurrency = configuration.getConcurrency(Pools.getNoThreadsInDefaultPool());
+        final int concurrency = configuration.getConcurrency();
         log.debug("Computing eigenvector centrality with " + iterations + " iterations.");
 
         List<Node> sourceNodes = configuration.get("sourceNodes", new ArrayList<>());
diff --git a/algo/src/main/java/org/neo4j/graphalgo/KShortestPathsProc.java b/algo/src/main/java/org/neo4j/graphalgo/KShortestPathsProc.java
@@ -23,7 +23,6 @@
 import org.neo4j.graphalgo.core.ProcedureConfiguration;
 import org.neo4j.graphalgo.core.utils.*;
 import org.neo4j.graphalgo.impl.walking.WalkPath;
-import org.neo4j.graphalgo.impl.yens.WeightedPath;
 import org.neo4j.graphalgo.impl.yens.WeightedPathExporter;
 import org.neo4j.graphalgo.impl.yens.YensKShortestPaths;
 import org.neo4j.graphalgo.results.AbstractResultBuilder;
@@ -39,7 +38,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.IntPredicate;
 import java.util.stream.Stream;
 
 /**
diff --git a/algo/src/main/java/org/neo4j/graphalgo/PageRankProc.java b/algo/src/main/java/org/neo4j/graphalgo/PageRankProc.java
@@ -20,14 +20,12 @@
 
 import org.neo4j.graphalgo.api.Graph;
 import org.neo4j.graphalgo.api.GraphFactory;
-import org.neo4j.graphalgo.api.HugeGraph;
 import org.neo4j.graphalgo.core.GraphLoader;
 import org.neo4j.graphalgo.core.ProcedureConfiguration;
 import org.neo4j.graphalgo.core.utils.Pools;
 import org.neo4j.graphalgo.core.utils.ProgressTimer;
 import org.neo4j.graphalgo.core.utils.TerminationFlag;
 import org.neo4j.graphalgo.core.utils.paged.AllocationTracker;
-import org.neo4j.graphalgo.impl.pagerank.PageRankResult;
 import org.neo4j.graphalgo.impl.Algorithm;
 import org.neo4j.graphalgo.impl.pagerank.PageRankAlgorithm;
 import org.neo4j.graphalgo.impl.results.CentralityResult;
@@ -47,7 +45,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.IntStream;
 import java.util.stream.LongStream;
 import java.util.stream.Stream;
 
@@ -171,7 +168,7 @@ private CentralityResult runAlgorithm(
         double dampingFactor = configuration.get(CONFIG_DAMPING, DEFAULT_DAMPING);
         int iterations = configuration.getIterations(DEFAULT_ITERATIONS);
         final int batchSize = configuration.getBatchSize();
-        final int concurrency = configuration.getConcurrency(Pools.getNoThreadsInDefaultPool());
+        final int concurrency = configuration.getConcurrency();
         log.debug("Computing page rank with damping of " + dampingFactor + " and " + iterations + " iterations.");
 
         List<Node> sourceNodes = configuration.get("sourceNodes", new ArrayList<>());
diff --git a/algo/src/main/java/org/neo4j/graphalgo/ShortestPathProc.java b/algo/src/main/java/org/neo4j/graphalgo/ShortestPathProc.java
@@ -185,7 +185,7 @@ public Stream<DijkstraResult> dijkstra(
 
         return Stream.of(builder.build());
     }
-    
+
     @Procedure("algo.shortestPath.astar.stream")
     @Description("CALL algo.shortestPath.astar.stream(startNode:Node, endNode:Node, weightProperty:String, propertyKeyLat:String," +
     		"propertyKeyLon:String, {nodeQuery:'labelName', relationshipQuery:'relationshipName', direction:'BOTH', defaultValue:1.0}) " +
@@ -220,7 +220,7 @@ public Stream<ShortestPathAStar.Result> astarStream(
                 graph.release();
                 return Stream.empty();
             }
-    		
+
     		return new ShortestPathAStar(graph, api)
     				.withProgressLogger(ProgressLogger.wrap(log, "ShortestPath(AStar)"))
     				.withTerminationFlag(TerminationFlag.wrap(transaction))
diff --git a/algo/src/main/java/org/neo4j/graphalgo/TraverseProc.java b/algo/src/main/java/org/neo4j/graphalgo/TraverseProc.java
@@ -19,7 +19,6 @@
 package org.neo4j.graphalgo;
 
 import org.neo4j.graphalgo.api.Graph;
-import org.neo4j.graphalgo.api.HugeGraph;
 import org.neo4j.graphalgo.core.GraphLoader;
 import org.neo4j.graphalgo.core.ProcedureConfiguration;
 import org.neo4j.graphalgo.core.heavyweight.HeavyGraph;
@@ -28,7 +27,6 @@
 import org.neo4j.graphalgo.impl.walking.WalkPath;
 import org.neo4j.graphalgo.impl.walking.WalkResult;
 import org.neo4j.graphdb.Direction;
-import org.neo4j.graphdb.Path;
 import org.neo4j.kernel.api.KernelTransaction;
 import org.neo4j.kernel.internal.GraphDatabaseAPI;
 import org.neo4j.logging.Log;
diff --git a/algo/src/main/java/org/neo4j/graphalgo/impl/AverageDegreeCentrality.java b/algo/src/main/java/org/neo4j/graphalgo/impl/AverageDegreeCentrality.java
@@ -27,7 +27,6 @@
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.SortedMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -47,7 +46,7 @@ public AverageDegreeCentrality(
             Direction direction
     ) {
         if (concurrency <= 0) {
-            concurrency = Pools.DEFAULT_QUEUE_SIZE;
+            concurrency = Pools.DEFAULT_CONCURRENCY;
         }
 
         this.graph = graph;
diff --git a/algo/src/main/java/org/neo4j/graphalgo/impl/degree/WeightedDegreeCentrality.java b/algo/src/main/java/org/neo4j/graphalgo/impl/degree/WeightedDegreeCentrality.java
@@ -51,7 +51,7 @@ public WeightedDegreeCentrality(
             Direction direction
     ) {
         if (concurrency <= 0) {
-            concurrency = Pools.DEFAULT_QUEUE_SIZE;
+            concurrency = Pools.DEFAULT_CONCURRENCY;
         }
 
         this.graph = graph;
diff --git a/algo/src/main/java/org/neo4j/graphalgo/impl/pagerank/PageRank.java b/algo/src/main/java/org/neo4j/graphalgo/impl/pagerank/PageRank.java
@@ -227,7 +227,7 @@ private ComputeSteps createComputeSteps(
             DegreeComputer degreeComputer,
             long nodeCount) {
         if (concurrency <= 0) {
-            concurrency = Pools.DEFAULT_QUEUE_SIZE;
+            concurrency = Pools.DEFAULT_CONCURRENCY;
         }
         final int expectedParallelism = Math.min(
                 concurrency,
diff --git a/algo/src/test/java/org/neo4j/graphalgo/similarity/SimilarityStreamGeneratorTest.java b/algo/src/test/java/org/neo4j/graphalgo/similarity/SimilarityStreamGeneratorTest.java
@@ -318,4 +318,4 @@ private TerminationFlag terminationFlag() {
         return terminationFlag;
     }
 
-}
+}
diff --git a/benchmark/src/main/java/org/neo4j/graphalgo/bench/LongIteratorsBenchmark.java b/benchmark/src/main/java/org/neo4j/graphalgo/bench/LongIteratorsBenchmark.java
@@ -2,7 +2,7 @@
 
 import org.neo4j.collection.primitive.PrimitiveLongCollections;
 import org.neo4j.collection.primitive.PrimitiveLongIterator;
-import org.neo4j.graphalgo.core.huge.HugeIdMap;
+import org.neo4j.graphalgo.core.huge.loader.HugeIdMap;
 import org.neo4j.graphalgo.core.utils.RandomLongIterator;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
diff --git a/benchmark/src/main/java/org/neo4j/graphalgo/bench/PageRankBenchmarkLdbc.java b/benchmark/src/main/java/org/neo4j/graphalgo/bench/PageRankBenchmarkLdbc.java
@@ -88,7 +88,7 @@ public CentralityResult run() throws Exception {
                 0.85,
                 LongStream.empty(),
                 Pools.DEFAULT,
-                Pools.getNoThreadsInDefaultPool(),
+                Pools.DEFAULT_CONCURRENCY,
                 batchSize)
                 .compute(iterations)
                 .result();
diff --git a/benchmark/src/main/java/org/neo4j/graphalgo/bench/SimilarityStreamGeneratorBenchmark.java b/benchmark/src/main/java/org/neo4j/graphalgo/bench/SimilarityStreamGeneratorBenchmark.java
@@ -19,16 +19,12 @@
 package org.neo4j.graphalgo.bench;
 
 import org.neo4j.graphalgo.core.ProcedureConfiguration;
-import org.neo4j.graphalgo.core.utils.Intersections;
 import org.neo4j.graphalgo.core.utils.TerminationFlag;
 import org.neo4j.graphalgo.similarity.*;
 import org.neo4j.helpers.collection.MapUtil;
 import org.openjdk.jmh.annotations.*;
 import org.openjdk.jmh.infra.Blackhole;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Random;
 import java.util.concurrent.TimeUnit;
 
 import static org.mockito.Mockito.mock;
diff --git a/benchmark/src/main/java/org/neo4j/graphalgo/bench/UFBenchmarkCombination.java b/benchmark/src/main/java/org/neo4j/graphalgo/bench/UFBenchmarkCombination.java
@@ -22,7 +22,6 @@
 import org.neo4j.graphalgo.core.utils.Pools;
 import org.neo4j.graphalgo.core.utils.paged.AllocationTracker;
 import org.neo4j.graphalgo.impl.UnionFindAlgo;
-import org.openjdk.jmh.annotations.Param;
 
 import static org.neo4j.graphalgo.impl.UnionFindAlgo.NOTHING;
 
diff --git a/benchmark/src/main/java/org/neo4j/graphalgo/bench/WeightedPageRankBenchmarkLdbc.java b/benchmark/src/main/java/org/neo4j/graphalgo/bench/WeightedPageRankBenchmarkLdbc.java
@@ -111,7 +111,7 @@ public CentralityResult run() throws Exception {
                 0.85,
                 LongStream.empty(),
                 Pools.DEFAULT,
-                Pools.getNoThreadsInDefaultPool(),
+                Pools.DEFAULT_CONCURRENCY,
                 batchSize,
                 cacheWeights)
                 .compute(iterations)
diff --git a/core/src/main/java/org/neo4j/graphalgo/api/GraphSetup.java b/core/src/main/java/org/neo4j/graphalgo/api/GraphSetup.java
@@ -159,17 +159,7 @@ public GraphSetup(
      * Setup Graph to load any label, any relationship, no property in single threaded mode
      */
     public GraphSetup() {
-        this(null);
-    }
-
-    /**
-     * Setup graph to load any label, any relationship, no property but
-     * in multithreaded mode (depends on the actual executor)
-     *
-     * @param executor executor service
-     */
-    public GraphSetup(ExecutorService executor) {
-        this(null, null, null, 1.0, executor);
+        this(null, null, null, 1.0, null);
     }
 
     public GraphSetup(
diff --git a/core/src/main/java/org/neo4j/graphalgo/core/GraphLoader.java b/core/src/main/java/org/neo4j/graphalgo/core/GraphLoader.java
@@ -73,10 +73,10 @@ public class GraphLoader {
     private ExecutorService executorService;
     private double relWeightDefault = 0.0;
     private double nodeWeightDefault = 0.0;
-    private Map<String, Object> params = new HashMap<>();
+    private final Map<String, Object> params = new HashMap<>();
     private double nodePropDefault = 0.0;
     private int batchSize = ParallelUtil.DEFAULT_BATCH_SIZE;
-    private int concurrency = Pools.DEFAULT_CONCURRENCY;
+    private int concurrency;
 
     private DuplicateRelationshipsStrategy duplicateRelationshipsStrategy = DuplicateRelationshipsStrategy.NONE;
 
@@ -93,6 +93,7 @@ public class GraphLoader {
     public GraphLoader(GraphDatabaseAPI api) {
         this.api = Objects.requireNonNull(api);
         this.executorService = null;
+        this.concurrency = Pools.DEFAULT_CONCURRENCY;
     }
 
     /**
@@ -103,6 +104,7 @@ public GraphLoader(GraphDatabaseAPI api) {
     public GraphLoader(GraphDatabaseAPI api, ExecutorService executorService) {
         this.api = Objects.requireNonNull(api);
         this.executorService = Objects.requireNonNull(executorService);
+        this.concurrency = Pools.DEFAULT_CONCURRENCY;
     }
 
     /**
@@ -185,7 +187,7 @@ public GraphLoader withConcurrency(int newConcurrency) {
         if (newConcurrency <= 0) {
             throw new IllegalArgumentException("concurrency: " + newConcurrency);
         }
-        this.concurrency = newConcurrency;
+        this.concurrency = Pools.allowedConcurrency(newConcurrency);
         return this;
     }
 
@@ -504,7 +506,7 @@ private GraphFactory invokeConstructor(MethodHandle constructor) {
         }
     }
 
-    public  GraphSetup toSetup() {
+    public GraphSetup toSetup() {
         return new GraphSetup(
                     label,
                     null,
diff --git a/core/src/main/java/org/neo4j/graphalgo/core/ProcedureConfiguration.java b/core/src/main/java/org/neo4j/graphalgo/core/ProcedureConfiguration.java
@@ -54,7 +54,7 @@ public class ProcedureConfiguration {
 
     private final Map<String, Object> config;
 
-    public ProcedureConfiguration(Map<String, Object> config) {
+    private ProcedureConfiguration(Map<String, Object> config) {
         this.config = new HashMap<>(config);
     }
 
@@ -252,19 +252,15 @@ public int getBatchSize(int defaultValue) {
         return getNumber(ProcedureConstants.BATCH_SIZE_PARAM, defaultValue).intValue();
     }
 
-    /**
-     * TODO
-     *
-     * @return
-     */
-    public int getConcurrency(int defaultValue) {
-        return getNumber(ProcedureConstants.CONCURRENCY, defaultValue).intValue();
-    }
-
     public int getConcurrency() {
         return getConcurrency(Pools.DEFAULT_CONCURRENCY);
     }
 
+    public int getConcurrency(int defaultValue) {
+        int requestedConcurrency = getNumber(ProcedureConstants.CONCURRENCY, defaultValue).intValue();
+        return Pools.allowedConcurrency(requestedConcurrency);
+    }
+
     public String getDirectionName() {
         return getDirectionName(ProcedureConstants.DIRECTION_DEFAULT);
     }
@@ -389,7 +385,7 @@ public <V> V get(String newKey, String oldKey, V defaultValue) {
         Object value = config.get(newKey);
         if (null == value) {
             value = config.get(oldKey);
-            if(null == value) {
+            if (null == value) {
                 return defaultValue;
             }
         }
diff --git a/core/src/main/java/org/neo4j/graphalgo/core/utils/ConcurrencyConfig.java b/core/src/main/java/org/neo4j/graphalgo/core/utils/ConcurrencyConfig.java
@@ -0,0 +1,33 @@
+package org.neo4j.graphalgo.core.utils;
+
+final class ConcurrencyConfig {
+
+    private static final String PROCESSORS_OVERRIDE_PROPERTY = "neo4j.graphalgo.processors";
+    private static final int MAX_CE_CONCURRENCY = 4;
+
+    final int maxConcurrency;
+    final int defaultConcurrency;
+
+    static ConcurrencyConfig of() {
+        Integer definedProcessors = null;
+        try {
+            definedProcessors = Integer.getInteger(PROCESSORS_OVERRIDE_PROPERTY);
+        } catch (SecurityException ignored) {
+        }
+        if (definedProcessors == null) {
+            definedProcessors = Runtime.getRuntime().availableProcessors();
+        }
+        boolean isOnEnterprise = Package.getPackage("org.neo4j.kernel.impl.enterprise") != null;
+        return new ConcurrencyConfig(definedProcessors, isOnEnterprise);
+    }
+
+    /* test-private */ ConcurrencyConfig(int availableProcessors, boolean isOnEnterprise) {
+        if (isOnEnterprise) {
+            maxConcurrency = Integer.MAX_VALUE;
+            defaultConcurrency = availableProcessors;
+        } else {
+            maxConcurrency = MAX_CE_CONCURRENCY;
+            defaultConcurrency = Math.min(availableProcessors, MAX_CE_CONCURRENCY);
+        }
+    }
+}
diff --git a/core/src/main/java/org/neo4j/graphalgo/core/utils/Pools.java b/core/src/main/java/org/neo4j/graphalgo/core/utils/Pools.java
diff --git a/core/src/main/java/org/neo4j/graphalgo/core/write/Exporter.java b/core/src/main/java/org/neo4j/graphalgo/core/write/Exporter.java
diff --git a/tests/src/test/java/org/neo4j/graphalgo/core/ProcedureConfigurationTest.java b/tests/src/test/java/org/neo4j/graphalgo/core/ProcedureConfigurationTest.java
diff --git a/tests/src/test/java/org/neo4j/graphalgo/core/utils/ConcurrencyConfigTest.java b/tests/src/test/java/org/neo4j/graphalgo/core/utils/ConcurrencyConfigTest.java

Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,7 @@ public WeightedDegreeCentrality(`
`51`	`51`	`Direction direction`
`52`	`52`	`) {`
`53`	`53`	`if (concurrency <= 0) {`
`54`		`- concurrency = Pools.DEFAULT_QUEUE_SIZE;`
	`54`	`+ concurrency = Pools.DEFAULT_CONCURRENCY;`
`55`	`55`	`}`
`56`	`56`
`57`	`57`	`this.graph = graph;`