ATLAS Offline Software
ByteStreamMergeOutputSvc.cxx
Go to the documentation of this file.
1 /*
2  Copyright (C) 2002-2024 CERN for the benefit of the ATLAS collaboration
3 */
4 
7 
8 #include "GaudiKernel/GenericAddress.h"
9 #include "GaudiKernel/MsgStream.h"
10 #include "GaudiKernel/ISvcLocator.h"
11 
12 #include "eformat/SourceIdentifier.h"
13 
14 #include <map>
15 
17 typedef std::map<uint32_t, ROBF*> ROBMAP;
18 
19 // Constructor.
20 ByteStreamMergeOutputSvc::ByteStreamMergeOutputSvc(const std::string& name, ISvcLocator* svcloc) :
21  base_class(name,svcloc),
22  m_bsOutputStreamName(name)
23 {
24 }
25 
26 // Destructor.
28 }
29 
30 // setup input and output paths
32  ATH_CHECK( m_outSvc.retrieve() );
33  ATH_CHECK( m_inSvc.retrieve() );
34 
35  return(StatusCode::SUCCESS);
36 }
37 
38 // ROBs from the L2 and event filter needs to be treated specially since the module id
39 // is used online to identify which node processed an event and it therefore varies from
40 // event to event - here we remap it to have module id 0, since there should only be one per event
42  eformat::helper::SourceIdentifier id = eformat::helper::SourceIdentifier(source_id);
43  if (id.subdetector_id() == eformat::TDAQ_LVL2 || id.subdetector_id() == eformat::TDAQ_EVENT_FILTER) {
44  return(eformat::helper::SourceIdentifier(id.subdetector_id(), 0).code());
45  } else {
46  return(source_id);
47  }
48 }
49 
50 // Read the next event.
52  // get original event
53  const RawEvent* orgEvent = m_inSvc->currentEvent();
54 
55  ATH_MSG_DEBUG("original BS size = " << 4 * orgEvent->fragment_size_word());
56  ATH_MSG_DEBUG("athena BS size = " << 4 * newEvent->fragment_size_word());
57 
58  // do the merge...
59  // get all the ROBFragments
60  const size_t MAX_ROBFRAGMENTS = 2048;
61  OFFLINE_FRAGMENTS_NAMESPACE::PointerType orgRobF[MAX_ROBFRAGMENTS];
62  OFFLINE_FRAGMENTS_NAMESPACE::PointerType newRobF[MAX_ROBFRAGMENTS];
63  size_t orgrobcount = orgEvent->children(orgRobF, MAX_ROBFRAGMENTS);
64  if (orgrobcount == MAX_ROBFRAGMENTS) {
65  ATH_MSG_ERROR("ROB buffer overflow");
66  return false;
67  }
68  size_t newrobcount = newEvent->children(newRobF,MAX_ROBFRAGMENTS);
69  if (newrobcount == MAX_ROBFRAGMENTS) {
70  ATH_MSG_ERROR("ROB buffer overflow");
71  return false;
72  }
73 
74  ROBMAP robsToAdd;
75  // loop over all ROBs
76  for (size_t irob = 0; irob < orgrobcount; ++irob) {
77  ROBF* rob = new ROBF(orgRobF[irob]);
78  robsToAdd[reducedROBid(rob->source_id())] = rob;
79  ATH_MSG_DEBUG("original ROBFragment, src ID = " << std::hex << rob->source_id());
80  }
81  // now add/overwrite with newly created robs
82  for (size_t irob = 0; irob < newrobcount; ++irob) {
83  ROBF* rob = new ROBF(newRobF[irob]);
84  ROBMAP::const_iterator it = robsToAdd.find(reducedROBid(rob->source_id()));
85  if (it != robsToAdd.end()) {
86  delete it->second;
87  ATH_MSG_DEBUG("overwriting ROBFragment with src ID = " << std::hex << rob->source_id());
88  }
89  robsToAdd[reducedROBid(rob->source_id())] = rob;
90  ATH_MSG_DEBUG("new ROBFragment, src ID = " << std::hex << rob->source_id());
91  }
92  RawEventWrite* mergedEventWrite = new RawEventWrite();
93  // copy header
94  const RawEvent *event = orgEvent;
95  if (m_overwriteHeader) {
96  event = newEvent;
97  }
98  mergedEventWrite->source_id(event->source_id());
99  mergedEventWrite->bc_time_seconds(event->bc_time_seconds());
100  mergedEventWrite->bc_time_nanoseconds(event->bc_time_nanoseconds());
101  mergedEventWrite->global_id(event->global_id());
102  mergedEventWrite->run_type(event->run_type());
103  mergedEventWrite->run_no(event->run_no());
104  mergedEventWrite->lumi_block(event->lumi_block());
105  mergedEventWrite->lvl1_id(event->lvl1_id());
106  mergedEventWrite->bc_id(event->bc_id());
107  mergedEventWrite->lvl1_trigger_type(event->lvl1_trigger_type());
109  event->status(tmp);
110  mergedEventWrite->status(event->nstatus(), tmp);
111  event->lvl1_trigger_info(tmp);
112  mergedEventWrite->lvl1_trigger_info(event->nlvl1_trigger_info(), tmp);
113  event->lvl2_trigger_info(tmp);
114  mergedEventWrite->lvl2_trigger_info(event->nlvl2_trigger_info(), tmp);
115  event->event_filter_info(tmp);
116  mergedEventWrite->event_filter_info(event->nevent_filter_info(), tmp);
117  event->stream_tag(tmp);
118  mergedEventWrite->stream_tag(event->nstream_tag(), tmp);
119  mergedEventWrite->checksum_type(event->checksum_type());
120  // copy robs
121  for(ROBMAP::iterator it = robsToAdd.begin(), itEnd = robsToAdd.end(); it != itEnd; ++it) {
122  mergedEventWrite->append(it->second);
123  }
124  // convert RawEventWrite to RawEvent
125  uint32_t rawSize = mergedEventWrite->size_word();
127  uint32_t count = eformat::write::copy(*(mergedEventWrite->bind()), buffer, rawSize);
128  if (count != rawSize) {
129  ATH_MSG_ERROR("Memcopy failed " << count << " " << rawSize);
130  return false;
131  }
132  RawEvent newRawEvent(buffer);
133  StatusCode sc = m_outSvc->putEvent(&newRawEvent) ? StatusCode::SUCCESS : StatusCode::FAILURE;
134  for(ROBMAP::iterator it = robsToAdd.begin(), itEnd = robsToAdd.end(); it != itEnd; ++it) {
135  delete it->second; it->second = 0;
136  }
137  delete mergedEventWrite; mergedEventWrite = 0;
138  delete [] buffer;
139  if (sc != StatusCode::SUCCESS) {
140  ATH_MSG_ERROR("Failed to put RawEvent");
141  }
142  return(true);
143 }
144 
145 bool ByteStreamMergeOutputSvc::putEvent(const RawEvent* /*re*/, const EventContext& /*ctx*/) {
146  ATH_MSG_FATAL(name() << " does not implement the context-aware putEvent method");
147  return false;
148 }
xAOD::iterator
JetConstituentVector::iterator iterator
Definition: JetConstituentVector.cxx:68
RawEventWrite
OFFLINE_FRAGMENTS_NAMESPACE_WRITE::FullEventFragment RawEventWrite
data type for writing raw event
Definition: RawEvent.h:39
ATH_MSG_FATAL
#define ATH_MSG_FATAL(x)
Definition: AthMsgStreamMacros.h:34
OFFLINE_FRAGMENTS_NAMESPACE::DataType
uint32_t DataType
Definition: RawEvent.h:24
xAOD::uint32_t
setEventNumber uint32_t
Definition: EventInfo_v1.cxx:127
ByteStreamMergeOutputSvc::m_outSvc
ServiceHandle< IByteStreamOutputSvc > m_outSvc
Definition: ByteStreamMergeOutputSvc.h:39
RawEvent
OFFLINE_FRAGMENTS_NAMESPACE::FullEventFragment RawEvent
data type for reading raw event
Definition: RawEvent.h:37
skel.it
it
Definition: skel.GENtoEVGEN.py:396
XMLtoHeader.count
count
Definition: XMLtoHeader.py:85
AthenaPoolTestRead.sc
sc
Definition: AthenaPoolTestRead.py:27
ByteStreamMergeOutputSvc::putEvent
virtual bool putEvent(const RawEvent *re) override
Implementation of the IByteStreamOutputSvc interface methods.
Definition: ByteStreamMergeOutputSvc.cxx:51
ByteStreamMergeOutputSvc::ByteStreamMergeOutputSvc
ByteStreamMergeOutputSvc(const std::string &name, ISvcLocator *svcloc)
Constructors:
Definition: ByteStreamMergeOutputSvc.cxx:20
ROBF
OFFLINE_FRAGMENTS_NAMESPACE_WRITE::ROBFragment ROBF
Definition: ByteStreamMergeOutputSvc.cxx:16
ByteStreamMergeOutputSvc::m_inSvc
ServiceHandle< IByteStreamInputSvc > m_inSvc
Definition: ByteStreamMergeOutputSvc.h:38
ByteStreamMergeOutputSvc.h
This file contains the class definition for the ByteStreamMergeOutputSvc class.
OFFLINE_FRAGMENTS_NAMESPACE::PointerType
const DataType * PointerType
Definition: RawEvent.h:25
histSizes.code
code
Definition: histSizes.py:129
createCoolChannelIdFile.buffer
buffer
Definition: createCoolChannelIdFile.py:12
ATH_MSG_ERROR
#define ATH_MSG_ERROR(x)
Definition: AthMsgStreamMacros.h:33
event
POOL::TEvent event(POOL::TEvent::kClassAccess)
EL::StatusCode
::StatusCode StatusCode
StatusCode definition for legacy code.
Definition: PhysicsAnalysis/D3PDTools/EventLoop/EventLoop/StatusCode.h:22
ATH_MSG_DEBUG
#define ATH_MSG_DEBUG(x)
Definition: AthMsgStreamMacros.h:29
ByteStreamMergeOutputSvc::initialize
virtual StatusCode initialize() override
Definition: ByteStreamMergeOutputSvc.cxx:31
ByteStreamMergeOutputSvc::~ByteStreamMergeOutputSvc
virtual ~ByteStreamMergeOutputSvc()
Destructor.
Definition: ByteStreamMergeOutputSvc.cxx:27
ByteStreamMergeOutputSvc::reducedROBid
uint32_t reducedROBid(uint32_t)
Definition: ByteStreamMergeOutputSvc.cxx:41
ByteStreamMergeOutputSvc::m_overwriteHeader
Gaudi::Property< bool > m_overwriteHeader
Definition: ByteStreamMergeOutputSvc.h:42
ATH_CHECK
#define ATH_CHECK
Definition: AthCheckMacros.h:40
DeMoUpdate.tmp
string tmp
Definition: DeMoUpdate.py:1167
name
std::string name
Definition: Control/AthContainers/Root/debug.cxx:228
ByteStream.h
OFFLINE_FRAGMENTS_NAMESPACE_WRITE::ROBFragment
eformat::write::ROBFragment ROBFragment
Definition: RawEvent.h:33
calibdata.copy
bool copy
Definition: calibdata.py:27
ROBMAP
std::map< uint32_t, ROBF * > ROBMAP
Definition: ByteStreamMergeOutputSvc.cxx:17