dispatchernode.java

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

                e1.printStackTrace();
            }
            //oob=null;
            //iob=null;
            return false;
        }
        return true;
    } //initDispatcher

    public boolean onPoll()
    {
        if (nodeSocket == null)
        {
            if (!init())
            {
                return false;
            }
        }
        try
        {
            os.println("<Dispatcher poll>");
            os.flush();
//            System.out.println("<Dispatcher poll>");
            //nodeInfo=null;
            for (int i = 0; i < 6; i++)
            {
                tmpString.add(is.readLine());
            }
            synchronized (nodeInfo)
            {
                //nodeInfo = (DispatcherInfo) iob.readObject();
                nodeInfo.setIdleWorkerCount(Integer.parseInt( (String)
                    tmpString.get(3)));
                nodeInfo.setDispatcherLoad( (String) tmpString.get(5));
                nodeInfo.setDispatcherInfo( (String) tmpString.get(0),
                                           "", "", "", "",
                                           Integer.parseInt( (String) tmpString.
                    get(1)),
                                           Integer.parseInt( (String) tmpString.
                    get(2)),
                                           Boolean.getBoolean( (String)
                    tmpString.get(4)),
                                           null, false);
            }
            tmpString.clear();
//            System.out.println(nodeInfo.getWorkerCount());
            os.println("<poll finish>");
            os.flush();
//            System.out.println("<poll finish>");
        } // try
        catch (Exception e)
        {
            System.out.println(e.toString());
            try
            {
                nodeSocket.close();
                is.close();
                os.close();
            }
            catch (Exception e1)
            {
                e1.printStackTrace();
                return false;
            }
            return false;
        } // catch
        return true;
    } // onPoll

//    private boolean initDispatcher()
//    {
//        try
//        {
//        	Socket socket = new Socket(ipAddr, taskPort);
//            BufferedReader iss = new BufferedReader(new InputStreamReader(socket.
//               getInputStream()));
//            PrintWriter oss = new PrintWriter(socket.getOutputStream());
//            ObjectOutputStream ob = new ObjectOutputStream(socket.getOutputStream());
//        }
//        catch (Exception e)
//        {
//            System.out.println(e.toString());
//            try
//            {
//                if (nodeSocket != null)
//                    nodeSocket.close();
//                if (is != null)
//                    is.close();
//                if (os != null)
//                    os.close();
//            }
//            catch (Exception e1)
//            {
//                e1.printStackTrace();
//            }
//            //oob=null;
//            //iob=null;
//            return false;
//        }
//        return true;
//    } //initDispatcher

        /* public void setDispatcherInfo(String dispatcherId, String firstMointroId,
                           String secondMointorId, int jobCount,
                           int workerCount,
                           int workerLoad, int dispatcherLoad,
                           boolean canAcceptJob,Vector dispatcherJob)
      {
          this.nodeInfo.setDispatcherInfo(dispatcherId, firstMointroId,
          secondMointorId, jobCount, workerCount,
                                 workerLoad, dispatcherLoad,
                                 canAcceptJob,dispatcherJob);
      } //setDispatcherInfo*/

    private void cancelJobHandle(String cancelJobId)
    {
//        if (nodeSocket == null)
//        {
//            if (!initDispatcher())
//            {
//                return false;
//            }
//        }
        Socket socket = null;
        BufferedReader iss = null;
        PrintWriter oss = null;
        ObjectOutputStream ob = null;

        try
        {
            try
            {
                socket = new Socket(ipAddr, taskPort);
                iss = new BufferedReader(new InputStreamReader(socket.
                    getInputStream()));
                oss = new PrintWriter(socket.getOutputStream());
                ob = new ObjectOutputStream(socket.getOutputStream());
            }
            catch (Exception e)
            {
                e.printStackTrace();
                return;
            }
            String tmpStr = "";
            oss.println("<Dispatcher cancelJob>");
            oss.println(cancelJobId);
            oss.flush();

            tmpStr = iss.readLine();
            if (!tmpStr.equals("<cancelJob finish>"))
            {

//                cancelJobFlag = false;
//                cancelJobId = "";
                System.out.println("communication protocol error!");

            }

        }
        catch (Exception e)
        {
            e.printStackTrace();
            //System.out.println(e.toString());
        }
        finally
        {
            try
            {
                socket.close();
            }
            catch (Exception e1)
            {
                e1.printStackTrace();

            }

        }
    }

    private void newJobHandle(MonitorJob tmpNewJob, int newJobCount)
    {
//        if (nodeSocket == null)
//        {
//            if (!initDispatcher())
//            {
//                return false;
//            }
//        }
//        OutputStream oos;
        Socket socket = null;
        BufferedReader iss = null;
        PrintWriter oss = null;
        ObjectOutputStream ob = null;
        try
        {
            try
            {
                socket = new Socket(ipAddr, taskPort);
                System.out.println(taskPort);
                iss = new BufferedReader(new InputStreamReader(socket.
                    getInputStream()));
                oss = new PrintWriter(socket.getOutputStream());
                ob = new ObjectOutputStream(socket.getOutputStream());
            }
            catch (Exception e)
            {
                e.printStackTrace();
                return;
            }
            String tmpStr = "";
            oss.println("<Job transmission>");
            oss.flush();
            oss.println(newJobCount);
            oss.flush();

//            oos = nodeSocket.getOutputStream();
//            oos.write("<Job transmission>\n".getBytes());
//            oos.flush();
//            oos.write(Integer.toString(newJobCount).getBytes());
//            oos.write("\n".getBytes());
//            oos.flush();
            tmpStr = iss.readLine();
            if (!tmpStr.equals("<Send Object>"))
            {
                //iss.reset();
                System.out.println("communication protocol error!");
                return;
            } // if
            //oob = new ObjectOutputStream(nodeSocket.getOutputStream());
            ob.writeObject(tmpNewJob);
            ob.flush();
            //oob.close();

            /*
                         tmpStr = is.readLine();
                         if (tmpStr.equals("<tansmission finish>"))
                         {
                newJobFlag = false;
                tmpNewJob = null;
                newJobCount = 0;
                return true;
                         }
             */
//            newJobFlag = false;
//            tmpNewJob = null;
//            newJobCount = 0;
        }
        catch (Exception e)
        {
            e.printStackTrace();
            //System.out.println(e.toString());
        }
        finally
        {
            try
            {
                socket.close();
            }
            catch (Exception e1)
            {
                e1.printStackTrace();
            }

        }
    }

} // DispatcherNode