rms/org.eclipse.ptp.rm.smoa.core/src/org/eclipse/ptp/rm/smoa/core/rmsystem/JobThread.java - ptp/org.eclipse.ptp - Git at Google

 /*******************************************************************************
  * Copyright (c) 2010 Poznan Supercomputing and Networking Center
  * 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:
  *     Jan Konczak (PSNC) - initial implementation
  ******************************************************************************/

 package org.eclipse.ptp.rm.smoa.core.rmsystem;

 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.util.BitSet;
 import java.util.HashMap;
 import java.util.Map;

 import org.eclipse.core.filesystem.IFileStore;
 import org.eclipse.core.runtime.CoreException;
 import org.eclipse.ptp.core.attributes.AttributeManager;
 import org.eclipse.ptp.core.elements.IPJob;
 import org.eclipse.ptp.core.elements.IPMachine;
 import org.eclipse.ptp.core.elements.IPNode;
 import org.eclipse.ptp.core.elements.IPResourceManager;
 import org.eclipse.ptp.core.elements.attributes.JobAttributes;
 import org.eclipse.ptp.core.elements.attributes.ProcessAttributes;
 import org.eclipse.ptp.remote.core.IRemoteFileManager;
 import org.eclipse.ptp.remote.core.PTPRemoteCorePlugin;
 import org.eclipse.ptp.rm.smoa.core.rmsystem.PoolingIntervalsAndStatic.SMOAJobState;
 import org.eclipse.ptp.rm.smoa.core.rservices.SMOAFileStore;
 import org.eclipse.ptp.rm.smoa.core.util.NotifyShell;

 import com.smoa.comp.sdk.SMOAFactory;
 import com.smoa.comp.sdk.exceptions.FileNotFoundException;
 import com.smoa.comp.sdk.exceptions.NotAuthorizedException;
 import com.smoa.comp.sdk.exceptions.UnknownActivityIdentifierException;
 import com.smoa.comp.sdk.types.ActivityEndpointReference;
 import com.smoa.comp.sdk.types.SMOAActivityStatus;

 /**
  * Thread for monitoring a single job - checks state and transfers standard
  * output. Allows job's termination.
  */
 class JobThread extends Thread {
 	// The RM that created the job
 	private final SMOAResourceManager rm;
 	private final IPResourceManager pRM;

 	// The eclipse job control object
 	private final IPJob jobControl;

 	// The SMOA job control object
 	private final ActivityEndpointReference activityIdentifier;

 	private final SMOAFactory factory;
 	private final IRemoteFileManager fileManager;

 	// Remote files used by launch
 	private final IFileStore out;
 	private final IFileStore err;
 	private final IFileStore sh;
 	private final IFileStore machinefile;

 	// Offsets for the output files
 	private int stdOutOffset = 0;
 	private int stdErrOffset = 0;

 	/**
 	 * Constructs the job thread for monitoring a single job
 	 */
 	public JobThread(SMOAResourceManager rm, SMOAFactory factory, ActivityEndpointReference activityEndpointReference,
 			IPJob jobControl, String out, String err, String sh, String machinefile) {
 		this.rm = rm;
 		this.pRM = (IPResourceManager) rm.getAdapter(IPResourceManager.class);
 		this.factory = factory;
 		this.activityIdentifier = activityEndpointReference;
 		this.jobControl = jobControl;

 		this.setPriority(MIN_PRIORITY);
 		this.setName("JobListener for " + activityIdentifier.getActivityUUID()); //$NON-NLS-1$

 		final IRemoteFileManager fileManager_t = PTPRemoteCorePlugin.getDefault()
 				.getRemoteServices(rm.getControlConfiguration().getRemoteServicesId())
 				.getFileManager(rm.getControlConfiguration().getConnection());

 		fileManager = fileManager_t;

 		this.out = fileManager.getResource(out);
 		this.err = fileManager.getResource(err);
 		if (sh != null) {
 			this.sh = fileManager.getResource(sh);
 		} else {
 			this.sh = null;
 		}
 		if (machinefile != null) {
 			this.machinefile = fileManager.getResource(machinefile);
 		} else {
 			this.machinefile = null;
 		}

 	}

 	/** Adds given text to standard error, and makes it appear on console */
 	void appendStdErr(String stderr) {
 		final AttributeManager outManager = new AttributeManager();
 		outManager.addAttribute(ProcessAttributes.getStderrAttributeDefinition().create(stderr));
 		final BitSet bs = new BitSet();
 		bs.set(jobControl.getProcessJobRanks().nextSetBit(0));
 		jobControl.addProcessAttributes(bs, outManager);
 	}

 	/** Adds given text to standard output, and makes it appear on console */
 	void appendStdOut(String stdout) {
 		final AttributeManager outManager = new AttributeManager();
 		outManager.addAttribute(ProcessAttributes.getStdoutAttributeDefinition().create(stdout));
 		final BitSet bs = new BitSet();
 		bs.set(jobControl.getProcessJobRanks().nextSetBit(0));
 		jobControl.addProcessAttributes(bs, outManager);
 	}

 	/**
 	 * Changes job state
 	 */
 	void changeState(JobAttributes.State newState) {
 		if (JobAttributes.State.RUNNING.equals(newState)) {
 			enteredRunPhase();
 		}

 		final AttributeManager am = new AttributeManager();
 		am.addAttribute(JobAttributes.getStateAttributeDefinition().create(newState));
 		rm.getMonitor().getRuntimeSystem().changeJob(jobControl.getID(), am);
 	}

 	/**
 	 * Called when the process state becomes RUNNING.
 	 *
 	 * This is the proper moment for identifying where the processes are
 	 * located, so that debug routing files may be written.
 	 */
 	private void enteredRunPhase() {
 		if (machinefile != null) {
 			try {
 				final BufferedReader reader = new BufferedReader(new InputStreamReader(machinefile.openInputStream(0, null)));

 				final IPMachine machine = pRM.getMachines()[0];

 				final Map<String, String> nodesByName = new HashMap<String, String>();
 				for (final IPNode ipNode : machine.getNodes()) {
 					nodesByName.put(ipNode.getName(), ipNode.getID());
 				}

 				String line;
 				int i = 0;
 				while ((line = reader.readLine()) != null) {
 					final BitSet bs = new BitSet();
 					bs.set(i++);
 					final AttributeManager attrs = new AttributeManager();

 					if (!nodesByName.containsKey(line)) {
 						rm.getMonitor().getRuntimeSystem().addUnknownNode(machine, line);
 						nodesByName.put(line, ((Integer) nodesByName.size()).toString());
 						NotifyShell.open(Messages.JobThread_UnknownNode_title, Messages.JobThread_UnknownNode_text_1 + line
 								+ Messages.JobThread_UnknownNode_text_2);
 						attrs.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create(nodesByName.get(line)));
 					} else {
 						attrs.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create(nodesByName.get(line)));
 					}
 					jobControl.addProcessesByJobRanks(bs, attrs);
 				}
 				return;
 			} catch (final CoreException e) {
 				NotifyShell.open(Messages.JobThread_ErrorOpeningRemote, e.toString());

 				final BitSet bs = new BitSet(1);
 				bs.set(0);
 				final AttributeManager jam = new AttributeManager();
 				jam.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create("0")); //$NON-NLS-1$
 				jobControl.addProcessesByJobRanks(bs, jam);
 			} catch (final IOException e) {
 				NotifyShell.open(Messages.JobThread_ErrorReadingRemote, e.toString());

 				if (jobControl.getProcessJobRanks() != null && !jobControl.getProcessJobRanks().isEmpty()) {
 					return;
 				}

 				final BitSet bs = new BitSet(1);
 				bs.set(0);
 				final AttributeManager jam = new AttributeManager();
 				jam.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create("0")); //$NON-NLS-1$
 				jobControl.addProcessesByJobRanks(bs, jam);
 			}
 		} else {
 			final BitSet bs = new BitSet(1);
 			bs.set(0);
 			final AttributeManager jam = new AttributeManager();
 			jam.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create("0")); //$NON-NLS-1$
 			jobControl.addProcessesByJobRanks(bs, jam);
 		}
 	}

 	/**
 	 * Called if an exception has been thrown while monitoring job
 	 */
 	void exceptionCaught(Exception e) {
 		final AttributeManager am = new AttributeManager();

 		am.addAttribute(JobAttributes.getStatusAttributeDefinition().create(Messages.JobThread_ExceptionByMonitoring));

 		am.addAttribute(JobAttributes.getStateAttributeDefinition().create(JobAttributes.State.COMPLETED));

 		am.addAttribute(PoolingIntervalsAndStatic.exceptionAttrDef.create(e.getLocalizedMessage()));

 		NotifyShell.open(Messages.JobThread_ExceptionByMonitoring, e.getLocalizedMessage());

 		rm.getMonitor().getRuntimeSystem().changeJob(jobControl.getID(), am);
 	}

 	/** Executed after the job reached terminal state */
 	private void jobFinished(SMOAActivityStatus status) {

 		try {
 			// If the job jumped from queued to finished state, we didn't
 			// add any processes yet
 			if (jobControl.getProcessJobRanks() == null || jobControl.getProcessJobRanks().isEmpty()) {
 				final BitSet bs = new BitSet(1);
 				bs.set(0);
 				final AttributeManager jam = new AttributeManager();
 				jam.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create("0")); //$NON-NLS-1$
 				jobControl.addProcessesByJobRanks(bs, jam);
 			}

 			processOutAndErr();

 			try {
 				out.delete(0, null);
 				err.delete(0, null);
 				if (sh != null) {
 					sh.delete(0, null);
 				}
 				if (machinefile != null) {
 					machinefile.delete(0, null);
 				}
 			} catch (final CoreException e) {
 				NotifyShell.open(Messages.JobThread_ErrorDeletingTempFiles_title, Messages.JobThread_ErrorDeletingTempFiles_text
 						+ e.getLocalizedMessage());
 				e.printStackTrace();
 			}
 		} catch (final RuntimeException e) {
 			// Happens if there is a problem with stdout/err file
 		}

 		final AttributeManager am = new AttributeManager();

 		String jobStatus = null;

 		switch (PoolingIntervalsAndStatic.getEquivalentJobState(status)) {
 		case Finished:
 			jobStatus = Messages.JobThread_JobStateFinisedWithStatus + status.getEndStatus().getExitStatus();
 			break;
 		case Failed:
 			jobStatus = Messages.JobThread_JobStateFailed;
 			break;
 		case Cancelled:
 			jobStatus = Messages.JobThread_JobStateCancelled;
 			break;
 		default:
 			jobStatus = Messages.JobThread_JobStateUnknown;
 		}

 		am.addAttribute(JobAttributes.getStatusAttributeDefinition().create(jobStatus));

 		JobAttributes.State state;
 		state = JobAttributes.State.COMPLETED;
 		am.addAttribute(JobAttributes.getStateAttributeDefinition().create(state));

 		rm.getMonitor().getRuntimeSystem().changeJob(jobControl.getID(), am);
 	}

 	/**
 	 * Takes care about reading out and err streams and forwards them on console
 	 */
 	private void processOutAndErr() {

 		/* Out */
 		InputStream is;
 		try {
 			if (out instanceof SMOAFileStore) {
 				is = ((SMOAFileStore) out).openInputStream(0, null, stdOutOffset);
 			} else {
 				is = out.openInputStream(0, null);
 				is.skip(stdOutOffset);
 			}

 			final byte[] buffer = new byte[512];

 			for (int count = is.read(buffer); count > 0; count = is.read(buffer)) {
 				stdOutOffset += count;
 				appendStdOut(new String(buffer).substring(0, count));
 			}
 		} catch (final CoreException e) {
 			if (e.getCause() instanceof FileNotFoundException) {
 				return;
 			}
 			throw new RuntimeException(e);
 		} catch (final IOException e) {
 			NotifyShell.open(Messages.JobThread_ErrorOut, e.getLocalizedMessage());
 		}

 		/* Err */

 		try {
 			if (err instanceof SMOAFileStore) {
 				is = ((SMOAFileStore) err).openInputStream(0, null, stdErrOffset);
 			} else {
 				is = err.openInputStream(0, null);
 				is.skip(stdErrOffset);
 			}

 			final byte[] buffer = new byte[512];

 			for (int count = is.read(buffer); count > 0; count = is.read(buffer)) {
 				stdErrOffset += count;
 				if (count == buffer.length) {
 					appendStdErr(new String(buffer));
 				} else {
 					appendStdErr(new String(buffer).substring(0, count));
 				}
 			}
 		} catch (final CoreException e) {
 			if (e.getCause() instanceof FileNotFoundException) {
 				return;
 			}
 			throw new RuntimeException(e);
 		} catch (final IOException e) {
 			NotifyShell.open(Messages.JobThread_ErrorErr, e.getLocalizedMessage());
 		}
 	}

 	/** Loop for monitoring task */
 	@Override
 	public void run() {
 		try {

 			SMOAActivityStatus status = factory.getActivityStatus(activityIdentifier);

 			SMOAActivityStatus prevStatus = null;

 			long nextStateCheck = System.currentTimeMillis();
 			long nextOutCheck = nextStateCheck;

 			// Till the state is not final, we pool the job and out streams
 			while (true) {

 				// Status
 				if (nextStateCheck <= System.currentTimeMillis()) {
 					nextStateCheck = System.currentTimeMillis() + PoolingIntervalsAndStatic.getPoolingIntervalTask();

 					final SMOAJobState statusS = PoolingIntervalsAndStatic.getEquivalentJobState(status);

 					SMOAJobState prevstatusS = null;

 					if (prevStatus != null) {
 						prevstatusS = PoolingIntervalsAndStatic.getEquivalentJobState(prevStatus);
 					}

 					if (prevStatus == null || !statusS.equals(prevstatusS)) {
 						switch (statusS) {
 						case Stage_in:
 						case Held:
 						case Queued:
 							changeState(JobAttributes.State.STARTING);
 							break;
 						case Stage_out:
 						case Executing:
 							changeState(JobAttributes.State.RUNNING);
 							break;
 						case Suspended:
 							changeState(JobAttributes.State.SUSPENDED);
 							break;

 						case Cancelled:
 							break;
 						case Failed:
 							break;
 						case Finished:
 							break;
 						}
 						prevStatus = status;
 					}
 					status = factory.getActivityStatus(activityIdentifier);

 					if (status.isFinalState()) {
 						break;
 					}
 				}

 				// Out
 				final JobAttributes.State state = jobControl.getAttribute(JobAttributes.getStateAttributeDefinition()).getValue();
 				if (state == JobAttributes.State.RUNNING && nextOutCheck <= System.currentTimeMillis()) {
 					nextOutCheck = System.currentTimeMillis() + PoolingIntervalsAndStatic.getPoolingIntervalOut();

 					processOutAndErr();
 				}
 				try {
 					final long nextCheck = Math.min(nextOutCheck - System.currentTimeMillis(),
 							nextStateCheck - System.currentTimeMillis());
 					if (nextCheck > 0) {
 						sleep(nextCheck);
 					}
 				} catch (final InterruptedException e) {
 					// The job has been terminated, ignore
 				}
 			}

 			jobFinished(status);

 		} catch (final NotAuthorizedException e) {
 			exceptionCaught(e);
 		} catch (final UnknownActivityIdentifierException e) {
 			exceptionCaught(e);
 		}

 		rm.removeJobThread(jobControl.getID());
 	}

 	/**
 	 * Terminates the monitored job
 	 */
 	public boolean terminate() {
 		try {
 			factory.terminateActivity(activityIdentifier);
 			this.interrupt();
 		} catch (final Exception e) {
 			return false;
 		}
 		return true;
 	}
 }
	/*******************************************************************************
	* Copyright (c) 2010 Poznan Supercomputing and Networking Center
	* 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:
	* Jan Konczak (PSNC) - initial implementation
	******************************************************************************/

	package org.eclipse.ptp.rm.smoa.core.rmsystem;

	import java.io.BufferedReader;
	import java.io.IOException;
	import java.io.InputStream;
	import java.io.InputStreamReader;
	import java.util.BitSet;
	import java.util.HashMap;
	import java.util.Map;

	import org.eclipse.core.filesystem.IFileStore;
	import org.eclipse.core.runtime.CoreException;
	import org.eclipse.ptp.core.attributes.AttributeManager;
	import org.eclipse.ptp.core.elements.IPJob;
	import org.eclipse.ptp.core.elements.IPMachine;
	import org.eclipse.ptp.core.elements.IPNode;
	import org.eclipse.ptp.core.elements.IPResourceManager;
	import org.eclipse.ptp.core.elements.attributes.JobAttributes;
	import org.eclipse.ptp.core.elements.attributes.ProcessAttributes;
	import org.eclipse.ptp.remote.core.IRemoteFileManager;
	import org.eclipse.ptp.remote.core.PTPRemoteCorePlugin;
	import org.eclipse.ptp.rm.smoa.core.rmsystem.PoolingIntervalsAndStatic.SMOAJobState;
	import org.eclipse.ptp.rm.smoa.core.rservices.SMOAFileStore;
	import org.eclipse.ptp.rm.smoa.core.util.NotifyShell;

	import com.smoa.comp.sdk.SMOAFactory;
	import com.smoa.comp.sdk.exceptions.FileNotFoundException;
	import com.smoa.comp.sdk.exceptions.NotAuthorizedException;
	import com.smoa.comp.sdk.exceptions.UnknownActivityIdentifierException;
	import com.smoa.comp.sdk.types.ActivityEndpointReference;
	import com.smoa.comp.sdk.types.SMOAActivityStatus;

	/**
	* Thread for monitoring a single job - checks state and transfers standard
	* output. Allows job's termination.
	*/
	class JobThread extends Thread {
	// The RM that created the job
	private final SMOAResourceManager rm;
	private final IPResourceManager pRM;

	// The eclipse job control object
	private final IPJob jobControl;

	// The SMOA job control object
	private final ActivityEndpointReference activityIdentifier;

	private final SMOAFactory factory;
	private final IRemoteFileManager fileManager;

	// Remote files used by launch
	private final IFileStore out;
	private final IFileStore err;
	private final IFileStore sh;
	private final IFileStore machinefile;

	// Offsets for the output files
	private int stdOutOffset = 0;
	private int stdErrOffset = 0;

	/**
	* Constructs the job thread for monitoring a single job
	*/
	public JobThread(SMOAResourceManager rm, SMOAFactory factory, ActivityEndpointReference activityEndpointReference,
	IPJob jobControl, String out, String err, String sh, String machinefile) {
	this.rm = rm;
	this.pRM = (IPResourceManager) rm.getAdapter(IPResourceManager.class);
	this.factory = factory;
	this.activityIdentifier = activityEndpointReference;
	this.jobControl = jobControl;

	this.setPriority(MIN_PRIORITY);
	this.setName("JobListener for " + activityIdentifier.getActivityUUID()); //$NON-NLS-1$

	final IRemoteFileManager fileManager_t = PTPRemoteCorePlugin.getDefault()
	.getRemoteServices(rm.getControlConfiguration().getRemoteServicesId())
	.getFileManager(rm.getControlConfiguration().getConnection());

	fileManager = fileManager_t;

	this.out = fileManager.getResource(out);
	this.err = fileManager.getResource(err);
	if (sh != null) {
	this.sh = fileManager.getResource(sh);
	} else {
	this.sh = null;
	}
	if (machinefile != null) {
	this.machinefile = fileManager.getResource(machinefile);
	} else {
	this.machinefile = null;
	}

	}

	/** Adds given text to standard error, and makes it appear on console */
	void appendStdErr(String stderr) {
	final AttributeManager outManager = new AttributeManager();
	outManager.addAttribute(ProcessAttributes.getStderrAttributeDefinition().create(stderr));
	final BitSet bs = new BitSet();
	bs.set(jobControl.getProcessJobRanks().nextSetBit(0));
	jobControl.addProcessAttributes(bs, outManager);
	}

	/** Adds given text to standard output, and makes it appear on console */
	void appendStdOut(String stdout) {
	final AttributeManager outManager = new AttributeManager();
	outManager.addAttribute(ProcessAttributes.getStdoutAttributeDefinition().create(stdout));
	final BitSet bs = new BitSet();
	bs.set(jobControl.getProcessJobRanks().nextSetBit(0));
	jobControl.addProcessAttributes(bs, outManager);
	}

	/**
	* Changes job state
	*/
	void changeState(JobAttributes.State newState) {
	if (JobAttributes.State.RUNNING.equals(newState)) {
	enteredRunPhase();
	}

	final AttributeManager am = new AttributeManager();
	am.addAttribute(JobAttributes.getStateAttributeDefinition().create(newState));
	rm.getMonitor().getRuntimeSystem().changeJob(jobControl.getID(), am);
	}

	/**
	* Called when the process state becomes RUNNING.
	*
	* This is the proper moment for identifying where the processes are
	* located, so that debug routing files may be written.
	*/
	private void enteredRunPhase() {
	if (machinefile != null) {
	try {
	final BufferedReader reader = new BufferedReader(new InputStreamReader(machinefile.openInputStream(0, null)));

	final IPMachine machine = pRM.getMachines()[0];

	final Map<String, String> nodesByName = new HashMap<String, String>();
	for (final IPNode ipNode : machine.getNodes()) {
	nodesByName.put(ipNode.getName(), ipNode.getID());
	}

	String line;
	int i = 0;
	while ((line = reader.readLine()) != null) {
	final BitSet bs = new BitSet();
	bs.set(i++);
	final AttributeManager attrs = new AttributeManager();

	if (!nodesByName.containsKey(line)) {
	rm.getMonitor().getRuntimeSystem().addUnknownNode(machine, line);
	nodesByName.put(line, ((Integer) nodesByName.size()).toString());
	NotifyShell.open(Messages.JobThread_UnknownNode_title, Messages.JobThread_UnknownNode_text_1 + line
	+ Messages.JobThread_UnknownNode_text_2);
	attrs.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create(nodesByName.get(line)));
	} else {
	attrs.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create(nodesByName.get(line)));
	}
	jobControl.addProcessesByJobRanks(bs, attrs);
	}
	return;
	} catch (final CoreException e) {
	NotifyShell.open(Messages.JobThread_ErrorOpeningRemote, e.toString());

	final BitSet bs = new BitSet(1);
	bs.set(0);
	final AttributeManager jam = new AttributeManager();
	jam.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create("0")); //$NON-NLS-1$
	jobControl.addProcessesByJobRanks(bs, jam);
	} catch (final IOException e) {
	NotifyShell.open(Messages.JobThread_ErrorReadingRemote, e.toString());

	if (jobControl.getProcessJobRanks() != null && !jobControl.getProcessJobRanks().isEmpty()) {
	return;
	}

	final BitSet bs = new BitSet(1);
	bs.set(0);
	final AttributeManager jam = new AttributeManager();
	jam.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create("0")); //$NON-NLS-1$
	jobControl.addProcessesByJobRanks(bs, jam);
	}
	} else {
	final BitSet bs = new BitSet(1);
	bs.set(0);
	final AttributeManager jam = new AttributeManager();
	jam.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create("0")); //$NON-NLS-1$
	jobControl.addProcessesByJobRanks(bs, jam);
	}
	}

	/**
	* Called if an exception has been thrown while monitoring job
	*/
	void exceptionCaught(Exception e) {
	final AttributeManager am = new AttributeManager();

	am.addAttribute(JobAttributes.getStatusAttributeDefinition().create(Messages.JobThread_ExceptionByMonitoring));

	am.addAttribute(JobAttributes.getStateAttributeDefinition().create(JobAttributes.State.COMPLETED));

	am.addAttribute(PoolingIntervalsAndStatic.exceptionAttrDef.create(e.getLocalizedMessage()));

	NotifyShell.open(Messages.JobThread_ExceptionByMonitoring, e.getLocalizedMessage());

	rm.getMonitor().getRuntimeSystem().changeJob(jobControl.getID(), am);
	}

	/** Executed after the job reached terminal state */
	private void jobFinished(SMOAActivityStatus status) {

	try {
	// If the job jumped from queued to finished state, we didn't
	// add any processes yet
	if (jobControl.getProcessJobRanks() == null \|\| jobControl.getProcessJobRanks().isEmpty()) {
	final BitSet bs = new BitSet(1);
	bs.set(0);
	final AttributeManager jam = new AttributeManager();
	jam.addAttribute(ProcessAttributes.getNodeIdAttributeDefinition().create("0")); //$NON-NLS-1$
	jobControl.addProcessesByJobRanks(bs, jam);
	}

	processOutAndErr();

	try {
	out.delete(0, null);
	err.delete(0, null);
	if (sh != null) {
	sh.delete(0, null);
	}
	if (machinefile != null) {
	machinefile.delete(0, null);
	}
	} catch (final CoreException e) {
	NotifyShell.open(Messages.JobThread_ErrorDeletingTempFiles_title, Messages.JobThread_ErrorDeletingTempFiles_text
	+ e.getLocalizedMessage());
	e.printStackTrace();
	}
	} catch (final RuntimeException e) {
	// Happens if there is a problem with stdout/err file
	}

	final AttributeManager am = new AttributeManager();

	String jobStatus = null;

	switch (PoolingIntervalsAndStatic.getEquivalentJobState(status)) {
	case Finished:
	jobStatus = Messages.JobThread_JobStateFinisedWithStatus + status.getEndStatus().getExitStatus();
	break;
	case Failed:
	jobStatus = Messages.JobThread_JobStateFailed;
	break;
	case Cancelled:
	jobStatus = Messages.JobThread_JobStateCancelled;
	break;
	default:
	jobStatus = Messages.JobThread_JobStateUnknown;
	}

	am.addAttribute(JobAttributes.getStatusAttributeDefinition().create(jobStatus));

	JobAttributes.State state;
	state = JobAttributes.State.COMPLETED;
	am.addAttribute(JobAttributes.getStateAttributeDefinition().create(state));

	rm.getMonitor().getRuntimeSystem().changeJob(jobControl.getID(), am);
	}

	/**
	* Takes care about reading out and err streams and forwards them on console
	*/
	private void processOutAndErr() {

	/* Out */
	InputStream is;
	try {
	if (out instanceof SMOAFileStore) {
	is = ((SMOAFileStore) out).openInputStream(0, null, stdOutOffset);
	} else {
	is = out.openInputStream(0, null);
	is.skip(stdOutOffset);
	}

	final byte[] buffer = new byte[512];

	for (int count = is.read(buffer); count > 0; count = is.read(buffer)) {
	stdOutOffset += count;
	appendStdOut(new String(buffer).substring(0, count));
	}
	} catch (final CoreException e) {
	if (e.getCause() instanceof FileNotFoundException) {
	return;
	}
	throw new RuntimeException(e);
	} catch (final IOException e) {
	NotifyShell.open(Messages.JobThread_ErrorOut, e.getLocalizedMessage());
	}

	/* Err */

	try {
	if (err instanceof SMOAFileStore) {
	is = ((SMOAFileStore) err).openInputStream(0, null, stdErrOffset);
	} else {
	is = err.openInputStream(0, null);
	is.skip(stdErrOffset);
	}

	final byte[] buffer = new byte[512];

	for (int count = is.read(buffer); count > 0; count = is.read(buffer)) {
	stdErrOffset += count;
	if (count == buffer.length) {
	appendStdErr(new String(buffer));
	} else {
	appendStdErr(new String(buffer).substring(0, count));
	}
	}
	} catch (final CoreException e) {
	if (e.getCause() instanceof FileNotFoundException) {
	return;
	}
	throw new RuntimeException(e);
	} catch (final IOException e) {
	NotifyShell.open(Messages.JobThread_ErrorErr, e.getLocalizedMessage());
	}
	}

	/** Loop for monitoring task */
	@Override
	public void run() {
	try {

	SMOAActivityStatus status = factory.getActivityStatus(activityIdentifier);

	SMOAActivityStatus prevStatus = null;

	long nextStateCheck = System.currentTimeMillis();
	long nextOutCheck = nextStateCheck;

	// Till the state is not final, we pool the job and out streams
	while (true) {

	// Status
	if (nextStateCheck <= System.currentTimeMillis()) {
	nextStateCheck = System.currentTimeMillis() + PoolingIntervalsAndStatic.getPoolingIntervalTask();

	final SMOAJobState statusS = PoolingIntervalsAndStatic.getEquivalentJobState(status);

	SMOAJobState prevstatusS = null;

	if (prevStatus != null) {
	prevstatusS = PoolingIntervalsAndStatic.getEquivalentJobState(prevStatus);
	}

	if (prevStatus == null \|\| !statusS.equals(prevstatusS)) {
	switch (statusS) {
	case Stage_in:
	case Held:
	case Queued:
	changeState(JobAttributes.State.STARTING);
	break;
	case Stage_out:
	case Executing:
	changeState(JobAttributes.State.RUNNING);
	break;
	case Suspended:
	changeState(JobAttributes.State.SUSPENDED);
	break;

	case Cancelled:
	break;
	case Failed:
	break;
	case Finished:
	break;
	}
	prevStatus = status;
	}
	status = factory.getActivityStatus(activityIdentifier);

	if (status.isFinalState()) {
	break;
	}
	}

	// Out
	final JobAttributes.State state = jobControl.getAttribute(JobAttributes.getStateAttributeDefinition()).getValue();
	if (state == JobAttributes.State.RUNNING && nextOutCheck <= System.currentTimeMillis()) {
	nextOutCheck = System.currentTimeMillis() + PoolingIntervalsAndStatic.getPoolingIntervalOut();

	processOutAndErr();
	}
	try {
	final long nextCheck = Math.min(nextOutCheck - System.currentTimeMillis(),
	nextStateCheck - System.currentTimeMillis());
	if (nextCheck > 0) {
	sleep(nextCheck);
	}
	} catch (final InterruptedException e) {
	// The job has been terminated, ignore
	}
	}

	jobFinished(status);

	} catch (final NotAuthorizedException e) {
	exceptionCaught(e);
	} catch (final UnknownActivityIdentifierException e) {
	exceptionCaught(e);
	}

	rm.removeJobThread(jobControl.getID());
	}

	/**
	* Terminates the monitored job
	*/
	public boolean terminate() {
	try {
	factory.terminateActivity(activityIdentifier);
	this.interrupt();
	} catch (final Exception e) {
	return false;
	}
	return true;
	}
	}