package accrue.analysis.pointer;

import java.util.Collections;
import java.util.ConcurrentModificationException;
import java.util.LinkedHashSet;
import java.util.Set;
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.AtomicLong;

import polyglot.frontend.ExtensionInfo;
import polyglot.main.Report;
import polyglot.types.ProcedureInstance;
import polyglot.util.ErrorInfo;
import accrue.analysis.AnalysisTopics;
import accrue.analysis.pointer.StmtRegistrar.CodeInfo;
import accrue.analysis.pointer.analyses.HeapAbstractionFactory;

public class PointsToEngineMultiThread extends PointsToEngine {

    /**
     * When a node is changed, should we process the dependent statements all at once?
     */
    private static final boolean CHUNK_DEPENDENCY_SETS = true;

    public PointsToEngineMultiThread(HeapAbstractionFactory haf) {
        super(haf);
    }

    /**
     * The dependencies map records which StmtAndContexts (i.e., pointer statements 
     * in a context) depend on the points-to set of a given graph node. More precisely,
     * if we have StmtAndContext sac such that sac is in the set dependencies.get(n),
     * then if the points-to set of n changes (i.e., this.pointsTo(n) increases), then
     * we need to re-process sac.
     * 
     *  The correctness of multithreaded protocol relies the dependency sets being synchronized.
     *  See the code for getOrCreateDependencySet.
     */
    private ConcurrentHashMap<PointsToGraphNode, Set<StmtAndContext>> dependencies =
            new ConcurrentHashMap<PointsToGraphNode, Set<StmtAndContext>>(16,
                                                                          0.75f,
                                                                          numThreads());

    private PointsToGraph g;
    private StmtRegistrar registrar;

    private final ExecutorServiceCounter execService =
            new ExecutorServiceCounter(Executors.newFixedThreadPool(this.numThreads()));

//            new ExecutorServiceCounter(new ForkJoinPool(this.numThreads()));

    int numThreads() {
        return Runtime.getRuntime().availableProcessors();
    }

    @Override
    protected PointsToGraph solveImpl(StmtRegistrar registrar,
            ExtensionInfo extInfo) {
        return solveMultithreaded(registrar, extInfo);

        // The following code performs a single threaded solve, and a multi-threaded solve,
        // and makes sure they agree.
//        long startA = System.currentTimeMillis();
//        System.out.println("starting single threaded pointer solve...");
//        singleThreadG =
//                new PointsToEngineSingleThread(heapAbstractionFactory()).solve(registrar,
//                                                                               extInfo);
//        singleThreadG.setSolved();
//        long startB = System.currentTimeMillis();
//        System.out.println("starting multithreaded pointer solve...");
//        PointsToGraph g = solveMultithreaded(registrar, extInfo);
//        g.setSolved();
//        this.dependencies = null; // TEMP: clear this out early to save space
//        long startC = System.currentTimeMillis();
//        System.out.println("  sing threaded time " + (startC - startB) + "ms");
//        System.out.println("    domain size is "
//                + singleThreadG.getDomain().size());
//        System.out.println("    total size is " + singleThreadG.totalSize());
//        System.out.println("  multithreaded time " + (startB - startA) + "ms");
//        System.out.println("    domain size is " + g.getDomain().size());
//        System.out.println("    total size is " + g.totalSize());
//        if (g.equals(singleThreadG)) {
//            System.out.println("EQUIVALENT GRAPHS");
//        }
//        else {
//
//            System.out.println("FAILBLOG TOTALLY");
//            throw new Error("Multithreaded pointer analysis disagrees with single threaded");
//        }
//        return g;
    }

    public PointsToGraph solveMultithreaded(StmtRegistrar registrar,
            ExtensionInfo extInfo) {
        this.registrar = registrar;
        this.g = new PointsToGraphSync(registrar, this);
        if (Report.should_report(AnalysisTopics.POINTSTO, 1)) {
            Report.report(1, "Starting PointsTo Engine");
        }

        // Add all the initial stmts and their contexts to the list.
        for (PointsToStmt s : registrar.stmts()) {
            for (CContext context : g.getContexts(s.codeInstance)) {
                execService.processStmtAndContext(new StmtAndContext(s, context));
            }
        }

        while (execService.containsPending()) {
            this.checkTimeout();
            long timeout = 100000L; // default 100 seconds
            if (this.timeoutTime() != null) {
                timeout =
                        Math.abs(this.timeoutTime()
                                - System.currentTimeMillis());
            }
            execService.waitUntilAllFinished(timeout);
        }
        // all the tasks are done.
        // Shut down the executer service
        execService.shutdownAndAwaitTermination();

        // for safety, do a pass through again...
//        checkReachedFixedpoint(registrar, g);

        long end = System.currentTimeMillis();

        if (Report.should_report(AnalysisTopics.POINTSTO, 1)) {
            Report.report(1, "Processed statements (" + (end - startTime())
                    + " millisecs)");
            Report.report(1, "  number of nodes: " + g.nodeSize()
                    + "   numer of edges : " + g.edgeSize());
        }

        if (Report.should_report(AnalysisTopics.POINTSTOGRAPH, 2)) {
            g.dumpPointsToGraph(2);
        }
        else if (Report.should_report(AnalysisTopics.POINTSTOGRAPH, 1)) {
            g.dumpPointsToGraph(1);
        }

        // record warnings about missing code.
        for (ErrorInfo ei : registrar.missingCode()) {
            extInfo.compiler().errorQueue().enqueue(ei);
        }

        return g;
    }

    /**
     * 
     * @param stmt
     * @param context 
     * @param codeInstance 
     * @param r
     */
    @Override
    void recordRead(StmtAndContext sac, PointsToGraphNode r) {
        getOrCreateDependencySet(r).add(sac);
    }

    @Override
    void newContext(ProcedureInstance callee, CContext calleeContext) {
        CodeInfo codeInfo = registrar.getCodeInfo(callee);
        if (codeInfo != null) {
            for (PointsToStmt s : registrar.stmtsForCode(callee)) {
                execService.processStmtAndContext(new StmtAndContext(s,
                                                                     calleeContext));
            }
        }
    }

    @Override
    void changedNode(PointsToGraphNode n) {
        Set<StmtAndContext> d = getOrCreateDependencySet(n);
        if (CHUNK_DEPENDENCY_SETS) {
            if (d.isEmpty()) {
                // nothing to do!
                return;
            }

            // we create a (non-synchronized) copy of d to use for the
            // chunked task. This is because things added later to d will
            // have already used the updated pointstoset of n, so no need to 
            // rerun it.
            Set<StmtAndContext> s = new LinkedHashSet<StmtAndContext>();

            // synchronize on d before iterating on it: it is a synchronized set.
            synchronized (d) {
                for (StmtAndContext sac : d) {
                    s.add(sac);
                }
            }
            execService.processStmtAndContextSet(s);
        }
        else {
            // synchronize on the set before we iterate over it.
            synchronized (d) {
                for (StmtAndContext sac : d) {
                    execService.processStmtAndContext(sac);
                }
            }
        }
    }

    private Set<StmtAndContext> getOrCreateDependencySet(PointsToGraphNode n) {
        // Double-Checked Lock pattern, works with ConcurrentHashMap in Java 5.0 and later
        Set<StmtAndContext> set = this.dependencies.get(n);
        if (set == null) {
            // Note that we use a synchronized set for the dependency sets.
            // This is required so that we get a happens-before relation between
            // all reads and writes to a dependency set. This ensures the correctness
            // of the protocol: If the points-to set of node n changes, then
            // any StmtAndContext that may have read an old version of the points-to
            // set will be evaluated again.
            Set<StmtAndContext> newSet =
                    Collections.synchronizedSet(new LinkedHashSet<StmtAndContext>());

            // It's possible that another thread created the Set for the current callee
            set = dependencies.putIfAbsent(n, newSet);
            if (set == null) {
                // put succeeded, use new value
                set = newSet;
            }
        }
        return set;
    }

    class ExecutorServiceCounter {
        private ExecutorService exec;
        private AtomicLong numTasks;
        private Set<StmtAndContext> enqueuedSacs =
                Collections.newSetFromMap(new ConcurrentHashMap<StmtAndContext, Boolean>(16,
                                                                                         0.75f,
                                                                                         numThreads()));

        public ExecutorServiceCounter(ExecutorService exec) {
            this.exec = exec;
            this.numTasks = new AtomicLong(0);
        }

        public void shutdownAndAwaitTermination() {
            exec.shutdown();
            boolean finished = false;
            // keep waiting until we successfully finish all the outstanding tasks
            do {
                try {
                    finished =
                            exec.awaitTermination(Long.MAX_VALUE,
                                                  TimeUnit.HOURS);
                }
                catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            } while (!finished);

        }

        public void processStmtAndContext(StmtAndContext sac) {
            boolean added = enqueuedSacs.add(sac);
            if (added) {
                this.numTasks.incrementAndGet();
                exec.execute(new RunnableStmtAndContext(sac));

            }
            else {
                // was already in the queue.
            }
        }

        public void processStmtAndContextSet(Set<StmtAndContext> sacs) {
            // always add to the queue, don't bother checking if something is there already
            this.numTasks.incrementAndGet();
            exec.execute(new RunnableStmtAndContextSet(sacs));
        }

        public void finishedTask() {
            if (this.numTasks.decrementAndGet() == 0) {
                // Notify anyone that was waiting.
                synchronized (this) {
                    this.notifyAll();
                }
            }
        }

        public boolean containsPending() {
            return numTasks.get() > 0;
        }

        public long numRemainingTasks() {
            return numTasks.get();
        }

        public void waitUntilAllFinished(long timeOutMillis) {
            if (this.containsPending()) {
                synchronized (this) {
                    try {
                        // wait until we are meant to time out.
                        this.wait(timeOutMillis);
                    }
                    catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        public class RunnableStmtAndContext implements Runnable {
            private final StmtAndContext sac;

            public RunnableStmtAndContext(StmtAndContext stmtAndContext) {
                this.sac = stmtAndContext;
            }

            @Override
            public void run() {
                try {
                    // remove the sac from the set of sacs that are waiting to be executed.
                    enqueuedSacs.remove(sac);
                    sac.stmt.process(sac.context,
                                     g,
                                     registrar,
                                     PointsToEngineMultiThread.this.heapAbstractionFactory(),
                                     sac);
                    execService.finishedTask();
                    if (Report.should_report(AnalysisTopics.POINTSTO, 3))
                        Report.report(3, "number of remaining tasks: "
                                + execService.numRemainingTasks()
                                + "\n just processed " + sac.stmt.toString());
                }
                catch (ConcurrentModificationException e) {
                    System.err.println("ConcurrentModificationException!!!");
                    e.printStackTrace();
                    System.exit(0);
                    // No seriously DIE!
                    Runtime.getRuntime().halt(0);
                }
            }
        }

        public class RunnableStmtAndContextSet implements Runnable {
            private final Set<StmtAndContext> sacs;

            public RunnableStmtAndContextSet(Set<StmtAndContext> stmtAndContexts) {
                this.sacs = stmtAndContexts;
            }

            @Override
            public void run() {
                try {
                    for (StmtAndContext sac : sacs) {
                        sac.stmt.process(sac.context,
                                         g,
                                         registrar,
                                         PointsToEngineMultiThread.this.heapAbstractionFactory(),
                                         sac);
                    }
                    execService.finishedTask();
                    if (Report.should_report(AnalysisTopics.POINTSTO, 3))
                        Report.report(3, "number of remaining tasks: "
                                + execService.numRemainingTasks()
                                + "\n just processed " + sacs.toString());

                }
                catch (ConcurrentModificationException e) {
                    System.err.println("ConcurrentModificationException!!!");
                    e.printStackTrace();
                    System.exit(0);
                    // No seriously DIE!
                    Runtime.getRuntime().halt(0);
                }
            }
        }
    }

}
