added the basic API for dealing with Fork/Join in a convenient way

added PI estimation using the Monte-Carlo method as an example on how to use that API

Author: mguenther

.gitignore

@@ -0,0 +1,2 @@
+target/
+*.iml

pom.xml

@@ -1,18 +1,37 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
-  <modelVersion>4.0.0</modelVersion>
-  <groupId>com.mgu.parallel</groupId>
-  <artifactId>parallelism-in-java</artifactId>
-  <packaging>jar</packaging>
-  <version>1.0-SNAPSHOT</version>
-  <name>parallelism-in-java</name>
-  <url>http://maven.apache.org</url>
-  <dependencies>
-    <dependency>
-      <groupId>junit</groupId>
-      <artifactId>junit</artifactId>
-      <version>3.8.1</version>
-      <scope>test</scope>
-    </dependency>
-  </dependencies>
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+    <modelVersion>4.0.0</modelVersion>
+    <groupId>com.mgu.parallel</groupId>
+    <artifactId>parallelism-in-java</artifactId>
+    <packaging>jar</packaging>
+    <version>0.1.0-SNAPSHOT</version>
+    <name>parallelism-in-java</name>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.1</version>
+                <configuration>
+                    <source>1.8</source>
+                    <target>1.8</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.11</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
 </project>

src/main/java/com/mgu/parallel/AbstractTaskScheduler.java

@@ -0,0 +1,13 @@
+package com.mgu.parallel;
+
+import java.util.concurrent.ForkJoinTask;
+import java.util.function.Supplier;
+
+abstract public class AbstractTaskScheduler implements TaskScheduler {
+
+    @Override
+    public <A, B> Tuple2<A, B> parallel(final Supplier<A> taskA, final Supplier<B> taskB) {
+        final ForkJoinTask<B> right = schedule(taskB);
+        return new Tuple2<A, B>(taskA.get(), right.join());
+    }
+}

src/main/java/com/mgu/parallel/App.java

@@ -0,0 +1,43 @@
+package com.mgu.parallel;
+
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinTask;
+import java.util.concurrent.ForkJoinWorkerThread;
+import java.util.concurrent.RecursiveTask;
+import java.util.function.Supplier;
+
+public class DefaultTaskScheduler extends AbstractTaskScheduler {
+
+    private final ForkJoinPool forkJoinPool;
+
+    public DefaultTaskScheduler() {
+        this(ForkJoinPool.commonPool());
+    }
+
+    public DefaultTaskScheduler(final ForkJoinPool forkJoinPool) {
+        this.forkJoinPool = forkJoinPool;
+    }
+
+    @Override
+    public <T> ForkJoinTask<T> schedule(final Supplier<T> body) {
+
+        final RecursiveTask<T> task = new RecursiveTask<T>() {
+            @Override
+            protected T compute() {
+                return body.get();
+            }
+        };
+
+        if (isForkJoinWorkerThread()) {
+            task.fork();
+        } else {
+            forkJoinPool.execute(task);
+        }
+
+        return task;
+    }
+
+    private boolean isForkJoinWorkerThread() {
+        return Thread.currentThread() instanceof ForkJoinWorkerThread;
+    }
+}

src/main/java/com/mgu/parallel/DefaultTaskScheduler.java

@@ -0,0 +1,30 @@
+package com.mgu.parallel;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ForkJoinTask;
+import java.util.function.Supplier;
+
+public class Schedulers {
+
+    public static final DefaultTaskScheduler SCHEDULER = new DefaultTaskScheduler();
+
+    private static <T> ForkJoinTask<T> task(final Supplier<T> body) {
+        return SCHEDULER.schedule(body);
+    }
+
+    public static <A, B> Tuple2<A, B> parallel(final Supplier<A> taskA, final Supplier<B> taskB) {
+        return SCHEDULER.parallel(taskA, taskB);
+    }
+
+    public static <A, B, C, D> Tuple4<A, B, C, D> parallel(
+            final Supplier<A> taskA,
+            final Supplier<B> taskB,
+            final Supplier<C> taskC,
+            final Supplier<D> taskD) throws ExecutionException, InterruptedException {
+        final ForkJoinTask<A> ta = task(taskA);
+        final ForkJoinTask<B> tb = task(taskB);
+        final ForkJoinTask<C> tc = task(taskC);
+        final ForkJoinTask<D> td = task(taskD);
+        return new Tuple4<>(ta.join(), tb.join(), tc.join(), td.get());
+    }
+}

src/main/java/com/mgu/parallel/Schedulers.java

@@ -0,0 +1,11 @@
+package com.mgu.parallel;
+
+import java.util.concurrent.ForkJoinTask;
+import java.util.function.Supplier;
+
+public interface TaskScheduler {
+
+    <T>ForkJoinTask<T> schedule(Supplier<T> body);
+
+    <A, B> Tuple2<A, B> parallel(Supplier<A> taskA, Supplier<B> taskB);
+}

src/main/java/com/mgu/parallel/TaskScheduler.java

@@ -0,0 +1,12 @@
+package com.mgu.parallel;
+
+public class Tuple2<A, B> {
+
+    public final A a;
+    public final B b;
+
+    public Tuple2(final A a, final B b) {
+        this.a = a;
+        this.b = b;
+    }
+}

src/main/java/com/mgu/parallel/Tuple2.java

@@ -0,0 +1,16 @@
+package com.mgu.parallel;
+
+public class Tuple4<A, B, C, D> {
+
+    public final A a;
+    public final B b;
+    public final C c;
+    public final D d;
+
+    public Tuple4(final A a, final B b, final C c, final D d) {
+        this.a = a;
+        this.b = b;
+        this.c = c;
+        this.d = d;
+    }
+}

src/main/java/com/mgu/parallel/Tuple4.java

@@ -0,0 +1,50 @@
+package com.mgu.parallel.pi;
+
+import com.mgu.parallel.Tuple4;
+
+import java.util.Random;
+import java.util.concurrent.ExecutionException;
+
+import static com.mgu.parallel.Schedulers.parallel;
+
+public class PiEstimation {
+
+    public double computeSequentially(final int iterations) {
+        return 4.0 * monteCarloCount(iterations) / iterations;
+    }
+
+    public double computeParallel(final int iterations) throws ExecutionException, InterruptedException {
+        final Tuple4<Integer, Integer, Integer, Integer> result = parallel(
+                () -> monteCarloCount(iterations/4),
+                () -> monteCarloCount(iterations/4),
+                () -> monteCarloCount(iterations/4),
+                () -> monteCarloCount(iterations-3*iterations/4));
+        return 4.0 * (result.a + result.b + result.c + result.d) / iterations;
+    }
+
+    private int monteCarloCount(final int iterations) {
+        final Random randomX = new Random();
+        final Random randomY = new Random();
+        int hits = 0;
+        for (int i = 0; i < iterations; i++) {
+            final double x = randomX.nextDouble();
+            final double y = randomY.nextDouble();
+            if (x*x + y*y < 1) hits++;
+        }
+        return hits;
+    }
+
+    public static void main(String[] args) throws ExecutionException, InterruptedException {
+        final PiEstimation piMonteCarlo = new PiEstimation();
+        long start = System.nanoTime();
+        final double piSeq = piMonteCarlo.computeSequentially(50000000);
+        long end = System.nanoTime();
+        long duration = (end - start) / 1000000;
+        System.out.println("Computed PI (" + piSeq + ") sequentially in " + duration + " ms.");
+        start = System.nanoTime();
+        final double piPar = piMonteCarlo.computeParallel(50000000);
+        end = System.nanoTime();
+        duration = (end - start) / 1000000;
+        System.out.println("Computed PI (" + piPar + ") parallel in " + duration + " ms.");
+    }
+}