Helper class for data store serialization. More...

#include <HLTStreamHelper.h>

Collaboration diagram for HLTStreamHelper:

[legend]

Public Member Functions
void	initialize ()
	Initialize this class. Call this e.g. in the first event.

std::unique_ptr< ZMQNoIdMessage >	stream (bool addPersistentDurability, bool streamTransientObjects)
	Stream the data store into an event message. Add ROI as additional message (if valid).

std::unique_ptr< ZMQNoIdMessage >	streamRaw ()
	Stream the data store into an event message and add SendHeader and SendTrailer around the message. Add ROI as additional message (if valid).

std::unique_ptr< ZMQNoIdMessage >	streamHistograms (bool compressed=true)
	Stream all objects derived from TH1 into a message. Only the last subfolder is streamed by prefixing the histogram names with "<subfolder>/".

void	read (std::unique_ptr< ZMQNoIdMessage > message)
	Read in a ZMQ message and rebuilt the data store from it.

void	registerStoreObjects (bool addExpressRecoObjects)
	Register all needed store objects, either only the raw data, ROIs and event meta data (for HLT) or additional objects (for express reco).

Private Member Functions
zmq::message_t	getROIMessageIfViable () const
	If the ROI payload data storobject is filled, write out the roi message (otherwise an empty message)

Private Attributes
StreamHelper	m_streamHelper
	We use the framework stream helper.

StoreObjPtr< EventMetaData >	m_eventMetaData
	Store Objects for HLT use.

StoreArray< RawSVD >	m_rawSVDs
	Store Objects for HLT use.

StoreArray< RawCDC >	m_rawCDCs
	Store Objects for HLT use.

StoreArray< RawTOP >	m_rawTOPs
	Store Objects for HLT use.

StoreArray< RawARICH >	m_rawARICHs
	Store Objects for HLT use.

StoreArray< RawECL >	m_rawECLs
	Store Objects for HLT use.

StoreArray< RawKLM >	m_rawKLMs
	Store Objects for HLT use.

StoreArray< RawTRG >	m_rawTRGs
	Store Objects for HLT use.

StoreArray< RawFTSW >	m_rawFTSWs
	Store Objects for HLT use.

StoreObjPtr< ROIpayload >	m_roiPayload
	Store Objects for HLT use.

StoreObjPtr< RandomGenerator >	m_randomGenerator
	Additional Store Objects for ExpressReco use.

StoreObjPtr< SoftwareTriggerResult >	m_softwareTriggerResult
	Additional Store Objects for ExpressReco use.

StoreObjPtr< SoftwareTrigger::SoftwareTriggerVariables >	m_softwareTriggerVariables
	Additional Store Objects for ExpressReco use.

StoreObjPtr< TRGSummary >	m_triggerSummary
	Additional Store Objects for ExpressReco use.

StoreArray< RawPXD >	m_rawPXDs
	Additional Store Objects for ExpressReco use.

StoreArray< ROIid >	m_rois
	Additional Store Objects for ExpressReco use.

std::vector< char >	m_outputBuffer
	Temporary buffer for storing the compressed result.

unsigned int	m_maximalCompressedSize = 100'000'000
	Maximal size of the compression buffer.

Detailed Description

Helper class for data store serialization.

Definition at line 42 of file HLTStreamHelper.h.

Member Function Documentation

◆ getROIMessageIfViable()

zmq::message_t getROIMessageIfViable ( ) const

private

If the ROI payload data storobject is filled, write out the roi message (otherwise an empty message)

Definition at line 137 of file HLTStreamHelper.cc.

{
  zmq::message_t roiMessage;
 
  if (m_roiPayload.isValid()) {
    const size_t length = m_roiPayload->getPacketLengthByte();
    const char* data = reinterpret_cast<const char*>(m_roiPayload->getRootdata());
 
    roiMessage = zmq::message_t(length);
    char* tbuffer = roiMessage.data<char>();
    memcpy(tbuffer, data, length);
  }
 
  return roiMessage;
}

◆ initialize()

void initialize ( )

Initialize this class. Call this e.g. in the first event.

Definition at line 58 of file HLTStreamHelper.cc.

{
  m_streamHelper.initialize(0, true);
}

◆ read()

void read ( std::unique_ptr< ZMQNoIdMessage > message )

Read in a ZMQ message and rebuilt the data store from it.

Definition at line 188 of file HLTStreamHelper.cc.

{
  if (message->isMessage(EMessageTypes::c_eventMessage)) {
    m_streamHelper.read(std::move(message));
  } else if (message->isMessage(EMessageTypes::c_rawDataMessage)) {
    int* eventBuffer = message->getMessagePart<1>().data<int>();
 
    SendHeader sndhdr;
    sndhdr.SetBuffer(eventBuffer);
    int npackedevts = sndhdr.GetNumEventsinPacket();
    if (npackedevts != 1) {
      B2WARNING("Strange SendHeader : ");
      //    for (int i = 0; i < sndhdr.SENDHDR_NWORDS; i++) {
      for (int i = 0; i < 10; i++) {
        B2WARNING(std::hex << * (sndhdr.GetBuffer() + i));
      }
 
      B2WARNING("Raw2DsModule::number of events in packet is not 1. This process gets stuck here. Please ABORT the system. (Please see discussion of daqcore channel in https://b2rc.kek.jp/ on 2017. Nov. 30. about why this is not FATAL message.");
      sleep(86400);
    }
    int ncprs = sndhdr.GetNumNodesinPacket();
    int nwords = sndhdr.GetTotalNwords() - SendHeader::SENDHDR_NWORDS - SendTrailer::SENDTRL_NWORDS;
 
    // Get buffer header
    int* bufbody = eventBuffer + SendHeader::SENDHDR_NWORDS;
 
    // Unpack buffer
    RawDataBlock tempdblk;
    tempdblk.SetBuffer(bufbody, nwords, false, npackedevts, ncprs);
 
    unsigned long long int mtime = 0;
 
    int store_time_flag = 0;
    unsigned int error_flag = 0;
 
    // Store data contents in Corresponding RawXXXX
    for (int cprid = 0; cprid < ncprs * npackedevts; cprid++) {
      // Pick up one COPPER and copy data in a temporary buffer
      int nwds_buf = tempdblk.GetBlockNwords(cprid);
      int* cprbuf = new int[nwds_buf];
      memcpy(cprbuf, tempdblk.GetBuffer(cprid), nwds_buf * 4);
 
      // Check FTSW
      if (tempdblk.CheckFTSWID(cprid)) {
        RawFTSW* ftsw = m_rawFTSWs.appendNew();
        ftsw->SetBuffer(cprbuf, nwds_buf, 1, 1, 1);
 
        // Tentative for DESY TB 2017
        unsigned int utime = (unsigned int)(ftsw->GetTTUtime(0));
        unsigned int ctime = (unsigned int)(ftsw->GetTTCtime(0));
        mtime = 1000000000 * (unsigned long long int)utime + (unsigned long long int)(std::round(ctime / 0.127216));
        store_time_flag = 1;
        continue;
      } else if (store_time_flag == 0) {
        // Tentative until RawFTSW data stream is established. 2018.5.28
        // Not store RawCOPPER here. 2018.11.23
        RawCOPPER tempcpr_time;
        tempcpr_time.SetBuffer(cprbuf, nwds_buf, false, 1, 1);
        unsigned int utime = (unsigned int)(tempcpr_time.GetTTUtime(0));
        unsigned int ctime = (unsigned int)(tempcpr_time.GetTTCtime(0));
        mtime = 1000000000 * (unsigned long long int)utime + (unsigned long long int)(std::round(ctime / 0.127216));
        store_time_flag = 1;
      }
 
      RawCOPPER tempcpr;
      tempcpr.SetBuffer(cprbuf, nwds_buf, false, 1, 1);
      int subsysid = tempcpr.GetNodeID(0);
      error_flag |= (unsigned int)(tempcpr.GetDataType(0));
 
      // Switch to each detector and register RawXXX
      if ((subsysid & DETECTOR_MASK) == CDC_ID) {
        (m_rawCDCs.appendNew())->SetBuffer(cprbuf, nwds_buf, 1, 1, 1);
      } else if ((subsysid & DETECTOR_MASK) == SVD_ID) {
        (m_rawSVDs.appendNew())->SetBuffer(cprbuf, nwds_buf, 1, 1, 1);
      } else if ((subsysid & DETECTOR_MASK) == BECL_ID) {
        (m_rawECLs.appendNew())->SetBuffer(cprbuf, nwds_buf, 1, 1, 1);
      } else if ((subsysid & DETECTOR_MASK) == EECL_ID) {
        (m_rawECLs.appendNew())->SetBuffer(cprbuf, nwds_buf, 1, 1, 1);
      } else if ((subsysid & DETECTOR_MASK) == TOP_ID) {
        (m_rawTOPs.appendNew())->SetBuffer(cprbuf, nwds_buf, 1, 1, 1);
      } else if ((subsysid & DETECTOR_MASK) == ARICH_ID) {
        (m_rawARICHs.appendNew())->SetBuffer(cprbuf, nwds_buf, 1, 1, 1);
      } else if ((subsysid & DETECTOR_MASK) == BKLM_ID) {
        (m_rawKLMs.appendNew())->SetBuffer(cprbuf, nwds_buf, 1, 1, 1);
      } else if ((subsysid & DETECTOR_MASK) == EKLM_ID) {
        (m_rawKLMs.appendNew())->SetBuffer(cprbuf, nwds_buf, 1, 1, 1);
      } else if (((subsysid & DETECTOR_MASK) & 0xF0000000) == TRGDATA_ID) {
        (m_rawTRGs.appendNew())->SetBuffer(cprbuf, nwds_buf, 1, 1, 1);
      } else {
        // Do not store Unknown RawCOPPER object. 2018.11.25
        B2WARNING("Unknown COPPER ID : ");
        for (int i = 0; i < 12; i++) {
          B2WARNING(std::hex << cprbuf[i]);
        }
        B2FATAL("Unknown COPPER ID is found. CPRID = " << std::hex << subsysid << " Please check. Exiting...");
      }
 
      // Check the magic number of the Raw(ROB) header and trailer
      if ((cprbuf[nwds_buf - (RawTrailer_latest::RAWTRAILER_NWORDS - RawTrailer_latest::POS_TERM_WORD)]
           != RawTrailer_latest::MAGIC_WORD_TERM_TRAILER) ||
          ((cprbuf[RawHeader_latest::POS_VERSION_HDRNWORDS] & RawHeader_latest::MAGIC_MASK)
           != RawHeader_latest::MAGIC_WORD)) {
        if (nwds_buf < 8) {
          std::ostringstream oss;
          for (int i = 0; i < nwds_buf; i++) {
            oss << "0x" << std::setw(8) << std::setfill('0') << std::hex << cprbuf[i] << " ";
          }
          B2ERROR("All words of the error event: " << oss.str() <<
                  "[Subsystem ID = " << subsysid << " ], [N_words = " << nwds_buf << "]");
        } else {
          std::ostringstream oss_first, oss_last;
          // Print the first 8 words
          for (int i = 0; i < 8; i++) {
            oss_first << std::setw(8) << std::setfill('0') << std::hex << cprbuf[i] << " ";
          }
          // Print the last 8 words
          for (int i = nwds_buf - 8; i < nwds_buf; i++) {
            oss_last << std::setw(8) << std::setfill('0') << std::hex << cprbuf[i] << " ";
          }
          B2ERROR("The first 8 words of the error event: " << oss_first.str() <<
                  "[Subsystem ID = " << subsysid << " ], [N_words = " << nwds_buf << "]");
          B2ERROR("The  last 8 words of the error event: " << oss_last.str()  <<
                  "[Subsystem ID = " << subsysid << " ], [N_words = " << nwds_buf << "]");
        }
        B2FATAL("Invalid magic number in Raw Trailer and/or Header. Exiting...");
      }
    }
 
    if (store_time_flag != 1) {
      B2FATAL("No time information could be extracted from Data. That should not happen. Exiting...");
    }
 
    m_eventMetaData.create();
    m_eventMetaData->setExperiment(sndhdr.GetExpNum());
    m_eventMetaData->setRun(sndhdr.GetRunNum());
    m_eventMetaData->setSubrun(sndhdr.GetSubRunNum());
    m_eventMetaData->setEvent(sndhdr.GetEventNumber());
    m_eventMetaData->setTime(mtime);
 
    if (error_flag) {
      if (error_flag & RawHeader_latest::B2LINK_PACKET_CRC_ERROR) {
        m_eventMetaData->addErrorFlag(EventMetaData::c_B2LinkPacketCRCError);
        B2WARNING("Raw2Ds: c_B2LinkPacketCRCError flag was set in EventMetaData.");
      }
      if (error_flag & RawHeader_latest::B2LINK_EVENT_CRC_ERROR) {
        m_eventMetaData->addErrorFlag(EventMetaData::c_B2LinkEventCRCError);
        B2WARNING("Raw2Ds: c_B2LinkEventCRCError flag was set in EventMetaData.");
      }
    }
  }
}

◆ registerStoreObjects()

void registerStoreObjects ( bool addExpressRecoObjects )

Register all needed store objects, either only the raw data, ROIs and event meta data (for HLT) or additional objects (for express reco).

Definition at line 63 of file HLTStreamHelper.cc.

{
  m_eventMetaData.registerInDataStore();
  m_rawSVDs.registerInDataStore();
  m_rawCDCs.registerInDataStore();
  m_rawTOPs.registerInDataStore();
  m_rawARICHs.registerInDataStore();
  m_rawECLs.registerInDataStore();
  m_rawKLMs.registerInDataStore();
  m_rawTRGs.registerInDataStore();
  m_rawFTSWs.registerInDataStore();
 
  if (addExpressRecoObjects) {
    m_randomGenerator.registerInDataStore(DataStore::c_DontWriteOut);
    m_softwareTriggerResult.registerInDataStore();
    m_softwareTriggerVariables.registerInDataStore();
    m_triggerSummary.registerInDataStore();
    m_rawPXDs.registerInDataStore();
    m_roiPayload.registerInDataStore();
    m_rois.registerInDataStore("ROIs");
  }
}

◆ stream()

std::unique_ptr< ZMQNoIdMessage > stream	(	bool	addPersistentDurability,
		bool	streamTransientObjects
	)

Stream the data store into an event message. Add ROI as additional message (if valid).

Definition at line 128 of file HLTStreamHelper.cc.

{
  const auto eventMessage = m_streamHelper.stream(addPersistentDurability, streamTransientObjects);
 
  zmq::message_t roiMessage = getROIMessageIfViable();
  auto zmqMessage = ZMQMessageFactory::createMessage(EMessageTypes::c_eventMessage, eventMessage, std::move(roiMessage));
  return zmqMessage;
}

◆ streamHistograms()

std::unique_ptr< ZMQNoIdMessage > streamHistograms ( bool compressed = true )

Stream all objects derived from TH1 into a message. Only the last subfolder is streamed by prefixing the histogram names with "<subfolder>/".

Definition at line 153 of file HLTStreamHelper.cc.

{
  B2ASSERT("Event Meta Data not set!", m_eventMetaData.isValid());
 
  Belle2::MsgHandler msgHandler;
  streamHistogramImpl(gDirectory, msgHandler);
 
  auto evtMessage = std::unique_ptr<EvtMessage>(msgHandler.encode_msg(Belle2::ERecordType::MSG_EVENT));
 
  EventMetaData& eventMetaData = *m_eventMetaData;
  auto eventInformationString = TBufferJSON::ToJSON(&eventMetaData);
  zmq::message_t additionalEventMessage(eventInformationString.Data(), eventInformationString.Length());
 
  if (not compressed) {
    return Belle2::ZMQMessageFactory::createMessage(Belle2::EMessageTypes::c_rawDataMessage, evtMessage,
                                                    std::move(additionalEventMessage));
  }
 
  // TODO: do I want to use compression everywhere? That is probably not worth it! -> test this!
  if (m_outputBuffer.empty()) {
    m_outputBuffer.resize(m_maximalCompressedSize, 0);
  }
 
  B2DEBUG(10, "Size before compression " << evtMessage->size());
  int size = m_maximalCompressedSize;
  size = LZ4_compress_default(evtMessage->buffer(), &m_outputBuffer[0], evtMessage->size(), size);
  B2ASSERT("Compression failed", size > 0);
  B2DEBUG(10, "Size after compression " << size);
 
  zmq::message_t message(&m_outputBuffer[0], size);
 
  return Belle2::ZMQMessageFactory::createMessage(Belle2::EMessageTypes::c_compressedDataMessage,
                                                  std::move(message), std::move(additionalEventMessage));
}

◆ streamRaw()

std::unique_ptr< ZMQNoIdMessage > streamRaw ( )

Stream the data store into an event message and add SendHeader and SendTrailer around the message. Add ROI as additional message (if valid).

Definition at line 86 of file HLTStreamHelper.cc.

{
  const auto eventMessage = m_streamHelper.stream(false, false);
 
  // Fill Header and Trailer
  SendHeader hdr;
  SendTrailer trl;
 
  // Number of total words
  int msgsize = (eventMessage->size() - 1) / sizeof(int) + 1;
  int total_nwrds = msgsize +  hdr.GetHdrNwords() + trl.GetTrlNwords();
 
  // Fill header and trailer
  hdr.SetNwords(total_nwrds);
  hdr.SetNumEventsinPacket(1);
  hdr.SetNumNodesinPacket(1);
  hdr.SetEventNumber(m_eventMetaData->getEvent());
  //  hdr.SetExpRunWord(evtmeta->getRun());
  hdr.SetSubRunNum(m_eventMetaData->getSubrun());     // modified on Apr. 20, 2016 by SY
  hdr.SetRunNum(m_eventMetaData->getRun());
  hdr.SetExpNum(m_eventMetaData->getExperiment());
  hdr.SetNodeID(300);
 
  zmq::message_t rawMessage(total_nwrds * sizeof(int));
  int* buffer = rawMessage.data<int>();
 
  // Fill header
  memcpy(buffer, hdr.GetBuffer(), hdr.GetHdrNwords()*sizeof(int));
 
  // Fill EvtMessage
  memcpy(buffer + hdr.GetHdrNwords(), eventMessage->buffer(), eventMessage->size());
 
  // Fill trailer
  memcpy(buffer + hdr.GetHdrNwords() + msgsize, trl.GetBuffer(),
         trl.GetTrlNwords()*sizeof(int));
 
  zmq::message_t roiMessage = getROIMessageIfViable();
  auto zmqMessage = ZMQMessageFactory::createMessage(EMessageTypes::c_rawDataMessage, std::move(rawMessage), std::move(roiMessage));
 
  return zmqMessage;
}

Member Data Documentation

◆ m_eventMetaData

StoreObjPtr<EventMetaData> m_eventMetaData

private

Store Objects for HLT use.

Definition at line 67 of file HLTStreamHelper.h.

◆ m_maximalCompressedSize

unsigned int m_maximalCompressedSize = 100'000'000

private

Maximal size of the compression buffer.

Definition at line 103 of file HLTStreamHelper.h.

◆ m_outputBuffer

std::vector<char> m_outputBuffer

private

Temporary buffer for storing the compressed result.

Definition at line 101 of file HLTStreamHelper.h.

◆ m_randomGenerator

StoreObjPtr<RandomGenerator> m_randomGenerator

private

Additional Store Objects for ExpressReco use.

Definition at line 88 of file HLTStreamHelper.h.

◆ m_rawARICHs

StoreArray<RawARICH> m_rawARICHs

private

Store Objects for HLT use.

Definition at line 75 of file HLTStreamHelper.h.

◆ m_rawCDCs

StoreArray<RawCDC> m_rawCDCs

private

Store Objects for HLT use.

Definition at line 71 of file HLTStreamHelper.h.

◆ m_rawECLs

StoreArray<RawECL> m_rawECLs

private

Store Objects for HLT use.

Definition at line 77 of file HLTStreamHelper.h.

◆ m_rawFTSWs

StoreArray<RawFTSW> m_rawFTSWs

private

Store Objects for HLT use.

Definition at line 83 of file HLTStreamHelper.h.

◆ m_rawKLMs

StoreArray<RawKLM> m_rawKLMs

private

Store Objects for HLT use.

Definition at line 79 of file HLTStreamHelper.h.

◆ m_rawPXDs

StoreArray<RawPXD> m_rawPXDs

private

Additional Store Objects for ExpressReco use.

Definition at line 96 of file HLTStreamHelper.h.

◆ m_rawSVDs

StoreArray<RawSVD> m_rawSVDs

private

Store Objects for HLT use.

Definition at line 69 of file HLTStreamHelper.h.

◆ m_rawTOPs

StoreArray<RawTOP> m_rawTOPs

private

Store Objects for HLT use.

Definition at line 73 of file HLTStreamHelper.h.

◆ m_rawTRGs

StoreArray<RawTRG> m_rawTRGs

private

Store Objects for HLT use.

Definition at line 81 of file HLTStreamHelper.h.

◆ m_roiPayload

StoreObjPtr<ROIpayload> m_roiPayload

private

Store Objects for HLT use.

Definition at line 85 of file HLTStreamHelper.h.

◆ m_rois

StoreArray<ROIid> m_rois

private

Additional Store Objects for ExpressReco use.

Definition at line 98 of file HLTStreamHelper.h.

◆ m_softwareTriggerResult

StoreObjPtr<SoftwareTriggerResult> m_softwareTriggerResult

private

Additional Store Objects for ExpressReco use.

Definition at line 90 of file HLTStreamHelper.h.

◆ m_softwareTriggerVariables

StoreObjPtr<SoftwareTrigger::SoftwareTriggerVariables> m_softwareTriggerVariables

private

Additional Store Objects for ExpressReco use.

Definition at line 92 of file HLTStreamHelper.h.

◆ m_streamHelper

StreamHelper m_streamHelper

private

We use the framework stream helper.

Definition at line 64 of file HLTStreamHelper.h.

◆ m_triggerSummary

StoreObjPtr<TRGSummary> m_triggerSummary

private

Additional Store Objects for ExpressReco use.

Definition at line 94 of file HLTStreamHelper.h.

The documentation for this class was generated from the following files:

daq/hbasf2/utils/include/HLTStreamHelper.h
daq/hbasf2/utils/src/HLTStreamHelper.cc

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Member Function Documentation

◆ getROIMessageIfViable()

◆ initialize()

◆ read()

◆ registerStoreObjects()

◆ stream()

◆ streamHistograms()

◆ streamRaw()

Member Data Documentation

◆ m_eventMetaData

◆ m_maximalCompressedSize

◆ m_outputBuffer

◆ m_randomGenerator

◆ m_rawARICHs

◆ m_rawCDCs

◆ m_rawECLs

◆ m_rawFTSWs

◆ m_rawKLMs

◆ m_rawPXDs

◆ m_rawSVDs

◆ m_rawTOPs

◆ m_rawTRGs

◆ m_roiPayload

◆ m_rois

◆ m_softwareTriggerResult

◆ m_softwareTriggerVariables

◆ m_streamHelper

◆ m_triggerSummary