monitornode.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.Socket;
import java.text.ParseException;

/**
 * <p>Title: </p>
 * <p>Description: MonitorNode类负责monitor组内其他类的</p>
 * <p>Copyright: Copyright (c) 2004</p>
 * <p>Company: </p>
 * @author not attributable
 * @version 1.0
 */

public class MonitorNode
    extends Thread
{
    // monitor node运行正常
    public static final int MONITOR_ALIVE = 0;
    // monitor node没有响应
    public static final int MONITOR_CHOKE = 1;
    // 由于InterruptedException异常，线程已经停止运行
    public static final int MONITOR_DEAD = 2;

    private static final int POLL_INTERVAL = 60 * 1000;
    private static final int TIME_OUT = 1000;

    private String monitorId;
    private String ipAddr;
    private int monitorPort;
    private DispatcherGroup dispatcherGroupArray[];
    private MonitorInfo monitorInfo;
    private int status;
    private Socket nodeSocket;
    private BufferedReader is;
    private PrintWriter os;
    private MonitorConfiuration monConfig;
    //JButton jButton1 = new JButton();

    public static int getBracketStr(String inStr, int beginIndex, StringBuffer bracketStr)
    throws Exception
    {
        if (beginIndex >= inStr.length() || beginIndex < 0)
        {
            throw new ParseException("BracketStr parse error!", beginIndex);
        } // if
        int begin = inStr.indexOf('<', beginIndex);
        if (begin < 0)
        {
            throw new ParseException("BracketStr parse error!", begin);
        } // if
        int end = inStr.indexOf('>', begin);
        if (end < 0)
        {
            throw new ParseException("BracketStr parse error!", end);
        } // if
        //String bracketStr=inStr.substring(begin, end+1);
        bracketStr.replace(0, bracketStr.length(),inStr.substring(begin+1, end));
        //bracketStr.replaceAll(bracketStr, inStr.substring(begin, end + 1));
        return end;
        // getBracketStr

    }

    public MonitorNode(MonitorConfiuration monconf, int port, String monitorid)
    {
        monConfig = monconf;
        this.monitorPort=monConfig.getQueryPort();
        this.ipAddr = monConfig.getMonitorIp();
        monitorPort = port;
        this.monitorId = monitorid;
        nodeSocket = null;
        this.is = null;
        this.os = null;
        this.monitorInfo=null;
    } // MonitorNode

    public MonitorInfo getMonitorInfo()
    {
        return monitorInfo;
    } // GetMonitorInfo

    public MonitorConfiuration getMonConf()
    {
        return this.monConfig;
    }

    public void run()
    {
        /*
                 if (!this.initMonitorNode())
                 {
            return;
                 } // if
         */
        while (true)
        {
            onPoll();
            try
            {
                Thread.sleep(POLL_INTERVAL);
            } //try
            catch (InterruptedException e)
            {
            //return;
            } // catch
        } // while
        ///return;
    } // run

    public DispatcherGroup getDispatcherGroup(int index)
    {
        return dispatcherGroupArray[index];
    } // GetDispatcherGroup

    public boolean onPoll()
    {
        if (nodeSocket == null)
        {
            if (!initMonitorNode())
            {
                return false;
            }
        } // if
        String tmpStr;
        try
        {
            os.println("<monitor poll>");
            os.flush();
            // 第一项是monitorId
            tmpStr = is.readLine();
            //int endIndex = 0;
            StringBuffer tmpMonitorId = new StringBuffer();
            int endIndex = MonitorNode.getBracketStr(tmpStr, 0, tmpMonitorId);
            // 第二项是jobCount
            tmpStr = is.readLine();
            StringBuffer tmpNum = new StringBuffer();
            endIndex = MonitorNode.getBracketStr(tmpStr, endIndex, tmpNum);
            int tmpJobCount = Integer.parseInt(new String(tmpNum));
            // 第三项是workerLoad
            tmpStr = is.readLine();
            endIndex = MonitorNode.getBracketStr(tmpStr, endIndex, tmpNum);
            int tmpWorkerLoad = Integer.parseInt(new String(tmpNum));
            // 第四项是dispatcherLoad
            tmpStr = is.readLine();
            endIndex = MonitorNode.getBracketStr(tmpStr, endIndex, tmpNum);
            int tmpDispatcherLoad = Integer.parseInt(new String(tmpNum));
            // 第五项是dispatcherGroupCount
            tmpStr = is.readLine();
            endIndex = MonitorNode.getBracketStr(tmpStr, endIndex, tmpNum);
            int tmpdispatcherGroupCount = Integer.parseInt(new String(tmpNum));
            // 第6项是dispatcherCount
            tmpStr = is.readLine();
            endIndex = MonitorNode.getBracketStr(tmpStr, endIndex, tmpNum);
            int tmpDispatcherCount = Integer.parseInt(new String(tmpNum));
            // 第7项是workerCount
            tmpStr = is.readLine();
            endIndex = MonitorNode.getBracketStr(tmpStr, endIndex, tmpNum);
            int tmpWorkerCount = Integer.parseInt(new String(tmpNum));

            if (this.monitorInfo == null)
            {
                monitorInfo = new MonitorInfo();
            } // if
            monitorInfo.setInfo(new String(tmpMonitorId),
                                tmpJobCount,
                                tmpWorkerLoad,
                                tmpDispatcherLoad,
                                tmpdispatcherGroupCount,
                                tmpDispatcherCount,
                                tmpWorkerCount,
                                true);

            this.status = MonitorNode.MONITOR_ALIVE;
            os.println("<poll finish>");
            os.flush();
            os.close();
            is.close();
        } // try
        catch (Exception e)
        {
            //nodeSocket.close();
            nodeSocket = null;
            this.status = MonitorNode.MONITOR_CHOKE;
            return false;
        } // catch

        return true;
    } // onPoll

    private boolean initMonitorNode()
    {
        try
        {
            this.nodeSocket = new Socket(ipAddr, monitorPort);
            nodeSocket.setSoTimeout(TIME_OUT);
            is = new BufferedReader(new InputStreamReader(nodeSocket.
                getInputStream()));
            os = new PrintWriter(nodeSocket.getOutputStream());
        } // try
        catch (Exception e)
        {
            nodeSocket = null;
            is = null;
            os = null;
            this.status = MonitorNode.MONITOR_DEAD;
            return false;
        } // catch
        status = MonitorNode.MONITOR_ALIVE;
        return true;
    } // initMonitorNode

} // MonitorNode