WatcherStreamFileReader Class Reference

#include <WatcherStreamFileReader.h>

Public Member Functions
void	closeFile ()
const edm::streamer::InitMsgView *	getHeader ()
edm::streamer::StreamerInputFile *	getInputFile ()
const edm::streamer::EventMsgView *	getNextEvent ()
const bool	newHeader ()
	WatcherStreamFileReader (edm::ParameterSet const &pset)
	~WatcherStreamFileReader ()

Private Member Functions
void	moveJustReadFile ()

Private Attributes
std::string	corruptedDir_
bool	end_
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::streamer::StreamerInputFile >	streamerInputFile_
int	timeOut_
std::string	tokenFile_
int	verbosity_

Detailed Description

Definition at line 26 of file WatcherStreamFileReader.h.

Constructor & Destructor Documentation

WatcherStreamFileReader()

WatcherStreamFileReader::WatcherStreamFileReader

(

edm::ParameterSet const &

pset

)

Definition at line 115 of file WatcherStreamFileReader.cc.

References visDQMUpload::buf, corruptedDir_, DeadROC_duringRun::dir, heppy_check::dirs, Exception, f, mps_fire::i, inprocessDir_, eostools::mkdir(), processedDir_, edm_modernize_messagelogger::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.";
      }
    }
  }
}

~WatcherStreamFileReader()

WatcherStreamFileReader::~WatcherStreamFileReader

(

)

Definition at line 155 of file WatcherStreamFileReader.cc.

{}

Member Function Documentation

closeFile()

void WatcherStreamFileReader::closeFile

(

)

Definition at line 392 of file WatcherStreamFileReader.cc.

{}

getHeader()

const InitMsgView * WatcherStreamFileReader::getHeader

(

)

Definition at line 159 of file WatcherStreamFileReader.cc.

References Exception, getInputFile(), RecoTauValidation_cfi::header, and makeListRunsInFiles::inputFile.

                                                      {
  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;
}

getInputFile()

StreamerInputFile * WatcherStreamFileReader::getInputFile

(

)

Definition at line 191 of file WatcherStreamFileReader.cc.

References visDQMUpload::buf, HltBtagPostValidation_cff::c, mps_setup::cmd, filterCSVwithJSON::copy, corruptedDir_, gather_cfg::cout, mps_fire::dest, BTVHLTOfflineSource_cfi::dirname, dt, end_, Exception, f, ztee::fd, MillePedeFileConverter_cfg::fileName, fileName_, filePatterns_, filesInQueue_, free(), mps_fire::i, inprocessDir_, inputDir_, createfilelist::int, dqmiodumpmetadata::n, now(), alignCSCRings::s, findQualityFiles::size, edm_modernize_messagelogger::stat, streamerInputFile_, submitPVValidationJobs::t, hcalRecHitTable_cff::time, timeOut_, tokenFile_, and verbosity_.

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

                                                         {
  char* lineptr = nullptr;
  size_t n = 0;
  static stringstream cmd;
  static bool cmdSet = false;
  static char curDir[PATH_MAX > 0 ? PATH_MAX : 4096];

  if (!cmdSet) {
    cmd.str("");
    //    cmd << "/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.
    //    cmd << "/bin/ls " << inputDir_ << " | egrep '(";
    cmd << "/bin/find " << inputDir_ << " -maxdepth 2 -print | egrep '(";
    //TODO: validate patternDir (see ;, &&, ||) and escape special character
    if (filePatterns_.empty())
      return nullptr;
    if (getcwd(curDir, sizeof(curDir)) == nullptr) {
      throw cms::Exception("WatcherSource") << "Failed to retreived working directory path: " << strerror(errno);
    }

    for (unsigned i = 0; i < filePatterns_.size(); ++i) {
      if (i > 0)
        cmd << "|";
      //     if(filePatterns_[i].size()>0 && filePatterns_[0] != "/"){//relative path
      //       cmd << curDir << "/";
      //     }
      cmd << filePatterns_[i];
    }
    cmd << ")' | sort";

    cout << "[WatcherSource " << now() << "]"
         << " Command to retrieve input files: " << cmd.str() << "\n";
    cmdSet = true;
  }

  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(cmd.str().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() << "]"
                 << " Executing " << 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[buf2.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) << " (Error no " << errno << ")";
        }

        fileName_ = dest;

        cout << "[WatcherSource " << now() << "]"
             << " Opening file " << fileName_ << "\n"
             << flush;
        streamerInputFile_ = std::make_unique<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();
}

getNextEvent()

const EventMsgView * WatcherStreamFileReader::getNextEvent

(

)

Definition at line 176 of file WatcherStreamFileReader.cc.

References end_, getInputFile(), makeListRunsInFiles::inputFile, edm::shared_memory::channel_names::kStop, and moveJustReadFile().

                                                          {
  if (end_) {
    moveJustReadFile();
    return nullptr;
  }

  StreamerInputFile* inputFile;
  if ((inputFile = getInputFile()) != nullptr and inputFile->next() == StreamerInputFile::Next::kStop) {
    moveJustReadFile();
    return nullptr;
  }

  return inputFile == nullptr ? nullptr : inputFile->currentRecord();
}

moveJustReadFile()

void WatcherStreamFileReader::moveJustReadFile ( )

private

Definition at line 394 of file WatcherStreamFileReader.cc.

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

Referenced by getNextEvent().

                                               {
  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() << "]"
  //<< " Executing (in closeFile())" << cmd.str() << "\n";
  //int i = system(cmd.str().c_str());
  //cout << "move command done at " << now() << "\n";
  vector<char> buf(fileName_.size() + 1);
  copy(fileName_.begin(), fileName_.end(), buf.begin());
  buf[buf.size() - 1] = 0;
  string dest = processedDir_ + "/" + basename(&buf[0]);
  if (verbosity_)
    cout << "[WatcherSource " << now() << "]"
         << " Moving " << fileName_ << " to " << dest << "... ";
  int i = rename(fileName_.c_str(), dest.c_str());
  if (i != 0) {
    throw cms::Exception("WatcherSource")
        << "Failed to move processed file '" << fileName_ << "'"
        << " to processed directory '" << processedDir_ << ": " << strerror(errno) << " (Error no " << errno << ")";

    //Stop further processing to prevent endless loop:
    end_ = true;
  }
  if (verbosity_)
    cout << "Done at " << now() << "\n";

  //  cout << flush;
}

newHeader()

const bool WatcherStreamFileReader::newHeader

(

)

Definition at line 157 of file WatcherStreamFileReader.cc.

References getInputFile().

{ return getInputFile() != nullptr; }

Member Data Documentation

corruptedDir_

std::string WatcherStreamFileReader::corruptedDir_

private

Directory where file must be moved if file is unreadble (e.g empty size)

Definition at line 59 of file WatcherStreamFileReader.h.

Referenced by getInputFile(), and WatcherStreamFileReader().

end_

bool WatcherStreamFileReader::end_

private

Definition at line 73 of file WatcherStreamFileReader.h.

Referenced by getInputFile(), getNextEvent(), and moveJustReadFile().

fileName_

std::string WatcherStreamFileReader::fileName_

private

Definition at line 65 of file WatcherStreamFileReader.h.

Referenced by getInputFile(), and moveJustReadFile().

filePatterns_

std::vector<std::string> WatcherStreamFileReader::filePatterns_

private

Streamer file name pattern list

Definition at line 47 of file WatcherStreamFileReader.h.

Referenced by getInputFile().

filesInQueue_

std::deque<std::string> WatcherStreamFileReader::filesInQueue_

private

Definition at line 71 of file WatcherStreamFileReader.h.

Referenced by getInputFile().

inprocessDir_

std::string WatcherStreamFileReader::inprocessDir_

private

Directory where file are moved during processing

Definition at line 51 of file WatcherStreamFileReader.h.

Referenced by getInputFile(), and WatcherStreamFileReader().

inputDir_

std::string WatcherStreamFileReader::inputDir_

private

Directory to look for streamer files

Definition at line 43 of file WatcherStreamFileReader.h.

Referenced by getInputFile().

processedDir_

std::string WatcherStreamFileReader::processedDir_

private

Directory where file must be moved once processed

Definition at line 55 of file WatcherStreamFileReader.h.

Referenced by moveJustReadFile(), and WatcherStreamFileReader().

streamerInputFile_

std::unique_ptr<edm::streamer::StreamerInputFile> WatcherStreamFileReader::streamerInputFile_

private

Cached input file stream

Definition at line 63 of file WatcherStreamFileReader.h.

Referenced by getInputFile(), and moveJustReadFile().

timeOut_

int WatcherStreamFileReader::timeOut_

private

Definition at line 69 of file WatcherStreamFileReader.h.

Referenced by getInputFile().

tokenFile_

std::string WatcherStreamFileReader::tokenFile_

private

Definition at line 67 of file WatcherStreamFileReader.h.

Referenced by getInputFile(), and WatcherStreamFileReader().

verbosity_

int WatcherStreamFileReader::verbosity_

private

Definition at line 75 of file WatcherStreamFileReader.h.

Referenced by getInputFile(), and moveJustReadFile().