Getting Started With ForkJoinPool - Map & Reduce of Java

ForkJoinPool - Java's Map & Reduce

package com.thekarna.fjp;

import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;

/**
 * This class glues all code required for FJP Demo.
 * In many cases given implementation can perform worse then single threaded max finder
 * So don't use this to benchmark, 
 * this is only a Hello World Kind of program to get started with FJP   
 * @author DVG 
 *
 */
public class MaxFinderFJP {
	
	/**
	 * This class extends RecursiveTask class
	 * RecursiveTask class is quite similar to Thread/Runnable/Callable in context to ExecutorService
	 * And compute method of this class is similar to run/call method of Runnable/Callable
	 * compute method is the place where you will write your problem solution and also logic to divide & conquer
	 * You can also extend RecursiveAction but that return void
	 * @author DVG
	 *
	 * @param <T>
	 */
	@SuppressWarnings("all")
	private static class MaxFinderTask<T> extends RecursiveTask<T> {
		private static final int THRESHOLD = 10000;
		private T targetArray[] = null;
		private int low;
		private int high;
		
		/**
		 * Parameterized so the callee can pass data
		 * @param arr
		 */
		public MaxFinderTask(T arr[]) {
			targetArray = arr;
			low = 0;
			high = targetArray.length;
		}
		
		private MaxFinderTask(T arr[], int low, int high) {
			targetArray = arr;
			this.low = low;
			this.high = high;
		}

		/**
		 * Bread & Butter for ForkJoinPool 
		 * All Logic Go Below
		 * @return
		 */
		@Override
		protected T compute() {
			
			/*
				If Task is small then compute it normally as diving more then required will
				result in negative way
			 */
			if (high - low <=MaxFinderTask.THRESHOLD) {
				return this.computeNormal();
			}
			int mid = (high - low) / 2;
			T result = null;
			T result2 = null;
			//Divide The Big Task in Smaller Task So All Processor Can Get There Share
			MaxFinderTask<T> leftTask = new MaxFinderTask<>(this.targetArray, low, low+mid);
			MaxFinderTask<T> rightTask = new MaxFinderTask<>(this.targetArray, low+mid, high);
			//Fork 1st Task, Fork is a non-blocking call so current thread will move ahead.
			leftTask.fork();
			//Call compute directly which will result in again divide & conquer
			result2 = rightTask.compute();
			//Calling Join Result in blocking call but Divide & conquer will go ahead in newly created leftTask
			result = leftTask.join();
			//Merge Results and Return
			return ((Comparable) result).compareTo(((Comparable) result2)) > -1 ? result
					: result2;
		}
		
		/**
		 * This Method Found Max in Normal Way
		 * @return
		 */
		protected T computeNormal() {
			if (this.targetArray.length == 1) {
				return this.targetArray[0];
			}
			Comparable cMax =  (Comparable)this.targetArray[0];
			for(int i=low; i<high; i++) {
				Comparable obj =  (Comparable)this.targetArray[i];
				if(obj.compareTo(cMax) > 0) {
					cMax = obj;
				}
			}
			return (T)cMax;
		}
	}

	/**
	 * This is where all things get going on
	 * @param args
	 */
	public static void main(String[] args) {
		//Create a array of millions/billion entries
		Integer d[] = new Integer[100_000_0];
		
		//Fill it with random data
		for (int i = 0; i < d.length; i++) {
			Random r = new Random(System.nanoTime());
			d[i] = r.nextInt();
		}
		final long startTime = System.currentTimeMillis();
		//Create a ForkJoinPool
		final ForkJoinPool forkJoinPool = new ForkJoinPool();
		
		//Create Object of MaxFinderTask and Call Invoke which is a blocking call
		//You can also use submit method and can use submit which returns Future
		//final Future<Integer> futureResult = forkJoinPool.submit(new MaxFinderTask<Integer>(d));
		final Integer result = forkJoinPool.invoke(new MaxFinderTask<Integer>(d));
		//final Future<Integer> futureResult = forkJoinPool.submit(new MaxFinderTask<Integer>(d));
		forkJoinPool.shutdown();
		
		System.out.println("Max == " + result);
		System.out.println("Time Taken FJP == " + (System.currentTimeMillis() - startTime));

	}
}