使用 OGCE 向网格资源提交作业

				
package org.globus.cog.core.interfaces;
import java.util.Calendar;

public interface Task
{
    public static final int JOB_SUBMISSION = 1;
    public static final int FILE_TRANSFER = 2;
    public static final int INFORMATION_QUERY = 3;

    public void setName(String name);
    public String getName();

    public void setIdentity(Identity id);
    public Identity getIdentity();

    public void setType(int type);
    public int getType();

    public void setProvider(String provider);
    public String getProvider();

    public void setSpecification(Specification specification);
    public Specification getSpecification();

    public void setSecurityContext(SecurityContext security);
    public SecurityContext getSecurityContext();

    public void setServiceContact(ServiceContact servicecontact);
    public ServiceContact getServiceContact();

    public void setStdOutput(String output);
    public String getStdOutput();

    public void setStdError(String error);
    public String getStdError();

    public void setStatus(Status status);
    public void setStatus(int status);
    public Status getStatus();

    public void setAttribute(String name, Object value);
    public Object getAttribute(String name);

    public void addStatusListener(StatusListener listener);
    public void removeStatusListener(StatusListener listener);

    public void addOutputListener(OutputListener listener);
    public void removeOutputListener(OutputListener listener);

    public void fromXML(String task);
    public String toXML();

    public void fromString(String task);
    public String toString();

    public boolean isUnsubmitted();
    public boolean isActive();
    public boolean isCompleted();
    public boolean isSuspended();
    public boolean isFailed();
    public boolean isCanceled();
    public Calendar getSubmittedTime();
    public Calendar getCompletedTime();
}

				
// This program is distributed under the following licence:
// http://www.globus.org/toolkit/download/license.html
package org.globus.cog.core.interfaces;

public interface Specification
{
  public static final int JOB_SUBMISSION = 1;
  public static final int FILE_TRANSFER = 2;
  public static final int INFORMATION_QUERY = 3;

  public void setType(int type);
  public int getType();

public void setSpecification(String specification);
public String getSpecification();
}

				
// This program is distributed under the following license:
// http://www.globus.org/toolkit/download/license.html
package org.globus.cog.core.interfaces;

import java.util.Enumeration;

public interface JobSpecification extends Specification
{
  public void setExecutable(String executable);
  public String getExecutable();

  public void setDirectory(String directory);
  public String getDirectory();

  public void setEnvironment(String environment);
  public String getEnvironment();

  public void setArguments(String arguments);
  public String getArguments();

  public void setStdOutput(String output);
  public String getStdOutput();

  public void setStdInput(String input);
  public String getStdInput();

  public void setStdError(String error);
  public String getStdError();
    
  public void setCount(int count);
  public Integer getCount();

  public void setBatchJob(boolean bool);
  public boolean isBatchJob();

  public void setRedirected(boolean bool);
  public boolean isRedirected();

  public void setLocalExecutable(boolean bool);
  public boolean isLocalExecutable();

  public void setAttribute(String name, String value);
  public String getAttribute(String name);
  public Enumeration getAllAttributes();
}

				
// This program is distributed under the following license:
// http://www.globus.org/toolkit/download/license.html

package org.globus.cog.core.interfaces;

public interface FileSpecification extends Specification {
  public void setSourceServer(String server);
  public String getSourceServer();
  public void setDestinationServer(String server);
  public String getDestinationServer();
  public void setSourceDirectory(String directory);
  public String getSourceDirectory();
  public void setDestinationDirectory(String directory);
  public String getDestinationDirectory();
  public void setSourceFile(String file);
  public String getSourceFile();
  public void setDestinationFile(String file);
  public String getDestinationFile();
  public void setSource(String source);
  public String getSource();
  public void setDestination(String destination);
  public String getDestination();
  public void setBinary(boolean bool);
  public boolean isBinary();
  public void setNotpt(boolean bool);
  public boolean isNotpt();
  public void setDcau(boolean bool);
  public boolean isDcau();
  public void setBlockSize(int size);
  public int getBlockSize();
  public void setTcpBufferSize(int size);
  public int getTcpBufferSize();
  public void setParallelStreams(int value);
  public int getParallelStreams();
  public void setThirdParty(boolean bool);
  public boolean isThirdParty();
  public void setAttribute(String name, Object value);
  public Object getAttribute(String name);
}

				
// This program is distributed under the following license:
// http://www.globus.org/toolkit/download/license.html

package org.globus.cog.core.interfaces;

import java.util.Enumeration;

import org.globus.cog.core.impl.common.ActiveTaskException;
import org.globus.cog.core.impl.common.IllegalSpecException;
import org.globus.cog.core.impl.common.InvalidSecurityContextException;
import org.globus.cog.core.impl.common.InvalidServiceContactException;
import org.globus.cog.core.impl.common.TaskSubmissionException;

  public interface TaskHandler {
  public static final int GENERIC = 1;
  public static final int GT2 = 2;
  public static final int GT3 = 3;

  public void setType(int type);
  public int getType();

  public void submit(Task task) 
    throws IllegalSpecException,
           InvalidSecurityContextException,
           InvalidServiceContactException,
           TaskSubmissionException;

  public void suspend(Task task)
    throws InvalidSecurityContextException, TaskSubmissionException;

  public void resume(Task task)
    throws InvalidSecurityContextException, TaskSubmissionException;

  public void cancel(Task task)
    throws InvalidSecurityContextException, TaskSubmissionException;

  public void remove(Task task) throws ActiveTaskException;

  public Task[] getAllTasks();
  public Enumeration getActiveTasks();
  public Enumeration getFailedTasks();
  public Enumeration getCompletedTasks();
  public Enumeration getSuspendedTasks();
  public Enumeration getResumedTasks();
  public Enumeration getCanceledTasks();
}

				
// This program is distributed under the following license:
// http://www.globus.org/toolkit/download/license.html

package org.globus.cog.core.interfaces;

import java.util.Calendar;

  public interface Status {
  public static final int UNSUBMITTED = 0;
  public static final int SUBMITTED = 1;
  public static final int ACTIVE = 2;
  public static final int SUSPENDED = 3;
  public static final int RESUMED = 4;
  public static final int FAILED = 5;
  public static final int CANCELED = 6;
  public static final int COMPLETED = 7;

  public abstract void setStatus(int status);
  public abstract int getStatus();
  public abstract void setPrevStatus(int status);
  public abstract int getPrevStatus();
  public abstract void setException(Exception exception);
  public abstract Exception getException();
  public abstract void setMessage(String message);
  public abstract String getMessage();
  public void setTime(Calendar time);
  public Calendar getTime();
}

				
package org.globus.cog.core.examples.ssh;

import org.globus.cog.core.impl.common.StatusEvent;
import org.globus.cog.core.impl.common.task.IllegalSpecException;
import org.globus.cog.core.impl.common.task.InvalidSecurityContextException;
import org.globus.cog.core.impl.common.task.InvalidServiceContactException;
import org.globus.cog.core.impl.common.task.JobSpecificationImpl;
import org.globus.cog.core.impl.common.task.SecurityContextImpl;
import org.globus.cog.core.impl.common.task.ServiceContactImpl;
import org.globus.cog.core.impl.common.task.TaskImpl;
import org.globus.cog.core.impl.common.task.TaskSubmissionException;
import org.globus.cog.core.impl.ssh.TaskHandlerImpl;
import org.globus.cog.core.interfaces.JobSpecification;
import org.globus.cog.core.interfaces.ServiceContact;
import org.globus.cog.core.interfaces.Status;
import org.globus.cog.core.interfaces.StatusListener;
import org.globus.cog.core.interfaces.Task;
import org.globus.cog.core.interfaces.TaskHandler;

public class JobSubmission implements StatusListener {
  private Task task;

  public JobSubmission() {
     prepareTask();
     submitTask();

     boolean done = false;
     while (!done) {
        //check status every seconds
        try {
           Thread.sleep(1000);
        }
        catch (Exception e) {
           e.printStackStrace();
        }
        String status = task.getStatus().getStatusString();
        if (status.equalsIgnoreCase("completed")) {
           System.out.println("ssh stdout: " + task.getStdOutput());
           done = true;
        }
        else if (status.equalsIgnoreCase("failed")) {
           System.out.println(ssh sterr: " + task.getStdError());
           done = true;
        }
     }
  }

  private void prepareTask() {
     this.task = new TaskImpl("mySSHTask", Task.JOB_SUBMISSION);
     JobSpecification spec = new JobSpecificationImpl();
     spec.setExecutable("/bin/ls -al");
     spec.setRedirected(true);
     this.task.setSpecification(spec);

     SecurityContextImpl securityContext = new SecurityContextImpl();
     securityContext.setAttribute("ssh-username", "valid user name");
     securityContext.setAttribute("ssh-password", "valid user password");
     this.task.setSecurityContext(securityContext);
     ServiceContact service = new ServiceContactImpl();
     service.setIP("a valid ssh server IP address");
     this.task.setServiceContact(service);
     this.task.addStatusListener(this);
  }

  private void submitTask() {
     TaskHandler handler = new TaskHandlerImpl();
     try {
        handler.submit(this.task);
     }
     catch (InvalidSecurityContextException ise) {
        ise.printStackTrace();
        System.exit(1);
     }
     catch (TaskSubmissionException tse) {
        tse.printStackTrace();
        System.exit(1);
     }
     catch (IllegalSpecException ispe) {
        ispe.printStackTrace();
        System.exit(1);
     }
     catch (InvalidServiceContactException isce) {
        isce.printStackTrace();
        System.exit(1);
     }
  }

  public void statusChanged(StatusEvent event) {
     Status status = event.getStatus();
     System.out.println("Status changed to " + status.getStatusString());
  }

  public static void main(String arg[]) {
     new JobSubmission();
  }
}

				
/****************************************************
 * 
 * Submit O G C E job.
 * 
 ****************************************************/
private void submitOgceJob(JobInfo job)
	throws
		IllegalSpecException,
		InvalidSecurityContextException,
		InvalidServiceContactException,
		TaskSubmissionException {

	T.logEntry();

	TaskHandler taskHandler = new GenericTaskHandler();
	JobSubmitter submitter = new JobSubmitter();

	//Retrieve target resource type from our database (could be GT2/GT3)
	GridResource resource =
		Db.getResourceInfoByName(job.getTargetResource());
	if (resource == null) {
		T.log('e', "RESOURCE: " + job.getTargetResource() + " NOT FOUND");
		return;
	}
	job.setTargetResourceType(resource.getType());

	// Create OGCE task 		
	Task task = submitter.createTask(user, job);

	//Save job id in our JobInfo object
	long id = task.getIdentity().getValue();
	job.setGlobusJobID(String.valueOf(id));

	// Save the job submit time in our JobInfo object
	String submitTime = String.valueOf(System.currentTimeMillis());
	job.setStartTime(submitTime);

	// Submit the job
	T.log('i', "Submitting job");
	submitter.submitTask(task, taskHandler);

	// Get the status just after submit
	String status = subDb.getJobStatus(job.getGlobusJobID());
	job.setStatus(status);

	T.logExit();
	return;
}

GridResource resource =
	Db.getResourceInfoByName(job.getTargetResource());

	Task task = submitter.createTask(user, job);
	submitter.submitTask(task, taskHandler);

				
package com.ibm.gdc.gt3.jobs;

/**
 * Submit a job to a Grid resource using OGCE classes
 * provided by the Java CogKit 2.0
 * @author Montpellier Pssc
 */
public class JobSubmitter implements StatusListener {

   /**
    * Create Task
    */
   public Task createTask(ISDGUser user, JobInfo job) {
      1. Create a JobSubmission task object
      2. According to the target resource type, call 
      prepareGT2Task() or prepareGT3Task() method 
      to set task specification.
      3. Return the created task object
   }

   /**
    * Set GT3 Task specifications
    */
   private void prepareGT3Task(Task task, JobInfo job, ISDGUser user) {
      1. Create a TaskSpecification object
      2. Set Task specification values like executable name and parameters.
      3. Set GT3 Task security context
      4. Set GT3 Task contact (MasterForkManagedJob grid service reference)
      5. Add Task status listener
   }

   /**
    * Set GT2 Task specifications
    */
   private void prepareGT2Task(Task task, JobInfo job, ISDGUser user) {
      1. Create a TaskSpecification object
      2. Set Task specification values like executable name and parameters.
      3. Set GT2 Task security context
      4. Set GT2 Task contact (Target resource host name)
      5. Add Task status listener
   }

   /**
    * Submit GT2/GT3 task
    */
   public void submitTask(Task task, TaskHandler handler) {
      1. Use the TaskHandler to submit the job (This is common to GT2 and GT3)
      2. Handle exceptions.
   }

   /**
    * Convert GlobusCredential to GSSCredential
    */
   private GSSCredential globus2gssCredential(ISDGUser user) {
      1. Utility method to convert credential from GT2 to GT3 format
      2. Return GT3 credential (GSSCredential object)
   }

   /**
    * Task status listener
    * (This method is mandatory because the JobSubmitter class implements StatusListener.)
    */
   public void statusChanged(StatusEvent event) {
      1. Receive the new status event and retrieve the calling Task.
      2. Save the job status in the database
      3. If the status is COMPLETED, retrieve the job 
      output from the Task and save it in the database.
   }

}

public Task createTask(ISDGUser user, JobInfo job) {
   T.logEntry();

   // Create a job submission task object
   Task task = new TaskImpl(job.getJobName(), Task.JOB_SUBMISSION);

   if (job.getTargetResourceType().equalsIgnoreCase("GT2")) {
      prepareGT2Task(task, job, user);
   }
   else if (job.getTargetResourceType().equalsIgnoreCase("GT3")) {
      prepareGT3Task(task, job, user);
   }
   else {
      task = null;
      T.log('e', "Invalid Task Provider: " + job.getTargetResourceType());
   }
   T.logExit();
   return task;
}

				
/**
 * GT3 Task initialization
 * Prepare task using received job and user data.
 */
private void prepareGT3Task(Task task, JobInfo job, ISDGUser user) {
   T.logEntry();
   
   // Set the service provider type for this task. This is used at
   // submission time by the generic task handler to determine the class 
   // name of the TaskHandler to be loaded.
   task.setProvider("GT3");
   
   // Create a Specification object for the task
   JobSpecification spec = new JobSpecificationImpl();
   
   T.log('d', "Setting GT3 task specification");
   // Specify the command to be executed on the Grid resource.
   // The received job object contains all job information
   // entered by the user.
   spec.setExecutable(job.getCmdName());
   T.log('d', "Task executable: [" + spec.getExecutable() + "]");
   
   // Set location of the executable.
   // If the executable resides on the client machine, it will be 
   // automatically staged from the client machine to the Grid resource.
   if (job.getStageExecutable() != null) {
      if (job.getStageExecutable().equalsIgnoreCase("true")) {
         spec.setLocalExecutable(true);
       }
   }
   
   // Set the arguments (if any) for the executable.
   if (job.getCmdArgs() != null && job.getCmdArgs().length() > 0) {
      spec.setArguments(job.getCmdArgs());
      T.log('d', "Task arguments: [" + spec.getArguments() + "]");
   }
   // Specify the Grid resource working directory in which
   // the command is to be executed.
   if (job.getCmdDir() != null && job.getCmdDir().length()> 0) {
      spec.setDirectory(job.getCmdDir());
      T.log('d', "Task directory: [" + spec.getDirectory() + "]");
   }
   
   // Specify environment variable to be set prior to the
   // execution of the command
   if (job.getCmdEnv() != null && job.getCmdEnv().length() > 0) {
      spec.setEnvironment(job.getCmdEnv());
      T.log('d', "Task environment: [" + spec.getEnvironment() + "]");
   }
   
   // Specify the name of the file from which the input must be retrieved.
   if (job.getStdIn() != null) {
      if (job.getStdIn().length() > 0) {
         spec.setStdInput(job.getStdIn());
      }
   }
   
   // Specify the name of the file where the output must be redirected to. 
   // If left blank or not specified, the output is streamed to the 
   // standard output.
   if (job.getStdOut() != null) {
      if (job.getStdOut().length() > 0) {
         spec.setStdOutput(job.getStdOut());
      }
   }
   
   // Specify the name of the file where the error messages must be 
   // redirected to. If left blank, the errors are streamed to the 
   // standard error.
   if (job.getStdErr() != null) {
      if (job.getStdErr().length() > 0) {
         spec.setStdError(job.getStdErr());
      }
   }
   
   //Specify if the output and error streams are to be redirected
   //to the client. (This field is set only if both stdOut and stdErr
   // have not been set by the user)
   if ((spec.getStdOutput() == null || spec.getStdOutput().length() == 0)) {
      spec.setRedirected(true);
   }
   else {
      spec.setRedirected(false);
   }
   
   // Specify if the remote execution is to occur in batch mode.
   // If true, the client will not be notified of any status changes 
   // on the server side. It is the responsibility of the client to 
   // obtain the final output from the server by some
   // asynchronous mechanism. 
   // If false, the client will interactively receive status notifications
   // from the server as well as the final output or error.
   spec.setBatchJob(
       (job.getBatch().equalsIgnoreCase("true")) ? true : false);
   T.log('d', "Task batch: [" + spec.isBatchJob() + "]");
   if (spec.isBatchJob()) {
      spec.setRedirected(false);
   }
   T.log('d', "Task redirected: [" + spec.isRedirected() + "]");
   
   // Attach above specifications to the task
   task.setSpecification(spec);
   
   // Convert GlobusCredential (the user proxy) to GSSCredential
   // The user proxy is part of the received ISDGUser object.
   T.log('d', "Converting user proxy to GSSCredential");
   GSSCredential cred = globus2gssCredential(user);
   
   // Set the Task security context
   T.log('d', "Setting GT3 task security context");
   org.globus.cog.core.impl.gt3.GlobusSecurityContextImpl securityContext =
   	new org.globus.cog.core.impl.gt3.GlobusSecurityContextImpl();
   securityContext.setCredentials(cred);
   task.setSecurityContext(securityContext);
   
   // Set the task contact. In GT3, it is the URL of the MasterForkJobFactoryService
   T.log('d', "Setting GT3 task contact");
   String serviceName =
        job.getTargetResource()
                + ":"
                + 8080
                + "/ogsa/services/base/gram/MasterForkManagedJobFactoryService";
   
   ServiceContact service = new ServiceContactImpl(serviceName);
   task.setServiceContact(service);
   
   // Add the status listener. The JobSubmitter itself (this) is specified as
   // status listener because it provides a statusChanged() method.
   T.log('d', "Adding  GT3 task status listener");
   task.addStatusListener(this);
   T.logExit();
  }

				
1. The provider name must be changed in order 
to use the right TaskHandler class at submission time.

   // Set the service provider type for this task
   task.setProvider("GT2");

2. The security context has a different type than GT3.

   // Set the task security context
   org.globus.cog.core.impl.gt2.GlobusSecurityContextImpl securityContext =
             new org.globus.cog.core.impl.gt2.GlobusSecurityContextImpl();
   securityContext.setCredentials(cred);
   task.setSecurityContext(securityContext);

3. With GT2, the contact is the host name of 
the target resource that the job is submitted to. 
This information exists in the JobInfo object (job) received as parameter. 

   // Set the task contact
   String serviceName = job.getTargetResource();
   ServiceContact service = new ServiceContactImpl(serviceName);
   task.setServiceContact(service);

4. All other task specification, executable name, 
parameters, etc. are set the same way as for GT3.

				
/**
 * Submit GT2/GT3 task
 */
public void submitTask(Task task, TaskHandler handler) {
   T.logEntry();
   try {
      T.log('d', "Calling Taskhandler to submit the task");
      handler.submit(task);
   }
   catch (InvalidSecurityContextException ise) {
      T.log('d', "Security Exception");
      T.logException('e', ise);
   }
   catch (TaskSubmissionException tse) {
      T.log('d', "TaskSubmission Exception");
      T.logException('e', tse);
   }
   catch (IllegalSpecException ispe) {
      T.log('d', "Specification Exception");
      T.logException('e', ispe);
   }
   catch (InvalidServiceContactException isce) {
      T.log('d', "Service Contact Exception");
      T.logException('e', isce);
   }
   catch (Exception ex) {
      T.log('d', "Exception while submitting job");
      T.logException('e', ex);
   }
   T.logExit();
   }

				
/**
 * 
 * Task status listener
 * 
 */
 public void statusChanged(StatusEvent event) {
   T.logEntry();
   Status status = event.getStatus();
   Task task = event.getSource();
   long id = task.getIdentity().getValue();
   switch (status.getStatus()) {
      case 0 :
         T.log('d', "Status changed to " + Const.UNSUBMITTED);
         Db.setJobStatus(String.valueOf(id), Const.UNSUBMITTED);
      break;
      case 1 :
         T.log('d', "Status changed to " + Const.SUBMITTED);
         Db.setJobStatus(String.valueOf(id), Const.SUBMITTED);
         break;
      case 2 :
         T.log('d', "Status changed to " + Const.ACTIVE);
         Db.setJobStatus(String.valueOf(id), Const.ACTIVE);
         break;
      case 3 :
         T.log('d', "Status changed to " + Const.SUSPENDED);
         Db.setJobStatus(String.valueOf(id), Const.SUSPENDED);
         break;
      case 4 :
         T.log('d', "Status changed to " + Const.RESUMED);
         Db.setJobStatus(String.valueOf(id), Const.RESUMED);
         break;
      case 5 :
         T.log('d', "Status changed to " + Const.FAILED);
         Db.setJobStatus(String.valueOf(id), Const.FAILED);
         break;
      case 6 :
         T.log('d', "Status changed to " + Const.CANCELED);
         Db.setJobStatus(String.valueOf(id), Const.CANCELED);
         break;
      case 7 :
         T.log('d', "Status changed to COMPLETED");
         Db.setJobStatus(String.valueOf(id), Const.COMPLETED);
         if (task.getStdOutput() != null)
            if (task.getStdOutput().length() > 0)
               Db.setJobOutput(String.valueOf(id), task.getStdOutput());
         break;
      default :
         T.log('d', "Status changed to INVALID VALUE !!");
         break;
   }
   T.logExit();
}