mpjrun.java

MPI for java for Distributed Programming

      try {
	fileAppender = new DailyRollingFileAppender(
			    new PatternLayout(
			    " %-5p %c %x - %m\n" ),
			    mpjHomeDir+"/logs/mpjrun.log",
			    "yyyy-MM-dd-a" );

	Logger rootLogger = Logger.getRootLogger() ;
	rootLogger.addAppender( fileAppender);
	LoggerRepository rep =  rootLogger.getLoggerRepository() ;
	rootLogger.setLevel ((Level) Level.ALL );
	//rep.setThreshold((Level) Level.OFF ) ;
	logger = Logger.getLogger( "runtime" );
      }
      catch(Exception e) {
	throw new MPJRuntimeException(e) ;
      }
    }  
  }

  private void printUsage() { 
    System.out.println(   
      "\n\n #########################################################"+     
      "\n mpirun.[bat/sh] [options] -jar file.jar"+
      "\n                 <package>className"+
      "\n                 [application arguments]"+
      "\n OPTIONS "+
      "\n   -np val            -- 1"+ 
      "\n   -dev val           -- niodev"+
      "\n   -dport val         -- 10000"+ 
      "\n   -wdir val          -- $MPJ_HOME/bin"+ 
      "\n   -mpjport val       -- 20000"+  
      "\n   -mxboardnum val    -- 0"+  
      "\n   -sport val         -- 15000"+
      "\n   -psl val           -- 128Kbytes"+ 
      "\n   -machinesfile val  -- machines"+ 
      "\n   -localloader"+ 
      "\n   -h                 -- print this usage information"+ 
      "\n   ...any JVM arguments..."+
 "\n Note: Value on the right in front of each option is the default value"+ 
 "\n Note: 'MPJ_HOME' variable must be set"+
      "\n\n #########################################################" );  
  }

  /**
   * Parses the input ...
   */
  private void processInput(String args[]) {

    if (args.length < 2) {
      printUsage() ;
      System.exit(0);  
    }
    
    boolean append = false;
    boolean beforeJar = true ; 
    
    for (int i = 0; i < args.length; i++) {

      if(args[i].equals("-np")) {
        nprocs = new Integer(args[i+1]).intValue();
        i++;
      }

      else if(args[i].equals("-h")) {
        printUsage();
        System.exit(0); 
      }
      
      else if (args[i].equals("-dport")) {
        D_SER_PORT = new Integer(args[i+1]).intValue();
        i++;
      }
      
      else if (args[i].equals("-localloader")) {
        loader = "useLocalLoader" ; 	      
      }
      
      else if (args[i].equals("-dev")) {
        deviceName = args[i+1];
        i++;
      } 

      else if (args[i].equals("-machinesfile")) {
        machinesFile = args[i+1];
        i++;
      }

      else if (args[i].equals("-wdir")) {
        wdir = args[i+1];
        i++;
      }

      else if(args[i].equals("-psl")) {
        psl = new Integer(args[i+1]).intValue();
        i++;
      }
      
      else if (args[i].equals("-mpjport")) {
        MPJ_SERVER_PORT = new Integer(args[i+1]).intValue();
        i++;
      }
      
      else if (args[i].equals("-mxboardnum")) {
        mxBoardNum = new Integer(args[i+1]).intValue();
        i++;
      }
      
      else if (args[i].equals("-cp")) {
        jvmArgs.add("-cp");
	jvmArgs.add(args[i+1]);
        i++;
      }
      
      else if (args[i].equals("-sport")) {
        S_PORT = new Integer(args[i+1]).intValue();
        i++;
      }
      
      else if(args[i].equals("-class")) {
        codeBase = System.getProperty("user.dir");	      
	className = args[i+1];
	beforeJar = false ; 
	i++;
      }
      
      else if(args[i].equals("-jar")) {
        File tFile = new File( args[i+1] );
	File absFile = tFile.getAbsoluteFile();
	
	if(tFile.exists() || loader.equals("useLocalLoader")) {
          jarName = tFile.getName() ;
	  codeBase = absFile.getParent();
	  beforeJar = false ; 
	  i++;
	}
	else {
          throw new MPJRuntimeException("mpjrun cannot find the jar file <"+
			  args[i+1]+">. Make sure this is the right path.");	
	}
	
      }

      else {
	      
        //these have to be jvm options ...  		
        if(beforeJar) {
          if(args[i].startsWith("-")) { 		
	    jvmArgs.add(args[i]); 
	  }
          else {
            codeBase = System.getProperty("user.dir");	      
 	    className = args[i];
	    beforeJar = false ; 
          }
	}
	
        //these have to be app arguments ...		
	else {
          appArgs.add(args[i]);		
	}

      }

    }

    jArgs = jvmArgs.toArray(new String[0]);
    aArgs = appArgs.toArray(new String[0]);

    if(DEBUG && logger.isDebugEnabled()) {

      logger.debug("###########################"); 	    
      logger.debug("-appargs: <"+applicationArgs+">");
      logger.debug("-dport: <"+D_SER_PORT+">");
      logger.debug("-mpjport: <"+MPJ_SERVER_PORT+">");
      logger.debug("-sport: <"+S_PORT+">");
      logger.debug("-np: <"+nprocs+">");
      logger.debug("$MPJ_HOME: <"+mpjHomeDir+">");
      logger.debug("-dir: <"+codeBase+">"); 
      logger.debug("-dev: <"+deviceName+">");
      logger.debug("-psl: <"+psl+">");
      logger.debug("-jarName: <"+jarName+">");
      logger.debug("jvmArgs.length: <"+jArgs.length+">");
      logger.debug("jarName : <"+jarName+">");
      logger.debug("className : <"+className+">");
      logger.debug("codeBase : <"+codeBase+">");
      

      for(int i=0; i<jArgs.length ; i++) {
        logger.debug(" jvmArgs["+i+"]: <"+jArgs[i]+">");	      
      }
      
      logger.debug("appArgs.length: <"+aArgs.length+">");

      for(int i=0; i<aArgs.length ; i++) {
        logger.debug(" appArgs["+i+"]: <"+aArgs[i]+">");	      
      }
      
      
      logger.debug("###########################"); 	    
    }

  }

  private synchronized void Wait() throws Exception {
    if (wait) {
      logger.debug("Waiting ...");
      this.wait();
      logger.debug("Unwaiting ...");
    }

    wait = true;
  }

  private synchronized void Notify() {
    logger.debug("Notifying ...");
    this.notify();
    wait = false;
  }

  private void assignTasks() throws Exception {
	  
    PrintStream cout = null;
    int rank = 0;
    String name = null;

    try {
      cfos = new FileOutputStream(CONF_FILE);
    }
    catch (FileNotFoundException fnfe) {}

    cout = new PrintStream(cfos);
    int noOfMachines = machineVector.size();
    cout.println("# Number of Processes");
    cout.println(nprocs);
    cout.println("# Protocol Switch Limit");
    cout.println(psl);
    cout.println("# Entry, HOST_NAME/IP@SERVERPORT@RANK");

    if (nprocs < noOfMachines) {

      if(DEBUG && logger.isDebugEnabled()) { 
        logger.debug("Processes Requested " + nprocs +
                  " are less than than machines " + noOfMachines);
        logger.debug("Adding 1 processes to the first " + nprocs +
                  " items");
      }

      for (int i = 0; i < nprocs; i++) {
        //name=(String)machineVector.get(i);
        //name=InetAddress.getByName(name).getHostName();
        //name=InetAddress.getByAddress( name.getBytes() ).getHostName();
        procsPerMachineTable.put( (String) machineVector.get(i),
                                 new Integer(1));
	 
	if(deviceName.equals("niodev")) { 
          cout.println(name + "@" + MPJ_SERVER_PORT +
                       "@" + (rank++));
	} else if(deviceName.equals("mxdev")) { 
          cout.println(name + "@" + mxBoardNum+
                       "@" + (rank++));
	}
	
        if(DEBUG && logger.isDebugEnabled()) { 
          logger.debug("procPerMachineTable==>" + procsPerMachineTable);
	}

      }

    }
    else if (nprocs > noOfMachines) {

      logger.debug("Processes Requested " + nprocs +
                  " are greater than than machines " + noOfMachines);
      int divisor = nprocs / noOfMachines;
      logger.debug("divisor " + divisor);
      int remainder = nprocs % noOfMachines;
      logger.debug("remainder " + remainder);

      for (int i = 0; i < noOfMachines; i++) {
	      
        if (i < remainder) {
		
          procsPerMachineTable.put( (String) machineVector.get(i),
                                   new Integer(divisor + 1));
          if(DEBUG && logger.isDebugEnabled()) { 
            logger.debug("procPerMachineTable==>" + procsPerMachineTable);
	  }
	  
          //name=(String)machineVector.get(i);
          //name=InetAddress.getByAddress( name.getBytes() ).getHostName();
          //name=InetAddress.getByName(name).getHostName();

          for (int j = 0; j < (divisor + 1); j++) {
            if(deviceName.equals("niodev")) { 		  
              cout.println( (String) machineVector.get(i) + "@" +
                           (MPJ_SERVER_PORT + (j * 2)) + "@" + (rank++));
	    } else if(deviceName.equals("mxdev")) { 
              cout.println( (String) machineVector.get(i) + "@" +
                           (mxBoardNum+j) + "@" + (rank++));
	    }
          }
	  
        }
	
        else if (divisor > 0) {
          procsPerMachineTable.put( (String) machineVector.get(i),
                                   new Integer(divisor));
	  
          if(DEBUG && logger.isDebugEnabled()) { 
            logger.debug("procPerMachineTable==>" + procsPerMachineTable);
	  }

          //name=(String)machineVector.get(i);
          //name=InetAddress.getByAddress( name.getBytes() ).getHostName();
          for (int j = 0; j < divisor; j++) {
            if(deviceName.equals("niodev")) { 		  
              cout.println( (String) machineVector.get(i) + "@" +
                           (MPJ_SERVER_PORT + (j * 2)) + "@" + (rank++));
	    } else if(deviceName.equals("mxdev")) { 
              cout.println( (String) machineVector.get(i) + "@" +
                           (mxBoardNum+j) + "@" + (rank++));
	    }
          }
        }
      }

    }
    else if (nprocs == noOfMachines) {

      if(DEBUG && logger.isDebugEnabled()) { 
        logger.debug("Processes Requested " + nprocs +
                  " are equal to machines " + noOfMachines);
        logger.debug("Adding a process each into the hashtable");
      }
      
      for (int i = 0; i < nprocs; i++) {
        procsPerMachineTable.put( (String) machineVector.get(i), 
                                  new Integer(1));
	if(deviceName.equals("niodev")) { 
          cout.println( (String) machineVector.get(i) + "@" + MPJ_SERVER_PORT +
                       "@" + (rank++));
	} else if(deviceName.equals("mxdev")) { 
          cout.println( (String) machineVector.get(i) + "@" +
                       (mxBoardNum) + "@" + (rank++));
	}
	
        if(DEBUG && logger.isDebugEnabled()) { 
          logger.debug("procPerMachineTable==>" + procsPerMachineTable);
	}
      }

    }

  }

  private void machinesSanityCheck() throws Exception {
	  
    for(int i=0 ; i<machineVector.size() ; i++) {
	    
      String host = (String) machineVector.get(i) ;

      try {
        InetAddress add = InetAddress.getByName(host);
      } catch( Exception e) {
        throw new MPJRuntimeException (e);	      
      }
      
    }

  }

  /* assume 'machines'is in the current directory */
  public void readMachineFile() throws Exception {

    BufferedReader reader = null;

    try {