rpccomm.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;

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

public class RPCComm extends Thread
{
    public static int dispatchUnit=20;

    public static int serverPort=1885;
    private DispatcherGroup dispatcherGroup;
    MonitorJobManagement jobManager;
    private ServerSocket serverSock;

    public RPCComm(DispatcherGroup disgroup, MonitorJobManagement jobmanager)
    {
        this.jobManager=jobmanager;
        dispatcherGroup=disgroup;
    } // RPCComm

    public void run()
    {
        
        Socket clientSock;
        try
        {
            serverSock=new ServerSocket(serverPort);
            while ( (clientSock = serverSock.accept()) != null)
            {
                RPCCommHandler rpcHandler=new RPCCommHandler(clientSock, this.dispatcherGroup,jobManager);
                rpcHandler.start();
            } // while
        } // try
        catch (Exception e)
        {
            System.out.println(e.toString());
            serverSock=null;
            clientSock=null;
        } // catch
        System.out.println("rpccomm's quit!");
        return;
    } // run
    
    public void quit()
    {
    	System.out.println("RPCComm start to quitting...");
    	try{
    		if(!this.serverSock.isClosed())
    		    this.serverSock.close();
    	}
    	catch(Exception e)
		{
    		
		}
    	System.out.println("RPCComm's quitting finished!");
    	return;
    }
} // RPCComm

class RPCCommHandler extends Thread
{
    public DispatcherGroup dispatcherGroup;
    Socket clientSock;
    MonitorJobManagement jobManager;

    RPCCommHandler(Socket clientsock, DispatcherGroup disgroup, MonitorJobManagement jobmanager)
    {
    	
        this.jobManager=jobmanager;
        dispatcherGroup = disgroup;
        clientSock = clientsock;
        
    } // RPCCommHandler

    public void run()
    {
        BufferedReader is;
        PrintWriter os;
        if (clientSock == null)
        {
            return;
        } //

        try
        {
            is = new BufferedReader(new InputStreamReader(clientSock.
                getInputStream()));
            os = new PrintWriter(clientSock.getOutputStream());
            String lineStr = is.readLine();
            if (lineStr.equals("<new subtask>"))
            {
                // 处理新子任务的申请
                newTaskHandler(is, os);
            } // if
            // 还需要处理查询作业信息的代码吗？
        }
        catch (Exception e)
        {
            System.out.println(e.toString());
        } // catch

        is = null;
        os = null;
        clientSock = null;
        return;
    } // run

    private void newTaskHandler(BufferedReader is, PrintWriter os)
        throws Exception
    {
        os.println("<new subtask next>");
        os.flush();
        String lineStr = is.readLine();
        String jobid=lineStr;
//        System.out.println(lineStr);
        MonitorJob job=null;
        if ((job=this.jobManager.GetJob(jobid))==null)
        {
            throw new ParseException("protocol parse error!", 0);
        }
        lineStr = is.readLine();
        int taskCount=Integer.parseInt(lineStr);
        dispatcherGroup.getJob(is, os, job,taskCount);
        os.flush();
        this.clientSock.close();
    } // newTaskHandler

}