package pulling;


import java.io.DataInputStream;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;

import pulling.availability_imposer.CPU_EatersLauncher;
import pulling.availability_predection.AvailabilityManager;
import pulling.executing.ServerSideExecuter;
import pulling.executing.ServerSideExecuter_BuiltIn;

/**
 * This class is responsible for managing the Puller module on every server.
 * It is also responsible for communication with the central Mapper.
 * 
 * @author Majd Kokaly and Ben Kybartas
 *
 */
public class Puller {
	/**
	 * The executer is responsible for the execution of jobs. 
	 */
	private ServerSideExecuter executer;
	
	/**
	 * This module is responsible for managing availability info and logging.
	 */
	private AvailabilityManager availabilityManager;
	
	/**
	 * This module is able to launch processes to eat CPU time
	 */
	private CPU_EatersLauncher cpuEater;

	/**
	 * This is the port that this Puller listens on 
	 * and receives messages from Mapper.
	 */
	private int pullerPort = 37934;
	
	/**
	 * The message START that Mapper sends.
	 */
	final static private String START = "start";
	
	/**
	 * The message STOP that Mapper sends.
	 */
	final static private String STOP = "stop";
	
	/**
	 * The message PING that Mapper sends.
	 */
	final static private String PING = "ping";
	
	/**
	 * The message RESTART that Mapper sends.
	 */
	final static private String RESTART = "restart";
	
	final static private int  NON_ACTIVE = -3;
	
	
	public Puller() {

	}

	public AvailabilityManager getAvailabilityManager() {
		return availabilityManager;
	}

	public void setAvailabilityManager(AvailabilityManager availabilityManager) {
		this.availabilityManager = availabilityManager;
	}

	public CPU_EatersLauncher getCpuEater() {
		return cpuEater;
	}

	public void setCpuEater(CPU_EatersLauncher cpuEater) {
		this.cpuEater = cpuEater;
	}

	public int getPullerPort() {
		return pullerPort;
	}

	public void setPullerPort(int pullerPort) {
		this.pullerPort = pullerPort;
	}


	/**
	 * This method is invoked from the main method it initialize the Availability Manager object.
	 * 
	 * @param systemResolution the system resolution. Which is the time that the Availability Manager object send a new value of aj
	 * @param mapperHostName it is the host name of the Mapper machine. 
	 * @param mode this defines the mode in which the Availability Manager works
	 * @param weights is the array of weights. Please refer to the Thesis document Section Availability model.
	 * @param c Please refer to the Thesis document Section Availability model.
	 */
	public void startAvailabilityManager(double systemResolution, String mapperHostName, int mode, double[] weights,
			double c, double availability) {
		if (this.getAvailabilityManager() != null)
			this.getAvailabilityManager().setAlive(false);

		if (mode == AvailabilityManager.WEIGHTED_MEAN)
			this.setAvailabilityManager(new AvailabilityManager((int) systemResolution, mapperHostName, weights, availability));

		if (mode == AvailabilityManager.RECURSIVE)
			this.setAvailabilityManager(new AvailabilityManager((int) systemResolution, mapperHostName, c, availability));

		/* ---- Start Availability Manager */
		this.getAvailabilityManager().startThread();
	}

	/**
	 * This method start a CPU_Eater that eats a percentage of the CPU for an specific amount of time
	 * @param percentage the percentage to be eaten from the CPU
	 * @param forHowLong the time in minutes for how long for the eater to work.
	 */
	public void startCPUEater(double percentage, double forHowLong) {
		this.setCpuEater(new CPU_EatersLauncher(percentage, forHowLong));
		this.getCpuEater().startEaters();
	}

	/**
	 * This is the main loop.
	 * The puller keeps running in an infinite loop until the Mapper module sends a kill message.
	 * @param args 
	 */
	public static void main(String[] args) {
		Puller puller = new Puller();

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

		try {
			serverSocket = new ServerSocket(puller.getPullerPort());
		} catch (IOException e1) {
			e1.printStackTrace();
		}
		while (true) {
			
			String str = null;

			String sysResolutionString = "";
			double sysResolution = 4;

			String availabilityString = "";
			double availability = 0.3;

			String forHowLongString = "";
			double forHowLong = 3; // In minutes

			String mapperHostName = "";

			String availabilityModeString = "";
			int mode = AvailabilityManager.WEIGHTED_MEAN;

			String N_String = "";
			int n = 4;

			String weightsString = "";
			double[] weights = {};

			String cString = "";
			double c = 0.5;

			
			
			/* ---- Waiting to be Started ---- */

			try {
				System.out.println("Puller (" + java.net.InetAddress.getLocalHost().getCanonicalHostName() + ") is waiting to be started: ");
			} catch (UnknownHostException e1) {
				e1.printStackTrace();
			}
			try {
				// Message format is
				// start#systemResolution#mapperHostName#password#percentage#forHowLong#AvailabilityManagerMode#N#Weights#C
				socket = serverSocket.accept();
				in = new DataInputStream(socket.getInputStream());

				int r;

				str = new String();

				/* ---- Reading First Word (Command) ---- */
				while ((r = in.read()) != -1 && r != '#') {
					str = str + ((char) r);
				}
				
				if (str.equalsIgnoreCase(STOP)) {
					System.out.print("Exited");
					System.exit(1);
				}

				if (!str.equalsIgnoreCase(START)) {
					continue;
				}

				/* ---- Reading System Resolution ---- */
				while ((r = in.read()) != -1 && r != '#') {
					sysResolutionString = sysResolutionString + ((char) r);
				}

				/* ---- Reading Mapper Hostname ---- */
				while ((r = in.read()) != -1 && r != '#') {
					mapperHostName = mapperHostName + ((char) r);
				}

				/* ---- Reading Percentage for CPU Eater ---- */
				while ((r = in.read()) != -1 && r != '#') {
					availabilityString = availabilityString + ((char) r);
				}

				/* ---- Reading period Length for CPU period ---- */
				while ((r = in.read()) != -1 && r != '#') {
					forHowLongString = forHowLongString + ((char) r);
				}

				/* ---- availability Mode String ---- */
				while ((r = in.read()) != -1 && r != '#') {
					availabilityModeString = availabilityModeString + ((char) r);
				}

				/* ---- Reading N String ---- */
				while ((r = in.read()) != -1 && r != '#') {
					N_String = N_String + ((char) r);
				}

				/* ---- Reading WeightsString ---- */
				while ((r = in.read()) != -1 && r != '#') {
					weightsString = weightsString + ((char) r);
				}

				/* ---- Reading the C String ---- */
				while ((r = in.read()) != -1 && r != '#') {
					cString = cString + ((char) r);
				}

				

				sysResolution = Double.parseDouble(sysResolutionString);
				availability = Double.parseDouble(availabilityString);
				forHowLong = Double.parseDouble(forHowLongString);

				if (availabilityModeString.trim().charAt(0) == 'R' || availabilityModeString.trim().charAt(0) == 'r')
					mode = AvailabilityManager.RECURSIVE;
				
				if (availabilityModeString.trim().charAt(0) == 'N' || availabilityModeString.trim().charAt(0) == 'n')
					mode = Puller.NON_ACTIVE;
				// else mode = AvailabilityManager.WEIGHTED_MEAN;

				n = Integer.parseInt(N_String);

				weights = parseWeightsString(weightsString, n);

				c = Double.parseDouble(cString);

			} catch (IOException e) {
				e.printStackTrace();
			} catch (NumberFormatException ex) {
				ex.printStackTrace();
			}
			
			if ( mode != Puller.NON_ACTIVE ) {
				puller.startAvailabilityManager(sysResolution, mapperHostName, mode, weights, c, availability);
				System.out.println("Availability module started ");
			} else
				System.out.println("Availability module did NOT start.");
			
			if (availability != 1) {
				//puller.startCPUEater(1-availability, forHowLong);
				System.out.println("Availability Altered:" + (1-availability));
			} else
				System.out.println("Availability Normal");
			
			puller.executer = new ServerSideExecuter_BuiltIn(mapperHostName);
			puller.executer.startExecuter();
			puller.executer.setAvailability(availability);
			System.out.println("Executer started ");

			int r;
			String command = "";

			
			while (true) {
				try {
					socket = serverSocket.accept();
					in = new DataInputStream(socket.getInputStream());

					command = "";
					while ((r = in.read()) != -1) {
						command = command + (char) r;
					}
					System.out.println("command:" + command);

					if (command.equalsIgnoreCase(STOP)) {
						if (puller.executer != null)
							puller.executer.stopExecuter();
						if (puller.cpuEater != null)
							puller.cpuEater.stopEaters();
						if (puller.availabilityManager != null)
							puller.availabilityManager.stopThread();
						System.out.print("Exited");
						System.exit(1);
					}
					
					if (command.equalsIgnoreCase(PING)) {
						System.out.println("Pinged");
						continue;
					}
					
					if (command.equalsIgnoreCase(RESTART)) {
						System.out.println("RESTARTED");
						if (puller.executer != null)
							puller.executer.stopExecuter();
						if (puller.cpuEater != null)
							puller.cpuEater.stopEaters();
						if (puller.availabilityManager != null)
							puller.availabilityManager.stopThread();
						break;
					}

				} catch (IOException e) {
					e.printStackTrace();
				}
				
				
			}

		
		}

		/* ---- Start All the threads ---- */

		/* ---- Waiting to be Shut Off ---- */
	}

	public static double[] parseWeightsString(String weightString, int n) {
		double[] weights = new double[n];

		int i = 0, beginIndex = 0, wIndex = 0;
		// 1.3;4.7;4.1
		while (i < weightString.length()) {
			if (weightString.charAt(i) == ';') {
				weights[wIndex++] = Double.parseDouble(weightString.substring(beginIndex, i));
				beginIndex = i + 1;
			}
			i++;
		}
		weights[wIndex] = Double.parseDouble(weightString.substring(beginIndex));
		return weights;
	}
}
