more printin

Author: mneedham

core/src/main/java/org/neo4j/graphalgo/core/neo4jview/GraphView.java

@@ -44,6 +44,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Random;
 import java.util.function.Consumer;
 import java.util.function.IntPredicate;
 import java.util.function.ToDoubleFunction;

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

@@ -701,7 +701,6 @@ private static void runWithConcurrency(
                 return;
             }
 
-
             System.out.println("[ParallelUtil#runWithConcurrency] submit more tasks");
             // submit all remaining tasks
             int tries = 0;
@@ -719,7 +718,7 @@ private static void runWithConcurrency(
                 }
                 if (!completionService.trySubmit(ts) && !completionService.hasTasks()) {
                     if (++tries >= maxWaitRetries) {
-                        System.out.println("[ParallelUtil#runWithConcurrency] exceeded max wait retriesS");
+                        System.out.println("[ParallelUtil#runWithConcurrency] exceeded max wait retries");
                         break;
                     }
                     LockSupport.parkNanos(waitNanos);
@@ -890,6 +889,7 @@ protected void done() {
                 pool = (ThreadPoolExecutor) executor;
                 availableConcurrency = pool.getCorePoolSize();
                 int capacity = Math.max(targetConcurrency, availableConcurrency) + 1;
+                System.out.println("[ParallelUtil#runWithConcurrency] capacity = " + capacity + " [target:" + targetConcurrency + ",available:" + availableConcurrency + "]");
                 completionQueue = new ArrayBlockingQueue<>(capacity);
             } else {
                 pool = null;
@@ -937,11 +937,13 @@ void cancelAll() {
         }
 
         private boolean canSubmit() {
-            int activeCount = pool.getActiveCount();
-            boolean canSubmit = pool == null || activeCount < availableConcurrency;
+            int activeCount = 0;
+            boolean canSubmit = pool == null || (activeCount = pool.getActiveCount()) < availableConcurrency;
 
             if(!canSubmit) {
                 System.out.println("[ParallelUtil#runWithConcurrency] unable to submit task and pool:" + pool + ", activeCount:" + activeCount + ", availableConcurrency:" + availableConcurrency);
+            } else {
+                System.out.println("[ParallelUtil#runWithConcurrency] submitted task and pool:" + pool + ", activeCount:" + activeCount + ", availableConcurrency:" + availableConcurrency);
             }
 
             return canSubmit;
@@ -996,6 +998,7 @@ public T next() {
 
         void pushBack(T element) {
             if (pushedElement != null) {
+                System.out.println("[ParallelUtil#runWithConcurrency] unable to reschedule task");
                 throw new IllegalArgumentException("Cannot push back twice");
             }
             pushedElement = element;

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:3, concurrency:2, write:true, graph:'"+graphImpl+"', direction: 'BOTH'}) YIELD writeMillis, write, writeProperty, iterations",
+                "CALL algo.eigenvector('Character', 'INTERACTS_SEASON1', {batchSize:50, 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:2, concurrency:2, graph:'"+graphImpl+"', direction: 'BOTH'}) " +
+                "CALL algo.eigenvector.stream('Character', 'INTERACTS_SEASON1', {batchSize:100, concurrency:2, graph:'"+graphImpl+"', direction: 'BOTH'}) " +
                         "YIELD nodeId, score " +
                         "RETURN nodeId, score " +
                         "ORDER BY score DESC " +