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_istreambuf *in;
  24   boost::iostreams::filtering_ostreambuf *out;
  25 };
  26
  27 configcache::configcache(const string& cacheid, const int& nequi, const int& nskip, const string& datadir, char **configmem, const int& configMemSize, const int& cachemode,
  28                          ostream *_log){
  29   log = _log;
  30
  31   NEQUI = nequi;
  32   NSKIP = nskip;
  33   DATADIR = datadir;
  34   CACHEID = cacheid;
  35
  36   if ( cacheid.find("_") != -1 ) {
  37     if(log) *log << "CCACHE: Invalid cacheid \"" << cacheid << "\" given. Cacheids must not contain underscores!" << endl << flush;
  38     exit(1);
  39   }
  40
  41   configMem = (char*)malloc(configMemSize);
  42   tmpConfig = (char*)malloc(configMemSize);
  43
  44   *configmem = configMem;
  45   configSize = configMemSize;
  46
  47   ioBuffers = new iobuffers;
  48   ioBuffers->in = NULL;
  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 int configcache::readHeader()
 155 {
 156   long unsigned int headersize;
 157
 158   if( readDataToMem((char *)&headersize, sizeof(long unsigned int)) == sizeof(long unsigned int) && inFile.is_open() ) {
 159     if ( headersize == 0 )
 160       return HEADER_READLAST;
 161
 162     pair<unsigned long, void *> newHeader;
 163
 164     if( readDataToMem((char *)&newHeader.first, sizeof(unsigned long)) == sizeof(unsigned long) && inFile.is_open() ) {
 165       newHeader.second = malloc(headersize);
 166
 167       if( readDataToMem((char *)newHeader.second, headersize) == headersize && inFile.is_open() ) {
 168         headerStore.push_back(newHeader);
 169         return HEADER_READOK;
 170       }
 171       else {
 172         if(log) *log << "CCACHE: Could not read heade-data! Closing dat-file: " << openFileDesc.filename << endl << flush;
 173         inFile.close();
 174         return HEADER_READERR;
 175       }
 176     }
 177     else {
 178       if(log) *log << "CCACHE: Could not read headerid-hash! Closing dat-file: " << openFileDesc.filename << endl << flush;
 179       inFile.close();
 180       return HEADER_READERR;
 181     }
 182   }
 183   else {
 184     if(log) *log << "CCACHE: Could not read header size. Closing dat-file: " << openFileDesc.filename << endl << flush;
 185     inFile.close();
 186     return HEADER_READERR;
 187   }
 188 }
 189
 190 bool configcache::readAllHeaders()
 191 {
 192   int readHeaderStatus;
 193
 194   deleteHeaderStore();
 195
 196   do {
 197     readHeaderStatus = readHeader();
 198   }
 199   while ( readHeaderStatus == HEADER_READOK );
 200
 201   if ( readHeaderStatus == HEADER_READLAST ) return true;
 202   else if ( readHeaderStatus == HEADER_READERR ) return false;
 203 }
 204
 205 void * configcache::getHeader(const string& headerid) {
 206   for (vector< pair<unsigned long, void *> >::iterator headerStoreIt = headerStore.begin(); headerStoreIt != headerStore.end(); ++headerStoreIt)
 207     if ( headerStoreIt->first == hash(headerid) )
 208       return headerStoreIt->second;
 209
 210   return NULL;
 211 }
 212
 213 void configcache::readConfig(bool *readnewconfig, int *nequileft, vector<unsigned long> *excludeFileHashes)
 214 {
 215   *readnewconfig = false;
 216
 217   if( DATADIR == "" || !(MODE==CACHE_MODE_RO||MODE==CACHE_MODE_RW) ) return;
 218
 219   if(refetchDataFiles){
 220     refetchDataFiles = false;
 221     fetchDataFiles();
 222   }
 223
 224   while(true)
 225     {
 226       vector<infiledesc>::iterator inFileIt = getNextInfile(excludeFileHashes);
 227       int iDidVirtualSkips;
 228
 229       if( (!inFile.is_open()) && inFileIt == inFiles.end() ) {
 230         if (*readnewconfig)
 231           *nequileft = nequileft_internal;
 232         return;
 233       }
 234
 235       while( (!inFile.is_open()) && inFiles.size() > 0 ) {
 236         string inFileParaString;
 237
 238         openFileDesc = *inFileIt;
 239
 240         if (openFileDesc.nequi < NEQUI)
 241           doVirtualEquilibration = true;
 242         else
 243           doVirtualEquilibration = false;
 244
 245         firstUsedConfig = true;
 246
 247         if(log) *log << "CCACHE: Opening dat-file: " << inFileIt->filename << endl << flush;
 248         inFile.open( (DATADIR + "/" + inFileIt->filename).c_str(), std::ios::binary );
 249
 250         if( openFileDesc.superextended ) {
 251           getline( inFile, inFileParaString );
 252           if( inFileParaString != paraString() ) {
 253             if(log) *log << "CCACHE: Parastring does not match. Closing dat-file..." << endl << flush;
 254             inFile.close();
 255           }
 256         }
 257
 258         inFiles.erase(inFileIt);
 259
 260         if( !inFile.is_open() ) continue;
 261
 262         ioBuffers->in = new boost::iostreams::filtering_istreambuf;
 263         ioBuffers->in->push( boost::iostreams::bzip2_decompressor() );
 264         ioBuffers->in->push(inFile);
 265       }
 266
 267       if( inFile.is_open() )
 268         {
 269           if (doVirtualEquilibration) {
 270             if(log) *log << "CCACHE: Trying virtual equilibration." << endl << flush;
 271             doVirtualEquilibration = false;
 272             for (iDidVirtualSkips=0; iDidVirtualSkips < (NEQUI-openFileDesc.nequi)/openFileDesc.nskip; iDidVirtualSkips++) {
 273               if( readFullBlock(tmpConfig, configSize) != configSize || ! inFile.is_open() )
 274                 break;
 275               else if ( (NEQUI-openFileDesc.nequi) - (iDidVirtualSkips+1)*openFileDesc.nskip < nequileft_internal ) {
 276                 memcpy(configMem, tmpConfig, configSize);
 277                 nequileft_internal = NEQUI - openFileDesc.nequi - (iDidVirtualSkips+1)*openFileDesc.nskip;
 278                 *readnewconfig = true;
 279                 firstUsedConfig = false;
 280               }
 281             }
 282           }
 283
 284           if( readFullBlock(tmpConfig, configSize) == configSize && inFile.is_open() )
 285             {
 286               memcpy(configMem, tmpConfig, configSize);
 287               *readnewconfig = true;
 288               if (firstUsedConfig) {
 289                 firstUsedConfig = false;
 290                 if (openFileDesc.nequi < NEQUI)
 291                   nequileft_internal = NEQUI - openFileDesc.nequi - iDidVirtualSkips*openFileDesc.nskip;
 292                 else
 293                   nequileft_internal = NEQUI - openFileDesc.nequi;
 294               }
 295               nequileft_internal -= openFileDesc.nskip;
 296               *nequileft = nequileft_internal;
 297               return;
 298             }
 299           else {
 300             if(log) *log << "CCACHE: Could not read configuration. Closing dat-file: " << openFileDesc.filename << endl << flush;
 301             inFile.close();
 302           }
 303         }
 304     }
 305 }
 306
 307 void configcache::openOutFile(int actnequi)
 308 {
 309   time_t secstamp = time(NULL);
 310   int iseq=0;
 311
 312   while (true) {
 313     outFileName.str("");
 314     outFileName << DATADIR << "/" << secstamp << "." << iseq << "_" << getFileId(actnequi, true, false) << "_.sdat.tmp";
 315
 316     int tmpfd = open(outFileName.str().c_str(), O_CREAT | O_EXCL, 0644);
 317
 318     if ( tmpfd != -1 ) {
 319       close(tmpfd);
 320       break;
 321     }
 322     else if ( errno != EEXIST ) {
 323       if(log) *log << "CCACHE: Could not create cachefile!" << endl << flush;
 324       exit(1);
 325     }
 326
 327     iseq++;
 328   }
 329
 330   outFile.open( outFileName.str().c_str(), std::ios::binary );
 331
 332   outFile << paraString() << endl;
 333
 334   ioBuffers->out = new boost::iostreams::filtering_ostreambuf;
 335   ioBuffers->out->push(boost::iostreams::bzip2_compressor());
 336   ioBuffers->out->push(outFile);
 337 }
 338
 339 void configcache::writeHeader(const string& headerid, const char *header, long unsigned int size, int actnequi) {
 340   unsigned long headeridhash;
 341
 342   if( DATADIR == "" || !(MODE==CACHE_MODE_WO||MODE==CACHE_MODE_RW) ) return;
 343
 344   if(!outFile.is_open())
 345     openOutFile(actnequi);
 346
 347   headeridhash = hash(headerid);
 348
 349   boost::iostreams::write(*ioBuffers->out, (char*)&size, sizeof(long unsigned int));
 350   boost::iostreams::write(*ioBuffers->out, (char*)&headeridhash, sizeof(unsigned long));
 351   boost::iostreams::write(*ioBuffers->out, header, size);
 352 }
 353
 354 void configcache::writeConfig(int actnequi)
 355 {
 356   long unsigned int zeroheader=0;
 357
 358   if ( DATADIR == "" || !(MODE==CACHE_MODE_WO||MODE==CACHE_MODE_RW) ) return;
 359
 360   if ( ! outFile.is_open() )
 361     openOutFile(actnequi);
 362
 363   boost::iostreams::write(*ioBuffers->out, (char*)&zeroheader, sizeof(long unsigned int));
 364
 365   boost::iostreams::write(*ioBuffers->out, configMem, configSize);
 366 }
 367
 368 void configcache::addPara(const string& parid, const double& val){
 369   parameter newPara;
 370   newPara.id = parid;
 371   newPara.val = val;
 372   Paras.push_back(newPara);
 373 }
 374
 375 int configcache::getParIndex(const string& parid){
 376   for(int ipara=0; ipara<Paras.size(); ipara++)
 377     if(Paras[ipara].id == parid) return ipara;
 378 }
 379
 380 void configcache::setPara(const string& parid, const double& value){
 381   Paras[getParIndex(parid)].val = value;
 382
 383   finishOutFile();
 384   if(ioBuffers->in != NULL) { delete ioBuffers->in; ioBuffers->in=NULL; }
 385   inFile.close();
 386   inFiles.clear();
 387
 388   refetchDataFiles = true;
 389   nequileft_internal = NEQUI;
 390 }
 391
 392 configcache::~configcache()
 393 {
 394   finishOutFile();
 395   delete ioBuffers->in;
 396   ioBuffers->in = NULL;
 397 }
 398
 399 void configcache::finishOutFile()
 400 {
 401   if( ioBuffers->out != NULL )
 402     {
 403       delete ioBuffers->out;
 404       ioBuffers->out = NULL;
 405     }
 406
 407   if( outFile.is_open() )
 408     {
 409       outFile.close();
 410       rename( outFileName.str().c_str(), outFileName.str().substr(0, outFileName.str().size()-4).c_str() );
 411     }
 412 }
 413
 414 int configcache::readFullBlock(char *tmpData, long unsigned int dataSize)
 415 {
 416   /* try to read header */
 417   if ( openFileDesc.extended )
 418     if ( ! readAllHeaders() )
 419       return -1;
 420
 421   /* read data */
 422   return readDataToMem(tmpData, dataSize);
 423 }
 424
 425 int configcache::readDataToMem(char *tmpData, long unsigned int dataSize)
 426 {
 427   int readturn = -1;
 428
 429   if ( dataSize == 0 ) return 0;
 430
 431   try { readturn = boost::iostreams::read(*ioBuffers->in, tmpData, dataSize); }
 432   catch(boost::iostreams::bzip2_error& error) {
 433     if(log) *log << "CCACHE: Caught bzip2 exception with error code: " << error.error() << endl << flush;
 434     inFile.close();
 435   }
 436   catch (std::exception const& ex) {
 437     if(log) *log << "CCACHE: Caught exception: " << ex.what() << endl << flush;
 438     inFile.close();
 439   }
 440   catch( ... ) {
 441     if(log) *log << "CCACHE: Caught unknown exception while reading." << endl << flush;
 442     inFile.close();
 443   }
 444
 445   return readturn;
 446 }
 447
 448 unsigned long configcache::hash(const string& str)
 449 {
 450   unsigned long hash = 5381;
 451
 452   for(string::const_iterator it=str.begin();it!=str.end();it++)
 453     hash = ((hash << 5) + hash) + *it; /* hash * 33 + character */
 454
 455   return hash;
 456 }
 457
 458 void configcache::deleteHeaderStore()
 459 {
 460   while ( headerStore.size() > 0 ) {
 461     free(headerStore.back().second);
 462     headerStore.pop_back();
 463   }
 464 }
 465
 466 vector<infiledesc>::iterator configcache::getNextInfile(vector<unsigned long> *excludeFileHashes) {
 467   for (vector<infiledesc>::iterator init = inFiles.begin(); init != inFiles.end(); ++init) {
 468     if (excludeFileHashes != NULL) {
 469       bool excludethisfile = false;
 470
 471       for (vector<unsigned long>::iterator exit = excludeFileHashes->begin(); exit != excludeFileHashes->end(); ++exit)
 472         if ( *exit == hash(init->filename) ) {
 473           excludethisfile = true;
 474           break;
 475         }
 476
 477       if (excludethisfile)
 478         continue;
 479     }
 480     return init;
 481   }
 482   return inFiles.end();
 483 }