Line data Source code
1 : #ifndef ARTDAQ_ARTDAQ_ARTMODULES_ARTDAQOUTPUT_HH_
2 : #define ARTDAQ_ARTDAQ_ARTMODULES_ARTDAQOUTPUT_HH_
3 :
4 : #include "TRACE/tracemf.h" // Pre-empt TRACE/trace.h from Fragment.hh.
5 : #include "artdaq-core/Data/Fragment.hh"
6 :
7 : #include "artdaq-core/Data/RawEvent.hh"
8 : #include "artdaq-core/Data/detail/ParentageMap.hh"
9 : #include "artdaq/DAQdata/Globals.hh"
10 : #include "artdaq/DAQdata/NetMonHeader.hh"
11 :
12 : #include "art/Framework/Core/OutputModule.h"
13 : #include "art/Framework/Principal/EventPrincipal.h"
14 : #include "art/Framework/Principal/OutputHandle.h"
15 : #include "art/Framework/Principal/RunPrincipal.h"
16 : #include "art/Framework/Principal/SubRunPrincipal.h"
17 : #include "art/Persistency/Provenance/ModuleContext.h"
18 : #include "art/Persistency/Provenance/ProcessHistoryRegistry.h"
19 : #include "art_root_io/setup.h"
20 :
21 : #include "canvas/Persistency/Common/WrappedTypeID.h"
22 : #include "canvas/Persistency/Provenance/BranchDescription.h"
23 : #include "canvas/Persistency/Provenance/BranchKey.h"
24 : #include "canvas/Persistency/Provenance/ParentageRegistry.h"
25 : #include "canvas/Persistency/Provenance/ProcessConfiguration.h"
26 : #include "canvas/Persistency/Provenance/ProcessConfigurationID.h"
27 : #include "canvas/Persistency/Provenance/ProcessHistoryID.h"
28 : #include "canvas/Persistency/Provenance/ProductList.h"
29 : #include "canvas/Persistency/Provenance/ProductProvenance.h"
30 : #include "canvas/Persistency/Provenance/RunAuxiliary.h"
31 : #include "canvas/Persistency/Provenance/SubRunAuxiliary.h"
32 : #include "canvas/Utilities/Exception.h"
33 : #include "fhiclcpp/ParameterSet.h"
34 : #include "fhiclcpp/ParameterSetRegistry.h"
35 :
36 : #include <TBufferFile.h>
37 : #include <TClass.h>
38 : #include <TList.h>
39 : #include <TStreamerInfo.h>
40 :
41 : #include <unistd.h>
42 : #include <algorithm>
43 : #include <iomanip>
44 : #include <iostream>
45 : #include <iterator>
46 : #include <memory>
47 : #include <mutex>
48 : #include <sstream>
49 : #include <string>
50 : #include <vector>
51 :
52 : #define TLVL_ENTER_EXIT 33
53 : #define TLVL_SENDINIT 39
54 : #define TLVL_SENDINIT_VERBOSE1 40
55 : #define TLVL_SENDINIT_VERBOSE2 41
56 : #define TLVL_WRITEDATAPRODUCTS 42
57 : #define TLVL_WRITEDATAPRODUCTS_VERBOSE 43
58 : #define TLVL_WRITE 44
59 : #define TLVL_WRITERUN 45
60 : #define TLVL_WRITERUN_VERBOSE 46
61 : #define TLVL_WRITESUBRUN 47
62 : #define TLVL_WRITESUBRUN_VERBOSE 48
63 : #define TLVL_EXTRACTPRODUCTS 49
64 : #define TLVL_EXTRACTPRODUCTS_VERBOSE 50
65 : #define TLVL_BEGINJOB 51
66 : #define TLVL_BEGINJOB_VERBOSE 52
67 :
68 : namespace art {
69 : class ArtdaqOutput;
70 : } // namespace art
71 :
72 : static artdaq::FragmentPtr outputFrag = nullptr;
73 : static std::mutex outputFragMutex;
74 0 : inline char* Fragment_ReAllocChar(char* dataPtr, size_t size, size_t /*oldsize*/)
75 : {
76 0 : if (outputFrag != nullptr && dataPtr == reinterpret_cast<char*>(outputFrag->dataBegin())) // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
77 : {
78 0 : outputFrag->resizeBytes(size);
79 :
80 0 : return reinterpret_cast<char*>(outputFrag->dataBegin()); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
81 : }
82 0 : return nullptr;
83 : }
84 :
85 : /// <summary>
86 : /// This is the base class for artdaq OutputModules, providing the serialization interface for art Events.
87 : /// </summary>
88 : class art::ArtdaqOutput : public art::OutputModule
89 : {
90 : public:
91 : /// <summary>
92 : /// ArtdaqOutput Constructor
93 : /// </summary>
94 : /// <param name="ps">ParameterSet used to configure art::OutputModule</param>
95 0 : explicit ArtdaqOutput(fhicl::ParameterSet const& ps)
96 0 : : OutputModule(ps), productList_(), raw_data_label_(ps.get<std::string>("raw_data_label", "daq"))
97 : {
98 0 : root::setup();
99 0 : }
100 :
101 : /// <summary>
102 : /// Destructor
103 : /// </summary>
104 0 : virtual ~ArtdaqOutput() = default;
105 :
106 : protected:
107 : /// <summary>
108 : /// Perform actions necessary for opening files. No-op, but derived classes may override
109 : /// </summary>
110 0 : virtual void openFile(FileBlock const&)
111 : {
112 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "Begin/End: ArtdaqOutput::openFile(const FileBlock&)";
113 0 : }
114 :
115 : /// <summary>
116 : /// Perform actions necessary for closing files. No-op, but derived classes may override
117 : /// </summary>
118 0 : virtual void closeFile() { TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "Begin/End: ArtdaqOutput::closeFile()"; }
119 :
120 : /// <summary>
121 : /// Perform actions necessary after closing the input file. No-op, but derived classes may override
122 : /// </summary>
123 0 : virtual void respondToCloseInputFile(FileBlock const&)
124 : {
125 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "Begin/End: ArtdaqOutput::"
126 0 : "respondToCloseOutputFiles(FileBlock const&)";
127 0 : }
128 :
129 : /// <summary>
130 : /// Perform actions necessary after closing the output file(s). No-op, but derived classes may override
131 : /// </summary>
132 0 : virtual void respondToCloseOutputFiles(FileBlock const&)
133 : {
134 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "Begin/End: ArtdaqOutput::"
135 0 : "respondToCloseOutputFiles(FileBlock const&)";
136 0 : }
137 :
138 : virtual void beginJob();
139 :
140 : /// <summary>
141 : /// Perform End-of-Job actions. No-op, but derived classes may override
142 : /// </summary>
143 0 : virtual void endJob()
144 : {
145 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "Begin/End: ArtdaqOutput::endJob()";
146 0 : }
147 :
148 : /// <summary>
149 : /// Perform Begin Run actions. Derived classes should implement beginRun_ instead.
150 : /// </summary>
151 : /// <param name="rp">RunPrincipal of new run</param>
152 0 : void beginRun(RunPrincipal const& rp) final
153 : {
154 0 : if (!initMsgSent_)
155 : {
156 0 : art::ProcessHistory ph = rp.processHistory();
157 0 : ph.push_back(moduleDescription().processConfiguration());
158 0 : art::ProcessHistoryRegistry::emplace(ph.id(), ph);
159 0 : send_init_message();
160 0 : }
161 0 : beginRun_(rp);
162 0 : }
163 : /// <summary>
164 : /// Perform Begin Run actions. No-op, but derived classes may override
165 : /// </summary>
166 0 : virtual void beginRun_(RunPrincipal const&) {}
167 :
168 : /// <summary>
169 : /// Perform Begin SubRun actions. Derived classes should implement beginSubRun_ instead.
170 : /// </summary>
171 : /// <param name="srp">SubRunPrincipal of new subrun</param>
172 0 : void beginSubRun(SubRunPrincipal const& srp) final
173 : {
174 0 : beginSubRun_(srp);
175 0 : }
176 : /// <summary>
177 : /// Perform Begin SubRun actions. No-op, but derived classes may override
178 : /// </summary>
179 0 : virtual void beginSubRun_(SubRunPrincipal const&) {}
180 :
181 : /// <summary>
182 : /// Perform actions for each event. Derived classes should implement event_ instead.
183 : /// </summary>
184 : /// <param name="ep">EventPrincipal of event</param>
185 0 : void event(EventPrincipal const& ep) final
186 : {
187 0 : event_(ep);
188 0 : }
189 : /// <summary>
190 : /// Perform actions for each event. No-op, but derived classes may override
191 : /// </summary>
192 0 : virtual void event_(EventPrincipal const&) {}
193 :
194 : /// <summary>
195 : /// Write an EventPrincipal to TBufferFile and send
196 : /// </summary>
197 : /// <param name="ep">EventPrincipal to write</param>
198 : void write(EventPrincipal& ep) final;
199 :
200 : /// <summary>
201 : /// Write a RunPrincipal to TBufferFile and send
202 : /// </summary>
203 : /// <param name="rp">RunPrincipal to write</param>
204 : void writeRun(RunPrincipal& rp) final;
205 :
206 : /// <summary>
207 : /// Write a SubRunPrincipal to TBufferFile and send
208 : /// </summary>
209 : /// <param name="srp">SubRunPrincipal to write</param>
210 : void writeSubRun(SubRunPrincipal& srp) final;
211 :
212 : /// <summary>
213 : /// Extract the data products from a Principal and write them to the TBufferFile
214 : /// </summary>
215 : /// <param name="msg">Output TBufferFile</param>
216 : /// <param name="principal">Principal from which to extract products</param>
217 : /// <param name="bkv">Branch Keys for data products</param>
218 : void writeDataProducts(std::unique_ptr<TBufferFile>& msg, const Principal& principal, std::vector<BranchKey*>& bkv);
219 :
220 : /// <summary>
221 : /// Send an init message downstream.
222 : /// </summary>
223 : void send_init_message();
224 :
225 : /// <summary>
226 : /// Send the serialized art Event downstream. Artdaq output modules should define this function.
227 : /// </summary>
228 : /// <param name="msg">Serialized art Event</param>
229 : virtual void SendMessage(artdaq::FragmentPtr& msg) = 0;
230 :
231 : private:
232 : ArtdaqOutput(ArtdaqOutput const&) = delete;
233 : ArtdaqOutput(ArtdaqOutput&&) = delete;
234 : ArtdaqOutput& operator=(ArtdaqOutput const&) = delete;
235 : ArtdaqOutput& operator=(ArtdaqOutput&&) = delete;
236 :
237 : bool initMsgSent_{false};
238 : ProductList productList_;
239 : size_t last_fragment_size_{10};
240 : artdaq::Fragment::sequence_id_t last_sequence_id_{0};
241 : artdaq::Fragment::timestamp_t last_timestamp_{0};
242 : std::string raw_data_label_;
243 :
244 0 : std::unique_ptr<TBufferFile> prepareMessage(artdaq::Fragment::sequence_id_t seqID, artdaq::Fragment::timestamp_t ts, artdaq::Fragment::type_t type)
245 : {
246 0 : artdaq::NetMonHeader hdr{};
247 0 : outputFragMutex.lock();
248 0 : outputFrag = std::make_unique<artdaq::Fragment>(last_fragment_size_, seqID, my_rank, type, hdr, ts);
249 0 : auto msg = std::make_unique<TBufferFile>(TBuffer::kWrite, last_fragment_size_ * sizeof(artdaq::RawDataType), outputFrag->dataBegin(), kFALSE, &Fragment_ReAllocChar);
250 0 : msg->SetWriteMode();
251 :
252 0 : if (seqID > last_sequence_id_) last_sequence_id_ = seqID;
253 0 : if (ts > last_timestamp_) last_timestamp_ = ts;
254 :
255 0 : return msg;
256 0 : }
257 :
258 0 : void sendMessage(std::unique_ptr<TBufferFile>& msg)
259 : {
260 0 : artdaq::NetMonHeader hdr{};
261 0 : hdr.data_length = static_cast<uint64_t>(msg->Length());
262 0 : outputFrag->updateMetadata(hdr);
263 0 : outputFrag->resizeBytes(hdr.data_length);
264 0 : last_fragment_size_ = std::ceil(msg->Length() / static_cast<double>(sizeof(artdaq::RawDataType)));
265 :
266 0 : SendMessage(outputFrag);
267 0 : outputFragMutex.unlock();
268 0 : }
269 : };
270 :
271 0 : inline void art::ArtdaqOutput::send_init_message()
272 : {
273 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "Begin: ArtdaqOutput::send_init_message()";
274 : //
275 : // Get the classes we will need.
276 : //
277 : // static TClass* string_class = TClass::GetClass("std::string");
278 : // if (string_class == nullptr) {
279 : // throw art::Exception(art::errors::DictionaryNotFound) <<
280 : // "ArtdaqOutput static send_init_message(): "
281 : // "Could not get TClass for std::string!";
282 : //}
283 0 : static TClass* product_list_class = TClass::GetClass("std::map<art::BranchKey,art::BranchDescription>");
284 0 : if (product_list_class == nullptr)
285 : {
286 0 : throw art::Exception(art::errors::DictionaryNotFound) << "ArtdaqOutput::send_init_message(): " // NOLINT(cert-err60-cpp)
287 : "Could not get TClass for "
288 0 : "map<art::BranchKey,art::BranchDescription>!";
289 : }
290 : // typedef std::map<const ProcessHistoryID,ProcessHistory> ProcessHistoryMap;
291 : // TClass* process_history_map_class = TClass::GetClass(
292 : // "std::map<const art::ProcessHistoryID,art::ProcessHistory>");
293 : // FIXME: Replace the "2" here with a use of the proper enum value!
294 0 : static TClass* process_history_map_class = TClass::GetClass("std::map<const art::Hash<2>,art::ProcessHistory>");
295 0 : if (process_history_map_class == nullptr)
296 : {
297 0 : throw art::Exception(art::errors::DictionaryNotFound) << "ArtdaqOutput::send_init_message(): " // NOLINT(cert-err60-cpp)
298 : "Could not get class for "
299 0 : "std::map<const art::Hash<2>,art::ProcessHistory>!";
300 : }
301 : // static TClass* parentage_map_class = TClass::GetClass(
302 : // "std::map<const art::ParentageID,art::Parentage>");
303 0 : static TClass* parentage_map_class = TClass::GetClass("art::ParentageMap");
304 0 : if (parentage_map_class == nullptr)
305 : {
306 0 : throw art::Exception(art::errors::DictionaryNotFound) << "ArtdaqOutput::send_init_message(): " // NOLINT(cert-err60-cpp)
307 0 : "Could not get class for ParentageMap.";
308 : }
309 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "parentage_map_class: " << static_cast<void*>(parentage_map_class);
310 :
311 : //
312 : // Construct and send the init message.
313 : //
314 0 : auto msg = prepareMessage(0, artdaq::Globals::my_art_id_, artdaq::Fragment::InitFragmentType);
315 : //
316 : // Stream the message type code.
317 : //
318 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Streaming message type code ...";
319 0 : msg->WriteULong(1);
320 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Finished streaming message type code.";
321 :
322 : //
323 : // Stream Class info
324 : //
325 : // ELF: 6/11/2019: This is being done so that if the receiver is a newer version of art/ROOT, it can still understand our version.
326 0 : TList infos;
327 : std::vector<std::string> classNames{"std::map<art::BranchKey,art::BranchDescription>", "std::map<const art::Hash<2>,art::ProcessHistory>", "art::ParentageMap",
328 0 : "art::BranchKey", "art::ProductProvenance", "art::RunAuxiliary", "art::SubRunAuxiliary", "art::EventAuxiliary"};
329 0 : for (auto& className : classNames)
330 : {
331 0 : TClass* class_ptr = TClass::GetClass(className.c_str());
332 0 : if (class_ptr == nullptr)
333 : {
334 0 : throw art::Exception(art::errors::DictionaryNotFound) << "ArtdaqOutput::send_init_message: Could not get TClass for " << className << "!"; // NOLINT(cert-err60-cpp)
335 : }
336 0 : infos.Add(class_ptr->GetStreamerInfo());
337 : }
338 0 : msg->WriteObject(&infos);
339 :
340 : //
341 : // Stream the ParameterSetRegistry.
342 : //
343 0 : ULong_t ps_cnt = fhicl::ParameterSetRegistry::size();
344 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): parameter set count: " << ps_cnt;
345 0 : msg->WriteULong(ps_cnt);
346 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Streaming parameter sets ...";
347 0 : for (auto I = std::begin(fhicl::ParameterSetRegistry::get()), E = std::end(fhicl::ParameterSetRegistry::get());
348 0 : I != E; ++I)
349 : {
350 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "Pset ID " << I->first << ": " << I->second.to_string();
351 0 : std::string pset_str = I->second.to_string();
352 : // msg->WriteObjectAny(&pset_str, string_class);
353 0 : msg->WriteStdString(pset_str);
354 0 : }
355 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Finished streaming parameter sets.";
356 :
357 : //
358 : // Stream the MasterProductRegistry.
359 : //
360 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Streaming Product List sz=" << productList_.size() << "...";
361 0 : msg->WriteObjectAny(&productList_, product_list_class);
362 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Finished streaming Product List.";
363 :
364 0 : art::ProcessHistoryMap phr;
365 0 : for (auto const& pr : art::ProcessHistoryRegistry::get())
366 : {
367 0 : phr.emplace(pr);
368 : }
369 : //
370 : // Dump the ProcessHistoryRegistry.
371 : //
372 0 : TLOG(TLVL_SENDINIT_VERBOSE2, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Dumping ProcessHistoryRegistry ...";
373 : // typedef std::map<const ProcessHistoryID,ProcessHistory>
374 : // ProcessHistoryMap;
375 0 : TLOG(TLVL_SENDINIT_VERBOSE2, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): phr: size: " << phr.size();
376 0 : for (auto I = phr.begin(), E = phr.end(); I != E; ++I)
377 : {
378 0 : std::ostringstream OS;
379 0 : I->first.print(OS);
380 0 : TLOG(TLVL_SENDINIT_VERBOSE2, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): phr: id: '" << OS.str() << "'";
381 0 : }
382 : //
383 : // Stream the ProcessHistoryRegistry.
384 : //
385 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Streaming ProcessHistoryRegistry ...";
386 : // typedef std::map<const ProcessHistoryID,ProcessHistory>
387 : // ProcessHistoryMap;
388 0 : const art::ProcessHistoryMap& phm = phr;
389 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): phm: size: " << phm.size();
390 0 : msg->WriteObjectAny(&phm, process_history_map_class);
391 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Finished streaming ProcessHistoryRegistry.";
392 :
393 : //
394 : // Stream the ParentageRegistry.
395 : //
396 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Streaming ParentageRegistry ..." << static_cast<void*>(parentage_map_class);
397 0 : art::ParentageMap parentageMap{};
398 0 : for (auto const& pr : art::ParentageRegistry::get())
399 : {
400 0 : parentageMap.emplace(pr.first, pr.second);
401 : }
402 :
403 0 : msg->WriteObjectAny(&parentageMap, parentage_map_class);
404 :
405 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Finished streaming ParentageRegistry.";
406 :
407 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Sending init message";
408 0 : sendMessage(msg);
409 0 : TLOG(TLVL_SENDINIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): Done sending init message, sleeping to ensure delivery";
410 0 : usleep(1000000); // Sleep to allow peer init messages to be sent/received
411 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "ArtdaqOutput::send_init_message(): END";
412 0 : initMsgSent_ = true;
413 0 : }
414 :
415 0 : inline void art::ArtdaqOutput::writeDataProducts(std::unique_ptr<TBufferFile>& msg, const Principal& principal, std::vector<BranchKey*>& bkv)
416 : {
417 0 : TLOG(TLVL_WRITEDATAPRODUCTS, "ArtdaqOutput") << "Begin: ArtdaqOutput::writeDataProducts(...)";
418 : //
419 : // Fetch the class dictionaries we need for
420 : // writing out the data products.
421 : //
422 0 : static TClass* branch_key_class = TClass::GetClass("art::BranchKey");
423 0 : if (branch_key_class == nullptr)
424 : {
425 0 : throw art::Exception(art::errors::DictionaryNotFound) << "ArtdaqOutput::writeDataProducts(...): " // NOLINT(cert-err60-cpp)
426 0 : "Could not get TClass for art::BranchKey!";
427 : }
428 0 : static TClass* prdprov_class = TClass::GetClass("art::ProductProvenance");
429 0 : if (prdprov_class == nullptr)
430 : {
431 0 : throw art::Exception(art::errors::DictionaryNotFound) << "ArtdaqOutput::writeDataProducts(...): " // NOLINT(cert-err60-cpp)
432 0 : "Could not get TClass for art::ProductProvenance!";
433 : }
434 :
435 : //
436 : // Calculate the data product count.
437 : //
438 0 : ULong_t prd_cnt = 0;
439 : // std::map<art::BranchID, std::shared_ptr<art::Group>>::const_iterator
440 0 : for (auto I = principal.begin(), E = principal.end(); I != E; ++I)
441 : {
442 0 : auto const& productDescription = I->second->productDescription();
443 0 : auto const& refs = keptProducts()[productDescription.branchType()];
444 0 : bool found = false;
445 0 : for (auto const& ref : refs)
446 : {
447 0 : if (ref.second == productDescription)
448 : {
449 0 : found = true;
450 0 : break;
451 : }
452 : }
453 0 : if (!I->second->productAvailable() || !found)
454 : {
455 0 : continue;
456 : }
457 0 : ++prd_cnt;
458 : }
459 : //
460 : // Write the data product count.
461 : //
462 0 : TLOG(TLVL_WRITEDATAPRODUCTS, "ArtdaqOutput") << "ArtdaqOutput::writeDataProducts(...): Streaming product count: " +
463 0 : std::to_string(prd_cnt);
464 0 : msg->WriteULong(prd_cnt);
465 0 : TLOG(TLVL_WRITEDATAPRODUCTS, "ArtdaqOutput") << "ArtdaqOutput::writeDataProducts(...): Finished streaming product count.";
466 :
467 : //
468 : // Loop over the groups in the RunPrincipal and
469 : // write out the data products.
470 : //
471 : // Note: We need this vector of keys because the ROOT I/O mechanism
472 : // requires that each object inserted in the message has a
473 : // unique address, so we force that by holding on to each
474 : // branch key manufactured in the loop until after we are
475 : // done constructing the message.
476 : //
477 0 : bkv.reserve(prd_cnt);
478 : // std::map<art::BranchID, std::shared_ptr<art::Group>>::const_iterator
479 0 : for (auto I = principal.begin(), E = principal.end(); I != E; ++I)
480 : {
481 0 : auto const& productDescription = I->second->productDescription();
482 0 : auto const& refs = keptProducts()[productDescription.branchType()];
483 0 : bool found = false;
484 0 : for (auto const& ref : refs)
485 : {
486 0 : if (ref.second == productDescription)
487 : {
488 0 : found = true;
489 0 : break;
490 : }
491 : }
492 0 : if (!I->second->productAvailable() || !found)
493 : {
494 0 : continue;
495 : }
496 0 : const BranchDescription& bd(I->second->productDescription());
497 0 : bkv.push_back(new BranchKey(bd));
498 0 : TLOG(TLVL_WRITEDATAPRODUCTS_VERBOSE, "ArtdaqOutput")
499 0 : << "ArtdaqOutput::writeDataProducts(...): Dumping branch key of class: '"
500 0 : << bkv.back()->friendlyClassName_ << "' modlbl: '" << bkv.back()->moduleLabel_ << "' instnm: '"
501 0 : << bkv.back()->productInstanceName_ << "' procnm: '" << bkv.back()->processName_ << "'";
502 0 : TLOG(TLVL_WRITEDATAPRODUCTS, "ArtdaqOutput") << "ArtdaqOutput::writeDataProducts(...): "
503 0 : "Streaming branch key of class: '"
504 0 : << bd.producedClassName() << "' modlbl: '" << bd.moduleLabel() << "' instnm: '"
505 0 : << bd.productInstanceName() << "' procnm: '" << bd.processName() << "'";
506 0 : msg->WriteObjectAny(bkv.back(), branch_key_class);
507 :
508 0 : TLOG(TLVL_WRITEDATAPRODUCTS, "ArtdaqOutput") << "ArtdaqOutput::writeDataProducts(...): "
509 0 : "Streaming product of class: '"
510 0 : << bd.producedClassName() << "' modlbl: '" << bd.moduleLabel() << "' instnm: '"
511 0 : << bd.productInstanceName() << "' procnm: '" << bd.processName() << "'";
512 :
513 0 : OutputHandle oh = principal.getForOutput(bd.productID(), true);
514 0 : const EDProduct* prd = oh.wrapper();
515 0 : TLOG(TLVL_WRITEDATAPRODUCTS, "ArtdaqOutput") << "Class for branch " << bd.wrappedName() << " is "
516 0 : << static_cast<void*>(TClass::GetClass(bd.wrappedName().c_str()));
517 0 : msg->WriteObjectAny(prd, TClass::GetClass(bd.wrappedName().c_str()));
518 0 : TLOG(TLVL_WRITEDATAPRODUCTS, "ArtdaqOutput") << "ArtdaqOutput::writeDataProducts(...): "
519 0 : "Streaming product provenance of class: '"
520 0 : << bd.producedClassName() << "' modlbl: '" << bd.moduleLabel() << "' instnm: '"
521 0 : << bd.productInstanceName() << "' procnm: '" << bd.processName() << "'";
522 :
523 0 : const ProductProvenance* prdprov = I->second->productProvenance().get();
524 :
525 0 : msg->WriteObjectAny(prdprov, prdprov_class);
526 0 : }
527 0 : TLOG(TLVL_WRITEDATAPRODUCTS, "ArtdaqOutput") << "End: ArtdaqOutput::writeDataProducts(...)";
528 0 : }
529 :
530 0 : inline void art::ArtdaqOutput::write(EventPrincipal& ep)
531 : {
532 : //
533 : // Write an Event message.
534 : //
535 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "Begin: ArtdaqOutput::write(EventPrincipal& ep)";
536 0 : if (!initMsgSent_)
537 : {
538 0 : send_init_message();
539 : }
540 : //
541 : // Get root classes needed for I/O.
542 : //
543 0 : static TClass* run_aux_class = TClass::GetClass("art::RunAuxiliary");
544 0 : if (run_aux_class == nullptr)
545 : {
546 0 : throw art::Exception(art::errors::DictionaryNotFound) << "ArtdaqOutput::write: " // NOLINT(cert-err60-cpp)
547 0 : "Could not get TClass for art::RunAuxiliary!";
548 : }
549 0 : static TClass* subrun_aux_class = TClass::GetClass("art::SubRunAuxiliary");
550 0 : if (subrun_aux_class == nullptr)
551 : {
552 0 : throw art::Exception(art::errors::DictionaryNotFound) << "ArtdaqOutput::write: " // NOLINT(cert-err60-cpp)
553 0 : "Could not get TClass for art::SubRunAuxiliary!";
554 : }
555 0 : static TClass* event_aux_class = TClass::GetClass("art::EventAuxiliary");
556 0 : if (event_aux_class == nullptr)
557 : {
558 0 : throw art::Exception(art::errors::DictionaryNotFound) << "ArtdaqOutput::write: " // NOLINT(cert-err60-cpp)
559 0 : "Could not get TClass for art::EventAuxiliary!";
560 : }
561 :
562 : // Subrun number starts at 1
563 0 : TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Setting Output Fragment Header Fields";
564 0 : auto seqID = (static_cast<uint64_t>(ep.eventID().subRun()) << 32) + ep.eventID().event();
565 :
566 0 : art::ProcessTag tag("", processName());
567 0 : auto res = ep.getMany(art::ModuleContext::invalid(), art::WrappedTypeID::make<artdaq::detail::RawEventHeader>(), art::MatchAllSelector(), tag);
568 :
569 0 : artdaq::Fragment::timestamp_t ts = 0;
570 :
571 0 : for (auto const& qr : res)
572 : {
573 0 : Handle<artdaq::detail::RawEventHeader> handle{qr};
574 0 : if (handle.isValid())
575 : {
576 0 : if (handle->timestamp > ts) ts = handle->timestamp;
577 : }
578 0 : }
579 0 : TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Data Fragment Header Fields: SeqID: " << seqID << ", timestamp: " << ts;
580 :
581 : //
582 : // Setup message buffer.
583 : //
584 0 : auto msg = prepareMessage(seqID, ts, artdaq::Fragment::DataFragmentType);
585 : //
586 : // Write message type code.
587 : //
588 0 : TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Streaming message type code ...";
589 0 : msg->WriteULong(static_cast<ULong_t>(artdaq::NetMonHeader::MessageType::Event));
590 0 : TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Finished streaming message type code.";
591 :
592 : //
593 : // Write RunAuxiliary.
594 : //
595 0 : TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Streaming RunAuxiliary ...";
596 0 : msg->WriteObjectAny(&ep.subRunPrincipal().runPrincipal().runAux(), run_aux_class);
597 0 : TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Finished streaming RunAuxiliary.";
598 :
599 : //
600 : // Write SubRunAuxiliary.
601 : //
602 0 : TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Streaming SubRunAuxiliary ...";
603 0 : msg->WriteObjectAny(&ep.subRunPrincipal().subRunAux(), subrun_aux_class);
604 0 : TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Finished streaming SubRunAuxiliary.";
605 :
606 : //
607 : // Write EventAuxiliary.
608 : //
609 : {
610 0 : TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Streaming EventAuxiliary ...";
611 0 : msg->WriteObjectAny(&ep.eventAux(), event_aux_class);
612 0 : TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Finished streaming EventAuxiliary.";
613 : }
614 :
615 : //
616 : // Write data products.
617 : //
618 0 : std::vector<BranchKey*> bkv;
619 0 : writeDataProducts(msg, ep, bkv);
620 :
621 0 : TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Sending message";
622 0 : sendMessage(msg);
623 0 : TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Done sending message";
624 :
625 : //
626 : // Delete the branch keys we created for the message.
627 : //
628 0 : for (auto I = bkv.begin(), E = bkv.end(); I != E; ++I)
629 : {
630 0 : delete *I;
631 0 : *I = 0;
632 : }
633 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "End: ArtdaqOutput::write(EventPrincipal& ep)";
634 0 : }
635 :
636 0 : inline void art::ArtdaqOutput::writeSubRun(SubRunPrincipal& srp)
637 : { //
638 : // Write an SubRun message.
639 : //
640 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "Begin: ArtdaqOutput::writeSubRun(SubRunPrincipal& srp)";
641 0 : if (!initMsgSent_)
642 : {
643 : // TLOG(TLVL_WARNING, "ArtdaqOutput") << "Not sending Subrun message before Event!";
644 : // return;
645 0 : send_init_message();
646 : }
647 :
648 : //
649 : // Get root classes needed for I/O.
650 : //
651 0 : static TClass* run_aux_class = TClass::GetClass("art::RunAuxiliary");
652 0 : if (run_aux_class == nullptr)
653 : {
654 0 : throw art::Exception(art::errors::DictionaryNotFound) << "ArtdaqOutput::writeSubRun: " // NOLINT(cert-err60-cpp)
655 0 : "Could not get TClass for art::RunAuxiliary!";
656 : }
657 0 : static TClass* subrun_aux_class = TClass::GetClass("art::SubRunAuxiliary");
658 0 : if (subrun_aux_class == nullptr)
659 : {
660 0 : throw art::Exception(art::errors::DictionaryNotFound) << "ArtdaqOutput::writeSubRun: " // NOLINT(cert-err60-cpp)
661 0 : "Could not get TClass for art::SubRunAuxiliary!";
662 : }
663 : //
664 : // Begin preparing message.
665 : //
666 0 : auto msg = prepareMessage(static_cast<uint64_t>(srp.subRun()) << 32, srp.subRun() + 1, artdaq::Fragment::SubrunDataFragmentType);
667 : //
668 : // Write message type code.
669 : //
670 : {
671 0 : TLOG(TLVL_WRITESUBRUN, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: streaming message type code ...";
672 0 : msg->WriteULong(static_cast<ULong_t>(artdaq::NetMonHeader::MessageType::Subrun));
673 0 : TLOG(TLVL_WRITESUBRUN, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: finished streaming message type code.";
674 : }
675 : //
676 : // Write RunAuxiliary.
677 : //
678 0 : TLOG(TLVL_WRITESUBRUN, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: Streaming RunAuxiliary ...";
679 0 : msg->WriteObjectAny(&srp.runPrincipal().runAux(), run_aux_class);
680 0 : TLOG(TLVL_WRITESUBRUN, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: Finished streaming RunAuxiliary.";
681 :
682 : //
683 : // Write SubRunAuxiliary.
684 : //
685 : {
686 0 : TLOG(TLVL_WRITESUBRUN, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: streaming SubRunAuxiliary ...";
687 :
688 0 : TLOG(TLVL_WRITESUBRUN_VERBOSE, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: dumping ProcessHistoryRegistry ...";
689 : // typedef std::map<const ProcessHistoryID,ProcessHistory>
690 : // ProcessHistoryMap;
691 0 : for (auto I = std::begin(art::ProcessHistoryRegistry::get()), E = std::end(art::ProcessHistoryRegistry::get());
692 0 : I != E; ++I)
693 : {
694 0 : std::ostringstream OS;
695 0 : I->first.print(OS);
696 0 : TLOG(TLVL_WRITESUBRUN_VERBOSE, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: phr: id: '" << OS.str() << "'";
697 0 : OS.str("");
698 0 : TLOG(TLVL_WRITESUBRUN_VERBOSE, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: phr: data.size(): " << I->second.data().size();
699 0 : if (!I->second.data().empty())
700 : {
701 0 : I->second.data().back().id().print(OS);
702 0 : TLOG(TLVL_WRITESUBRUN_VERBOSE, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: phr: data.back().id(): '" << OS.str() << "'";
703 : }
704 0 : }
705 0 : if (!srp.subRunAux().processHistoryID().isValid())
706 : {
707 0 : TLOG(TLVL_WRITESUBRUN_VERBOSE, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: ProcessHistoryID: 'INVALID'";
708 : }
709 : else
710 : {
711 0 : std::ostringstream OS;
712 0 : srp.subRunAux().processHistoryID().print(OS);
713 0 : TLOG(TLVL_WRITESUBRUN_VERBOSE, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: ProcessHistoryID: '" << OS.str() << "'";
714 0 : OS.str("");
715 0 : ProcessHistory processHistory;
716 0 : ProcessHistoryRegistry::get(srp.subRunAux().processHistoryID(), processHistory);
717 0 : if (!processHistory.data().empty())
718 : {
719 : // FIXME: Print something special on invalid id() here!
720 0 : processHistory.data().back().id().print(OS);
721 0 : TLOG(TLVL_WRITESUBRUN_VERBOSE, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: ProcessConfigurationID: '" << OS.str() << "'";
722 0 : OS.str("");
723 0 : OS << processHistory.data().back();
724 0 : TLOG(TLVL_WRITESUBRUN_VERBOSE, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: ProcessConfiguration: '" << OS.str();
725 : }
726 0 : }
727 0 : msg->WriteObjectAny(&srp.subRunAux(), subrun_aux_class);
728 0 : TLOG(TLVL_WRITESUBRUN, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: streamed SubRunAuxiliary.";
729 : }
730 : //
731 : // Write data products.
732 : //
733 0 : std::vector<BranchKey*> bkv;
734 0 : writeDataProducts(msg, srp, bkv);
735 :
736 0 : TLOG(TLVL_WRITESUBRUN, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: Sending message";
737 0 : sendMessage(msg);
738 0 : TLOG(TLVL_WRITESUBRUN, "ArtdaqOutput") << "ArtdaqOutput::writeSubRun: Done sending message";
739 :
740 : //
741 : // Delete the branch keys we created for the message.
742 : //
743 0 : for (auto I = bkv.begin(), E = bkv.end(); I != E; ++I)
744 : {
745 0 : delete *I;
746 0 : *I = 0;
747 : }
748 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "End: ArtdaqOutput::writeSubRun(const SubRunPrincipal& srp)";
749 0 : }
750 :
751 0 : inline void art::ArtdaqOutput::writeRun(RunPrincipal& rp)
752 : { //
753 : // Write an Run message.
754 : //
755 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "Begin: ArtdaqOutput::writeRun(const RunPrincipal& rp)";
756 : (void)rp;
757 0 : if (!initMsgSent_)
758 : {
759 : // TLOG(TLVL_WARNING, "ArtdaqOutput") << "Not sending Run message before Event!";
760 : // return;
761 0 : send_init_message();
762 : }
763 :
764 : //
765 : // Fetch the class dictionaries we need for
766 : // writing out the auxiliary information.
767 : //
768 0 : static TClass* run_aux_class = TClass::GetClass("art::RunAuxiliary");
769 0 : assert(run_aux_class != nullptr && "writeRun: Could not get TClass for art::RunAuxiliary!");
770 : //
771 : // Begin preparing message.
772 : //
773 0 : auto msg = prepareMessage(artdaq::Fragment::InvalidSequenceID - 1, rp.run() + 1, artdaq::Fragment::RunDataFragmentType);
774 : //
775 : // Write message type code.
776 : //
777 : {
778 0 : TLOG(TLVL_WRITERUN, "ArtdaqOutput") << "ArtdaqOutput::writeRun: streaming message type code ...";
779 0 : msg->WriteULong(static_cast<ULong_t>(artdaq::NetMonHeader::MessageType::Run));
780 0 : TLOG(TLVL_WRITERUN, "ArtdaqOutput") << "ArtdaqOutput::writeRun: finished streaming message type code.";
781 : }
782 : //
783 : // Write RunAuxiliary.
784 : //
785 : {
786 0 : TLOG(TLVL_WRITERUN, "ArtdaqOutput") << "ArtdaqOutput::writeRun: streaming RunAuxiliary ...";
787 0 : msg->WriteObjectAny(&rp.runAux(), run_aux_class);
788 0 : TLOG(TLVL_WRITERUN, "ArtdaqOutput") << "ArtdaqOutput::writeRun: streamed RunAuxiliary.";
789 : }
790 : //
791 : // Write data products.
792 : //
793 0 : std::vector<BranchKey*> bkv;
794 0 : writeDataProducts(msg, rp, bkv);
795 :
796 0 : TLOG(TLVL_WRITERUN, "ArtdaqOutput") << "ArtdaqOutput::writeRun: Sending message";
797 0 : sendMessage(msg);
798 0 : TLOG(TLVL_WRITERUN, "ArtdaqOutput") << "ArtdaqOutput::writeRun: Done sending message";
799 :
800 0 : for (auto I = bkv.begin(), E = bkv.end(); I != E; ++I)
801 : {
802 0 : delete *I;
803 0 : *I = 0;
804 : }
805 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "End: ArtdaqOutput::writeRun(RunPrincipal& rp)";
806 0 : }
807 :
808 0 : inline void art::ArtdaqOutput::beginJob()
809 : {
810 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "Begin: ArtdaqOutput::beginJob()";
811 0 : auto const& products = keptProducts();
812 :
813 : // std::array<Selections, NumBranchTypes>
814 0 : for (auto const& selections : products)
815 : {
816 : // Selections = std::map<ProductID, BranchDescription>
817 0 : for (auto const& prod_pair : selections)
818 : {
819 0 : auto const& productDescription = prod_pair.second;
820 0 : auto const& branchKey = BranchKey(productDescription);
821 :
822 0 : if (!productList_.count(branchKey))
823 : {
824 0 : TLOG(TLVL_BEGINJOB_VERBOSE, "ArtdaqOutput") << "ArtdaqOutput::beginJob:"
825 0 : << "Adding branch key to productList of class: '"
826 0 : << branchKey.friendlyClassName_ << "' modlbl: '" << branchKey.moduleLabel_ << "' instnm: '"
827 0 : << branchKey.productInstanceName_ << "' procnm: '" << branchKey.processName_ << "'"
828 0 : << ", description: " << productDescription.wrappedName();
829 :
830 0 : productList_[branchKey] = productDescription;
831 : }
832 0 : }
833 : }
834 0 : TLOG(TLVL_ENTER_EXIT, "ArtdaqOutput") << "End: ArtdaqOutput::beginJob() Product list sz=" << productList_.size();
835 :
836 : // send_init_message();
837 0 : }
838 :
839 : #endif // ARTDAQ_ARTDAQ_ARTMODULES_ARTDAQOUTPUT_HH_
|
