Parallel pagerank (#178)

* Add parallel page rank

* Run only 1 warmup/measurement for LDBC benchmark

* Allow LightGraph to be run in parallel

* Only park for 100 nanos instead of sleep for 100 millis when the queue is full

* Avoid allocation of task list, of tasks cannot be run in parallel

* Make sure to sync all scores before running a new iteration

* Add test for parallel execution only

* Reorder LBDC params

Author: knutwalker

algo/src/main/java/org/neo4j/graphalgo/PageRankProc.java

@@ -1,11 +1,10 @@
 package org.neo4j.graphalgo;
 
-import algo.Pools;
 import org.neo4j.graphalgo.api.Graph;
 import org.neo4j.graphalgo.api.GraphFactory;
 import org.neo4j.graphalgo.core.GraphLoader;
 import org.neo4j.graphalgo.core.ProcedureConfiguration;
-import org.neo4j.graphalgo.core.heavyweight.HeavyGraphFactory;
+import org.neo4j.graphalgo.core.utils.Pools;
 import org.neo4j.graphalgo.core.utils.ProgressTimer;
 import org.neo4j.graphalgo.impl.PageRank;
 import org.neo4j.graphalgo.impl.PageRankExporter;
@@ -104,21 +103,25 @@ private double[] evaluate(
 
         double dampingFactor = configuration.get(CONFIG_DAMPING, DEFAULT_DAMPING);
         int iterations = configuration.getIterations(DEFAULT_ITERATIONS);
+        final int batchSize = configuration.getBatchSize();
         log.debug("Computing page rank with damping of " + dampingFactor + " and " + iterations + " iterations.");
-        PageRank pageRankAlgo = new PageRank(
+
+        PageRank algo = new PageRank(
+                Pools.DEFAULT,
+                batchSize,
                 graph,
                 graph,
                 graph,
                 graph,
                 dampingFactor);
 
-        statsBuilder.timeEval(() -> pageRankAlgo.compute(iterations));
+        statsBuilder.timeEval(() -> algo.compute(iterations));
 
         statsBuilder
                 .withIterations(iterations)
                 .withDampingFactor(dampingFactor);
 
-        return pageRankAlgo.getPageRank();
+        return algo.getPageRank();
     }
 
     private void write(
@@ -130,15 +133,14 @@ private void write(
             log.debug("Writing results");
             String propertyName = configuration.getWriteProperty(DEFAULT_SCORE_PROPERTY);
             int batchSize = configuration.getBatchSize();
-            statsBuilder.timeWrite(() -> {
-                new PageRankExporter(
-                        batchSize,
-                        api,
-                        graph,
-                        graph,
-                        propertyName,
-                        Pools.DEFAULT).write(scores);
-            });
+            statsBuilder.timeWrite(() -> new PageRankExporter(
+                    batchSize,
+                    api,
+                    graph,
+                    graph,
+                    propertyName,
+                    Pools.DEFAULT)
+                    .write(scores));
             statsBuilder
                     .withWrite(true)
                     .withProperty(propertyName);