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 #include <boost/filesystem.hpp>
  15
  16 using namespace std;
  17
  18 string writeout::tstamp(const long& timestamp)
  19 {
  20   stringstream sstr;
  21   if(!timestamp) sstr << time (NULL);
  22   else sstr << timestamp;
  23   return sstr.str();
  24 }
  25
  26 void writeout::newsub(string subname) {
  27   of[subname] = new ofstream;
  28
  29   if ( fulldir != "" ) {
  30     if(rank>0) of[subname]->open( (fulldir + "/rank" + cRank + "-" + subname + ".tmp").c_str() );
  31     else of[subname]->open( (fulldir + "/" + signature + "-" + subname + ".dat").c_str() );
  32
  33     if ( !of[subname]->is_open() ) {
  34       logf << "WRITEOUT: Could not open output-file!" << endl << flush;
  35       exit(1);
  36     }
  37
  38     buf[subname] = of[subname]->rdbuf();
  39   }
  40   else
  41     buf[subname] = cout.rdbuf();
  42
  43   out[subname] = new ostream(buf[subname]);
  44 }
  45
  46 writeout::writeout(const string& wdir, const string& _signature,
  47                    const int& _rank, const int& procs, const long& timestamp)
  48 {
  49   fulldir = "";
  50   signature = _signature;
  51   rank = _rank;
  52
  53   if(wdir != ""){
  54     numprocs = procs;
  55     sprintf(cRank, "%d", rank);
  56     fulldir = wdir + "/" + tstamp(timestamp) + "_" + signature + ".tmp";
  57
  58     if (rank==0)
  59       mkdir(fulldir.c_str(), 0775);
  60     else {
  61       while ( ! ( boost::filesystem::exists(fulldir) && boost::filesystem::is_directory(fulldir) ) ) {
  62         cerr << "WRITOUT: Waiting for rank 0 to create out directory." << endl << flush;
  63         sleep(1);
  64       }
  65     }
  66
  67     logf.open( (fulldir + "/rank" + cRank + ".log").c_str() );
  68
  69     if ( !logf.is_open() ) {
  70       cerr << "WRITEOUT: Could not open log-file!" << endl << flush;
  71       exit(1);
  72     }
  73
  74     logf << "[ " << timestring() << " ] Log starts here." << endl;
  75
  76     logbuf = logf.rdbuf();
  77   }
  78   else{
  79     logbuf = cerr.rdbuf();
  80   }
  81   log = new ostream(logbuf);
  82 }
  83
  84 string writeout::timestring()
  85 {
  86   time_t rawtime;
  87   string timestring;
  88   time( &rawtime );
  89   timestring = asctime( localtime( &rawtime ) );
  90   return timestring.substr(0, timestring.size()-1);;
  91 }
  92
  93 writeout::~writeout()
  94 {
  95   if(fulldir != "") {
  96     for (map<string,ofstream*>::iterator ofit = of.begin(); ofit != of.end(); ++ofit) {
  97       if( cRank[0] == '0' ) {
  98         int jobsdone=0;
  99         while(jobsdone<numprocs-1) {
 100           string nextfile;
 101           if( (nextfile = getdatfile(ofit->first)) == "" )
 102             sleep(1);
 103           else {
 104             logf << "collecting " << nextfile << endl;
 105
 106             ifstream myfile( (fulldir + "/" + nextfile).c_str() );
 107             while(true) {
 108               string line;
 109               getline(myfile, line);
 110               if( !myfile.good() ) break;
 111               *ofit->second << line << endl << flush;
 112             }
 113             myfile.close();
 114             remove( (fulldir + "/" + nextfile).c_str() );
 115             jobsdone++;
 116           }
 117         }
 118         *ofit->second << "#end" << endl << flush;
 119         ofit->second->close();
 120       }
 121       else {
 122         ofit->second->close();
 123         rename((fulldir + "/rank" + cRank + "-" + ofit->first + ".tmp").c_str(),
 124                (fulldir + "/rank" + cRank + "-" + ofit->first + ".part").c_str());
 125       }
 126     }
 127     if( cRank[0] == '0' )
 128       rename( fulldir.c_str(), fulldir.substr(0, fulldir.length()-4).c_str() );
 129   }
 130   logf << "[ " << timestring() << " ] Log ends here." << endl;
 131   logf.close();
 132 }
 133
 134 string writeout::getdatfile(string subname)
 135 {
 136   string myfile;
 137   DIR *dp;
 138   struct dirent *dirp;
 139
 140   if((dp  = opendir(fulldir.c_str())) == NULL) {
 141     logf << "Error(" << errno << ") opening " << fulldir << endl;
 142     closedir(dp);
 143     return "";
 144   }
 145
 146   while ((dirp = readdir(dp)) != NULL)
 147     {
 148       myfile = string(dirp->d_name);
 149
 150       if(myfile.length() > 3 && myfile.substr(myfile.length()-4) == "part" &&
 151          subname == myfile.substr( myfile.find("-")+1, myfile.rfind(".")-myfile.find("-")-1 ) ) {
 152         closedir(dp);
 153         return myfile;
 154       }
 155     }
 156
 157   closedir(dp);
 158   return "";
 159 }