Improve randomness for LPA (#858)

* Enable LPA to control its randomness
* Add Random Iterators with proper random iteration
* Add random iteration benchmark
* Use proper randomness in LPA

Author: knutwalker

algo/src/main/java/org/neo4j/graphalgo/impl/BaseLabelPropagation.java

@@ -18,24 +18,38 @@
  */
 package org.neo4j.graphalgo.impl;
 
+import com.carrotsearch.hppc.HashOrderMixing;
+import com.carrotsearch.hppc.LongDoubleHashMap;
+import com.carrotsearch.hppc.LongDoubleScatterMap;
+import com.carrotsearch.hppc.cursors.LongDoubleCursor;
+import org.neo4j.collection.primitive.PrimitiveIntIterator;
+import org.neo4j.collection.primitive.PrimitiveLongIterator;
 import org.neo4j.graphalgo.api.Graph;
 import org.neo4j.graphalgo.api.WeightMapping;
+import org.neo4j.graphalgo.core.utils.LazyBatchCollection;
 import org.neo4j.graphalgo.core.utils.ParallelUtil;
+import org.neo4j.graphalgo.core.utils.ProgressLogger;
+import org.neo4j.graphalgo.core.utils.RandomLongIterable;
 import org.neo4j.graphalgo.core.utils.paged.AllocationTracker;
+import org.neo4j.graphalgo.core.utils.paged.BitUtil;
 import org.neo4j.graphdb.Direction;
 
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
+import java.util.Random;
 import java.util.concurrent.ExecutorService;
 
+import static com.carrotsearch.hppc.Containers.DEFAULT_EXPECTED_ELEMENTS;
+import static com.carrotsearch.hppc.HashContainers.DEFAULT_LOAD_FACTOR;
+
 abstract class BaseLabelPropagation<
         G extends Graph,
         W extends WeightMapping,
-        L extends LabelPropagationAlgorithm.Labels,
-        Self extends BaseLabelPropagation<G, W, L, Self>
+        Self extends BaseLabelPropagation<G, W, Self>
         > extends LabelPropagationAlgorithm<Self> {
 
-    static final int[] EMPTY_INTS = new int[0];
-    static final long[] EMPTY_LONGS = new long[0];
+    private static final long[] EMPTY_LONGS = new long[0];
 
     private G graph;
     private final long nodeCount;
@@ -46,14 +60,14 @@ abstract class BaseLabelPropagation<
     final int concurrency;
     final ExecutorService executor;
 
-    private L labels;
+    private Labels labels;
     private long ranIterations;
     private boolean didConverge;
 
     BaseLabelPropagation(
             G graph,
-             W nodeProperties,
-             W nodeWeights,
+            W nodeProperties,
+            W nodeWeights,
             int batchSize,
             int concurrency,
             ExecutorService executor,
@@ -68,22 +82,24 @@ abstract class BaseLabelPropagation<
         this.nodeWeights = nodeWeights;
     }
 
-    abstract L initialLabels(long nodeCount, AllocationTracker tracker);
-
+    abstract Labels initialLabels(long nodeCount, AllocationTracker tracker);
 
-    abstract List<BaseStep> baseSteps(
+    abstract Initialization initStep(
             final G graph,
-            final L labels,
+            final Labels labels,
             final W nodeProperties,
             final W nodeWeights,
             final Direction direction,
-            final boolean randomizeOrder);
+            final ProgressLogger progressLogger,
+            final RandomProvider randomProvider,
+            final RandomLongIterable nodes
+    );
 
     @Override
-    final Self compute(
+    Self compute(
             Direction direction,
             long maxIterations,
-            boolean randomizeOrder) {
+            RandomProvider random) {
         if (maxIterations <= 0L) {
             throw new IllegalArgumentException("Must iterate at least 1 time");
         }
@@ -95,10 +111,12 @@ final Self compute(
         ranIterations = 0L;
         didConverge = false;
 
-        List<BaseStep> baseSteps = baseSteps(graph, labels, nodeProperties, nodeWeights, direction, randomizeOrder);
+        List<BaseStep> baseSteps = baseSteps(direction, random);
 
-        for (long i = 0L; i < maxIterations; i++) {
-            ParallelUtil.runWithConcurrency(concurrency, baseSteps, executor);
+        long currentIteration = 0L;
+        while (running() && currentIteration < maxIterations) {
+            ParallelUtil.runWithConcurrency(concurrency, baseSteps, terminationFlag, executor);
+            ++currentIteration;
         }
 
         long maxIteration = 0L;
@@ -121,10 +139,6 @@ final Self compute(
         return me();
     }
 
-    final BaseStep asStep(Initialization initialization) {
-        return new BaseStep(initialization);
-    }
-
     @Override
     public final long ranIterations() {
         return ranIterations;
@@ -146,6 +160,47 @@ public Self release() {
         return me();
     }
 
+    private List<BaseStep> baseSteps(Direction direction, RandomProvider random) {
+
+        long nodeCount = graph.nodeCount();
+        long batchSize = adjustBatchSize(nodeCount, (long) this.batchSize);
+
+        Collection<RandomLongIterable> nodeBatches = LazyBatchCollection.of(
+                nodeCount,
+                batchSize,
+                (start, length) -> new RandomLongIterable(start, start + length, random.randomForNewIteration()));
+
+        int threads = nodeBatches.size();
+        List<BaseStep> tasks = new ArrayList<>(threads);
+        for (RandomLongIterable iter : nodeBatches) {
+            Initialization initStep = initStep(
+                    graph,
+                    labels,
+                    nodeProperties,
+                    nodeWeights,
+                    direction,
+                    getProgressLogger(),
+                    random,
+                    iter
+            );
+            BaseStep task = new BaseStep(initStep);
+            tasks.add(task);
+        }
+        ParallelUtil.runWithConcurrency(concurrency, tasks, terminationFlag, executor);
+        return tasks;
+    }
+
+    private long adjustBatchSize(long nodeCount, long batchSize) {
+        if (batchSize <= 0L) {
+            batchSize = 1L;
+        }
+        batchSize = BitUtil.nextHighestPowerOfTwo(batchSize);
+        while (((nodeCount + batchSize + 1L) / batchSize) > (long) Integer.MAX_VALUE) {
+            batchSize = batchSize << 1;
+        }
+        return batchSize;
+    }
+
     static abstract class Initialization implements Step {
         abstract void setExistingLabels();
 
@@ -164,12 +219,40 @@ public final Step next() {
 
     static abstract class Computation implements Step {
 
+        final RandomProvider randomProvider;
+        private final Labels existingLabels;
+        private final ProgressLogger progressLogger;
+        private final double maxNode;
+        private final LongDoubleHashMap votes;
+
         private boolean didChange = true;
-        private long iteration = 0L;
+        long iteration = 0L;
+
+        Computation(
+                final Labels existingLabels,
+                final ProgressLogger progressLogger,
+                final long maxNode,
+                final RandomProvider randomProvider) {
+            this.randomProvider = randomProvider;
+            this.existingLabels = existingLabels;
+            this.progressLogger = progressLogger;
+            this.maxNode = (double) maxNode;
+            if (randomProvider.isRandom()) {
+                Random random = randomProvider.randomForNewIteration();
+                this.votes = new LongDoubleHashMap(
+                        DEFAULT_EXPECTED_ELEMENTS,
+                        (double) DEFAULT_LOAD_FACTOR,
+                        HashOrderMixing.constant(random.nextLong()));
+            } else {
+                this.votes = new LongDoubleScatterMap();
+            }
+        }
 
         abstract boolean computeAll();
 
-        abstract void release();
+        abstract void forEach(long nodeId);
+
+        abstract double weightOf(long nodeId, long candidate);
 
         @Override
         public final void run() {
@@ -182,10 +265,68 @@ public final void run() {
             }
         }
 
+        final boolean iterateAll(PrimitiveIntIterator nodeIds) {
+            boolean didChange = false;
+            while (nodeIds.hasNext()) {
+                long nodeId = (long) nodeIds.next();
+                didChange = compute(nodeId, didChange);
+                progressLogger.logProgress((double) nodeId, maxNode);
+            }
+            return didChange;
+        }
+
+        final boolean iterateAll(PrimitiveLongIterator nodeIds) {
+            boolean didChange = false;
+            while (nodeIds.hasNext()) {
+                long nodeId = nodeIds.next();
+                didChange = compute(nodeId, didChange);
+                progressLogger.logProgress((double) nodeId, maxNode);
+            }
+            return didChange;
+        }
+
+        final boolean compute(long nodeId, boolean didChange) {
+            votes.clear();
+            long partition = existingLabels.labelFor(nodeId);
+            long previous = partition;
+            forEach(nodeId);
+            double weight = Double.NEGATIVE_INFINITY;
+            for (LongDoubleCursor vote : votes) {
+                if (weight < vote.value) {
+                    weight = vote.value;
+                    partition = vote.key;
+                }
+            }
+            if (partition != previous) {
+                existingLabels.setLabelFor(nodeId, partition);
+                return true;
+            }
+            return didChange;
+        }
+
+        final void castVote(long nodeId, long candidate) {
+            double weight = weightOf(nodeId, candidate);
+            long partition = existingLabels.labelFor(candidate);
+            votes.addTo(partition, weight);
+        }
+
         @Override
         public final Step next() {
             return this;
         }
+
+        final void release() {
+            // the HPPC release() method allocates new arrays
+            // the clear() method overwrite the existing keys with the default value
+            // we want to throw away all data to allow for GC collection instead.
+
+            if (votes.keys != null) {
+                votes.keys = EMPTY_LONGS;
+                votes.clear();
+                votes.keys = null;
+                votes.values = null;
+            }
+        }
     }
 
     interface Step extends Runnable {