/* ************************************************************************
#
#  DivConq
#
#  http://divconq.com/
#
#  Copyright:
#    Copyright 2012 eTimeline, LLC. All rights reserved.
#
#  License:
#    See the license.txt file in the project's top-level directory for details.
#
#  Authors:
#    * Andy White
#
************************************************************************ */
package divconq.work;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicLong;

import divconq.lang.Hub;
import divconq.lang.TaskContext;
import divconq.log.Logger;
import divconq.xml.XElement;

public class WorkPool {
	protected LinkedBlockingQueue<ITaskInfo> queue = new LinkedBlockingQueue<ITaskInfo>();
	protected ConcurrentHashMap<Integer, Worker> pool = new ConcurrentHashMap<Integer, Worker>();
	
	protected int size = 16;
	
	protected AtomicLong totalThreadsCreated = new AtomicLong();
	protected AtomicLong totalThreadsHung = new AtomicLong();		// based on timeout
	
	protected AtomicLong totalTasksSubmitted = new AtomicLong();
	protected AtomicLong totalTasksStarted = new AtomicLong();
	protected AtomicLong totalTasksCompleted = new AtomicLong();
	
	protected boolean shutdown = false;
	protected ScheduledFuture<?> clock = null;
	protected long scheduleFreq = 150;
	
	public void init(XElement config) {
		// TODO load settings
	}
	
	public void init(int threads) {
		this.size = threads;
	}
	
	public int howBusy() {
		return this.activeThreadCount() * 100 / this.size; 
	}
	
	public int activeThreadCount() {
		return (int) (this.totalTasksStarted.longValue() - this.totalTasksCompleted.longValue()); 
	}
	
	public int threadCount() {
		return this.size; 
	}
	
	public long threadsCreated() {
		return this.totalThreadsCreated.get();
	}
	
	public void incThreadsCreated() {
		this.totalThreadsCreated.intValue();
	}
	
	public long threadsHung() {
		return this.totalThreadsHung.get();
	}
	
	public void incThreadsHung() {
		this.totalThreadsHung.intValue();
	}
	
	public long tasksSubmitted() {
		return this.totalTasksSubmitted.get();
	}
	
	public void incTasksSubmitted() {
		this.totalTasksSubmitted.intValue();
	}
	
	public long tasksStarted() {
		return this.totalTasksStarted.get();
	}
	
	public void incTasksStarted() {
		this.totalTasksStarted.intValue();
	}
	
	public long tasksCompleted() {
		return this.totalTasksCompleted.get();
	}
	
	public void incTasksComlpeted() {
		this.totalTasksCompleted.intValue();
	}
	
	public ITaskInfo submit(IRunnable task) {
		return this.submit(new TaskInfo(task));
	}
	
	public ITaskInfo submit(ITaskInfo task) {
		if (this.shutdown)
			return null;
		
		if (task.context() == null) {
			TaskContext.getHubContext().warn("Work missing context: " + task);
			return null;
		}
		
		this.queue.add(task);
		
		this.totalTasksSubmitted.incrementAndGet();
		
		return task;
	}
	
	public void start() {
		for (int i = 0; i < this.size; i++) 
			this.initSlot(i);
		
		this.clock = Hub.instance.getClock().schedulePeriodicInternal(new Runnable() {
			@Override
			public void run() {
				// after 15 minutes TODO config
				long stamp = System.currentTimeMillis() - (15 * 60 * 1000);
				
				for (Worker w : WorkPool.this.pool.values()) 
					w.stopTimeout(stamp);
			}
		}, this.scheduleFreq);
	}
	
	protected void initSlot(int slot) {
		if (!this.shutdown) {
			Worker work = new Worker();
			this.pool.put(slot, work);
			work.start(this, slot);
		}
		else
			this.pool.remove(slot);
		
		TaskContext tc = TaskContext.getHubContext();
		tc.trace("Thread Pool slot " + slot + " changed, now have " + this.pool.size() + " threads");
	}
	
	// TODO convert to return OpResult, change the Logger call below to go to OR
	public void stop() {
		Logger.trace("Work Pool Stopping");
		
		this.shutdown = true;
		
		this.clock.cancel(false);
		
		Logger.trace("Work Pool Stopping Nice");
		
		for (Worker w : this.pool.values()) 
			w.stopNice();
		
		Logger.trace("Work Pool Waiting");
				
		// wait a minute for things to finish up.
		for (int i = 0; i < 60; i++) {
			if (this.pool.size() == 0)
				break;
			
			try {
				Thread.sleep(1);
			}
			catch (Exception x) {				
			}
		}
		
		Logger.trace("Work Pool Size: " + this.pool.size());
		
		Logger.trace("Work Pool Interrupt Remaining Workers");
		
		for (Worker w : this.pool.values()) 
			w.stop();
		
		Logger.trace("Work Pool Stopped");
	}
	
	public ITaskInfo take() throws InterruptedException {
		return this.queue.take();
	}
}
