statstrackerthread.java

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

/*
 * StatsTrackerThread.java
 *
 * Created on May 10, 2005, 11:24 AM
 */

package jwsgrid.resourcemanager.priv;

import jwsgrid.resourcemanager.ResourceManager;

import jwsgrid.util.Log;

import jwsgrid.xsd.*;
import jwsgrid.xsd.jobhostdynamicinfo.*;

import jwsgrid.wsdl.Proxy;

import jwsgrid.ws.jobhost.JobHostWSSEI;

import java.sql.Connection;
import java.sql.Timestamp;

import java.util.Calendar;
import java.util.List;

/**
 *
 * @author sean
 */
public class StatsTrackerThread implements Runnable
{    
    ////////////////////////////////////////////////////////////////////////////
    // private attributes
    //   
    
    private static final int MS_PER_SEC = 1000;
    private static final int MS_PER_MIN = MS_PER_SEC * 60;
    private static final int MS_PER_HOUR = MS_PER_MIN * 60;   
    
    private ResourceManager resMgr = null;
    private boolean shutdownRequest = false;
    
    ////////////////////////////////////////////////////////////////////////////
    // public methods
    //      
    
    /** Creates a new instance of StatsTrackerThread */
    public StatsTrackerThread( ResourceManager resMgr ) 
    {
        this.resMgr = resMgr;
    }
    
    public ResourceManager getResMgr() 
    {
        return resMgr;
    }    
    
    public void stop() 
    {
        shutdownRequest = true;
    }    
    
    public void run()
    {
        getResMgr().logMessage( 
                Log.MSG_PRI_MAX, 
                "job host statistics tracker thread started" ); 
        
        // set job host statistics reset timestamp
        Timestamp resetTs = new Timestamp( 
                Calendar.getInstance().getTimeInMillis() + (MS_PER_HOUR * 24) );          
        
        
        // main loop, continue until shutdown is requested and all
        // threads are stopped
        while ( !shutdownRequest )
        {
            try
            {            
                //
                // get a list of all registered job hosts
                //
                List<LdapJobHost> jobHostList = null;
                try
                {
                    jobHostList = Ldap.getJobHosts();

                } catch ( Exception ex ) {
                    getResMgr().logMessage( 
                            Log.MSG_PRI_MAX, 
                            "could not obtain a list of job hosts, the following " +
                            "LDAP error occured: " + ex.getMessage() );

                    continue;
                }

                //
                // update job host dynamic information
                //            
                for ( int i = 0; i < jobHostList.size() && !shutdownRequest; i++ )
                {
                    DynamicJobHostInfo dynamicInfo = null;
                    List<DynamicCpuInfoType> dynCpuInfoList = null;        
                    List<DynamicFsInfoType> dynFsInfoList = null;                
                    JobHostWSSEI jobHostProxy = null;
                    String xmlDynamicInfoDoc = null;

                    getResMgr().logMessage( 
                            Log.MSG_PRI_MIN, 
                            "updating dynamic job host information for '" +
                            jobHostList.get( i ).wsaddr + "'" );                

                    // get a handle on the job host and retrieve its dynamic info
                    try
                    {
                        String jobHostWsAddr = jobHostList.get( i ).wsaddr;

                        if ( jobHostWsAddr.endsWith( "/" ) )
                        {
                            jobHostWsAddr = jobHostWsAddr.substring( 
                                    0, 
                                    jobHostWsAddr.length() - 1 );
                        }                       
                        
                        jobHostProxy = Proxy.getJobHostProxy( jobHostWsAddr );

                        xmlDynamicInfoDoc = jobHostProxy.getDynamicInfo();

                    } catch ( Exception ex ) {
                        // could not contact the job host, remove it from registry
                        getResMgr().logMessage( 
                                Log.MSG_PRI_MAX, 
                                "job host '" + jobHostList.get( i ).wsaddr + 
                                "' could not be contacted; removing it from " +
                                "the registry: " + ex.getMessage() );                     
                        try
                        {
                            Ldap.unregisterJobHost( 
                                    jobHostList.get( i ).wsaddr, 
                                    jobHostList.get( i ).clusterId );

                        } catch ( Exception e ) {}
                        
                        continue;
                    }

                    // create a dynamic host info object from the document returned
                    try 
                    {
                        dynamicInfo = JobHost.createDynamicInfoObj( 
                                xmlDynamicInfoDoc );

                    } catch ( Exception ex ) {
                        // invalid object, remove job host from registry
                        getResMgr().logMessage( 
                                Log.MSG_PRI_MAX, 
                                "job host '" + jobHostList.get( i ).wsaddr + 
                                "' returned an invalid dynamic information " +
                                "document; removing it from the registry" ); 
                        try
                        {
                            Ldap.unregisterJobHost( 
                                    jobHostList.get( i ).wsaddr, 
                                    jobHostList.get( i ).clusterId );

                        } catch ( Exception e ) {}  
                        
                        continue;
                    }  

                    // extract some of the multi-valued job host information
                    dynCpuInfoList = dynamicInfo.getCpuInfo();
                    dynFsInfoList = dynamicInfo.getFsInfo();

                    Integer cpuUserTime = new Integer( 
                            dynCpuInfoList.get( 0 ).getUserTime() );
                    Integer cpuIdleTime = new Integer( 
                            dynCpuInfoList.get( 0 ).getIdleTime() );
                    Integer cpuSysTime = new Integer( 
                            dynCpuInfoList.get( 0 ).getSystemTime() );
                    Integer cpuLoad = new Integer( 
                            cpuUserTime.intValue() + cpuSysTime.intValue() );        

                    // update statistics
                    Connection conn = null;
                    try
                    {
                        conn = Sql.getConnection();

                        Sql.CpuLoadStats loadInfo = Sql.getCpuLoadStats( 
                                conn, jobHostList.get( i ).wsaddr );

                        if ( loadInfo == null )
                        {
                            Sql.insertCpuLoadStats( 
                                    conn, 
                                    jobHostList.get( i ).wsaddr, 
                                    cpuLoad.intValue() );

                            loadInfo = Sql.getCpuLoadStats( 
                                    conn, jobHostList.get( i ).wsaddr );
                        }

                        int loadTot = loadInfo.getTotal() + cpuLoad.intValue();
                        int updateCount = loadInfo.getUpdateCount() + 1;

                        Sql.updateCpuLoadStats( 
                                conn, 
                                jobHostList.get( i ).wsaddr , 
                                updateCount, 
                                loadTot );

                        cpuLoad = new Integer( loadTot / updateCount );
                        
                        Ldap.updateJobHostDynamicInfo(
                                jobHostList.get( i ).wsaddr,
                                jobHostList.get( i ).clusterId,
                                cpuUserTime,
                                cpuIdleTime,
                                cpuSysTime,
                                cpuLoad,
                                new Integer( dynamicInfo.getMemInfo().getFreeRam() ),                 
                                new Integer( dynFsInfoList.get( 0 ).getFreeSpace() ) );                        

                    } catch ( Exception ex ) {
                        getResMgr().logMessage( 
                                Log.MSG_PRI_MAX, 
                                "could not update dynamic job host information " +
                                "for '" + jobHostList.get( i ).wsaddr + 
                                "': " + ex.getMessage() ); 

                    } finally {
                        try
                        {
                            if ( conn != null )
                            {
                                conn.close();
                                conn = null;
                            }

                        } catch ( Exception ex ) {}
                    }   

                    Thread.sleep( 50 );                 
                }  

                // 
                // reset jobhost statistics tracking every 24 hours
                //

                Timestamp currTs = new Timestamp( 
                        Calendar.getInstance().getTimeInMillis() );

                if ( currTs.compareTo( resetTs ) >= 0 ) 
                {    
                    getResMgr().logMessage( 
                            Log.MSG_PRI_MAX, 
                            "re-setting job host statistics" ); 
        
                    resetTs = new Timestamp( 
                            Calendar.getInstance().getTimeInMillis() + 
                            MS_PER_MIN * (MS_PER_HOUR * 24) );   

                    // reset cpu tracking stats   
                    Connection conn = null;
                    try
                    {
                        for ( int i = 0; i < jobHostList.size() && !shutdownRequest; i++ )
                        {    

                            try
                            {
                                conn = Sql.getConnection();

                                Sql.CpuLoadStats loadInfo = Sql.getCpuLoadStats( 
                                        conn, jobHostList.get( i ).wsaddr );

                                if ( loadInfo == null )
                                {
                                    continue;
                                }

                                Sql.updateCpuLoadStats( 
                                        conn, 
                                        jobHostList.get( i ).wsaddr , 
                                        1, 
                                        loadInfo.getTotal() / loadInfo.getUpdateCount() );

                            } catch ( Exception ex ) {
                                getResMgr().logMessage( 
                                        Log.MSG_PRI_MAX, 
                                        "could not reset cpu tracking statistics for " +
                                        "'" + jobHostList.get( i ).wsaddr + 
                                        "': " + ex.getMessage() ); 
                            }
                        }

                    } finally {
                        try
                        {
                            if ( conn != null )
                            {
                                conn.close();
                                conn = null;
                            }

                        } catch ( Exception ex ) {}
                    }                                 
                }
                
                Thread.sleep( MS_PER_MIN * 5 );
                
            } catch ( InterruptedException ie ) {}
        } 
        
        getResMgr().logMessage( 
                Log.MSG_PRI_MAX, 
                "job host statistics tracker thread stopped" );         
    }
    
}