Merge pull request #9338 from FlorentinD/rel-parallel-deadlock

Revert writing relationships in parallel

algo/src/main/java/org/neo4j/gds/algorithms/similarity/Neo4jDatabaseRelationshipWriter.java

@@ -26,7 +26,6 @@
 import org.neo4j.gds.api.RelationshipWithPropertyConsumer;
 import org.neo4j.gds.api.ResultStore;
 import org.neo4j.gds.config.ArrowConnectionInfo;
-import org.neo4j.gds.core.concurrency.Concurrency;
 import org.neo4j.gds.core.utils.ProgressTimer;
 import org.neo4j.gds.core.utils.progress.JobId;
 import org.neo4j.gds.core.utils.progress.TaskRegistryFactory;
@@ -51,7 +50,6 @@ public static WriteRelationshipResult writeRelationship(
         Log log,
         String taskName,
         TerminationFlag algorithmTerminationFlag,
-        Concurrency concurrency,
         Optional<ArrowConnectionInfo> arrowConnectionInfo,
         Optional<ResultStore> resultStore,
         RelationshipWithPropertyConsumer relationshipConsumer,
@@ -71,7 +69,6 @@ public static WriteRelationshipResult writeRelationship(
                 .withGraph(graph)
                 .withTerminationFlag(algorithmTerminationFlag)
                 .withProgressTracker(progressTracker)
-                .withConcurrency(concurrency)
                 .withArrowConnectionInfo(
                     arrowConnectionInfo,
                     graphStore.databaseInfo().remoteDatabaseId().map(DatabaseId::databaseName)

algo/src/main/java/org/neo4j/gds/algorithms/similarity/WriteRelationshipService.java

@@ -27,7 +27,6 @@
 import org.neo4j.gds.applications.algorithms.machinery.RequestScopedDependencies;
 import org.neo4j.gds.applications.algorithms.machinery.WriteContext;
 import org.neo4j.gds.config.ArrowConnectionInfo;
-import org.neo4j.gds.core.concurrency.Concurrency;
 import org.neo4j.gds.core.utils.progress.JobId;
 import org.neo4j.gds.logging.Log;
 
@@ -51,7 +50,6 @@ public WriteRelationshipResult write(
         GraphStore graphStore,
         IdMap rootIdMap,
         String taskName,
-        Concurrency concurrency,
         Optional<ArrowConnectionInfo> arrowConnectionInfo,
         Optional<ResultStore> resultStore,
         RelationshipWithPropertyConsumer relationshipWithPropertyConsumer,
@@ -68,7 +66,6 @@ public WriteRelationshipResult write(
             log,
             taskName,
             requestScopedDependencies.getTerminationFlag(),
-            concurrency,
             arrowConnectionInfo,
             resultStore,
             relationshipWithPropertyConsumer,

applications/algorithms/path-finding/src/main/java/org/neo4j/gds/applications/algorithms/pathfinding/ShortestPathWriteStep.java

@@ -121,7 +121,6 @@ public RelationshipsWritten execute(
 
             // configure the exporter
             var relationshipStreamExporter = writeContext.getRelationshipStreamExporterBuilder()
-                .withConcurrency(configuration.writeConcurrency())
                 .withArrowConnectionInfo(
                     configuration.arrowConnectionInfo(),
                     graphStore.databaseInfo().remoteDatabaseId().map(DatabaseId::databaseName)

applications/algorithms/similarity/src/main/java/org/neo4j/gds/applications/algorithms/similarity/SimilarityWrite.java

@@ -106,7 +106,6 @@ Pair<RelationshipsWritten, Map<String, Object>> execute(
             graphStore,
             rootIdMap,
             label.value,
-            writeConfiguration.writeConcurrency(),
             writeConfiguration.arrowConnectionInfo(),
             resultStore,
             similarityDistributionBuilder.similarityConsumer(),

applications/graph-store-catalog/src/main/java/org/neo4j/gds/applications/graphstorecatalog/WriteNodeLabelApplication.java

@@ -74,7 +74,7 @@ WriteLabelResult compute(
                 )
                 .withResultStore(configuration.resolveResultStore(resultStore))
                 .withJobId(configuration.jobId())
-                .parallel(DefaultPool.INSTANCE, configuration.concurrency())
+                .parallel(DefaultPool.INSTANCE, configuration.writeConcurrency())
                 .build();
 
             try {

applications/graph-store-catalog/src/main/java/org/neo4j/gds/applications/graphstorecatalog/WriteRelationshipPropertiesApplication.java

@@ -57,7 +57,6 @@ WriteRelationshipPropertiesResult compute(
             .withRelationPropertyTranslator(Values::doubleValue)
             .withTerminationFlag(terminationFlag)
             .withProgressTracker(ProgressTracker.NULL_TRACKER)
-            .withConcurrency(configuration.concurrency())
             .withArrowConnectionInfo(
                 configuration.arrowConnectionInfo(),
                 graphStore.databaseInfo().remoteDatabaseId().map(DatabaseId::databaseName)

applications/graph-store-catalog/src/main/java/org/neo4j/gds/applications/graphstorecatalog/WriteRelationshipsApplication.java

@@ -26,7 +26,6 @@
 import org.neo4j.gds.api.ResultStore;
 import org.neo4j.gds.api.nodeproperties.ValueType;
 import org.neo4j.gds.config.ArrowConnectionInfo;
-import org.neo4j.gds.core.concurrency.Concurrency;
 import org.neo4j.gds.core.utils.ProgressTimer;
 import org.neo4j.gds.core.utils.progress.JobId;
 import org.neo4j.gds.core.utils.progress.TaskRegistryFactory;
@@ -83,7 +82,6 @@ WriteRelationshipResult compute(
                     relationshipExporterBuilder,
                     terminationFlag,
                     progressTracker,
-                    configuration.concurrency(),
                     configuration.arrowConnectionInfo(),
                     configuration.resolveResultStore(resultStore),
                     graphStore,
@@ -109,7 +107,6 @@ private long writeRelationshipType(
         RelationshipExporterBuilder relationshipExporterBuilder,
         TerminationFlag terminationFlag,
         ProgressTracker progressTracker,
-        Concurrency concurrency,
         Optional<ArrowConnectionInfo> arrowConnectionInfo,
         Optional<ResultStore> resultStore,
         GraphStore graphStore,
@@ -123,7 +120,6 @@ private long writeRelationshipType(
             .withIdMappingOperator(graph::toOriginalNodeId)
             .withGraph(graph)
             .withTerminationFlag(terminationFlag)
-            .withConcurrency(concurrency)
             .withArrowConnectionInfo(arrowConnectionInfo, graphStore.databaseInfo().remoteDatabaseId().map(DatabaseId::databaseName))
             .withResultStore(resultStore)
             .withJobId(jobId)

core-write/src/integrationTest/java/org/neo4j/gds/core/write/NativeRelationshipExporterTest.java

@@ -22,8 +22,6 @@
 import org.apache.commons.lang3.mutable.MutableInt;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
 import org.neo4j.gds.BaseTest;
 import org.neo4j.gds.Orientation;
 import org.neo4j.gds.PropertyMapping;
@@ -35,7 +33,6 @@
 import org.neo4j.gds.api.Graph;
 import org.neo4j.gds.api.GraphStore;
 import org.neo4j.gds.core.Aggregation;
-import org.neo4j.gds.core.concurrency.Concurrency;
 import org.neo4j.gds.core.utils.progress.EmptyTaskRegistryFactory;
 import org.neo4j.gds.core.utils.progress.tasks.TaskProgressTracker;
 import org.neo4j.gds.core.utils.progress.tasks.Tasks;
@@ -96,17 +93,15 @@ void doesNotExportWhenNotAllowed() {
         assertThatExceptionOfType(AuthorizationViolationException.class)
             .isThrownBy(() -> exporter.write("OUT_TYPE"));
     }
-    @ParameterizedTest
-    @ValueSource(ints = {1,2,4})
-    void exportRelationships(int concurrency) {
-        var exporter = setupExportTest(concurrency, /* includeProperties */ true);
+    @Test
+    void exportRelationships() {
+        var exporter = setupExportTest( /* includeProperties */ true);
         exporter.write("FOOBAR", "weight");
         validateWrittenGraph();
     }
 
-    @ParameterizedTest
-    @ValueSource(ints = {1,2,4})
-    void exportRelationshipsWithLongProperties(int concurrency) {
+    @Test
+    void exportRelationshipsWithLongProperties() {
         clearDb();
         runQuery(NODE_QUERY_PART + RELS_QUERY_PART);
 
@@ -125,7 +120,6 @@ void exportRelationshipsWithLongProperties(int concurrency) {
                 graphStore.getGraph(RelationshipType.of("NEW_REL"), Optional.of("newWeight")),
                 RUNNING_TRUE
             )
-            .withConcurrency(new Concurrency(concurrency))
             .withRelationPropertyTranslator(relProperty -> Values.longValue((long) relProperty))
             .build()
             .write("NEW_REL", "newWeight");
@@ -137,7 +131,7 @@ void exportRelationshipsWithLongProperties(int concurrency) {
 
     @Test
     void exportRelationshipsWithoutProperties() {
-        var exporter = setupExportTest(/* includeProperties */ false);
+        var exporter = setupExportTest(/* includeProperties */  false);
         exporter.write("FOOBAR");
         validateWrittenGraphWithoutProperties();
 
@@ -177,7 +171,7 @@ void exportRelationshipsWithAfterWriteConsumer() {
 
     @Test
     void exportRelationshipsWithAfterWriteConsumerAndNoProperties() {
-        var exporter = setupExportTest(/* includeProperties */ false);
+        var exporter = setupExportTest(true);
         MutableInt count = new MutableInt();
         exporter.write("FOOBAR", "weight", (sourceNodeId, targetNodeId, property) -> {
             count.increment();
@@ -241,10 +235,6 @@ void progressLogging() {
     }
 
     private RelationshipExporter setupExportTest(boolean includeProperties) {
-        return setupExportTest(1, includeProperties);
-    }
-
-    private RelationshipExporter setupExportTest(int concurrency, boolean includeProperties) {
         // create graph to export
         clearDb();
         runQuery(NODE_QUERY_PART + RELS_QUERY_PART);
@@ -265,7 +255,6 @@ private RelationshipExporter setupExportTest(int concurrency, boolean includePro
         return NativeRelationshipExporter
             .builder(TestSupport.fullAccessTransaction(db), fromGraph, RUNNING_TRUE)
             .withBatchSize(1)
-            .withConcurrency(new Concurrency(concurrency))
             .build();
     }
 

core-write/src/integrationTest/java/org/neo4j/gds/core/write/NativeRelationshipPropertiesExporterTest.java

@@ -20,8 +20,7 @@
 package org.neo4j.gds.core.write;
 
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.api.Test;
 import org.neo4j.gds.BaseTest;
 import org.neo4j.gds.Orientation;
 import org.neo4j.gds.RelationshipProjection;
@@ -30,7 +29,6 @@
 import org.neo4j.gds.api.DefaultValue;
 import org.neo4j.gds.api.GraphStore;
 import org.neo4j.gds.core.Aggregation;
-import org.neo4j.gds.core.concurrency.Concurrency;
 import org.neo4j.gds.core.utils.progress.tasks.ProgressTracker;
 import org.neo4j.gds.termination.TerminationFlag;
 import org.neo4j.values.storable.Values;
@@ -63,9 +61,8 @@ void setup() {
         runQuery(DB_CYPHER);
     }
 
-    @ParameterizedTest
-    @ValueSource(ints = {1, 2, 3, 4})
-    void shouldWriteRelationshipsWithMultipleProperties(int concurrency) {
+    @Test
+    void shouldWriteRelationshipsWithMultipleProperties() {
         GraphStore graphStore = new StoreLoaderBuilder().databaseService(db)
             .putRelationshipProjectionsWithIdentifier(
                 "PAID",
@@ -80,7 +77,6 @@ void shouldWriteRelationshipsWithMultipleProperties(int concurrency) {
             TestSupport.fullAccessTransaction(db),
             graphStore,
             Values::doubleValue,
-            new Concurrency(concurrency),
             1,
             ProgressTracker.NULL_TRACKER,
             TerminationFlag.RUNNING_TRUE

core-write/src/integrationTest/java/org/neo4j/gds/core/write/NativeRelationshipStreamExporterTest.java

@@ -209,7 +209,6 @@ void exportExceedsBufferSize(int concurrencyValue) {
 
         var exporter = NativeRelationshipStreamExporter
             .builder(TestSupport.fullAccessTransaction(db), graph, relationshipStream, TerminationFlag.RUNNING_TRUE)
-            .withConcurrency(concurrency)
             .withBatchSize(batchSize)
             .build();
 

core-write/src/main/java/org/neo4j/gds/core/write/NativeRelationshipExporter.java

@@ -28,7 +28,6 @@
 import org.neo4j.gds.api.RelationshipWithPropertyConsumer;
 import org.neo4j.gds.compat.Neo4jProxy;
 import org.neo4j.gds.compat.Write;
-import org.neo4j.gds.core.concurrency.Concurrency;
 import org.neo4j.gds.core.concurrency.DefaultPool;
 import org.neo4j.gds.core.concurrency.RunWithConcurrency;
 import org.neo4j.gds.core.utils.partition.Partition;
@@ -49,7 +48,6 @@ public final class NativeRelationshipExporter extends StatementApi implements Re
     private final Graph graph;
     private final LongUnaryOperator toOriginalId;
     private final RelationshipPropertyTranslator propertyTranslator;
-    private final Concurrency concurrency;
     private final long batchSize;
     private final TerminationFlag terminationFlag;
     private final ProgressTracker progressTracker;
@@ -84,7 +82,6 @@ public static RelationshipExporterBuilder builder(
         Graph graph,
         LongUnaryOperator toOriginalId,
         RelationshipPropertyTranslator propertyTranslator,
-        Concurrency concurrency,
         long batchSize,
         TerminationFlag terminationFlag,
         ProgressTracker progressTracker
@@ -93,7 +90,6 @@ public static RelationshipExporterBuilder builder(
         this.graph = graph;
         this.toOriginalId = toOriginalId;
         this.propertyTranslator = propertyTranslator;
-        this.concurrency = concurrency;
         this.batchSize = batchSize;
         this.terminationFlag = terminationFlag;
         this.progressTracker = progressTracker;
@@ -139,7 +135,7 @@ private void write(int relationshipTypeToken, int propertyKeyToken, @Nullable Re
         try {
             RunWithConcurrency
                 .builder()
-                .concurrency(concurrency)
+                .concurrency(RelationshipExporterBuilder.TYPED_DEFAULT_WRITE_CONCURRENCY)
                 .tasks(tasks)
                 .maxWaitRetries(Integer.MAX_VALUE)
                 .waitTime(10L, TimeUnit.MICROSECONDS)

core-write/src/main/java/org/neo4j/gds/core/write/NativeRelationshipExporterBuilder.java

@@ -38,7 +38,6 @@ public NativeRelationshipExporter build() {
             graph,
             toOriginalId,
             propertyTranslator,
-            concurrency,
             batchSize,
             terminationFlag,
             progressTracker

core-write/src/main/java/org/neo4j/gds/core/write/NativeRelationshipPropertiesExporter.java

@@ -26,9 +26,6 @@
 import org.neo4j.gds.api.GraphStore;
 import org.neo4j.gds.compat.Neo4jProxy;
 import org.neo4j.gds.compat.Write;
-import org.neo4j.gds.core.concurrency.Concurrency;
-import org.neo4j.gds.core.concurrency.DefaultPool;
-import org.neo4j.gds.core.concurrency.RunWithConcurrency;
 import org.neo4j.gds.core.utils.partition.DegreePartition;
 import org.neo4j.gds.core.utils.partition.PartitionUtils;
 import org.neo4j.gds.core.utils.progress.tasks.ProgressTracker;
@@ -38,7 +35,6 @@
 import org.neo4j.gds.utils.StatementApi;
 
 import java.util.List;
-import java.util.concurrent.TimeUnit;
 import java.util.function.LongUnaryOperator;
 import java.util.stream.Collectors;
 
@@ -49,7 +45,6 @@ public class NativeRelationshipPropertiesExporter extends StatementApi implement
     private final RelationshipPropertyTranslator propertyTranslator;
     private final ProgressTracker progressTracker;
 
-    private final Concurrency concurrency;
     private final long batchSize;
 
     private final TerminationFlag terminationFlag;
@@ -58,15 +53,13 @@ public class NativeRelationshipPropertiesExporter extends StatementApi implement
         TransactionContext tx,
         GraphStore graphStore,
         RelationshipPropertyTranslator propertyTranslator,
-        Concurrency concurrency,
         long batchSize,
         ProgressTracker progressTracker,
         TerminationFlag terminationFlag
     ) {
         super(tx);
         this.graphStore = graphStore;
         this.propertyTranslator = propertyTranslator;
-        this.concurrency = concurrency;
         this.batchSize = batchSize;
         this.progressTracker = progressTracker;
         this.terminationFlag = terminationFlag;
@@ -104,16 +97,9 @@ public void write(
 
         progressTracker.beginSubTask();
         try {
-            RunWithConcurrency
-                .builder()
-                .concurrency(concurrency)
-                .tasks(tasks)
-                .maxWaitRetries(Integer.MAX_VALUE)
-                .waitTime(10L, TimeUnit.MICROSECONDS)
-                .terminationFlag(terminationFlag)
-                .executor(DefaultPool.INSTANCE)
-                .mayInterruptIfRunning(false)
-                .run();
+            for (Runnable task : tasks) {
+                task.run();
+            }
         } finally {
             progressTracker.endSubTask();
         }
@@ -136,7 +122,6 @@ private Runnable createBatchRunnable(
             partition.consume(nodeId -> {
                 relationshipIterator.forEachRelationship(nodeId, writeConsumer);
             });
-
         });
     }
 

core-write/src/main/java/org/neo4j/gds/core/write/NativeRelationshipPropertiesExporterBuilder.java

@@ -35,7 +35,6 @@ public RelationshipPropertiesExporter build() {
             transactionContext,
             graphStore,
             propertyTranslator,
-            concurrency,
             batchSize,
             progressTracker,
             terminationFlag