#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 >())

	StreamerInputFile (std::string const &name, 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)

storage::IOSize	readBytes (char *buf, storage::IOSize nBytes)

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_

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

edm::shared_memory::channel_names::kStop

constexpr char const *const kStop

Definition: channel_names.h:34

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.

References Skims_PA_cff::name, 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, LFN);
     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) {}

edm::StreamerInputFile::StreamerInputFile

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

Definition: StreamerInputFile.cc:21

Skims_PA_cff.name

name

Definition: Skims_PA_cff.py:17

◆ 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.

References currentFile_, currProto_, currRun_, names, 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).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().

19 { closeStreamerFile(); }

edm::StreamerInputFile::closeStreamerFile

void closeStreamerFile()

Definition: StreamerInputFile.cc:106

Member Function Documentation

◆ closeStreamerFile()

void edm::StreamerInputFile::closeStreamerFile ( )

Test bit if a new header is encountered

Definition at line 106 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 211 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 40 of file StreamerInputFile.h.

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

40 { return currentEvMsg_.get(); }

edm::propagate_const::get

constexpr element_type const * get() const

Definition: propagate_const.h:64

edm::StreamerInputFile::currentEvMsg_

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

Definition: StreamerInputFile.h:67

◆ logFileAction()

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

private

Definition at line 288 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 43 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 177 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 191 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 69 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()

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

private

Definition at line 114 of file StreamerInputFile.cc.

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

Referenced by readEventMessage(), and readStartMessage().

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

◆ readEventMessage()

int edm::StreamerInputFile::readEventMessage ( )

private

Definition at line 225 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 next().

                                           {
     if (endOfFile_)
       return 0;
 
     using namespace edm::storage;
     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;
   }

◆ readStartMessage()

void edm::StreamerInputFile::readStartMessage ( )

private

Not an init message should return

Definition at line 140 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(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
   }

◆ skipBytes()

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

private

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

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

37 { return startMsg_.get(); }

edm::propagate_const::get

constexpr element_type const * get() const

Definition: propagate_const.h:64

edm::StreamerInputFile::startMsg_

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

Definition: StreamerInputFile.h:66

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<edm::storage::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_