#include <StreamerOutputModuleBase.h>

Inheritance diagram for edm::StreamerOutputModuleBase:

Public Member Functions
	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 (const std::string &iProcessName, std::vector< const char * > &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	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	beginRun (RunPrincipal const &, ModuleCallingContext const *) override

virtual void	doOutputEvent (EventMsgBuilder const &msg) const =0

virtual void	doOutputHeader (InitMsgBuilder const &init_message) const =0

virtual void	endJob () override

virtual void	endRun (RunPrincipal const &, ModuleCallingContext const *) override

Trig	getTriggerResults (EDGetTokenT< TriggerResults > const &token, EventPrincipal const &ep, ModuleCallingContext const *) const

std::auto_ptr< EventMsgBuilder >	serializeEvent (EventPrincipal const &e, ModuleCallingContext const *mcc)

std::auto_ptr< InitMsgBuilder >	serializeRegistry ()

void	setHltMask (EventPrincipal const &e, ModuleCallingContext const *)

void	setLumiSection ()

virtual void	start () const =0

virtual void	stop () const =0

virtual void	write (EventPrincipal const &e, ModuleCallingContext const *) override

virtual void	writeLuminosityBlock (LuminosityBlockPrincipal const &, ModuleCallingContext const *) override

virtual void	writeRun (RunPrincipal const &, ModuleCallingContext const *) override

Private Attributes
int	compressionLevel_

std::vector< unsigned char >	hltbits_

unsigned int	hltsize_

Strings	hltTriggerSelections_

char	host_name_ [255]

std::vector< bool >	l1bit_

uint32	lumi_

int	lumiSectionInterval_

int	maxEventSize_

uint32	origSize_

uint32	outputModuleId_

SelectedProducts const *	selections_

SerializeDataBuffer	serializeDataBuffer_

StreamSerializer	serializer_

double	timeInSecSinceUTC

edm::EDGetTokenT < edm::TriggerResults >	trToken_

bool	useCompression_

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

Definition at line 19 of file StreamerOutputModuleBase.h.

Constructor & Destructor Documentation

edm::StreamerOutputModuleBase::StreamerOutputModuleBase ( ParameterSet const & ps )

explicit

Definition at line 58 of file StreamerOutputModuleBase.cc.

References SerializeDataBuffer::bufs_, compressionLevel_, FDEBUG, edm::EventSelector::getEventSelectionVString(), hltTriggerSelections_, host_name_, maxEventSize_, serializeDataBuffer_, timeInSecSinceUTC, and useCompression_.

                                                                            :
     one::OutputModuleBase::OutputModuleBase(ps),
     one::OutputModule<one::WatchRuns, one::WatchLuminosityBlocks>(ps),
     selections_(&keptProducts()[InEvent]),
     maxEventSize_(ps.getUntrackedParameter<int>("max_event_size")),
     useCompression_(ps.getUntrackedParameter<bool>("use_compression")),
     compressionLevel_(ps.getUntrackedParameter<int>("compression_level")),
     lumiSectionInterval_(ps.getUntrackedParameter<int>("lumiSection_interval")),
     serializer_(selections_),
     serializeDataBuffer_(),
     hltsize_(0),
     lumi_(0),
     l1bit_(0),
     hltbits_(0),
     origSize_(0),
     host_name_(),
     trToken_(consumes<edm::TriggerResults>(edm::InputTag("TriggerResults"))),
     hltTriggerSelections_(),
     outputModuleId_(0) {
     // no compression as default value - we need this!
 
     // test luminosity sections
     struct timeval now;
     struct timezone dummyTZ;
     gettimeofday(&now, &dummyTZ);
     timeInSecSinceUTC = static_cast<double>(now.tv_sec) + (static_cast<double>(now.tv_usec)/1000000.0);
 
     if(useCompression_ == true) {
       if(compressionLevel_ <= 0) {
         FDEBUG(9) << "Compression Level = " << compressionLevel_
                   << " no compression" << std::endl;
         compressionLevel_ = 0;
         useCompression_ = false;
       } else if(compressionLevel_ > 9) {
         FDEBUG(9) << "Compression Level = " << compressionLevel_
                   << " using max compression level 9" << std::endl;
         compressionLevel_ = 9;
       }
     }
     serializeDataBuffer_.bufs_.resize(maxEventSize_);
     int got_host = gethostname(host_name_, 255);
     if(got_host != 0) strncpy(host_name_, "noHostNameFoundOrTooLong", sizeof(host_name_));
     //loadExtraClasses();
 
     // 25-Jan-2008, KAB - pull out the trigger selection request
     // which we need for the INIT message
     hltTriggerSelections_ = EventSelector::getEventSelectionVString(ps);
   }

edm::StreamerOutputModuleBase::~StreamerOutputModuleBase ( )

virtual

Definition at line 107 of file StreamerOutputModuleBase.cc.

107 {}

Member Function Documentation

void edm::StreamerOutputModuleBase::beginJob ( void )

overrideprivatevirtual

Reimplemented from edm::one::OutputModuleBase.

Reimplemented in evf::RecoEventOutputModuleForFU< Consumer >.

Definition at line 124 of file StreamerOutputModuleBase.cc.

124 {}

void edm::StreamerOutputModuleBase::beginRun	(	RunPrincipal const &	,
		ModuleCallingContext const *
	)

overrideprivatevirtual

Definition at line 110 of file StreamerOutputModuleBase.cc.

References doOutputHeader(), SerializeDataBuffer::header_buf_, serializeDataBuffer_, serializeRegistry(), and start().

                                                                                      {
     start();
     std::auto_ptr<InitMsgBuilder>  init_message = serializeRegistry();
     doOutputHeader(*init_message);
     serializeDataBuffer_.header_buf_.clear();
     serializeDataBuffer_.header_buf_.shrink_to_fit();
   }

virtual void edm::StreamerOutputModuleBase::doOutputEvent ( EventMsgBuilder const & msg ) const

privatepure virtual

Implemented in evf::RecoEventOutputModuleForFU< Consumer >, and edm::StreamerOutputModule< Consumer >.

Referenced by write().

virtual void edm::StreamerOutputModuleBase::doOutputHeader ( InitMsgBuilder const & init_message ) const

privatepure virtual

Implemented in evf::RecoEventOutputModuleForFU< Consumer >, and edm::StreamerOutputModule< Consumer >.

Referenced by beginRun().

void edm::StreamerOutputModuleBase::endJob ( void )

overrideprivatevirtual

Reimplemented from edm::one::OutputModuleBase.

Definition at line 127 of file StreamerOutputModuleBase.cc.

References stop().

                                    {
     stop();  // for closing of files, notify storage manager, etc.
   }

void edm::StreamerOutputModuleBase::endRun	(	RunPrincipal const &	,
		ModuleCallingContext const *
	)

overrideprivatevirtual

Definition at line 119 of file StreamerOutputModuleBase.cc.

References stop().

                                                                                    {
     stop();
   }

void edm::StreamerOutputModuleBase::fillDescription ( ParameterSetDescription & desc )

static

Definition at line 306 of file StreamerOutputModuleBase.cc.

References edm::ParameterSetDescription::addUntracked(), and edm::OutputModule::fillDescription().

Referenced by edm::StreamerOutputModule< Consumer >::fillDescriptions(), and evf::RecoEventOutputModuleForFU< Consumer >::fillDescriptions().

                                                                          {
     desc.addUntracked<int>("max_event_size", 7000000)
         ->setComment("Starting size in bytes of the serialized event buffer.");
     desc.addUntracked<bool>("use_compression", true)
         ->setComment("If True, compression will be used to write streamer file.");
     desc.addUntracked<int>("compression_level", 1)
         ->setComment("ROOT compression level to use.");
     desc.addUntracked<int>("lumiSection_interval", 0)
         ->setComment("If 0, use lumi section number from event.\n"
                      "If not 0, the interval in seconds between fake lumi sections.");
     OutputModule::fillDescription(desc);
   }

Trig edm::StreamerOutputModuleBase::getTriggerResults	(	EDGetTokenT< TriggerResults > const &	token,
		EventPrincipal const &	ep,
		ModuleCallingContext const *	mcc
	)		const

private

Definition at line 202 of file StreamerOutputModuleBase.cc.

References edm::convert_handle(), edm::PrincipalGetAdapter::getByToken_(), edm::one::OutputModuleBase::moduleDescription(), eostools::move(), edm::PRODUCT_TYPE, query::result, and edm::PrincipalGetAdapter::setConsumer().

Referenced by setHltMask().

                                                                                                                                                        {
     //This cast is safe since we only call const functions of the EventPrincipal after this point
     PrincipalGetAdapter adapter(const_cast<EventPrincipal&>(ep), moduleDescription());
     adapter.setConsumer(this);
     Trig result;
     auto bh = adapter.getByToken_(TypeID(typeid(TriggerResults)),PRODUCT_TYPE, token, mcc);
     convert_handle(std::move(bh), result);
     return result;
   }

std::auto_ptr< EventMsgBuilder > edm::StreamerOutputModuleBase::serializeEvent	(	EventPrincipal const &	e,
		ModuleCallingContext const *	mcc
	)

private

Definition at line 254 of file StreamerOutputModuleBase.cc.

References SerializeDataBuffer::adler32_chksum(), SerializeDataBuffer::bufferPointer(), SerializeDataBuffer::bufs_, compressionLevel_, filterCSVwithJSON::copy, SerializeDataBuffer::currentEventSize(), SerializeDataBuffer::currentSpaceUsed(), edm::EventID::event(), hltbits_, hltsize_, host_name_, edm::EventPrincipal::id(), l1bit_, lumi_, edm::EventPrincipal::luminosityBlock(), lumiSectionInterval_, visualization-live-secondInstance_cfg::msg, origSize_, outputModuleId_, edm::EventID::run(), edm::one::OutputModuleBase::selectorConfig(), serializeDataBuffer_, edm::StreamSerializer::serializeEvent(), serializer_, setHltMask(), setLumiSection(), alcazmumu_cfi::src, and useCompression_.

Referenced by write().

                                                                                                    {
     //Lets Build the Event Message first
 
     //Following is strictly DUMMY Data for L! Trig and will be replaced with actual
     // once figured out, there is no logic involved here.
     l1bit_.push_back(true);
     l1bit_.push_back(true);
     l1bit_.push_back(false);
     //End of dummy data
 
     setHltMask(e, mcc);
 
     if (lumiSectionInterval_ == 0) {
       lumi_ = e.luminosityBlock();
     } else {
       setLumiSection();
     }
 
     serializer_.serializeEvent(e, selectorConfig(), useCompression_, compressionLevel_, serializeDataBuffer_, mcc);
 
     // resize bufs_ to reflect space used in serializer_ + header
     // I just added an overhead for header of 50000 for now
     unsigned int src_size = serializeDataBuffer_.currentSpaceUsed();
     unsigned int new_size = src_size + 50000;
     if(serializeDataBuffer_.bufs_.size() < new_size) serializeDataBuffer_.bufs_.resize(new_size);
 
     std::auto_ptr<EventMsgBuilder>
       msg(new EventMsgBuilder(&serializeDataBuffer_.bufs_[0], serializeDataBuffer_.bufs_.size(), e.id().run(),
                               e.id().event(), lumi_, outputModuleId_, 0,
                               l1bit_, (uint8*)&hltbits_[0], hltsize_,
                               (uint32)serializeDataBuffer_.adler32_chksum(), host_name_) );
     msg->setOrigDataSize(origSize_); // we need this set to zero
 
     // copy data into the destination message
     // an alternative is to have serializer only to the serialization
     // in serializeEvent, and then call a new member "getEventData" that
     // takes the compression arguments and a place to put the data.
     // This will require one less copy.  The only catch is that the
     // space provided in bufs_ should be at least the uncompressed
     // size + overhead for header because we will not know the actual
     // compressed size.
 
     unsigned char* src = serializeDataBuffer_.bufferPointer();
     std::copy(src,src + src_size, msg->eventAddr());
     msg->setEventLength(src_size);
     if(useCompression_) msg->setOrigDataSize(serializeDataBuffer_.currentEventSize());
 
     l1bit_.clear();  //Clear up for the next event to come.
     return msg;
   }

std::auto_ptr< InitMsgBuilder > edm::StreamerOutputModuleBase::serializeRegistry ( )

private

Definition at line 144 of file StreamerOutputModuleBase.cc.

References SerializeDataBuffer::adler32_chksum(), edm::one::OutputModuleBase::branchIDLists(), SerializeDataBuffer::bufferPointer(), edm::Hash< I >::compactForm(), filterCSVwithJSON::copy, SerializeDataBuffer::currentSpaceUsed(), edm::one::OutputModuleBase::description(), edm::getAllTriggerNames(), edm::getReleaseVersion(), SerializeDataBuffer::header_buf_, hltsize_, hltTriggerSelections_, dttmaxenums::L, edm::ModuleDescription::mainParameterSetID(), edm::one::OutputModuleBase::moduleDescription(), edm::ModuleDescription::moduleLabel(), outputModuleId_, edm::OutputModule::processName(), edm::one::OutputModuleBase::processName(), DTTTrigCorrFirst::run, serializeDataBuffer_, serializer_, edm::StreamSerializer::serializeRegistry(), alcazmumu_cfi::src, AlCaHLTBitMon_QueryRunRegistry::string, and edm::one::OutputModuleBase::thinnedAssociationsHelper().

Referenced by beginRun().

                                               {
 
     serializer_.serializeRegistry(serializeDataBuffer_, *branchIDLists(), *thinnedAssociationsHelper());
 
     // resize bufs_ to reflect space used in serializer_ + header
     // I just added an overhead for header of 50000 for now
     unsigned int src_size = serializeDataBuffer_.currentSpaceUsed();
     unsigned int new_size = src_size + 50000;
     if(serializeDataBuffer_.header_buf_.size() < new_size) serializeDataBuffer_.header_buf_.resize(new_size);
 
     //Build the INIT Message
     //Following values are strictly DUMMY and will be replaced
     // once available with Utility function etc.
     uint32 run = 1;
 
     //Get the Process PSet ID
     ParameterSetID toplevel = moduleDescription().mainParameterSetID();
 
     //In case we need to print it
     //  cms::Digest dig(toplevel.compactForm());
     //  cms::MD5Result r1 = dig.digest();
     //  std::string hexy = r1.toString();
     //  std::cout << "HEX Representation of Process PSetID: " << hexy << std::endl;
 
     Strings hltTriggerNames = getAllTriggerNames();
     hltsize_ = hltTriggerNames.size();
 
     //L1 stays dummy as of today
     Strings l1_names;  //3
     l1_names.push_back("t1");
     l1_names.push_back("t10");
     l1_names.push_back("t2");
 
     //Setting the process name to HLT
     std::string processName = OutputModule::processName();
 
     std::string moduleLabel = description().moduleLabel();
     uLong crc = crc32(0L, Z_NULL, 0);
     Bytef const* buf = (Bytef const*)(moduleLabel.data());
     crc = crc32(crc, buf, moduleLabel.length());
     outputModuleId_ = static_cast<uint32>(crc);
 
     std::auto_ptr<InitMsgBuilder> init_message(
         new InitMsgBuilder(&serializeDataBuffer_.header_buf_[0], serializeDataBuffer_.header_buf_.size(),
                            run, Version((uint8 const*)toplevel.compactForm().c_str()),
                            getReleaseVersion().c_str() , processName.c_str(),
                            moduleLabel.c_str(), outputModuleId_,
                            hltTriggerNames, hltTriggerSelections_, l1_names,
                            (uint32)serializeDataBuffer_.adler32_chksum()));
 
     // copy data into the destination message
     unsigned char* src = serializeDataBuffer_.bufferPointer();
     std::copy(src, src + src_size, init_message->dataAddress());
     init_message->setDataLength(src_size);
     return init_message;
   }

void edm::StreamerOutputModuleBase::setHltMask	(	EventPrincipal const &	e,
		ModuleCallingContext const *	mcc
	)

private

Definition at line 213 of file StreamerOutputModuleBase.cc.

References getTriggerResults(), hltbits_, hltsize_, i, edm::HandleBase::isValid(), edm::hlt::Pass, parseEventContent::prod, and trToken_.

Referenced by serializeEvent().

                                                                                                {
 
     hltbits_.clear();  // If there was something left over from last event
 
     Handle<TriggerResults> const& prod = getTriggerResults(trToken_, e, mcc);
     //Trig const& prod = getTrigMask(e);
     std::vector<unsigned char> vHltState;
 
     if (prod.isValid()) {
       for(std::vector<unsigned char>::size_type i=0; i != hltsize_ ; ++i) {
         vHltState.push_back(((prod->at(i)).state()));
       }
     } else {
      // We fill all Trigger bits to valid state.
      for(std::vector<unsigned char>::size_type i=0; i != hltsize_ ; ++i) {
            vHltState.push_back(hlt::Pass);
       }
     }
     //Pack into member hltbits_
     packIntoString(vHltState, hltbits_);
 
     //This is Just a printing code.
     //std::cout << "Size of hltbits:" << hltbits_.size() << std::endl;
     //for(unsigned int i=0; i != hltbits_.size() ; ++i) {
     //  printBits(hltbits_[i]);
     //}
     //std::cout << "\n";
   }

void edm::StreamerOutputModuleBase::setLumiSection ( )

private

Definition at line 244 of file StreamerOutputModuleBase.cc.

References lumi_, lumiSectionInterval_, and timeInSecSinceUTC.

Referenced by serializeEvent().

                                            {
     struct timeval now;
     struct timezone dummyTZ;
     gettimeofday(&now, &dummyTZ);
     double timeInSec = static_cast<double>(now.tv_sec) + (static_cast<double>(now.tv_usec)/1000000.0) - timeInSecSinceUTC;
     // what about overflows?
     if(lumiSectionInterval_ > 0) lumi_ = static_cast<uint32>(timeInSec/lumiSectionInterval_) + 1;
   }

virtual void edm::StreamerOutputModuleBase::start ( ) const

privatepure virtual

Implemented in evf::RecoEventOutputModuleForFU< Consumer >, and edm::StreamerOutputModule< Consumer >.

Referenced by progressbar.ProgressBar::__next__(), and beginRun().

virtual void edm::StreamerOutputModuleBase::stop ( ) const

privatepure virtual

Implemented in evf::RecoEventOutputModuleForFU< Consumer >, and edm::StreamerOutputModule< Consumer >.

Referenced by endJob(), and endRun().

void edm::StreamerOutputModuleBase::write	(	EventPrincipal const &	e,
		ModuleCallingContext const *	mcc
	)

overrideprivatevirtual

Implements edm::one::OutputModuleBase.

Definition at line 138 of file StreamerOutputModuleBase.cc.

References doOutputEvent(), visualization-live-secondInstance_cfg::msg, and serializeEvent().

                                                                                           {
     std::auto_ptr<EventMsgBuilder> msg = serializeEvent(e, mcc);
     doOutputEvent(*msg); // You can't use msg in StreamerOutputModuleBase after this point
   }