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& shortid)
  66 {
  67   stringstream fileid;
  68
  69   if(!shortid) fileid << CACHEID << "_" << actnequi << "_" << NSKIP;
  70
  71   fileid << "_" << hash( paraString() );
  72
  73   return fileid.str();
  74 }
  75
  76 void configcache::fetchDataFiles()
  77 {
  78   struct dirent *de=NULL;
  79   DIR *d=NULL;
  80   static infiledesc filedesc;
  81
  82   d=opendir(DATADIR.c_str());
  83   if(d != NULL){
  84     while(de = readdir(d)){
  85       string filename = de->d_name;
  86       if(isValidInFile(filename, &filedesc))
  87         {
  88           inFiles.push_back(filedesc);
  89         }
  90     }
  91   }
  92 }
  93
  94 bool configcache::isValidInFile(const string& infile, infiledesc *filedesc)
  95 {
  96   char *inchar, *inParts;
  97   string truncIn, truncOut;
  98
  99   filedesc->filename = infile;
 100
 101   if( infile.size() < 4 ) return false;
 102
 103   if( infile.substr(infile.size()-4) == ".dat" )
 104     filedesc->extended = false;
 105   else if( infile.substr(infile.size()-4) == "edat" )
 106     filedesc->extended = true;
 107   else
 108     return false;
 109
 110   inchar = new char [infile.size()+1];
 111   strcpy (inchar, infile.c_str());
 112
 113   inParts = strtok( inchar, "_" );
 114   for(int iPart=0; inParts!=NULL; iPart++)
 115     {
 116       if( iPart>3 ) { truncIn += "_"; truncIn += inParts; }
 117
 118       switch(iPart)
 119         {
 120         case 1: if(inParts != CACHEID)
 121             return false;
 122           break;
 123         case 2:
 124           filedesc->nequi = atoi(inParts);
 125           break;
 126         case 3:
 127           if(atoi(inParts) != NSKIP)
 128             return false;
 129           filedesc->nskip = atoi(inParts);
 130           break;
 131         }
 132       inParts = strtok( NULL, "_");
 133     }
 134   truncIn = truncIn.substr(0, truncIn.size()-4);
 135
 136   delete[] inchar;
 137
 138   if( truncIn.find( getFileId(NEQUI, true) + "_" ) == string::npos ) return false;
 139
 140   return true;
 141 }
 142
 143 int configcache::readHeader()
 144 {
 145   long unsigned int headersize;
 146
 147   if( readDataToMem((char *)&headersize, sizeof(long unsigned int)) == sizeof(long unsigned int) && inFile.is_open() ) {
 148     if ( headersize == 0 )
 149       return HEADER_READLAST;
 150
 151     pair<unsigned long, void *> newHeader;
 152
 153     if( readDataToMem((char *)&newHeader.first, sizeof(unsigned long)) == sizeof(unsigned long) && inFile.is_open() ) {
 154       newHeader.second = malloc(headersize);
 155
 156       if( readDataToMem((char *)newHeader.second, headersize) == headersize && inFile.is_open() ) {
 157         headerStore.push_back(newHeader);
 158         return HEADER_READOK;
 159       }
 160       else {
 161         if(log) *log << "CCACHE: Could not read heade-data! Closing dat-file: " << openFileDesc.filename << endl << flush;
 162         inFile.close();
 163         return HEADER_READERR;
 164       }
 165     }
 166     else {
 167       if(log) *log << "CCACHE: Could not read headerid-hash! Closing dat-file: " << openFileDesc.filename << endl << flush;
 168       inFile.close();
 169       return HEADER_READERR;
 170     }
 171   }
 172   else {
 173     if(log) *log << "CCACHE: Could not read header size. Closing dat-file: " << openFileDesc.filename << endl << flush;
 174     inFile.close();
 175     return HEADER_READERR;
 176   }
 177 }
 178
 179 bool configcache::readAllHeaders()
 180 {
 181   int readHeaderStatus;
 182
 183   deleteHeaderStore();
 184
 185   do {
 186     readHeaderStatus = readHeader();
 187   }
 188   while ( readHeaderStatus == HEADER_READOK );
 189
 190   if ( readHeaderStatus == HEADER_READLAST ) return true;
 191   else if ( readHeaderStatus == HEADER_READERR ) return false;
 192 }
 193
 194 void * configcache::getHeader(const string& headerid) {
 195   for (vector< pair<unsigned long, void *> >::iterator headerStoreIt = headerStore.begin(); headerStoreIt != headerStore.end(); ++headerStoreIt)
 196     if ( headerStoreIt->first == hash(headerid) )
 197       return headerStoreIt->second;
 198
 199   return NULL;
 200 }
 201
 202 void configcache::readConfig(bool *readnewconfig, int *nequileft, vector<unsigned long> *excludeFileHashes)
 203 {
 204   *readnewconfig = false;
 205
 206   if( DATADIR == "" || !(MODE==CACHE_MODE_RO||MODE==CACHE_MODE_RW) ) return;
 207
 208   if(refetchDataFiles){
 209     refetchDataFiles = false;
 210     fetchDataFiles();
 211   }
 212
 213   while(true)
 214     {
 215       vector<infiledesc>::iterator inFileIt = getNextInfile(excludeFileHashes);
 216       int iDidVirtualSkips;
 217
 218       if( (!inFile.is_open()) && inFileIt == inFiles.end() ) {
 219         if (*readnewconfig)
 220           *nequileft = nequileft_internal;
 221         return;
 222       }
 223
 224       while( (!inFile.is_open()) && inFiles.size() > 0 ) {
 225         string inFileParaString;
 226
 227         openFileDesc = *inFileIt;
 228
 229         if (openFileDesc.nequi < NEQUI)
 230           doVirtualEquilibration = true;
 231         else
 232           doVirtualEquilibration = false;
 233
 234         firstUsedConfig = true;
 235
 236         if(log) *log << "CCACHE: Opening dat-file: " << inFileIt->filename << endl << flush;
 237         inFile.open( (DATADIR + "/" + inFileIt->filename).c_str(), std::ios::binary );
 238
 239         getline( inFile, inFileParaString );
 240         if( inFileParaString != paraString() ) {
 241           if(log) *log << "CCACHE: Parastring does not match. Closing dat-file..." << endl << flush;
 242           inFile.close();
 243         }
 244
 245         inFiles.erase(inFileIt);
 246
 247         if( !inFile.is_open() ) continue;
 248
 249         ioBuffers->in = new boost::iostreams::filtering_istreambuf;
 250         ioBuffers->in->push( boost::iostreams::bzip2_decompressor() );
 251         ioBuffers->in->push(inFile);
 252       }
 253
 254       if( inFile.is_open() )
 255         {
 256           if (doVirtualEquilibration) {
 257             if(log) *log << "CCACHE: Trying virtual equilibration." << endl << flush;
 258             doVirtualEquilibration = false;
 259             for (iDidVirtualSkips=0; iDidVirtualSkips < (NEQUI-openFileDesc.nequi)/openFileDesc.nskip; iDidVirtualSkips++) {
 260               if( readFullBlock(tmpConfig, configSize) != configSize || ! inFile.is_open() )
 261                 break;
 262               else if ( (NEQUI-openFileDesc.nequi) - (iDidVirtualSkips+1)*openFileDesc.nskip < nequileft_internal ) {
 263                 memcpy(configMem, tmpConfig, configSize);
 264                 nequileft_internal = NEQUI - openFileDesc.nequi - (iDidVirtualSkips+1)*openFileDesc.nskip;
 265                 *readnewconfig = true;
 266                 firstUsedConfig = false;
 267               }
 268             }
 269           }
 270
 271           if( readFullBlock(tmpConfig, configSize) == configSize && inFile.is_open() )
 272             {
 273               memcpy(configMem, tmpConfig, configSize);
 274               *readnewconfig = true;
 275               if (firstUsedConfig) {
 276                 firstUsedConfig = false;
 277                 if (openFileDesc.nequi < NEQUI)
 278                   nequileft_internal = NEQUI - openFileDesc.nequi - iDidVirtualSkips*openFileDesc.nskip;
 279                 else
 280                   nequileft_internal = NEQUI - openFileDesc.nequi;
 281               }
 282               nequileft_internal -= openFileDesc.nskip;
 283               *nequileft = nequileft_internal;
 284               return;
 285             }
 286           else {
 287             if(log) *log << "CCACHE: Could not read configuration. Closing dat-file: " << openFileDesc.filename << endl << flush;
 288             inFile.close();
 289           }
 290         }
 291     }
 292 }
 293
 294 void configcache::openOutFile(int actnequi)
 295 {
 296   time_t secstamp = time(NULL);
 297   int iseq=0;
 298
 299   while (true) {
 300     outFileName.str("");
 301     outFileName << DATADIR << "/" << secstamp << "." << iseq << "_" << getFileId(actnequi) << "_.edat.tmp";
 302
 303     int tmpfd = open(outFileName.str().c_str(), O_CREAT | O_EXCL, 0644);
 304
 305     if ( tmpfd != -1 ) {
 306       close(tmpfd);
 307       break;
 308     }
 309     else if ( errno != EEXIST ) {
 310       if(log) *log << "CCACHE: Could not create cachefile!" << endl << flush;
 311       exit(1);
 312     }
 313
 314     iseq++;
 315   }
 316
 317   outFile.open( outFileName.str().c_str(), std::ios::binary );
 318
 319   outFile << paraString() << endl;
 320
 321   ioBuffers->out = new boost::iostreams::filtering_ostreambuf;
 322   ioBuffers->out->push(boost::iostreams::bzip2_compressor());
 323   ioBuffers->out->push(outFile);
 324 }
 325
 326 void configcache::writeHeader(const string& headerid, const char *header, long unsigned int size, int actnequi) {
 327   unsigned long headeridhash;
 328
 329   if( DATADIR == "" || !(MODE==CACHE_MODE_WO||MODE==CACHE_MODE_RW) ) return;
 330
 331   if(!outFile.is_open())
 332     openOutFile(actnequi);
 333
 334   headeridhash = hash(headerid);
 335
 336   boost::iostreams::write(*ioBuffers->out, (char*)&size, sizeof(long unsigned int));
 337   boost::iostreams::write(*ioBuffers->out, (char*)&headeridhash, sizeof(unsigned long));
 338   boost::iostreams::write(*ioBuffers->out, header, size);
 339 }
 340
 341 void configcache::writeConfig(int actnequi)
 342 {
 343   long unsigned int zeroheader=0;
 344
 345   if ( DATADIR == "" || !(MODE==CACHE_MODE_WO||MODE==CACHE_MODE_RW) ) return;
 346
 347   if ( ! outFile.is_open() )
 348     openOutFile(actnequi);
 349
 350   boost::iostreams::write(*ioBuffers->out, (char*)&zeroheader, sizeof(long unsigned int));
 351
 352   boost::iostreams::write(*ioBuffers->out, configMem, configSize);
 353 }
 354
 355 void configcache::addPara(const string& parid, const double& val){
 356   parameter newPara;
 357   newPara.id = parid;
 358   newPara.val = val;
 359   Paras.push_back(newPara);
 360 }
 361
 362 int configcache::getParIndex(const string& parid){
 363   for(int ipara=0; ipara<Paras.size(); ipara++)
 364     if(Paras[ipara].id == parid) return ipara;
 365 }
 366
 367 void configcache::setPara(const string& parid, const double& value){
 368   Paras[getParIndex(parid)].val = value;
 369
 370   finishOutFile();
 371   if(ioBuffers->in != NULL) { delete ioBuffers->in; ioBuffers->in=NULL; }
 372   inFile.close();
 373   inFiles.clear();
 374
 375   refetchDataFiles = true;
 376   nequileft_internal = NEQUI;
 377 }
 378
 379 configcache::~configcache()
 380 {
 381   finishOutFile();
 382   delete ioBuffers->in;
 383   ioBuffers->in = NULL;
 384 }
 385
 386 void configcache::finishOutFile()
 387 {
 388   if( ioBuffers->out != NULL )
 389     {
 390       delete ioBuffers->out;
 391       ioBuffers->out = NULL;
 392     }
 393
 394   if( outFile.is_open() )
 395     {
 396       outFile.close();
 397       rename( outFileName.str().c_str(), outFileName.str().substr(0, outFileName.str().size()-4).c_str() );
 398     }
 399 }
 400
 401 int configcache::readFullBlock(char *tmpData, long unsigned int dataSize)
 402 {
 403   /* try to read header */
 404   if ( openFileDesc.extended )
 405     if ( ! readAllHeaders() )
 406       return -1;
 407
 408   /* read data */
 409   return readDataToMem(tmpData, dataSize);
 410 }
 411
 412 int configcache::readDataToMem(char *tmpData, long unsigned int dataSize)
 413 {
 414   int readturn = -1;
 415
 416   if ( dataSize == 0 ) return 0;
 417
 418   try { readturn = boost::iostreams::read(*ioBuffers->in, tmpData, dataSize); }
 419   catch(boost::iostreams::bzip2_error& error) {
 420     if(log) *log << "CCACHE: Caught bzip2 exception with error code: " << error.error() << endl << flush;
 421     inFile.close();
 422   }
 423   catch (std::exception const& ex) {
 424     if(log) *log << "CCACHE: Caught exception: " << ex.what() << endl << flush;
 425     inFile.close();
 426   }
 427   catch( ... ) {
 428     if(log) *log << "CCACHE: Caught unknown exception while reading." << endl << flush;
 429     inFile.close();
 430   }
 431
 432   return readturn;
 433 }
 434
 435 unsigned long configcache::hash(const string& str)
 436 {
 437   unsigned long hash = 5381;
 438
 439   for(string::const_iterator it=str.begin();it!=str.end();it++)
 440     hash = ((hash << 5) + hash) + *it; /* hash * 33 + character */
 441
 442   return hash;
 443 }
 444
 445 void configcache::deleteHeaderStore()
 446 {
 447   while ( headerStore.size() > 0 ) {
 448     free(headerStore.back().second);
 449     headerStore.pop_back();
 450   }
 451 }
 452
 453 vector<infiledesc>::iterator configcache::getNextInfile(vector<unsigned long> *excludeFileHashes) {
 454   for (vector<infiledesc>::iterator init = inFiles.begin(); init != inFiles.end(); ++init) {
 455     if (excludeFileHashes != NULL) {
 456       bool excludethisfile = false;
 457
 458       for (vector<unsigned long>::iterator exit = excludeFileHashes->begin(); exit != excludeFileHashes->end(); ++exit)
 459         if ( *exit == hash(init->filename) ) {
 460           excludethisfile = true;
 461           break;
 462         }
 463
 464       if (excludethisfile)
 465         continue;
 466     }
 467     return init;
 468   }
 469   return inFiles.end();
 470 }