add parallel impl, but it sucks

Author: ar.r.lysenko

src/main/java/org/itmo/bfs/impl/ParallelBfs.java

@@ -1,12 +1,82 @@
 package org.itmo.bfs.impl;
 
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicIntegerArray;
+import java.util.stream.IntStream;
 import org.itmo.Graph;
 import org.itmo.bfs.Bfs;
 
 public class ParallelBfs implements Bfs {
 
+    private final int threadNumber;
+
+    public ParallelBfs(int threadNumber) {
+        this.threadNumber = threadNumber;
+    }
+
     @Override
     public void bfs(Graph graph, int startVertex) {
+        AtomicInteger activeNodes = new AtomicInteger(1);
+        AtomicIntegerArray distances = new AtomicIntegerArray(graph.getSize());
+        ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
+
+        queue.add(startVertex);
+        distances.set(startVertex, 0);
+
+        IntStream.range(0, threadNumber)
+                .mapToObj(i ->
+                        new Thread(new BfsWorker(graph, distances, queue, activeNodes))
+                )
+                .peek(Thread::start)
+                .forEach(thread -> {
+                    try {
+                        thread.join();
+                    } catch (Exception e) {
+                        throw new RuntimeException(e);
+                    }
+                });
+    }
+
+    private record BfsWorker(
+            Graph graph,
+            AtomicIntegerArray distances,
+            Queue<Integer> queue,
+            AtomicInteger activeNodes
+    ) implements Runnable {
+
+        @Override
+        public void run() {
+            var adjList = graph.getAdjList();
+            while (activeNodes.get() > 0) {
+                var currentVertex = queue.poll();
+                if (currentVertex == null) {
+                    continue;
+                }
+                int d = distances.get(currentVertex);
+                for (var neighbor : adjList[currentVertex]) {
+                    if (updateDistanceIfLower(neighbor, d - 1)) {
+                        queue.add(neighbor);
+                        activeNodes.incrementAndGet();
+                    }
+                }
+                activeNodes.decrementAndGet();
+            }
+        }
+
+        private boolean updateDistanceIfLower(int node, int newDistance) {
+            while (true) {
+                int currentDistance = distances.get(node);
+
+                if (newDistance >= currentDistance || currentDistance != 0) {
+                    return false;
+                }
 
+                if (distances.compareAndSet(node, currentDistance, newDistance)) {
+                    return true;
+                }
+            }
+        }
     }
 }

src/test/java/org/itmo/BFSTest.java

@@ -28,6 +28,24 @@ public void kek() {
         System.out.println("ForkJoin " + time2);
     }
 
+    @Test
+    public void lotOfConnectionsTest() {
+        int size = 1_000_000;
+        int connections = 15_000_000;
+        Random r = new Random(42);
+
+        System.out.println("Generating graph of size " + size + " ...wait");
+        Graph g = new RandomGraphGenerator().generateGraph(r, size, connections);
+        System.out.println("Generation completed!\nStarting bfs");
+
+        long serialTime = executeSerialBfsAndGetTime(g);
+        long parallelTime = executeParallelBfsAndGetTime(g);
+
+        System.out.println("Times for " + size + " vertices and " + connections + " connections: ");
+        System.out.println("Serial: " + serialTime);
+        System.out.println("Parallel: " + parallelTime);
+    }
+
     @Test
     public void bfsTest() throws IOException {
         int[] sizes = new int[]{10, 100, 1000, 10_000, 10_000, 50_000, 100_000, 1_000_000, 2_000_000};
@@ -40,9 +58,11 @@ public void bfsTest() throws IOException {
                 Graph g = new RandomGraphGenerator().generateGraph(r, sizes[i], connections[i]);
                 System.out.println("Generation completed!\nStarting bfs");
                 long serialTime = executeSerialBfsAndGetTime(g);
+                long forkJoinTime = executeForkJoinBfs(g);
                 long parallelTime = executeParallelBfsAndGetTime(g);
                 fw.append("Times for " + sizes[i] + " vertices and " + connections[i] + " connections: ");
                 fw.append("\nSerial: " + serialTime);
+                fw.append("\nForkJoin: " + forkJoinTime);
                 fw.append("\nParallel: " + parallelTime);
                 fw.append("\n--------\n");
             }
@@ -59,7 +79,8 @@ private long executeSerialBfsAndGetTime(Graph g) {
     }
 
     private long executeParallelBfsAndGetTime(Graph g) {
-        var bfs = new ParallelBfs(1);
+        var runtime = Runtime.getRuntime().availableProcessors() * 2;
+        var bfs = new ParallelBfs(runtime);
         long startTime = System.nanoTime();
         bfs.bfs(g, 0);
         long endTime = System.nanoTime();

tmp/results.txt

@@ -1,36 +1,45 @@
 Times for 10 vertices and 50 connections: 
 Serial: 0
-Parallel: 0
+ForkJoin: 3
+Parallel: 6
 --------
 Times for 100 vertices and 500 connections: 
-Serial: 0
-Parallel: 0
+Serial: 1
+ForkJoin: 1
+Parallel: 38
 --------
 Times for 1000 vertices and 5000 connections: 
-Serial: 1
-Parallel: 0
+Serial: 2
+ForkJoin: 2
+Parallel: 19
 --------
 Times for 10000 vertices and 50000 connections: 
 Serial: 4
-Parallel: 0
+ForkJoin: 18
+Parallel: 10
 --------
 Times for 10000 vertices and 100000 connections: 
-Serial: 3
-Parallel: 0
+Serial: 2
+ForkJoin: 17
+Parallel: 11
 --------
 Times for 50000 vertices and 1000000 connections: 
-Serial: 22
-Parallel: 0
+Serial: 27
+ForkJoin: 15
+Parallel: 34
 --------
 Times for 100000 vertices and 1000000 connections: 
-Serial: 21
-Parallel: 0
+Serial: 33
+ForkJoin: 30
+Parallel: 40
 --------
 Times for 1000000 vertices and 10000000 connections: 
-Serial: 419
-Parallel: 0
+Serial: 600
+ForkJoin: 251
+Parallel: 649
 --------
 Times for 2000000 vertices and 10000000 connections: 
-Serial: 755
-Parallel: 0
+Serial: 932
+ForkJoin: 539
+Parallel: 971
 --------