ReadWriteModule.cxx

Go to the documentation of this file.

#include "IoModules/ReadWriteModule.h"
#include <unistd.h>
#include <cstdio>
#include <cstring>
#include <iostream>
#include <cassert>
#include "TFile.h"
#include "EventDataModel/EventHeader.h"
using namespace io;

//......................................................................

ReadWriteModule::ReadWriteModule() :
  fRunNumber   (  0     ),
  fEventNumber (  0     ),
  fReadOK      (  false ), 
  fFileIndex   ( -1     ), 
  fInFile      (  0     ), 
  fEvtHandle   (  0     ), 
  fOwnHandle   (  true  ),
  fOutFile     (  0     ),
  fNwrite      (  0     ),
  fFlushFreq   (  100   ),
  fOutSizeLimit(  0     )
{ 
  // fEvtHandle pointer is set to 0 so that classes which inherit can
  // change the pointer in their constructor before the handle is ever
  // accessed by IO module
}

//......................................................................
ReadWriteModule::~ReadWriteModule() 
{ 
  if (fInFile) {
    fInFile->Close();
    delete fInFile;
    fInFile = 0;
  }
  if (fOutFile) {
    fOutFile->Write();
    fOutFile->Close();
    delete fOutFile;
    fOutFile = 0;
  }
  if (fOwnHandle && fEvtHandle) {
    delete fEvtHandle;
  }
  std::vector<Filter*>::iterator itr;
  std::vector<Filter*>::iterator itrEnd (fInFilterList.end());
  itr    = fInFilterList.begin();
  itrEnd = fInFilterList.end();
  for (; itr!=itrEnd; ++itr) if (*itr) delete (*itr);
  itr    = fOutFilterList.begin();
  itrEnd = fOutFilterList.end();
  for (; itr!=itrEnd; ++itr) if (*itr) delete (*itr);
}

//......................................................................
int ReadWriteModule::AddFile(const char* file_regexp) 
{
  // If nobody created event handle, we should do it now
  if (!fEvtHandle) fEvtHandle = new EventHandle;

  // Cheap way to get list of files matching the expression (assumes
  // some flavor of unix
  char tmp_name[128] = {"/tmp/IOMOD.XXXXXX"};
  mkstemp(tmp_name);
  std::string cmd;
  cmd  = "(ls ";
  cmd += file_regexp;
  cmd += " > ";
  cmd += tmp_name;
  cmd += ") >& /dev/null";
  system(cmd.c_str());

  // Open the temp file and get the list of files which matched
  static const int s = 256;
  char             buff[s];
  FILE*            fp     = fopen(tmp_name,"r");
  int              nfiles = 0;
  std::string      file;
  while ( fgets(buff,s-1,fp) != NULL ) {
    int len = strlen(buff);
    while (buff[len]==' ' || buff[len]=='\n' || buff[len]=='\0') --len;
    buff[len+1]='\0';
    file = buff;
    fFileList.push_back(file);
    ++nfiles;
  }
  fclose(fp);

  // Remove the temporary file
  unlink(tmp_name);

  if(strstr(file_regexp,"dcache:dcap")!=0){
     std::string f=(std::string)file_regexp;
     fFileList.push_back(f);
  }
  else if  (strstr(file_regexp, "root://") != 0) {
     std::string f=(std::string)file_regexp;
     fFileList.push_back(f);
  }

  // Setup the first file in the list
  if (fFileIndex<0 && fFileList.size()>0) {
    fFileIndex = 0;
    fEvtHandle->Clear();
    if (this->SetupInputFile()) {
      fReadOK = true;
    }
    else {
      fReadOK = false;
    }
  }
  return nfiles;
}

//......................................................................
int ReadWriteModule::RemoveFile(const char* /*file_regexp*/) {
  return 0; 
}

//......................................................................
int ReadWriteModule::GoToFile(const char* file) 
{ 
  unsigned int i;
  for (i=0; i<fFileList.size(); ++i) {
    if (fFileList[i] == file) {
      fFileIndex = i;
      this->SetupInputFile();
      return 1;
    }
  }
  // File not in list
  return 0;
}

//......................................................................
int ReadWriteModule::AdvanceFile(int n) 
{ 
  if (n<=0) return 0;
  
  int indexMax  = fFileList.size()-1;
  int indexSave = fFileIndex;

  fFileIndex += n;
  if (fFileIndex > indexMax) {
    fFileIndex = indexMax+1;
    fReadOK    = false;
    return 0;
  }

  // Do something to open new file...
  this->SetupInputFile();

  return (fFileIndex-indexSave);
}

//......................................................................
int ReadWriteModule::RewindFile(int n) 
{
  if (n<=0) return 0;
  
  int indexMin  = 0;
  int indexSave = fFileIndex;

  fFileIndex -= n;
  if (fFileIndex < indexMin) {
    fFileIndex = 0;
    fReadOK    = false;
    return 0;
  }

  // Do something to open new file...
  this->SetupInputFile();

  return (indexSave-fFileIndex);
}

//......................................................................
int ReadWriteModule::GoTo(int run, int event) 
{ 
  // Get an initial guess at how many events to skip
  this->UpdateEventNumbers();
  int nSkipEvent = abs(event - fEventNumber); 
  if (run != fRunNumber) nSkipEvent += 1000;
  
  // This assumes that the files are sorted in run/event order
  for (;;) {
    // Check if we're done
    if (fRunNumber == run && fEventNumber == event) return 1;
    
    // Advance/Rewind in the data stream
    while (run<fRunNumber || event<fEventNumber) { 
      int nrew = this->Rewind(nSkipEvent);
      this->UpdateEventNumbers();
      if (nrew == 0) break;
    }
    while (run>fRunNumber || event>fEventNumber) {
      int nadv = this->Advance(nSkipEvent);
      this->UpdateEventNumbers();
      if (nadv == 0) break;
    }

    // Check if we've found the right event
    if (fRunNumber == run && fEventNumber == event) return 1;

    // If we were stepping one-by-one and couldn't find the event,
    // stop searching
    if (nSkipEvent <= 1) return 0;

    // Reduce step size for next search
    nSkipEvent = nSkipEvent/2;
  }
  return 0;
}

//......................................................................
int ReadWriteModule::Advance(int n) 
{ 
  assert(fEvtHandle);

  int ndone = 0;
  while (ndone < n) {
    // Try to advance far enough in the current file to match the request
    int ntry = n-ndone;
    int ndid;
    ndid = fEvtHandle->Advance(ntry);
    ndone += ndid;
    // If the current file doesn't have enough events to fill the
    // request, go to the next file
    if (ndid<ntry) {
      int advance = this->AdvanceFile();
      // If there is no next file, stop and return the number of records
      // actually advanced
      if (advance<1) {
        fReadOK = false;
        return ndone;
      }
      ++ndone; // count the file boundary as one advance
    }
  }
  // Check if the current event passed filters, if not, advance one
  if (this->CheckInFilters(*fEvtHandle) == false) {
    ndone += this->Advance(1);
  }
  this->UpdateEventNumbers();
  return ndone;
}

//......................................................................
int ReadWriteModule::Rewind(int n) 
{
  int ndone = 0;
  while (ndone < n) {
    // Try to rewind far enough in the current file to finish
    int ntry = n-ndone;
    int ndid = fEvtHandle->Rewind(ntry);
    ndone += ndid;
    // Not enough events in current file, go to previous file and 
    if (ndid<ntry) {
      int rewind = this->RewindFile();
      if (rewind<1) {
        std::cout << "Reached start of event stream.\n";
        return ndone;
      }
      ++ndone; // count the file boundary as one rewind
    }
  }
  // Check if the current event passed filters, if not, rewind one
  if (this->CheckInFilters(*fEvtHandle) == false) {
    ndone += this->Rewind(1);
  }
  return ndone;
}

//......................................................................
int ReadWriteModule::Reload() 
{ 
  fEvtHandle->ClearLoadFlags();
  return 1;
}

//......................................................................
void ReadWriteModule::SetOutputFileName(const char* n) 
{
  std::string fname(n);
  if (!fEvtHandle) fEvtHandle = new EventHandle;
  if (fOutFile && (fname==fOutFile->GetName())) return;

  if (fOutFile) { 
    fEvtHandle->Close();
    fOutFile->Flush(); 
    fOutFile->Write(); 
    delete fOutFile; 
    fOutFile = 0;
  }

  fOutFile = new TFile(fname.c_str(),"RECREATE","Event Data File",1);
  fEvtHandle->SetupOutputFile(fOutFile);
}

//......................................................................
void ReadWriteModule::Report() const 
{
  std::cout << "Files in list:" << std::endl;
  for (int i=0; i<(int)fFileList.size(); ++i) {
    if (i==fFileIndex) std::cout << ">";
    else               std::cout << " ";
    std::cout << "[" << i << "] " << fFileList[i];
    if (i==fFileIndex) std::cout << " (i="<<fEvtHandle->Index()<<")";
    std::cout << std::endl;
  }
}

//......................................................................
const char* ReadWriteModule::CurrentFile() const 
{
  if (fFileList.size()>0) {
    if (fFileIndex>=0 && fFileIndex<(int)fFileList.size()) {
      return fFileList[fFileIndex].c_str();
    }
  }
  return "";
}

//......................................................................
const char* ReadWriteModule::FileName(int i) const
{
  if (i>=0 && i<(int)fFileList.size()) return fFileList[i].c_str();
  return 0;
}

//......................................................................
edm::EventHandle& ReadWriteModule::GetEvent() 
{
  if (!fEvtHandle) fEvtHandle = new EventHandle;
  return *fEvtHandle; 
}

//......................................................................
void ReadWriteModule::UpdateEventNumbers() 
{
  edm::EventHandle& ev = this->GetEvent();
  edm::EventHeader& eh = ev.Header();
  fRunNumber           = eh.Run();
  fEventNumber         = eh.Event();
}

//......................................................................
int ReadWriteModule::WriteEvent() 
{
  if (fOutFile==0) {
    std::cout << "ReadWriteModule: No output file set." << std::endl;
    return 0;
  }
  int aok = fEvtHandle->Write();
  if (aok) {
    ++fNwrite;
    if (fNwrite%fFlushFreq == 0) { fOutFile->Flush(); }
  }
  if ((fOutSizeLimit > 0) &&
      (fOutSizeLimit < fOutFile->GetSize())) {
    // We should start writing to a new file
    std::string fname = fOutFile->GetName();

    std::string::size_type pos = fname.find("_Evt");
    if (pos == std::string::npos) {
      pos = fname.find(".root");
    }
    fname = fname.substr(0, pos);
    char evtStr[256];
    sprintf(evtStr, "_Evt%06d.root", fEventNumber);
    fname += evtStr;
    std::cout << "Starting new file: " << fname << std::endl;
    SetOutputFileName(fname.c_str());
  }
  return aok;
}

//......................................................................

bool ReadWriteModule::ReadOK() { return fReadOK; }

//......................................................................

int ReadWriteModule::SetupInputFile() 
{
//======================================================================
// Setup a new file
//======================================================================
  if (fFileIndex<0 || fFileIndex>=(int)fFileList.size()) {
    fReadOK = false;
    return 0;
  }
  
  // Check if the filename has changed
  if (fInFile && fInFile->GetName()!=fFileList[fFileIndex]) {
    fInFile->Flush();
    fInFile->Close();
    delete fInFile;
    fInFile = 0;
    fReadOK = false;
  }

  // Open the new file
  fInFile = TFile::Open(fFileList[fFileIndex].c_str());
  if (fInFile==0) {
    std::cout << "Failed to open file '" 
              << fFileList[fFileIndex] << "' for read.\n";
//    std::cout<<"Is it a dcache file? "<<endl;
//    if(strstr(fFileList[fFileIndex].c_str(),"dcache:dcap")!=0){
//       fInFile = TFile::Open(fFileList[fFileIndex].c_str())
    fReadOK = false;
    return 0;
  }

  // Sync. the event handle to the new file
  if (fEvtHandle->SetupInputFile(fInFile)) {
    this->UpdateEventNumbers();
    fReadOK = true;
    return 1;
  }
  
  // Error conditions fall through to here
  fReadOK = false;
  return 0;
}

//......................................................................

void ReadWriteModule::AdoptInputFilter(Filter* f) { 
  fInFilterList.push_back(f); 
}

//......................................................................
void ReadWriteModule::AdoptOutputFilter(Filter* f) {
  fOutFilterList.push_back(f);
}

//......................................................................
bool ReadWriteModule::CheckInFilters(edm::EventHandle& evd) 
{
  std::vector<Filter*>::iterator itr    (fInFilterList.begin());
  std::vector<Filter*>::iterator itrEnd (fInFilterList.end());
  for (; itr!=itrEnd; ++itr) {
    bool ret = (**itr)(evd);
    if (ret==false) return false; // Failed filter
  }
  return true; // Passed all filters
}

//......................................................................
bool ReadWriteModule::CheckOutFilters(edm::EventHandle& evd) 
{
  std::vector<Filter*>::iterator itr    (fOutFilterList.begin());
  std::vector<Filter*>::iterator itrEnd (fOutFilterList.end());
  for (; itr!=itrEnd; ++itr) {
    bool ret = (**itr)(evd);
    if (ret==false) return false; // Failed filter
  }
  return true; // Passed all filters
}

//......................................................................
void ReadWriteModule::SetEventHandle(EventHandle* handle)
{
  assert(!fEvtHandle);
  fEvtHandle = handle;
  fOwnHandle = false;
}

//......................................................................
void ReadWriteModule::Close()
{
  if (fOutFile) {
    std::cout << "Wrote " << fNwrite << " events. Closing file."
              << std::endl;
    fOutFile->Write();
    fOutFile->Close();
    delete fOutFile;
    fOutFile = 0;
  }
}

//......................................................................
void ReadWriteModule::SetOutSizeLimit(int mbLimit) 
{
  fOutSizeLimit = mbLimit; fOutSizeLimit *= 1000000;
}

---