#include <RecoEventOutputModuleForFU.h>

Inheritance diagram for evf::RecoEventOutputModuleForFU< Consumer >:

Public Member Functions
	RecoEventOutputModuleForFU (edm::ParameterSet const &ps)

virtual	~RecoEventOutputModuleForFU ()

Public Member Functions inherited from edm::StreamerOutputModuleBase
	StreamerOutputModuleBase (ParameterSet const &ps)

virtual	~StreamerOutputModuleBase ()

Public Member Functions inherited from edm::one::OutputModule< one::WatchRuns, one::WatchLuminosityBlocks >
	OutputModule (edm::ParameterSet const &iPSet)

Public Member Functions inherited from edm::one::OutputModuleBase
BranchIDLists const *	branchIDLists ()

std::array< bool, NumBranchTypes > const &	hasNewlyDroppedBranch () const

SelectedProductsForBranchType const &	keptProducts () const

int	maxEvents () const

const ModuleDescription &	moduleDescription () const

OutputModuleBase &	operator= (OutputModuleBase const &)=delete

	OutputModuleBase (ParameterSet const &pset)

	OutputModuleBase (OutputModuleBase const &)=delete

std::string const &	processName () const

int	remainingEvents () const

bool	selected (BranchDescription const &desc) const

void	selectProducts (ProductRegistry const &preg, ThinnedAssociationsHelper const &)

ThinnedAssociationsHelper const *	thinnedAssociationsHelper () const

bool	wantAllEvents () const

virtual	~OutputModuleBase ()

Public Member Functions inherited from edm::EDConsumerBase
std::vector< ConsumesInfo >	consumesInfo () const

	EDConsumerBase ()

ProductHolderIndexAndSkipBit	indexFrom (EDGetToken, BranchType, TypeID const &) const

void	itemsMayGet (BranchType, std::vector< ProductHolderIndexAndSkipBit > &) const

void	itemsToGet (BranchType, std::vector< ProductHolderIndexAndSkipBit > &) const

std::vector < ProductHolderIndexAndSkipBit > const &	itemsToGetFromEvent () const

void	labelsForToken (EDGetToken iToken, Labels &oLabels) const

void	modulesDependentUpon (std::string const &iProcessName, std::string const &iModuleLabel, bool iPrint, std::vector< char const * > &oModuleLabels) const

void	modulesWhoseProductsAreConsumed (std::vector< ModuleDescription const * > &modules, ProductRegistry const &preg, std::map< std::string, ModuleDescription const * > const &labelsToDesc, std::string const &processName) const

bool	registeredToConsume (ProductHolderIndex, bool, BranchType) const

bool	registeredToConsumeMany (TypeID const &, BranchType) const

void	updateLookup (BranchType iBranchType, ProductHolderIndexHelper const &)

virtual	~EDConsumerBase ()

Static Public Member Functions
static void	fillDescriptions (edm::ConfigurationDescriptions &descriptions)

Static Public Member Functions inherited from edm::StreamerOutputModuleBase
static void	fillDescription (ParameterSetDescription &desc)

Static Public Member Functions inherited from edm::one::OutputModuleBase
static const std::string &	baseType ()

static void	fillDescription (ParameterSetDescription &desc)

static void	fillDescriptions (ConfigurationDescriptions &descriptions)

static void	prevalidate (ConfigurationDescriptions &)

Private Member Functions
virtual void	beginJob () override

virtual void	beginLuminosityBlock (edm::LuminosityBlockPrincipal const &, edm::ModuleCallingContext const *) override

virtual void	doOutputEvent (EventMsgBuilder const &msg) override

virtual void	doOutputHeader (InitMsgBuilder const &init_message) override

virtual void	endLuminosityBlock (edm::LuminosityBlockPrincipal const &, edm::ModuleCallingContext const *) override

void	initRun ()

virtual void	start () override

virtual void	stop () override

Private Attributes
jsoncollector::IntJ	accepted_

std::auto_ptr< Consumer >	c_

jsoncollector::IntJ	errorEvents_

jsoncollector::IntJ	fileAdler32_

jsoncollector::StringJ	filelist_

jsoncollector::IntJ	filesize_

evf::FastMonitoringService *	fms_

jsoncollector::IntJ	hltErrorEvents_

jsoncollector::StringJ	inputFiles_

boost::shared_ptr < jsoncollector::FastMonitor >	jsonMonitor_

jsoncollector::StringJ	mergeType_

boost::filesystem::path	openDatChecksumFilePath_

boost::filesystem::path	openDatFilePath_

unsigned char *	outBuf_ = 0

jsoncollector::DataPointDefinition	outJsonDef_

jsoncollector::IntJ	processed_

bool	readAdler32Check_ =false

jsoncollector::IntJ	retCodeMask_

std::string	stream_label_

jsoncollector::StringJ	transferDestination_

Additional Inherited Members
Public Types inherited from edm::one::OutputModuleBase
typedef OutputModuleBase	ModuleType

Public Types inherited from edm::EDConsumerBase
typedef ProductLabels	Labels

Protected Member Functions inherited from edm::one::OutputModuleBase
void	configure (OutputModuleDescription const &desc)

ModuleDescription const &	description () const

void	doBeginJob ()

bool	doBeginLuminosityBlock (LuminosityBlockPrincipal const &lbp, EventSetup const &c, ModuleCallingContext const *)

bool	doBeginRun (RunPrincipal const &rp, EventSetup const &c, ModuleCallingContext const *)

void	doEndJob ()

bool	doEndLuminosityBlock (LuminosityBlockPrincipal const &lbp, EventSetup const &c, ModuleCallingContext const *)

bool	doEndRun (RunPrincipal const &rp, EventSetup const &c, ModuleCallingContext const *)

bool	doEvent (EventPrincipal const &ep, EventSetup const &c, ActivityRegistry , ModuleCallingContext const )

void	doPreallocate (PreallocationConfiguration const &)

std::map< BranchID::value_type, BranchID::value_type > const &	droppedBranchIDToKeptBranchID ()

ParameterSetID	selectorConfig () const

void	setEventSelectionInfo (std::map< std::string, std::vector< std::pair< std::string, int > > > const &outputModulePathPositions, bool anyProductProduced)

Protected Member Functions inherited from edm::EDConsumerBase
template<typename ProductType , BranchType B = InEvent>
EDGetTokenT< ProductType >	consumes (edm::InputTag const &tag)

EDGetToken	consumes (const TypeToGet &id, edm::InputTag const &tag)

template<BranchType B>
EDGetToken	consumes (TypeToGet const &id, edm::InputTag const &tag)

ConsumesCollector	consumesCollector ()
	Use a ConsumesCollector to gather consumes information from helper functions. More...

template<typename ProductType , BranchType B = InEvent>
void	consumesMany ()

void	consumesMany (const TypeToGet &id)

template<BranchType B>
void	consumesMany (const TypeToGet &id)

template<typename ProductType , BranchType B = InEvent>
EDGetTokenT< ProductType >	mayConsume (edm::InputTag const &tag)

EDGetToken	mayConsume (const TypeToGet &id, edm::InputTag const &tag)

template<BranchType B>
EDGetToken	mayConsume (const TypeToGet &id, edm::InputTag const &tag)

Detailed Description

template<typename Consumer>
class evf::RecoEventOutputModuleForFU< Consumer >

Definition at line 25 of file RecoEventOutputModuleForFU.h.

Constructor & Destructor Documentation

template<typename Consumer >

evf::RecoEventOutputModuleForFU< Consumer >::RecoEventOutputModuleForFU ( edm::ParameterSet const & ps )

explicit

Consumers are supposed to provide

void doOutputHeader(InitMsgBuilder const& init_message) void doOutputEvent(EventMsgBuilder const& msg) void start() void stop() static void fillDescription(ParameterSetDescription&)

Definition at line 77 of file RecoEventOutputModuleForFU.h.

References Exception, evf::RecoEventOutputModuleForFU< Consumer >::fms_, evf::RecoEventOutputModuleForFU< Consumer >::stream_label_, and AlCaHLTBitMon_QueryRunRegistry::string.

                                                                                             :
     edm::one::OutputModuleBase::OutputModuleBase(ps),
     edm::StreamerOutputModuleBase(ps),
     c_(new Consumer(ps)),
     stream_label_(ps.getParameter<std::string>("@module_label")),
     processed_(0),
     accepted_(0),
     errorEvents_(0),
     retCodeMask_(0),
     filelist_(),
     filesize_(0),
     inputFiles_(),
     fileAdler32_(1),
     transferDestination_(),
     mergeType_(),
     hltErrorEvents_(0),
     outBuf_(new unsigned char[1024*1024])
   {
     //replace hltOutoputA with stream if the HLT menu uses this convention
     std::string testPrefix="hltOutput";
     if (stream_label_.find(testPrefix)==0) 
       stream_label_=std::string("stream")+stream_label_.substr(testPrefix.size());
 
     if (stream_label_.find("_")!=std::string::npos) {
       throw cms::Exception("RecoEventOutputModuleForFU")
         << "Underscore character is reserved can not be used for stream names in FFF, but was detected in stream name -: " << stream_label_;
     }
 
     std::string stream_label_lo = stream_label_;
     boost::algorithm::to_lower(stream_label_lo);
     auto streampos = stream_label_lo.rfind("stream");
     if (streampos !=0 && streampos!=std::string::npos)
       throw cms::Exception("RecoEventOutputModuleForFU")
         << "stream (case-insensitive) sequence was found in stream suffix. This is reserved and can not be used for names in FFF based HLT, but was detected in stream name";
 
     fms_ = (evf::FastMonitoringService *)(edm::Service<evf::MicroStateService>().operator->());
   }

template<typename Consumer >

evf::RecoEventOutputModuleForFU< Consumer >::~RecoEventOutputModuleForFU ( )

virtual

Definition at line 183 of file RecoEventOutputModuleForFU.h.

183 {}

Member Function Documentation

template<typename Consumer >

void evf::RecoEventOutputModuleForFU< Consumer >::beginJob ( void )

overrideprivatevirtual

Reimplemented from edm::StreamerOutputModuleBase.

Definition at line 259 of file RecoEventOutputModuleForFU.h.

References evf::MergeTypeDAT.

   {
     //get stream transfer destination
     transferDestination_ = edm::Service<evf::EvFDaqDirector>()->getStreamDestinations(stream_label_);
     mergeType_ = edm::Service<evf::EvFDaqDirector>()->getStreamMergeType(stream_label_,evf::MergeTypeDAT);
   }

template<typename Consumer >

void evf::RecoEventOutputModuleForFU< Consumer >::beginLuminosityBlock	(	edm::LuminosityBlockPrincipal const &	ls,
		edm::ModuleCallingContext const *
	)

overrideprivatevirtual

Definition at line 268 of file RecoEventOutputModuleForFU.h.

References edm::LuminosityBlockPrincipal::luminosityBlock().

   {
     //edm::LogInfo("RecoEventOutputModuleForFU") << "begin lumi";
     openDatFilePath_ = edm::Service<evf::EvFDaqDirector>()->getOpenDatFilePath(ls.luminosityBlock(),stream_label_);
     openDatChecksumFilePath_ = edm::Service<evf::EvFDaqDirector>()->getOpenDatFilePath(ls.luminosityBlock(),stream_label_);
     c_->setOutputFile(openDatFilePath_.string());
     filelist_ = openDatFilePath_.filename().string();
   }

template<typename Consumer >

void evf::RecoEventOutputModuleForFU< Consumer >::doOutputEvent ( EventMsgBuilder const & msg )

overrideprivatevirtual

Implements edm::StreamerOutputModuleBase.

Definition at line 244 of file RecoEventOutputModuleForFU.h.

                                                                                 {
     accepted_.value()++;
     c_->doOutputEvent(msg); // You can't use msg in RecoEventOutputModuleForFU after this point
   }

template<typename Consumer >

void evf::RecoEventOutputModuleForFU< Consumer >::doOutputHeader ( InitMsgBuilder const & init_message )

overrideprivatevirtual

Implements edm::StreamerOutputModuleBase.

Definition at line 207 of file RecoEventOutputModuleForFU.h.

References cms::Adler32(), Exception, alcazmumu_cfi::src, and AlCaHLTBitMon_QueryRunRegistry::string.

   {
     c_->doOutputHeader(init_message);
 
     const std::string openIniFileName = edm::Service<evf::EvFDaqDirector>()->getOpenInitFilePath(stream_label_);
     struct stat istat;
     stat(openIniFileName.c_str(), &istat);
     //read back file to check integrity of what was written
     off_t readInput=0;
     uint32_t adlera=1,adlerb=0;
     FILE *src = fopen(openIniFileName.c_str(),"r");
     while (readInput<istat.st_size)
     {
       size_t toRead=  readInput+1024*1024 < istat.st_size ? 1024*1024 : istat.st_size-readInput;
       fread(outBuf_,toRead,1,src);
       cms::Adler32((const char*)outBuf_,toRead,adlera,adlerb);
       readInput+=toRead;
     }
     fclose(src);
     //free output buffer if micromerge is not done by the module
     if (edm::Service<evf::EvFDaqDirector>()->microMergeDisabled()) {
       delete [] outBuf_;
       outBuf_=nullptr;
     }
     uint32_t adler32c = (adlerb << 16) | adlera;
     if (adler32c != c_->get_adler32_ini()) {
       throw cms::Exception("RecoEventOutputModuleForFU") << "Checksum mismatch of ini file -: " << openIniFileName
                            << " expected:" << c_->get_adler32_ini() << " obtained:" << adler32c;
     }
     else {
       edm::LogWarning("RecoEventOutputModuleForFU") << "Ini file checksum -: "<< stream_label_ << " " << adler32c;
       boost::filesystem::rename(openIniFileName,edm::Service<evf::EvFDaqDirector>()->getInitFilePath(stream_label_));
     }
   }

template<typename Consumer >

void evf::RecoEventOutputModuleForFU< Consumer >::endLuminosityBlock	(	edm::LuminosityBlockPrincipal const &	ls,
		edm::ModuleCallingContext const *
	)

overrideprivatevirtual

Definition at line 278 of file RecoEventOutputModuleForFU.h.

References cms::Adler32(), Exception, edm::LuminosityBlockPrincipal::luminosityBlock(), NULL, alcazmumu_cfi::src, and AlCaHLTBitMon_QueryRunRegistry::string.

   {
     //edm::LogInfo("RecoEventOutputModuleForFU") << "end lumi";
     long filesize=0;
     fileAdler32_.value() = c_->get_adler32();
     c_->closeOutputFile();
     bool abortFlag = false;
     processed_.value() = fms_->getEventsProcessedForLumi(ls.luminosityBlock(),&abortFlag);
 
     if (abortFlag) {
         edm::LogInfo("RecoEventOutputModuleForFU") << "output suppressed";
         return;
     }
     
     if(processed_.value()!=0) {
 
       //lock
       struct stat istat;
       if (!edm::Service<evf::EvFDaqDirector>()->microMergeDisabled()) {
         FILE *des = edm::Service<evf::EvFDaqDirector>()->maybeCreateAndLockFileHeadForStream(ls.luminosityBlock(),stream_label_);
 
         std::string deschecksum = edm::Service<evf::EvFDaqDirector>()->getMergedDatChecksumFilePath(ls.luminosityBlock(), stream_label_);
 
         struct stat istat;
         FILE * cf = NULL;
         uint32_t mergedAdler32=1;
         //get adler32 accumulated checksum for the merged file
         if (!stat(deschecksum.c_str(), &istat)) {
           if (istat.st_size) {
             cf = fopen(deschecksum.c_str(),"r");
             if (!cf) throw cms::Exception("RecoEventOutputModuleForFU") << "Unable to open checksum file -: " << deschecksum.c_str();
             fscanf(cf,"%u",&mergedAdler32);
             fclose(cf);
           }
           else edm::LogWarning("RecoEventOutputModuleForFU") << "Checksum file size is empty -: "<< deschecksum.c_str();
         }
 
         FILE *src = fopen(openDatFilePath_.string().c_str(),"r");
 
         stat(openDatFilePath_.string().c_str(), &istat);
         off_t readInput=0;
         uint32_t adlera=1;
         uint32_t adlerb=0;
         while (readInput<istat.st_size) {
           size_t toRead=  readInput+1024*1024 < istat.st_size ? 1024*1024 : istat.st_size-readInput;
           fread(outBuf_,toRead,1,src);
           fwrite(outBuf_,toRead,1,des);
           if (readAdler32Check_)
             cms::Adler32((const char*)outBuf_,toRead,adlera,adlerb);
           readInput+=toRead;
           filesize+=toRead;
         }
 
         //write new string representation of the checksum value
         cf = fopen(deschecksum.c_str(),"w");
         if (!cf) throw cms::Exception("RecoEventOutputModuleForFU") << "Unable to open or rewind checksum file for writing -:" << deschecksum.c_str();
 
         //write adler32 combine to checksum file 
         mergedAdler32 = adler32_combine(mergedAdler32,fileAdler32_.value(),filesize);
 
         fprintf(cf,"%u",mergedAdler32);
         fclose(cf);
 
         edm::Service<evf::EvFDaqDirector>()->unlockAndCloseMergeStream();
         fclose(src);
 
         if (readAdler32Check_ && ((adlerb << 16) | adlera) != fileAdler32_.value()) {
 
           throw cms::Exception("RecoEventOutputModuleForFU") << "Adler32 checksum mismatch after reading file -: " 
                                                            << openDatFilePath_.string() <<" in LS " << ls.luminosityBlock() << std::endl;
         }
       }
       else  { //no micromerge by HLT
         stat(openDatFilePath_.string().c_str(), &istat);
         filesize = istat.st_size;
         boost::filesystem::rename(openDatFilePath_.string().c_str(), edm::Service<evf::EvFDaqDirector>()->getDatFilePath(ls.luminosityBlock(),stream_label_));
       }
     } else {
       //return if not in empty lumisection mode
       if (!edm::Service<evf::EvFDaqDirector>()->emptyLumisectionMode()) {
         remove(openDatFilePath_.string().c_str());
         return;
       }
       filelist_ = "";
       fileAdler32_.value()=-1;
     }
 
     //remove file
     remove(openDatFilePath_.string().c_str());
     filesize_=filesize;
 
     jsonMonitor_->snap(ls.luminosityBlock());
     const std::string outputJsonNameStream =
       edm::Service<evf::EvFDaqDirector>()->getOutputJsonFilePath(ls.luminosityBlock(),stream_label_);
     jsonMonitor_->outputFullJSON(outputJsonNameStream,ls.luminosityBlock());
 
     // reset monitoring params
     accepted_.value() = 0;
     filelist_ = "";
   }

template<typename Consumer >

void evf::RecoEventOutputModuleForFU< Consumer >::fillDescriptions ( edm::ConfigurationDescriptions & descriptions )

static

Definition at line 251 of file RecoEventOutputModuleForFU.h.

References edm::ConfigurationDescriptions::add(), and edm::StreamerOutputModuleBase::fillDescription().

                                                                                                    {
     edm::ParameterSetDescription desc;
     edm::StreamerOutputModuleBase::fillDescription(desc);
     Consumer::fillDescription(desc);
     descriptions.add("EvFOutputModule", desc);
   }

template<typename Consumer >

void evf::RecoEventOutputModuleForFU< Consumer >::initRun ( void )

private

Definition at line 116 of file RecoEventOutputModuleForFU.h.

References jsoncollector::DataPointDefinition::ADLER32, jsoncollector::DataPointDefinition::BINARYOR, jsoncollector::DataPointDefinition::CAT, ws_sso_content_reader::content, LogDebug, jsoncollector::DataPointDefinition::MERGE, jsoncollector::DataPointDefinition::SAME, jsoncollector::JSONSerializer::serialize(), contentValuesCheck::ss, AlCaHLTBitMon_QueryRunRegistry::string, jsoncollector::DataPointDefinition::SUM, and jsoncollector::FileIO::writeStringToFile().

   {
     std::string baseRunDir = edm::Service<evf::EvFDaqDirector>()->baseRunDir();
     readAdler32Check_ =  edm::Service<evf::EvFDaqDirector>()->outputAdler32Recheck();
     LogDebug("RecoEventOutputModuleForFU") << "writing .dat files to -: " << baseRunDir;
     // create open dir if not already there
     edm::Service<evf::EvFDaqDirector>()->createRunOpendirMaybe();
 
     processed_.setName("Processed");
     accepted_.setName("Accepted");
     errorEvents_.setName("ErrorEvents");
     retCodeMask_.setName("ReturnCodeMask");
     filelist_.setName("Filelist");
     filesize_.setName("Filesize");
     inputFiles_.setName("InputFiles");
     fileAdler32_.setName("FileAdler32");
     transferDestination_.setName("TransferDestination");
     mergeType_.setName("MergeType");
     hltErrorEvents_.setName("HLTErrorEvents");
 
     outJsonDef_.setDefaultGroup("data");
     outJsonDef_.addLegendItem("Processed","integer",jsoncollector::DataPointDefinition::SUM);
     outJsonDef_.addLegendItem("Accepted","integer",jsoncollector::DataPointDefinition::SUM);
     outJsonDef_.addLegendItem("ErrorEvents","integer",jsoncollector::DataPointDefinition::SUM);
     outJsonDef_.addLegendItem("ReturnCodeMask","integer",jsoncollector::DataPointDefinition::BINARYOR);
     outJsonDef_.addLegendItem("Filelist","string",jsoncollector::DataPointDefinition::MERGE);
     outJsonDef_.addLegendItem("Filesize","integer",jsoncollector::DataPointDefinition::SUM);
     outJsonDef_.addLegendItem("InputFiles","string",jsoncollector::DataPointDefinition::CAT);
     outJsonDef_.addLegendItem("FileAdler32","integer",jsoncollector::DataPointDefinition::ADLER32);
     outJsonDef_.addLegendItem("TransferDestination","string",jsoncollector::DataPointDefinition::SAME);
     outJsonDef_.addLegendItem("MergeType","string",jsoncollector::DataPointDefinition::SAME);
     outJsonDef_.addLegendItem("HLTErrorEvents","integer",jsoncollector::DataPointDefinition::SUM);
     std::stringstream tmpss,ss;
     tmpss << baseRunDir << "/open/" << "output_" << getpid() << ".jsd";
     ss << baseRunDir << "/" << "output_" << getpid() << ".jsd";
     std::string outTmpJsonDefName = tmpss.str();
     std::string outJsonDefName = ss.str();
 
     edm::Service<evf::EvFDaqDirector>()->lockInitLock();
     struct stat   fstat;
     if (stat (outJsonDefName.c_str(), &fstat) != 0) { //file does not exist
       LogDebug("RecoEventOutputModuleForFU") << "writing output definition file -: " << outJsonDefName;
       std::string content;
       jsoncollector::JSONSerializer::serialize(&outJsonDef_,content);
       jsoncollector::FileIO::writeStringToFile(outTmpJsonDefName, content);
       boost::filesystem::rename(outTmpJsonDefName,outJsonDefName);
     }
     edm::Service<evf::EvFDaqDirector>()->unlockInitLock();
 
     jsonMonitor_.reset(new jsoncollector::FastMonitor(&outJsonDef_,true));
     jsonMonitor_->setDefPath(outJsonDefName);
     jsonMonitor_->registerGlobalMonitorable(&processed_,false);
     jsonMonitor_->registerGlobalMonitorable(&accepted_,false);
     jsonMonitor_->registerGlobalMonitorable(&errorEvents_,false);
     jsonMonitor_->registerGlobalMonitorable(&retCodeMask_,false);
     jsonMonitor_->registerGlobalMonitorable(&filelist_,false);
     jsonMonitor_->registerGlobalMonitorable(&filesize_,false);
     jsonMonitor_->registerGlobalMonitorable(&inputFiles_,false);
     jsonMonitor_->registerGlobalMonitorable(&fileAdler32_,false);
     jsonMonitor_->registerGlobalMonitorable(&transferDestination_,false);
     jsonMonitor_->registerGlobalMonitorable(&mergeType_,false);
     jsonMonitor_->registerGlobalMonitorable(&hltErrorEvents_,false);
     jsonMonitor_->commit(nullptr);
 
   }

template<typename Consumer >

void evf::RecoEventOutputModuleForFU< Consumer >::start ( void )

overrideprivatevirtual

Implements edm::StreamerOutputModuleBase.

Definition at line 187 of file RecoEventOutputModuleForFU.h.

References AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by progressbar.ProgressBar::__next__(), Types.LuminosityBlockRange::cppID(), and Types.EventRange::cppID().

   {
     initRun();
     const std::string openInitFileName = edm::Service<evf::EvFDaqDirector>()->getOpenInitFilePath(stream_label_);
     edm::LogInfo("RecoEventOutputModuleForFU") << "start() method, initializing streams. init stream -: "  
                                            << openInitFileName;
     c_->setInitMessageFile(openInitFileName);
     c_->start();
     
   }