ByteStreamMergeOutputSvc.cxx

Go to the documentation of this file.

/*
  Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration
*/
 
#include "ByteStreamMergeOutputSvc.h"
#include "ByteStreamData/ByteStream.h"
 
#include "GaudiKernel/GenericAddress.h"
#include "GaudiKernel/MsgStream.h"
#include "GaudiKernel/ISvcLocator.h"
 
#include "eformat/SourceIdentifier.h"
 
#include <map>
 
using ROBF = OFFLINE_FRAGMENTS_NAMESPACE_WRITE::ROBFragment;
using ROBMAP = std::map<uint32_t, ROBF*>;
 
// Constructor.
ByteStreamMergeOutputSvc::ByteStreamMergeOutputSvc(const std::string& name, ISvcLocator* svcloc) :
   base_class(name,svcloc)
{
   // cppcheck-suppress useInitializationList; deprecated Property constructor
   m_bsOutputStreamName = name;
}
 
// setup input and output paths
StatusCode ByteStreamMergeOutputSvc::initialize() {
   ATH_CHECK( m_outSvc.retrieve() );
   ATH_CHECK( m_inSvc.retrieve() );
 
   return StatusCode::SUCCESS;
}
 
// ROBs from the L2 and event filter needs to be treated specially since the module id
// is used online to identify which node processed an event and it therefore varies from
// event to event - here we remap it to have module id 0, since there should only be one per event
uint32_t ByteStreamMergeOutputSvc::reducedROBid(uint32_t source_id) {
   eformat::helper::SourceIdentifier id = eformat::helper::SourceIdentifier(source_id);
   if (id.subdetector_id() == eformat::TDAQ_LVL2 || id.subdetector_id() == eformat::TDAQ_EVENT_FILTER) {
      return(eformat::helper::SourceIdentifier(id.subdetector_id(), 0).code());
   } else {
      return(source_id);
   }
}
 
// Read the next event.
bool  ByteStreamMergeOutputSvc::putEvent(const RawEvent* newEvent) {
   // get original event
   const RawEvent* orgEvent = m_inSvc->currentEvent();
 
   ATH_MSG_DEBUG("original BS size = " << 4 * orgEvent->fragment_size_word());
   ATH_MSG_DEBUG("athena BS size = " << 4 * newEvent->fragment_size_word());
 
   // do the merge...
 
   ROBMAP robsToAdd;
   // loop over all ROBs
   auto orgIter = orgEvent->child_iter();
   while (OFFLINE_FRAGMENTS_NAMESPACE::PointerType fp = orgIter.next()) {
      ROBF* rob = new ROBF(fp);
      robsToAdd[reducedROBid(rob->source_id())] = rob;
      ATH_MSG_DEBUG("original ROBFragment, src ID = " << std::hex << rob->source_id());
   }
   // now add/overwrite with newly created robs
   auto newIter = newEvent->child_iter();
   while (OFFLINE_FRAGMENTS_NAMESPACE::PointerType fp = newIter.next()) {
      ROBF* rob = new ROBF(fp);
      ROBMAP::const_iterator it = robsToAdd.find(reducedROBid(rob->source_id()));
      if (it != robsToAdd.end()) {
        delete it->second;
        ATH_MSG_DEBUG("overwriting ROBFragment with src ID = " << std::hex << rob->source_id());
      }
      robsToAdd[reducedROBid(rob->source_id())] = rob;
      ATH_MSG_DEBUG("new ROBFragment, src ID = " << std::hex << rob->source_id());
   }
   RawEventWrite* mergedEventWrite = new RawEventWrite();
   // copy header
   const RawEvent *event = orgEvent;
   if (m_overwriteHeader) {
      event = newEvent;
   }
   mergedEventWrite->source_id(event->source_id());
   mergedEventWrite->bc_time_seconds(event->bc_time_seconds());
   mergedEventWrite->bc_time_nanoseconds(event->bc_time_nanoseconds());
   mergedEventWrite->global_id(event->global_id());
   mergedEventWrite->run_type(event->run_type());
   mergedEventWrite->run_no(event->run_no());
   mergedEventWrite->lumi_block(event->lumi_block());
   mergedEventWrite->lvl1_id(event->lvl1_id());
   mergedEventWrite->bc_id(event->bc_id());
   mergedEventWrite->lvl1_trigger_type(event->lvl1_trigger_type());
   const OFFLINE_FRAGMENTS_NAMESPACE::DataType *tmp;
   event->status(tmp);
   mergedEventWrite->status(event->nstatus(), tmp);
   event->lvl1_trigger_info(tmp);
   mergedEventWrite->lvl1_trigger_info(event->nlvl1_trigger_info(), tmp);
   event->lvl2_trigger_info(tmp);
   mergedEventWrite->lvl2_trigger_info(event->nlvl2_trigger_info(), tmp);
   event->event_filter_info(tmp);
   mergedEventWrite->event_filter_info(event->nevent_filter_info(), tmp);
   event->stream_tag(tmp);
   mergedEventWrite->stream_tag(event->nstream_tag(), tmp);
   mergedEventWrite->checksum_type(event->checksum_type());
   // copy robs
   for(ROBMAP::iterator it = robsToAdd.begin(), itEnd = robsToAdd.end(); it != itEnd; ++it) {
      mergedEventWrite->append(it->second);
   }
   // convert RawEventWrite to RawEvent
   uint32_t rawSize = mergedEventWrite->size_word();
   OFFLINE_FRAGMENTS_NAMESPACE::DataType* buffer = new OFFLINE_FRAGMENTS_NAMESPACE::DataType[rawSize];
   uint32_t count = eformat::write::copy(*(mergedEventWrite->bind()), buffer, rawSize);
   if (count != rawSize) {
      ATH_MSG_ERROR("Memcopy failed " << count << " " << rawSize);
      return false;
   }
   RawEvent newRawEvent(buffer);
   StatusCode sc = m_outSvc->putEvent(&newRawEvent) ? StatusCode::SUCCESS : StatusCode::FAILURE;
   for(ROBMAP::iterator it = robsToAdd.begin(), itEnd = robsToAdd.end(); it != itEnd; ++it) {
      delete it->second; it->second = nullptr;
   }
   delete mergedEventWrite; mergedEventWrite = nullptr;
   delete [] buffer;
   if (sc != StatusCode::SUCCESS) {
      ATH_MSG_ERROR("Failed to put RawEvent");
   }
   return(true);
}
 
bool ByteStreamMergeOutputSvc::putEvent(const RawEvent* /*re*/, const EventContext& /*ctx*/) {
  ATH_MSG_FATAL(name() << " does not implement the context-aware putEvent method");
  return false;
}