configcache.cpp

   1 #include "configcache.h"
   2
   3 #include <stdlib.h>
   4 #include <iostream>
   5 #include <time.h>
   6 #include <dirent.h>
   7 #include <errno.h>
   8 #include <sys/stat.h>
   9 #include <fcntl.h>
  10
  11 #include <boost/iostreams/filtering_streambuf.hpp>
  12 #include <boost/iostreams/stream.hpp>
  13 #include <boost/iostreams/filter/bzip2.hpp>
  14 #include <boost/iostreams/device/array.hpp>
  15 #include <boost/iostreams/copy.hpp>
  16
  17 #define HEADER_READOK   0
  18 #define HEADER_READERR  1
  19 #define HEADER_READLAST 2
  20
  21 struct configcache::iobuffers
  22 {
  23   boost::iostreams::filtering_ostreambuf *out;
  24 };
  25
  26 configcache::configcache(const string& cacheid, const int& nequi, const int& nskip, const string& datadir, char **configmem, const int& configMemSize, const int& cachemode,
  27                          ostream *_log){
  28   log = _log;
  29
  30   dataReader = new datread(configMemSize, log);
  31
  32   NEQUI = nequi;
  33   NSKIP = nskip;
  34   DATADIR = datadir;
  35   CACHEID = cacheid;
  36
  37   if ( cacheid.find("_") != -1 ) {
  38     if(log) *log << "CCACHE: Invalid cacheid \"" << cacheid << "\" given. Cacheids must not contain underscores!" << endl << flush;
  39     exit(1);
  40   }
  41
  42   configMem = (char*)malloc(configMemSize);
  43   tmpConfig = (char*)malloc(configMemSize);
  44
  45   *configmem = configMem;
  46   configSize = configMemSize;
  47
  48   ioBuffers = new iobuffers;
  49   ioBuffers->out = NULL;
  50
  51   MODE = cachemode;
  52
  53   refetchDataFiles = false;
  54 }
  55
  56 string configcache::paraString() {
  57   stringstream parastring;
  58
  59   for(int ipara=0; ipara<Paras.size(); ipara++)
  60     parastring << "_" << Paras[ipara].id << Paras[ipara].val;
  61
  62   return parastring.str();
  63 }
  64
  65 string configcache::getFileId(int actnequi, const bool& superextended, const bool& shortid)
  66 {
  67   stringstream fileid;
  68
  69   if(!shortid) fileid << CACHEID << "_" << actnequi << "_" << NSKIP;
  70
  71   if( superextended )
  72     fileid << "_" << hash( paraString() );
  73   else
  74     fileid << paraString();
  75
  76   return fileid.str();
  77 }
  78
  79 void configcache::fetchDataFiles()
  80 {
  81   struct dirent *de=NULL;
  82   DIR *d=NULL;
  83   static infiledesc filedesc;
  84
  85   d=opendir(DATADIR.c_str());
  86   if(d != NULL){
  87     while(de = readdir(d)){
  88       string filename = de->d_name;
  89       if(isValidInFile(filename, &filedesc))
  90         {
  91           inFiles.push_back(filedesc);
  92         }
  93     }
  94   }
  95 }
  96
  97 bool configcache::isValidInFile(const string& infile, infiledesc *filedesc)
  98 {
  99   char *inchar, *inParts;
 100   string truncIn, truncOut;
 101
 102   filedesc->filename = infile;
 103
 104   if( infile.size() < 4 ) return false;
 105
 106   if( infile.substr(infile.size()-4) == ".dat" ) {
 107     filedesc->extended = false;
 108     filedesc->superextended = false;
 109   }
 110   else if( infile.substr(infile.size()-4) == "edat" ) {
 111     filedesc->extended = true;
 112     filedesc->superextended = false;
 113   }
 114   else if( infile.substr(infile.size()-4) == "sdat" ) {
 115     filedesc->extended = true;
 116     filedesc->superextended = true;
 117   }
 118   else
 119     return false;
 120
 121   inchar = new char [infile.size()+1];
 122   strcpy (inchar, infile.c_str());
 123
 124   inParts = strtok( inchar, "_" );
 125   for(int iPart=0; inParts!=NULL; iPart++)
 126     {
 127       if( iPart>3 ) { truncIn += "_"; truncIn += inParts; }
 128
 129       switch(iPart)
 130         {
 131         case 1: if(inParts != CACHEID)
 132             return false;
 133           break;
 134         case 2:
 135           filedesc->nequi = atoi(inParts);
 136           break;
 137         case 3:
 138           if(atoi(inParts) != NSKIP)
 139             return false;
 140           filedesc->nskip = atoi(inParts);
 141           break;
 142         }
 143       inParts = strtok( NULL, "_");
 144     }
 145   truncIn = truncIn.substr(0, truncIn.size()-4);
 146
 147   delete[] inchar;
 148
 149   if( truncIn.find( getFileId(NEQUI, filedesc->superextended, true) + "_" ) == string::npos ) return false;
 150
 151   return true;
 152 }
 153
 154 void * configcache::getHeader(const string& headerid) {
 155   return dataReader->getHeader(headerid);
 156 }
 157
 158 void configcache::readConfig(bool *readnewconfig, int *nequileft, vector<unsigned long> *excludeFileHashes)
 159 {
 160   *readnewconfig = false;
 161
 162   if( DATADIR == "" || !(MODE==CACHE_MODE_RO||MODE==CACHE_MODE_RW) ) return;
 163
 164   if(refetchDataFiles){
 165     refetchDataFiles = false;
 166     fetchDataFiles();
 167   }
 168
 169   while(true)
 170     {
 171       vector<infiledesc>::iterator inFileIt = getNextInfile(excludeFileHashes);
 172       int iDidVirtualSkips;
 173
 174       if( (!dataReader->fisopen()) && inFileIt == inFiles.end() ) {
 175         if (*readnewconfig)
 176           *nequileft = nequileft_internal;
 177         return;
 178       }
 179
 180       while( (!dataReader->fisopen()) && inFiles.size() > 0 ) {
 181         string inFileParaString;
 182
 183         openFileDesc = *inFileIt;
 184
 185         if (openFileDesc.nequi < NEQUI)
 186           doVirtualEquilibration = true;
 187         else
 188           doVirtualEquilibration = false;
 189
 190         firstUsedConfig = true;
 191
 192         if(log) *log << "CCACHE: Opening dat-file: " << inFileIt->filename << endl << flush;
 193
 194         dataReader->openFile( DATADIR + "/" + inFileIt->filename );
 195
 196         if( openFileDesc.superextended ) {
 197           if( dataReader->getParaString() != paraString() ) {
 198             if(log) *log << "CCACHE: Parastring does not match. Closing dat-file..." << endl << flush;
 199             dataReader->closeFile();
 200           }
 201         }
 202
 203         inFiles.erase(inFileIt);
 204
 205         if( !dataReader->fisopen() ) continue;
 206       }
 207
 208       if( dataReader->fisopen() )
 209         {
 210           if (doVirtualEquilibration) {
 211             if(log) *log << "CCACHE: Trying virtual equilibration." << endl << flush;
 212             doVirtualEquilibration = false;
 213             for (iDidVirtualSkips=0; iDidVirtualSkips < (NEQUI-openFileDesc.nequi)/openFileDesc.nskip; iDidVirtualSkips++) {
 214               if( dataReader->readFullBlock(tmpConfig) < 0 )
 215                 break;
 216               else if ( (NEQUI-openFileDesc.nequi) - (iDidVirtualSkips+1)*openFileDesc.nskip < nequileft_internal ) {
 217                 memcpy(configMem, tmpConfig, configSize);
 218                 nequileft_internal = NEQUI - openFileDesc.nequi - (iDidVirtualSkips+1)*openFileDesc.nskip;
 219                 *readnewconfig = true;
 220                 firstUsedConfig = false;
 221               }
 222             }
 223           }
 224
 225           if( dataReader->readFullBlock(tmpConfig) >= 0 )
 226             {
 227               memcpy(configMem, tmpConfig, configSize);
 228               *readnewconfig = true;
 229               if (firstUsedConfig) {
 230                 firstUsedConfig = false;
 231                 if (openFileDesc.nequi < NEQUI)
 232                   nequileft_internal = NEQUI - openFileDesc.nequi - iDidVirtualSkips*openFileDesc.nskip;
 233                 else
 234                   nequileft_internal = NEQUI - openFileDesc.nequi;
 235               }
 236               nequileft_internal -= openFileDesc.nskip;
 237               *nequileft = nequileft_internal;
 238               return;
 239             }
 240         }
 241     }
 242 }
 243
 244 void configcache::openOutFile(int actnequi)
 245 {
 246   time_t secstamp = time(NULL);
 247   int iseq=0;
 248
 249   while (true) {
 250     outFileName.str("");
 251     outFileName << DATADIR << "/" << secstamp << "." << iseq << "_" << getFileId(actnequi, true, false) << "_.sdat.tmp";
 252
 253     int tmpfd = open(outFileName.str().c_str(), O_CREAT | O_EXCL, 0644);
 254
 255     if ( tmpfd != -1 ) {
 256       close(tmpfd);
 257       break;
 258     }
 259     else if ( errno != EEXIST ) {
 260       if(log) *log << "CCACHE: Could not create cachefile!" << endl << flush;
 261       exit(1);
 262     }
 263
 264     iseq++;
 265   }
 266
 267   outFile.open( outFileName.str().c_str(), std::ios::binary );
 268
 269   outFile << paraString() << endl;
 270
 271   ioBuffers->out = new boost::iostreams::filtering_ostreambuf;
 272   ioBuffers->out->push(boost::iostreams::bzip2_compressor());
 273   ioBuffers->out->push(outFile);
 274 }
 275
 276 void configcache::writeHeader(const string& headerid, const char *header, long unsigned int size, int actnequi) {
 277   unsigned long headeridhash;
 278
 279   if( DATADIR == "" || !(MODE==CACHE_MODE_WO||MODE==CACHE_MODE_RW) ) return;
 280
 281   if(!outFile.is_open())
 282     openOutFile(actnequi);
 283
 284   headeridhash = hash(headerid);
 285
 286   boost::iostreams::write(*ioBuffers->out, (char*)&size, sizeof(long unsigned int));
 287   boost::iostreams::write(*ioBuffers->out, (char*)&headeridhash, sizeof(unsigned long));
 288   boost::iostreams::write(*ioBuffers->out, header, size);
 289 }
 290
 291 void configcache::writeConfig(int actnequi)
 292 {
 293   long unsigned int zeroheader=0;
 294
 295   if ( DATADIR == "" || !(MODE==CACHE_MODE_WO||MODE==CACHE_MODE_RW) ) return;
 296
 297   if ( ! outFile.is_open() )
 298     openOutFile(actnequi);
 299
 300   boost::iostreams::write(*ioBuffers->out, (char*)&zeroheader, sizeof(long unsigned int));
 301
 302   boost::iostreams::write(*ioBuffers->out, configMem, configSize);
 303 }
 304
 305 void configcache::addPara(const string& parid, const double& val){
 306   parameter newPara;
 307   newPara.id = parid;
 308   newPara.val = val;
 309   Paras.push_back(newPara);
 310 }
 311
 312 int configcache::getParIndex(const string& parid){
 313   for(int ipara=0; ipara<Paras.size(); ipara++)
 314     if(Paras[ipara].id == parid) return ipara;
 315 }
 316
 317 void configcache::setPara(const string& parid, const double& value){
 318   Paras[getParIndex(parid)].val = value;
 319
 320   finishOutFile();
 321   dataReader->closeFile();
 322   inFiles.clear();
 323
 324   refetchDataFiles = true;
 325   nequileft_internal = NEQUI;
 326 }
 327
 328 configcache::~configcache()
 329 {
 330   finishOutFile();
 331   delete dataReader;
 332 }
 333
 334 void configcache::finishOutFile()
 335 {
 336   if( ioBuffers->out != NULL )
 337     {
 338       delete ioBuffers->out;
 339       ioBuffers->out = NULL;
 340     }
 341
 342   if( outFile.is_open() )
 343     {
 344       outFile.close();
 345       rename( outFileName.str().c_str(), outFileName.str().substr(0, outFileName.str().size()-4).c_str() );
 346     }
 347 }
 348
 349 unsigned long configcache::hash(const string& str)
 350 {
 351   unsigned long hash = 5381;
 352
 353   for(string::const_iterator it=str.begin();it!=str.end();it++)
 354     hash = ((hash << 5) + hash) + *it; /* hash * 33 + character */
 355
 356   return hash;
 357 }
 358
 359 vector<infiledesc>::iterator configcache::getNextInfile(vector<unsigned long> *excludeFileHashes) {
 360   for (vector<infiledesc>::iterator init = inFiles.begin(); init != inFiles.end(); ++init) {
 361     if (excludeFileHashes != NULL) {
 362       bool excludethisfile = false;
 363
 364       for (vector<unsigned long>::iterator exit = excludeFileHashes->begin(); exit != excludeFileHashes->end(); ++exit)
 365         if ( *exit == hash(init->filename) ) {
 366           excludethisfile = true;
 367           break;
 368         }
 369
 370       if (excludethisfile)
 371         continue;
 372     }
 373     return init;
 374   }
 375   return inFiles.end();
 376 }