workerjoin.java

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

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

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.InetAddress;

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

public class WorkerJoin
{
    private String m_mIpAddr;
    private int m_mPort;
    private String dIpAddr;
    private int dPort;
    private String workerId;
    private String dispatcherId;
    private int workerHeartbeatPort;
    


    //must initialized these two param in the begining.
    private String ipAddr;
    private String workerLoad="";

    private static final int TIME_OUT = 30*1000;

    private Socket nodeSocket = null;

    private BufferedReader is;
    private PrintWriter os;
    
    private String MonitorIpAddr;
    private int MonitorPort;
    private WorkerNode wn;

    public WorkerJoin(String MonitorIpAddr, int MonitorPort,WorkerNode wn)
    {
    	this.MonitorIpAddr=MonitorIpAddr;
    	this.MonitorPort=MonitorPort;
    	this.wn=wn;
    	try{
    	this.ipAddr=InetAddress.getLocalHost().getHostAddress();
    	}catch(Exception e)
		{
    		e.printStackTrace();
		}
    }
    
    public boolean join()
    {
        if (!joinInit(MonitorIpAddr, MonitorPort))
        {
            System.out.println("Fail to connect to a random monitor !");
            return false;
        }
        else
        {
            if (!joinInMonitor(m_mIpAddr, m_mPort))
            {
                System.out.println(
                    "Fail to connect to a special master monitor !");
                return false;
            }
            else
            {
            if(!joinInDispatcher(dIpAddr,dPort))
            {
                     System.out.println(
                    "Fail to connect to a dispatcher !");
                     return false;
            }
            else{
            wn.init(workerId,dispatcherId,dIpAddr,workerHeartbeatPort);
            return true;
            }
          }
        }
    }

    public boolean joinInit(String MonitorIpAddr, int MonitorPort)
    {
        boolean flag=false;
        if (!initCommunication(MonitorIpAddr, MonitorPort))
        {
        	ExceptionCode.exceptionCode=-2;
            return false;
        }
        try
        {
            os.println("worker join");
            os.flush();
            m_mIpAddr = is.readLine();
            m_mPort = Integer.parseInt(is.readLine());
            os.println("worker join finish");
            os.flush();
            flag=true;
        }
        catch (IOException e)
        {
            log(e);
        }
        finally
        {
            try
            {
                nodeSocket.close();
                is.close();
                os.close();
            }
            catch (IOException e)
            {
                log(e);
            }
        }
        if (!flag) ExceptionCode.exceptionCode=-1;
        return flag;

    }

    private boolean initCommunication(String ipAddr, int port)
    {
        try
        {
            this.nodeSocket = new Socket(ipAddr, port);
            nodeSocket.setSoTimeout(TIME_OUT);
            is = new BufferedReader(new InputStreamReader(nodeSocket.
                getInputStream()));
            os = new PrintWriter(nodeSocket.getOutputStream());
        }
        catch (Exception e)
        {
            nodeSocket = null;
            os = null;
            is = null;
            return false;
        }
        return true;
    }

    public boolean joinInMonitor(String MonitorIpAddr, int MonitorPort)
    {
        boolean flag=false;
        if (!initCommunication(MonitorIpAddr, MonitorPort))
        {
        	ExceptionCode.exceptionCode=-3;
            return false;
        }
        try
        {
            os.println("worker join in");
            os.flush();
            dIpAddr = is.readLine();
            //System.out.println(dIpAddr);
            dPort = Integer.parseInt(is.readLine());
            //System.out.println(dPort);
            if(is.readLine().equals("worker join in finish"))
            {
                flag = true;
            }else{
            //...
            }
        }
        catch (IOException e)
        {
            log(e);
        }
        finally
        {
            try
            {
                nodeSocket.close();
                is.close();
                os.close();
            }
            catch (IOException e)
            {
                log(e);
            }
        }
        if (!flag) ExceptionCode.exceptionCode=-1;
        return flag;
    }

    public boolean joinInDispatcher(String dispatcherIpAddr, int dispatcherPort)
   {
       boolean flag=false;
       if (!initCommunication(dispatcherIpAddr, dispatcherPort))
       {
       	   ExceptionCode.exceptionCode=-4;
           return false;
       }
       try
       {
           os.println("<Worker join>");
           os.flush();
           is.readLine();
           os.println(this.ipAddr);
           os.println(this.workerLoad);
           os.flush();

           //this.dIpAddr=is.readLine();
           this.workerId=is.readLine();
           this.dispatcherId=is.readLine();
           this.workerHeartbeatPort=Integer.parseInt(is.readLine());
           os.println("<join finish>");
           os.flush();
           System.out.println("the dispatcherid: "+dispatcherId);
           flag=true;
       }
       catch (IOException e)
       {
           log(e);
       }
       finally
       {
           try
           {
               nodeSocket.close();
               is.close();
               os.close();
           }
           catch (IOException e)
           {
               log(e);
           }
       }
       if (!flag) ExceptionCode.exceptionCode=-1;
       return flag;
   }


    private void log(Exception e)
    {
        e.printStackTrace();
    }

}