package pulling.executing;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.text.SimpleDateFormat;
import java.util.GregorianCalendar;
import java.util.Hashtable;

/**
 * The Built in server executer which is responsible for executing
 * the job when it is received by the server
 * 
 * @author Ben Kybartas and Majd Kokaly
 *
 */
public class ServerSideExecuter_BuiltIn extends ServerSideExecuter implements Runnable {

	private static final long serialVersionUID = 1L;

	private boolean alive = true;
	private Thread thisThread;
	private Hashtable<Long, JobInfo> executerJobTable;
	private int serverPort = 37935; // Port I listen on
	private String hostName;
	private String password = "";
	private int serverID;
	private String mapperHostName;
	private int completionServerPort = 37931;
	private double availability = 1;
	private int percent;

	private double dateSubmitted = 0;
	private String dateStarted = null;
	private String dateStopped = null;
	private static long ExecutionLayerIDCounter = 1; // job id on the
														// execution layer
	
	private double IdlePercent = 1;
	
	static class JobInfo {
		long ExecutionLayerJobID;
		long mapperJobID;
		GregorianCalendar dateStarted;
		GregorianCalendar dateStopped;
		String Result;

		public JobInfo(long ExecutionLayerJobID, long mapperJobID) {
			this.ExecutionLayerJobID = ExecutionLayerJobID;
			this.mapperJobID = mapperJobID;
		}

	}

	public ServerSideExecuter_BuiltIn(String mapperHostName) {
		executerJobTable = new Hashtable<Long, JobInfo>(2000);
		this.setMapperHostName(mapperHostName);
		try {
			this.setHostName(InetAddress.getLocalHost().getHostName());
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}
	}

	public synchronized boolean isAlive() {
		return alive;
	}

	public synchronized void setAlive(boolean alive) {
		this.alive = alive;
	}

	public int getServerPort() {
		return serverPort;
	}

	public void setServerPort(int serverPort) {
		this.serverPort = serverPort;
	}

	public String getHostName() {
		return hostName;
	}

	public void setHostName(String hostname) {
		this.hostName = hostname;
	}

	public String getPassword() {
		return password;
	}

	public void setPassword(String password) {
		this.password = password;
	}

	public int getServerID() {
		return serverID;
	}

	public void setServerID(int serverID) {
		this.serverID = serverID;
	}
	
	public double getAvailability()
	{
		return this.availability;
	}
	
	public void setAvailability(double avail)
	{
		this.availability = avail;
	}
	
	public String getMapperHostName() {
		return mapperHostName;
	}

	public void setMapperHostName(String mapperHostName) {
		this.mapperHostName = mapperHostName;
	}

	public int getCompletionServerPort() {
		return completionServerPort;
	}

	public void setCompletionServerPort(int completionServerPort) {
		this.completionServerPort = completionServerPort;
	}
	
	public void setIdlePercent (double Percent)
	{
		this.IdlePercent = Percent;
	}
	
	public double getIdlePercent()
	{
		return IdlePercent;
	}
	
	public void startExecuter() {
		thisThread = new Thread(this);
		thisThread.start();
		new Thread(new CPUUsageUpdater_BuiltIn(this)).start();
	}

	public void stopExecuter() {
		boolean wasAlive = this.isAlive();
		this.setAlive(false);
		if (wasAlive) {
			thisThread.interrupt();
			thisThread = null;
			if (serverSocket != null)
				try {
					serverSocket.close();
				} catch (IOException e) {
					e.printStackTrace();
				}	
			if (socket != null)
				try {
					socket.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
		}
	}

	public long deleteJob(long id, String hostname, String password) {
		return 0;
	}




	public String getDateStarted(long id, String hostname, String password) {
		return dateStarted;
	}

	public String getDateStopped(long id, String hostname, String password) {
		return dateStopped;
	}

	public double getDateSubmitted(long id, String hostname, String password) {
		return dateSubmitted;
	}
	
	@Override
	public String getJobStatus(long id, String hostname, String password) {
		return "The current percent completed is : " + percent;
	}

	@Override
	public double getPercentageDone(long id, String hostname, String password) {
		return percent;
	}

	@Override
	public long stopJob(long id, String hostname, String password) {
		return 0;
	}

	@Override
	public long submitLoopJob(String hostName, String password, String workingDirectory, long iter, double ratio) {
		return 0;
	}

	ServerSocket serverSocket = null;
	Socket socket = null;
	DataInputStream in = null;

	public void run() {

		int executionTime;
		String ratioString;
		int serverID;
		long mapperJobID;
		double ratio;

		int r;
		try {
			serverSocket = new ServerSocket(getServerPort());
		} catch (IOException e1) {
			e1.printStackTrace();
		}

		while (this.isAlive()) {

			executionTime = 0;
			ratioString = "";
			serverID = 0;
			mapperJobID = 0;
			ratio = 0;
			try {
				/*
				 * ---------------------Format of message received is
				 * -----------------------
				 */
				/* executionTime#ratio#serverID#mapperJobID */
				/* -------------------------------------------------------------------------- */
				socket = serverSocket.accept();
				if (!this.isAlive())
					return;

				in = new DataInputStream(socket.getInputStream());

				while ((r = in.read()) != -1 && r != '#') {
					if ((char) r - '0' >= 0) {
						executionTime = 10 * executionTime + ((char) r - '0');
					}
				}
				while ((r = in.read()) != -1 && r != '#') {
					ratioString = ratioString + ((char) r);
				}
				while ((r = in.read()) != -1 && r != '#') {
					if ((char) r - '0' >= 0) {
						serverID = 10 * serverID + ((char) r - '0');
					}
				}
				while ((r = in.read()) != -1 && r != '#') {
					if ((char) r - '0' >= 0) {
						mapperJobID = 10 * mapperJobID + ((char) r - '0');
					}
					dateSubmitted = new GregorianCalendar().getTimeInMillis();
				}

				

				ratio = Double.parseDouble(ratioString);

			} catch (IOException ex) {
				ex.printStackTrace();
			}
			this.setServerID(serverID);
			executerJobTable.put(ExecutionLayerIDCounter, new JobInfo(ExecutionLayerIDCounter, mapperJobID));
			dateStarted = new GregorianCalendar().getTime().toString();
			System.out.println("Job: " + mapperJobID + " started at " + dateStarted);
			
			executeJob(executionTime, ratio, ExecutionLayerIDCounter);
			
			dateStopped = new GregorianCalendar().getTime().toString();
			System.out.println("Job: " + mapperJobID + " completed at " + dateStopped);
			ExecutionLayerIDCounter++;
		}
	}
	
	public void executeJob(int executionTime, double ratio, long ExecutionLayerID) 
	{
		System.out.println("executionTime : " + executionTime + " Ratio : " + ratio);
		executerJobTable.get(ExecutionLayerID).dateStarted = new GregorianCalendar();
		double firstPercent = 1;
		int JobTime = (int)(executionTime * ratio);
		int SleepTime = 100;
		while (JobTime > 0)
		{
			try 
			{
				firstPercent = this.getIdlePercent();
				Thread.sleep(SleepTime);
				JobTime = JobTime - (int)(SleepTime * (avg(this.getIdlePercent(), firstPercent) * this.getAvailability()));
			} 
			catch (InterruptedException e) 
			{
				System.out.println("Job has not completed");
			}
		}
		executerJobTable.get(ExecutionLayerID).dateStopped = new GregorianCalendar();
		notifyMapper(ExecutionLayerID);
	}

	public double avg(double a, double b)
	{
			return (a + b)/2;
	}
	
	SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss Z");

	Socket mapperSocket = null;
	DataOutputStream out = null;

	public void notifyMapper(long ExecutionLayerID) {
		/*
		 * --------------------- Format of message sent is
		 * -----------------------
		 */
		/* TimeStamp#serverId#localAddress#jobID#secondaryJobID#dateStarted#dateDone */
		/* -------------------------------------------------------------------------- */
		String message;
		try {

			String dateStarted = formatter.format(executerJobTable.get(ExecutionLayerID).dateStarted.getTime());

			String dateDone = formatter.format(executerJobTable.get(ExecutionLayerID).dateStopped.getTime());

			message = formatter.format((new GregorianCalendar()).getTime()) + "#" + this.getServerID() + "#"
					+ this.getHostName() + "#" + executerJobTable.get(ExecutionLayerID).mapperJobID + "#"
					+ ExecutionLayerID + "#" + dateStarted + "#" + dateDone;
			mapperSocket = new Socket(this.getMapperHostName(), this.getCompletionServerPort());
			out = new DataOutputStream(mapperSocket.getOutputStream());
			out.writeBytes(message);
			this.dateSubmitted = 0;
			this.dateStarted = null;
			this.dateStopped = null;
		} catch (UnknownHostException e) {
			System.err.println("Don't know about mapper: " + this.getMapperHostName());
		} catch (IOException e) {
			System.err.println("FinishedThread.run: Couldn't Connect with the socket " + " the connection to: "
					+ this.getMapperHostName()
					+ " did not function properly. Make sure Completion Server is listening at "
					+ this.getCompletionServerPort());
		} finally {
			try {
				if (out != null) {
					out.flush();
					out.close();
				}
				if (mapperSocket != null)
					mapperSocket.close();
			} catch (IOException e) {
				e.printStackTrace();
			}

		}
	}

}
