/* ************************************************************************
#
#  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.locks.ReentrantLock;

import divconq.lang.TaskContext;

public class Worker implements Runnable {
	protected Thread thread = null;
	protected boolean busy = false;
	protected boolean stopping = false;
	protected long startStamp = 0;
	protected ReentrantLock lock = new ReentrantLock();
	protected int slot = 0;
	protected int jobs = 0;
	protected WorkPool pool = null;
	
	public void start(WorkPool v, int slot) {
		this.pool = v;
		this.slot = slot;
		this.thread = new Thread(this, "Worker_" + this.slot);		// TODO name
		this.thread.start();
	}
	
	@Override
	public void run() {
		TaskContext tc = TaskContext.getHubContext();
		tc.trace("Work pool thread started: " + this.slot);		// TODO name
		this.pool.incThreadsCreated();
		
		while (!this.stopping) {
			try {
				tc.trace("Work pool thread taking: " + this.slot);
				
				ITaskInfo r = this.pool.take();
				
				this.lock.lockInterruptibly();
				
				this.busy = true;
				this.startStamp = System.currentTimeMillis();
				
				this.lock.unlock();
				
				if (!this.stopping) {
					this.pool.incTasksStarted();
					
					tc.trace("Work pool thread running: " + this.slot);
					
					r.setThreadSlot(this.slot);
					r.run();
					this.pool.incTasksComlpeted();
				}
				
				if (Thread.currentThread().isInterrupted())
					break;
			}
			catch (InterruptedException x) {
				break;
			}
			catch (Exception x) {
				tc.warn("Work pool caught error: " + x);
			}
			catch (ThreadDeath x) {
				tc.warn("Work pool thread died: " + x);
				throw x;
			}
			
			this.busy = false;
			this.startStamp = 0;
		}
		
		tc.trace("Work pool thread stopped: " + this.slot);
	}

	public void stopTimeout(long timestamp) {
		if (!this.thread.isAlive()) 
			this.stopNice();
		
		if (this.busy && (this.startStamp < timestamp)) {
			TaskContext tc = TaskContext.getHubContext();
			tc.warn("Work pool thread hung: " + this.slot);
			this.pool.incThreadsHung();
			this.stop();
		}
	}
	
	public void stop() {
		this.stopping = true;
		
		try {
			if (this.thread.isAlive())
				this.thread.interrupt();
		}
		catch (Exception x) {				
		}
		
		this.pool.initSlot(this.slot);
	}
	
	public void stopNice() {
		this.stopping = true;
		
		try {
			this.lock.lockInterruptibly();
			
			// only if not busy
			try {
				if (!this.busy && this.thread.isAlive())
					this.thread.interrupt();
			}
			catch (Exception x) {				
			}
			finally {
				this.lock.unlock();
			}
		} 
		catch (InterruptedException x) {
		}
		
		this.pool.initSlot(this.slot);
	}
}