#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 20 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 40 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().

18 { closeStreamerFile(); }

edm::StreamerInputFile::closeStreamerFile

void closeStreamerFile()

Needs to be public because of forking.

Definition: StreamerInputFile.cc:88

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 88 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 194 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 35 of file StreamerInputFile.h.

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

Referenced by WatcherStreamFileReader::getNextEvent().

35 { return currentEvMsg_.get(); }

edm::propagate_const::get

element_type const * get() const

Definition: propagate_const.h:64

edm::StreamerInputFile::currentEvMsg_

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

Definition: StreamerInputFile.h:63

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

private

Definition at line 271 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 38 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().

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

bool edm::StreamerInputFile::next ( void )

Definition at line 158 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 174 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 63 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 96 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 209 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 122 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 108 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;
   }