usercommunication.java

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

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

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.LinkedList;
import java.util.Vector;

/**
 * @author Administrator
 *
 * TODO To change the template for this generated type comment go to
 * Window - Preferences - Java - Code Style - Code Templates
 */
/**
*
* <p>Title: </p>
* <p>Description:与用户主程序的通信，包括相应SetDispatcher，NewTask，TaskStatus，GetTaskResult</p>
* <p>Copyright: Copyright (c) 2004</p>
* <p>Company: </p>
* @author not attributable
* @version 1.0
*/

/*****
 *This class is unused, and the function has been done in the class "MonitorCommunication". 
 *
 */
public class UserCommunication
   extends Thread
{
   
	private int userPort=-1;
	private Vector jobList;
	private LinkedList jobApply;

   private static final int TIME_OUT = 1000;
   private boolean ucIsActive = true; //用于控制accept线程的终止
   private ServerSocket serverSocket; //与monitor通信的serversocket
   
//   DispatcherJobManagement jobManager;
   private Integer dummy;
   

   public UserCommunication(int u_port,Vector j_List,LinkedList j_Apply)
   {
   	userPort=u_port;
   	jobList=j_List;
   	jobApply=j_Apply;
   }

   public void run()
   {
       try
       {
           serverSocket = new ServerSocket(userPort);
           serverSocket.setSoTimeout(TIME_OUT);
           while (ucIsActive)
           {
               Socket socket = null;
               try
               {
                   socket = serverSocket.accept();
                   Process p = new Process(socket); //连接后处理
                   p.start();
               }
               catch (java.io.InterruptedIOException e)
               {
               //等待连接超时，返回循环开头；（无处理）
               }
               catch (IOException e)
               {
                   //error("I/O Error", e);
                   log(e);
               }
               catch (SecurityException e)
               {
                   log(e);
                   //error("An unauthorized client has attempted to connect",
                    //     e);
               }
               /*catch (Throwable e)
               {
                  log(e);
               error("Unexpected exception", e);
               }*/
           } // while
           try
           {
               serverSocket.close();
           }
           catch (IOException e)
           {
               log(e);
               //error("Error closing server socket", e);
           }
       } // try
       catch (IOException e)
       {
       log(e);
           //error("Error binding to port", e);
       }
       catch (SecurityException e)
       {
           log(e);
           //error("The security manager refused permission to bind to port",
              //   e);
       }
   } // run

   public void terminate()
   {
       ucIsActive = false;
   }

   /**
    *
    * <p>Title: </p>
    * <p>Description: 处理连接</p>
    * <p>Copyright: Copyright (c) 2004</p>
    * <p>Company: </p>
    * @author not attributable
    * @version 1.0
    */
   class Process
       extends Thread
   {
       Socket socket;
       private BufferedReader is;
       private PrintWriter os;
       //private ObjectOutputStream oob;
       //private ObjectInputStream iob;

       public Process(Socket sock)
       {
           this.socket = sock;
       }

       public void run()
       {
           //process the query from monitor
           if (socket == null)
           {
               return;
           }
           try
           {
               is = new BufferedReader(new InputStreamReader(socket.
                   getInputStream()));
               //oob = new ObjectOutputStream(socket.getOutputStream());
               //iob = new ObjectInputStream(socket.getInputStream());
               os = new PrintWriter(socket.getOutputStream());
               //协议传输数据
               String lineStr = is.readLine();

               if (lineStr.equals("<Dispatcher poll>"))
               {
                   os.println("<Dispatcher is OK!>"); //传送dispatcher正常运行标志
                   os.flush();
                   lineStr = is.readLine();
                   if (lineStr.equals("<poll finish>"))
                   {
                   //return; //如果不等。。。？？
                   }
               }
               else if (lineStr.equals("<new task>"))
               {
                   String tmpJobId = is.readLine(); //申请执行的任务ID
                   int n = Integer.parseInt(is.readLine()); //申请执行子任务的个数
                   int j = 0;
                   Job tmpjob=null;
                   synchronized(jobList){
                       for (; j < jobList.size(); j++)
                       {
                           if (tmpJobId.equals( ( (Job) (jobList.
                               elementAt(
                               j))).getJobId()))
                               break;
                       }
                       if(j>=jobList.size())
                       {
                       System.out.println("Don't find this job in jobList!");
                       }
                               else tmpjob=(Job) (jobList.elementAt(j));
                   }
                   //String[] s=tmpjob.newSubJob(n,dummy);
                   
                   for (int i = 0; i < n; i++)
                   {
                       synchronized (jobApply)
                       {
                       //	jobApply.addLast(s[i]);
                       }
                      // os.println(s[i]);
                   }
                   os.flush();
               }
               else if (lineStr.equals("<get subtask status>"))
               {//it's not necessary !the status can be got from datapoll.
                   String taskId = is.readLine();
                   //........task  status...........
                   lineStr = is.readLine();
                   if (lineStr.equals("<Get status finish>"))
                   {
                   //return; //...
                   }
               }

           }
           catch (IOException e)
           {
               log(e);
               //error("I/O error", e);
           }
           finally
           {
               try
               {
                   is.close();
                   socket.close();
                   os.close();
               }
               catch (IOException e)
               {
                   log(e);
                   //error("Error closing socket", e);
               }
           }
           return;
       } //run
   }
   

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