Merge pull request #9803 from IoannisPanagiotas/sllp-becomes-explicit

Sllp becomes explicit

Author: IoannisPanagiotas

algo/src/main/java/org/neo4j/gds/sllpa/SpeakerListenerLPA.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.gds.sllpa;
+
+import org.neo4j.gds.Algorithm;
+import org.neo4j.gds.api.Graph;
+import org.neo4j.gds.beta.pregel.ImmutablePregelResult;
+import org.neo4j.gds.beta.pregel.Pregel;
+import org.neo4j.gds.beta.pregel.PregelResult;
+import org.neo4j.gds.core.utils.progress.tasks.ProgressTracker;
+
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+
+
+public class SpeakerListenerLPA extends Algorithm<PregelResult> {
+    public static final String LABELS_PROPERTY = "communityIds";
+
+    private final Graph graph;
+    private final SpeakerListenerLPAConfig config;
+    private final ExecutorService executorService;
+    private final Optional<Long> seed;
+
+    public SpeakerListenerLPA(
+        Graph graph,
+        SpeakerListenerLPAConfig config,
+        ExecutorService executorService,
+        ProgressTracker progressTracker,
+        Optional<Long> seed
+    ) {
+        super(progressTracker);
+        this.graph = graph;
+        this.config = config;
+        this.executorService = executorService;
+        this.seed = seed;
+
+    }
+
+    @Override
+    public PregelResult compute() {
+
+        var computation = new SpeakerListenerLPAComputation(seed);
+
+        var pregelResult = Pregel.create(
+            graph,
+            config,
+            computation,
+            executorService,
+            progressTracker,
+            terminationFlag
+        ).run();
+
+        this.progressTracker.endSubTask();
+
+        return ImmutablePregelResult.builder()
+            .nodeValues(pregelResult.nodeValues())
+            .didConverge(pregelResult.didConverge())
+            .ranIterations(pregelResult.ranIterations())
+            .build();
+    }
+}

alpha/alpha-proc/src/main/java/org/neo4j/gds/pregel/SpeakerListenerLPA.java renamed to algo/src/main/java/org/neo4j/gds/sllpa/SpeakerListenerLPAComputation.java

@@ -17,64 +17,51 @@
  * You should have received a copy of the GNU General Public License
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
-package org.neo4j.gds.pregel;
+package org.neo4j.gds.sllpa;
 
 import com.carrotsearch.hppc.LongArrayList;
 import com.carrotsearch.hppc.LongIntScatterMap;
 import com.carrotsearch.hppc.cursors.LongIntCursor;
-import org.neo4j.gds.mem.MemoryEstimateDefinition;
-import org.neo4j.gds.annotation.Configuration;
 import org.neo4j.gds.api.nodeproperties.ValueType;
 import org.neo4j.gds.beta.pregel.Messages;
 import org.neo4j.gds.beta.pregel.Pregel;
 import org.neo4j.gds.beta.pregel.PregelComputation;
-import org.neo4j.gds.beta.pregel.PregelProcedureConfig;
 import org.neo4j.gds.beta.pregel.PregelSchema;
-import org.neo4j.gds.beta.pregel.annotation.PregelProcedure;
 import org.neo4j.gds.beta.pregel.context.ComputeContext;
 import org.neo4j.gds.beta.pregel.context.InitContext;
-import org.neo4j.gds.core.CypherMapWrapper;
+import org.neo4j.gds.mem.MemoryEstimateDefinition;
 import org.neo4j.gds.utils.CloseableThreadLocal;
 
 import java.util.Arrays;
 import java.util.Map;
+import java.util.Optional;
 import java.util.SplittableRandom;
 
-@PregelProcedure(
-    name = "gds.sllpa",
-    description = "The Speaker Listener Label Propagation algorithm is a fast algorithm for finding overlapping communities in a graph."
-)
-public class SpeakerListenerLPA implements PregelComputation<SpeakerListenerLPA.SpeakerListenerLPAConfig> {
+import static org.neo4j.gds.sllpa.SpeakerListenerLPA.LABELS_PROPERTY;
 
-    public static final String LABELS_PROPERTY = "communityIds";
+class SpeakerListenerLPAComputation implements PregelComputation<SpeakerListenerLPAConfig> {
 
     private final CloseableThreadLocal<SplittableRandom> random;
 
-    @SuppressWarnings("unused") // Needed for the @PregelProcedure annotation
-    public SpeakerListenerLPA() {
-        this(System.currentTimeMillis());
-    }
-
-    public SpeakerListenerLPA(long seed) {
-        var splittableRandom = new SplittableRandom(seed);
+    SpeakerListenerLPAComputation(Optional<Long> seed) {
+        var splittableRandom = new SplittableRandom(seed.orElse(System.currentTimeMillis()));
         random = CloseableThreadLocal.withInitial(splittableRandom::split);
     }
 
     @Override
-    public PregelSchema schema(SpeakerListenerLPAConfig config) {
-        return new PregelSchema.Builder()
-            .add(LABELS_PROPERTY, ValueType.LONG_ARRAY)
-            .build();
-    }
-
+    public void compute(ComputeContext<SpeakerListenerLPAConfig> context, Messages messages) {
+        var labels = context.longArrayNodeValue(LABELS_PROPERTY);
 
-    @Override
-    public MemoryEstimateDefinition estimateDefinition(boolean isAsynchronous) {
-        return () -> Pregel.memoryEstimation(
-            Map.of(LABELS_PROPERTY, ValueType.LONG_ARRAY),
-            false,
-            false
-        );
+        if (context.isInitialSuperstep()) {
+            labels[0] = context.nodeId();
+            context.sendToNeighbors(context.nodeId());
+        } else if (context.superstep() < context.config().propagationSteps()) {
+            listen(context, messages, labels);
+            speak(context, labels);
+        } else {
+            listen(context, messages, labels);
+            prune(context, labels);
+        }
     }
 
     @Override
@@ -85,25 +72,21 @@ public void init(InitContext<SpeakerListenerLPAConfig> context) {
         context.setNodeValue(LABELS_PROPERTY, initialLabels);
     }
 
+
     @Override
-    public void close() {
-        random.close();
+    public PregelSchema schema(SpeakerListenerLPAConfig config) {
+        return new PregelSchema.Builder()
+            .add(LABELS_PROPERTY, ValueType.LONG_ARRAY)
+            .build();
     }
 
     @Override
-    public void compute(ComputeContext<SpeakerListenerLPAConfig> context, Messages messages) {
-        var labels = context.longArrayNodeValue(LABELS_PROPERTY);
-
-        if (context.isInitialSuperstep()) {
-            labels[0] = context.nodeId();
-            context.sendToNeighbors(context.nodeId());
-        } else if (context.superstep() < context.config().propagationSteps()) {
-            listen(context, messages, labels);
-            speak(context, labels);
-        } else {
-            listen(context, messages, labels);
-            prune(context, labels);
-        }
+    public MemoryEstimateDefinition estimateDefinition(boolean isAsynchronous) {
+        return () -> Pregel.memoryEstimation(
+            Map.of(LABELS_PROPERTY, ValueType.LONG_ARRAY),
+            false,
+            false
+        );
     }
 
     private void listen(
@@ -161,28 +144,8 @@ private void prune(ComputeContext<SpeakerListenerLPAConfig> context, long[] labe
         );
     }
 
-
-    @Configuration
-    public interface SpeakerListenerLPAConfig extends PregelProcedureConfig {
-
-        default double minAssociationStrength() {
-            return 0.2;
-        }
-
-        static SpeakerListenerLPAConfig of(CypherMapWrapper userConfig) {
-            return new SpeakerListenerLPAConfigImpl(userConfig);
-        }
-
-        @Override
-        @Configuration.Ignore
-        default boolean isAsynchronous() {
-            return true;
-        }
-
-
-        @Configuration.Ignore
-        default int propagationSteps() {
-            return maxIterations() - 1;
-        }
+    @Override
+    public void close() {
+        random.close();
     }
 }

algo/src/main/java/org/neo4j/gds/sllpa/SpeakerListenerLPAMemoryEstimateDefinition.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.gds.sllpa;
+
+import org.neo4j.gds.api.nodeproperties.ValueType;
+import org.neo4j.gds.beta.pregel.Pregel;
+import org.neo4j.gds.mem.MemoryEstimateDefinition;
+import org.neo4j.gds.mem.MemoryEstimation;
+import org.neo4j.gds.mem.MemoryEstimations;
+
+import java.util.Map;
+
+import static org.neo4j.gds.sllpa.SpeakerListenerLPA.LABELS_PROPERTY;
+
+
+public class SpeakerListenerLPAMemoryEstimateDefinition  implements MemoryEstimateDefinition {
+
+    @Override
+    public MemoryEstimation memoryEstimation() {
+        return MemoryEstimations.builder()
+            .add(Pregel.memoryEstimation(
+                Map.of(LABELS_PROPERTY, ValueType.LONG_ARRAY),
+                false,
+                false
+            )).build();
+    }
+}

algo/src/main/java/org/neo4j/gds/sllpa/SpeakerListenerLPAProgressTrackerCreator.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.gds.sllpa;
+
+import org.neo4j.gds.core.utils.progress.tasks.Task;
+import org.neo4j.gds.core.utils.progress.tasks.Tasks;
+
+import java.util.List;
+
+public class SpeakerListenerLPAProgressTrackerCreator {
+
+    public static Task progressTask(long nodeCount, int maxIterations,String taskName) {
+        return Tasks.iterativeDynamic(
+            taskName,
+            () -> List.of(
+                Tasks.leaf("Compute iteration", nodeCount),
+                Tasks.leaf("Master compute iteration", nodeCount)
+            ),
+            maxIterations
+        );
+}
+}

algo/src/test/java/org/neo4j/gds/sllpa/SpeakerListenerLPAComputationTest.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.gds.sllpa;
+
+import org.junit.jupiter.api.Test;
+import org.neo4j.gds.utils.CloseableThreadLocal;
+
+import java.lang.invoke.MethodHandles;
+import java.util.Optional;
+import java.util.SplittableRandom;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.fail;
+
+class SpeakerListenerLPAComputationTest {
+
+    @Test
+    void closesThreadLocal() {
+
+        var computation = new SpeakerListenerLPAComputation(Optional.empty());
+        computation.close();
+
+        CloseableThreadLocal<SplittableRandom> threadLocal = null;
+        try {
+            //noinspection unchecked
+            threadLocal = (CloseableThreadLocal<SplittableRandom>) MethodHandles
+                .privateLookupIn(SpeakerListenerLPAComputation.class, MethodHandles.lookup())
+                .findGetter(SpeakerListenerLPAComputation.class, "random", CloseableThreadLocal.class)
+                .invoke(computation);
+        } catch (Throwable e) {
+            fail("couldn't inspect the field", e);
+        }
+        assertThatThrownBy(threadLocal::get).isInstanceOf(NullPointerException.class);
+    }
+
+}