writeout.cpp

   1 #include "writeout.h"
   2
   3 #include <sys/types.h>
   4 #include <sys/stat.h>
   5 #include <stdio.h>
   6 #include <stdlib.h>
   7 #include <iostream>
   8 #include <time.h>
   9 #include <sstream>
  10 #include <dirent.h>
  11 #include <errno.h>
  12 #include <unistd.h>
  13
  14 #ifndef MPI_DISABLED
  15 #include <mpi.h>
  16 #endif
  17
  18 using namespace std;
  19
  20 string writeout::longToStr (long arg)
  21 {
  22   stringstream ss;
  23   ss << arg;
  24   return ss.str();
  25 }
  26
  27 void writeout::newsub(string subname) {
  28   of[subname] = new ofstream;
  29
  30   if ( fulldir != "" ) {
  31     if(rank>0) of[subname]->open( (fulldir + "/rank" + cRank + "_" + subname + ".tmp").c_str() );
  32     else of[subname]->open( (fulldir + "/" + signature + "_" + subname + ".dat").c_str() );
  33
  34     if ( !of[subname]->is_open() ) {
  35       logf << "WRITEOUT: Could not open output-file!" << endl << flush;
  36       exit(1);
  37     }
  38
  39     buf[subname] = of[subname]->rdbuf();
  40   }
  41   else
  42     buf[subname] = cout.rdbuf();
  43
  44   out[subname] = new ostream(buf[subname]);
  45 }
  46
  47 writeout::writeout(const string& wdir, const string& _signature,
  48                    const int& _rank, const int& procs)
  49 {
  50   long timestamp;
  51   int iseq=0;
  52
  53   fulldir = "";
  54   signature = _signature;
  55   rank = _rank;
  56
  57   if(wdir != ""){
  58     numprocs = procs;
  59     sprintf(cRank, "%d", rank);
  60
  61     if (rank == 0) {
  62       timestamp = time(NULL);
  63
  64       while (true) {
  65         fulldir = wdir + "/" + longToStr(timestamp) + "." + longToStr(iseq) + "_" + signature + ".tmp";
  66         if ( mkdir(fulldir.c_str(), 0775) == 0 )
  67           break;
  68         else if ( errno != EEXIST ) {
  69           cerr << "WRITEOUT: Could not create outdir!" << endl << flush;
  70           break;
  71         }
  72         iseq++;
  73       }
  74
  75 #ifndef MPI_DISABLED
  76       for(int idest=1; idest<numprocs; idest++) {
  77         MPI_Send(&timestamp, 1, MPI_LONG, idest, 123, MPI_COMM_WORLD);
  78         MPI_Send(&iseq, 1, MPI_LONG, idest, 124, MPI_COMM_WORLD);
  79       }
  80 #endif
  81
  82     }
  83
  84 #ifndef MPI_DISABLED
  85     else if(rank>0) {
  86       MPI_Recv(&timestamp, 1, MPI_LONG, 0, 123, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
  87       MPI_Recv(&iseq, 1, MPI_LONG, 0, 124, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
  88       fulldir = wdir + "/" + longToStr(timestamp) + "." + longToStr(iseq) + "_" + signature + ".tmp";
  89     }
  90 #endif
  91
  92     logf.open( (fulldir + "/rank" + cRank + ".log").c_str() );
  93
  94     if ( !logf.is_open() ) {
  95       cerr << "WRITEOUT: Could not open log-file!" << endl << flush;
  96       exit(1);
  97     }
  98
  99     logf << "[ " << timestring() << " ] Log starts here." << endl;
 100
 101     logbuf = logf.rdbuf();
 102   }
 103   else{
 104     logbuf = cerr.rdbuf();
 105   }
 106   log = new ostream(logbuf);
 107 }
 108
 109 string writeout::timestring()
 110 {
 111   time_t rawtime;
 112   string timestring;
 113   time( &rawtime );
 114   timestring = asctime( localtime( &rawtime ) );
 115   return timestring.substr(0, timestring.size()-1);;
 116 }
 117
 118 writeout::~writeout()
 119 {
 120   if(fulldir != "") {
 121     for (map<string,ofstream*>::iterator ofit = of.begin(); ofit != of.end(); ++ofit) {
 122       if( cRank[0] == '0' ) {
 123         int jobsdone=0;
 124         while(jobsdone<numprocs-1) {
 125           string nextfile;
 126           if( (nextfile = getdatfile(ofit->first)) == "" )
 127             sleep(1);
 128           else {
 129             logf << "collecting " << nextfile << endl;
 130
 131             ifstream myfile( (fulldir + "/" + nextfile).c_str() );
 132             while(true) {
 133               string line;
 134               getline(myfile, line);
 135               if( !myfile.good() ) break;
 136               *ofit->second << line << endl << flush;
 137             }
 138             myfile.close();
 139             remove( (fulldir + "/" + nextfile).c_str() );
 140             jobsdone++;
 141           }
 142         }
 143         *ofit->second << "#end" << endl << flush;
 144         ofit->second->close();
 145       }
 146       else {
 147         ofit->second->close();
 148         rename((fulldir + "/rank" + cRank + "_" + ofit->first + ".tmp").c_str(),
 149                (fulldir + "/rank" + cRank + "_" + ofit->first + ".part").c_str());
 150       }
 151     }
 152     if( cRank[0] == '0' )
 153       rename( fulldir.c_str(), fulldir.substr(0, fulldir.length()-4).c_str() );
 154   }
 155   logf << "[ " << timestring() << " ] Log ends here." << endl;
 156   logf.close();
 157 }
 158
 159 string writeout::getdatfile(string subname)
 160 {
 161   string myfile;
 162   DIR *dp;
 163   struct dirent *dirp;
 164
 165   if((dp  = opendir(fulldir.c_str())) == NULL) {
 166     logf << "Error(" << errno << ") opening " << fulldir << endl;
 167     closedir(dp);
 168     return "";
 169   }
 170
 171   while ((dirp = readdir(dp)) != NULL)
 172     {
 173       myfile = string(dirp->d_name);
 174
 175       if(myfile.length() > 3 && myfile.substr(myfile.length()-4) == "part" &&
 176          subname == myfile.substr( myfile.find("_")+1, myfile.rfind(".")-myfile.find("_")-1 ) ) {
 177         closedir(dp);
 178         return myfile;
 179       }
 180     }
 181
 182   closedir(dp);
 183   return "";
 184 }