Line data Source code
1 : #ifndef ARTDAQ_ARTDAQ_ARTMODULES_ARTDAQINPUTHELPER_HH_
2 : #define ARTDAQ_ARTDAQ_ARTMODULES_ARTDAQINPUTHELPER_HH_
3 :
4 : #include "TRACE/tracemf.h" // Pre-empt TRACE/trace.h from Fragment.hh.
5 :
6 : #include "artdaq-core/Data/Fragment.hh"
7 : #include "artdaq/ArtModules/InputUtilities.hh"
8 :
9 : #include "artdaq-core/Data/MetadataFragment.hh"
10 : #include "artdaq-core/Data/detail/ParentageMap.hh"
11 : #include "artdaq-core/Utilities/TimeUtils.hh"
12 : #include "artdaq/ArtModules/ArtdaqFragmentNamingService.h"
13 : #include "artdaq/ArtModules/ArtdaqSharedMemoryServiceInterface.h"
14 : #include "artdaq/DAQdata/Globals.hh"
15 : #include "artdaq/DAQdata/NetMonHeader.hh"
16 :
17 : #include "art/Framework/Core/FileBlock.h"
18 : #include "art/Framework/Core/ProductRegistryHelper.h"
19 : #include "art/Framework/IO/Sources/SourceHelper.h"
20 : #include "art/Framework/IO/Sources/put_product_in_principal.h"
21 : #include "art/Framework/Principal/EventPrincipal.h"
22 : #include "art/Framework/Principal/RunPrincipal.h"
23 : #include "art/Framework/Principal/SubRunPrincipal.h"
24 : #include "art/Framework/Services/Registry/ServiceHandle.h"
25 :
26 : #include "art/Persistency/Provenance/ProcessHistoryRegistry.h"
27 : #include "art_root_io/setup.h"
28 :
29 : #include "canvas/Persistency/Common/EDProduct.h"
30 : #include "canvas/Persistency/Provenance/BranchDescription.h"
31 : #include "canvas/Persistency/Provenance/BranchKey.h"
32 : #include "canvas/Persistency/Provenance/FileFormatVersion.h"
33 : #if ART_HEX_VERSION < 0x31100
34 : #include "canvas/Persistency/Provenance/History.h"
35 : #else
36 : #include "canvas/Persistency/Provenance/Compatibility/History.h"
37 : #endif
38 : #include "canvas/Persistency/Provenance/ParentageRegistry.h"
39 : #include "canvas/Persistency/Provenance/ProcessHistory.h"
40 : #include "canvas/Persistency/Provenance/ProcessHistoryID.h"
41 : #include "canvas/Persistency/Provenance/ProductList.h"
42 : #include "canvas/Persistency/Provenance/ProductProvenance.h"
43 :
44 : #include "fhiclcpp/ParameterSet.h"
45 : #include "fhiclcpp/ParameterSetID.h"
46 : #include "fhiclcpp/ParameterSetRegistry.h"
47 :
48 : #include <TBufferFile.h>
49 : #include <TClass.h>
50 : #include <TList.h>
51 : #include <TStreamerInfo.h>
52 :
53 : #include <sys/time.h>
54 : #include <chrono>
55 : #include <cstdio>
56 : #include <iomanip>
57 : #include <iostream>
58 : #include <list>
59 : #include <map>
60 : #include <memory>
61 : #include <set>
62 : #include <sstream>
63 : #include <string>
64 : #include <unordered_map>
65 : #include <utility>
66 : #include <vector>
67 :
68 : #define CAN_REINIT 0
69 :
70 : namespace art {
71 : template<typename U>
72 : class ArtdaqInputHelper;
73 : } // namespace art
74 :
75 : /**
76 : * \brief This template class provides a unified interface for reading data into art
77 : * \tparam U The class responsible for delivering data
78 : *
79 : * JCF, May-27-2016
80 : * ArtdaqInputHelper is a template class which takes, as a parameter, a
81 : * class which it uses to receive data; the instance of this class is
82 : * called "communicationWrapper_". As of this writing, this wrapper
83 : * class is implemented by NetMonWrapper (for reading data into the
84 : * aggregator from the eventbuilder) and TransferWrapper (for reading
85 : * data into an art process). This class presents a unified approach
86 : * to handling art provenance, regardless of the communication
87 : * protocol used to read data in.
88 : */
89 : template<typename U>
90 : class art::ArtdaqInputHelper
91 : {
92 : public:
93 : /**
94 : * \brief Copy Constructor is deleted
95 : */
96 : ArtdaqInputHelper(const ArtdaqInputHelper&) = delete;
97 :
98 : /**
99 : * \brief Copy Assignment operator is deleted
100 : * \return ArtdaqInputHelper copy
101 : */
102 : ArtdaqInputHelper& operator=(const ArtdaqInputHelper&) = delete;
103 :
104 : /**
105 : * \brief ArtdaqInputHelper Destructor
106 : */
107 : ~ArtdaqInputHelper();
108 :
109 : /**
110 : * \brief ArtdaqInputHelper Constructor
111 : * \param ps ParameterSet used to confiugre communication wrapper class
112 : * \param helper An art::ProductRegistryHelper for registering products
113 : * \param pm An art::SourceHelper for handling provenance
114 : */
115 : ArtdaqInputHelper(const fhicl::ParameterSet& ps, art::ProductRegistryHelper& helper, art::SourceHelper const& pm);
116 :
117 : /**
118 : * \brief Called by art to close the input source. No-Op
119 : */
120 : void closeCurrentFile();
121 :
122 : /**
123 : * \brief Emulate reading a file
124 : * \param fb Output art::FileBlock object
125 : */
126 : void readFile(const std::string&, art::FileBlock*& fb);
127 :
128 : /**
129 : * \brief Whether additional events are expected from the source
130 : * \return True if ArtdaqInputHelper has not been shut down
131 : */
132 : bool hasMoreData() const;
133 :
134 : /**
135 : * \brief Read the next event from the communication wrapper
136 : * \param inR RunPrincipal input pointer
137 : * \param inSR SubRunPrincipal input pointer
138 : * \param outR RunPrincipal output pointer
139 : * \param outSR SubRunPrincipal output pointer
140 : * \param outE EventPrincipal output pointer
141 : * \return Whether an event was successfully read from the communication wrapper
142 : */
143 : bool readNext(art::RunPrincipal* const inR, art::SubRunPrincipal* const inSR, art::RunPrincipal*& outR,
144 : art::SubRunPrincipal*& outSR, art::EventPrincipal*& outE);
145 :
146 : private:
147 : ArtdaqInputHelper(ArtdaqInputHelper&&) = delete;
148 : ArtdaqInputHelper& operator=(ArtdaqInputHelper&&) = delete;
149 :
150 : void readAndConstructPrincipal(std::unique_ptr<TBufferFile>&, artdaq::NetMonHeader::MessageType, art::RunPrincipal* const,
151 : art::SubRunPrincipal* const, art::RunPrincipal*&, art::SubRunPrincipal*&,
152 : art::EventPrincipal*&);
153 :
154 : bool constructPrincipal(std::shared_ptr<ArtdaqEvent>, art::RunPrincipal* const,
155 : art::SubRunPrincipal* const, art::RunPrincipal*&, art::SubRunPrincipal*&,
156 : art::EventPrincipal*&);
157 :
158 : template<class T>
159 : void readDataProducts(std::list<std::unique_ptr<TBufferFile>>&, T* const&);
160 :
161 : void putInPrincipal(RunPrincipal* const&, std::unique_ptr<EDProduct>&&, const BranchDescription&,
162 : std::unique_ptr<const ProductProvenance>&&);
163 :
164 : void putInPrincipal(SubRunPrincipal* const&, std::unique_ptr<EDProduct>&&, const BranchDescription&,
165 : std::unique_ptr<const ProductProvenance>&&);
166 :
167 : void putInPrincipal(EventPrincipal* const&, std::unique_ptr<EDProduct>&&, const BranchDescription&,
168 : std::unique_ptr<const ProductProvenance>&&);
169 :
170 : std::pair<bool, bool> readFragments(std::unordered_map<artdaq::Fragment::type_t, std::unique_ptr<artdaq::Fragments>> const& eventMap, art::RunPrincipal* const theRun, art::SubRunPrincipal* const theSubRun, art::EventPrincipal* const theEvent);
171 :
172 : void readInitMessage();
173 :
174 : bool shutdownMsgReceived_;
175 : art::SourceHelper const& pm_;
176 : art::ProductRegistryHelper& helper_;
177 : U communicationWrapper_;
178 : ProductList* productList_;
179 : std::unique_ptr<art::History> history_to_use_;
180 : bool fragmentsOnlyMode_;
181 : std::string pretend_module_name; ///< The module name to store data under
182 : size_t bytesRead; ///< running total of number of bytes received
183 : std::chrono::steady_clock::time_point last_read_time; ///< Time last read was completed
184 : };
185 :
186 : template<typename U>
187 0 : art::ArtdaqInputHelper<U>::ArtdaqInputHelper(const fhicl::ParameterSet& ps, art::ProductRegistryHelper& helper,
188 : art::SourceHelper const& pm)
189 0 : : shutdownMsgReceived_(false)
190 0 : , pm_(pm)
191 0 : , helper_(helper)
192 0 : , communicationWrapper_(ps)
193 0 : , productList_()
194 0 : , fragmentsOnlyMode_(false)
195 0 : , pretend_module_name(ps.get<std::string>("raw_data_label", "daq"))
196 0 : , bytesRead(0)
197 0 : , last_read_time(std::chrono::steady_clock::now())
198 : {
199 0 : root::setup();
200 : // Instantiate ArtdaqSharedMemoryService to set up artdaq Globals and MetricManager
201 0 : art::ServiceHandle<ArtdaqSharedMemoryServiceInterface> shm;
202 :
203 : #if 0
204 : volatile bool loop = true;
205 : while (loop)
206 : {
207 : usleep(1000);
208 : }
209 : #endif
210 :
211 : // JCF, May-27-2016
212 :
213 : // Something will have to be done about the labeling of this class,
214 : // since it's just a template class- the user will care about the
215 : // specific instantiation when it comes to messages
216 :
217 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "Begin: ArtdaqInputHelper::ArtdaqInputHelper("
218 0 : << "const fhicl::ParameterSet& ps, "
219 0 : << "art::ProductRegistryHelper& helper, "
220 0 : << "const art::SourceHelper& pm)";
221 0 : readInitMessage();
222 :
223 0 : helper.reconstitutes<artdaq::detail::RawEventHeader, art::InEvent>(pretend_module_name, "RawEventHeader");
224 :
225 0 : if (ps.get<bool>("register_fragment_types", true))
226 : {
227 0 : TLOG(TLVL_DEBUG + 32, "ArtdaqInputHelper") << "Registering known Fragment labels from ArtdaqFragmentNamingServiceInterface";
228 :
229 0 : art::ServiceHandle<ArtdaqFragmentNamingServiceInterface> translator;
230 0 : helper.reconstitutes<artdaq::Fragments, art::InEvent>(pretend_module_name, translator->GetUnidentifiedInstanceName());
231 : // Workaround for #22979
232 0 : helper.reconstitutes<artdaq::Fragments, art::InRun>(pretend_module_name, translator->GetUnidentifiedInstanceName());
233 0 : helper.reconstitutes<artdaq::Fragments, art::InSubRun>(pretend_module_name, translator->GetUnidentifiedInstanceName());
234 :
235 0 : helper.reconstitutes<std::vector<artdaq::ArtdaqMetadata>, art::InRun>(pretend_module_name, "ArtdaqMetadata");
236 0 : helper.reconstitutes<std::vector<artdaq::ArtdaqMetadata>, art::InSubRun>(pretend_module_name, "ArtdaqMetadata");
237 :
238 0 : std::set<std::string> instance_names = translator->GetAllProductInstanceNames();
239 0 : for (const auto& set_iter : instance_names)
240 : {
241 0 : helper.reconstitutes<artdaq::Fragments, art::InEvent>(pretend_module_name, set_iter);
242 : }
243 0 : }
244 : //
245 : // Finished with init message.
246 : //
247 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "End: ArtdaqInputHelper::ArtdaqInputHelper("
248 0 : << "const fhicl::ParameterSet& ps, "
249 0 : << "art::ProductRegistryHelper& helper, "
250 0 : << "const art::SourceHelper& pm)";
251 0 : }
252 :
253 : template<typename U>
254 0 : void art::ArtdaqInputHelper<U>::readInitMessage()
255 : {
256 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "Going to receive init message";
257 0 : artdaq::FragmentPtrs initFrags = communicationWrapper_.receiveInitMessage();
258 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "Init message received";
259 :
260 0 : if (!initFrags.empty() && initFrags.front().get()->type() == artdaq::Fragment::EndOfDataFragmentType)
261 : {
262 0 : TLOG_ERROR("ArtdaqInputHelper") << "Received EndOfData as first broadcast! This process never received any data!";
263 0 : shutdownMsgReceived_ = true;
264 : }
265 : else
266 : {
267 0 : if (initFrags.empty() || initFrags.back().get()->dataSize() == 0)
268 : {
269 0 : TLOG(TLVL_DEBUG + 32, "ArtdaqInputHelper") << "No init message received or zero-size init message: Fragments-only mode activated! This is an EventBuilder!";
270 0 : fragmentsOnlyMode_ = true;
271 : }
272 : else
273 : {
274 0 : std::list<std::unique_ptr<TBufferFile>> msgs;
275 0 : for (auto& initFrag : initFrags)
276 : {
277 0 : auto header = initFrag->metadata<artdaq::NetMonHeader>();
278 0 : msgs.emplace_back(new TBufferFile(TBuffer::kRead, header->data_length, initFrag->dataBegin(), kFALSE, nullptr));
279 : }
280 :
281 0 : std::set<art::ProcessHistoryID> history_ids;
282 :
283 0 : for (auto& msg : msgs)
284 : {
285 : // This first unsigned long is the message type code, ignored here in the constructor
286 0 : ULong_t dummy = 0;
287 0 : msg->ReadULong(dummy);
288 :
289 : // ELF: 6/11/2019: This code is taken from TSocket::RecvStreamerInfos
290 0 : auto list = dynamic_cast<TList*>(msg->ReadObject(TList::Class()));
291 :
292 0 : TIter next(list);
293 : TStreamerInfo* info;
294 0 : TObjLink* lnk = list->FirstLink();
295 : // First call BuildCheck for regular class
296 0 : while (lnk)
297 : {
298 0 : info = dynamic_cast<TStreamerInfo*>(lnk->GetObject());
299 0 : TObject* element = info->GetElements()->UncheckedAt(0);
300 0 : Bool_t isstl = element && strcmp("This", element->GetName()) == 0;
301 0 : if (!isstl)
302 : {
303 0 : info->BuildCheck();
304 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "ArtdaqInputHelper: importing TStreamerInfo: " << info->GetName() << ", version = " << info->GetClassVersion();
305 : }
306 0 : lnk = lnk->Next();
307 : }
308 : // Then call BuildCheck for stl class
309 0 : lnk = list->FirstLink();
310 0 : while (lnk)
311 : {
312 0 : info = dynamic_cast<TStreamerInfo*>(lnk->GetObject());
313 0 : TObject* element = info->GetElements()->UncheckedAt(0);
314 0 : Bool_t isstl = element && strcmp("This", element->GetName()) == 0;
315 0 : if (isstl)
316 : {
317 0 : info->BuildCheck();
318 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "ArtdaqInputHelper: importing TStreamerInfo: " << info->GetName() << ", version = " << info->GetClassVersion();
319 : }
320 0 : lnk = lnk->Next();
321 : }
322 : // ELF: 6/11/2019: End TSocket snippet
323 :
324 : //
325 : // Read the ParameterSetRegistry.
326 : //
327 0 : ULong_t ps_cnt = 0;
328 0 : msg->ReadULong(ps_cnt);
329 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "ArtdaqInputHelper: parameter set count: " << ps_cnt;
330 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "ArtdaqInputHelper: reading parameter sets ...";
331 0 : for (ULong_t I = 0; I < ps_cnt; ++I)
332 : {
333 0 : std::string pset_str = ""; // = ReadObjectAny<std::string>(msg, "std::string", "ArtdaqInputHelper::ArtdaqInputHelper");
334 0 : msg->ReadStdString(pset_str);
335 :
336 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "ArtdaqInputHelper: parameter set: " << pset_str;
337 :
338 0 : fhicl::ParameterSet pset;
339 0 : pset = fhicl::ParameterSet::make(pset_str);
340 : // Force id calculation.
341 0 : pset.id();
342 0 : fhicl::ParameterSetRegistry::put(pset);
343 : }
344 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "ArtdaqInputHelper: finished reading parameter sets.";
345 :
346 : //
347 : // Read the MasterProductRegistry.
348 : //
349 0 : auto thisProductList = ReadObjectAny<art::ProductList>(
350 : msg, "std::map<art::BranchKey,art::BranchDescription>", "ArtdaqInputHelper::ArtdaqInputHelper");
351 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "ArtdaqInputHelper: Input Product list sz=" << thisProductList->size();
352 :
353 0 : bool productListInitialized = productList_ != nullptr;
354 0 : if (!productListInitialized) productList_ = thisProductList;
355 0 : for (auto I = thisProductList->begin(), E = thisProductList->end(); I != E; ++I)
356 : {
357 : #ifndef __OPTIMIZE__
358 0 : TLOG(TLVL_DEBUG + 50, "ArtdaqInputHelper") << "Branch key: class: '" << I->first.friendlyClassName_ << "' modlbl: '"
359 0 : << I->first.moduleLabel_ << "' instnm: '" << I->first.productInstanceName_ << "' procnm: '"
360 0 : << I->first.processName_ << "', branch description name: " << I->second.wrappedName();
361 : #endif
362 0 : if (productListInitialized)
363 : {
364 0 : productList_->emplace(*I);
365 : }
366 : }
367 :
368 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "ArtdaqInputHelper: Reading ProcessHistory";
369 0 : auto phm = ReadObjectAny<art::ProcessHistoryMap>(
370 : msg, "std::map<const art::Hash<2>,art::ProcessHistory>", "ArtdaqInputHelper::ArtdaqInputHelper");
371 0 : printProcessMap(*phm, "ArtdaqInputHelper's ProcessHistoryMap");
372 :
373 0 : for (auto& proc_hist : *phm)
374 : {
375 0 : history_ids.insert(proc_hist.second.id());
376 : }
377 :
378 0 : ProcessHistoryRegistry::put(*phm);
379 0 : printProcessMap(ProcessHistoryRegistry::get(), "ArtdaqInputHelper's ProcessHistoryRegistry");
380 :
381 : //
382 : // Read the ParentageRegistry.
383 : //
384 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "ArtdaqInputHelper: Reading ParentageMap";
385 0 : auto parentageMap = ReadObjectAny<ParentageMap>(msg, "art::ParentageMap", "ArtdaqInputHelper::ArtdaqInputHelper");
386 0 : ParentageRegistry::put(*parentageMap);
387 : }
388 :
389 : // We're going to make a fake History using the collected process histories!
390 0 : art::ProcessHistory fake_process_history;
391 0 : for (auto& hist : history_ids)
392 : {
393 0 : if (!hist.isValid())
394 : {
395 0 : TLOG(TLVL_WARNING, "ArtdaqInputHelper") << "Encountered invalid history ID!";
396 0 : continue;
397 0 : }
398 :
399 0 : ProcessHistory thisProcessHistory;
400 0 : if (ProcessHistoryRegistry::get(hist, thisProcessHistory))
401 : {
402 0 : for (auto& conf : thisProcessHistory)
403 : {
404 0 : if (auto e = fake_process_history.end();
405 0 : std::find(fake_process_history.begin(), e, conf) == e)
406 : {
407 0 : fake_process_history.push_back(conf);
408 : }
409 : }
410 : }
411 : }
412 0 : art::ProcessHistoryMap fake_process_history_map;
413 0 : fake_process_history_map[fake_process_history.id()] = fake_process_history;
414 0 : ProcessHistoryRegistry::put(fake_process_history_map);
415 0 : printProcessMap(ProcessHistoryRegistry::get(), "ArtdaqInputHelper's ProcessHistoryRegistry w/fake history");
416 :
417 0 : history_to_use_.reset(new History());
418 0 : history_to_use_->setProcessHistoryID(fake_process_history.id());
419 :
420 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper")
421 0 : << "ArtdaqInputHelper: Product list sz=" << productList_->size();
422 : #if 0
423 : for (auto I = productList_->begin(), E = productList_->end(); I != E; ++I)
424 : {
425 : TLOG(TLVL_DEBUG + 50, "ArtdaqInputHelper") << "Branch key: class: '" << I->first.friendlyClassName_ << "' modlbl: '"
426 : << I->first.moduleLabel_ << "' instnm: '" << I->first.productInstanceName_ << "' procnm: '"
427 : << I->first.processName_ << "', branch description name: " << I->second.wrappedName();
428 : }
429 : #endif
430 : // helper now owns productList_!
431 :
432 0 : helper_.productList(std::unique_ptr<art::ProductList>(productList_));
433 :
434 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "ArtdaqInputHelper: got product list";
435 0 : }
436 : }
437 0 : }
438 :
439 : template<typename U>
440 0 : art::ArtdaqInputHelper<U>::~ArtdaqInputHelper()
441 0 : {}
442 :
443 : template<typename U>
444 0 : void art::ArtdaqInputHelper<U>::closeCurrentFile()
445 : {
446 0 : TLOG(TLVL_DEBUG + 34, "ArtdaqInputHelper") << "Begin/End: ArtdaqInputHelper::closeCurrentFile()";
447 0 : }
448 :
449 : template<typename U>
450 0 : void art::ArtdaqInputHelper<U>::readFile(const std::string&, art::FileBlock*& fb)
451 : {
452 0 : TLOG(TLVL_DEBUG + 35, "ArtdaqInputHelper") << "Begin: ArtdaqInputHelper::"
453 : "readFile(const std::string& name, art::FileBlock*& fb)";
454 0 : fb = new art::FileBlock(art::FileFormatVersion(1, "ArtdaqInputHelper2013"), "");
455 0 : TLOG(TLVL_DEBUG + 35, "ArtdaqInputHelper") << "End: ArtdaqInputHelper::"
456 : "readFile(const std::string& name, art::FileBlock*& fb)";
457 0 : }
458 :
459 : template<typename U>
460 0 : bool art::ArtdaqInputHelper<U>::hasMoreData() const
461 : {
462 0 : TLOG(TLVL_DEBUG + 36, "ArtdaqInputHelper") << "Begin: ArtdaqInputHelper::hasMoreData()";
463 0 : if (shutdownMsgReceived_)
464 : {
465 0 : TLOG(TLVL_DEBUG + 36, "ArtdaqInputHelper") << "ArtdaqInputHelper::hasMoreData(): "
466 : "returning false on shutdownMsgReceived_.";
467 0 : TLOG(TLVL_DEBUG + 36, "ArtdaqInputHelper") << "End: ArtdaqInputHelper::hasMoreData()";
468 0 : return false;
469 : }
470 0 : TLOG(TLVL_DEBUG + 32, "ArtdaqInputHelper") << "ArtdaqInputHelper::hasMoreData(): "
471 : "returning true on not shutdownMsgReceived_.";
472 0 : TLOG(TLVL_DEBUG + 36, "ArtdaqInputHelper") << "End: ArtdaqInputHelper::hasMoreData()";
473 0 : return true;
474 : }
475 :
476 : template<typename U>
477 0 : void art::ArtdaqInputHelper<U>::readAndConstructPrincipal(std::unique_ptr<TBufferFile>& msg, artdaq::NetMonHeader::MessageType msg_type_code,
478 : art::RunPrincipal* const inR, art::SubRunPrincipal* const inSR,
479 : art::RunPrincipal*& outR, art::SubRunPrincipal*& outSR,
480 : art::EventPrincipal*& outE)
481 : {
482 : //
483 : // Process the message.
484 : //
485 0 : std::unique_ptr<art::RunAuxiliary> run_aux;
486 0 : std::unique_ptr<art::SubRunAuxiliary> subrun_aux;
487 0 : std::unique_ptr<art::EventAuxiliary> event_aux;
488 :
489 : // Establish default 'results'
490 0 : outR = nullptr;
491 0 : outSR = nullptr;
492 0 : outE = nullptr;
493 :
494 0 : art::Timestamp currentTime = 0;
495 : timespec hi_res_time;
496 0 : int retcode = clock_gettime(CLOCK_REALTIME, &hi_res_time);
497 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "hi_res_time tv_sec = " << hi_res_time.tv_sec
498 0 : << " tv_nsec = " << hi_res_time.tv_nsec << " (retcode = " << retcode << ")";
499 0 : if (retcode == 0)
500 : {
501 0 : currentTime = ((hi_res_time.tv_sec & 0xffffffff) << 32) | (hi_res_time.tv_nsec & 0xffffffff);
502 : }
503 : else
504 : {
505 0 : TLOG_ERROR("ArtdaqInputHelper")
506 0 : << "Unable to fetch a high-resolution time with clock_gettime for art::SubRun Timestamp. ";
507 : }
508 :
509 0 : TLOG(TLVL_DEBUG + 37, "ArtdaqInputHelper") << "inR: " << static_cast<void*>(inR) << " run " << (inR ? std::to_string(inR->run()) : "invalid")
510 0 : << ", inSR: " << static_cast<void*>(inSR) << " run " << (inSR ? std::to_string(inSR->run()) : "invalid")
511 0 : << ", subrun " << (inSR ? std::to_string(inSR->subRun()) : "invalid");
512 :
513 : // Process Run Aux
514 0 : TLOG(TLVL_DEBUG + 37, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
515 0 : << "processing Run auxiliary ...";
516 :
517 0 : run_aux.reset(ReadObjectAny<art::RunAuxiliary>(msg, "art::RunAuxiliary", "ArtdaqInputHelper::readAndConstructPrincipal"));
518 0 : run_aux->setProcessHistoryID(history_to_use_->processHistoryID());
519 0 : printProcessHistoryID("readAndConstructPrincipal", run_aux.get());
520 :
521 0 : TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
522 0 : << "inR: " << static_cast<void*>(inR) << " run/expected "
523 0 : << (inR ? std::to_string(inR->run()) : "invalid") << "/" << run_aux->run();
524 :
525 0 : if ((inR == nullptr) || !inR->runID().isValid() || (inR->run() != run_aux->run()))
526 : {
527 : // New run, either we have no input RunPrincipal, or the
528 : // input run number does not match the run number.
529 0 : TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: making RunPrincipal ...";
530 0 : outR = pm_.makeRunPrincipal(*run_aux);
531 : }
532 :
533 0 : TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
534 0 : << "finished processing Run auxiliary.";
535 :
536 0 : if (msg_type_code != artdaq::NetMonHeader::MessageType::Run) // SubRun or Event
537 : {
538 0 : TLOG(TLVL_DEBUG + 38, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
539 0 : << "processing SubRun auxiliary ...";
540 :
541 0 : subrun_aux.reset(
542 0 : ReadObjectAny<art::SubRunAuxiliary>(msg, "art::SubRunAuxiliary", "ArtdaqInputHelper::readAndConstructPrincipal"));
543 0 : printProcessHistoryID("readAndConstructPrincipal", subrun_aux.get());
544 :
545 : // HACK! Make the SR PHID match!
546 0 : printProcessHistoryID("readAndConstructPrincipal", subrun_aux.get());
547 0 : subrun_aux->setProcessHistoryID(run_aux->processHistoryID());
548 0 : printProcessHistoryID("readAndConstructPrincipal", subrun_aux.get());
549 :
550 0 : TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
551 0 : << "inSR: " << static_cast<void*>(inSR) << " run/expected "
552 0 : << (inSR ? std::to_string(inSR->run()) : "invalid") << "/" << subrun_aux->run()
553 0 : << ", subrun/expected " << (inSR ? std::to_string(inSR->subRun()) : "invalid") << "/"
554 0 : << subrun_aux->subRun();
555 :
556 0 : art::SubRunID subrun_check(subrun_aux->run(), subrun_aux->subRun());
557 0 : if (inSR == nullptr || !inSR->subRunID().isValid() || subrun_check != inSR->subRunID())
558 : {
559 : // New SubRun, either we have no input SubRunPrincipal, or the
560 : // input subRun number does not match the subRun number.
561 0 : TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
562 0 : << "making SubRunPrincipal ...";
563 0 : outSR = pm_.makeSubRunPrincipal(*subrun_aux);
564 : }
565 :
566 0 : TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
567 0 : << "finished processing SubRun auxiliary.";
568 : }
569 :
570 0 : if (msg_type_code == artdaq::NetMonHeader::MessageType::Event)
571 : { // Event message.
572 :
573 0 : TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
574 0 : << "processing Event auxiliary ...";
575 :
576 0 : event_aux.reset(
577 0 : ReadObjectAny<art::EventAuxiliary>(msg, "art::EventAuxiliary", "ArtdaqInputHelper::readAndConstructPrincipal"));
578 :
579 0 : TLOG(TLVL_DEBUG + 34, "ArtdaqInputHelper") << "readAndConstructPrincipal: making EventPrincipal ...";
580 : #if ART_HEX_VERSION < 0x31100
581 : auto historyPtr = std::unique_ptr<art::History>(new History(*(history_to_use_.get())));
582 : if (!art::ProcessHistoryRegistry::get().count(history_to_use_->processHistoryID()))
583 : {
584 : TLOG_WARNING("ArtdaqInputHelper") << "Stored history is not in ProcessHistoryRegistry, this event may have issues!";
585 : }
586 : outE = pm_.makeEventPrincipal(*event_aux, std::move(historyPtr));
587 : #else
588 0 : if (!art::ProcessHistoryRegistry::get().count(history_to_use_->processHistoryID()))
589 : {
590 0 : TLOG_WARNING("ArtdaqInputHelper") << "Stored history is not in ProcessHistoryRegistry, this event may have issues!";
591 : }
592 0 : event_aux->setProcessHistoryID(history_to_use_->processHistoryID());
593 0 : outE = pm_.makeEventPrincipal(*event_aux);
594 : #endif
595 0 : TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
596 0 : << "finished processing Event auxiliary.";
597 : }
598 0 : else if (msg_type_code == artdaq::NetMonHeader::MessageType::Subrun)
599 : {
600 0 : if (outSR == nullptr)
601 : {
602 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "SubrunDataFragment for current Subrun received, returning Flush event";
603 0 : art::EventID const evid(art::EventID::flushEvent(inSR->subRunID()));
604 0 : outE = pm_.makeEventPrincipal(evid, currentTime);
605 : }
606 : }
607 0 : else if (msg_type_code == artdaq::NetMonHeader::MessageType::Run)
608 : {
609 0 : if (outR == nullptr)
610 : {
611 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "RunDataFragment for current Run received, returning Flush subrun/event";
612 0 : art::EventID const evid(art::EventID::flushEvent(inR->runID()));
613 0 : if (inSR == nullptr || inSR->subRunID() != evid.subRunID())
614 : {
615 0 : outSR = pm_.makeSubRunPrincipal(evid.subRunID(), currentTime);
616 : }
617 0 : outE = pm_.makeEventPrincipal(evid, currentTime);
618 : }
619 : }
620 0 : }
621 :
622 : template<typename U>
623 0 : bool art::ArtdaqInputHelper<U>::constructPrincipal(std::shared_ptr<ArtdaqEvent> eventPtr, art::RunPrincipal* const inR, art::SubRunPrincipal* const inSR, art::RunPrincipal*& outR, art::SubRunPrincipal*& outSR, art::EventPrincipal*& outE)
624 : {
625 : // We return false, indicating we're done reading, if:
626 : // 1) we did not obtain an event, because we timed out and were
627 : // configured NOT to keep trying after a timeout, or
628 : // 2) the event we read was the end-of-data marker: a null
629 : // pointer
630 0 : if (eventPtr->FirstFragmentType() == artdaq::Fragment::EndOfDataFragmentType)
631 : {
632 0 : TLOG(TLVL_DEBUG + 32, "ArtdaqInputHelper") << "Received shutdown message, returning false";
633 0 : shutdownMsgReceived_ = true;
634 0 : return false;
635 : }
636 :
637 0 : if (!eventPtr->header)
638 : {
639 0 : TLOG_ERROR("ArtdaqInputHelper") << "No RawEventHeader received, cannot construct principals!";
640 0 : shutdownMsgReceived_ = true;
641 0 : return false;
642 : }
643 :
644 : // Check the number of fragments in the RawEvent. If we have a single
645 : // fragment and that fragment is marked as EndRun or EndSubrun we'll create
646 : // the special principals for that.
647 0 : art::Timestamp currentTime = 0;
648 : timespec hi_res_time;
649 0 : int retcode = clock_gettime(CLOCK_REALTIME, &hi_res_time);
650 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "hi_res_time tv_sec = " << hi_res_time.tv_sec
651 0 : << " tv_nsec = " << hi_res_time.tv_nsec << " (retcode = " << retcode << ")";
652 0 : if (retcode == 0)
653 : {
654 0 : currentTime = ((hi_res_time.tv_sec & 0xffffffff) << 32) | (hi_res_time.tv_nsec & 0xffffffff);
655 : }
656 : else
657 : {
658 0 : TLOG_ERROR("ArtdaqInputHelper")
659 0 : << "Unable to fetch a high-resolution time with clock_gettime for art::Event Timestamp. "
660 0 : << "The art::Event Timestamp will be zero for event " << eventPtr->header->event_id;
661 : }
662 :
663 : // make new run if inR is 0 or if the run has changed
664 0 : if (inR == nullptr || !inR->runID().isValid() || inR->run() != eventPtr->header->run_id)
665 : {
666 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "Making run principal with run_id " << eventPtr->header->run_id;
667 0 : outR = pm_.makeRunPrincipal(eventPtr->header->run_id, currentTime);
668 : }
669 :
670 0 : if (eventPtr->FirstFragmentType() == artdaq::Fragment::EndOfRunFragmentType)
671 : {
672 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "EndOfRunFragment received, returning Flush subrun/event";
673 0 : art::EventID const evid(art::EventID::flushEvent(outR != nullptr ? outR->runID() : inR->runID()));
674 0 : if (inSR == nullptr || inSR->subRunID() != evid.subRunID())
675 : {
676 0 : outSR = pm_.makeSubRunPrincipal(evid.subRunID(), currentTime);
677 : }
678 0 : outE = pm_.makeEventPrincipal(evid, currentTime);
679 0 : return true;
680 : }
681 :
682 : // make new subrun if inSR is 0 or if the subrun has changed
683 0 : art::SubRunID subrun_check(eventPtr->header->run_id, eventPtr->header->subrun_id);
684 0 : if (inSR == nullptr || !inSR->subRunID().isValid() || subrun_check != inSR->subRunID())
685 : {
686 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "Making subrun principal with subrun_id " << eventPtr->header->subrun_id;
687 0 : outSR = pm_.makeSubRunPrincipal(eventPtr->header->run_id, eventPtr->header->subrun_id, currentTime);
688 : }
689 :
690 0 : if (eventPtr->FirstFragmentType() == artdaq::Fragment::EndOfSubrunFragmentType)
691 : {
692 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "EndOfSubrunFragment received, returning Flush event";
693 0 : art::EventID const evid(art::EventID::flushEvent(outSR != nullptr ? outSR->subRunID() : inSR->subRunID()));
694 0 : outE = pm_.makeEventPrincipal(evid, currentTime);
695 0 : return true;
696 : }
697 :
698 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "Making event principal with event_id " << eventPtr->header->event_id;
699 0 : outE = pm_.makeEventPrincipal(eventPtr->header->run_id, eventPtr->header->subrun_id, eventPtr->header->event_id, currentTime);
700 0 : return true;
701 : }
702 :
703 : template<typename U>
704 : template<class T>
705 0 : void art::ArtdaqInputHelper<U>::readDataProducts(std::list<std::unique_ptr<TBufferFile>>& msgs, T* const& outPrincipal)
706 : {
707 0 : for (auto& msg : msgs)
708 : {
709 0 : ULong_t prd_cnt = 0;
710 : {
711 0 : TLOG(TLVL_DEBUG + 40, "ArtdaqInputHelper") << "readDataProducts: reading data product count ...";
712 0 : msg->ReadULong(prd_cnt);
713 0 : TLOG(TLVL_DEBUG + 40, "ArtdaqInputHelper") << "readDataProducts: product count: " << prd_cnt;
714 : }
715 : //
716 : // Read the data products.
717 : //
718 0 : for (ULong_t I = 0; I < prd_cnt; ++I)
719 : {
720 0 : std::unique_ptr<BranchKey> bk;
721 : {
722 0 : TLOG(TLVL_DEBUG + 40, "ArtdaqInputHelper") << "readDataProducts: Reading branch key.";
723 0 : bk.reset(ReadObjectAny<BranchKey>(msg, "art::BranchKey", "ArtdaqInputHelper::readDataProducts"));
724 : }
725 :
726 : #ifndef __OPTIMIZE__
727 0 : TLOG(TLVL_DEBUG + 41, "ArtdaqInputHelper") << "readDataProducts: got product class: '" << bk->friendlyClassName_ << "' modlbl: '"
728 0 : << bk->moduleLabel_ << "' instnm: '" << bk->productInstanceName_ << "' procnm: '"
729 0 : << bk->processName_;
730 : #endif
731 0 : ProductList::const_iterator iter;
732 : {
733 0 : TLOG(TLVL_DEBUG + 40, "ArtdaqInputHelper") << "readDataProducts: looking up product ...";
734 0 : iter = productList_->find(*bk);
735 0 : if (iter == productList_->end())
736 : {
737 0 : throw art::Exception(art::errors::ProductNotFound) // NOLINT(cert-err60-cpp)
738 : << "No product is registered for\n"
739 0 : << " process name: '" << bk->processName_ << "'\n"
740 0 : << " module label: '" << bk->moduleLabel_ << "'\n"
741 0 : << " product friendly class name: '" << bk->friendlyClassName_ << "'\n"
742 0 : << " product instance name: '" << bk->productInstanceName_ << "'\n";
743 : }
744 : }
745 : // Note: This must be a reference to the unique copy in
746 : // the master product registry!
747 0 : const BranchDescription& bd = iter->second;
748 0 : std::unique_ptr<EDProduct> prd;
749 : {
750 0 : TLOG(TLVL_DEBUG + 40, "ArtdaqInputHelper") << "readDataProducts: Reading product with wrapped name: " << bd.wrappedName()
751 0 : << ", TClass = " << static_cast<void*>(TClass::GetClass(bd.wrappedName().c_str()));
752 :
753 : // JCF, May-25-2016
754 : // Currently unclear why the templatized version of ReadObjectAny doesn't work here...
755 :
756 : // prd.reset(ReadObjectAny<EDProduct>(msg, bd.wrappedName()));
757 :
758 0 : void* p = msg->ReadObjectAny(TClass::GetClass(bd.wrappedName().c_str()));
759 0 : auto pp = reinterpret_cast<EDProduct*>(p); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
760 :
761 0 : TLOG(TLVL_DEBUG + 40, "ArtdaqInputHelper") << "readDataProducts: After ReadObjectAny(prd): p=" << p << ", EDProduct::isPresent: " << pp->isPresent();
762 0 : prd.reset(pp);
763 0 : p = nullptr;
764 : }
765 0 : std::unique_ptr<const ProductProvenance> prdprov;
766 : {
767 0 : TLOG(TLVL_DEBUG + 40, "ArtdaqInputHelper") << "readDataProducts: Reading product provenance.";
768 0 : prdprov.reset(ReadObjectAny<ProductProvenance>(msg, "art::ProductProvenance", "ArtdaqInputHelper::readDataProducts"));
769 : }
770 :
771 : {
772 0 : TLOG(TLVL_DEBUG + 40, "ArtdaqInputHelper") << "readDataProducts: inserting product: class: '" << bd.friendlyClassName()
773 0 : << "' modlbl: '" << bd.moduleLabel() << "' instnm: '" << bd.productInstanceName()
774 0 : << "' procnm: '" << bd.processName() << "' id: '" << bd.productID() << "'";
775 0 : putInPrincipal(outPrincipal, std::move(prd), bd, std::move(prdprov));
776 : }
777 : }
778 : }
779 0 : }
780 :
781 : template<typename U>
782 0 : void art::ArtdaqInputHelper<U>::putInPrincipal(RunPrincipal* const& rp, std::unique_ptr<EDProduct>&& prd,
783 : const BranchDescription& bd,
784 : std::unique_ptr<const ProductProvenance>&& prdprov)
785 : {
786 0 : rp->put(bd, std::move(prdprov), std::move(prd), std::make_unique<RangeSet>(RangeSet::forRun(rp->runID())));
787 0 : }
788 :
789 : template<typename U>
790 0 : void art::ArtdaqInputHelper<U>::putInPrincipal(SubRunPrincipal* const& srp, std::unique_ptr<EDProduct>&& prd,
791 : const BranchDescription& bd,
792 : std::unique_ptr<const ProductProvenance>&& prdprov)
793 : {
794 0 : srp->put(bd, std::move(prdprov), std::move(prd), std::make_unique<RangeSet>(RangeSet::forSubRun(srp->subRunID())));
795 0 : }
796 :
797 : template<typename U>
798 0 : void art::ArtdaqInputHelper<U>::putInPrincipal(EventPrincipal* const& ep, std::unique_ptr<EDProduct>&& prd,
799 : const BranchDescription& bd,
800 : std::unique_ptr<const ProductProvenance>&& prdprov)
801 : {
802 0 : TLOG(TLVL_DEBUG + 42, "ArtdaqInputHelper") << "EventPrincipal size before put: " << ep->size();
803 :
804 0 : ep->put(bd, std::move(prdprov), std::move(prd), std::make_unique<RangeSet>(RangeSet::invalid()));
805 :
806 0 : TLOG(TLVL_DEBUG + 42, "ArtdaqInputHelper") << "EventPrincipal size after put: " << ep->size();
807 0 : }
808 :
809 : template<typename U>
810 0 : std::pair<bool, bool> art::ArtdaqInputHelper<U>::readFragments(std::unordered_map<artdaq::Fragment::type_t, std::unique_ptr<artdaq::Fragments>> const& eventMap, art::RunPrincipal* const theRun, art::SubRunPrincipal* const theSubRun, art::EventPrincipal* const theEvent)
811 : {
812 : // Now read in Fragments
813 0 : double fragmentLatency = 0;
814 0 : double fragmentLatencyMax = 0.0;
815 0 : size_t fragmentCount = 0;
816 :
817 0 : bool eventProductsRead = false;
818 0 : bool subrunProductsRead = false;
819 :
820 0 : art::ServiceHandle<ArtdaqFragmentNamingServiceInterface> translator;
821 :
822 : // insert the Fragments of each type into the EventPrincipal
823 0 : for (auto& fragmentTypePair : eventMap)
824 : {
825 0 : auto type_code = fragmentTypePair.first;
826 0 : if (artdaq::Fragment::isSystemFragmentType(type_code) && type_code != artdaq::Fragment::ContainerFragmentType && type_code != artdaq::Fragment::EmptyFragmentType)
827 : {
828 0 : if (type_code == artdaq::Fragment::EndOfRunFragmentType || type_code == artdaq::Fragment::EndOfSubrunFragmentType)
829 : {
830 0 : std::unique_ptr<std::vector<artdaq::ArtdaqMetadata>> metadata_coll = std::make_unique<std::vector<artdaq::ArtdaqMetadata>>();
831 0 : for (auto& frag : *fragmentTypePair.second)
832 : {
833 0 : artdaq::MetadataFragment mf(frag);
834 0 : auto md = mf.get_metadata();
835 :
836 0 : TLOG(TLVL_DEBUG + 44, "ArtdaqInputHelper") << "Adding ArtdaqMetadata " << frag.fragmentID() << " to storage (sz=" << metadata_coll->size() + 1 << ")";
837 0 : metadata_coll->push_back(md);
838 : }
839 :
840 0 : TLOG(TLVL_DEBUG + 44, "ArtdaqInputHelper") << "Adding " << metadata_coll->size() << " ArtdaqMetadatas to Run.";
841 0 : if (type_code == artdaq::Fragment::EndOfRunFragmentType)
842 : {
843 0 : put_product_in_principal(std::move(metadata_coll), *theRun, pretend_module_name, "ArtdaqMetadata");
844 : }
845 0 : else if (type_code == artdaq::Fragment::EndOfSubrunFragmentType)
846 : {
847 0 : put_product_in_principal(std::move(metadata_coll), *theSubRun, pretend_module_name, "ArtdaqMetadata");
848 : }
849 0 : }
850 : else
851 : {
852 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "Skipping system Fragment with type " << static_cast<int>(type_code) << " ( " << translator->GetInstanceNameForType(type_code) << " )";
853 : }
854 0 : continue;
855 0 : }
856 0 : TLOG(TLVL_DEBUG + 33, "ArtdaqInputHelper") << "type is " << static_cast<int>(type_code) << ", number of fragments is " << fragmentTypePair.second->size();
857 :
858 0 : std::unordered_map<std::string, std::unique_ptr<artdaq::Fragments>> derived_fragments;
859 0 : for (auto& frag : *fragmentTypePair.second)
860 : {
861 0 : TLOG(TLVL_DEBUG + 44, "ArtdaqInputHelper") << "Processing Fragment with ID " << frag.fragmentID();
862 0 : bytesRead += frag.sizeBytes();
863 0 : auto latency_s = frag.getLatency(true);
864 0 : double latency = latency_s.tv_sec + (latency_s.tv_nsec / 1000000000.0);
865 :
866 0 : fragmentLatency += latency;
867 0 : fragmentCount++;
868 0 : if (latency > fragmentLatencyMax) fragmentLatencyMax = latency;
869 :
870 0 : std::pair<bool, std::string> instance_name_result =
871 : translator->GetInstanceNameForFragment(frag);
872 0 : std::string label = instance_name_result.second;
873 0 : if (!instance_name_result.first)
874 : {
875 0 : TLOG_WARNING("ArtdaqInputHelper")
876 0 : << "UnknownFragmentType: The product instance name mapping for fragment type \"" << static_cast<int>(type_code)
877 0 : << "\" is not known. Fragments of this "
878 0 : << "type will be stored in the event with an instance name of \"" << label << "\".";
879 : }
880 0 : if (!derived_fragments.count(label))
881 : {
882 0 : TLOG(TLVL_DEBUG + 44, "ArtdaqInputHelper") << "Creating output Fragment storage for label " << label;
883 0 : derived_fragments[label] = std::make_unique<artdaq::Fragments>();
884 : }
885 0 : TLOG(TLVL_DEBUG + 44, "ArtdaqInputHelper") << "Adding Fragment " << frag.fragmentID() << " to storage with label " << label << " (sz=" << derived_fragments[label]->size() + 1 << ")";
886 0 : derived_fragments[label]->emplace_back(std::move(frag));
887 : }
888 0 : for (auto& type : derived_fragments)
889 : {
890 0 : TLOG(TLVL_DEBUG + 44, "ArtdaqInputHelper") << "Adding " << type.second->size() << " Fragments with label " << type.first << " to event.";
891 0 : put_product_in_principal(std::move(type.second), *theEvent, pretend_module_name, type.first);
892 0 : eventProductsRead = true;
893 : }
894 : }
895 :
896 0 : if (metricMan)
897 : {
898 0 : metricMan->sendMetric("bytesRead", bytesRead, "B", 3, artdaq::MetricMode::LastPoint);
899 :
900 0 : metricMan->sendMetric("ArtdaqInputHelper Latency", fragmentLatency / fragmentCount, "s", 4, artdaq::MetricMode::Average);
901 0 : metricMan->sendMetric("ArtdaqInputHelper Maximum Latency", fragmentLatencyMax, "s", 4, artdaq::MetricMode::Maximum);
902 : }
903 :
904 0 : return std::make_pair(subrunProductsRead, eventProductsRead);
905 : }
906 :
907 : template<typename U>
908 0 : bool art::ArtdaqInputHelper<U>::readNext(art::RunPrincipal* const inR, art::SubRunPrincipal* const inSR,
909 : art::RunPrincipal*& outR, art::SubRunPrincipal*& outSR, art::EventPrincipal*& outE)
910 : {
911 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "Begin: ArtdaqInputHelper::readNext";
912 :
913 0 : if (shutdownMsgReceived_)
914 : {
915 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "Shutdown message already received, returning false";
916 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "End: ArtdaqInputHelper::readNext";
917 0 : return false;
918 : }
919 :
920 0 : auto read_start_time = std::chrono::steady_clock::now();
921 :
922 0 : std::shared_ptr<ArtdaqEvent> eventMap = communicationWrapper_.receiveMessages();
923 0 : auto got_event_time = std::chrono::steady_clock::now();
924 :
925 0 : if (eventMap == nullptr)
926 : {
927 0 : TLOG(TLVL_ERROR, "ArtdaqInputHelper") << "No Fragments received! Aborting...";
928 0 : shutdownMsgReceived_ = true;
929 0 : TLOG(TLVL_DEBUG + 45, "ArtdaqInputHelper") << "End: ArtdaqInputHelper::readNext";
930 0 : return false;
931 : }
932 :
933 0 : if (eventMap->FirstFragmentType() == artdaq::Fragment::EndOfDataFragmentType)
934 : {
935 0 : TLOG(TLVL_DEBUG + 45, "ArtdaqInputHelper") << "Shutdown message received!";
936 0 : shutdownMsgReceived_ = true;
937 0 : TLOG(TLVL_DEBUG + 45, "ArtdaqInputHelper") << "End: ArtdaqInputHelper::readNext";
938 0 : return false;
939 : }
940 :
941 0 : if (eventMap->FirstFragmentType() == artdaq::Fragment::InitFragmentType)
942 : {
943 : #if CAN_REINIT
944 : TLOG(TLVL_INFO, "ArtdaqInputHelper") << "Additional Init Message received! Attempting to register new products...";
945 : readInitMessage();
946 : #else
947 0 : TLOG(TLVL_WARNING, "ArtdaqInputHelper") << "Received additional Init Message! Check init_fragment_count configuration!";
948 : #endif
949 0 : TLOG(TLVL_DEBUG + 45, "ArtdaqInputHelper") << "End: ArtdaqInputHelper::readNext";
950 0 : return true;
951 : }
952 :
953 0 : if (fragmentsOnlyMode_)
954 : {
955 0 : if (eventMap->fragments.count(artdaq::Fragment::DataFragmentType) || eventMap->fragments.count(artdaq::Fragment::RunDataFragmentType) || eventMap->fragments.count(artdaq::Fragment::SubrunDataFragmentType))
956 : {
957 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "ArtdaqInputHelper::readNext unexpectedly got a message with a DataFragment. This art Event will NOT be reconstructed!";
958 : }
959 :
960 0 : auto firstFragmentType = eventMap->FirstFragmentType();
961 0 : TLOG(TLVL_DEBUG + 32, "ArtdaqInputHelper") << "First Fragment type is " << static_cast<int>(firstFragmentType);
962 0 : if (constructPrincipal(eventMap, inR, inSR, outR, outSR, outE))
963 : {
964 0 : readFragments(eventMap->fragments, outR ? outR : inR, outSR ? outSR : inSR, outE);
965 : }
966 : }
967 : else
968 : {
969 0 : std::list<std::unique_ptr<TBufferFile>> msgs;
970 0 : if (eventMap->fragments.count(artdaq::Fragment::RunDataFragmentType))
971 0 : for (auto& dataFrag : *(eventMap->fragments[artdaq::Fragment::RunDataFragmentType]))
972 : {
973 0 : if (!dataFrag.hasMetadata()) continue;
974 0 : auto header = dataFrag.metadata<artdaq::NetMonHeader>();
975 0 : msgs.emplace_back(new TBufferFile(TBuffer::kRead, header->data_length, dataFrag.dataBegin(), kFALSE, nullptr));
976 : }
977 0 : if (eventMap->fragments.count(artdaq::Fragment::SubrunDataFragmentType))
978 0 : for (auto& dataFrag : *(eventMap->fragments[artdaq::Fragment::SubrunDataFragmentType]))
979 : {
980 0 : if (!dataFrag.hasMetadata()) continue;
981 0 : auto header = dataFrag.metadata<artdaq::NetMonHeader>();
982 0 : msgs.emplace_back(new TBufferFile(TBuffer::kRead, header->data_length, dataFrag.dataBegin(), kFALSE, nullptr));
983 : }
984 0 : if (eventMap->fragments.count(artdaq::Fragment::DataFragmentType))
985 0 : for (auto& dataFrag : *(eventMap->fragments[artdaq::Fragment::DataFragmentType]))
986 : {
987 0 : if (!dataFrag.hasMetadata()) continue;
988 0 : auto header = dataFrag.metadata<artdaq::NetMonHeader>();
989 0 : msgs.emplace_back(new TBufferFile(TBuffer::kRead, header->data_length, dataFrag.dataBegin(), kFALSE, nullptr));
990 : }
991 :
992 : //
993 : // Read message type code.
994 : //
995 0 : artdaq::NetMonHeader::MessageType msg_type_code = artdaq::NetMonHeader::MessageType::Invalid;
996 0 : ULong_t msg_type_code_tmp = 0;
997 0 : for (auto& msg : msgs)
998 : {
999 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "ArtdaqInputHelper::readNext: "
1000 0 : << "getting message type code ...";
1001 0 : msg->ReadULong(msg_type_code_tmp);
1002 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "ArtdaqInputHelper::readNext: "
1003 0 : << "message type: " << msg_type_code_tmp;
1004 :
1005 0 : if (msg_type_code == artdaq::NetMonHeader::MessageType::Invalid)
1006 0 : msg_type_code = static_cast<artdaq::NetMonHeader::MessageType>(msg_type_code_tmp);
1007 0 : else if (msg_type_code != static_cast<artdaq::NetMonHeader::MessageType>(msg_type_code_tmp))
1008 : {
1009 0 : TLOG(TLVL_ERROR, "ArtdaqInputHelper") << "ArtdaqInputHelper::readNext: Received conflicting message type codes! Aborting...";
1010 :
1011 0 : shutdownMsgReceived_ = true;
1012 0 : TLOG(TLVL_DEBUG + 45, "ArtdaqInputHelper") << "End: ArtdaqInputHelper::readNext";
1013 0 : return false;
1014 : }
1015 : }
1016 :
1017 0 : for (auto& msg : msgs)
1018 : {
1019 0 : readAndConstructPrincipal(msg, msg_type_code, inR, inSR, outR, outSR, outE);
1020 : }
1021 : //
1022 : // Read per-event metadata needed to construct principal.
1023 : //
1024 0 : if (msg_type_code == artdaq::NetMonHeader::MessageType::Run)
1025 : {
1026 : // EndRun message.
1027 : // FIXME: We need to merge these into the input RunPrincipal.
1028 0 : readDataProducts(msgs, outR ? outR : inR);
1029 : }
1030 0 : else if (msg_type_code == artdaq::NetMonHeader::MessageType::Subrun)
1031 : {
1032 : // FIXME: We need to merge these into the input SubRunPrincipal.
1033 0 : readDataProducts(msgs, outSR ? outSR : inSR);
1034 : }
1035 0 : else if (msg_type_code == artdaq::NetMonHeader::MessageType::Event)
1036 : {
1037 : // Event message.
1038 0 : readDataProducts(msgs, outE);
1039 :
1040 0 : if (eventMap->fragments.size() > 1)
1041 : {
1042 0 : readFragments(eventMap->fragments, outR ? outR : inR, outSR ? outSR : inSR, outE);
1043 : }
1044 : }
1045 : else
1046 : {
1047 : // Did not have a valid message, try again
1048 0 : return false;
1049 : }
1050 0 : }
1051 :
1052 0 : if (outE != nullptr)
1053 : {
1054 0 : auto artHdrPtr = std::make_unique<artdaq::detail::RawEventHeader>();
1055 0 : auto daqHdrPtr = eventMap->header;
1056 :
1057 0 : if (daqHdrPtr != nullptr)
1058 : {
1059 0 : memcpy(artHdrPtr.get(), daqHdrPtr.get(), sizeof(artdaq::detail::RawEventHeader));
1060 0 : put_product_in_principal(std::move(artHdrPtr), *outE, pretend_module_name, "RawEventHeader");
1061 : }
1062 0 : }
1063 :
1064 0 : auto read_finish_time = std::chrono::steady_clock::now();
1065 0 : TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "readNext: bytesRead=" << bytesRead
1066 0 : << " metricMan=" << static_cast<void*>(metricMan.get());
1067 0 : if (metricMan)
1068 : {
1069 0 : metricMan->sendMetric("Avg Processing Time", artdaq::TimeUtils::GetElapsedTime(last_read_time, read_start_time),
1070 : "s", 2, artdaq::MetricMode::Average);
1071 0 : metricMan->sendMetric("Avg Input Wait Time", artdaq::TimeUtils::GetElapsedTime(read_start_time, got_event_time),
1072 : "s", 3, artdaq::MetricMode::Average);
1073 0 : metricMan->sendMetric("Avg Read Time", artdaq::TimeUtils::GetElapsedTime(got_event_time, read_finish_time), "s",
1074 : 3, artdaq::MetricMode::Average);
1075 : }
1076 :
1077 0 : TLOG(TLVL_DEBUG + 48, "ArtdaqInputHelper") << "End: ArtdaqInputHelper::readNext ret=" << std::boolalpha << (outR || outSR || outE);
1078 0 : last_read_time = std::chrono::steady_clock::now();
1079 0 : return outR || outSR || outE;
1080 0 : }
1081 :
1082 : #endif // ARTDAQ_ARTDAQ_ARTMODULES_ARTDAQINPUTHELPER_HH_
|
