SM_Terasort.java

/**
 * Created by lchen on 11/15/17.
 *
 * @author: Linlin Chen
 * @email: lchen96@hawk.iit.edu
 *
 * This file is used to test the terasort speed in single node with shared memory mode.
 * Quick sort is used to sort the text data
 *
 */


import java.io.*;
import java.util.*;
import java.lang.Runnable;
import java.util.concurrent.ArrayBlockingQueue;


public class SM_Terasort {
    /**
     * private variables
     */
    private final static String configfile = "SM_config.properties";     //config file

    private static String sortpath;                 //sort file path
    private static String resultpath;               //store the result path
    private static String sortfile;                 //generated sort file
    private static int LINESIZE;       //the size of one line in bytes, in generated sort file (default is 100B)
    private static long CHUNKSIZE;      //the size of one chunk, in number of lines, one chunk will be sorted independently to meet the memory requirement
    private static int NUMCHUNKS;      //number of such chunks
    private static int NUMTHREADS;     //number of threads

    private static int TXTLENGTH;      //for the file generated by gensort, only the top 10 characters are context to be sorted


    /**
     * public variables
     */
    public static volatile long filechunknum;                        //the file chunk number, which will be shared by multiple threads
    public static volatile long buffersize;                         //buffer size each thread can use to read and write file

    public static volatile double readDataGB;                      //data size in GB been read
    public static volatile  double writeDataGB;                    //data size in GB been write

    public ArrayList<ArrayList<String>> dataChunks;             //used to store the data chunks for each thread

    public final String newlinechar = System.getProperty("line.separator");     //system's new line separator character

    /**
     * protected variables
     */

    protected long readtime;        //file read time
    protected long writetime;       //file write time
    protected long sorttime;        //sort time
    protected double filesizeGB;       //sort file size in GB
    protected double totalreadfilesizeGB;      //total file read size in GB
    protected double totalwritefilesizeGB;     //total file write size in GB

    protected long filesizebyte;        //sort file size in byte


    protected int[] chunkNumsPerThrd;           //number of chunks for each thread
    protected int[] startChunkNum;             //1st chunk number for each thread
    protected int[] endChunkNum;                //last chunk number for each thread

    protected long[] startFilePosition;        //start position in the file for each thread
    protected long[] endFilePosition;        //end position in the file for each thread


    protected long [] readtimebythread;     //file read time by each thread in 1st round
    protected long [] writetimebythread;    //file write time by each thread in 1st round
    protected long [] sorttimebythread;     //file sort time by each thread in 1st round




    /**
     *
     * @param autoconfig: whether use auto configuration based on system's info
     */
    SM_Terasort (boolean autoconfig) {
        if (autoconfig) {
            System.out.println("Use system's info to automatically configure");
        }
        else {
            System.out.println("Load configuration from config file");
        }
        //load the configuration info
        if (!loadConfig(autoconfig)) {
            System.out.println("ERROR: Fail to load the config file");
            System.out.println("Exit the program now...");
            System.exit(1);
        }
        //list the system information,
        listMemInfo(autoconfig);

        setConfig();        // set other configurations, which needed be calculated based on the loaded info

        //initialize some members, which will be used in each thread
        chunkNumsPerThrd = new int [NUMTHREADS];
        startChunkNum = new int[NUMTHREADS];
        endChunkNum = new int[NUMTHREADS];

        startFilePosition = new long[NUMTHREADS];
        endFilePosition = new long[NUMTHREADS];

        readtimebythread = new long[NUMTHREADS];
        writetimebythread = new long[NUMTHREADS];
        sorttimebythread = new long[NUMTHREADS];

        for (int i = 0; i < NUMTHREADS; i++) {
            chunkNumsPerThrd[i] = 0;
            startChunkNum[i] = 0;
            endChunkNum[i] = 0;

            startFilePosition[i] = 0;
            endFilePosition[i] = 0;

            readtimebythread[i] = 0;
            writetimebythread[i] = 0;
            sorttimebythread[i] = 0;
        }

        this.dataChunks= new ArrayList<>(NUMTHREADS);
        for (int i = 0; i < NUMTHREADS; i++) {
            dataChunks.add(new ArrayList<String >(Math.toIntExact(CHUNKSIZE)));
        }

        readtime = 0;
        sorttime = 0;
        filesizeGB = bytestoGB(new File(sortpath, sortfile).length());
        totalreadfilesizeGB = 0;
        totalwritefilesizeGB = 0;

        workAssignPerThread();

    }

    /**
     * This method is used to load the config file from disk
     * Set some fields' values
     * @param autoconfig: whether use auto configuration based on system's info
     * @return: whether loading configs successfully
     */
    private boolean loadConfig (boolean autoconfig) {
        Properties prop = new Properties();
        //open the config file, and load all configurations
        FileInputStream in = null;
        try {
            in = new FileInputStream(this.configfile);
            prop.load(in);
            sortpath = prop.getProperty("sortpath");
            resultpath = prop.getProperty("resultpath");
            sortfile = prop.getProperty("sortfile");
            TXTLENGTH = Integer.parseInt(prop.getProperty("TXTLENGTH"));
            LINESIZE = Integer.parseInt(prop.getProperty("LINESIZE"));
            //if not auto config, then load this two settings from file
            if (!autoconfig) {
                CHUNKSIZE = Long.parseLong(prop.getProperty("CHUNKSIZE"));
                NUMTHREADS = Integer.parseInt(prop.getProperty("NUMTHREADS"));
            }
        } catch (IOException e1) {
            e1.printStackTrace();
            return false;
        } finally {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e2) {
                    e2.printStackTrace();
                }
            }
        }

        return true;
    }

    /**
     * This method is used to list the system memory's information.
     * @param autoconfig: whether use auto configuration based on system's info
     */
    private void listMemInfo (boolean autoconfig) {
        //run garbage collector
        System.gc();
        Runtime rm = Runtime.getRuntime();
        int numProcessor = rm.availableProcessors();
        long usedMem = rm.totalMemory() - rm.freeMemory();
        long availableFreeMem = rm.maxMemory() - usedMem;

        //print the system information
        System.out.println();
        System.out.println("Number of processors: " + numProcessor);
        System.out.println("Maximum memory: " +  getOptimalSize(rm.maxMemory()));
        System.out.println("Total memory: " +  getOptimalSize(rm.totalMemory()));
        System.out.println("Used memory: " + getOptimalSize(usedMem));
        System.out.println("Available free memory: " + getOptimalSize(availableFreeMem));

        if (autoconfig) {
            NUMTHREADS = numProcessor;
            //use half of availableFreeMem to sort, each thread will use one full chunk
            CHUNKSIZE = availableFreeMem /2 / NUMTHREADS;
        } else {
            while (CHUNKSIZE * NUMTHREADS * LINESIZE > availableFreeMem) {
                System.out.println("Total memory size to be assigned exceeds the available free memory size");
                CHUNKSIZE /= 2;
            }
        }
    }

    /**
     * transform size in bytes to size in GB, MB, or KB
     * @param bytes: size in bytes
     * @return: size in GB, MB or KB
     */
    private static double bytestoGB (long bytes) {
        return ((double)bytes / 1024 / 1024 / 1024);
    }
    private static double bytestoMB (long bytes) {
        return ((double)bytes / 1024 / 1024);
    }
    private static double bytestoKB (long bytes) {
        return ((double)bytes / 1024);
    }

    /**
     * For better reading, I write a function to automatically represent the bytes in a optimal representation
     * Will choose to use KB, MB or GB to represent the size according to the bytes size
     *
     */
    private static String getOptimalSize (long bytes) {
        double bytesinGB = bytestoGB(bytes);
        double bytesinMB = bytestoMB(bytes);
        double bytesinKB = bytestoKB(bytes);
        if (bytesinGB >= 1.0) {
            return String.format("%.2f GB", bytesinGB);
        } else if (bytesinMB >= 1.0) {
            return String.format("%.2f MB", bytesinMB);
        } else
            return String.format("%.2f KB", bytesinKB);
    }


    /**
     * Set other members' values, which need be calculated
     */
    private void setConfig () {
        readDataGB = 0;
        writeDataGB = 0;

        filesizebyte = new File(sortpath, sortfile).length();

        //make sure the chunksize can be divided by LINESIZE
        //CHUNKSIZE = CHUNKSIZE / LINESIZE * LINESIZE;

        NUMCHUNKS = (int) Math.ceil((double)new File(sortpath, sortfile).length() / LINESIZE / CHUNKSIZE);

        System.out.println();
        System.out.println("File to be sorted: " + sortpath + sortfile + " (" + getOptimalSize(new File(sortpath, sortfile).length()) + ")");
        System.out.println("Number of threads: " + NUMTHREADS);
        System.out.println("One line in Bytes: " + LINESIZE);
        System.out.println("Number of lines in one chunk: " + CHUNKSIZE);

    }

    /**
     * Assign the file block for each thread to sort
     * Will assign the file start pointer and end pointer for each thread
     *
     * Initialize some flag values, which will be used for each thread, to track the sorting task
     */
    void workAssignPerThread () {
        long filelength = new File(sortpath, sortfile).length();        //file size in bytes
//        for (int i = 0; i < NUMTHREADS - 1; i++) {
//            this.chunkNumsPerThrd[i] = NUMCHUNKS / NUMTHREADS;
//        }
//        this.chunkNumsPerThrd[NUMTHREADS-1] = NUMCHUNKS - (NUMTHREADS-1)*(NUMCHUNKS / NUMTHREADS);

        int totalchunks = 0;
        for (int i = 0; i < NUMTHREADS; i++) {
            chunkNumsPerThrd[i] = NUMCHUNKS / NUMTHREADS;
            totalchunks += chunkNumsPerThrd[i];
        }
        int restchunks = NUMCHUNKS - totalchunks;
        for (int i = 0; i < NUMTHREADS; i++) {
            if (restchunks > 0) {
                chunkNumsPerThrd[i] += 1;
                restchunks --;
            }
        }


        this.startChunkNum[0] = 0;
        this.startFilePosition[0] = 0;
        this.endChunkNum[0] = this.chunkNumsPerThrd[0] - 1;
        this.endFilePosition[0] = this.chunkNumsPerThrd[0] * CHUNKSIZE * LINESIZE;

        for (int i = 1; i < NUMTHREADS; i++) {
            this.startChunkNum[i] = this.endChunkNum[i-1] + 1;
            this.endChunkNum[i] = this.endChunkNum[i-1] + this.chunkNumsPerThrd[i];

            this.startFilePosition[i] = this.endFilePosition[i-1];
            this.endFilePosition[i] = this.endFilePosition[i-1] + this.chunkNumsPerThrd[0] * CHUNKSIZE * LINESIZE;
        }

    }


    /**
     * The merge sort helper
     * Implement all the logical flow for the sorting
     *
     * Firstly, divide files into chunks and solve it by each thread
     * After 1st sorting about each chunk, each thread merge the chunks solved by them into one large chunk
     * Lastly, merge the #THREAD large chunks merged by each thread, into one whole sort file
     * Finish the sorting
     */
    void mergeHelper () {
        //max number of chunks per thread must exist between either the 1st thread or last thread
        //int maxchunknumPerThrd = chunkNumsPerThrd[NUMTHREADS-1] > chunkNumsPerThrd[0] ? chunkNumsPerThrd[NUMTHREADS-1] : chunkNumsPerThrd[0];
        long starttime, endtime;

        System.out.println("First round sorting begins...");
        System.out.println("\tSplit files into " + NUMCHUNKS + " chunks and then sort each chunk");

        BufferedReader reader = null;

        /**
         * The first round sorting
         * read data into #threads chunks, then sort it with each thread
         * write sorted file back to the disk
         */
        try {
            reader = new BufferedReader(new FileReader(new File(sortpath, sortfile)));
            starttime = System.currentTimeMillis();
            //for (int chunknum = 0; chunknum < Math.ceil((double)NUMCHUNKS/NUMTHREADS); chunknum++) {
            //    readFileIntoChunks(reader);

            //the largest number of chunks can only be at the 1st thread's chunk num
            //So iterate chunkNumsPerThrd[0] times will ensure reading the whole file
            for (int chunknum = 0; chunknum < chunkNumsPerThrd[0]; chunknum ++) {
                //read data
                readFileIntoChunks(reader);

                endtime = System.currentTimeMillis();
                readtime += (endtime - starttime);

                //sort data and write data
                Thread[] chunksort = new Thread[NUMTHREADS];
                for (int thridx = 0; thridx < NUMTHREADS; thridx++) {
                    if (!dataChunks.get(thridx).isEmpty()) {
                        chunksort[thridx] = new Thread(new chunksSortByThread(thridx, chunknum));
                        chunksort[thridx].start();
                    }
                }
                for (int thridx = 0; thridx < NUMTHREADS; thridx++) {
                    try {
                        if (!dataChunks.get(thridx).isEmpty()) {
                            chunksort[thridx].join();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        } catch (IOException e1) {
            e1.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

 //       long avgreadtime = 0;
        long avgwritetime = 0;
        long avgsorttime = 0;
        for (int i = 0; i < NUMTHREADS; i++) {
   //         avgreadtime += readtimebythread[i];
            avgwritetime += writetimebythread[i];
            avgsorttime += sorttimebythread[i];
        }
     //   avgreadtime /= NUMTHREADS;
        avgwritetime /= NUMTHREADS;
        avgsorttime /= NUMTHREADS;


//        for (int chunkidx = 0; chunkidx < maxchunknumPerThrd; chunkidx ++) {
//            boolean[] ifthreadrun = new boolean[NUMTHREADS];        //if the corresponding thread is resumed
//            Arrays.fill(ifthreadrun, false);
//            Thread[] readthreads = new Thread[NUMTHREADS];
//
//            //read chunks to array by for each thread
//            starttime = System.currentTimeMillis();
//            for (int thrdidx = 0; thrdidx < NUMTHREADS; thrdidx ++) {
//                if (chunkNumsPerThrd[thrdidx] >= chunkidx) {
//                    dataChunks.get(thrdidx).clear();        //clear all the data stored in previous chunk
//                    ifthreadrun[thrdidx] = true;
//                    readthreads[thrdidx] = new Thread(new readChunksFromFile(thrdidx, startFilePosition[thrdidx] + chunkidx * CHUNKSIZE * LINESIZE));
//                    readthreads[thrdidx].start();
//                }
//            }
//            //wait each thread finishing reading data chunks
//            for (int thrdidx = 0; thrdidx < NUMTHREADS; thrdidx ++) {
//                if (ifthreadrun[thrdidx]) {
//                    try {
//                        readthreads[thrdidx].join();
//                    } catch (InterruptedException e) {
//                        e.printStackTrace();
//                    }
//                }
//            }
//            endtime = System.currentTimeMillis();
//            readtime += (endtime - starttime);
//
//            //each thread implement sorting of responsible chunks
//            Thread[] sortthreads = new Thread[NUMTHREADS];
//            starttime = System.currentTimeMillis();
//            for (int thrdidx = 0; thrdidx < NUMTHREADS; thrdidx ++) {
//                if (ifthreadrun[thrdidx]) {
//                    sortthreads[thrdidx] = new Thread(new firstRoundChunkSort(thrdidx));
//                    sortthreads[thrdidx].start();
//                }
//            }
//            //wait each thread finishing sorting
//            for (int thrdidx = 0; thrdidx < NUMTHREADS; thrdidx ++) {
//                if (ifthreadrun[thrdidx]) {
//                    try {
//                        sortthreads[thrdidx].join();
//                    } catch (InterruptedException e) {
//                        e.printStackTrace();
//                    }
//                }
//            }
//            endtime = System.currentTimeMillis();
//            //sorttime += (endtime - starttime);
//
//            //write chunks into file
//            Thread[] writethreads = new Thread[NUMTHREADS];
//            starttime = System.currentTimeMillis();
//            for(int thrdidx = 0; thrdidx < NUMTHREADS; thrdidx ++) {
//                if (ifthreadrun[thrdidx]) {
//                    writethreads[thrdidx] = new Thread(new writeChunksFromFile(thrdidx, startChunkNum[thrdidx] + chunkidx));
//                    writethreads[thrdidx].start();
//                }
//            }
//            //wait each thread finishing writing chunks
//            for (int thrdidx = 0; thrdidx < NUMTHREADS; thrdidx ++) {
//                if (ifthreadrun[thrdidx]) {
//                    try {
//                        writethreads[thrdidx].join();
//                    } catch (InterruptedException e) {
//                        e.printStackTrace();
//                    }
//                }
//            }
//            endtime = System.currentTimeMillis();
//            writetime += (endtime - starttime);
//
//        }
        totalreadfilesizeGB += filesizeGB;      //read the whole sort file for once
        totalwritefilesizeGB += filesizeGB;     //write the whole sort file to disk for once

        //delete the whole file
        //new File(sortpath, sortfile).delete();

        //release the consumed resource in memory
        dataChunks = null;
        System.gc();

        System.out.println();
        System.out.println("Second round sorting begins...");
        System.out.println("\tEach thread merges its own chunks into one whole file");
        starttime = System.currentTimeMillis();
        //each thread merge their own chunks into one whole file
        Thread[] mergethread = new Thread[NUMTHREADS];
        for (int thrdidx = 0; thrdidx < NUMTHREADS; thrdidx++) {
            mergethread[thrdidx] = new Thread(new mergeChunksIntoOneChunk(thrdidx));
            mergethread[thrdidx].start();
        }
        //wait each thread finishing merging chunks
        for (int thrdidx = 0; thrdidx < NUMTHREADS; thrdidx++) {
            try {
                mergethread[thrdidx].join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        endtime = System.currentTimeMillis();
        sorttime += (endtime - starttime);


        System.out.println();
        System.out.println("Third round sorting begins...");
        System.out.println("\tMerge chunk file merged by each thread into whole file, and finish the sorting");
        starttime = System.currentTimeMillis();
        //merge #threads chunks into one whole file and finish the sorting
        mergeSortedChunksPerThrdIntoOneWholeFile();
        endtime = System.currentTimeMillis();
        sorttime += (endtime - starttime);


        System.out.println();
        System.out.println("DONE!");


        System.out.println();
        System.out.println("Total data read: " + getOptimalSize(filesizebyte*3));
        System.out.println("Total data written: " + getOptimalSize(filesizebyte*3));

        System.out.println();
        System.out.println("First round data read time: " + timeinsecond(readtime) + "s");
        //System.out.println("First round data read size: " + getOptimalSize(filesizebyte));
        //System.out.println("First round data read time: " + timeinsecond(avgreadtime) + "s");
        //System.out.println("File read throughput： " + (bytestoMB(filesizebyte) / timeinsecond(readtime)) + " MB/S");
        System.out.println("File read throughput： " + (bytestoMB(filesizebyte) / (timeinsecond(readtime)/NUMTHREADS)) + " MB/S");
        System.out.println();
        //System.out.println("First round data written time: " + timeinsecond(writetime) + "s");
        System.out.println("First round data write time: " + timeinsecond(avgwritetime) + "s");
        System.out.println("First round data written size: " + getOptimalSize((filesizebyte)));
        //System.out.println("File written throughput： " + (bytestoMB(filesizebyte) / timeinsecond(writetime)) + " MB/S");
        System.out.println("File written throughput： " + (bytestoMB(filesizebyte) / (double)NUMTHREADS / timeinsecond(avgwritetime)) + " MB/S");
        System.out.println();

        System.out.println("First round data sort time: " + timeinsecond(avgsorttime) + "s");
        System.out.println();

        System.out.println("K-way merge sort takes time: " + timeinsecond(sorttime) + "s");

    }

    private static double timeinsecond (double ms) {
        return (ms/1000);
    }

    /**
     * This method is used to merge sorted chunks by each thread into one whole file
     * Finish the whole sorting procedure
     */
    void mergeSortedChunksPerThrdIntoOneWholeFile () {
        final int writebuffersize = Math.toIntExact(CHUNKSIZE/2 * NUMTHREADS);     //buffer size to buffering writing content
        final int readbuffersize = Math.toIntExact(CHUNKSIZE/2);       //read buffer size used for each file

        //Queue<String> readLinesPerFile;     //read buffer to cache each file's read contents

        //used to buffer the read data from each chunk file
        //will load again until the queue is empty
        ArrayBlockingQueue<String> [] readdatabuffer = new ArrayBlockingQueue [NUMTHREADS];
        for (int i = 0; i < NUMTHREADS; i++) {
            readdatabuffer[i] = new ArrayBlockingQueue<String>(readbuffersize);
        }

        //used for writing buffer
       // ArrayBlockingQueue<String> writedatabuffer = new ArrayBlockingQueue<String>(writebuffersize);
        ArrayList<String> writedatabuffer = new ArrayList<>(writebuffersize);

        //priority queue (heap), used as k-merge sort for speed augmenting
        PriorityQueue<priorityQueueElement> datachunksqueue;

        //file array, contains the chunk files which are left by each thread
        File[] chunkfiles;

        //initialize the priority queue, overriding a new comparator function.
        datachunksqueue = new PriorityQueue<>(NUMTHREADS, new Comparator<priorityQueueElement>() {
            //override a new comparator
            //compare two elements based on the first 10 characters
            @Override
            public int compare(priorityQueueElement o1, priorityQueueElement o2) {
                return o1.getData().substring(0, TXTLENGTH).compareTo(o2.getData().substring(0, TXTLENGTH));
            }
        });       //maintain a priority queue for k-way merge sort

        //filter the data chunks with ending "-2d"
        chunkfiles = new File(resultpath).listFiles(new FilenameFilter() {
            @Override
            public boolean accept(File dir, String name) {
                int endpos = 0;
                if ((endpos = name.indexOf('-')) == -1 || !name.endsWith("-2nd"))
                    return false;

                int fileidx = Integer.parseInt(name.substring(0, name.indexOf('-')));
                return fileidx >= 0 && fileidx < NUMTHREADS;
            }
        });                 //list all files start with integer within this thread's responsible chunk index

        //expected number of chunk files should equal to the number of threads
        if (chunkfiles.length != NUMTHREADS) {
            System.out.println("Error: chunk number is different with expected value: " + NUMTHREADS);
            System.exit(1);
        }

        int finishedfilenum = 0;            //number of files finished sorting, check whether all the chunk files have been read and sorted
        boolean finishflag[] = new boolean[NUMTHREADS];     //indicating whether reading is finished for each chunk file
        for (int chunkidx = 0; chunkidx < NUMTHREADS; chunkidx++) {
            finishflag[chunkidx] = false;
        }

        BufferedWriter bfwriter = null;
        HashMap <Integer, BufferedReader> filereaderMaps = new HashMap<>(NUMTHREADS);

        try {
            //initialize a buffer writer to write data into whole file
            bfwriter = new BufferedWriter(new FileWriter(new File(resultpath, "finished-" + sortfile)));

            //build a hashmap, store <file_id, readbuffer> pair
            for (int chunkidx = 0; chunkidx < NUMTHREADS; chunkidx ++) {
                filereaderMaps.put(chunkidx, new BufferedReader(new FileReader(chunkfiles[chunkidx])));
            }

            //load each chunk file into read buffer
            for (int chunkidx = 0; chunkidx < NUMTHREADS; chunkidx++) {
                loadReadBuffer(readdatabuffer[chunkidx], filereaderMaps.get(chunkidx));
            }

            for (int chunkidx = 0; chunkidx < NUMTHREADS; chunkidx++) {
                datachunksqueue.add(new priorityQueueElement(readdatabuffer[chunkidx].poll(), chunkidx));
            }

            priorityQueueElement element;
            while ((element = datachunksqueue.poll()) != null) {
                //get the string data, and its file index
                String data = element.getData();
                int idx = element.getChunkIdx();

                //if the write buffer is full, write all contents back to the disk
                if (writedatabuffer.size() == readbuffersize) {
                    writeBuffertoFile(writedatabuffer, bfwriter);
                    writedatabuffer.clear();
                }


                //put the minimum value to the write buffer
                writedatabuffer.add(data);

                if (finishedfilenum != NUMTHREADS) {     //not all files finished reading
                    //the corresponding read buffer is empty, load data from disk file to the read buffer
                    if (readdatabuffer[idx].isEmpty()) {
                        if (!finishflag[idx]) {         //idx-th file does not finish reading, keep reading data to read buffer
                            if (!loadReadBuffer(readdatabuffer[idx], filereaderMaps.get(idx))) {        //false, means idx-file finishes reading
                                finishflag[idx] = true;
                                finishedfilenum++;
                            }
                        }
                    }
                }
                if (!readdatabuffer[idx].isEmpty()) {       //successfully load data to the read buffer
                    //load the next data from the read buffer to the heap
                    datachunksqueue.add(new priorityQueueElement(readdatabuffer[idx].poll(), idx));
                }

            }
            writeBuffertoFile(writedatabuffer, bfwriter);

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            for (File file : chunkfiles) {
                file.delete();
            }
            if (bfwriter != null) {
                try {
                    bfwriter.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }
            for (int chunkidx = 0; chunkidx < NUMTHREADS; chunkidx ++) {
                try {
                    filereaderMaps.get(chunkidx).close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }

    /**
     * This method is used to load data from the given reader to a queue
     * queue size is fixed by the CHUNKSIZE
     * return true if the file not reaches the end; otherwise return false if the file finishes reading
     * @param readdatabuffer: the buffer to load the data from the file
     * @param reader: bufferedreader, reader for a given file chunk
     * @return: true, if the file can be further read; false, if the file reaches the end
     */
    private boolean loadReadBuffer (ArrayBlockingQueue<String>  readdatabuffer, BufferedReader reader) {
        boolean notreachfileend = true;       //indicate whether reach the end of the file

        String line;
        while(readdatabuffer.remainingCapacity() != 0) {
            try {
                if ((line = reader.readLine()) != null) {
                    readdatabuffer.put(line);
                } else {
                    notreachfileend = false;
                    break;
                }
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return notreachfileend;

    }


    /**
     * This method is used to write the write buffer back to the disk file, using BufferedWriter writer
     * @param writedatabuffer: the write buffer, buffering the writing contents
     * @param writer: the BufferedWriter
     */
    private void writeBuffertoFile (ArrayList<String> writedatabuffer, BufferedWriter writer) {
        for (String line : writedatabuffer) {
            try {
                writer.write(line);
                writer.write("\r\n");
                //writer.newLine();
            } catch (IOException e) {
                e.printStackTrace();
            }

        }

    }




    /**
     * Main function
     * @param args
     */
    public static void main (String[] args) {
        SM_Terasort smsort = new SM_Terasort(false);
        smsort.mergeHelper();

    }


    /**
     * This threads implements the sorting in first round for each chunk file
     * Sort each chunk of the file, file location based on start and end position
     * Nested class
     */
    class firstRoundChunkSort implements  Runnable {
        private int thrdID;     //thread ID number

        firstRoundChunkSort (int thrdID) {
            this.thrdID = thrdID;
        }

        @Override
        public  void run () {
           //sort the data chunks responsible by current thread, comparing the first TXTLENGTH string of each line
            Comparator<String> comparator = (String s1, String s2) -> {
                return s1.substring(0, TXTLENGTH).compareTo(s2.substring(0, TXTLENGTH));
            };
            Collections.sort(dataChunks.get(thrdID), comparator);
        }

    }

    /**
     * This threads implements the reading from the single file into chunks for firstRoundChunkSort
     * Seperate this function for the purpose of measuring I/O time.
     * Without measuring I/O time requirement, there is no need to seperate this
     */
    class readChunksFromFile implements Runnable {
        private int thrdID;     //thread ID number
        long position;          //seek position

        readChunksFromFile(int thrdID, long position) {
            this.thrdID = thrdID;
            this.position = position;
        }

        @Override
        public void run() {
            RandomAccessFile readfile = null;
            try {
                readfile = new RandomAccessFile(new File(sortpath, sortfile), "r");
                readfile.seek(position);        //seeks the file position to read desired space

                String line = null;
                for (long i = 0; i < CHUNKSIZE; i++) {
                    if ((line = readfile.readLine()) != null) {
                        dataChunks.get(thrdID).add(line);
                    } else {            //hit the file end
                        break;
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                if (readfile != null) {
                    try {
                        readfile.close();
                    } catch (IOException e1) {
                        e1.printStackTrace();
                    }
                }
            }

        }
    }

    /**
     * this threads implementing the writing chunks into each file
     */
    class writeChunksFromFile implements Runnable {
        private int thrdID;
        private int chunkidx;

        writeChunksFromFile(int thrdID, int chunkidx) {
            this.thrdID = thrdID;
            this.chunkidx = chunkidx;
        }

        @Override
        public void run() {
            BufferedWriter wtfile = null;
            try {
                wtfile = new BufferedWriter(new FileWriter(new File(resultpath, chunkidx + "-" + sortfile + "-1st")));
                for (String line : dataChunks.get(thrdID)) {
                    wtfile.write(line);
                    wtfile.write("\r\n");
                    //wtfile.newLine();
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                if (wtfile != null) {
                    try {
                        wtfile.close();
                    } catch (IOException e2) {
                        e2.printStackTrace();
                    }
                }
            }

        }
    }


    /**
     * This method is used to read the file sequentially into the data chunks
     * Then used for each thread to sort
     * I found multithread reading a same file will decrease the performace, so I warp with single thread read method
     * for faster speed reading, as I test the bottleneck would be the reading in multithreading
     */
    private void readFileIntoChunks (BufferedReader reader) {

        for (int thrdidx = 0; thrdidx < NUMTHREADS; thrdidx++) {
            dataChunks.get(thrdidx).clear();
            int linenum = 0;
            String line = null;
            try {
                while (linenum < CHUNKSIZE && (line = reader.readLine()) != null) {
                    dataChunks.get(thrdidx).add(line);
                    linenum++;
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * This thread is implemented for each thread to sort its corresponding file chunks
     * This will merge the read, sort, and write together for each thread
     * Will measure the read, sort and write time by each thread, and then calculating the average
     */
    class chunksSortByThread implements Runnable {
        private int threadID;
        private long skippos;

        private long chunkidx;      //chunkidx, used for writer file nameing
        private int chunknum;       //chunk number

        private long starttime;
        private long endtime;
        private long lasttime;

        //private BufferedReader filereader;
        //private RandomAccessFile filereader;

        //private ArrayList<String> readerbuffer;



        chunksSortByThread(int threadID, int chunknum) {
            this.threadID = threadID;
            skippos = startFilePosition[threadID];

            chunkidx = startChunkNum[threadID];
            this.chunknum = chunknum;

            starttime = 0;
            endtime = 0;
            lasttime = 0;

            //readerbuffer = new ArrayList<String>(Math.toIntExact(CHUNKSIZE));


//            try {
//                //filereader = new BufferedReader(new FileReader(new File(sortpath, sortfile)));
//                //filereader = new RandomAccessFile(new File(sortpath, sortfile), "r");
//                //filereader.skip(skippos);
//                //filereader.seek(skippos);
//
//            } catch (FileNotFoundException e) {
//                e.printStackTrace();
//            } catch (IOException e) {
//                e.printStackTrace();
//            }
        }

        @Override
        public void run() {

            //rewrite the comparator, only compare the 1st TEXTLENGTH string, which is 10 in our case
            Comparator<String> comparator = (String s1, String s2) -> {
                return s1.substring(0, TXTLENGTH).compareTo(s2.substring(0, TXTLENGTH));
            };

//            try {
                //for (int chunknum = 0; chunknum < chunkNumsPerThrd[threadID]; chunknum++) {
                    //readerbuffer.clear();
//                    int linenum = 0;
//                    String line = null;
//                    //read file testing
//                    starttime = System.currentTimeMillis();
//                    while (linenum < CHUNKSIZE && ((line = filereader.readLine()) != null)) {
//                        readerbuffer.add(line);
//                        linenum++;
//                    }
//                    endtime = System.currentTimeMillis();
//                    lasttime = endtime - starttime;
//                    readtimebythread[threadID] += lasttime;

                    //sorting testing
                    starttime = System.currentTimeMillis();
                    //Collections.sort(readerbuffer, comparator);
                    Collections.sort(dataChunks.get(threadID), comparator);
                    endtime = System.currentTimeMillis();
                    lasttime = endtime - starttime;
                    sorttimebythread[threadID] += lasttime;

                    //write file testing
                    BufferedWriter writer = null;
                    try {
                        writer = new BufferedWriter(new FileWriter(new File(resultpath, (chunkidx + chunknum) + "-" + sortfile + "-1st")));
                        starttime = System.currentTimeMillis();
                        //writeBuffertoFile(readerbuffer, writer);
                        writeBuffertoFile(dataChunks.get(threadID), writer);
                        endtime = System.currentTimeMillis();
                        lasttime = endtime - starttime;
                        writetimebythread[threadID] += lasttime;
                    } catch (IOException ee) {
                        ee.printStackTrace();
                    } finally {
                        if (writer != null) {
                            try {
                                writer.close();
                            } catch (IOException eee) {
                                eee.printStackTrace();
                            }
                        }
                    }

                //}
//            } catch (IOException e) {
//                e.printStackTrace();
//            } finally {
//                if (filereader != null) {
//                    try {
//                        filereader.close();
//                    } catch (IOException e) {
//                        e.printStackTrace();
//                    }
//                }
//           }
        }


    }




    /**
     * This threads merge all chunks responsible by each thread into one ordered chunk
     */
    class mergeChunksIntoOneChunk implements Runnable {
        private int threadID;
        private int chunknums;      //number of chunks responsible by this thread
        private int startchunkidx;      //start index for the chunks responsible by this thread
        PriorityQueue<priorityQueueElement> datachunksqueue;
        File[] chunkfiles;

        //private final static int buffersize = 1000;     //cache 1000 lines each time beforing writing back to file
        final int writebuffersize;     //buffer size to buffering writing content
        final int readbuffersize;       //read buffer size used for each file


        mergeChunksIntoOneChunk(int thrid) {
            this.threadID = thrid;
            this.chunknums = chunkNumsPerThrd[threadID];
            this.startchunkidx = startChunkNum[threadID];

            writebuffersize = Math.toIntExact(CHUNKSIZE/2);     //buffer size to buffering writing content
            readbuffersize = Math.toIntExact(CHUNKSIZE/2/chunknums);       //read buffer size used for each file

            datachunksqueue = new PriorityQueue<>(chunknums, new Comparator<priorityQueueElement>() {
                //override a new comparator
                //compare two elements based on the first 10 characters
                @Override
                public int compare(priorityQueueElement o1, priorityQueueElement o2) {
                    return o1.getData().substring(0, TXTLENGTH).compareTo(o2.getData().substring(0, TXTLENGTH));
                }
            });       //maintain a priority queue for k-way merge sort


            chunkfiles = new File(resultpath).listFiles(new FilenameFilter() {
                @Override
                public boolean accept(File dir, String name) {
                    int endpos = 0;
                    if ((endpos = name.indexOf('-')) == -1 || !name.endsWith("-1st"))
                        return false;

                    int fileidx = Integer.parseInt(name.substring(0, name.indexOf('-')));
                    return fileidx >= startChunkNum[threadID] && fileidx <= endChunkNum[threadID];
                }
            });                 //list all files start with integer within this thread's responsible chunk index

            if (chunkfiles.length != chunknums) {
                System.out.println("Error: chunk number is different with expected value " + chunknums + " for thread " + threadID);
                System.exit(1);
            }
        }

        @Override
        public void run() {
            //used to buffer the read data from each chunk file
            //will load again until the queue is empty
            ArrayBlockingQueue<String> [] readdatabuffer = new ArrayBlockingQueue [chunknums];
            for (int i = 0; i < chunknums; i++) {
                readdatabuffer[i] = new ArrayBlockingQueue<String>(readbuffersize);
            }

            //used for writing buffer
            // ArrayBlockingQueue<String> writedatabuffer = new ArrayBlockingQueue<String>(writebuffersize);
            ArrayList<String> writedatabuffer = new ArrayList<>(writebuffersize);


            int finishedfilenum = 0;            //number of files finished sorting, check whether all the chunk files have been read and sorted
            boolean finishflag[] = new boolean[chunknums];     //indicating whether reading is finished for each chunk file
            for (int chunkidx = 0; chunkidx < chunknums; chunkidx++) {
                finishflag[chunkidx] = false;
            }

            BufferedWriter bfwriter = null;
            HashMap <Integer, BufferedReader> filereaderMaps = new HashMap<>(chunknums);


            try {
                //initialize a buffer writer to write data into whole file
                bfwriter = new BufferedWriter(new FileWriter(new File(resultpath, threadID + "-" + sortfile + "-2nd")));

                //build a hashmap, store <file_id, readbuffer> pair
                for (int chunkidx = 0; chunkidx < chunknums; chunkidx ++) {
                    filereaderMaps.put(chunkidx, new BufferedReader(new FileReader(chunkfiles[chunkidx])));
                }

                //load each chunk file into read buffer
                for (int chunkidx = 0; chunkidx < chunknums; chunkidx++) {
                    loadReadBuffer(readdatabuffer[chunkidx], filereaderMaps.get(chunkidx));
                }

                for (int chunkidx = 0; chunkidx < chunknums; chunkidx++) {
                    datachunksqueue.add(new priorityQueueElement(readdatabuffer[chunkidx].poll(), chunkidx));
                }

                priorityQueueElement element;
                while ((element = datachunksqueue.poll()) != null) {
                    //get the string data, and its file index
                    String data = element.getData();
                    int idx = element.getChunkIdx();

                    //if the write buffer is full, write all contents back to the disk
                    if (writedatabuffer.size() == readbuffersize) {
                        writeBuffertoFile(writedatabuffer, bfwriter);
                        writedatabuffer.clear();
                    }

                    //put the minimum value to the write buffer
                    writedatabuffer.add(data);

                    if (finishedfilenum != chunknums) {     //not all files finished reading
                        //the corresponding read buffer is empty, load data from disk file to the read buffer
                        if (readdatabuffer[idx].isEmpty()) {
                            if (!finishflag[idx]) {         //idx-th file does not finish reading, keep reading data to read buffer
                                if (!loadReadBuffer(readdatabuffer[idx], filereaderMaps.get(idx))) {        //false, means idx-file finishes reading
                                    finishflag[idx] = true;
                                    finishedfilenum++;
                                }
                            }
                        }
                    }
                    if (!readdatabuffer[idx].isEmpty()) {       //successfully load data to the read buffer
                        //load the next data from the read buffer to the heap
                        datachunksqueue.add(new priorityQueueElement(readdatabuffer[idx].poll(), idx));
                    }

                }
                writeBuffertoFile(writedatabuffer, bfwriter);

            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                //delete the chunk files
                for (File file : chunkfiles) {
                    file.delete();
                }
                if (bfwriter != null) {
                    try {
                        bfwriter.close();
                    } catch (IOException e1) {
                        e1.printStackTrace();
                    }
                }
                for (int chunkidx = 0; chunkidx < NUMTHREADS; chunkidx ++) {
                    try {
                        filereaderMaps.get(chunkidx).close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }

        }


    }


    class priorityQueueElement {
        String data;
        int chunkidx;

        priorityQueueElement(String data, int chunkidx) {
            this.data = data;
            this.chunkidx = chunkidx;
        }

        public String getData () {return this.data;}
        public int getChunkIdx () {return this.chunkidx;}
    }



}