add retries to pagerank parallel tasks

Author: mneedham

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

@@ -108,4 +108,11 @@ void normalizeDeltas() {
 //        System.out.println(name + "\nnorm: " + l2Norm + "\nbefore: " + Arrays.toString(before) + "\nafter: " + Arrays.toString(deltas));
     }
 
+    @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;
 
 
@@ -356,17 +357,17 @@ private void run(int iterations) {
                 System.out.println("-------");
                 System.out.println("[iteration started] iteration:" + iteration);
                 // calculate scores
-                ParallelUtil.runWithConcurrency(concurrency, steps, pool);
+                ParallelUtil.runWithConcurrency(concurrency, steps, 3, 1, TimeUnit.SECONDS, pool);
 
                 // sync scores
                 System.out.println("[sync scores] iteration:" + iteration + ", steps:" + steps.size());
                 synchronizeScores();
-                ParallelUtil.runWithConcurrency(concurrency, steps, pool);
+                ParallelUtil.runWithConcurrency(concurrency, steps, 3, 1, TimeUnit.SECONDS, pool);
 
                 // normalize deltas
                 System.out.println("[norm computation] iteration:" + iteration + ", steps:" + steps.size());
                 normalizeDeltas(iteration);
-                ParallelUtil.runWithConcurrency(concurrency, steps, pool);
+                ParallelUtil.runWithConcurrency(concurrency, steps, 3, 1, TimeUnit.SECONDS, pool);
 
                 System.out.println("[iteration finished] iteration:" + iteration);
                 System.out.println("-------");

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

@@ -707,7 +707,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) {
@@ -888,6 +890,7 @@ protected void done() {
             if (executor instanceof ThreadPoolExecutor) {
                 pool = (ThreadPoolExecutor) executor;
                 availableConcurrency = pool.getCorePoolSize();
+//                availableConcurrency = 2;
                 int capacity = Math.max(targetConcurrency, availableConcurrency) + 1;
                 System.out.println("[ParallelUtil#runWithConcurrency] capacity = " + capacity + " [target:" + targetConcurrency + ",available:" + availableConcurrency + "]");
                 completionQueue = new ArrayBlockingQueue<>(capacity);
@@ -919,7 +922,6 @@ boolean submit(Runnable task) {
                 executor.execute(future);
                 return true;
             }
-            System.out.println("[ParallelUtil#runWithConcurrency] unable to submit task " + task);
             return false;
         }
 
@@ -942,6 +944,7 @@ private boolean canSubmit() {
 
             if(!canSubmit) {
                 System.out.println("[ParallelUtil#runWithConcurrency] unable to submit task and pool:" + pool + ", activeCount:" + activeCount + ", availableConcurrency:" + availableConcurrency);
+//                throw new RuntimeException();
             } else {
                 System.out.println("[ParallelUtil#runWithConcurrency] submitted task and pool:" + pool + ", activeCount:" + activeCount + ", availableConcurrency:" + availableConcurrency);
             }

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,7 +219,7 @@ public void testWriteBackUnderDifferentProperty() throws Exception {
     @Test
     public void testParallelWriteBack() throws Exception {
         runQuery(
-                "CALL algo.eigenvector('Character', 'INTERACTS_SEASON1', {batchSize:50, concurrency:2, write:true, graph:'"+graphImpl+"', direction: 'BOTH'}) YIELD writeMillis, write, writeProperty, iterations",
+                "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",