import java.util.StringTokenizer;
import java.util.Map;
import java.util.HashMap;
import java.io.BufferedReader;
import java.io.FileReader;

/** This is the parallel version of WordCount.java by way of
 * WordCountParallel.java.  The key change is that we use an
 * AtomicInteger that we can update in place to avoid the shenanigans
 * needed to perform updates in WordCountParallel.
 */

import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class WordCountParallelAtomicInt implements Runnable {
    private final String buffer;
    private final ConcurrentMap<String,AtomicInteger> counts;

    public WordCountParallelAtomicInt
        (String buffer, 
         ConcurrentMap<String,AtomicInteger> counts) 
    {
        this.counts = counts;
        this.buffer = buffer;
    }

    private final static String DELIMS = " :;,.{}()\t\n";
    private final static boolean printAll = false;

    /**
     * Looks for the last delimiter in the string, and returns its
     * index.
     */
    private static int findDelim(String buf) {
        for (int i = buf.length() - 1; i>=0; i--) {
            for (int j = 0; j < DELIMS.length(); j++) {
                char d = DELIMS.charAt(j);
                if (d == buf.charAt(i)) return i;
            }
        }
        return 0;
    }

    /** 
     * Reads in a chunk of the file into a string.  
     */
    private static String readFileAsString(BufferedReader reader, int size)
        throws java.io.IOException 
    {
        StringBuffer fileData = new StringBuffer(size);
        int numRead=0;

        while(size > 0) {
            int bufsz = 1024 > size ? size : 1024;
            char[] buf = new char[bufsz];
            numRead = reader.read(buf,0,bufsz);
            if (numRead == -1)
                break;
            String readData = String.valueOf(buf, 0, numRead);
            fileData.append(readData);
            size -= numRead;
        }
        return fileData.toString();
    }

    /**
     * Updates the count for each number of words.  
     * Assumes given map is synchronized.
     */
    private void updateCount(String q) {
        AtomicInteger cnt = counts.get(q);
        if (cnt == null) {
            // attempt to put 1 in the table.  If the old
            // value was null, then we are OK.  If not, then
            // some other thread put a value into the table
            // instead, so we fall through
            cnt = counts.put(q, new AtomicInteger(1));
            if (cnt == null) return;
        }
        cnt.incrementAndGet();
    } 

    /**
     * Main task : tokenizes the given buffer and counts words. 
     */
    public void run() {
        StringTokenizer st = new StringTokenizer(buffer,DELIMS);
        while (st.hasMoreTokens()) {
            String token = st.nextToken();
            //System.out.println("updating count for "+token);
            updateCount(token);
        }
    } 

    public static void main(String args[]) throws java.io.IOException {
        long startTime = System.currentTimeMillis();
        if (args.length < 1) {
            System.out.println("Usage: <file to count> [#threads] [chunksize]");
            System.exit(1);
        }
        int numThreads = 4;
        int chunksize = 1000;
        if (args.length >= 2)
            numThreads = Integer.valueOf(args[1]);
        if (args.length >= 3)
            chunksize = Integer.valueOf(args[2]);
        ExecutorService pool = Executors.newFixedThreadPool(numThreads);
        BufferedReader reader = new BufferedReader(new FileReader(args[0]));
        ConcurrentMap<String,AtomicInteger> m = 
            new ConcurrentHashMap<String,AtomicInteger>();
        String leftover = ""; // in case a string broken in half
        while (true) {
            String res = readFileAsString(reader,chunksize);
            if (res.equals("")) {
                if (!leftover.equals("")) 
                    new WordCountParallelAtomicInt(leftover,m).run();
                break;
            }
            int idx = findDelim(res);
            String taskstr = leftover + res.substring(0,idx);
            leftover = res.substring(idx,res.length());
            pool.submit(new WordCountParallelAtomicInt(taskstr,m));
        }
        pool.shutdown();
        try {
            pool.awaitTermination(1,TimeUnit.DAYS);
        } catch (InterruptedException e) {
            System.out.println("Pool interrupted!");
            System.exit(1);
        }
        long endTime = System.currentTimeMillis();
        long elapsed = endTime - startTime;
        int total = 0;
        for (Map.Entry<String,AtomicInteger> entry : m.entrySet()) {
            int count = entry.getValue().get();
            if (printAll)
                System.out.format("%-30s %d\n",entry.getKey(),count);
            total += count;
        }
        System.out.println("Total words = "+total);
        System.out.println("Total time = "+elapsed+" ms");
    }
}