edm::StreamerInputFile Class Reference

#include <StreamerInputFile.h>

Public Types
enum	Next { Next::kEvent, Next::kFile, Next::kStop }

Public Member Functions
void	closeStreamerFile ()
EventMsgView const *	currentRecord () const
bool	newHeader ()
Next	next ()
bool	openNextFile ()
InitMsgView const *	startMessage () const
	StreamerInputFile (std::string const &name, std::string const &LFN, std::shared_ptr< EventSkipperByID > eventSkipperByID=std::shared_ptr< EventSkipperByID >(), unsigned int prefetchMBytes=0)
	StreamerInputFile (std::string const &name, std::shared_ptr< EventSkipperByID > eventSkipperByID=std::shared_ptr< EventSkipperByID >(), unsigned int prefetchMBytes=0)
	StreamerInputFile (std::vector< FileCatalogItem > const &names, std::shared_ptr< EventSkipperByID > eventSkipperByID=std::shared_ptr< EventSkipperByID >(), unsigned int prefetchMBytes=0)
	~StreamerInputFile ()

Private Member Functions
bool	compareHeader ()
void	logFileAction (char const *msg)
void	openStreamerFile (std::string const &name, std::string const &LFN)
std::pair< storage::IOSize, char * >	readBytes (char *buf, storage::IOSize nBytes, bool zeroCopy, unsigned int skippedHdr=0)
int	readEventMessage ()
void	readStartMessage ()
storage::IOOffset	skipBytes (storage::IOSize nBytes)

Private Attributes
edm::propagate_const< std::shared_ptr< EventMsgView > >	currentEvMsg_
unsigned int	currentFile_
std::string	currentFileName_
bool	currentFileOpen_
uint32	currProto_
uint32	currRun_
bool	endOfFile_
std::vector< char >	eventBuf_
edm::propagate_const< std::shared_ptr< EventSkipperByID > >	eventSkipperByID_
std::vector< char >	headerBuf_
bool	multiStreams_
bool	newHeader_
edm::propagate_const< std::shared_ptr< InitMsgView > >	startMsg_
edm::propagate_const< std::unique_ptr< edm::storage::Storage > >	storage_
std::vector< FileCatalogItem >	streamerNames_
std::vector< char >	tempBuf_
unsigned int	tempLen_ = 0
unsigned int	tempPos_ = 0

Detailed Description

Definition at line 19 of file StreamerInputFile.h.

Member Enumeration Documentation

Next

enum edm::StreamerInputFile::Next

strong

Enumerator
kEvent
kFile
kStop

Definition at line 37 of file StreamerInputFile.h.

{ kEvent, kFile, kStop };

Constructor & Destructor Documentation

StreamerInputFile() [1/3]

edm::StreamerInputFile::StreamerInputFile ( std::string const &  name, std::string const &  LFN, std::shared_ptr< EventSkipperByID >  eventSkipperByID = std::shared_ptr<EventSkipperByID>(), unsigned int  prefetchMBytes = 0  )

explicit

Reads a Streamer file

Definition at line 21 of file StreamerInputFile.cc.

References Skims_PA_cff::name, openStreamerFile(), and readStartMessage().

      : startMsg_(),
        currentEvMsg_(),
        headerBuf_(1000 * 1000),
        eventBuf_(1000 * 1000 * 7),
        tempBuf_(1024 * 1024 * prefetchMBytes),
        currentFile_(0),
        streamerNames_(),
        multiStreams_(false),
        currentFileName_(),
        currentFileOpen_(false),
        eventSkipperByID_(eventSkipperByID),
        currRun_(0),
        currProto_(0),
        newHeader_(false),
        storage_(),
        endOfFile_(false) {
    openStreamerFile(name, LFN);
    readStartMessage();
  }

StreamerInputFile() [2/3]

edm::StreamerInputFile::StreamerInputFile ( std::string const &  name, std::shared_ptr< EventSkipperByID >  eventSkipperByID = std::shared_ptr<EventSkipperByID>(), unsigned int  prefetchMBytes = 0  )

explicit

Definition at line 45 of file StreamerInputFile.cc.

      : StreamerInputFile(name, name, eventSkipperByID, prefetchMBytes) {}

StreamerInputFile() [3/3]

edm::StreamerInputFile::StreamerInputFile ( std::vector< FileCatalogItem > const &  names, std::shared_ptr< EventSkipperByID >  eventSkipperByID = std::shared_ptr<EventSkipperByID>(), unsigned int  prefetchMBytes = 0  )

explicit

Multiple Streamer files

Definition at line 50 of file StreamerInputFile.cc.

References currentFile_, currProto_, currRun_, names, openStreamerFile(), readStartMessage(), and startMsg_.

      : startMsg_(),
        currentEvMsg_(),
        headerBuf_(1000 * 1000),
        eventBuf_(1000 * 1000 * 7),
        tempBuf_(1024 * 1024 * prefetchMBytes),
        currentFile_(0),
        streamerNames_(names),
        multiStreams_(true),
        currentFileName_(),
        currentFileOpen_(false),
        eventSkipperByID_(eventSkipperByID),
        currRun_(0),
        currProto_(0),
        newHeader_(false),
        endOfFile_(false) {
    openStreamerFile(names.at(0).fileName(0), names.at(0).logicalFileName());
    ++currentFile_;
    readStartMessage();
    currRun_ = startMsg_->run();
    currProto_ = startMsg_->protocolVersion();
  }

~StreamerInputFile()

edm::StreamerInputFile::~StreamerInputFile

(

)

Definition at line 19 of file StreamerInputFile.cc.

References closeStreamerFile().

{ closeStreamerFile(); }

Member Function Documentation

closeStreamerFile()

void edm::StreamerInputFile::closeStreamerFile

(

)

Test bit if a new header is encountered

Definition at line 112 of file StreamerInputFile.cc.

References currentFileOpen_, logFileAction(), and storage_.

Referenced by openStreamerFile(), and ~StreamerInputFile().

                                            {
    if (currentFileOpen_ && storage_) {
      storage_->close();
      logFileAction("  Closed file ");
    }
    currentFileOpen_ = false;
  }

compareHeader()

bool edm::StreamerInputFile::compareHeader ( )

private

Compares current File header with the newly opened file header Returns false in case of mismatch

Definition at line 253 of file StreamerInputFile.cc.

References currentFile_, currProto_, currRun_, Exception, edm::errors::MismatchedInputFiles, readStartMessage(), startMsg_, and streamerNames_.

Referenced by openNextFile().

                                        {
    //Get the new header
    readStartMessage();

    //Values from new Header should match up
    if (currRun_ != startMsg_->run() || currProto_ != startMsg_->protocolVersion()) {
      throw Exception(errors::MismatchedInputFiles, "StreamerInputFile::compareHeader")
          << "File " << streamerNames_.at(currentFile_).fileNames()[0]
          << "\nhas different run number or protocol version than previous\n";
      return false;
    }
    return true;
  }

currentRecord()

EventMsgView const* edm::StreamerInputFile::currentRecord ( ) const

inline

Points to File Start Header/Message

Definition at line 43 of file StreamerInputFile.h.

References currentEvMsg_, and edm::propagate_const< T >::get().

{ return currentEvMsg_.get(); }

logFileAction()

void edm::StreamerInputFile::logFileAction ( char const *  msg )

private

Definition at line 379 of file StreamerInputFile.cc.

References currentFileName_, edm::FlushMessageLog(), and mps_check::msg.

Referenced by closeStreamerFile(), and openStreamerFile().

                                                       {
    LogAbsolute("fileAction") << std::setprecision(0) << TimeOfDay() << msg << currentFileName_;
    FlushMessageLog();
  }

newHeader()

bool edm::StreamerInputFile::newHeader ( )

inline

Points to current Record

Definition at line 46 of file StreamerInputFile.h.

References newHeader_, and createJobs::tmp.

                     {
      bool tmp = newHeader_;
      newHeader_ = false;
      return tmp;
    } 

next()

StreamerInputFile::Next edm::StreamerInputFile::next

(

void

)

Definition at line 219 of file StreamerInputFile.cc.

References currentFile_, kEvent, kFile, kStop, multiStreams_, newHeader_, readEventMessage(), and streamerNames_.

                                                {
    if (this->readEventMessage()) {
      return Next::kEvent;
    }
    if (multiStreams_) {
      //Try opening next file
      if (currentFile_ <= streamerNames_.size() - 1) {
        newHeader_ = true;
        return Next::kFile;
      }
    }
    return Next::kStop;
  }

openNextFile()

bool edm::StreamerInputFile::openNextFile

(

)

Definition at line 233 of file StreamerInputFile.cc.

References compareHeader(), currentFile_, endOfFile_, FDEBUG, openStreamerFile(), startMsg_, and streamerNames_.

                                       {
    if (currentFile_ <= streamerNames_.size() - 1) {
      FDEBUG(10) << "Opening file " << streamerNames_.at(currentFile_).fileNames()[0].c_str() << std::endl;

      openStreamerFile(streamerNames_.at(currentFile_).fileNames()[0],
                       streamerNames_.at(currentFile_).logicalFileName());

      // If start message was already there, then compare the
      // previous and new headers
      if (startMsg_) {
        FDEBUG(10) << "Comparing Header" << std::endl;
        compareHeader();
      }
      ++currentFile_;
      endOfFile_ = false;
      return true;
    }
    return false;
  }

openStreamerFile()

void edm::StreamerInputFile::openStreamerFile ( std::string const &  name, std::string const &  LFN  )

private

Definition at line 75 of file StreamerInputFile.cc.

References cms::Exception::addContext(), edm::check(), cms::Exception::clearMessage(), closeStreamerFile(), currentFileName_, currentFileOpen_, MillePedeFileConverter_cfg::e, Exception, edm::errors::FileOpenError, get, logFileAction(), Skims_PA_cff::name, edm::storage::IOFlags::OpenRead, findQualityFiles::size, and storage_.

Referenced by openNextFile(), and StreamerInputFile().

                                                                                      {
    closeStreamerFile();

    currentFileName_ = name;

    // Check if the logical file name was found.
    if (currentFileName_.empty()) {
      // LFN not found in catalog.
      throw cms::Exception("LogicalFileNameNotFound", "StreamerInputFile::openStreamerFile()\n")
          << "Logical file name '" << LFN << "' was not found in the file catalog.\n"
          << "If you wanted a local file, you forgot the 'file:' prefix\n"
          << "before the file name in your configuration file.\n";
      return;
    }

    logFileAction("  Initiating request to open file ");

    using namespace edm::storage;
    IOOffset size = -1;
    if (StorageFactory::get()->check(name, &size)) {
      try {
        storage_ = StorageFactory::get()->open(name, IOFlags::OpenRead);
      } catch (cms::Exception& e) {
        Exception ex(errors::FileOpenError, "", e);
        ex.addContext("Calling StreamerInputFile::openStreamerFile()");
        ex.clearMessage();
        ex << "Error Opening Streamer Input File: " << name << "\n";
        throw ex;
      }
    } else {
      throw Exception(errors::FileOpenError, "StreamerInputFile::openStreamerFile")
          << "Error Opening Streamer Input File, file does not exist: " << name << "\n";
    }
    currentFileOpen_ = true;
    logFileAction("  Successfully opened file ");
  }

readBytes()

std::pair< storage::IOSize, char * > edm::StreamerInputFile::readBytes ( char *  buf, storage::IOSize  nBytes, bool  zeroCopy, unsigned int  skippedHdr = 0  )

private

Definition at line 120 of file StreamerInputFile.cc.

References cms::Exception::addContext(), visDQMUpload::buf, edm::errors::FileReadError, dqmiodumpmetadata::n, storage_, tempBuf_, tempLen_, and tempPos_.

Referenced by readEventMessage(), and readStartMessage().

                                                                                          {
    storage::IOSize n = 0;
    //returned pointer should point to the beginning of the header
    //even if we read event payload that comes afterwards
    char* ptr = buf - skippedHdr;
    try {
      if (!tempBuf_.empty()) {
        if (tempPos_ == tempLen_) {
          n = storage_->read(&tempBuf_[0], tempBuf_.size());
          tempPos_ = 0;
          tempLen_ = n;
          if (n == 0)
            return std::pair<storage::IOSize, char*>(0, ptr);
        }
        if (nBytes <= tempLen_ - tempPos_) {
          //zero-copy can't done when header start address is in the previous buffer
          if (!zeroCopy || skippedHdr > tempPos_) {
            memcpy(buf, &tempBuf_[0] + tempPos_, nBytes);
            tempPos_ += nBytes;
          } else {
            //pass pointer to the prebuffer address (zero copy) at the beginning of the header
            ptr = &tempBuf_[0] + tempPos_ - skippedHdr;
            tempPos_ += nBytes;
          }
          n = nBytes;
        } else {
          //crossing buffer boundary
          auto len = tempLen_ - tempPos_;
          memcpy(buf, &tempBuf_[0] + tempPos_, len);
          tempPos_ += len;
          char* tmpPtr = buf + len;
          n = len + readBytes(tmpPtr, nBytes - len, false).first;
        }
      } else
        n = storage_->read(buf, nBytes);
    } catch (cms::Exception& ce) {
      Exception ex(errors::FileReadError, "", ce);
      ex.addContext("Calling StreamerInputFile::readBytes()");
      throw ex;
    }
    return std::pair<storage::IOSize, char*>(n, ptr);
  }

readEventMessage()

int edm::StreamerInputFile::readEventMessage ( )

private

Definition at line 267 of file StreamerInputFile.cc.

References HeaderView::code(), convert32(), convert64(), currentEvMsg_, endOfFile_, Header::EVENT, EventHeader::event_, eventBuf_, eventSkipperByID_, Exception, edm::errors::FileReadError, headerBuf_, mps_fire::i, Header::INIT, EventHeader::lumi_, Header::PADDING, readBytes(), EventHeader::run_, HeaderView::size(), and skipBytes().

Referenced by next().

                                          {
    if (endOfFile_)
      return 0;

    using namespace edm::storage;
    bool eventRead = false;
    unsigned hdrSkipped = 0;
    while (!eventRead) {
      IOSize nWant = sizeof(EventHeader);
      IOSize nGot = readBytes(&eventBuf_[hdrSkipped], nWant - hdrSkipped, false).first + hdrSkipped;
      while (nGot == nWant) {
        //allow padding before next event or end of file.
        //event header starts with code 0 - 17, so 0xff (Header:PADDING) uniquely represents padding
        bool headerFetched = false;
        for (size_t i = 0; i < nGot; i++) {
          if ((unsigned char)eventBuf_[i] != Header::PADDING) {
            //no padding 0xff
            if (i != 0) {
              memmove(&eventBuf_[0], &eventBuf_[i], nGot - i);
              //read remainder of the header
              nGot = nGot - i + readBytes(&eventBuf_[nGot - i], i, false).first;
            }
            headerFetched = true;
            break;
          }
        }
        if (headerFetched)
          break;
        //read another block
        nGot = readBytes(&eventBuf_[0], nWant, false).first;
      }
      if (nGot == 0) {
        // no more data available
        endOfFile_ = true;
        return 0;
      }
      if (nGot != nWant) {
        for (size_t i = 0; i < nGot; i++) {
          if ((unsigned char)eventBuf_[i] != Header::PADDING)
            throw edm::Exception(errors::FileReadError, "StreamerInputFile::readEventMessage")
                << "Failed reading streamer file, first read in readEventMessage\n"
                << "Requested " << nWant << " bytes, read function returned " << nGot
                << " bytes, non-padding at offset " << i;
        }
        //padded 0xff only
        endOfFile_ = true;
        return 0;
      }
      uint32 eventSize;
      {
        HeaderView head(&eventBuf_[0]);
        uint32 code = head.code();

        // If it is not an event then something is wrong.
        eventSize = head.size();
        if (code != Header::EVENT) {
          if (code == Header::INIT) {
            edm::LogWarning("StreamerInputFile") << "Found another INIT header in the file. It will be skipped";
            if (eventSize < sizeof(EventHeader)) {
              //very unlikely case that EventHeader is larger than total INIT size inserted in the middle of the file
              hdrSkipped = nGot - eventSize;
              memmove(&eventBuf_[0], &eventBuf_[eventSize], hdrSkipped);
              continue;
            }
            if (headerBuf_.size() < eventSize)
              headerBuf_.resize(eventSize);
            memcpy(&headerBuf_[0], &eventBuf_[0], nGot);
            readBytes(&headerBuf_[nGot], eventSize, true, nGot);
            //do not parse this header and proceed to the next event
            continue;
          }
          throw Exception(errors::FileReadError, "StreamerInputFile::readEventMessage")
              << "Failed reading streamer file, unknown code in event header\n"
              << "code = " << code << "\n";
        }
      }
      if (eventSize <= sizeof(EventHeader)) {
        throw edm::Exception(errors::FileReadError, "StreamerInputFile::readEventMessage")
            << "Failed reading streamer file, event header size from data too small\n";
      }
      eventRead = true;
      if (eventSkipperByID_) {
        EventHeader* evh = (EventHeader*)(&eventBuf_[0]);
        if (eventSkipperByID_->skipIt(convert32(evh->run_), convert32(evh->lumi_), convert64(evh->event_))) {
          eventRead = false;
        }
      }
      nWant = eventSize - sizeof(EventHeader);
      if (eventRead) {
        if (eventBuf_.size() < eventSize)
          eventBuf_.resize(eventSize);

        auto res = readBytes(&eventBuf_[sizeof(EventHeader)], nWant, true, sizeof(EventHeader));
        if (res.first != nWant) {
          throw Exception(errors::FileReadError, "StreamerInputFile::readEventMessage")
              << "Failed reading streamer file, second read in readEventMessage\n"
              << "Requested " << nWant << " bytes, read function returned " << res.first << " bytes\n";
        }
        currentEvMsg_ =
            std::make_shared<EventMsgView>((void*)res.second);  // propagate_const<T> has no reset() function
      } else {
        nGot = skipBytes(nWant);
        if (nGot != nWant) {
          throw Exception(errors::FileReadError, "StreamerInputFile::readEventMessage")
              << "Failed reading streamer file, skip event in readEventMessage\n"
              << "Requested " << nWant << " bytes skipped, seek function returned " << nGot << " bytes\n";
        }
      }
    }
    return 1;
  }

readStartMessage()

void edm::StreamerInputFile::readStartMessage ( )

private

Not an init message should return

Definition at line 180 of file StreamerInputFile.cc.

References HeaderView::code(), Exception, edm::errors::FileReadError, headerBuf_, Header::INIT, readBytes(), HeaderView::size(), and startMsg_.

Referenced by compareHeader(), and StreamerInputFile().

                                           {
    using namespace edm::storage;
    IOSize nWant = sizeof(InitHeader);
    IOSize nGot = readBytes(&headerBuf_[0], nWant, false).first;
    if (nGot != nWant) {
      throw Exception(errors::FileReadError, "StreamerInputFile::readStartMessage")
          << "Failed reading streamer file, first read in readStartMessage\n";
    }

    uint32 headerSize;
    {
      HeaderView head(&headerBuf_[0]);
      uint32 code = head.code();
      if (code != Header::INIT) 
      {
        throw Exception(errors::FileReadError, "StreamerInputFile::readStartMessage")
            << "Expecting an init Message at start of file\n";
        return;
      }
      headerSize = head.size();
    }

    if (headerBuf_.size() < headerSize)
      headerBuf_.resize(headerSize);

    if (headerSize > sizeof(InitHeader)) {
      nWant = headerSize - sizeof(InitHeader);
      auto res = readBytes(&headerBuf_[sizeof(InitHeader)], nWant, true, sizeof(InitHeader));
      if (res.first != nWant) {
        throw Exception(errors::FileReadError, "StreamerInputFile::readStartMessage")
            << "Failed reading streamer file, second read in readStartMessage\n";
      }
      startMsg_ = std::make_shared<InitMsgView>(res.second);  // propagate_const<T> has no reset() function
    } else {
      throw Exception(errors::FileReadError, "StreamerInputFile::readStartMessage")
          << "Failed reading streamer file, init header size from data too small\n";
    }
  }

skipBytes()

storage::IOOffset edm::StreamerInputFile::skipBytes ( storage::IOSize  nBytes )

private

Definition at line 166 of file StreamerInputFile.cc.

References cms::Exception::addContext(), edm::storage::Storage::CURRENT, edm::errors::FileReadError, dqmiodumpmetadata::n, and storage_.

Referenced by readEventMessage().

                                                                   {
    storage::IOOffset n = 0;
    try {
      // We wish to return the number of bytes skipped, not the final offset.
      n = storage_->position(0, storage::Storage::CURRENT);
      n = storage_->position(nBytes, storage::Storage::CURRENT) - n;
    } catch (cms::Exception& ce) {
      Exception ex(errors::FileReadError, "", ce);
      ex.addContext("Calling StreamerInputFile::skipBytes()");
      throw ex;
    }
    return n;
  }

startMessage()

InitMsgView const* edm::StreamerInputFile::startMessage ( ) const

inline

Moves the handler to next Event Record

Definition at line 40 of file StreamerInputFile.h.

References edm::propagate_const< T >::get(), and startMsg_.

{ return startMsg_.get(); }

Member Data Documentation

currentEvMsg_

edm::propagate_const<std::shared_ptr<EventMsgView> > edm::StreamerInputFile::currentEvMsg_

private

Definition at line 73 of file StreamerInputFile.h.

Referenced by currentRecord(), and readEventMessage().

currentFile_

unsigned int edm::StreamerInputFile::currentFile_

private

Definition at line 82 of file StreamerInputFile.h.

Referenced by compareHeader(), next(), openNextFile(), and StreamerInputFile().

currentFileName_

std::string edm::StreamerInputFile::currentFileName_

private

True if Multiple Streams are Read

Definition at line 85 of file StreamerInputFile.h.

Referenced by logFileAction(), and openStreamerFile().

currentFileOpen_

bool edm::StreamerInputFile::currentFileOpen_

private

Definition at line 86 of file StreamerInputFile.h.

Referenced by closeStreamerFile(), and openStreamerFile().

currProto_

uint32 edm::StreamerInputFile::currProto_

private

Definition at line 91 of file StreamerInputFile.h.

Referenced by compareHeader(), and StreamerInputFile().

currRun_

uint32 edm::StreamerInputFile::currRun_

private

Definition at line 90 of file StreamerInputFile.h.

Referenced by compareHeader(), and StreamerInputFile().

endOfFile_

bool edm::StreamerInputFile::endOfFile_

private

Definition at line 97 of file StreamerInputFile.h.

Referenced by openNextFile(), and readEventMessage().

eventBuf_

std::vector<char> edm::StreamerInputFile::eventBuf_

private

Buffer to store file Header

Definition at line 76 of file StreamerInputFile.h.

Referenced by readEventMessage().

eventSkipperByID_

edm::propagate_const<std::shared_ptr<EventSkipperByID> > edm::StreamerInputFile::eventSkipperByID_

private

Definition at line 88 of file StreamerInputFile.h.

Referenced by readEventMessage().

headerBuf_

std::vector<char> edm::StreamerInputFile::headerBuf_

private

Definition at line 75 of file StreamerInputFile.h.

Referenced by readEventMessage(), and readStartMessage().

multiStreams_

bool edm::StreamerInputFile::multiStreams_

private

names of Streamer files

Definition at line 84 of file StreamerInputFile.h.

Referenced by next().

newHeader_

bool edm::StreamerInputFile::newHeader_

private

Definition at line 93 of file StreamerInputFile.h.

Referenced by newHeader(), and next().

startMsg_

edm::propagate_const<std::shared_ptr<InitMsgView> > edm::StreamerInputFile::startMsg_

private

Definition at line 72 of file StreamerInputFile.h.

Referenced by compareHeader(), openNextFile(), readStartMessage(), startMessage(), and StreamerInputFile().

storage_

edm::propagate_const<std::unique_ptr<edm::storage::Storage> > edm::StreamerInputFile::storage_

private

Definition at line 95 of file StreamerInputFile.h.

Referenced by closeStreamerFile(), openStreamerFile(), readBytes(), and skipBytes().

streamerNames_

std::vector<FileCatalogItem> edm::StreamerInputFile::streamerNames_

private

keeps track of which file is in use at the moment

Definition at line 83 of file StreamerInputFile.h.

Referenced by compareHeader(), next(), and openNextFile().

tempBuf_

std::vector<char> edm::StreamerInputFile::tempBuf_

private

Buffer to store Event Data

Definition at line 78 of file StreamerInputFile.h.

Referenced by readBytes().

tempLen_

unsigned int edm::StreamerInputFile::tempLen_ = 0

private

Buffer to store prefetched bytes

Definition at line 79 of file StreamerInputFile.h.

Referenced by readBytes().

tempPos_

unsigned int edm::StreamerInputFile::tempPos_ = 0

private

Definition at line 80 of file StreamerInputFile.h.

Referenced by readBytes().