add retries to pagerank parallel tasks

Author: mneedham

algo/src/main/java/org/neo4j/graphalgo/impl/pagerank/EigenvectorCentralityComputeStep.java

@@ -100,4 +100,11 @@ void normalizeDeltas() {
         }
     }
 
+    @Override
+    public String toString() {
+        return "EigenvectorCentralityComputeStep{" +
+                "partitionSize=" + partitionSize +
+                ", startNode=" + startNode +
+                '}';
+    }
 }

algo/src/main/java/org/neo4j/graphalgo/impl/pagerank/PageRank.java

@@ -31,6 +31,7 @@
 
 import java.util.*;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.LongStream;
 
 
@@ -353,15 +354,15 @@ private void run(int iterations) {
             ParallelUtil.runWithConcurrency(concurrency, steps, pool);
             for (int i = 0; i < iterations && running(); i++) {
                 // calculate scores
-                ParallelUtil.runWithConcurrency(concurrency, steps, pool);
+                ParallelUtil.runWithConcurrency(concurrency, steps, 3, 1, TimeUnit.SECONDS, pool);
 
                 // sync scores
                 synchronizeScores();
-                ParallelUtil.runWithConcurrency(concurrency, steps, pool);
+                ParallelUtil.runWithConcurrency(concurrency, steps, 3, 1, TimeUnit.SECONDS, pool);
 
                 // normalize deltas
                 normalizeDeltas();
-                ParallelUtil.runWithConcurrency(concurrency, steps, pool);
+                ParallelUtil.runWithConcurrency(concurrency, steps, 3, 1, TimeUnit.SECONDS, pool);
             }
         }
 

core/src/main/java/org/neo4j/graphalgo/core/utils/ParallelUtil.java

@@ -705,7 +705,9 @@ private static void runWithConcurrency(
             while (ts.hasNext()) {
                 if (completionService.hasTasks()) {
                     try {
+                        System.out.println("[ParallelUtil#runWithConcurrency] waiting for task to finish... " + executor);
                         completionService.awaitNext();
+                        System.out.println("[ParallelUtil#runWithConcurrency] task finished... " + executor);
                     } catch (ExecutionException e) {
                         error = Exceptions.chain(error, e.getCause());
                     } catch (CancellationException ignore) {
@@ -882,6 +884,7 @@ protected void done() {
             if (executor instanceof ThreadPoolExecutor) {
                 pool = (ThreadPoolExecutor) executor;
                 availableConcurrency = pool.getCorePoolSize();
+//                availableConcurrency = 2;
                 int capacity = Math.max(targetConcurrency, availableConcurrency) + 1;
                 completionQueue = new ArrayBlockingQueue<>(capacity);
             } else {

core/src/main/java/org/neo4j/graphalgo/core/utils/Pools.java

@@ -42,6 +42,7 @@ public class Pools {
             DEFAULT_CONCURRENCY = definedProcessors;
         } else {
             DEFAULT_CONCURRENCY = Runtime.getRuntime().availableProcessors();
+//            DEFAULT_CONCURRENCY = 2;
         }
     }
 

tests/src/test/java/org/neo4j/graphalgo/algo/EigenvectorCentralityProcIntegrationTest.java

@@ -219,10 +219,12 @@ public void testWriteBackUnderDifferentProperty() throws Exception {
     @Test
     public void testParallelWriteBack() throws Exception {
         runQuery(
-                "CALL algo.eigenvector('Character', 'INTERACTS_SEASON1', {batchSize:3, write:true, graph:'"+graphImpl+"', direction: 'BOTH'}) YIELD writeMillis, write, writeProperty",
-                row -> assertTrue(
-                        "write time not set",
-                        row.getNumber("writeMillis").intValue() >= 0));
+                "CALL algo.eigenvector('Character', 'INTERACTS_SEASON1', {batchSize:3, concurrency:2, write:true, graph:'"+graphImpl+"', direction: 'BOTH'}) YIELD writeMillis, write, writeProperty, iterations",
+                row -> {
+                    assertTrue(
+                            "write time not set",
+                            row.getNumber("writeMillis").intValue() >= 0);
+                });
 
         assertResult("eigenvector", expected);
     }