Feedback

README.md

@@ -1,3 +1,4 @@
+[![Review Assignment Due Date](https://classroom.github.com/assets/deadline-readme-button-22041afd0340ce965d47ae6ef1cefeee28c7c493a6346c4f15d667ab976d596c.svg)](https://classroom.github.com/a/AwTYhPar)
 # Лабораторная работа № 1: определение достижимости параллелизма и реализация параллельных алгоритмов.
 
 Шаги выполнения:

artifacts/jmh/graph-size.txt

@@ -0,0 +1,29 @@
+Benchmark                       (graphSize)  Mode  Cnt    Score     Error  Units
+JmhBfsTest.forkJoinBfs                  100  avgt    4    0.043 ±   0.026  ms/op
+JmhBfsTest.forkJoinBfs                 1000  avgt    4    0.226 ±   0.066  ms/op
+JmhBfsTest.forkJoinBfs                10000  avgt    4    2.217 ±   0.288  ms/op
+JmhBfsTest.forkJoinBfs                50000  avgt    4   11.955 ±   2.038  ms/op
+JmhBfsTest.forkJoinBfs               100000  avgt    4   23.961 ±   2.808  ms/op
+JmhBfsTest.forkJoinBfs              1000000  avgt    4  234.495 ±  69.199  ms/op
+JmhBfsTest.forkJoinBfs              2000000  avgt    4  437.866 ± 152.712  ms/op
+JmhBfsTest.parallelBfs                  100  avgt    4    2.571 ±   0.988  ms/op
+JmhBfsTest.parallelBfs                 1000  avgt    4    2.468 ±   0.873  ms/op
+JmhBfsTest.parallelBfs                10000  avgt    4    4.318 ±   1.568  ms/op
+JmhBfsTest.parallelBfs                50000  avgt    4   12.274 ±   1.017  ms/op
+JmhBfsTest.parallelBfs               100000  avgt    4   26.078 ±   1.949  ms/op
+JmhBfsTest.parallelBfs              1000000  avgt    4  459.465 ± 286.051  ms/op
+JmhBfsTest.parallelBfs              2000000  avgt    4  772.640 ± 393.633  ms/op
+JmhBfsTest.parallelFrontierBfs          100  avgt    4    2.187 ±   1.741  ms/op
+JmhBfsTest.parallelFrontierBfs         1000  avgt    4    2.117 ±   0.881  ms/op
+JmhBfsTest.parallelFrontierBfs        10000  avgt    4    5.444 ±   3.305  ms/op
+JmhBfsTest.parallelFrontierBfs        50000  avgt    4   16.782 ±  11.075  ms/op
+JmhBfsTest.parallelFrontierBfs       100000  avgt    4   32.159 ±   9.104  ms/op
+JmhBfsTest.parallelFrontierBfs      1000000  avgt    4  354.178 ±  72.852  ms/op
+JmhBfsTest.parallelFrontierBfs      2000000  avgt    4  626.251 ± 216.028  ms/op
+JmhBfsTest.serialBfs                    100  avgt    4    0.002 ±   0.001  ms/op
+JmhBfsTest.serialBfs                   1000  avgt    4    0.046 ±   0.005  ms/op
+JmhBfsTest.serialBfs                  10000  avgt    4    0.721 ±   0.172  ms/op
+JmhBfsTest.serialBfs                  50000  avgt    4    5.449 ±   3.297  ms/op
+JmhBfsTest.serialBfs                 100000  avgt    4   20.219 ±   4.305  ms/op
+JmhBfsTest.serialBfs                1000000  avgt    4  383.653 ±  34.153  ms/op
+JmhBfsTest.serialBfs                2000000  avgt    4  622.543 ±  70.652  ms/op

artifacts/jmh/threads.txt

@@ -0,0 +1,15 @@
+Benchmark                               (runtime)  Mode  Cnt    Score     Error  Units
+JmhBfsResourceTest.parallelBfs                  1  avgt    4  811.736 ± 532.065  ms/op
+JmhBfsResourceTest.parallelBfs                  2  avgt    4  800.443 ± 213.065  ms/op
+JmhBfsResourceTest.parallelBfs                  4  avgt    4  786.551 ± 188.788  ms/op
+JmhBfsResourceTest.parallelBfs                  6  avgt    4  746.592 ± 310.785  ms/op
+JmhBfsResourceTest.parallelBfs                 12  avgt    4  726.113 ± 108.879  ms/op
+JmhBfsResourceTest.parallelBfs                 16  avgt    4  746.983 ± 143.357  ms/op
+JmhBfsResourceTest.parallelBfs                 24  avgt    4  787.820 ± 211.231  ms/op
+JmhBfsResourceTest.parallelFrontierBfs          1  avgt    4  967.328 ± 479.491  ms/op
+JmhBfsResourceTest.parallelFrontierBfs          2  avgt    4  623.180 ± 302.594  ms/op
+JmhBfsResourceTest.parallelFrontierBfs          4  avgt    4  558.315 ± 282.455  ms/op
+JmhBfsResourceTest.parallelFrontierBfs          6  avgt    4  732.789 ± 298.786  ms/op
+JmhBfsResourceTest.parallelFrontierBfs         12  avgt    4  798.175 ± 286.191  ms/op
+JmhBfsResourceTest.parallelFrontierBfs         16  avgt    4  746.080 ± 201.253  ms/op
+JmhBfsResourceTest.parallelFrontierBfs         24  avgt    4  775.214 ± 253.786  ms/op

build.gradle.kts

@@ -1,6 +1,9 @@
 plugins {
     kotlin("jvm") version "1.9.20"
+    java
     application
+    id("me.champeau.jmh") version "0.7.3"
+    // id("org.openjdk.jcstress") version "0.15"
 }
 
 group = "org.itmo"
@@ -12,16 +15,44 @@ repositories {
 
 dependencies {
     testImplementation(kotlin("test"))
+    testImplementation("org.openjdk.jcstress:jcstress-core:0.16")
+    testAnnotationProcessor("org.openjdk.jcstress:jcstress-core:0.16")
 }
 
 tasks.test {
     useJUnitPlatform()
 }
 
 kotlin {
-    jvmToolchain(8)
+    jvmToolchain(21)
 }
 
 application {
     mainClass.set("MainKt")
-}
+}
+
+jmh {
+    jvmArgs = listOf("--enable-native-access=ALL-UNNAMED", "--add-opens=java.base/java.lang=ALL-UNNAMED")
+}
+
+sourceSets {
+    test {
+        java.srcDirs("src/test/kotlin", "src/test/java")
+    }
+}
+
+// JCStress runner task: runs JCStress tests located on the test runtime classpath
+// Use: ./gradlew jcstress [-PjcstressArgs="-v -m quick"]
+tasks.register<JavaExec>("jcstress") {
+    group = "verification"
+    description = "Run JCStress stress tests"
+    mainClass.set("org.openjdk.jcstress.Main")
+    classpath = sourceSets.test.get().runtimeClasspath
+    dependsOn("testClasses")
+
+    val argsProp = project.findProperty("jcstressArgs") as String?
+    if (!argsProp.isNullOrBlank()) {
+        args = argsProp.split("\\s+".toRegex())
+    }
+}
+

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.2-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.7-bin.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

src/jmh/kotlin/org/itmo/bfs/JmhBfsResourceTest.kt

@@ -0,0 +1,63 @@
+package org.itmo.bfs
+
+import org.itmo.Graph
+import org.itmo.RandomGraphGenerator
+import org.itmo.bfs.impl.ParallelBfs
+import org.itmo.bfs.impl.ParallelFrontierBfs
+import org.openjdk.jmh.annotations.Benchmark
+import org.openjdk.jmh.annotations.BenchmarkMode
+import org.openjdk.jmh.annotations.Fork
+import org.openjdk.jmh.annotations.Measurement
+import org.openjdk.jmh.annotations.Mode
+import org.openjdk.jmh.annotations.OutputTimeUnit
+import org.openjdk.jmh.annotations.Param
+import org.openjdk.jmh.annotations.Scope
+import org.openjdk.jmh.annotations.Setup
+import org.openjdk.jmh.annotations.State
+import org.openjdk.jmh.annotations.Warmup
+import org.openjdk.jmh.infra.Blackhole
+import java.util.concurrent.TimeUnit
+import kotlin.math.min
+
+@Fork(1)
+@State(Scope.Benchmark)
+@Warmup(iterations = 1, time = 5)
+@Measurement(iterations = 4, time = 5)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+open class JmhBfsResourceTest {
+
+    @Param("1", "2", "4", "6", "12", "16", "24")
+    private var runtime = 0
+
+    private var connections = 0
+    private lateinit var graph: Graph
+
+    private lateinit var parallelFrontierBfs: Bfs
+    private lateinit var parallelBfs: Bfs
+
+    @Setup
+    fun setup() {
+        val graphSize = 2_000_000
+        connections = min(graphSize * 10, 10_000_000)
+
+        println("Генерация графа размером $graphSize с $connections рёбрами...")
+
+        graph = RandomGraphGenerator().generateGraph(java.util.Random(42), graphSize, connections)
+
+        println("Генерация завершена.")
+
+        parallelBfs = ParallelBfs(runtime)
+        parallelFrontierBfs = ParallelFrontierBfs(runtime)
+    }
+
+    @Benchmark
+    fun parallelBfs(bh: Blackhole) {
+        bh.consume(parallelBfs.bfs(graph, 0))
+    }
+
+    @Benchmark
+    fun parallelFrontierBfs(bh: Blackhole) {
+        bh.consume(parallelFrontierBfs.bfs(graph, 0))
+    }
+}

src/jmh/kotlin/org/itmo/bfs/JmhBfsTest.kt

@@ -0,0 +1,75 @@
+package org.itmo.bfs
+
+import org.itmo.Graph
+import org.itmo.RandomGraphGenerator
+import org.itmo.bfs.impl.ForkJoinBfs
+import org.itmo.bfs.impl.ParallelBfs
+import org.itmo.bfs.impl.ParallelFrontierBfs
+import org.itmo.bfs.impl.SimpleBfs
+import org.openjdk.jmh.annotations.Benchmark
+import org.openjdk.jmh.annotations.BenchmarkMode
+import org.openjdk.jmh.annotations.Fork
+import org.openjdk.jmh.annotations.Measurement
+import org.openjdk.jmh.annotations.Mode
+import org.openjdk.jmh.annotations.OutputTimeUnit
+import org.openjdk.jmh.annotations.Param
+import org.openjdk.jmh.annotations.Scope
+import org.openjdk.jmh.annotations.Setup
+import org.openjdk.jmh.annotations.State
+import org.openjdk.jmh.annotations.Warmup
+import org.openjdk.jmh.infra.Blackhole
+import java.util.concurrent.TimeUnit
+import kotlin.math.min
+
+@Fork(1)
+@State(Scope.Benchmark)
+@Warmup(iterations = 1, time = 5)
+@Measurement(iterations = 4, time = 5)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+open class JmhBfsTest {
+
+    @Param("10", "100", "1000", "10000", "50000", "100000", "1000000", "2000000")
+    private var graphSize = 0
+
+    private var connections = 0
+    private lateinit var graph: Graph
+
+    private val runtime = Runtime.getRuntime().availableProcessors() * 2
+
+    private val simpleBfs = SimpleBfs()
+    private val forkJoinBfs = ForkJoinBfs()
+    private val parallelFrontierBfs = ParallelFrontierBfs(runtime)
+    private val parallelBfs = ParallelBfs(runtime)
+
+    @Setup
+    fun setup() {
+        connections = min(graphSize * 10, 10_000_000)
+
+        println("Генерация графа размером $graphSize с $connections рёбрами...")
+
+        graph = RandomGraphGenerator().generateGraph(java.util.Random(42), graphSize, connections)
+
+        println("Генерация завершена.")
+    }
+
+    @Benchmark
+    fun serialBfs(bh: Blackhole) {
+        bh.consume(simpleBfs.bfs(graph, 0))
+    }
+
+    @Benchmark
+    fun parallelBfs(bh: Blackhole) {
+        bh.consume(parallelBfs.bfs(graph, 0))
+    }
+
+    @Benchmark
+    fun parallelFrontierBfs(bh: Blackhole) {
+        bh.consume(parallelFrontierBfs.bfs(graph, 0))
+    }
+
+    @Benchmark
+    fun forkJoinBfs(bh: Blackhole) {
+        bh.consume(forkJoinBfs.bfs(graph, 0))
+    }
+}

src/main/java/org/itmo/Graph.java

@@ -1,16 +1,13 @@
 package org.itmo;
 
 import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
 
-class Graph {
-    private final int V;
+public class Graph {
+    private final int size;
     private final ArrayList<Integer>[] adjList;
 
     Graph(int vertices) {
-        this.V = vertices;
+        this.size = vertices;
         adjList = new ArrayList[vertices];
         for (int i = 0; i < vertices; ++i) {
             adjList[i] = new ArrayList<>();
@@ -23,28 +20,12 @@ void addEdge(int src, int dest) {
         }
     }
 
-    void parallelBFS(int startVertex) {
+    public int getSize() {
+        return size;
     }
 
-    //Generated by ChatGPT
-    void bfs(int startVertex) {
-        boolean[] visited = new boolean[V];
-
-        LinkedList<Integer> queue = new LinkedList<>();
-
-        visited[startVertex] = true;
-        queue.add(startVertex);
-
-        while (!queue.isEmpty()) {
-            startVertex = queue.poll();
-
-            for (int n : adjList[startVertex]) {
-                if (!visited[n]) {
-                    visited[n] = true;
-                    queue.add(n);
-                }
-            }
-        }
+    // not safe, but today I don't care
+    public ArrayList<Integer>[] getAdjList() {
+        return adjList;
     }
-
 }

src/main/java/org/itmo/bfs/Bfs.java

@@ -0,0 +1,7 @@
+package org.itmo.bfs;
+
+import org.itmo.Graph;
+
+public interface Bfs {
+    void bfs(Graph graph, int startVertex);
+}

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

@@ -0,0 +1,33 @@
+package org.itmo.bfs.impl;
+
+import java.util.Collection;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicIntegerArray;
+import org.itmo.Graph;
+import org.itmo.bfs.Bfs;
+
+public class ForkJoinBfs implements Bfs {
+
+    @Override
+    public void bfs(Graph graph, int startVertex) {
+        var visited = new AtomicIntegerArray(graph.getSize());
+        visited.set(startVertex, 1);
+
+        // we can use faster collection, bz we don't remove elements and queue is sequential (linearizable) on CAS
+        Collection<Integer> frontier = new ConcurrentLinkedQueue<>();
+        frontier.add(startVertex);
+
+        while (!frontier.isEmpty()) {
+            Queue<Integer> nextFrontier = new ConcurrentLinkedQueue<>();
+
+            // actually this need bulk optimization for unbalanced graphs
+            frontier.parallelStream()
+                    .flatMap(v -> graph.getAdjList()[v].stream())
+                    .filter(neighbor -> visited.compareAndSet(neighbor, 0, 1))
+                    .forEach(nextFrontier::add);
+
+            frontier = nextFrontier;
+        }
+    }
+}