tasktransfer.java

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

/*
 * Created on 2004-12-8
 *
 * TODO To change the template for this generated file go to
 * Window - Preferences - Java - Code Style - Code Templates
 */
package cn.edu.hust.cgcl.biogrid.worker;

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

/**
 * @author Administrator
 *
 * TODO To change the template for this generated type comment go to
 * Window - Preferences - Java - Code Style - Code Templates
 */
public class TaskTransfer {
	
	private boolean isActive=true;
	private Socket nodeSocket=null;
	private String dIpAddr="";
	private int taskPort;
	private String workerId="";
	
	private WorkerInfo nodeInfo;
	
	private BufferedReader is;
	private PrintWriter os;
	private ObjectInputStream iob;
	public TaskTransfer(WorkerInfo Info)
	{
		this.nodeInfo=Info;
		this.dIpAddr=nodeInfo.get_f_d_Ip();
		this.taskPort=nodeInfo.get_f_d_port();
		this.workerId=nodeInfo.getWorkerId();
	}
	
	public boolean taskFinished(String proPath,String dataPath,String resultPath)
	{
		String pPath=new String(proPath);
		String dPath=new String(dataPath);
		String rPath=new String(resultPath);
		
		if (nodeSocket == null)
        {
        	if (!initCommunication())
            {
                return false;
            }
        }
		try
        {
            os.println(this.workerId);
            //System.out.println("to dispatcher :<subjob hava finished>");
            os.println("<subjob hava finished>");
            os.flush();
            os.println(nodeInfo.sj.getJobId());
            os.println(nodeInfo.sj.getSubJobId());
            os.flush();
            if (is.readLine().equals("<dispatcher hava known>"))
            {
                nodeInfo.setFinishFlag(false);
                if(Parameter.workSpaceFlag==1)
                {
                //计算完毕一个子任务后删除程序、数据和结果。
                //System.out.println("delete path: "+pPath+dPath+rPath);	
                /*FileHandle.delAllFile(pPath);
                FileHandle.delAllFile(dPath);
                FileHandle.delAllFile(rPath);*/
                }
                //System.out.println("<dispatcher hava known>");
                return true;
            }
            System.out.println("protocol parse error!");
            return false;
        }catch(IOException e)
		{
        	e.printStackTrace();
        	return false;
		}
        finally
		{
        	try{
        		nodeSocket.close();
        		nodeSocket=null;
        	}
        	catch(Exception e)
			{
        		e.printStackTrace();
			}
		}
	}
	
	public void workerQuit()
	{
		System.out.println("WorkerQuit handler quitting...");
		if (nodeSocket == null)
        {
        	if (!initCommunication())
            {
                return ;
            }
        }
		try{
		os.println(this.workerId);
		os.println("<worker quit>");
		os.flush();
        if (is.readLine().equals("<worker quit agree>"))
        {
            this.terminate();
            //WorkerDB wdb=new WorkerDB();
            //wdb.createDB(nodeInfo);
            //return;
        }
        //System.out.println("protocol parse error!");
		}catch(IOException e)
		{
			e.printStackTrace();
		}
		finally
		{
        	try{
        		nodeSocket.close();
        		nodeSocket=null;
        	}
        	catch(Exception e)
			{
        		e.printStackTrace();
			}
		}
		System.out.println("WorkerQuit handler quit!");
	}

	  private boolean initCommunication()
	    {
	        try
	        {
	            this.nodeSocket = new Socket(dIpAddr,taskPort);
	            //nodeSocket.setSoTimeout(TIME_OUT);
	            
	            is = new BufferedReader(new InputStreamReader(nodeSocket.
	                getInputStream()));  
	            os = new PrintWriter(nodeSocket.getOutputStream());
	            iob=new ObjectInputStream(nodeSocket.getInputStream());
	        }
	        catch (Exception e)
	        {
	        	try{
	            if (nodeSocket != null) nodeSocket.close();
	        	}
	        	catch(Exception e1)
				{
	        		e1.printStackTrace();
				}
	            //iob=null;
	            return false;
	        }
	        return true;
	    }

//	    public void quit()
//	    {
//	    quitFlag=true;
//	    }

	    private void terminate()
	    {
	    isActive=false;
	    }
}