import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;

/**
 * This is the final transformation.  Now the calls to sum() occur in
 * parallel thanks to creating the class Sum which implements
 * Callable<Double>.  There are two parallel routines.  The first
 * creates an array of Future<Double> objects that result from
 * submission to the executor, and then calls get() for each of their
 * results.  The second uses CompletionService to allow calls to get()
 * to happen in any order; the array of Futures is not needed.
 */
public class ArrayAverageParallel {

    /**
     * Chunked unit of work for computing the sum.
     */
    private static double sum(int start, int end, Double arr[]) {
        double tmp = 0;
        for (int i = start; i<end; i++) {
            tmp += arr[i];
        }
        return tmp;
    }

    /**
     * Callable for comptuing (portion of) a sum in a separate task
     */
    private static class Sum implements Callable<Double> {
        private Double[] elems;
        private int start, end;
        public Sum(Double[] elems, int start, int end) {
            this.elems = elems;
            this.start = start;
            this.end = end;
        }
        public Double call() {
            return sum(start,end,arr);
        }
    }

    private static int num = 2500000;
    private static Double arr[];
    private static int units = 10;
    private static int nThreads = 10;

    /**
     * Single-threaded computation
     */
    private static void computeST() {
        long start = System.currentTimeMillis();
        double sum = new Sum(arr,0,num).call();
        double result = sum / (double)num;
        long end = System.currentTimeMillis();
        System.out.println("Result (ST, "+(end-start)+" ms) = "+result);
    }

    /**
     * Explicit management of futures
     */
    private static void computeMTFut(ExecutorService executor) {
        long start = System.currentTimeMillis();
        Future<Double>[] futResults = (Future<Double>[])new Future[units];
        for (int i = 0; i<units; i++) {
            int startIdx = i*(num/units);
            int endIdx = (i == units-1) ? num : (i+1)*(num/units);
            futResults[i] = executor.submit(new Sum(arr,startIdx,endIdx));
        }
        try {
            double tmp = 0;
            for (int i = 0; i<units; i++) {
                Double result = futResults[i].get();
                tmp += result;
            }
            tmp = tmp / (double)num;
            long end = System.currentTimeMillis();
            System.out.println("Result (MT-Fut, "+(end-start)+" ms) = "+tmp);
        } catch (Exception e) {
            System.out.println("failure: exception on retrieval");
        }
    }

    /**
     * Future completion managed by completion service
     */
    private static void computeMTCompletion(ExecutorService executor) {
        CompletionService completionService = new
            ExecutorCompletionService(executor);
        long start = System.currentTimeMillis();
        for (int i = 0; i<units; i++) {
            int startIdx = i*(num/units);
            int endIdx = (i == units-1) ? num : (i+1)*(num/units);
            completionService.submit(new Sum(arr,startIdx,endIdx));
        }
        try {
            double tmp = 0;
            for (int i = 0; i<units; i++) {
                Future<Double> futResult = completionService.take();
                tmp += futResult.get();
            }
            tmp = tmp / (double)num;
            long end = System.currentTimeMillis();
            System.out.println("Result (MT-FutComp, "+(end-start)+" ms) = "+tmp);
        } catch (Exception e) {
            System.out.println("failure: exception on retrieval");
        }
    }

    /** 
     * Creates an array of the given size with random contents.
     */
    static private Random rand = new Random();
    public static Double[] createRand(int num) {
        Double[] result = new Double[num];
        for(int i = 0 ; i < num; i++)
            result[i] = rand.nextDouble();
        //result[i] = 2.0;
        return result;
    }

    public static void main(String args[]) {
        boolean doSingle = true;
        boolean doMT = true;
        boolean doMTComp = true;
        if (args.length >= 1)
            num = new Integer(args[0]);
        if (args.length >= 2)
            units = new Integer(args[1]);
        if (args.length >= 3)
            nThreads = new Integer(args[2]);
        if (args.length >= 4) {
            if (args[3].charAt(0) == 'S') {
                doMT = false;
                doMTComp = false;
            }
            else if (args[3].charAt(0) == 'F') {
                doSingle = false;
                doMTComp = false;
            }
            else if (args[3].charAt(0) == 'C') {
                doSingle = false;
                doMT = false;
            }
        }
        System.out.println("array size = "+num);
        System.out.println("(MT) units = "+units);
        System.out.println("(MT) threads = "+nThreads);
        System.out.println("Constructing input array ...");
        System.out.flush();

        arr = createRand(num);

        System.out.println("Starting runs ...");
        System.out.flush();

        // compute the result: execute with futures
        if (doMT || doMTComp) {
            ExecutorService executor = Executors.newFixedThreadPool(nThreads);
            if (doMT)
                computeMTFut(executor);
            
            // compute the result: execute with completion queue
            if (doMTComp)
                computeMTCompletion(executor);

            // cleanup
            executor.shutdown();
        }

        // compute the result: single-threaded
        if (doSingle)
            computeST();

    }
}