#include <StreamerOutputModuleBase.h>

Inheritance diagram for edm::StreamerOutputModuleBase:

Public Member Functions
	StreamerOutputModuleBase (ParameterSet const &ps)
virtual	~StreamerOutputModuleBase ()
Static Public Member Functions
static void	fillDescription (ParameterSetDescription &desc)
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_

Detailed Description

Definition at line 14 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),
    host_name_(),
    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;
      }
    }
    serialize_databuffer.bufs_.resize(maxEventSize_);
    int got_host = gethostname(host_name_, 255);
    if(got_host != 0) strncpy(host_name_, "noHostNameFoundOrTooLong", sizeof(host_name_));
    //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 104 of file StreamerOutputModuleBase.cc.

{}

Member Function Documentation

void edm::StreamerOutputModuleBase::beginJob ( void ) [private, virtual]

Reimplemented from edm::OutputModule.

Definition at line 119 of file StreamerOutputModuleBase.cc.

{}

void edm::StreamerOutputModuleBase::beginRun ( RunPrincipal const & ) [private, virtual]

Reimplemented from edm::OutputModule.

Definition at line 107 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 [private, pure virtual]

Implemented in edm::StreamerOutputModule< Consumer >.

Referenced by write().

virtual void edm::StreamerOutputModuleBase::doOutputHeader ( InitMsgBuilder const & init_message ) const [private, pure virtual]

Implemented in edm::StreamerOutputModule< Consumer >.

Referenced by beginRun().

void edm::StreamerOutputModuleBase::endJob ( void ) [private, virtual]

Reimplemented from edm::OutputModule.

Definition at line 122 of file StreamerOutputModuleBase.cc.

References stop().

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

void edm::StreamerOutputModuleBase::endRun ( RunPrincipal const & ) [private, virtual]

Reimplemented from edm::OutputModule.

Definition at line 114 of file StreamerOutputModuleBase.cc.

References stop().

                                                      {
    stop();
  }

void edm::StreamerOutputModuleBase::fillDescription ( ParameterSetDescription & desc ) [static]

Reimplemented from edm::OutputModule.

Definition at line 291 of file StreamerOutputModuleBase.cc.

References edm::ParameterSetDescription::addUntracked().

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 239 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_, lumiQueryAPI::msg, origSize_, outputModuleId_, edm::EventID::run(), edm::OutputModule::selectorConfig(), 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);

    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 139 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_, instance, dttmaxenums::L, edm::ModuleDescription::moduleLabel(), outputModuleId_, edm::OutputModule::processName(), DTTTrigCorrFirst::run, serializer_, edm::StreamSerializer::serializeRegistry(), and alcazmumu_cfi::src.

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;

    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, Version((uint8*)toplevel.compactForm().c_str()),
                           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 198 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 229 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 [private, pure virtual]

Implemented in edm::StreamerOutputModule< Consumer >.

Referenced by beginRun().

virtual void edm::StreamerOutputModuleBase::stop ( ) const [private, pure virtual]

Implemented in edm::StreamerOutputModule< Consumer >.

Referenced by endJob(), and endRun().

void edm::StreamerOutputModuleBase::write ( EventPrincipal const & e ) [private, virtual]

Implements edm::OutputModule.

Definition at line 133 of file StreamerOutputModuleBase.cc.

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

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

void edm::StreamerOutputModuleBase::writeLuminosityBlock ( LuminosityBlockPrincipal const & ) [private, virtual]

Implements edm::OutputModule.

Definition at line 130 of file StreamerOutputModuleBase.cc.