job.java

这是一个基于计算网格的web service。它用java编写。一旦安装完成

/*
 * JobImpl.java
 *
 * Created on April 11, 2005, 9:25 AM
 */

package jwsgrid.jobhost.priv;


import jwsgrid.xsd.jobdescription.*;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;

import java.lang.Thread;
import java.lang.Process;
import java.lang.ProcessBuilder;

import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.util.Stack;


/**
 *
 * @author sean
 */
public class Job implements Runnable 
{
    
    ////////////////////////////////////////////////////////////////////////////
    // public attributes
    //  

    public static final String[] JOB_TYPES = 
    {
        "windows/executable",
        "windows/command",
        "windows/batch",
        "java/jar",
        "java/class",
        "linux/shell", 
        "linux/binary",
        "solaris/shell",
        "solaris/binary"
    };     
    
    //
    // these should be revamped and moved into JobHost
    //
    public static final int STATUS_WAITING = 10;
    public static final int STATUS_RUNNING = 11; 
    public static final int STATUS_STOPPED = 12;
    public static final int STATUS_ABORTED = 13;
    public static final int STATUS_COMPLETE = 14;
    
    public static final String FILE_SEP = System.getProperty( "file.separator" );  
    
    private static final int SLEEP_INTERVAL = 50;
    private static final int DATA_BLOCK_SIZE = 512;
    
    ////////////////////////////////////////////////////////////////////////////
    // private attributes
    //      
    
    private boolean stopRequest = false;
    private ProcessBuilder procBuilder = null;
    private Process jobProc = null;
    
    private String id = null;
    private File deployDir = null;   
    private String type = null;
    private String execFileName = null;        
    private int status = 0;    
    private int exitValue = 0;    
    private List<String> inputFileList = null;
    private Exception jobException = null;
    
    ////////////////////////////////////////////////////////////////////////////
    // public methods
    //      
    
    public Job( 
            String id,
            File deployDir,  
            String type,
            String execFileName, 
            List<String> inputArgList,
            List<String> inputFileList,
            List<EnvVarType> envVarList ) 
    throws Exception {
        Map<String,String> procBuildEnv = null;
        Vector<String> procArgList = new Vector();
        
        this.id = id;
        this.type = type;
        this.deployDir = deployDir;
        this.inputFileList = inputFileList;
        
        if ( execFileName.startsWith( FILE_SEP ) ) {
            this.execFileName = execFileName.substring( 1, execFileName.length() );
            
        } else {
            this.execFileName = execFileName;            
        }
        
        File file = new File( deployDir.getPath().toString() + FILE_SEP + execFileName );
        
        if ( !file.exists()) {
            status = STATUS_ABORTED;
            jobException = new Exception( 
                    "execution file '" + file.getPath().toString() + "' " +
                    "does not exist" );
            throw jobException;
        }
        
        if ( file.isDirectory() ) {
            status = STATUS_ABORTED;            
            jobException = new Exception( 
                    "execution file '" + file.getPath().toString() + "' " +
                    "is a directory" );  
            throw jobException;
        }  
        
        // check for java job -- these are run in seperate JVMs
        if ( type.compareTo( "java/jar" ) == 0 ||
                type.compareTo( "java/class" ) == 0 ) {
            String jdkHome = System.getProperty( "java.home" );
                        
            if ( jdkHome == null ) {
                status = STATUS_ABORTED;
                jobException = new Exception( 
                        "could not determine java installation directory" );
                throw jobException;
            }
            
            String javaBin = jdkHome + FILE_SEP + "bin" + FILE_SEP + "java";
            
            file = new File( javaBin );            
            if ( !file.exists() || file.isDirectory() ) {
                
                javaBin = jdkHome + FILE_SEP + "bin" + FILE_SEP + "java.exe";
                file = new File( javaBin );            
                if ( !file.exists() || file.isDirectory() ) {                
                    status = STATUS_ABORTED;
                    jobException = new Exception( 
                            "could not determine path to java binary" ); 
                    throw jobException;
                }
            }
            
            // add java binary to process builder args
            procArgList.add( javaBin );
            
            if ( type.compareTo( "java/jar" ) == 0 ) {
                procArgList.add( "-jar" );                
            } 
        }
        
        // add main executable to process builder args
        procArgList.add( deployDir.getPath().toString() + FILE_SEP + execFileName );
        
        // set input args
        for ( int i = 0; i < inputArgList.size(); i++ ) {
            procArgList.add( inputArgList.get( i ) );
        }      
        
        // convert vector to an array
        String[] procArgArr = new String[ procArgList.size() ];
        for ( int i = 0; i < procArgList.size(); i++ ) {
            procArgArr[ i ] = procArgList.get( i );
        }
        
        // remove FILE_SEP from start of input files, if necessary
        for ( int i = 0; i < this.inputFileList.size(); i++ ) {
            String path = this.inputFileList.get( i );
            if ( path.startsWith( FILE_SEP ) ) {
                path = path.substring( 1, path.length() );
                this.inputFileList.set( i, path );
            }
        }
        
        // create process builder with given args
        this.procBuilder = new ProcessBuilder( procArgArr );
        
        // setup environment vars
        if ( envVarList != null ) {
            procBuildEnv = this.procBuilder.environment();
            
            for ( int i = 0; i < envVarList.size(); i++ ) {
                EnvVarType env = envVarList.get( i );
                procBuildEnv.put( env.getName(), env.getValue() );
            }
        }
        
        // set working directory
        this.procBuilder.directory( deployDir );
        
        // set status to NEW
        this.status = STATUS_WAITING;        
    }
    
    public String getId() {
        return id;
    }
    
    public int getStatus() {
        return status;
    }   
    
    public int getExitValue() {
        return exitValue;
    }
    
    public Exception getException() {
        return jobException;
    }   
    
    public void stop() {   
        stopRequest = true;

        // terminate process
        if ( jobProc != null ) {
            jobProc.destroy();
            jobProc = null;
        }
    } 
    
    public void run() {
        BufferedInputStream inputStreamIn = null;
        BufferedInputStream errStreamIn = null;
        BufferedOutputStream inputStreamOut = null;
        BufferedOutputStream errStreamOut = null;
        byte[] buf = new byte[ DATA_BLOCK_SIZE ];

        try {
            if ( stopRequest ) {
                status = STATUS_STOPPED;
                return;
            }
            
            jobProc = procBuilder.start();
            status = STATUS_RUNNING;
            
            inputStreamIn = 
                    new BufferedInputStream( jobProc.getInputStream() );

            errStreamIn = 
                    new BufferedInputStream( jobProc.getErrorStream() );  
            
            
            File stdinFile = File.createTempFile( "stdout-", ".out", deployDir );
            File stderrFile = File.createTempFile( "stderr-", ".out", deployDir );         
                        
            inputStreamOut =
                    new BufferedOutputStream( 
                        new FileOutputStream( stdinFile ) );
            
            errStreamOut =
                    new BufferedOutputStream( 
                        new FileOutputStream( stderrFile ) );            
            
            while( !stopRequest ) {
                if ( inputStreamIn.available() > 0 ) {
                    int bCnt = inputStreamIn.read( buf, 0, DATA_BLOCK_SIZE );
                    inputStreamOut.write( buf, 0, bCnt );
                }
                
                if ( errStreamIn.available() > 0 ) {
                    int bCnt = errStreamIn.read( buf, 0, DATA_BLOCK_SIZE );
                    errStreamOut.write( buf, 0, bCnt );
                }                
                
                try {
                    // try to capture process's exit value, if process
                    // is still running, an exception will be thrown...
                    // catch it and continue, but if the process is
                    // complete, break out of this loop
                    exitValue = jobProc.exitValue();
                    
                } catch ( IllegalThreadStateException itse ) {
                    try {
                        Thread.sleep( SLEEP_INTERVAL );
                        
                    } catch ( InterruptedException interruptEx ) {}
                    
                    continue;
                }
                
                break;
            }
            
            if ( stopRequest ) {
                status = STATUS_STOPPED;
                
            } else {
                status = STATUS_COMPLETE;
            } 
            
        } catch ( IOException ioEx ) {
            // killing the process can cause a bad file descriptor
            // while trying to read/write stdout/stderr; ignore   
            // unless this isn't a stop request
            if ( !stopRequest ) {
                jobException = ioEx;
                status = STATUS_ABORTED;
            }
            
        } catch ( Exception ex ) {
            jobException = ex;
            status = STATUS_ABORTED;
                
        } finally {
            try {
                if ( inputStreamOut != null ) {
                    inputStreamOut.flush();
                    inputStreamOut.close();
                    inputStreamOut = null;
                }
                
                if ( errStreamOut != null ) {
                    errStreamOut.flush();
                    errStreamOut.close();
                    errStreamOut = null;
                }                
                
                if ( inputStreamIn != null ) {
                    inputStreamIn.close();
                    inputStreamIn = null;
                }
                
                if ( errStreamIn != null ) {
                    errStreamIn.close();
                    errStreamIn = null;
                }
                
            } catch ( Exception ex ) {}
        }         
    } 
    
    public List<String> generatedOutputFileList() throws Exception {
        JobOuputFileFiler filter = new JobOuputFileFiler();
        Vector<String> outputFileList = new Vector();
        String deployDirName = deployDir.getPath().toString() + FILE_SEP;
        File file = null;
        
        // add executable file to filter list
        file = new File( deployDirName + execFileName );
        filter.add( file );
        
        // add input files to filter list
        for ( int i = 0; i < inputFileList.size(); i++ ) {
            file = new File( deployDirName + inputFileList.get( i ) );
            filter.add( file );            
        } 
        
        Stack dirStack = new Stack();
        dirStack.push( deployDir );
                        
        while ( !dirStack.isEmpty() ) {
            // enter directory
            File dir = (File) dirStack.pop();
            File[] contents = dir.listFiles();
            
            // push all sub-directories onto stack
            for( int i = 0; i < contents.length; i++ ) {
                if ( contents[ i ].isDirectory() && 
                        contents[ i ].compareTo( dir ) != 0 ) {
                    dirStack.push( contents[ i ] );
                }
            }
            
            // extract files that don't match filter
            contents = dir.listFiles( filter );
            for( int i = 0; i < contents.length; i++ ) {
                if ( contents[ i ].isFile() ) {
                    outputFileList.add( contents[ i ].getPath() );
                }
            }
        }

        return outputFileList;
    }          
}