#include <StreamerOutputModuleBase.h>

Inheritance diagram for edm::StreamerOutputModuleBase:

Public Member Functions
	StreamerOutputModuleBase (ParameterSet const &ps)

virtual	~StreamerOutputModuleBase ()

Public Member Functions inherited from edm::OutputModule
BranchChildren const &	branchChildren () const

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

SelectionsArray const &	keptProducts () const

int	maxEvents () const

	OutputModule (ParameterSet const &pset)

std::string const &	processName () const

int	remainingEvents () const

bool	selected (BranchDescription const &desc) const

void	selectProducts ()

bool	wantAllEvents () const

virtual	~OutputModule ()

Static Public Member Functions
static void	fillDescription (ParameterSetDescription &desc)

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

static void	fillDescription (ParameterSetDescription &desc)

static void	fillDescriptions (ConfigurationDescriptions &descriptions)

Private Member Functions
virtual void	beginJob ()

virtual void	beginRun (RunPrincipal const &)

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

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

virtual void	endJob ()

virtual void	endRun (RunPrincipal const &)

std::auto_ptr< EventMsgBuilder >	serializeEvent (EventPrincipal const &e)

std::auto_ptr< InitMsgBuilder >	serializeRegistry ()

void	setHltMask (EventPrincipal const &e)

void	setLumiSection ()

virtual void	start () const =0

virtual void	stop () const =0

virtual void	write (EventPrincipal const &e)

virtual void	writeLuminosityBlock (LuminosityBlockPrincipal const &)

virtual void	writeRun (RunPrincipal const &)

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_

Selections const *	selections_

StreamSerializer	serializer_

double	timeInSecSinceUTC

bool	useCompression_

Additional Inherited Members
Public Types inherited from edm::OutputModule
typedef OutputModule	ModuleType

typedef OutputWorker	WorkerType

Protected Member Functions inherited from edm::OutputModule
CurrentProcessingContext const *	currentContext () const

ModuleDescription const &	description () const

void	doBeginJob ()

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

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

void	doEndJob ()

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

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

bool	doEvent (EventPrincipal const &ep, EventSetup const &c, CurrentProcessingContext const *cpc)

Trig	getTriggerResults (Event const &ep) const

Trig	getTriggerResults (EventPrincipal const &ep) const

ParameterSetID	selectorConfig () const

Detailed Description

Definition at line 15 of file StreamerOutputModuleBase.h.

Constructor & Destructor Documentation

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

explicit

Definition at line 56 of file StreamerOutputModuleBase.cc.

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

                                                                            :
     OutputModule(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_),
     hltsize_(0),
     lumi_(0),
     l1bit_(0),
     hltbits_(0),
     origSize_(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;
       }
     }
     serialize_databuffer.bufs_.resize(maxEventSize_);
     int got_host = gethostname(host_name_, 255);
     if(got_host != 0) strcpy(host_name_, "noHostNameFoundOrTooLong");
     //loadExtraClasses();
     // do the line below instead of loadExtraClasses() to avoid Root errors
     RootAutoLibraryLoader::enable();
 
     // 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 100 of file StreamerOutputModuleBase.cc.

100 {}

Member Function Documentation

void edm::StreamerOutputModuleBase::beginJob ( void )

privatevirtual

Reimplemented from edm::OutputModule.

Definition at line 115 of file StreamerOutputModuleBase.cc.

115 {}

void edm::StreamerOutputModuleBase::beginRun ( RunPrincipal const & )

privatevirtual

Reimplemented from edm::OutputModule.

Definition at line 103 of file StreamerOutputModuleBase.cc.

References doOutputHeader(), serializeRegistry(), and start().

                                                         {
     start();
     std::auto_ptr<InitMsgBuilder>  init_message = serializeRegistry();
     doOutputHeader(*init_message);
   }

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

privatepure virtual

Implemented in edm::StreamerOutputModule< Consumer >.

Referenced by write().

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

privatepure virtual

Implemented in edm::StreamerOutputModule< Consumer >.

Referenced by beginRun().

void edm::StreamerOutputModuleBase::endJob ( void )

privatevirtual

Reimplemented from edm::OutputModule.

Definition at line 118 of file StreamerOutputModuleBase.cc.

References stop().

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

void edm::StreamerOutputModuleBase::endRun ( RunPrincipal const & )

privatevirtual

Reimplemented from edm::OutputModule.

Definition at line 110 of file StreamerOutputModuleBase.cc.

References stop().

                                                       {
     stop();
   }

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

static

Definition at line 289 of file StreamerOutputModuleBase.cc.

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

Referenced by edm::StreamerOutputModule< 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);
   }

std::auto_ptr< EventMsgBuilder > edm::StreamerOutputModuleBase::serializeEvent ( EventPrincipal const & e )

private

Definition at line 237 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_, runTheMatrix::msg, origSize_, outputModuleId_, edm::EventID::run(), edm::OutputModule::selectorConfig(), edm::StreamSerializer::serializeEvent(), serializer_, setHltMask(), setLumiSection(), align_tpl::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);
 
     if (lumiSectionInterval_ == 0) {
       lumi_ = e.luminosityBlock();
     } else {
       setLumiSection();
     }
 
     serializer_.serializeEvent(e, selectorConfig(), useCompression_, compressionLevel_, serialize_databuffer);
 
     // resize bufs_ to reflect space used in serializer_ + header
     // I just added an overhead for header of 50000 for now
     unsigned int src_size = serialize_databuffer.currentSpaceUsed();
     unsigned int new_size = src_size + 50000;
     if(serialize_databuffer.bufs_.size() < new_size) serialize_databuffer.bufs_.resize(new_size);
 
     std::auto_ptr<EventMsgBuilder>
       msg(new EventMsgBuilder(&serialize_databuffer.bufs_[0], serialize_databuffer.bufs_.size(), e.id().run(),
                               e.id().event(), lumi_, outputModuleId_, 0,
                               l1bit_, (uint8*)&hltbits_[0], hltsize_,
                               (uint32)serialize_databuffer.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 = serialize_databuffer.bufferPointer();
     std::copy(src,src + src_size, msg->eventAddr());
     msg->setEventLength(src_size);
     if(useCompression_) msg->setOrigDataSize(serialize_databuffer.currentEventSize());
 
     l1bit_.clear();  //Clear up for the next event to come.
     return msg;
   }

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

private

Definition at line 135 of file StreamerOutputModuleBase.cc.

References SerializeDataBuffer::adler32_chksum(), SerializeDataBuffer::bufferPointer(), edm::Hash< I >::compactForm(), filterCSVwithJSON::copy, SerializeDataBuffer::currentSpaceUsed(), edm::OutputModule::description(), edm::getAllTriggerNames(), edm::pset::getProcessParameterSetID(), edm::getReleaseVersion(), SerializeDataBuffer::header_buf_, hltsize_, hltTriggerSelections_, host_name_, edm::detail::ThreadSafeRegistry< KEY, T, E >::instance(), dttmaxenums::L, moduleLabel(), edm::ModuleDescription::moduleLabel(), outputModuleId_, edm::OutputModule::processName(), DTTTrigCorrFirst::run, serializer_, edm::StreamSerializer::serializeRegistry(), align_tpl::src, and v.

Referenced by beginRun().

                                               {
 
     serializer_.serializeRegistry(serialize_databuffer);
 
     // resize bufs_ to reflect space used in serializer_ + header
     // I just added an overhead for header of 50000 for now
     unsigned int src_size = serialize_databuffer.currentSpaceUsed();
     unsigned int new_size = src_size + 50000;
     if(serialize_databuffer.header_buf_.size() < new_size) serialize_databuffer.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
     pset::Registry* reg = pset::Registry::instance();
     ParameterSetID toplevel = pset::getProcessParameterSetID(reg);
 
     //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;
 
     //Setting protocol version V
     Version v(8,(uint8*)toplevel.compactForm().c_str());
 
     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* buf = (Bytef*) moduleLabel.data();
     crc = crc32(crc, buf, moduleLabel.length());
     outputModuleId_ = static_cast<uint32>(crc);
 
     std::auto_ptr<InitMsgBuilder> init_message(
         new InitMsgBuilder(&serialize_databuffer.header_buf_[0], serialize_databuffer.header_buf_.size(),
                            run, v, getReleaseVersion().c_str() , processName.c_str(),
                            moduleLabel.c_str(), outputModuleId_,
                            hltTriggerNames, hltTriggerSelections_, l1_names,
                            (uint32)serialize_databuffer.adler32_chksum(), host_name_));
 
     // copy data into the destination message
     unsigned char* src = serialize_databuffer.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 )

private

Definition at line 196 of file StreamerOutputModuleBase.cc.

References edm::OutputModule::getTriggerResults(), hltbits_, hltsize_, i, edm::HandleBase::isValid(), edm::hlt::Pass, parseEventContent::prod, and evf::utils::state.

Referenced by serializeEvent().

                                                               {
 
     hltbits_.clear();  // If there was something left over from last event
 
     Handle<TriggerResults> const& prod = getTriggerResults(e);
     //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 227 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 edm::StreamerOutputModule< Consumer >.

Referenced by beginRun().

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

privatepure virtual

Implemented in edm::StreamerOutputModule< Consumer >.

Referenced by endJob(), and endRun().

void edm::StreamerOutputModuleBase::write ( EventPrincipal const & e )

privatevirtual

Implements edm::OutputModule.

Definition at line 129 of file StreamerOutputModuleBase.cc.

References doOutputEvent(), runTheMatrix::msg, and serializeEvent().

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