/*******************************************************************************
 * Copyright (c) 2010 IBM Corporation.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * Contributors:
 *     IBM Corporation - initial API and implementation
 ******************************************************************************/
package org.eclipse.ptp.rm.generic.core.rtsystem;

import java.util.BitSet;
import java.util.HashMap;
import java.util.Map;

import org.eclipse.core.runtime.CoreException;
import org.eclipse.core.runtime.IProgressMonitor;
import org.eclipse.osgi.util.NLS;
import org.eclipse.ptp.core.PTPCorePlugin;
import org.eclipse.ptp.core.attributes.AttributeManager;
import org.eclipse.ptp.core.attributes.IAttribute;
import org.eclipse.ptp.core.elementcontrols.IPJobControl;
import org.eclipse.ptp.core.elements.IPJob;
import org.eclipse.ptp.core.elements.IPQueue;
import org.eclipse.ptp.core.elements.IResourceManager;
import org.eclipse.ptp.core.elements.attributes.ProcessAttributes;
import org.eclipse.ptp.remote.core.IRemoteProcessBuilder;
import org.eclipse.ptp.rm.core.MPIJobAttributes;
import org.eclipse.ptp.rm.core.rtsystem.AbstractToolRuntimeSystem;
import org.eclipse.ptp.rm.core.rtsystem.AbstractToolRuntimeSystemJob;
import org.eclipse.ptp.rm.core.utils.DebugUtil;
import org.eclipse.ptp.rm.core.utils.IInputStreamListener;
import org.eclipse.ptp.rm.core.utils.InputStreamObserver;
import org.eclipse.ptp.rm.generic.core.GenericRMCorePlugin;
import org.eclipse.ptp.rm.generic.core.messages.Messages;

public class GenericRMRuntimeSystemJob extends AbstractToolRuntimeSystemJob {
	private InputStreamObserver stderrObserver;
	private InputStreamObserver stdoutObserver;

	public GenericRMRuntimeSystemJob(String jobID, String queueID, String name, AbstractToolRuntimeSystem rtSystem,
			AttributeManager attrMgr) {
		super(jobID, queueID, name, rtSystem, attrMgr);
	}

	/**
	 * Terminate all processes.
	 */
	private void terminateProcesses() {
		final GenericRMRuntimeSystem rtSystem = (GenericRMRuntimeSystem) getRtSystem();
		final IResourceManager rm = PTPCorePlugin.getDefault().getUniverse().getResourceManager(rtSystem.getRmID());
		if (rm != null) {
			final IPQueue queue = rm.getQueueById(getQueueID());
			if (queue != null) {
				final IPJob ipJob = queue.getJobById(getJobID());
				if (ipJob != null) {

					/*
					 * Mark all running and starting processes as finished.
					 */

					AttributeManager attrMrg = new AttributeManager();
					attrMrg.addAttribute(ProcessAttributes.getStateAttributeDefinition().create(ProcessAttributes.State.COMPLETED));
					final BitSet procJobRanks = ipJob.getProcessJobRanks();
					rtSystem.changeProcesses(ipJob.getID(), procJobRanks, attrMrg);
				}
			}
		}
	}

	/**
	 * Add a process to the job
	 * 
	 * @param job
	 * @param proc
	 */
	protected void addProcess(IPJob job) {
		GenericRMRuntimeSystem rts = (GenericRMRuntimeSystem) getRtSystem();
		rts.createProcesses(job.getID(), 1);
		final BitSet processIndices = new BitSet();
		processIndices.set(0);
		AttributeManager attrMgr = new AttributeManager();
		attrMgr.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create(rts.getNodeID()));
		attrMgr.addAttribute(ProcessAttributes.getStateAttributeDefinition().create(ProcessAttributes.State.RUNNING));
		getRtSystem().changeProcesses(job.getID(), processIndices, attrMgr);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.eclipse.ptp.rm.core.rtsystem.AbstractToolRuntimeSystemJob#
	 * doBeforeExecution(org.eclipse.core.runtime.IProgressMonitor,
	 * org.eclipse.ptp.remote.core.IRemoteProcessBuilder)
	 */
	@Override
	protected void doBeforeExecution(IProgressMonitor monitor, IRemoteProcessBuilder builder) throws CoreException {
		// nothing
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.eclipse.ptp.rm.core.rtsystem.AbstractToolRuntimeSystemJob#
	 * doExecutionCleanUp(org.eclipse.core.runtime.IProgressMonitor)
	 */
	@Override
	protected void doExecutionCleanUp(IProgressMonitor monitor) {
		if (getProcess() != null) {
			getProcess().destroy();
			setProcess(null);
		}
		if (getStderrObserver() != null) {
			getStderrObserver().kill();
			setStderrObserver(null);
		}
		if (getStdoutObserver() != null) {
			getStdoutObserver().kill();
			setStdoutObserver(null);
		}
		// TODO: more cleanup?
		terminateProcesses();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.eclipse.ptp.rm.core.rtsystem.AbstractToolRuntimeSystemJob#
	 * doExecutionFinished(org.eclipse.core.runtime.IProgressMonitor)
	 */
	@Override
	protected void doExecutionFinished(IProgressMonitor monitor) throws CoreException {
		terminateProcesses();
		if (getProcess().exitValue() != 0) {
			if (!terminateJobFlag) {
				changeJobStatusMessage(NLS.bind(Messages.GenericRMRuntimeSystemJob_Exception_ExecutionFailedWithExitValue,
						new Integer(getProcess().exitValue())));
				changeJobStatus(MPIJobAttributes.Status.ERROR);
			}

			DebugUtil
					.trace(DebugUtil.RTS_JOB_TRACING,
							"RTS job #{0}: ignoring exit value {1} because job was forced to terminate by user", getJobID(), new Integer(getProcess().exitValue())); //$NON-NLS-1$
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.eclipse.ptp.rm.core.rtsystem.AbstractToolRuntimeSystemJob#
	 * doExecutionStarted(org.eclipse.core.runtime.IProgressMonitor)
	 */
	@Override
	protected void doExecutionStarted(IProgressMonitor monitor) throws CoreException {
		/*
		 * Create processes for the job.
		 */
		final IPJob job = PTPCorePlugin.getDefault().getUniverse().getResourceManager(getRtSystem().getRmID())
				.getQueueById(getQueueID()).getJobById(getJobID());
		addProcess(job);

		/*
		 * We only require procZero if we're using OMPI 1.2.x or 1.3.[0-3].
		 * Other versions use XML for stdout and stderr.
		 */
		final BitSet procZero = new BitSet();
		if (job.hasProcessByJobRank(0)) {
			procZero.set(0);
		}

		/*
		 * 
		 * Listener that saves stdout.
		 */
		final IInputStreamListener stdoutListener = new IInputStreamListener() {
			public void newBytes(byte[] bytes, int length) {
				String line = new String(bytes, 0, length);
				if (!procZero.isEmpty()) {
					final AttributeManager attributes = new AttributeManager(ProcessAttributes.getStdoutAttributeDefinition()
							.create(line));
					((IPJobControl) job).addProcessAttributes(procZero, attributes);
				}
				DebugUtil.trace(DebugUtil.RTS_JOB_OUTPUT_TRACING, "RTS job #{0}: {1}", getJobID(), line); //$NON-NLS-1$
			}

			public void streamClosed() {
				// No need to do anything
			}

			public void streamError(Exception e) {
				DebugUtil.trace(DebugUtil.RTS_JOB_TRACING_MORE, "RTS job #{0}: stdout stream: {0}", e); //$NON-NLS-1$
				GenericRMCorePlugin.log(e);
			}
		};

		/*
		 * 
		 * Listener that saves stderr.
		 */
		final IInputStreamListener stderrListener = new IInputStreamListener() {
			public void newBytes(byte[] bytes, int length) {
				String line = new String(bytes, 0, length);
				if (!procZero.isEmpty()) {
					final AttributeManager attributes = new AttributeManager(ProcessAttributes.getStderrAttributeDefinition()
							.create(line));
					((IPJobControl) job).addProcessAttributes(procZero, attributes);
				}
				DebugUtil.error(DebugUtil.RTS_JOB_OUTPUT_TRACING, "RTS job #{0}: {1}", getJobID(), line); //$NON-NLS-1$
			}

			public void streamClosed() {
				//
			}

			public void streamError(Exception e) {
				DebugUtil.trace(DebugUtil.RTS_JOB_TRACING_MORE, "RTS job #{0}: stderr stream: {0}", e); //$NON-NLS-1$
				GenericRMCorePlugin.log(e);
			}
		};

		setStderrObserver(new InputStreamObserver(getProcess().getErrorStream()));
		getStderrObserver().addListener(stderrListener);
		getStderrObserver().start();

		setStdoutObserver(new InputStreamObserver(getProcess().getInputStream()));
		getStdoutObserver().addListener(stdoutListener);
		getStdoutObserver().start();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.eclipse.ptp.rm.core.rtsystem.AbstractToolRuntimeSystemJob#
	 * doPrepareExecution(org.eclipse.core.runtime.IProgressMonitor)
	 */
	@Override
	protected void doPrepareExecution(IProgressMonitor monitor) throws CoreException {
		// Nothing to do
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.eclipse.ptp.rm.core.rtsystem.AbstractToolRuntimeSystemJob#
	 * doRetrieveToolBaseSubstitutionAttributes()
	 */
	@Override
	protected IAttribute<?, ?, ?>[] doRetrieveToolBaseSubstitutionAttributes() throws CoreException {
		// TODO make macros available for environment variables and work
		// directory.
		return null;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see org.eclipse.ptp.rm.core.rtsystem.AbstractToolRuntimeSystemJob#
	 * doRetrieveToolCommandSubstitutionAttributes
	 * (org.eclipse.ptp.core.attributes.AttributeManager, java.lang.String,
	 * java.util.Map)
	 */
	@Override
	protected IAttribute<?, ?, ?>[] doRetrieveToolCommandSubstitutionAttributes(AttributeManager baseSubstitutionAttributeManager,
			String directory, Map<String, String> environment) {
		// No extra variables need to be set.
		return null;
	}

	@Override
	protected HashMap<String, String> doRetrieveToolEnvironment() throws CoreException {
		// No extra environment variable needs to be set.
		return null;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * org.eclipse.ptp.rm.core.rtsystem.AbstractToolRuntimeSystemJob#doTerminateJob
	 * ()
	 */
	@Override
	protected void doTerminateJob() {
		// Empty implementation.
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * org.eclipse.ptp.rm.core.rtsystem.AbstractToolRuntimeSystemJob#doWaitExecution
	 * (org.eclipse.core.runtime.IProgressMonitor)
	 */
	@Override
	protected void doWaitExecution(IProgressMonitor monitor) throws CoreException {
		/*
		 * Wait until both stdout and stderr stop because stream are closed.
		 * This means that the process has finished.
		 */
		DebugUtil.trace(DebugUtil.RTS_JOB_TRACING_MORE, "RTS job #{0}: waiting stderr thread to finish", getJobID()); //$NON-NLS-1$
		try {
			getStderrObserver().join();
		} catch (InterruptedException e1) {
			// Ignore
		}

		DebugUtil.trace(DebugUtil.RTS_JOB_TRACING_MORE, "RTS job #{0}: waiting stdout thread to finish", getJobID()); //$NON-NLS-1$
		try {
			getStdoutObserver().join();
		} catch (InterruptedException e1) {
			// Ignore
		}

		/*
		 * Still experience has shown that remote process might not have yet
		 * terminated, although stdout and stderr is closed.
		 */
		DebugUtil.trace(DebugUtil.RTS_JOB_TRACING_MORE, "RTS job #{0}: waiting mpi process to finish completely", getJobID()); //$NON-NLS-1$
		try {
			getProcess().waitFor();
		} catch (InterruptedException e) {
			// Ignore
		}

		DebugUtil.trace(DebugUtil.RTS_JOB_TRACING_MORE, "RTS job #{0}: completely finished", getJobID()); //$NON-NLS-1$
	}

	/**
	 * @return the stderrObserver
	 */
	protected InputStreamObserver getStderrObserver() {
		return stderrObserver;
	}

	/**
	 * @return the stdoutObserver
	 */
	protected InputStreamObserver getStdoutObserver() {
		return stdoutObserver;
	}

	/**
	 * @param stderrObserver
	 *            the stderrObserver to set
	 */
	protected void setStderrObserver(InputStreamObserver stderrObserver) {
		this.stderrObserver = stderrObserver;
	}

	/**
	 * @param stdoutObserver
	 *            the stdoutObserver to set
	 */
	protected void setStdoutObserver(InputStreamObserver stdoutObserver) {
		this.stdoutObserver = stdoutObserver;
	}
}