Improve Label Propagation (#432)

Orienting on [Near linear time algorithm to detect community structures in large-scale networks][1], change LPA as follows:

- Randomize the order in which we calculate the nodes
    - Since we are using iterators and want to avoid the allocation of an extra random-order-array, we're randomly skipping and reinserting elements while iterating. This does not truly shuffle the order but just mixes it up a bit.
- Run in asynchronous mode
    - We write and read to the result array directly during iteration, which achieves the asynchronous execution __per-thread__
    - Across threads, we may read older data during a single iteration, but this is ok
    - We implicitly interpret a stale read as "not yet having been in this iteration" which further simulates shuffling of the input. It is as if this node would have been processed afterwards
    - This differes from the stale read issue in #270 in that it's only during a single iteration, not across multiple iterations
- Early terminate if no changes have happened during an iteration
    - A change only occurs if a node does not have the label that most of its neighbours have
    - The paper suggest to confirm this sitaution after every itertation instead of using the 'label-has-changed' semantics, but this would require 2 passes per iteration, effectively doubling the runtime
    - For the sake of performace, we sacrifice situations on which we could converge earlier or at all for completing the algortihm faster

Note that the algorithm does not guarantee a deterministic result, different invocatoins could lead to different clustering, based on the actual graph. This is as designed by the authors of the aforementioned paper.

Fixes #270

[1]: https://arxiv.org/pdf/0709.2938.pdf

* Sometime the cluster converges in just 2 iterations

Author: knutwalker

algo/src/main/java/org/neo4j/graphalgo/LabelPropagationProc.java

@@ -1,6 +1,5 @@
 package org.neo4j.graphalgo;
 
-import com.carrotsearch.hppc.IntDoubleMap;
 import org.neo4j.graphalgo.core.GraphLoader;
 import org.neo4j.graphalgo.core.ProcedureConfiguration;
 import org.neo4j.graphalgo.core.heavyweight.HeavyGraph;
@@ -10,7 +9,7 @@
 import org.neo4j.graphalgo.core.utils.ProgressTimer;
 import org.neo4j.graphalgo.core.utils.TerminationFlag;
 import org.neo4j.graphalgo.core.write.Exporter;
-import org.neo4j.graphalgo.core.write.OptionalIntDoubleMapTranslator;
+import org.neo4j.graphalgo.core.write.IntArrayTranslator;
 import org.neo4j.graphalgo.impl.LabelPropagation;
 import org.neo4j.graphalgo.results.LabelPropagationStats;
 import org.neo4j.graphdb.Direction;
@@ -52,7 +51,7 @@ public final class LabelPropagationProc {
     @Description("CALL algo.labelPropagation(" +
             "label:String, relationship:String, direction:String, " +
             "{iterations:1, weightProperty:'weight', partitionProperty:'partition', write:true, concurrency:4}) " +
-            "YIELD nodes, iterations, loadMillis, computeMillis, writeMillis, write, weightProperty, partitionProperty - " +
+            "YIELD nodes, iterations, didConverge, loadMillis, computeMillis, writeMillis, write, weightProperty, partitionProperty - " +
             "simple label propagation kernel")
     public Stream<LabelPropagationStats> labelPropagation(
             @Name(value = "label", defaultValue = "") String label,
@@ -86,10 +85,7 @@ public Stream<LabelPropagationStats> labelPropagation(
                 concurrency,
                 stats);
 
-        IntDoubleMap labels = compute(direction, iterations, batchSize, concurrency, graph, stats);
-
-        stats.nodes(labels.size());
-
+        int[] labels = compute(direction, iterations, batchSize, concurrency, graph, stats);
         if (configuration.isWriteFlag(DEFAULT_WRITE) && partitionProperty != null) {
             write(concurrency, partitionProperty, graph, labels, stats);
         }
@@ -122,7 +118,7 @@ private HeavyGraph load(
         }
     }
 
-    private IntDoubleMap compute(
+    private int[] compute(
             Direction direction,
             int iterations,
             int batchSize,
@@ -133,10 +129,18 @@ private IntDoubleMap compute(
             ExecutorService pool = batchSize > 0 ? Pools.DEFAULT : null;
             batchSize = Math.max(1, batchSize);
             final LabelPropagation labelPropagation = new LabelPropagation(graph, batchSize, concurrency, pool);
-            final IntDoubleMap result = labelPropagation
-                    .withProgressLogger(ProgressLogger.wrap(log, "LabelPropagation"))
+            labelPropagation
+                    .withProgressLogger(ProgressLogger.wrap(
+                            log,
+                            "LabelPropagation"))
                     .withTerminationFlag(TerminationFlag.wrap(transaction))
                     .compute(direction, iterations);
+            final int[] result = labelPropagation.labels();
+
+            stats.iterations(labelPropagation.ranIterations());
+            stats.didConverge(labelPropagation.didConverge());
+            stats.nodes(result.length);
+
             labelPropagation.release();
             graph.release();
             return result;
@@ -147,7 +151,7 @@ private void write(
             int concurrency,
             String partitionKey,
             HeavyGraph graph,
-            IntDoubleMap labels,
+            int[] labels,
             LabelPropagationStats.Builder stats) {
         stats.write(true);
         try (ProgressTimer timer = stats.timeWrite()) {
@@ -158,7 +162,7 @@ private void write(
                     .write(
                             partitionKey,
                             labels,
-                            OptionalIntDoubleMapTranslator.INSTANCE
+                            IntArrayTranslator.INSTANCE
                 );
         }
     }