package com.rationalcoding.sort;import java.util.ArrayList;import java.util.Arrays;import java.util.concurrent.ExecutionException;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Future;/** *  * @author Phani *  *         This is implementation of a multi threaded merge sort. First input *         array is divided into small chunks. Each chunk is sorted individually *         using Arrays.sort method which implements insertion sort Then all the *         chunks are merged in bottom up manner by doubling the width after *         each step *  */public class MultiThreadedMergeSort extends SequentialMergeSort {    private final int DEFAULT_POOL_SIZE = 100;    private final int DEFAULT_CHUNK_SIZE = 1000;    private ExecutorService pool = Executors.newFixedThreadPool(DEFAULT_POOL_SIZE);    private int[] arr;    @SuppressWarnings("rawtypes")    @Override    public void sort(int[] arr) {        this.arr = arr;        // handle null inputs        if (arr == null) {            throw new IllegalArgumentException("Input array cannot be null");        }        if (arr.length == 0 || arr.length == 1) {            // already sorted return            return;        }        // width is chunks we are diving the array into        int width = DEFAULT_CHUNK_SIZE;        if (width > 1) {            // apply insertion sort on chunks and then merge the chunks            ArrayList<Future> subTasks = new ArrayList<Future>();            for (int i = 0; i < arr.length; i = i + width) {                int start = i;                int end = i + width;                if (end > arr.length) {                    end = arr.length;                }                // add the runnables to pool                subTasks.add(pool.submit(new InsertionSortRunnable(start, end)));            }            // wait for the tasks to finish            // join all the threads to main thread            for (Future f : subTasks) {                try {                    f.get();                } catch (InterruptedException e) {                    e.printStackTrace();                } catch (ExecutionException e) {                    e.printStackTrace();                }            }        }        // apply merging on already sorted chunks        for (; width < arr.length; width = width * 2) {            ArrayList<Future> tasks = new ArrayList<Future>();            for (int i = 0; i < arr.length; i = i + 2 * width) {                int rStart = i;                int rEnd = i + width - 1;                int lEnd = i + 2 * width - 1;                if (lEnd >= arr.length) {                    lEnd = arr.length - 1;                }                tasks.add(pool.submit(new MergeSortRunnable(rStart, rEnd, lEnd)));            }            // wait for all threads to finish            for (Future f : tasks) {                try {                    f.get();                } catch (InterruptedException e) {                    e.printStackTrace();                } catch (ExecutionException e) {                    e.printStackTrace();                }            }        }    }    private class InsertionSortRunnable implements Runnable {        private int start;        private int end;        public InsertionSortRunnable(int start, int end) {            this.start = start;            this.end = end;        }        @Override        public void run() {            Arrays.sort(arr, start, end);        }    }    private class MergeSortRunnable implements Runnable {        private int start;        private int end;        private int mid;        public MergeSortRunnable(int start, int mid, int end) {            this.start = start;            this.end = end;            this.mid = mid;        }        @Override        public void run() {            if (start < end && mid <= end) {                merge(arr, start, mid, end);            }        }    }}