#include <StreamerOutputModuleCommon.h>

Inheritance diagram for edm::StreamerOutputModuleCommon:

Public Member Functions
SerializeDataBuffer *	getSerializerBuffer ()

std::unique_ptr< EventMsgBuilder >	serializeEvent (SerializeDataBuffer &sbuf, EventForOutput const &e, Handle< TriggerResults > const &triggerResults, ParameterSetID const &selectorCfg)

std::unique_ptr< InitMsgBuilder >	serializeRegistry (SerializeDataBuffer &sbuf, BranchIDLists const &branchLists, ThinnedAssociationsHelper const &helper, std::string const &processName, std::string const &moduleLabel, ParameterSetID const &toplevel, SendJobHeader::ParameterSetMap const *psetMap)

	StreamerOutputModuleCommon (ParameterSet const &ps, SelectedProducts const *selections, std::string const &moduleLabel)

	~StreamerOutputModuleCommon ()

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

Protected Attributes
std::unique_ptr < SerializeDataBuffer >	serializerBuffer_

Private Member Functions
void	setHltMask (EventForOutput const &e, Handle< TriggerResults > const &triggerResults, std::vector< unsigned char > &hltbits) const

Private Attributes
StreamerCompressionAlgo	compressionAlgo_

std::string	compressionAlgoStr_

int	compressionLevel_

unsigned int	hltsize_

Strings	hltTriggerSelections_

char	host_name_ [255]

int	lumiSectionInterval_

int	maxEventSize_

uint32	outputModuleId_

StreamSerializer	serializer_

double	timeInSecSinceUTC

bool	useCompression_

Detailed Description

Definition at line 21 of file StreamerOutputModuleCommon.h.

Constructor & Destructor Documentation

edm::StreamerOutputModuleCommon::StreamerOutputModuleCommon	(	ParameterSet const &	ps,
		SelectedProducts const *	selections,
		std::string const &	moduleLabel
	)

explicit

Definition at line 26 of file StreamerOutputModuleCommon.cc.

References visDQMUpload::buf, compressionAlgo_, compressionAlgoStr_, compressionLevel_, Exception, FDEBUG, edm::getAllTriggerNames(), edm::EventSelector::getEventSelectionVString(), hltsize_, hltTriggerSelections_, host_name_, dttmaxenums::L, edm::LZMA, outputModuleId_, timeInSecSinceUTC, edm::UNCOMPRESSED, useCompression_, edm::ZLIB, and edm::ZSTD.

       :
 
         serializer_(selections),
         useCompression_(ps.getUntrackedParameter<bool>("use_compression")),
         compressionAlgoStr_(ps.getUntrackedParameter<std::string>("compression_algorithm")),
         compressionLevel_(ps.getUntrackedParameter<int>("compression_level")),
         lumiSectionInterval_(ps.getUntrackedParameter<int>("lumiSection_interval")),
         hltsize_(0),
         host_name_(),
         hltTriggerSelections_(),
         outputModuleId_(0) {
     //limits initially set for default ZLIB option
     int minCompressionLevel = 1;
     int maxCompressionLevel = 9;
 
     // 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 (compressionAlgoStr_ == "ZLIB") {
         compressionAlgo_ = ZLIB;
       } else if (compressionAlgoStr_ == "LZMA") {
         compressionAlgo_ = LZMA;
         minCompressionLevel = 0;
       } else if (compressionAlgoStr_ == "ZSTD") {
         compressionAlgo_ = ZSTD;
         maxCompressionLevel = 20;
       } else if (compressionAlgoStr_ == "UNCOMPRESSED") {
         compressionLevel_ = 0;
         useCompression_ = false;
         compressionAlgo_ = UNCOMPRESSED;
       } else
         throw cms::Exception("StreamerOutputModuleCommon", "Compression type unknown")
             << "Unknown compression algorithm " << compressionAlgoStr_;
 
       if (compressionLevel_ < minCompressionLevel) {
         FDEBUG(9) << "Compression Level = " << compressionLevel_ << " no compression" << std::endl;
         compressionLevel_ = 0;
         useCompression_ = false;
         compressionAlgo_ = UNCOMPRESSED;
       } else if (compressionLevel_ > maxCompressionLevel) {
         FDEBUG(9) << "Compression Level = " << compressionLevel_ << " using max compression level "
                   << maxCompressionLevel << std::endl;
         compressionLevel_ = maxCompressionLevel;
         compressionAlgo_ = UNCOMPRESSED;
       }
     } else
       compressionAlgo_ = UNCOMPRESSED;
 
     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);
 
     Strings const& hltTriggerNames = edm::getAllTriggerNames();
     hltsize_ = hltTriggerNames.size();
 
     //Checksum of the module label
     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);
   }

edm::StreamerOutputModuleCommon::~StreamerOutputModuleCommon ( )

Definition at line 100 of file StreamerOutputModuleCommon.cc.

100 {}

Member Function Documentation

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

static

Definition at line 275 of file StreamerOutputModuleCommon.cc.

References edm::ParameterSetDescription::addUntracked(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by edm::StreamerOutputModuleBase::fillDescription(), evf::EvFOutputModule::fillDescriptions(), and evf::GlobalEvFOutputModule::fillDescriptions().

                                                                                 {
     desc.addUntracked<int>("max_event_size", 7000000)->setComment("Obsolete parameter.");
     desc.addUntracked<bool>("use_compression", true)
         ->setComment("If True, compression will be used to write streamer file.");
     desc.addUntracked<std::string>("compression_algorithm", "ZLIB")
         ->setComment("Compression algorithm to use: UNCOMPRESSED, ZLIB, LZMA or ZSTD");
     desc.addUntracked<int>("compression_level", 1)->setComment("Compression level to use on serialized ROOT events");
     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.");
   }

SerializeDataBuffer * edm::StreamerOutputModuleCommon::getSerializerBuffer ( )

Definition at line 288 of file StreamerOutputModuleCommon.cc.

References serializerBuffer_.

Referenced by edm::StreamerOutputModuleBase::beginRun(), and edm::StreamerOutputModuleBase::write().

                                                                        {
     auto* ptr = serializerBuffer_.get();
     if (!ptr) {
       serializerBuffer_ = std::make_unique<SerializeDataBuffer>();
       ptr = serializerBuffer_.get();
     }
     return ptr;
   }

std::unique_ptr< EventMsgBuilder > edm::StreamerOutputModuleCommon::serializeEvent	(	SerializeDataBuffer &	sbuf,
		EventForOutput const &	e,
		Handle< TriggerResults > const &	triggerResults,
		ParameterSetID const &	selectorCfg
	)

Definition at line 204 of file StreamerOutputModuleCommon.cc.

References SerializeDataBuffer::adler32_chksum(), SerializeDataBuffer::bufferPointer(), SerializeDataBuffer::comp_buf_, compressionAlgo_, compressionLevel_, filterCSVwithJSON::copy, SerializeDataBuffer::currentEventSize(), SerializeDataBuffer::currentSpaceUsed(), edm::EventID::event(), Exception, SerializeDataBuffer::header_buf_, hltsize_, host_name_, edm::EventForOutput::id(), dqmdumpme::lumi, edm::EventForOutput::luminosityBlock(), lumiSectionInterval_, mps_check::msg, outputModuleId_, SerializeDataBuffer::reserve_size, edm::EventID::run(), edm::StreamSerializer::serializeEvent(), serializer_, setHltMask(), timeInSecSinceUTC, and useCompression_.

Referenced by edm::StreamerOutputModuleBase::write().

                                          {
     constexpr unsigned int reserve_size = SerializeDataBuffer::reserve_size;
     //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.
     std::vector<bool> l1bit = {true, true, false};
     //End of dummy data
 
     std::vector<unsigned char> hltbits;
     setHltMask(e, triggerResults, hltbits);
 
     uint32 lumi;
     if (lumiSectionInterval_ == 0) {
       lumi = e.luminosityBlock();
     } else {
       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;
     }
 
     serializer_.serializeEvent(sbuf, e, selectorCfg, compressionAlgo_, compressionLevel_, reserve_size);
 
     // resize header_buf_ to reserved size on first written event
     if (sbuf.header_buf_.size() < reserve_size)
       sbuf.header_buf_.resize(reserve_size);
 
     auto msg = std::make_unique<EventMsgBuilder>(&sbuf.header_buf_[0],
                                                  sbuf.comp_buf_.size(),
                                                  e.id().run(),
                                                  e.id().event(),
                                                  lumi,
                                                  outputModuleId_,
                                                  0,
                                                  l1bit,
                                                  (uint8*)&hltbits[0],
                                                  hltsize_,
                                                  (uint32)sbuf.adler32_chksum(),
                                                  host_name_);
 
     // 50000 bytes is reserved for header as has been the case with previous version which did one extra copy of event data
     uint32 headerSize = msg->headerSize();
     if (headerSize > reserve_size)
       throw cms::Exception("StreamerOutputModuleCommon", "Header Overflow")
           << " header of size " << headerSize << "bytes is too big to fit into the reserved buffer space";
 
     //set addresses to other buffer and copy constructed header there
     msg->setBufAddr(&sbuf.comp_buf_[reserve_size - headerSize]);
     msg->setEventAddr(sbuf.bufferPointer());
     std::copy(&sbuf.header_buf_[0], &sbuf.header_buf_[headerSize], (char*)(&sbuf.comp_buf_[reserve_size - headerSize]));
 
     unsigned int src_size = sbuf.currentSpaceUsed();
     msg->setEventLength(src_size);  //compressed size
     if (useCompression_)
       msg->setOrigDataSize(
           sbuf.currentEventSize());  //uncompressed size (or 0 if no compression -> streamer input source requires this)
     else
       msg->setOrigDataSize(0);
 
     return msg;
   }

std::unique_ptr< InitMsgBuilder > edm::StreamerOutputModuleCommon::serializeRegistry	(	SerializeDataBuffer &	sbuf,
		BranchIDLists const &	branchLists,
		ThinnedAssociationsHelper const &	helper,
		std::string const &	processName,
		std::string const &	moduleLabel,
		ParameterSetID const &	toplevel,
		SendJobHeader::ParameterSetMap const *	psetMap
	)

Definition at line 102 of file StreamerOutputModuleCommon.cc.

References SerializeDataBuffer::adler32_chksum(), SerializeDataBuffer::bufferPointer(), edm::Hash< I >::compactForm(), filterCSVwithJSON::copy, SerializeDataBuffer::currentSpaceUsed(), edm::getAllTriggerNames(), edm::getReleaseVersion(), SerializeDataBuffer::header_buf_, hltTriggerSelections_, outputModuleId_, submitPVValidationJobs::run, serializer_, edm::StreamSerializer::serializeRegistry(), and alcazmumu_cfi::src.

Referenced by edm::StreamerOutputModuleBase::beginRun().

                                                    {
     if (psetMap) {
       serializer_.serializeRegistry(sbuf, branchLists, helper, *psetMap);
     } else {
       serializer_.serializeRegistry(sbuf, branchLists, helper);
     }
     // resize header_buf_ to reflect space used in serializer_ + header
     // I just added an overhead for header of 50000 for now
     unsigned int src_size = sbuf.currentSpaceUsed();
     unsigned int new_size = src_size + 50000;
     if (sbuf.header_buf_.size() < new_size)
       sbuf.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
 
     //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;
 
     //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");
 
     Strings const& hltTriggerNames = edm::getAllTriggerNames();
 
     auto init_message = std::make_unique<InitMsgBuilder>(&sbuf.header_buf_[0],
                                                          sbuf.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)sbuf.adler32_chksum());
 
     // copy data into the destination message
     unsigned char* src = sbuf.bufferPointer();
     std::copy(src, src + src_size, init_message->dataAddress());
     init_message->setDataLength(src_size);
     return init_message;
   }

void edm::StreamerOutputModuleCommon::setHltMask	(	EventForOutput const &	e,
		Handle< TriggerResults > const &	triggerResults,
		std::vector< unsigned char > &	hltbits
	)		const

private

Definition at line 163 of file StreamerOutputModuleCommon.cc.

References PVValHelper::fill(), hltsize_, mps_fire::i, edm::HandleBase::isValid(), and edm::hlt::Pass.

Referenced by serializeEvent().

                                                                                        {
     hltbits.clear();
 
     std::vector<unsigned char> vHltState;
 
     if (triggerResults.isValid()) {
       for (std::vector<unsigned char>::size_type i = 0; i != hltsize_; ++i) {
         vHltState.push_back(((triggerResults->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
     if (!vHltState.empty()) {
       unsigned int packInOneByte = 4;
       unsigned int sizeOfPackage = 1 + ((vHltState.size() - 1) / packInOneByte);  //Two bits per HLT
 
       hltbits.resize(sizeOfPackage);
       std::fill(hltbits.begin(), hltbits.end(), 0);
 
       for (std::vector<unsigned char>::size_type i = 0; i != vHltState.size(); ++i) {
         unsigned int whichByte = i / packInOneByte;
         unsigned int indxWithinByte = i % packInOneByte;
         hltbits[whichByte] = hltbits[whichByte] | (vHltState[i] << (indxWithinByte * 2));
       }
     }
 
     //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";
   }