mpjdaemon.java

MPI for java for Distributed Programming

	        // .. this line is generating exception which kills the 
		//    daemon ... I think we need a try catch here and if
		//    any exception is generated, then we will have to 
		//    goto back to selector.select() method ..
		//
		//    this is Windows 2000 specific error ..i have not 
		//    seen this error on Windows XP ..
	      
	      try { 
		      
                if((readInt = socketChannel.read(lilBuffer)) == -1) {

                  if(DEBUG && logger.isDebugEnabled()) { 
                    logger.debug(" The daemon has received an End of "+
	  	  		  "Stream signal") ; 
	    	    logger.debug(" checking if this channel is still open");
		  }
		  
		  if(socketChannel.isOpen()) {
                    if(DEBUG && logger.isDebugEnabled()) { 
                      logger.debug("closing the channel");
		    }
		    socketChannel.close() ; 			  
		  }

                  if(DEBUG && logger.isDebugEnabled()) { 
                    logger.debug("continuing to select()");
		  }
		  continue ; 
                  //END_OF_STREAM signal .... 

                }
	      }
	      catch(Exception innerException) {
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("Exception in selector thread, message is"+
				  innerException.getMessage() );
                  logger.debug (" continuing to select() method ..."); 
		}
		continue; 
	      }
		      

                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("READ_EVENT (read)" + readInt);
		}

              lilBuffer.flip();
              lilBuffer.get(lilArray, 0, 4);
              String read = new String(lilArray);
              if(DEBUG && logger.isDebugEnabled()) { 
                logger.debug ("READ_EVENT (String)<" + read + ">");
	      }

              if (read.equals("url-")) {
 	        if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("url-");
		}
                int length = lilBuffer.getInt();
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("URL Length -->" + length);
		}
                lilBuffer.clear();
                buffer.limit(length);
                socketChannel.read(buffer);
                byte[] byteArray = new byte[length];
                buffer.flip();
                buffer.get(byteArray, 0, length);
                URL = new String(byteArray);
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("URL:<" + URL + ">");
		}
		
		if(URL.endsWith(".jar")) {
                  className = null ;  			
		}
		
                buffer.clear();
              }
	      
              if (read.equals("mul-")) {
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("mul-");
		}
                int length = lilBuffer.getInt();
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("URL Length -->" + length);
		}
                lilBuffer.clear();
                buffer.limit(length);
                socketChannel.read(buffer);
                byte[] byteArray = new byte[length];
                buffer.flip();
                buffer.get(byteArray, 0, length);
                mpjURL = new String(byteArray);
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("mpjURL:<" + mpjURL + ">");
		}
                buffer.clear();
              }
	      
              if (read.equals("cls-")) {
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("cls-");
		}
                int length = lilBuffer.getInt();
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("className length -->" + length);
		}
                lilBuffer.clear();
                buffer.limit(length);
                socketChannel.read(buffer);
                byte[] byteArray = new byte[length];
                buffer.flip();
                buffer.get(byteArray, 0, length);
                className = new String(byteArray);
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("className :<" + className + ">");
		}
                buffer.clear();
              }

              if (read.equals("app-")) {
		      
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("app-");
		}
                int length = lilBuffer.getInt();
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("Application args Length -->" + length);
		}
                lilBuffer.clear();

		for(int j=0 ; j<length ; j++) {
                  lilBuffer2.position(0); lilBuffer2.limit(4); 			
                  socketChannel.read(lilBuffer2);
		  lilBuffer2.flip();
                  int argLen = lilBuffer2.getInt();
                  buffer.limit(argLen);
                  socketChannel.read(buffer);
                  byte[] t = new byte[argLen];
                  buffer.flip();
                  buffer.get(t,0,argLen);
		  appArgs.add(new String(t)); 
		  buffer.clear(); 
		  lilBuffer2.clear();
		}
                 
		//for loop to create a new array ...
                buffer.clear();
		
              }
              else if (read.equals("num-")) {
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("num-");
		}
                int length = lilBuffer.getInt();
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("should be 4, isit ? -->" + length);
		}
                lilBuffer.clear();
                socketChannel.read(lilBuffer2);
                lilBuffer2.flip();
                processes = lilBuffer2.getInt();
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("Num of processes ==>" + processes);
		}
                lilBuffer2.clear();
              }

              else if (read.equals("arg-")) {
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("arg-");
		}
                int length = lilBuffer.getInt();
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("argu len -->"+length);
		}
                lilBuffer.clear();

		for(int j=0 ; j<length ; j++) {
                  lilBuffer2.position(0); lilBuffer2.limit(4); 			
                  socketChannel.read(lilBuffer2);
		  lilBuffer2.flip();
                  int argLen = lilBuffer2.getInt();
                  buffer.limit(argLen);
                  socketChannel.read(buffer);
                  byte[] t = new byte[argLen];
                  buffer.flip();
                  buffer.get(t,0,argLen);
		  jvmArgs.add(new String(t)); 
		  buffer.clear(); 
		  lilBuffer2.clear();
		}
                 
		//for loop to create a new array ...
                buffer.clear();
		
              }

              else if (read.equals("dev-")) {
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("dev-");
		}
                int length = lilBuffer.getInt();
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("dev-Length -->" + length);
		}
                lilBuffer.clear();
                buffer.limit(length);
                socketChannel.read(buffer);
                byte[] byteArray = new byte[length];
                buffer.flip();
                buffer.get(byteArray, 0, length);
                deviceName = new String(byteArray);
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("Device Name :<" + deviceName + ">");
		}
                buffer.clear();
              }
	      
              else if (read.equals("ldr-")) {
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("ldr-");
		}
                int length = lilBuffer.getInt();
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("ldr-Length -->" + length);
		}
                lilBuffer.clear();
                buffer.limit(length);
                socketChannel.read(buffer);
                byte[] byteArray = new byte[length];
                buffer.flip();
                buffer.get(byteArray, 0, length);
                loader = new String(byteArray);
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("ldr:<"+loader+">");
		}
                buffer.clear();
              }

              else if (read.equals("wdr-")) {
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("wdr-");
		}
                int length = lilBuffer.getInt();
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("wdr-Length -->" + length);
		}
                lilBuffer.clear();
                buffer.limit(length);
                socketChannel.read(buffer);
                byte[] byteArray = new byte[length];
                buffer.flip();
                buffer.get(byteArray, 0, length);
                wdir = new String(byteArray);
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("wdir :<"+wdir+">");
		}
                buffer.clear();
              }

              else if (read.equals("*GO*")) {
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("GO");
		}
                lilBuffer.clear();
                startExecution ();

              }
              else if (read.equals("kill")) {
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("processing kill event");
		}
                MPJProcessPrintStream.stop();
                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("Stopping the output");
		}

                try {
                  if(DEBUG && logger.isDebugEnabled()) { 
                    logger.debug ("peerChannel is closed or what ?" +
                                peerChannel.isOpen());
		  }

                  if (peerChannel.isOpen()) {
                    if(DEBUG && logger.isDebugEnabled()) { 
                      logger.debug ("Closing it ...");
		    }
                    peerChannel.close();
                  }
                }
                catch (Exception e) {}

                  if(DEBUG && logger.isDebugEnabled()) { 
                    logger.debug ("Killling the process");
		  }
                try {
                  synchronized (MPJDaemon.this) {
                    if (processVector != null) {
                      synchronized (processVector) {

                        for (int i = 0; i < processVector.length; i++) {
                          processVector[i].destroy();
                        }

                        kill_signal = true;
                      }
                    }
                  }
                }
                catch (Exception e) {} 
//no matter what happens, we cant let this thread
//die, coz otherwise, the daemon will die as well..
//maybe you wanne stop the output handler threads as well.

                if(DEBUG && logger.isDebugEnabled()) { 
                  logger.debug ("Killed it");
		}
                buffer.clear();
                lilBuffer.clear();

              }

            } //end if key.isReadable()

            else if (key.isWritable()) {

              if(DEBUG && logger.isDebugEnabled()) { 
                logger.debug(
                    "In, WRITABLE, so changing the interestOps to READ_ONLY");
	      }
              key.interestOps(SelectionKey.OP_READ);

            }

          } //end while iterator
        } //end while
      }
      catch (Exception ioe1) {
        if(DEBUG && logger.isDebugEnabled()) { 
          logger.debug("Exception in selector thread " + ioe1.getMessage());
	}
        ioe1.printStackTrace();
        //System.exit(0);
      }
    } //end run()
  }; //end selectorThread which is an inner class 

  Runnable outputHandler = new Runnable() {

    public void run() {

      Process p = null;

      synchronized (processVector) {
        p = processVector[pos];
      }

      outputHandlerSem.release() ; 
      String line = "";
      InputStream outp = p.getInputStream();
      BufferedReader reader = new BufferedReader(new InputStreamReader(outp));
       
      if(DEBUG && logger.isDebugEnabled()) { 
        logger.debug( "outputting ...");
      }

      try {
        do {
          if (!line.equals("")) {
            line.trim(); 
 
            synchronized (this) {
              System.out.println(line);
              //if(DEBUG && logger.isDebugEnabled()) { 
              //  logger.debug(line);
	      //}
            } 
          }
        }  while ( (line = reader.readLine()) != null); 
        // && !kill_signal); 
      }
      catch (Exception e) {
        if(DEBUG && logger.isDebugEnabled()) { 
          logger.debug ("outputHandler =>" + e.getMessage());
	}
        e.printStackTrace();
      } 
    } //end run.
  }; //end outputHandler.

  public static void main(String args[]) {
    try {
      MPJDaemon dae = new MPJDaemon(args);
    }
    catch (Exception e) {
      e.printStackTrace();
    }
  }

            //"-Xloggc:" + hostName + ".gc",
            //"-XX:+PrintGCDetails",
            //"-XX:+PrintGCTimeStamps",
            //"-XX:+PrintGCApplicationConcurrentTime",
            //"-XX:+PrintGCApplicationStoppedTime",
            //"-Xnoclassgc",
            //"-XX:MinHeapFreeRatio=5",
            //"-XX:MaxHeapFreeRatio=5",
            //"-Xms512M", "-Xmx512M",
            //"-DSIZE=1000", "-DITERATIONS=100",
            //"-Xdebug",
            //"-Xrunjdwp:transport=dt_socket,address=11000,server=y,suspend=n",
}