source: trunk/oar/mpilauncher.cpp @ 24

#include <sys/types.h>
#include <dirent.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <vector>
#include <string>
#include <iostream>
#include <fstream>

#include <mpi.h>

using namespace std;

/*
 * These two variables sets the command to launch in each sub-directory
 * and the arguments required. The name of the root directory of the
 * datasets is given as an argument when starting the MPIlauncher. const
 * string
 * commandToLaunch("$HOME/SOURCES/MES-PROJETS/MPILAUNCHER/appli.exe");
 * const string("-l -info");
 */

/*
 * getListOfDir(string rep, vector<string> &files)
 *
 * Gets the list of tjhe subdirectories in the directory rep and returns
 * them in a vector array of strings. Directories ";" and ".." are removed
 * from the vector array.
 */
int getListOfDir(string rep, vector < string > &files) {
   DIR *dir;
   struct dirent *ent;
   int count;

   count = 0;
   dir = opendir(rep.c_str());
   if (dir != NULL) {
      /*
       * search for the files within directory
       */
      while ((ent = readdir(dir)) != NULL)
         if (ent->d_type == DT_DIR) {
            if (strcmp(ent->d_name, ".") *
                  strcmp(ent->d_name, "..")) {
               count++;
               files.push_back(string(ent->d_name));
               }
            }
      closedir(dir);
      }
   else {
      cerr << "Directory " << rep.c_str() << " not found" << endl;
      }
   return count;
   }

/*
 * getListOfCommand(string rep, vector<string> &files) Gets the list of
 * commands in the ascii file fich. One command per line (no wrap) in this
 * first version.
 */
int getListOfCommand(const string & fich, vector < string > &commands) {
   int count;
   string line;

   count = 0;
   commands.clear();

   std::ifstream infile(fich.c_str(), std::ios_base::in);
   while (getline(infile, line, '\n')) {
      // remove all trailing blanks
      while (line.size() > 0 && isspace(line[line.size() - 1]))
         line.erase(line.size() - 1);

      // no empty line
      if (line.size() > 0) {
         commands.push_back(line);
         count++;
         }
      }
   return count;
   }

/*
 * Main program
 */
int main(int argc, char **argv) {
   vector < string > files = vector < string > ();
   string rep;          // root directory where to find
   // subdirectories
   int rank, size, nbdir, stride;
   int nbcmd;
   /*
    * These two variables sets the command to launch in each
    * sub-directory and the arguments required. The name of the root
    * directory of the datasets is given as an argument when starting the
    * MPIlauncher.
    */
   string commandToLaunch;
   string arguments;
   string commandfile;

   string finalCommand; // command to execute

   MPI_Init(&argc, &argv);      /* starts MPI */
   MPI_Comm_rank(MPI_COMM_WORLD, &rank);        /* get current process id */
   MPI_Comm_size(MPI_COMM_WORLD, &size);        /* get number of processes
                                                 */

   if (argc < 3) {
      cout << "USAGE:" << endl;
      cout << argv[0] << " root_directory  command [arguments]" <<
           endl;
      cout << argv[0] << " -f  command_file" << endl;
      }
   else {
      if (strcmp(argv[1], "-f") == 0) {
         /*
          * processing a command file now.
          */
         commandfile = argv[2];
         nbcmd = getListOfCommand(commandfile, files);
         // if (rank==0){
         // cout<<"Command file is "<<commandfile<<endl;
         // for (unsigned int i = 0;i < nbcmd;i++)
         // cout<<"["<<files[i]<<"]"<<endl;
         // }

         // Number of commands should divide by number of cpus
         // if (nbcmd%size != 0) {
         // if(rank ==0)
         // cerr <<nbcmd<<" command(s) to process in "<<commandfile
         // <<" cannot fit on "<<size<<" processe(s)"
         // <<endl<<"FAILED"<<endl;
         // }
         // else {
         // execute the command
         int reste;
         stride = nbcmd / size;
         for (unsigned int i = 0; i < stride; i++) {
            cerr << "Process " << rank << " execute " <<
                 files[stride * rank + i] << endl;
            system(files[stride * rank + i].c_str());
            }
         // remaining command lines
         reste = nbcmd - stride * size;
         if (rank > 0 && rank <= reste) {
            cerr << "Process " << rank << " execute " <<
                 files[nbcmd - rank] << endl;
            system(files[nbcmd - rank].c_str());
            }
         // }
         }
      else {
         /*
          * processing a list of dir now
          */
         rep = string(argv[1]);
         commandToLaunch = string(argv[2]);
         for (int i = 3; i < argc; i++)
            (arguments += argv[i]) += " ";

         nbdir = getListOfDir(rep, files);

         // Number of dir should divide by number of cpus
         if (nbdir % size != 0) {
            if (rank == 0)
               cerr << nbdir <<
                    " dataset(s) to process cannot fit on "
                    << size << " processes" << endl <<
                    "FAILED" << endl;
            }
         else {
            // execute the command
            stride = nbdir / size;
            for (unsigned int i = 0; i < stride; i++) {
               string finalCommand("cd ");
               finalCommand += rep;
               finalCommand += "/";
               finalCommand +=
                  files[stride * rank + i];
               finalCommand += ";";
               finalCommand += commandToLaunch;
               finalCommand += " ";
               finalCommand += arguments;
               // cout<<"On "<<rank<<" execute "<<finalCommand<<endl;
               system(finalCommand.c_str());
               }
            }
         }
      }
   MPI_Finalize();

   return 0;
   }

/*
 * documentation POD format
 *
 * =head1 NAME
 *
 *  mpilauncher - parallel execute lot of small job via mpi
 *
 * =head1 SYNOPSIS
 *
 *  mpilauncher root_folder command [args]
 *  mpilauncher -f command_file
 *
 * =cut
 *
 */