monitorgroup.java

分布式计算平台P2HP-1的源代码；P2HP-1是基于P2P的高性能计算平台

package cn.edu.hust.cgcl.biogrid.monitor;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.text.ParseException;
import java.util.Vector;

/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2004</p>
 * <p>Company: </p>
 * @author not attributable
 * @version 1.0
 */

public class MonitorGroup
{
    public final static int serverPort = 3456;

    private Vector monitorVec;
    private MonitorInqueryHandler inqueryHandler;
    private MonitorInfo monitorInfo;

    public MonitorGroup(MonitorInfo moninfo, MonitorNode firstnode)
    {
        monitorVec = new Vector();
        if (firstnode != null)
        {
            monitorVec.add(firstnode);
        } // if
        monitorInfo = moninfo;
        inqueryHandler = new MonitorInqueryHandler(serverPort, monitorInfo);
        inqueryHandler.start();
    } // MonitorGroup

    public MonitorNode getMonitor(int index)
    {
        return (MonitorNode) monitorVec.get(index);
    } // GetMonitor

    public MonitorNode getHeaviestWorkerMonitor()
        throws Exception
    {
        MonitorNode heaviestNode = null;
        synchronized (monitorVec)
        {
            int count = monitorVec.size();
            if (count < 1)
            {
                throw new ParseException("Protocol parse error!", 0);
            } // if
            int tmpload = 0;
            for (int i = 0; i < count; i++)
            {
                MonitorNode tmpnode = (MonitorNode) monitorVec.elementAt(i);
                if (tmpnode.getMonitorInfo().getWorkerLoad() > tmpload)
                {
                    tmpload = tmpnode.getMonitorInfo().getDispatcherLoad();
                    heaviestNode = tmpnode;
                } // if
            } // for
        } // synchronized
        if (heaviestNode == null)
        {
            throw new ParseException("Protocol parse error!", 0);
        } //if
        return heaviestNode;
    } // getHeaviestWorkerMonitor

    public MonitorNode getHeaviestDispMonitor()
        throws Exception
    {
        MonitorNode heaviestNode = null;
        synchronized (monitorVec)
        {
            int count = monitorVec.size();
            if (count < 1)
            {
                throw new ParseException("Protocol parse error!", 0);
            }
            int tmpload = 0;
            for (int i = 0; i < count; i++)
            {
                MonitorNode tmpnode = (MonitorNode) monitorVec.elementAt(i);
                if (tmpnode.getMonitorInfo().getDispatcherLoad() > tmpload)
                {
                    tmpload = tmpnode.getMonitorInfo().getDispatcherLoad();
                    heaviestNode = tmpnode;
                } // if
            } // for
        } // synchronized
        if (heaviestNode == null)
        {
            throw new ParseException("Protocol parse error!", 0);
        } //if
        return heaviestNode;
    } // getHeaviestDispMonitor

    public MonitorNode getLowestMonitorNode(Vector exceptIdVec)
        throws Exception
    {
        MonitorNode lowestNode = null;
        synchronized (monitorVec)
        {
            int count = monitorVec.size();
            if (count < 1)
            {
                throw new ParseException("Protocol parse error!", 0);
            }
            int tmpload = 100;
            for (int i = 0; i < count; i++)
            {
                MonitorNode tmpnode = (MonitorNode) monitorVec.elementAt(i);
                /*
                     MonitorInfo tmpinfo1 = tmpnode.getMonitorInfo();
                     MonitorInfo tmpinfo2 = lowestNode.getMonitorInfo();
                 */
                // 下面有问题！！！
                if (tmpnode.getMonitorInfo().getWorkerLoad() < tmpload)
                {
                    String tmpmonid = tmpnode.getMonConf().getMonitorId();
                    if (exceptIdVec != null)
                    {
                        int j = 0;
                        for (; j < exceptIdVec.size(); j++)
                        {
                            if (tmpmonid.equals( (String) exceptIdVec.elementAt(
                                j)))
                            {
                                break;
                            } // if
                        } // for
                        if (j < exceptIdVec.size())
                        {
                            continue;
                        } // if
                    } // if
                    tmpload = tmpnode.getMonitorInfo().getWorkerLoad();
                    lowestNode = tmpnode;
                } // if
            } // for
        } // synchronized
        if (lowestNode == null)
        {
            throw new ParseException("Protocol parse error!", 0);
        }
        return lowestNode;
    } // getLowestMonitorNode

    public int getMonitorCount()
    {
        return this.monitorVec.size();
    } // getMonitorCount

    public void getMonitorGroupInfo()
    {
    } // GetMonitorGroupInfo

    public void monitorJoinIn()
    {

    } // monitorJoinIn

} // MonitorGroup

class MonitorInqueryHandler
    extends Thread
{
    int serverPort;
    MonitorInfo monitorInfo;

    public MonitorInqueryHandler(int port, MonitorInfo monitorinfo)
    {
        monitorInfo = monitorinfo;
        serverPort = port;
    } // MonitorHandler

    public void run()
    {
        ServerSocket serverSock;
        Socket clientSock;
        try
        {
            serverSock = new ServerSocket(this.serverPort);
            while ( (clientSock = serverSock.accept()) != null)
            {
                MonitorPollHandler pollHandler = new MonitorPollHandler(
                    clientSock, monitorInfo);
                pollHandler.start();
            } // while
        } // try
        catch (Exception e)
        {
        } // catch
        clientSock = null;
        serverSock = null;

        return;
    } // run

    private void onInquery()
    {
        return;
    } // onInquery
    /*
         private void onPoll()
         {
        return;
         } // onPoll
     */
} // NodeServerHandler

class MonitorPollHandler
    extends Thread
{
    private static final int TIME_OUT = 3000;
    //Vector monitorVec;
    private Socket clientSock;
    private MonitorInfo monitorInfo;

    public MonitorPollHandler(Socket sock, MonitorInfo monitorinfo)
    {
        clientSock = sock;
        monitorInfo = monitorinfo;
    } // MonitorPollHandler

    public void run()
    {
        BufferedReader is=null;
        PrintWriter os=null;
        if (clientSock == null)
        {
            return;
        } // if
        try
        {
            clientSock.setSoTimeout(TIME_OUT);
            is = new BufferedReader(new InputStreamReader(clientSock.
                getInputStream()));
            os = new PrintWriter(clientSock.getOutputStream());
            while (true)
            {
                String lineStr = is.readLine();
                if (lineStr.equals("<monitor poll>"))
                {
                    os.println(monitorInfo.getMonitorId());
                    os.println(monitorInfo.getJobCount());
                    os.println(monitorInfo.getWorkerLoad());
                    os.println(monitorInfo.getDispatcherLoad());
                    os.println(monitorInfo.getDispatcherGroupCount());
                    os.println(monitorInfo.getWorkerCount());
                    lineStr = is.readLine();
                    if (!lineStr.equals("<poll finish>"))
                    {
                        throw new ParseException("Protocol parse error!", 0);
                    } // if
                }
                else
                {
                    throw new ParseException("Protocol parse error!", 0);
                } // if

            } // while
        }
        catch (Exception e)
        {
            e.printStackTrace();
        } // catch
        finally
        {
            try
            {
                clientSock.close();
                is.close();
                os.close();
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
            clientSock = null;
            is = null;
            os = null;
        } // finally
        return;
    } // run

    private void onPoll()
    {
        return;
    } // onPoll

} // NodeServerHandler