#include <StreamerInputFile.h>

Public Member Functions
void	closeStreamerFile ()
	Needs to be public because of forking. More...

EventMsgView const *	currentRecord () const

bool	newHeader ()

bool	next ()

InitMsgView const *	startMessage () const

	StreamerInputFile (std::string const &name, std::shared_ptr< EventSkipperByID > eventSkipperByID=std::shared_ptr< EventSkipperByID >())

	StreamerInputFile (std::vector< std::string > const &names, std::shared_ptr< EventSkipperByID > eventSkipperByID=std::shared_ptr< EventSkipperByID >())

	~StreamerInputFile ()

Private Member Functions
bool	compareHeader ()

void	logFileAction (char const *msg)

bool	openNextFile ()

void	openStreamerFile (std::string const &name)

IOSize	readBytes (char *buf, IOSize nBytes)

int	readEventMessage ()

void	readStartMessage ()

IOOffset	skipBytes (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< Storage > >	storage_

std::vector< std::string >	streamerNames_

Detailed Description

Definition at line 18 of file StreamerInputFile.h.

Constructor & Destructor Documentation

edm::StreamerInputFile::StreamerInputFile	(	std::string const &	name,
		std::shared_ptr< EventSkipperByID >	eventSkipperByID = `std::shared_ptr<EventSkipperByID>()`
	)

explicit

Reads a Streamer file

Definition at line 22 of file StreamerInputFile.cc.

References openStreamerFile(), and readStartMessage().

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

edm::StreamerInputFile::StreamerInputFile	(	std::vector< std::string > const &	names,
		std::shared_ptr< EventSkipperByID >	eventSkipperByID = `std::shared_ptr<EventSkipperByID>()`
	)

explicit

Multiple Streamer files

Definition at line 43 of file StreamerInputFile.cc.

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

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

edm::StreamerInputFile::~StreamerInputFile ( )

Definition at line 18 of file StreamerInputFile.cc.

References closeStreamerFile().

                                         {
     closeStreamerFile();
   }

Member Function Documentation

void edm::StreamerInputFile::closeStreamerFile ( )

Needs to be public because of forking.

Test bit if a new header is encountered

Definition at line 97 of file StreamerInputFile.cc.

References currentFileOpen_, logFileAction(), and storage_.

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

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

bool edm::StreamerInputFile::compareHeader ( )

private

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

Definition at line 206 of file StreamerInputFile.cc.

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

Referenced by newHeader(), and 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_)
         << "\nhas different run number or protocol version than previous\n";
       return false;
     }
     newHeader_ = true;
     return true;
   }

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

inline

Points to File Start Header/Message

Definition at line 36 of file StreamerInputFile.h.

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

Referenced by WatcherStreamFileReader::getNextEvent().

36 { return currentEvMsg_.get(); }

edm::propagate_const::get

element_type const * get() const

Definition: propagate_const.h:68

edm::StreamerInputFile::currentEvMsg_

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

Definition: StreamerInputFile.h:61

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

private

Definition at line 285 of file StreamerInputFile.cc.

References currentFileName_, and edm::FlushMessageLog().

Referenced by closeStreamerFile(), newHeader(), and openStreamerFile().

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

bool edm::StreamerInputFile::newHeader ( )

inline

Points to current Record

Definition at line 39 of file StreamerInputFile.h.

References closeStreamerFile(), compareHeader(), logFileAction(), mps_check::msg, newHeader_, openNextFile(), openStreamerFile(), readBytes(), readEventMessage(), readStartMessage(), skipBytes(), AlCaHLTBitMon_QueryRunRegistry::string, and tmp.

Referenced by WatcherStreamFileReader::newHeader().

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

edm::StreamerInputFile::newHeader_

bool newHeader_

Definition: StreamerInputFile.h:77

tmp

std::vector< std::vector< double > > tmp

Definition: MVATrainer.cc:100

bool edm::StreamerInputFile::next ( void )

Definition at line 168 of file StreamerInputFile.cc.

References endOfFile_, multiStreams_, openNextFile(), and readEventMessage().

Referenced by WatcherStreamFileReader::getNextEvent().

                                {
     if(this->readEventMessage()) {
          return true;
     }
     if(multiStreams_) {
        //Try opening next file
        if(openNextFile()) {
           endOfFile_ = false;
           if(this->readEventMessage()) {
              return true;
           }
        }
     }
     return false;
   }

bool edm::StreamerInputFile::openNextFile ( )

private

Definition at line 184 of file StreamerInputFile.cc.

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

Referenced by newHeader(), and next().

                                        {
 
      if(currentFile_ <= streamerNames_.size() - 1) {
        FDEBUG(10) << "Opening file "
                   << streamerNames_.at(currentFile_).c_str() << std::endl;
 
        openStreamerFile(streamerNames_.at(currentFile_));
 
        // If start message was already there, then compare the
        // previous and new headers
        if(startMsg_) {
           FDEBUG(10) << "Comparing Header" << std::endl;
           if(!compareHeader()) {
               return false;
           }
        }
        ++currentFile_;
        return true;
      }
      return false;
   }

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

private

Definition at line 67 of file StreamerInputFile.cc.

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

Referenced by newHeader(), openNextFile(), and StreamerInputFile().

                                                            {
 
     closeStreamerFile();
 
     currentFileName_ = name;
     logFileAction("  Initiating request to open file ");
 
     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 ");
   }

IOSize edm::StreamerInputFile::readBytes	(	char *	buf,
		IOSize	nBytes
	)

private

Definition at line 105 of file StreamerInputFile.cc.

References cms::Exception::addContext(), edm::errors::FileReadError, gen::n, and storage_.

Referenced by newHeader(), readEventMessage(), and readStartMessage().

                                                               {
     IOSize n = 0;
     try {
       n = storage_->read(buf, nBytes);
     }
     catch(cms::Exception& ce) {
       Exception ex(errors::FileReadError, "", ce);
       ex.addContext("Calling StreamerInputFile::readBytes()");
       throw ex;
     }
     return n;
   }

int edm::StreamerInputFile::readEventMessage ( )

private

Definition at line 224 of file StreamerInputFile.cc.

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

Referenced by newHeader(), and next().

                                           {
     if(endOfFile_) return 0;
 
     bool eventRead = false;
     while(!eventRead) {
 
       IOSize nWant = sizeof(EventHeader);
       IOSize nGot = readBytes(&eventBuf_[0], nWant);
       if(nGot == 0) {
         // no more data available
         endOfFile_ = true;
         return 0;
       }
       if(nGot != nWant) {
         throw edm::Exception(errors::FileReadError, "StreamerInputFile::readEventMessage")
           << "Failed reading streamer file, first read in readEventMessage\n"
           << "Requested " << nWant << " bytes, read function returned " << nGot << " bytes\n";
       }
       HeaderView head(&eventBuf_[0]);
       uint32 code = head.code();
 
       // If it is not an event then something is wrong.
       if(code != Header::EVENT) {
         throw Exception(errors::FileReadError, "StreamerInputFile::readEventMessage")
           << "Failed reading streamer file, unknown code in event header\n"
           << "code = " << code << "\n";
       }
       uint32 eventSize = head.size();
       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);
         nGot = readBytes(&eventBuf_[sizeof(EventHeader)], nWant);
         if(nGot != nWant) {
           throw Exception(errors::FileReadError, "StreamerInputFile::readEventMessage")
             << "Failed reading streamer file, second read in readEventMessage\n"
             << "Requested " << nWant << " bytes, read function returned " << nGot << " bytes\n";
         }
       } 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";
         }
       }
     }
     currentEvMsg_ = std::make_shared<EventMsgView>((void*)&eventBuf_[0]); // propagate_const<T> has no reset() function
     return 1;
   }

void edm::StreamerInputFile::readStartMessage ( )

private

Not an init message should return

Definition at line 133 of file StreamerInputFile.cc.

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

Referenced by compareHeader(), newHeader(), and StreamerInputFile().

                                            {
     IOSize nWant = sizeof(HeaderView);
     IOSize nGot = readBytes(&headerBuf_[0], nWant);
     if(nGot != nWant) {
       throw Exception(errors::FileReadError, "StreamerInputFile::readStartMessage")
         << "Failed reading streamer file, first read in readStartMessage\n";
     }
 
     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;
     }
 
     uint32 headerSize = head.size();
     if(headerBuf_.size() < headerSize) headerBuf_.resize(headerSize);
 
     if(headerSize > sizeof(HeaderView)) {
       nWant = headerSize - sizeof(HeaderView);
       nGot = readBytes(&headerBuf_[sizeof(HeaderView)], nWant);
       if(nGot != nWant) {
         throw Exception(errors::FileReadError, "StreamerInputFile::readStartMessage")
           << "Failed reading streamer file, second read in readStartMessage\n";
       }
     } else {
       throw Exception(errors::FileReadError, "StreamerInputFile::readStartMessage")
         << "Failed reading streamer file, init header size from data too small\n";
     }
 
     startMsg_ = std::make_shared<InitMsgView>(&headerBuf_[0]); // propagate_const<T> has no reset() function
   }

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

private

Definition at line 118 of file StreamerInputFile.cc.

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

Referenced by newHeader(), and readEventMessage().

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