package com.hammurapi.common.concurrent;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import com.hammurapi.common.HammurapiException;

/**
 * In-JMV tracking executor service.
 * @author Pavel Vlasov
 *
 */
public class LocalTrackingExecutorService extends AbstractExecutorService implements TrackingExecutorService {
        
        private boolean oneOff;
        
        private AtomicLong joinDone = new AtomicLong(-1);
        
        private final AtomicLong TASK_COUNTER = new AtomicLong();

        private String name;
        
//      private final ThreadLocal<Tracking> contextTask = new ThreadLocal<Tracking>(); 
                
        /**
         * @param master Master executor.
         * @param oneOff If true, join can be invoked only once. No tasks can be submitted to the executor once
         * join() returns.
         */
        public LocalTrackingExecutorService(ExecutorService master, boolean oneOff, String name) {
                this.master = master;
                this.oneOff = oneOff;
                this.name = name;
        }

        @Override
        public void shutdown() {
                master.shutdown();
        }

        @Override
        public List<Runnable> shutdownNow() {
                return master.shutdownNow();
        }

        @Override
        public boolean isShutdown() {
                return master.isShutdown();
        }

        @Override
        public boolean isTerminated() {
                return master.isTerminated();
        }

        @Override
        public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
                return master.awaitTermination(timeout, unit);
        }

        @Override
        public void execute(Runnable command) {
                checkJoinDone();                
                if (command instanceof DelegatingRunnableFuture) {
                        DelegatingRunnableFuture<Object> drf = (DelegatingRunnableFuture<Object>) command;
                        if (drf.getCommand()!=null) {
                                drf.setTarget(master.submit(new TrackingRunnable(drf.getCommand()), drf.getValue()));
                        } else {
                                drf.setTarget(master.submit((Callable) new TrackingCallable(drf.getTask())));
                        }
                } else {
                        master.execute(new TrackingRunnable(command));
                }
        }
        
        @Override
        protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
                return new DelegatingRunnableFuture<T>(callable);
        }
        
        protected <T extends Object> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
                return new DelegatingRunnableFuture<T>(runnable, value);          
        };

        private void checkJoinDone() {
                if (oneOff && joinDone.get()!=-1) {
                        throw new IllegalStateException("Can't execute any more tasks");
                }
        }
        
//      private Lock serviceLock = new ReentrantLock();
        
        private Thread joinThread;

        /**
         * Executes and removes pending tasks, exits when all tasks are completed.
         */
        @Override
        public void join() throws InterruptedException {
                join(0);
        }
        
        public boolean join(long timeout) throws InterruptedException {
                long start = timeout>0 ? System.currentTimeMillis() : 0;
                checkJoinDone();                
                joinThread = Thread.currentThread();
                while (true) {
                        Runnable toRun = null;
                        boolean hasRunning = false;
                        synchronized (tasks) {
                                for (Tracking task: tasks) {
                                        if (task.state==TaskState.PENDING) {
                                                toRun = task;
                                                break;
                                        } else if (task.state==TaskState.RUNNING && task.runner!=Thread.currentThread()) {
                                                hasRunning = true;
                                        }
                                }
                                if (toRun==null) {
                                        if (!hasRunning) {
                                                if (oneOff) {
                                                        joinDone.set(TASK_COUNTER.get());
                                                }
                                                return true;
                                        }
                                        tasks.wait(10); // Wakes up if all tasks finish execution.
                                        if (start>0) {
                                                if (System.currentTimeMillis()-start>timeout) {
                                                        return false;
                                                }
                                        }
                                }
                        }
                        
                        if (toRun!=null) {
                                toRun.run();                    
                        }
                }
        }
        
        private ExecutorService master; 
        
//      private Collection<Thread> runners = new LinkedList<Thread>();
        
        // Tasks which are not yet being executed.
        private LinkedList<Tracking> tasks = new LinkedList<Tracking>();
        
        private enum TaskState {
                PENDING,
                RUNNING,
                DONE
        }
        
        private abstract class Tracking implements Runnable {
                volatile TaskState state;
                volatile Thread runner;                         
        }
        
        private class TrackingRunnable extends Tracking {
                
                private Runnable master;
//              private final long taskID = TASK_COUNTER.incrementAndGet(); 
//              private Tracking submitter = contextTask.get();
                
                TrackingRunnable(Runnable master) {                     
                        this.master = master;
                        synchronized (tasks) {
                                state = TaskState.PENDING;
                                tasks.add(this);
                        }
                }
                
                @Override
                public void run() {
                        synchronized (tasks) {
                                if (state!=TaskState.PENDING) {
                                        // Already running.
                                        return;
                                }
                                runner = Thread.currentThread();
                                state = TaskState.RUNNING;
//                              contextTask.set(this);
                        }
                        try {
                                master.run();
                        } finally {
                                synchronized (tasks) {
                                        runner = null;
                                        state = TaskState.DONE;
//                                      contextTask.set(null);
                                        tasks.remove(this);
                                        boolean hasIncompleteTasks = false;
                                        for (Tracking task: tasks) {
                                                if (task.state!=TaskState.DONE) {
                                                        hasIncompleteTasks = true;
                                                        break;
                                                }
                                        }
                                        if (!hasIncompleteTasks) {
                                                tasks.notifyAll();
                                        }
                                }
                        }
                }
        }

        private class TrackingCallable<V> extends Tracking implements Callable<V> {
                
                volatile TaskState state;
                volatile Thread runner;
                
                private Callable<V> master;
//              private final long taskID = TASK_COUNTER.incrementAndGet(); 
//              private TrackingRunnable submitter = contextTask.get();
                
                TrackingCallable(Callable<V> master) {                    
                        this.master = master;
                        synchronized (tasks) {
                                state = TaskState.PENDING;
                                tasks.add(this);
                        }
                }
                
                private Object[] result = {null};
                
                @Override
                public V call() throws Exception {
                        synchronized (tasks) {
                                if (state==TaskState.DONE) {
                                        return (V) result[0];
                                }
                                if (state==TaskState.RUNNING) {
                                        if (runner == Thread.currentThread()) {
                                                throw new IllegalStateException();
                                        }
                                        synchronized (result) {
                                                result.wait();
                                                return (V) result[0];
                                        }
                                }
                                runner = Thread.currentThread();
                                state = TaskState.RUNNING;
//                              contextTask.set(this);
                        }
                        try {
                                result[0] = master.call();
                                synchronized (result) {
                                        result.notifyAll();
                                }
                                return (V) result[0];
                        } finally {
                                synchronized (tasks) {
                                        runner = null;
                                        state = TaskState.DONE;
//                                      contextTask.set(null);
                                        tasks.remove(this);
                                        boolean hasIncompleteTasks = false;
                                        for (Tracking task: tasks) {
                                                if (task.state!=TaskState.DONE) {
                                                        hasIncompleteTasks = true;
                                                        break;
                                                }
                                        }
                                        if (!hasIncompleteTasks) {
                                                tasks.notifyAll();
                                        }
                                }
                        }
                }

                @Override
                public void run() {
                        try {
                                call();
                        } catch (Exception e) {
                                throw new HammurapiException(e); 
                        }                       
                }
        }
        
}