#include <WatcherStreamFileReader.h>

Public Member Functions
void	closeFile ()

const InitMsgView *	getHeader ()

edm::StreamerInputFile *	getInputFile ()

const EventMsgView *	getNextEvent ()

const bool	newHeader ()

	WatcherStreamFileReader (edm::ParameterSet const &pset)

	~WatcherStreamFileReader ()

Private Attributes
std::string	corruptedDir_

std::string	curDir_

bool	end_

std::string	fileListCmd_

std::string	fileName_

std::vector< std::string >	filePatterns_

std::deque< std::string >	filesInQueue_

std::string	inprocessDir_

std::string	inputDir_

std::string	processedDir_

std::unique_ptr< edm::StreamerInputFile >	streamerInputFile_

int	timeOut_

std::string	tokenFile_

int	verbosity_

Detailed Description

Definition at line 26 of file WatcherStreamFileReader.h.

Constructor & Destructor Documentation

WatcherStreamFileReader::WatcherStreamFileReader ( edm::ParameterSet const & pset )

Definition at line 113 of file WatcherStreamFileReader.cc.

References corruptedDir_, gather_cfg::cout, curDir_, DeadROC_duringRun::dir, heppy_check::dirs, Exception, f, fileListCmd_, filePatterns_, mps_fire::i, inprocessDir_, inputDir_, eostools::mkdir(), now(), processedDir_, hgcalPlots::stat, and tokenFile_.

     : inputDir_(pset.getParameter<std::string>("inputDir")),
       filePatterns_(pset.getParameter<std::vector<std::string> >("filePatterns")),
       inprocessDir_(pset.getParameter<std::string>("inprocessDir")),
       processedDir_(pset.getParameter<std::string>("processedDir")),
       corruptedDir_(pset.getParameter<std::string>("corruptedDir")),
       tokenFile_(pset.getUntrackedParameter<std::string>("tokenFile", "watcherSourceToken")),
       timeOut_(pset.getParameter<int>("timeOutInSec")),
       end_(false),
       verbosity_(pset.getUntrackedParameter<int>("verbosity", 0)) {
   struct stat buf;
   if (stat(tokenFile_.c_str(), &buf)) {
     FILE* f = fopen(tokenFile_.c_str(), "w");
     if (f) {
       fclose(f);
     } else {
       throw cms::Exception("WatcherSource") << "Failed to create token file.";
     }
   }
   vector<string> dirs;
   dirs.push_back(inprocessDir_);
   dirs.push_back(processedDir_);
   dirs.push_back(corruptedDir_);
 
   for (unsigned i = 0; i < dirs.size(); ++i) {
     const string& dir = dirs[i];
     struct stat fileStat;
     if (0 == stat(dir.c_str(), &fileStat)) {
       if (!S_ISDIR(fileStat.st_mode)) {
         throw cms::Exception("[WatcherSource]") << "File " << dir << " exists but is not a directory "
                                                 << " as expected.";
       }
     } else {  //directory does not exists, let's try to create it
       if (0 != mkdir(dir.c_str(), 0755)) {
         throw cms::Exception("[WatcherSource]") << "Failed to create directory " << dir << " for writing data.";
       }
     }
   }
 
   std::stringstream fileListCmdBuf;
   fileListCmdBuf.str("");
   //    fileListCmdBuf << "/bin/ls -rt " << inputDir_ << " | egrep '(";
   //by default ls will sort the file alphabetically which will results
   //in ordering the files in increasing LB number, which is the desired
   //order.
   //    fileListCmdBuf << "/bin/ls " << inputDir_ << " | egrep '(";
   fileListCmdBuf << "/bin/find " << inputDir_ << " -maxdepth 2 -print | egrep '(";
   //TODO: validate patternDir (see ;, &&, ||) and escape special character
   if (filePatterns_.empty())
     throw cms::Exception("WacherSource", "filePatterns parameter is empty");
   char curDir[PATH_MAX > 0 ? PATH_MAX : 4096];
   if (getcwd(curDir, sizeof(curDir)) == nullptr) {
     throw cms::Exception("WatcherSource") << "Failed to retreived working directory path: " << strerror(errno);
   }
   curDir_ = curDir;
 
   for (unsigned i = 0; i < filePatterns_.size(); ++i) {
     if (i > 0)
       fileListCmdBuf << "|";
     //     if(filePatterns_[i].size()>0 && filePatterns_[0] != "/"){//relative path
     //       fileListCmdBuf << curDir << "/";
     //     }
     fileListCmdBuf << filePatterns_[i];
   }
   fileListCmdBuf << ")' | sort";
 
   fileListCmd_ = fileListCmdBuf.str();
 
   cout << "[WatcherSource " << now() << "]"
        << " Command to retrieve input files: " << fileListCmd_ << "\n";
 }

WatcherStreamFileReader::~WatcherStreamFileReader ( )

Definition at line 185 of file WatcherStreamFileReader.cc.

185 {}

Member Function Documentation

void WatcherStreamFileReader::closeFile ( )

Definition at line 392 of file WatcherStreamFileReader.cc.

References mps_setup::cmd, gather_cfg::cout, end_, Exception, fileName_, mps_fire::i, now(), processedDir_, streamerInputFile_, and verbosity_.

Referenced by getNextEvent(), and Vispa.Main.Application.Application::tabCloseRequest().

                                         {
   if (streamerInputFile_.get() == nullptr)
     return;
   //delete the streamer input file:
   streamerInputFile_.reset();
   stringstream cmd;
   //TODO: validation of processDir
   cmd << "/bin/mv -f \"" << fileName_ << "\" \"" << processedDir_ << "/.\"";
   if (verbosity_)
     cout << "[WatcherSource " << now() << "]"
          << " Excuting " << cmd.str() << "\n";
   int i = system(cmd.str().c_str());
   if (i != 0) {
     throw cms::Exception("WatcherSource") << "Failed to move processed file '" << fileName_ << "'"
                                           << " to processed directory '" << processedDir_ << "'\n";
     //Stop further processing to prevent endless loop:
     end_ = true;
   }
   cout << flush;
 }

const InitMsgView * WatcherStreamFileReader::getHeader ( )

Definition at line 192 of file WatcherStreamFileReader.cc.

References InitMsgView::code(), Exception, getInputFile(), RecoTauValidation_cfi::header, Header::INIT, dtResolutionTest_cfi::inputFile, and edm::StreamerInputFile::startMessage().

                                                       {
   edm::StreamerInputFile* inputFile = getInputFile();
 
   //TODO: shall better send an exception...
   if (inputFile == nullptr) {
     throw cms::Exception("WatcherSource") << "No input file found.";
   }
 
   const InitMsgView* header = inputFile->startMessage();
 
   if (header->code() != Header::INIT)  //INIT Msg
     throw cms::Exception("readHeader", "WatcherStreamFileReader")
         << "received wrong message type: expected INIT, got " << header->code() << "\n";
 
   return header;
 }

edm::StreamerInputFile * WatcherStreamFileReader::getInputFile ( )

Definition at line 225 of file WatcherStreamFileReader.cc.

References HltBtagPostValidation_cff::c, filterCSVwithJSON::copy, corruptedDir_, gather_cfg::cout, curDir_, mps_fire::dest, BTVHLTOfflineSource_cfi::dirname, dt, end_, Exception, f, ztee::fd, fileListCmd_, MillePedeFileConverter_cfg::fileName, fileName_, filesInQueue_, mps_fire::i, inprocessDir_, inputDir_, createfilelist::int, dqmiodumpmetadata::n, now(), alignCSCRings::s, findQualityFiles::size, hgcalPlots::stat, streamerInputFile_, OrderedSet::t, ntuplemaker::time, timeOut_, tokenFile_, and verbosity_.

Referenced by getHeader(), getNextEvent(), and newHeader().

                                                             {
   char* lineptr = nullptr;
   size_t n = 0;
 
   struct stat buf;
 
   if (stat(tokenFile_.c_str(), &buf) != 0) {
     end_ = true;
   }
 
   bool waiting = false;
   static bool firstWait = true;
   timeval waitStart;
   //if no cached input file, look for new files until one is found:
   if (!end_ && streamerInputFile_.get() == nullptr) {
     fileName_.assign("");
 
     //check if we have file in the queue, if not look for new files:
     while (filesInQueue_.empty()) {
       if (stat(tokenFile_.c_str(), &buf) != 0) {
         end_ = true;
         break;
       }
       FILE* s = popen(fileListCmd_.c_str(), "r");
       if (s == nullptr) {
         throw cms::Exception("WatcherSource") << "Failed to retrieve list of input file: " << strerror(errno);
       }
 
       ssize_t len;
       while (!feof(s)) {
         if ((len = getline(&lineptr, &n, s)) > 0) {
           //remove end-of-line character:
           lineptr[len - 1] = 0;
           string fileName;
           if (lineptr[0] != '/') {
             if (!inputDir_.empty() && inputDir_[0] != '/') {  //relative path
               fileName.assign(curDir_);
               fileName.append("/");
               fileName.append(inputDir_);
             } else {
               fileName.assign(inputDir_);
             }
             fileName.append("/");
           }
           fileName.append(lineptr);
           filesInQueue_.push_back(fileName);
           if (verbosity_)
             cout << "[WatcherSource " << now() << "]"
                  << " File to process: '" << fileName << "'\n";
         }
       }
       while (!feof(s))
         fgetc(s);
       pclose(s);
       if (filesInQueue_.empty()) {
         if (!waiting) {
           cout << "[WatcherSource " << now() << "]"
                << " No file found. Waiting for new file...\n";
           cout << flush;
           waiting = true;
           gettimeofday(&waitStart, nullptr);
         } else if (!firstWait) {
           timeval t;
           gettimeofday(&t, nullptr);
           float dt = (t.tv_sec - waitStart.tv_sec) * 1. + (t.tv_usec - waitStart.tv_usec) * 1.e-6;
           if ((timeOut_ >= 0) && (dt > timeOut_)) {
             cout << "[WatcherSource " << now() << "]"
                  << " Having waited for new file for " << (int)dt << " sec. "
                  << "Timeout exceeded. Exits.\n";
             //remove(tokenFile_.c_str()); //we do not delete the token, otherwise sorting process on the monitoring farm will not be restarted by the runloop.sh script.
             end_ = true;
             break;
           }
         }
       }
       sleep(1);
     }  //end of file queue update
     firstWait = false;
     free(lineptr);
     lineptr = nullptr;
 
     while (streamerInputFile_.get() == nullptr && !filesInQueue_.empty()) {
       fileName_ = filesInQueue_.front();
       filesInQueue_.pop_front();
       int fd = open(fileName_.c_str(), 0);
       if (fd != 0) {
         struct stat buf;
         off_t size = -1;
         //check that file transfer is finished, by monitoring its size:
         time_t t = time(nullptr);
         for (;;) {
           fstat(fd, &buf);
           if (verbosity_)
             cout << "file size: " << buf.st_size << ", prev size: " << size << "\n";
           if (buf.st_size == size)
             break;
           else
             size = buf.st_size;
           if (difftime(t, buf.st_mtime) > 60)
             break;  //file older then 1 min=> tansfer must be finished
           sleep(1);
         }
 
         if (fd != 0 && buf.st_size == 0) {  //file is empty. streamer reader
           //                   does not like empty file=> skip it
           stringstream c;
           c << "/bin/mv -f \"" << fileName_ << "\" \"" << corruptedDir_ << "/.\"";
           if (verbosity_)
             cout << "[WatcherSource " << now() << "]"
                  << " Excuting " << c.str() << "\n";
           int i = system(c.str().c_str());
           if (i != 0) {
             //throw cms::Exception("WatcherSource")
             cout << "[WatcherSource " << now() << "] "
                  << "Failed to move empty file '" << fileName_ << "'"
                  << " to corrupted directory '" << corruptedDir_ << "'\n";
           }
           continue;
         }
 
         close(fd);
 
         vector<char> buf1(fileName_.size() + 1);
         copy(fileName_.begin(), fileName_.end(), buf1.begin());
         buf1[buf1.size() - 1] = 0;
 
         vector<char> buf2(fileName_.size() + 1);
         copy(fileName_.begin(), fileName_.end(), buf2.begin());
         buf2[buf1.size() - 1] = 0;
 
         string dirnam(dirname(&buf1[0]));
         string filenam(basename(&buf2[0]));
 
         string dest = inprocessDir_ + "/" + filenam;
 
         if (verbosity_)
           cout << "[WatcherSource " << now() << "]"
                << " Moving file " << fileName_ << " to " << dest << "\n";
 
         stringstream c;
         c << "/bin/mv -f \"" << fileName_ << "\" \"" << dest << "/.\"";
 
         if (0 != rename(fileName_.c_str(), dest.c_str())) {
           //if(0!=system(c.str().c_str())){
           throw cms::Exception("WatcherSource")
               << "Failed to move file '" << fileName_ << "' "
               << "to processing directory " << inprocessDir_ << ": " << strerror(errno);
         }
 
         fileName_ = dest;
 
         cout << "[WatcherSource " << now() << "]"
              << " Opening file " << fileName_ << "\n"
              << flush;
         streamerInputFile_ = unique_ptr<edm::StreamerInputFile>(new edm::StreamerInputFile(fileName_));
 
         ofstream f(".watcherfile");
         f << fileName_;
       } else {
         cout << "[WatcherSource " << now() << "]"
              << " Failed to open file " << fileName_ << endl;
       }
     }  //loop on file queue to find one file which opening succeeded
   }
   return streamerInputFile_.get();
 }

const EventMsgView * WatcherStreamFileReader::getNextEvent ( )

Definition at line 209 of file WatcherStreamFileReader.cc.

References closeFile(), edm::StreamerInputFile::currentRecord(), end_, getInputFile(), dtResolutionTest_cfi::inputFile, and edm::StreamerInputFile::next().

                                                           {
   if (end_) {
     closeFile();
     return nullptr;
   }
 
   edm::StreamerInputFile* inputFile;
 
   //go to next input file, till no new event is found
   while ((inputFile = getInputFile()) != nullptr && inputFile->next() == 0) {
     closeFile();
   }
 
   return inputFile == nullptr ? nullptr : inputFile->currentRecord();
 }

const bool WatcherStreamFileReader::newHeader ( )

Definition at line 187 of file WatcherStreamFileReader.cc.

References getInputFile(), dtResolutionTest_cfi::inputFile, and edm::StreamerInputFile::newHeader().

                                               {
   edm::StreamerInputFile* inputFile = getInputFile();
   return inputFile ? inputFile->newHeader() : false;
 }