#include <StreamHandler.h>

Inheritance diagram for stor::StreamHandler:

Public Member Functions
void	closeAllFiles ()

void	closeFilesForLumiSection (const uint32_t lumiSection)

void	closeFilesForLumiSection (const uint32_t &lumiSection, std::string &)

void	closeTimedOutFiles (utils::TimePoint_t currentTime=utils::getCurrentTime())

	StreamHandler (const SharedResourcesPtr, const DbFileHandlerPtr)

void	writeEvent (const I2OChain &event)

virtual	~StreamHandler ()

Protected Types
typedef std::map< std::string, unsigned int >	CoreFileNamesMap

typedef boost::shared_ptr < FileHandler >	FileHandlerPtr

typedef std::vector < FileHandlerPtr >	FileHandlers

Protected Member Functions
virtual double	fractionToDisk () const =0

virtual FileHandlerPtr	getFileHandler (const I2OChain &event)

unsigned long long	getMaxFileSize () const

FilesMonitorCollection::FileRecordPtr	getNewFileRecord (const I2OChain &event)

virtual int	getStreamMaxFileSize () const =0

virtual FileHandlerPtr	newFileHandler (const I2OChain &event)=0

virtual std::string	streamLabel () const =0

Protected Attributes
const DbFileHandlerPtr	dbFileHandler_

const DiskWritingParams	diskWritingParams_

FileHandlers	fileHandlers_

const SharedResourcesPtr	sharedResources_

const StatisticsReporterPtr	statReporter_

const StreamsMonitorCollection::StreamRecordPtr	streamRecord_

CoreFileNamesMap	usedCoreFileNames_

Private Member Functions
bool	fileTooLarge (const FileHandlerPtr, const unsigned long &dataSize) const

std::string	getBaseFilePath (const uint32_t &runNumber, uint32_t fileCount) const

std::string	getCoreFileName (const uint32_t &runNumber, const uint32_t &lumiSection) const

unsigned int	getFileCounter (const std::string &coreFileName)

std::string	getFileSystem (const uint32_t &runNumber, uint32_t fileCount) const

StreamHandler &	operator= (StreamHandler const &)

	StreamHandler (StreamHandler const &)

Detailed Description

Abstract class to handle one stream written to disk.

Author:: mommsen

Revision:: 1.13.6.1

Date:: 2011/03/07 11:33:04

Definition at line 31 of file StreamHandler.h.

Member Typedef Documentation

typedef std::map<std::string, unsigned int> stor::StreamHandler::CoreFileNamesMap

protected

Definition at line 149 of file StreamHandler.h.

typedef boost::shared_ptr<FileHandler> stor::StreamHandler::FileHandlerPtr

protected

Definition at line 71 of file StreamHandler.h.

typedef std::vector<FileHandlerPtr> stor::StreamHandler::FileHandlers

protected

Definition at line 146 of file StreamHandler.h.

Constructor & Destructor Documentation

stor::StreamHandler::StreamHandler	(	const SharedResourcesPtr	sharedResources,
		const DbFileHandlerPtr	dbFileHandler
	)

Definition at line 16 of file StreamHandler.cc.

     :
   sharedResources_(sharedResources),
   statReporter_(sharedResources->statisticsReporter_),
   streamRecord_(statReporter_->getStreamsMonitorCollection().getNewStreamRecord()),
   diskWritingParams_(sharedResources->configuration_->getDiskWritingParams()),
   dbFileHandler_(dbFileHandler)
   {}

virtual stor::StreamHandler::~StreamHandler ( )

inlinevirtual

Definition at line 37 of file StreamHandler.h.

37 {};

stor::StreamHandler::StreamHandler ( StreamHandler const & )

private

Member Function Documentation

void stor::StreamHandler::closeAllFiles ( )

Gracefully close all open files

Definition at line 28 of file StreamHandler.cc.

References ExpressReco_HICollisions_FallBack::e, stor::AlarmHandler::ERROR, cmsCodeRules.cppFunctionSkipper::exception, edm::hlt::Exception, fileHandlers_, stor::FilesMonitorCollection::FileRecord::runEnded, statReporter_, and streamLabel().

Referenced by stor::DiskWriter::destroyStreams(), and stor::FaultyEventStreamHandler::getFileHandler().

   {
     std::string errorMsg = "Failed to close all files for stream " + streamLabel() + ": ";
     
     for (FileHandlers::const_iterator it = fileHandlers_.begin(),
            itEnd = fileHandlers_.end(); it != itEnd; ++it)
     {
       try
       {
         (*it)->closeFile(FilesMonitorCollection::FileRecord::runEnded);
       }
       catch(xcept::Exception& e)
       {
         XCEPT_DECLARE_NESTED( stor::exception::DiskWriting,
           sentinelException, errorMsg, e );
         statReporter_->alarmHandler()->
           notifySentinel(AlarmHandler::ERROR, sentinelException);
       }
       catch(std::exception &e)
       {
         errorMsg += e.what();
         XCEPT_DECLARE( stor::exception::DiskWriting,
           sentinelException, errorMsg );
         statReporter_->alarmHandler()->
           notifySentinel(AlarmHandler::ERROR, sentinelException);
       }
       catch(...)
       {
         errorMsg += "Unknown exception";
         XCEPT_DECLARE( stor::exception::DiskWriting,
           sentinelException, errorMsg );
         statReporter_->alarmHandler()->
           notifySentinel(AlarmHandler::ERROR, sentinelException);
       }
     }
     fileHandlers_.clear();
   }

void stor::StreamHandler::closeFilesForLumiSection ( const uint32_t lumiSection )

Close all files which belong to the given lumi section

Definition at line 89 of file StreamHandler.cc.

References fileHandlers_, and stor::FileHandler::isFromLumiSection().

   {
     fileHandlers_.erase(
       std::remove_if(fileHandlers_.begin(),
         fileHandlers_.end(),
         boost::bind(&FileHandler::isFromLumiSection,
           _1, lumiSection)),
       fileHandlers_.end());
   }

void stor::StreamHandler::closeFilesForLumiSection	(	const uint32_t &	lumiSection,
		std::string &	str
	)

Close all files which belong to the given lumi section and print number of files for this lumi section into the passed string.

Definition at line 79 of file StreamHandler.cc.

   {
     streamRecord_->reportLumiSectionInfo(lumiSection, str);
     closeFilesForLumiSection(lumiSection);
   }

void stor::StreamHandler::closeTimedOutFiles ( utils::TimePoint_t currentTime = utils::getCurrentTime() )

Close all files which are have not seen any recent events

Definition at line 67 of file StreamHandler.cc.

References fileHandlers_, and stor::FileHandler::tooOld().

Referenced by stor::DiskWriter::closeTimedOutFiles().

   {
     fileHandlers_.erase(
       std::remove_if(fileHandlers_.begin(),
         fileHandlers_.end(),
         boost::bind(&FileHandler::tooOld,
           _1, currentTime)),
       fileHandlers_.end());
   }

bool stor::StreamHandler::fileTooLarge	(	const FileHandlerPtr	,
		const unsigned long &	dataSize
	)		const

private

Return true if the file would become too large when adding dataSize in Bytes

virtual double stor::StreamHandler::fractionToDisk ( ) const

protectedpure virtual

Return the fraction-to-disk parameter

Implemented in stor::EventStreamHandler, stor::FRDStreamHandler, and stor::FaultyEventStreamHandler.

Referenced by stor::FaultyEventStreamHandler::FaultyEventStreamHandler().

std::string stor::StreamHandler::getBaseFilePath	(	const uint32_t &	runNumber,
		uint32_t	fileCount
	)		const

private

Get path w/o working directory

Definition at line 167 of file StreamHandler.cc.

Referenced by getNewFileRecord().

   {
     return diskWritingParams_.filePath_ + getFileSystem(runNumber, fileCount);
   }

std::string stor::StreamHandler::getCoreFileName	(	const uint32_t &	runNumber,
		const uint32_t &	lumiSection
	)		const

private

Get the core file name

Definition at line 200 of file StreamHandler.cc.

References python.StorageManager_cfg::streamLabel.

Referenced by getNewFileRecord().

   {
     std::ostringstream coreFileName;
     coreFileName << diskWritingParams_.setupLabel_
       << "." << std::setfill('0') << std::setw(8) << runNumber
       << "." << std::setfill('0') << std::setw(4) << lumiSection
       << "." << streamLabel()
       << "." << diskWritingParams_.fileName_
       << "." << std::setfill('0') << std::setw(2) << diskWritingParams_.smInstanceString_;
     
     return coreFileName.str();
   }

unsigned int stor::StreamHandler::getFileCounter ( const std::string & coreFileName )

private

Get the instance count of this core file name

Definition at line 217 of file StreamHandler.cc.

References pos, and usedCoreFileNames_.

Referenced by getNewFileRecord().

   {
     CoreFileNamesMap::iterator pos = usedCoreFileNames_.find(coreFileName);
     if (pos == usedCoreFileNames_.end())
     {
       usedCoreFileNames_.insert(pos, std::make_pair(coreFileName, 0));
       return 0;
     }
     else
     {
       ++(pos->second);
       return pos->second;
     }
   }

StreamHandler::FileHandlerPtr stor::StreamHandler::getFileHandler ( const I2OChain & event )

protectedvirtual

Get the file handler responsible for the event

Reimplemented in stor::FaultyEventStreamHandler.

Definition at line 110 of file StreamHandler.cc.

References fileHandlers_, newFileHandler(), and stor::I2OChain::totalDataSize().

Referenced by writeEvent().

   {
     for (
       FileHandlers::iterator it = fileHandlers_.begin(), itEnd = fileHandlers_.end();
       it != itEnd;
       ++it
     ) 
     {
       if ( (*it)->lumiSection() == event.lumiSection() )
       {
         if ( (*it)->tooLarge(event.totalDataSize()) )
         { 
           fileHandlers_.erase(it);
           break;
         }
         else
         {
           return (*it);
         }
       }
     }    
     return newFileHandler(event);
   }

std::string stor::StreamHandler::getFileSystem	(	const uint32_t &	runNumber,
		uint32_t	fileCount
	)		const

private

Get file system string

Definition at line 177 of file StreamHandler.cc.

   {
     // if the number of logical disks is not specified, don't
     // add a file system subdir to the path
     if (diskWritingParams_.nLogicalDisk_ <= 0) {return "";}
     
     unsigned int fileSystemNumber =
       (runNumber
         + atoi(diskWritingParams_.smInstanceString_.c_str())
         + fileCount)
       % diskWritingParams_.nLogicalDisk_;
     
     std::ostringstream fileSystem;
     fileSystem << "/" << std::setfill('0') << std::setw(2) << fileSystemNumber; 
     
     return fileSystem.str();
   }

unsigned long long stor::StreamHandler::getMaxFileSize ( ) const

protected

Return the maximum file size in bytes

Definition at line 233 of file StreamHandler.cc.

References diskWritingParams_, getStreamMaxFileSize(), and stor::DiskWritingParams::maxFileSizeMB_.

Referenced by stor::FRDStreamHandler::newFileHandler(), and stor::EventStreamHandler::newFileHandler().

   {
     const unsigned long long maxFileSizeMB =
       diskWritingParams_.maxFileSizeMB_ > 0 ? 
       diskWritingParams_.maxFileSizeMB_ : getStreamMaxFileSize();
     
     return ( maxFileSizeMB * 1024 * 1024 );
   }

FilesMonitorCollection::FileRecordPtr stor::StreamHandler::getNewFileRecord ( const I2OChain & event )

protected

Return a new file record for the event

Definition at line 136 of file StreamHandler.cc.

References ExpressReco_HICollisions_FallBack::e, getBaseFilePath(), getCoreFileName(), getFileCounter(), stor::FilesMonitorCollection::FileRecord::notClosed, sharedResources_, statReporter_, streamLabel(), and streamRecord_.

Referenced by stor::FRDStreamHandler::newFileHandler(), stor::EventStreamHandler::newFileHandler(), and stor::FaultyEventStreamHandler::newFileHandler().

   {
     FilesMonitorCollection::FileRecordPtr fileRecord =
       statReporter_->getFilesMonitorCollection().getNewFileRecord();
     
     try
     {
       fileRecord->runNumber = event.runNumber();
       fileRecord->lumiSection = event.lumiSection();
     }
     catch(stor::exception::IncompleteEventMessage &e)
     {
       fileRecord->runNumber = sharedResources_->configuration_->getRunNumber();
       fileRecord->lumiSection = 0;
     }
     fileRecord->streamLabel = streamLabel();
     fileRecord->baseFilePath =
       getBaseFilePath(fileRecord->runNumber, fileRecord->entryCounter);
     fileRecord->coreFileName =
       getCoreFileName(fileRecord->runNumber, fileRecord->lumiSection);
     fileRecord->fileCounter = getFileCounter(fileRecord->coreFileName);
     fileRecord->whyClosed = FilesMonitorCollection::FileRecord::notClosed;
     fileRecord->isOpen = true;
     
     streamRecord_->incrementFileCount(fileRecord->lumiSection);
     
     return fileRecord;
   }

virtual int stor::StreamHandler::getStreamMaxFileSize ( ) const

protectedpure virtual

Return the maximum file size for the stream in MB

Implemented in stor::FaultyEventStreamHandler, stor::EventStreamHandler, and stor::FRDStreamHandler.

Referenced by getMaxFileSize().

virtual FileHandlerPtr stor::StreamHandler::newFileHandler ( const I2OChain & event )

protectedpure virtual

Return a new file handler for the provided event

Implemented in stor::FaultyEventStreamHandler, stor::EventStreamHandler, and stor::FRDStreamHandler.

Referenced by getFileHandler().

StreamHandler& stor::StreamHandler::operator= ( StreamHandler const & )

private

virtual std::string stor::StreamHandler::streamLabel ( ) const

protectedpure virtual

Return the stream label

Implemented in stor::EventStreamHandler, stor::FaultyEventStreamHandler, and stor::FRDStreamHandler.

Referenced by closeAllFiles(), and getNewFileRecord().

void stor::StreamHandler::writeEvent ( const I2OChain & event )

Write the event to the stream file

Definition at line 100 of file StreamHandler.cc.

References getFileHandler(), statReporter_, streamRecord_, and stor::I2OChain::totalDataSize().

   {
     FileHandlerPtr handler = getFileHandler(event);
     handler->writeEvent(event);
     streamRecord_->addSizeInBytes(event.totalDataSize());
     statReporter_->getThroughputMonitorCollection().
       addDiskWriteSample(event.totalDataSize());
   }