#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::shared_ptr< EventSkipperByID > eventSkipperByID=std::shared_ptr< EventSkipperByID >())

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

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

	~StreamerInputFile ()

Private Member Functions
bool	compareHeader ()

void	logFileAction (char const *msg)

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

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< FileCatalogItem >	streamerNames_

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 34 of file StreamerInputFile.h.

34 { 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>()`
	)

explicit

Reads a Streamer file

Definition at line 21 of file StreamerInputFile.cc.

       : 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, LFN);
     readStartMessage();
   }

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

◆ StreamerInputFile() [2/3]

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

explicit

Definition at line 43 of file StreamerInputFile.cc.

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

◆ StreamerInputFile() [3/3]

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

explicit

Multiple Streamer files

Definition at line 46 of file StreamerInputFile.cc.

       : 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).fileName(0), names.at(0).logicalFileName());
     ++currentFile_;
     readStartMessage();
     currRun_ = startMsg_->run();
     currProto_ = startMsg_->protocolVersion();
   }

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

◆ ~StreamerInputFile()

edm::StreamerInputFile::~StreamerInputFile ( )

Definition at line 19 of file StreamerInputFile.cc.

19 { closeStreamerFile(); }

References closeStreamerFile().

Member Function Documentation

◆ closeStreamerFile()

void edm::StreamerInputFile::closeStreamerFile ( )

Test bit if a new header is encountered

Definition at line 105 of file StreamerInputFile.cc.

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

References currentFileOpen_, logFileAction(), and storage_.

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

◆ 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 209 of file StreamerInputFile.cc.

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

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

Referenced by openNextFile().

◆ currentRecord()

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

inline

Points to File Start Header/Message

Definition at line 40 of file StreamerInputFile.h.

40 { return currentEvMsg_.get(); }

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

◆ logFileAction()

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

private

Definition at line 285 of file StreamerInputFile.cc.

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

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

Referenced by closeStreamerFile(), and openStreamerFile().

◆ newHeader()

bool edm::StreamerInputFile::newHeader ( )

inline

Points to current Record

Definition at line 43 of file StreamerInputFile.h.

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

References newHeader_, and createJobs::tmp.

◆ next()

StreamerInputFile::Next edm::StreamerInputFile::next ( void )

Definition at line 175 of file StreamerInputFile.cc.

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

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

◆ openNextFile()

bool edm::StreamerInputFile::openNextFile ( )

Definition at line 189 of file StreamerInputFile.cc.

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

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

◆ openStreamerFile()

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

private

Definition at line 69 of file StreamerInputFile.cc.

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

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

Referenced by openNextFile(), and StreamerInputFile().

◆ readBytes()

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

private

Definition at line 113 of file StreamerInputFile.cc.

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

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

Referenced by readEventMessage(), and readStartMessage().

◆ readEventMessage()

int edm::StreamerInputFile::readEventMessage ( )

private

Definition at line 223 of file StreamerInputFile.cc.

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

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 next().

◆ readStartMessage()

void edm::StreamerInputFile::readStartMessage ( )

private

Not an init message should return

Definition at line 139 of file StreamerInputFile.cc.

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

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

Referenced by compareHeader(), and StreamerInputFile().

◆ skipBytes()

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

private

Definition at line 125 of file StreamerInputFile.cc.

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

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

Referenced by readEventMessage().

◆ startMessage()

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

inline

Moves the handler to next Event Record

Definition at line 37 of file StreamerInputFile.h.

37 { return startMsg_.get(); }

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

Member Data Documentation

◆ currentEvMsg_

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

private

Definition at line 67 of file StreamerInputFile.h.

Referenced by currentRecord(), and readEventMessage().

◆ currentFile_

unsigned int edm::StreamerInputFile::currentFile_

private

Buffer to store Event Data

Definition at line 72 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 75 of file StreamerInputFile.h.

Referenced by logFileAction(), and openStreamerFile().

◆ currentFileOpen_

bool edm::StreamerInputFile::currentFileOpen_

private

Definition at line 76 of file StreamerInputFile.h.

Referenced by closeStreamerFile(), and openStreamerFile().

◆ currProto_

uint32 edm::StreamerInputFile::currProto_

private

Definition at line 81 of file StreamerInputFile.h.

Referenced by compareHeader(), and StreamerInputFile().

◆ currRun_

uint32 edm::StreamerInputFile::currRun_

private

Definition at line 80 of file StreamerInputFile.h.

Referenced by compareHeader(), and StreamerInputFile().

◆ endOfFile_

bool edm::StreamerInputFile::endOfFile_

private

Definition at line 87 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 70 of file StreamerInputFile.h.

Referenced by readEventMessage().

◆ eventSkipperByID_

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

private

Definition at line 78 of file StreamerInputFile.h.

Referenced by readEventMessage().

◆ headerBuf_

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

private

Definition at line 69 of file StreamerInputFile.h.

Referenced by readStartMessage().

◆ multiStreams_

bool edm::StreamerInputFile::multiStreams_

private

names of Streamer files

Definition at line 74 of file StreamerInputFile.h.

Referenced by next().

◆ newHeader_

bool edm::StreamerInputFile::newHeader_

private

Definition at line 83 of file StreamerInputFile.h.

Referenced by newHeader(), and next().

◆ startMsg_

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

private

Definition at line 66 of file StreamerInputFile.h.

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

◆ storage_

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

private

Definition at line 85 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 73 of file StreamerInputFile.h.

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

Public Types

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Member Enumeration Documentation

◆ Next

Constructor & Destructor Documentation

◆ StreamerInputFile() [1/3]

◆ StreamerInputFile() [2/3]

◆ StreamerInputFile() [3/3]

◆ ~StreamerInputFile()

Member Function Documentation

◆ closeStreamerFile()

◆ compareHeader()

◆ currentRecord()

◆ logFileAction()

◆ newHeader()

◆ next()

◆ openNextFile()

◆ openStreamerFile()

◆ readBytes()

◆ readEventMessage()

◆ readStartMessage()

◆ skipBytes()

◆ startMessage()

Member Data Documentation

◆ currentEvMsg_

◆ currentFile_

◆ currentFileName_

◆ currentFileOpen_

◆ currProto_

◆ currRun_

◆ endOfFile_

◆ eventBuf_

◆ eventSkipperByID_

◆ headerBuf_

◆ multiStreams_

◆ newHeader_

◆ startMsg_

◆ storage_

◆ streamerNames_