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); 
			}			
		}
	}
	
}
