05. ArraySum Fork/Join

README.md

@@ -11,6 +11,9 @@
 * Create FixedThreadPool
 * Create ScheduledThreadPool
 
+### Introduce to ForkJoin Framework
+* ArraySum using Recursive Action
+
 ### Acknowledgments
 Inspiration, code snippets, etc.
 * [Java - MultiThreading](https://www.youtube.com/watch?v=9DvDheKRJ9Y&t=656s)

src/main/java/ArraySum.java

+import java.util.Arrays;
+import java.util.concurrent.RecursiveAction;
+
+public class ArraySum extends RecursiveAction {
+    public long sum;
+    int arr[], lo, hi;
+
+    public ArraySum(int[] arr, int lo, int hi) {
+        this.arr = arr;
+        this.lo = lo;
+        this.hi = hi;
+    }
+
+    public long computeSeq() {
+//        if(lo == hi)
+//            return arr[lo];
+//        else {
+//            int mid = (lo + hi) / 2;
+//            ArraySum left = new ArraySum(arr, lo, mid);
+//            ArraySum right = new ArraySum(arr, mid + 1, hi);
+//
+//            return left.computeSeq()+ right.computeSeq();
+//        }
+        for (int i = lo; i <= hi; ++i) {
+            sum += arr[i];
+        }
+        return sum;
+    }
+
+    @Override
+    protected void compute() {
+        if (hi - lo > 1_000_000) {
+            int mid = (lo + hi) / 2;
+            ArraySum left = new ArraySum(arr, lo, mid);
+            ArraySum right = new ArraySum(arr, mid + 1, hi);
+            left.fork();
+            right.compute();
+            left.join();
+            sum = left.sum + right.sum;
+        } else {
+//            sum =computeSeq();
+            for (int i = lo; i <= hi; ++i) {
+                sum += arr[i];
+            }
+        }
+    }
+
+    public void computeStream() {
+        sum = Arrays.stream(arr).asLongStream().parallel().sum();
+//        all the intermediate and final operation run in parallel
+//         Introduce to Map Reduce pattern credit to java streams
+    }
+}

src/test/java/ArraySumTest.java

+import junit.framework.TestCase;
+
+import java.util.Random;
+import java.util.concurrent.ForkJoinPool;
+import java.util.stream.IntStream;
+
+public class ArraySumTest extends TestCase {
+
+
+    private int[] randomArray(int size) {
+        Random rand = new Random();
+        int[] arr = new int[size];
+        for (int i = 0; i < size; i++) {
+            arr[i] = rand.nextInt(size);
+        }
+        /* Check  random array
+        Arrays.stream(arr).forEach(System.out::println);*/
+        return arr;
+    }
+
+    private int[] intStream(int size) {
+        IntStream intStream = IntStream.range(1, size);
+        return intStream.toArray();
+    }
+
+    public void testArraySumSeq() {
+
+        int size = 1000_000_000;
+        int[] arr = intStream(size);
+
+        ArraySum array = new ArraySum(arr, 0, arr.length - 1);
+        long start = System.currentTimeMillis();
+        long sum = array.computeSeq();
+        long endTimer = System.currentTimeMillis() - start;
+        System.out.printf("Sequential Time execution for Random Array of size %d is %d sum is %d\n", size, endTimer, sum);
+//        assertEquals(15,res);
+    }
+
+    public void testArraySumPP() {
+        System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","7");
+
+        int size = 1000_000_000;
+        int[] arr = intStream(size);
+
+        ArraySum array = new ArraySum(arr, 0, arr.length - 1);
+        long start = System.currentTimeMillis();
+        ForkJoinPool.commonPool().invoke(array);
+        long endTimer = System.currentTimeMillis() - start;
+        System.out.printf("Parallel Time execution for Random Array of size %d is %d ms sum is %d\n", size, endTimer, array.sum);
+//        assertEquals(15,res);
+    }
+
+    public void testArraySumStream() {
+        System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","7");
+
+        int size = 1000_000_000;
+        int[] arr = intStream(size);
+
+        ArraySum array = new ArraySum(arr, 0, arr.length - 1);
+        long start = System.currentTimeMillis();
+        array.computeStream();
+        long endTimer = System.currentTimeMillis() - start;
+        System.out.printf("Parallel Stream Time execution for Random Array of size %d is %d ms sum is %d\n", size, endTimer, array.sum);
+//        assertEquals(15,res);
+    }
+
+    public void resource() {
+        System.out.println(Runtime.getRuntime().availableProcessors());
+        System.out.println(ForkJoinPool.commonPool().getParallelism());
+    }
+
+}