Revert "bump the batzh size - I think the GraphView ComputeSteps are just taking too long"

This reverts commit 9824ffd.

Author: mneedham

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

@@ -93,13 +93,19 @@ void synchronizeScores(int[] allScores) {
             deltas[i] = delta;
             allScores[i] = 0;
         }
+
+        String name = String.format("[%s,%d,%d, %d, synchronizeScores]", this.getClass().getSimpleName(), this.startNode, this.partitionSize, this.iteration);
+        System.out.println(name + "\nbefore: " + Arrays.toString(before) + "\nafter: " + Arrays.toString(deltas));
     }
 
     @Override
     void normalizeDeltas() {
+//        double[] before = deltas.clone();
         for (int i = 0; i < deltas.length; i++) {
             deltas[i] = deltas[i] / l2Norm;
         }
+//        String name = String.format("[%s,%d,%d, %d]", this.getClass().getSimpleName(), this.startNode, this.partitionSize, this.iteration);
+//        System.out.println(name + "\nnorm: " + l2Norm + "\nbefore: " + Arrays.toString(before) + "\nafter: " + Arrays.toString(deltas));
     }
 
 }

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

@@ -350,18 +350,26 @@ CentralityResult getPageRank() {
 
         private void run(int iterations) {
             // initialize data structures
+            System.out.println("[pre iterations] init data structures");
             ParallelUtil.runWithConcurrency(concurrency, steps, pool);
             for (int iteration = 0; iteration < iterations && running(); iteration++) {
+                System.out.println("-------");
+                System.out.println("[iteration started] iteration:" + iteration);
                 // calculate scores
                 ParallelUtil.runWithConcurrency(concurrency, steps, pool);
 
                 // sync scores
+                System.out.println("[sync scores] iteration:" + iteration + ", steps:" + steps.size());
                 synchronizeScores();
                 ParallelUtil.runWithConcurrency(concurrency, steps, pool);
 
                 // normalize deltas
+                System.out.println("[norm computation] iteration:" + iteration + ", steps:" + steps.size());
                 normalizeDeltas(iteration);
                 ParallelUtil.runWithConcurrency(concurrency, steps, pool);
+
+                System.out.println("[iteration finished] iteration:" + iteration);
+                System.out.println("-------");
             }
         }
 
@@ -377,11 +385,14 @@ private double computeNorm(int iteration) {
             double l2Norm = 0.0;
             for (ComputeStep step : steps) {
                 double[] deltas = step.deltas();
+                System.out.println("[norm computation] iteration:" + iteration + ", deltas:" + Arrays.toString(deltas));
                 l2Norm += Arrays.stream(deltas).map(score -> score * score).sum();
             }
 
             l2Norm = Math.sqrt(l2Norm);
 
+            System.out.println("[norm computation] iteration:" + iteration  + ", l2Norm: " + l2Norm );
+
             l2Norm = l2Norm <= 0 ? 1 : l2Norm;
             return l2Norm;
         }

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

@@ -671,6 +671,7 @@ private static void runWithConcurrency(
             int maxWaitRetries,
             TerminationFlag terminationFlag,
             ExecutorService executor) {
+        System.out.println("[ParallelUtil#runWithConcurrency] running with " + tasks.size() + " tasks" + ", and concurrency of " + concurrency);
         if (!canRunInParallel(executor)
                 || tasks.size() == 1
                 || concurrency <= 1) {
@@ -691,6 +692,7 @@ private static void runWithConcurrency(
         // generally assumes that tasks.size is notably larger than concurrency
         try {
             //noinspection StatementWithEmptyBody - add first concurrency tasks
+            System.out.println("[ParallelUtil#runWithConcurrency] submit tasks");
             while (concurrency-- > 0
                     && terminationFlag.running()
                     && completionService.trySubmit(ts));
@@ -700,6 +702,7 @@ private static void runWithConcurrency(
             }
 
 
+            System.out.println("[ParallelUtil#runWithConcurrency] submit more tasks");
             // submit all remaining tasks
             int tries = 0;
             while (ts.hasNext()) {
@@ -716,15 +719,18 @@ private static void runWithConcurrency(
                 }
                 if (!completionService.trySubmit(ts) && !completionService.hasTasks()) {
                     if (++tries >= maxWaitRetries) {
+                        System.out.println("[ParallelUtil#runWithConcurrency] exceeded max wait retriesS");
                         break;
                     }
                     LockSupport.parkNanos(waitNanos);
                 }
             }
 
+            System.out.println("[ParallelUtil#runWithConcurrency] wait for tasks");
             // wait for all tasks to finish
             while (completionService.hasTasks() && terminationFlag.running()) {
                 try {
+                    System.out.println("[ParallelUtil#runWithConcurrency] waiting for next task");
                     completionService.awaitNext();
                 } catch (ExecutionException e) {
                     error = Exceptions.chain(error, e.getCause());
@@ -736,6 +742,7 @@ private static void runWithConcurrency(
         } finally {
             finishRunWithConcurrency(completionService, error);
         }
+        System.out.println("[ParallelUtil#runWithConcurrency] finished running with " + tasks.size() + " tasks");
     }
 
     private static void finishRunWithConcurrency(
@@ -912,6 +919,7 @@ boolean submit(Runnable task) {
                 executor.execute(future);
                 return true;
             }
+            System.out.println("[ParallelUtil#runWithConcurrency] unable to submit task " + task);
             return false;
         }
 
@@ -930,7 +938,13 @@ void cancelAll() {
 
         private boolean canSubmit() {
             int activeCount = pool.getActiveCount();
-            return pool == null || activeCount < availableConcurrency;
+            boolean canSubmit = pool == null || activeCount < availableConcurrency;
+
+            if(!canSubmit) {
+                System.out.println("[ParallelUtil#runWithConcurrency] unable to submit task and pool:" + pool + ", activeCount:" + activeCount + ", availableConcurrency:" + availableConcurrency);
+            }
+
+            return canSubmit;
         }
 
         private void stopFutures(Collection<Future<Void>> futures) {

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: 100, 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",
@@ -232,7 +232,7 @@ public void testParallelWriteBack() throws Exception {
     public void testParallelExecution() throws Exception {
         final Map<Long, Double> actual = new HashMap<>();
         runQuery(
-                "CALL algo.eigenvector.stream('Character', 'INTERACTS_SEASON1', {batchSize:100, concurrency:2, graph:'"+graphImpl+"', direction: 'BOTH'}) " +
+                "CALL algo.eigenvector.stream('Character', 'INTERACTS_SEASON1', {batchSize:2, concurrency:2, graph:'"+graphImpl+"', direction: 'BOTH'}) " +
                         "YIELD nodeId, score " +
                         "RETURN nodeId, score " +
                         "ORDER BY score DESC " +