package jp.hasc.hasctool.core.runtime;

import java.util.ArrayList;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import jp.hasc.hasctool.core.data.SignalMessage;
import jp.hasc.hasctool.core.messaging.MessageQueue;

/**
 * Task（スレッド）を管理します。ExecutorServiceのラッパです
 * @author iwasaki
 */
class TaskManager {
	/** logger for this class */
	private final static org.apache.commons.logging.Log LOG = org.apache.commons.logging.LogFactory
			.getLog(TaskManager.class);
	
	private RuntimeContext context_;
	private ExecutorService executorService_ = null;
	private ArrayList<AbstractTask> tasksToStart_ = new ArrayList<AbstractTask>();
	private ConcurrentLinkedQueue<Future<?>> runningTasks_ = new ConcurrentLinkedQueue<Future<?>>();
	//private ArrayList<FutureTask<?>> runningFutures_ = new ArrayList<FutureTask<?>>();
	public ExecutorService getExecutorService() { return executorService_; }
	
	public static final long SHUTDOWN_TIMEOUT_SECS = 5;
	
	public TaskManager(RuntimeContext context) {
		super();
		context_ = context;
		RuntimeContext parentContext = context.getParentContext();
		if (parentContext!=null) {
			// parentContextと、ExecutorServiceを共用する
			executorService_=parentContext.getTaskManager().getExecutorService();
		}else{
			executorService_=Executors.newCachedThreadPool();
		}
	}
	
	public void shutdownPhase1() {
		synchronized (this) {
			if (!started_) return;
		}
		LOG.debug("shotdown...");
		
		// clear all queues
		try{
			ArrayList<MessageQueue> queues = context_.getMessageQueues();
			synchronized (queues) { 
				for(MessageQueue queue:queues) {
					BlockingQueue<Object> bq = queue.getQueueImpl();
					if (!bq.isEmpty()) {
						bq.clear();
					}
					if (bq.remainingCapacity()>0) {
						bq.add(SignalMessage.END); // 強制的にEND送信
					}
				}
				queues.clear();
			}
		}catch(Throwable ex) {
			//ex.printStackTrace();
			LOG.warn("Throwable",ex);
		}
		
		if (!executorService_.isShutdown())
			executorService_.shutdown(); // Disable new tasks from being submitted
	}
	
	public void shutdownPhase2() {
		synchronized (this) {
			if (!started_) return;
		}
		// ExecutorService の JDK document よりコピー
		ExecutorService pool = executorService_;
		try {
			// Wait a while for existing tasks to terminate
			if (!pool.awaitTermination(SHUTDOWN_TIMEOUT_SECS, TimeUnit.SECONDS)) {
				pool.shutdownNow(); // Cancel currently executing tasks
				// Wait a while for tasks to respond to being cancelled
				if (!pool.awaitTermination(SHUTDOWN_TIMEOUT_SECS, TimeUnit.SECONDS))
						LOG.warn("Pool did not terminate");
			}
		} catch (InterruptedException ie) {
			// (Re-)Cancel if current thread also interrupted
			pool.shutdownNow();
			// Preserve interrupt status
			Thread.currentThread().interrupt();
		}
		synchronized (this) {
			started_=false;
			this.notifyAll();
		}
		
		//
		LOG.debug("terminated");
	}
	
	/*
	public void dispose() {
		// NOP
	}
	*/

	public void register(AbstractTask task) {
		synchronized (this) { 
			if (!started_) {
				// startAll()を呼んだ時に実行
				tasksToStart_.add(task);
			}
		}
	}
	
	private boolean started_=false;
	
	/**
	 * startAll()実行後に、個別に新規Taskを実行します。実行前なら何もしません。
	 */
	public void start(AbstractTask task) {
		synchronized (this) {
			if (!started_) {
				return;
				//throw new IllegalStateException();
			}
			runningTasks_.add(executorService_.submit(task));
		}
	}
	
	/**
	 * registerで登録したTaskを、全て実行開始します
	 */
	public void startAll() {
		LOG.debug("startAll");
		
		synchronized (this) {
			if (started_) throw new IllegalStateException();
			started_=true;
			runningTasks_.clear();
			for(AbstractTask task:tasksToStart_) {
				Future<Object> f = executorService_.submit(task);
				//LOG.debug("submit "+f+" with "+task);
				runningTasks_.add(f);
			}
			tasksToStart_.clear();
		}
		
		// thread to  wait termination
		executorService_.execute(new Runnable() {
			@Override
			public void run() {
				try {
					while(!runningTasks_.isEmpty()) {
						try {
							Future<?> f = runningTasks_.poll();
							if (f!=null) {
								//LOG.debug("get "+f);
								f.get();
							}
						} catch (ExecutionException ex) {
							//ex.printStackTrace();
							LOG.warn("ExecutionException",ex);
						}
					}
				} catch (InterruptedException ex) {
					//ex.printStackTrace();
					LOG.warn("InterruptedException",ex);
				}finally{
					if (!executorService_.isShutdown()) {
						LOG.debug("done");
					}else{
						LOG.debug("done (shutdowned)");
					}
					synchronized (TaskManager.this) {
						started_=false;
						TaskManager.this.notifyAll();
					}
					context_.onTasksTerminated();
				}
			}
		});
		/*
		synchronized (this) {
			started_=true;
			runningFutures_.clear();
			for(AbstractTask task : tasksToStart_) {
				FutureTask<Object> ft = new FutureTask<Object>(task);
				runningFutures_.add(ft);
				executorService_.execute(ft);
			}
			tasksToStart_.clear();
		}
		*/
	}
	
	/**
	 * 全てのタスクが終了するまでブロックします
	 * @throws InterruptedException 
	 */
	//@SuppressWarnings("unchecked")
	public boolean awaitTermination(long timeout) throws InterruptedException {
		synchronized (this) {
			long endAt=(timeout>0)?(System.currentTimeMillis()+timeout):0;
			while(started_) {
				if (timeout>0) {
					long dt=endAt-System.currentTimeMillis();
					if (dt<=0) return false;
					this.wait(dt);
				}else{
					this.wait();
				}
			}
		}
		return true;
	}
	
	public boolean isRunning() {
		return started_;
	}

	public void runDirect(AbstractTask task) throws InterruptedException {
		tasksToStart_.remove(task);
		//
		task.run();
	}

}
