Line data Source code
1 : // vim: set sw=2 expandtab :
2 : #include "TRACE/tracemf.h" // TLOG
3 : #include "artdaq-utilities/Plugins/MetricData.hh"
4 : #include "artdaq/DAQdata/Globals.hh"
5 : #define TRACE_NAME (app_name + "_RootDAQOut").c_str()
6 :
7 : #include "artdaq/ArtModules/ArtdaqSharedMemoryServiceInterface.h"
8 : #include "artdaq/ArtModules/RootDAQOutFile.h"
9 :
10 : #include "art/Framework/Core/OutputModule.h"
11 : #include "art/Framework/Core/RPManager.h"
12 : #include "art/Framework/Core/ResultsProducer.h"
13 : #include "art/Framework/IO/ClosingCriteria.h"
14 : #include "art/Framework/IO/FileStatsCollector.h"
15 : #include "art/Framework/IO/PostCloseFileRenamer.h"
16 : #include "art/Framework/IO/detail/SafeFileNameConfig.h"
17 : #include "art/Framework/IO/detail/logFileAction.h"
18 : #include "art/Framework/IO/detail/validateFileNamePattern.h"
19 : #include "art/Framework/Principal/EventPrincipal.h"
20 : #include "art/Framework/Principal/ResultsPrincipal.h"
21 : #include "art/Framework/Principal/RunPrincipal.h"
22 : #include "art/Framework/Principal/SubRunPrincipal.h"
23 : #include "art/Utilities/Globals.h"
24 : #include "art/Utilities/parent_path.h"
25 : #include "art/Utilities/unique_filename.h"
26 : #include "art_root_io/DropMetaData.h"
27 : #include "art_root_io/FastCloningEnabled.h"
28 : #include "art_root_io/RootFileBlock.h"
29 : #include "art_root_io/detail/rootOutputConfigurationTools.h"
30 : #include "art_root_io/setup.h"
31 : #include "canvas/Persistency/Provenance/ProductTables.h"
32 : #include "canvas/Utilities/Exception.h"
33 : #include "fhiclcpp/ParameterSet.h"
34 : #include "fhiclcpp/types/Atom.h"
35 : #include "fhiclcpp/types/ConfigurationTable.h"
36 : #include "fhiclcpp/types/OptionalAtom.h"
37 : #include "fhiclcpp/types/OptionalSequence.h"
38 : #include "fhiclcpp/types/Table.h"
39 : #include "fhiclcpp/types/TableFragment.h"
40 : #include "messagefacility/MessageLogger/MessageLogger.h"
41 :
42 : #include <fcntl.h>
43 : #include <sys/file.h>
44 : #include <unistd.h>
45 : #include <algorithm>
46 : #include <cerrno>
47 : #include <cstring>
48 : #include <filesystem>
49 : #include <fstream>
50 : #include <iomanip>
51 : #include <limits>
52 : #include <map>
53 : #include <memory>
54 : #include <mutex>
55 : #include <set>
56 : #include <string>
57 : #include <vector>
58 :
59 : using namespace std;
60 : using namespace hep::concurrency;
61 :
62 : namespace {
63 : string const dev_null{"/dev/null"};
64 :
65 0 : bool maxCriterionSpecified(art::ClosingCriteria const& cc)
66 : {
67 0 : auto fp = mem_fn(&art::ClosingCriteria::fileProperties);
68 0 : return (fp(cc).nEvents() !=
69 0 : art::ClosingCriteria::Defaults::unsigned_max()) ||
70 0 : (fp(cc).nSubRuns() !=
71 0 : art::ClosingCriteria::Defaults::unsigned_max()) ||
72 0 : (fp(cc).nRuns() != art::ClosingCriteria::Defaults::unsigned_max()) ||
73 0 : (fp(cc).size() != art::ClosingCriteria::Defaults::size_max()) ||
74 0 : (fp(cc).age().count() !=
75 0 : art::ClosingCriteria::Defaults::seconds_max());
76 : }
77 :
78 0 : auto shouldFastClone(bool const fastCloningSet,
79 : bool const fastCloning,
80 : bool const wantAllEvents,
81 : art::ClosingCriteria const& cc)
82 : {
83 0 : art::FastCloningEnabled enabled;
84 0 : if (fastCloningSet and not fastCloning)
85 : {
86 0 : enabled.disable(
87 : "RootDAQOut configuration explicitly disables fast cloning.");
88 0 : return enabled;
89 : }
90 :
91 0 : if (not wantAllEvents)
92 : {
93 0 : enabled.disable(
94 : "Event-selection has been specified in the RootDAQOut configuration.");
95 : }
96 0 : if (fastCloning && maxCriterionSpecified(cc) &&
97 0 : cc.granularity() < art::Granularity::InputFile)
98 : {
99 0 : enabled.disable(
100 : "File-switching has been requested at event, subrun, or "
101 : "run boundaries.");
102 : }
103 0 : return enabled;
104 0 : }
105 : } // namespace
106 :
107 : namespace art {
108 :
109 : class RootDAQOut final : public OutputModule
110 : {
111 : public:
112 : static constexpr char const* default_tmpDir{"<parent-path-of-filename>"};
113 :
114 : struct Config
115 : {
116 : using Name = fhicl::Name;
117 : using Comment = fhicl::Comment;
118 : template<typename T>
119 : using Atom = fhicl::Atom<T>;
120 : template<typename T>
121 : using OptionalAtom = fhicl::OptionalAtom<T>;
122 : fhicl::TableFragment<OutputModule::Config> omConfig;
123 : Atom<string> catalog{Name("catalog"), ""};
124 : OptionalAtom<bool> dropAllEvents{Name("dropAllEvents")};
125 : Atom<bool> dropAllSubRuns{Name("dropAllSubRuns"), false};
126 : OptionalAtom<bool> fastCloning{Name("fastCloning")};
127 : Atom<string> tmpDir{Name("tmpDir"), default_tmpDir};
128 0 : Atom<int> compressionLevel{Name("compressionLevel"), 7};
129 0 : Atom<unsigned> freePercent{Name("freePercent"), 0};
130 0 : Atom<unsigned> freeMB{Name("freeMB"), 0};
131 : Atom<int64_t> saveMemoryObjectThreshold{Name("saveMemoryObjectThreshold"),
132 0 : -1l};
133 0 : Atom<int64_t> treeMaxVirtualSize{Name("treeMaxVirtualSize"), -1};
134 0 : Atom<int> splitLevel{Name("splitLevel"), 1};
135 0 : Atom<int> basketSize{Name("basketSize"), 16384};
136 : Atom<bool> dropMetaDataForDroppedData{Name("dropMetaDataForDroppedData"),
137 : false};
138 : Atom<string> dropMetaData{Name("dropMetaData"), "NONE"};
139 : Atom<bool> writeParameterSets{Name("writeParameterSets"), true};
140 : fhicl::Table<ClosingCriteria::Config> fileProperties{
141 : Name("fileProperties"),
142 : Comment("The 'fileProperties' parameter is specified to enable "
143 : "output-file switching.")};
144 : fhicl::TableFragment<detail::SafeFileNameConfig> safeFileName;
145 0 : Atom<int> firstLoggerRank{Name("firstLoggerRank"), -1};
146 :
147 : struct NewSubStringForApp
148 : {
149 : fhicl::Atom<string> appName{fhicl::Name("appName")};
150 : fhicl::Atom<string> newString{fhicl::Name("newString")};
151 : };
152 : struct FileNameSubstitution
153 : {
154 : fhicl::Atom<string> targetString{fhicl::Name("targetString")};
155 : fhicl::Sequence<fhicl::Table<NewSubStringForApp>> replacementList{fhicl::Name("replacementList")};
156 : };
157 : fhicl::OptionalSequence<fhicl::Table<FileNameSubstitution>> fileNameSubstitutions{Name("fileNameSubstitutions")};
158 : Atom<string> summaryDir{Name("subrunRecordDir"), Comment("Directory for per-file CSV subrun record (subrun/event statistics). Empty = disabled."), ""};
159 :
160 0 : Config()
161 0 : {
162 : // Both RootDAQOut module and OutputModule use the "fileName"
163 : // FHiCL parameter. However, whereas in OutputModule the
164 : // parameter has a default, for RootDAQOut the parameter should
165 : // not. We therefore have to change the default flag setting
166 : // for 'OutputModule::Config::fileName'.
167 : using namespace fhicl::detail;
168 : ParameterBase* adjustFilename{
169 0 : const_cast<fhicl::Atom<string>*>(&omConfig().fileName)}; // NOLINT(cppcoreguidelines-pro-type-const-cast)
170 0 : adjustFilename->set_par_style(fhicl::par_style::REQUIRED);
171 0 : }
172 :
173 : struct KeysToIgnore
174 : {
175 : set<string>
176 0 : operator()() const
177 : {
178 0 : set<string> keys{OutputModule::Config::KeysToIgnore::get()};
179 0 : keys.insert("results");
180 0 : return keys;
181 0 : }
182 : };
183 : };
184 :
185 : using Parameters = fhicl::WrappedTable<Config, Config::KeysToIgnore>;
186 :
187 : ~RootDAQOut();
188 : explicit RootDAQOut(Parameters const&);
189 : RootDAQOut(RootDAQOut const&) = delete;
190 : RootDAQOut(RootDAQOut&&) = delete;
191 : RootDAQOut& operator=(RootDAQOut const&) = delete;
192 : RootDAQOut& operator=(RootDAQOut&&) = delete;
193 :
194 : void postSelectProducts() override;
195 : void beginJob() override;
196 : void endJob() override;
197 : void beginRun(RunPrincipal const&) override;
198 : void endRun(RunPrincipal const&) override;
199 : void beginSubRun(SubRunPrincipal const&) override;
200 : void endSubRun(SubRunPrincipal const&) override;
201 : void event(EventPrincipal const&) override;
202 :
203 : private:
204 : // Replace OutputModule Functions.
205 : string fileNameAtOpen() const;
206 : string fileNameAtClose(string const& currentFileName);
207 : string const& lastClosedFileName() const override;
208 : Granularity fileGranularity() const override;
209 : void openFile(FileBlock const&) override;
210 : void respondToOpenInputFile(FileBlock const&) override;
211 : void readResults(ResultsPrincipal const& resp) override;
212 : void respondToCloseInputFile(FileBlock const&) override;
213 : void incrementInputFileNumber() override;
214 : void write(EventPrincipal&) override;
215 : void writeSubRun(SubRunPrincipal&) override;
216 : void writeRun(RunPrincipal&) override;
217 : void setSubRunAuxiliaryRangeSetID(RangeSet const&) override;
218 : void setRunAuxiliaryRangeSetID(RangeSet const&) override;
219 : bool isFileOpen() const override;
220 : void setFileStatus(OutputFileStatus) override;
221 : bool requestsToCloseFile() const override;
222 : void startEndFile() override;
223 : void writeFileFormatVersion() override;
224 : void writeFileIndex() override;
225 : void writeProcessConfigurationRegistry() override;
226 : void writeProcessHistoryRegistry() override;
227 : void writeParameterSetRegistry() override;
228 : void writeProductDescriptionRegistry() override;
229 : void writeParentageRegistry() override;
230 : void doWriteFileCatalogMetadata(
231 : FileCatalogMetadata::collection_type const& md,
232 : FileCatalogMetadata::collection_type const& ssmd) override;
233 : void writeProductDependencies() override;
234 : void finishEndFile() override;
235 : void doRegisterProducts(ProductDescriptions& productsToProduce,
236 : ModuleDescription const& md) override;
237 : std::string modifyFilePattern(std::string const& /*inputPattern*/, Config const& /*config*/);
238 : void writeSummaryFile(std::string const& closedFileName);
239 :
240 : // Implementation Details.
241 : void doOpenFile();
242 :
243 : // Data Members.
244 : mutable std::recursive_mutex mutex_;
245 : string const catalog_;
246 : bool dropAllEvents_{false};
247 : bool dropAllSubRuns_;
248 : string const moduleLabel_;
249 : int inputFileCount_{};
250 : unique_ptr<RootDAQOutFile> rootOutputFile_{nullptr};
251 : FileStatsCollector fstats_;
252 : PostCloseFileRenamer fRenamer_;
253 : string const filePattern_;
254 : string tmpDir_;
255 : string lastClosedFileName_{};
256 : int const compressionLevel_;
257 : unsigned freePercent_;
258 : unsigned freeMB_;
259 : int64_t const saveMemoryObjectThreshold_;
260 : int64_t const treeMaxVirtualSize_;
261 : int const splitLevel_;
262 : int const basketSize_;
263 : DropMetaData dropMetaData_;
264 : bool dropMetaDataForDroppedData_;
265 : FastCloningEnabled fastCloningEnabled_{};
266 : // Set false only for cases where we are guaranteed never to need historical
267 : // ParameterSet information in the downstream file, such as when mixing.
268 : bool writeParameterSets_;
269 : ClosingCriteria fileProperties_;
270 : string summaryDir_;
271 : size_t filesOpenedInRun_{0};
272 : size_t filesClosedInRun_{0};
273 : struct SubrunStats
274 : {
275 : size_t nEvents{0};
276 : art::EventNumber_t firstEvent{std::numeric_limits<art::EventNumber_t>::max()};
277 : art::EventNumber_t lastEvent{0};
278 : };
279 : std::map<art::SubRunID, SubrunStats> subrunStats_;
280 : ProductDescriptions productsToProduce_{};
281 : ProductTables producedResultsProducts_{ProductTables::invalid()};
282 : RPManager rpm_;
283 : };
284 :
285 0 : RootDAQOut::~RootDAQOut() = default;
286 :
287 0 : RootDAQOut::RootDAQOut(Parameters const& config)
288 : : OutputModule{
289 0 : config().omConfig}
290 0 : , catalog_{config().catalog()}
291 0 : , dropAllSubRuns_{config().dropAllSubRuns()}
292 0 : , moduleLabel_{config.get_PSet().get<string>("module_label")}
293 0 : , fstats_{moduleLabel_, processName()}
294 0 : , fRenamer_{fstats_}
295 0 : , filePattern_{modifyFilePattern(config().omConfig().fileName(), config())}
296 0 : , tmpDir_{config().tmpDir() == default_tmpDir ? parent_path(filePattern_) : config().tmpDir()}
297 0 : , compressionLevel_{config().compressionLevel()}
298 0 : , freePercent_{config().freePercent()}
299 0 : , freeMB_{config().freeMB()}
300 0 : , saveMemoryObjectThreshold_{config().saveMemoryObjectThreshold()}
301 0 : , treeMaxVirtualSize_{config().treeMaxVirtualSize()}
302 0 : , splitLevel_{config().splitLevel()}
303 0 : , basketSize_{config().basketSize()}
304 0 : , dropMetaData_{config().dropMetaData()}
305 0 : , dropMetaDataForDroppedData_{config().dropMetaDataForDroppedData()}
306 0 : , writeParameterSets_{config().writeParameterSets()}
307 0 : , fileProperties_{config().fileProperties()}
308 0 : , summaryDir_{config().summaryDir()}
309 0 : , rpm_{config.get_PSet()}
310 : {
311 0 : TLOG(TLVL_INFO) << "RootDAQOut_module (s124 version) CONSTRUCTOR Start";
312 0 : bool const check_filename = config.get_PSet().has_key("fileProperties") and
313 0 : config().safeFileName().checkFileName();
314 0 : detail::validateFileNamePattern(check_filename, filePattern_);
315 :
316 : // Setup the streamers and error handlers.
317 0 : root::setup();
318 :
319 0 : bool const dropAllEventsSet{config().dropAllEvents(dropAllEvents_)};
320 0 : dropAllEvents_ = detail::shouldDropEvents(
321 0 : dropAllEventsSet, dropAllEvents_, dropAllSubRuns_);
322 : // N.B. Any time file switching is enabled at a boundary other than
323 : // InputFile, fastCloningEnabled_ ***MUST*** be deactivated. This is
324 : // to ensure that the Event tree from the InputFile is not
325 : // accidentally cloned to the output file before the output
326 : // module has seen the events that are going to be processed.
327 0 : bool fastCloningEnabled{true};
328 0 : bool const fastCloningSet{config().fastCloning(fastCloningEnabled)};
329 0 : fastCloningEnabled_ = shouldFastClone(
330 0 : fastCloningSet, fastCloningEnabled, wantAllEvents(), fileProperties_);
331 :
332 0 : if (auto const n = Globals::instance()->nschedules(); n > 1)
333 : {
334 0 : std::ostringstream oss;
335 0 : oss << "More than one schedule (" << n << ") is being used.";
336 0 : fastCloningEnabled_.disable(oss.str());
337 0 : }
338 :
339 0 : if (!writeParameterSets_)
340 : {
341 0 : mf::LogWarning("PROVENANCE")
342 0 : << "Output module " << moduleLabel_
343 0 : << " has parameter writeParameterSets set to false.\n"
344 : << "Parameter set provenance will not be available in subsequent "
345 0 : "jobs.\n"
346 : << "Check your experiment's policy on this issue to avoid future "
347 0 : "problems\n"
348 0 : << "with analysis reproducibility.\n";
349 : }
350 0 : }
351 :
352 0 : void RootDAQOut::openFile(FileBlock const& fb)
353 : {
354 0 : std::lock_guard sentry{mutex_};
355 : // Note: The file block here refers to the currently open
356 : // input file, so we can find out about the available
357 : // products by looping over the branches of the input
358 : // file data trees.
359 0 : if (!isFileOpen())
360 : {
361 0 : doOpenFile();
362 0 : respondToOpenInputFile(fb);
363 : }
364 0 : }
365 :
366 0 : void RootDAQOut::postSelectProducts()
367 : {
368 0 : std::lock_guard sentry{mutex_};
369 0 : if (isFileOpen())
370 : {
371 0 : rootOutputFile_->selectProducts();
372 : }
373 0 : }
374 :
375 0 : void RootDAQOut::respondToOpenInputFile(FileBlock const& fb)
376 : {
377 0 : std::lock_guard sentry{mutex_};
378 0 : ++inputFileCount_;
379 0 : if (!isFileOpen())
380 : {
381 0 : return;
382 : }
383 0 : auto const* rfb = dynamic_cast<RootFileBlock const*>(&fb);
384 0 : auto fastCloneThisOne = fastCloningEnabled_;
385 0 : if (!rfb)
386 : {
387 0 : fastCloneThisOne.disable("Input source does not read art/ROOT files.");
388 : }
389 : else
390 : {
391 0 : fastCloneThisOne.merge(rfb->fastClonable());
392 : }
393 0 : rootOutputFile_->beginInputFile(rfb, std::move(fastCloneThisOne));
394 0 : fstats_.recordInputFile(fb.fileName());
395 0 : }
396 :
397 0 : void RootDAQOut::readResults(ResultsPrincipal const& resp)
398 : {
399 0 : std::lock_guard sentry{mutex_};
400 0 : rpm_.for_each_RPWorker(
401 0 : [&resp](RPWorker& w) { w.rp().doReadResults(resp); });
402 0 : }
403 :
404 0 : void RootDAQOut::respondToCloseInputFile(FileBlock const& fb)
405 : {
406 0 : std::lock_guard sentry{mutex_};
407 0 : if (isFileOpen())
408 : {
409 0 : rootOutputFile_->respondToCloseInputFile(fb);
410 : }
411 0 : }
412 :
413 0 : void RootDAQOut::write(EventPrincipal& ep)
414 : {
415 0 : std::lock_guard sentry{mutex_};
416 0 : if (dropAllEvents_)
417 : {
418 0 : return;
419 : }
420 0 : if (hasNewlyDroppedBranch()[InEvent])
421 : {
422 0 : ep.addToProcessHistory();
423 0 : ep.refreshProcessHistoryID();
424 : }
425 0 : rootOutputFile_->writeOne(ep);
426 0 : fstats_.recordEvent(ep.eventID());
427 0 : auto const& eid = ep.eventID();
428 0 : auto& sr = subrunStats_[eid.subRunID()];
429 0 : ++sr.nEvents;
430 0 : if (eid.event() < sr.firstEvent) { sr.firstEvent = eid.event(); }
431 0 : if (eid.event() > sr.lastEvent) { sr.lastEvent = eid.event(); }
432 0 : }
433 :
434 0 : void RootDAQOut::setSubRunAuxiliaryRangeSetID(RangeSet const& rs)
435 : {
436 0 : std::lock_guard sentry{mutex_};
437 0 : rootOutputFile_->setSubRunAuxiliaryRangeSetID(rs);
438 0 : }
439 :
440 0 : void RootDAQOut::writeSubRun(SubRunPrincipal& sr)
441 : {
442 0 : std::lock_guard sentry{mutex_};
443 0 : if (dropAllSubRuns_)
444 : {
445 0 : return;
446 : }
447 0 : if (hasNewlyDroppedBranch()[InSubRun])
448 : {
449 0 : sr.addToProcessHistory();
450 : }
451 0 : rootOutputFile_->writeSubRun(sr);
452 0 : fstats_.recordSubRun(sr.subRunID());
453 0 : }
454 :
455 0 : void RootDAQOut::setRunAuxiliaryRangeSetID(RangeSet const& rs)
456 : {
457 0 : std::lock_guard sentry{mutex_};
458 0 : rootOutputFile_->setRunAuxiliaryRangeSetID(rs);
459 0 : }
460 :
461 0 : void RootDAQOut::writeRun(RunPrincipal& rp)
462 : {
463 0 : std::lock_guard sentry{mutex_};
464 0 : if (hasNewlyDroppedBranch()[InRun])
465 : {
466 0 : rp.addToProcessHistory();
467 : }
468 0 : rootOutputFile_->writeRun(rp);
469 0 : fstats_.recordRun(rp.runID());
470 0 : }
471 :
472 0 : void RootDAQOut::startEndFile()
473 : {
474 0 : std::lock_guard sentry{mutex_};
475 : auto resp = make_unique<ResultsPrincipal>(
476 0 : ResultsAuxiliary{}, moduleDescription().processConfiguration(), nullptr);
477 0 : resp->createGroupsForProducedProducts(producedResultsProducts_);
478 0 : resp->enableLookupOfProducedProducts();
479 0 : if (!producedResultsProducts_.descriptions(InResults).empty() ||
480 0 : hasNewlyDroppedBranch()[InResults])
481 : {
482 0 : resp->addToProcessHistory();
483 : }
484 0 : rpm_.for_each_RPWorker(
485 0 : [&resp](RPWorker& w) { w.rp().doWriteResults(*resp); });
486 0 : rootOutputFile_->writeResults(*resp);
487 0 : }
488 :
489 0 : void RootDAQOut::writeFileFormatVersion()
490 : {
491 0 : std::lock_guard sentry{mutex_};
492 0 : rootOutputFile_->writeFileFormatVersion();
493 0 : }
494 :
495 0 : void RootDAQOut::writeFileIndex()
496 : {
497 0 : std::lock_guard sentry{mutex_};
498 0 : rootOutputFile_->writeFileIndex();
499 0 : }
500 :
501 0 : void RootDAQOut::writeProcessConfigurationRegistry()
502 : {
503 0 : std::lock_guard sentry{mutex_};
504 0 : rootOutputFile_->writeProcessConfigurationRegistry();
505 0 : }
506 :
507 0 : void RootDAQOut::writeProcessHistoryRegistry()
508 : {
509 0 : std::lock_guard sentry{mutex_};
510 0 : rootOutputFile_->writeProcessHistoryRegistry();
511 0 : }
512 :
513 0 : void RootDAQOut::writeParameterSetRegistry()
514 : {
515 0 : std::lock_guard sentry{mutex_};
516 0 : if (writeParameterSets_)
517 : {
518 0 : rootOutputFile_->writeParameterSetRegistry();
519 : }
520 0 : }
521 :
522 0 : void RootDAQOut::writeProductDescriptionRegistry()
523 : {
524 0 : std::lock_guard sentry{mutex_};
525 0 : rootOutputFile_->writeProductDescriptionRegistry();
526 0 : }
527 :
528 0 : void RootDAQOut::writeParentageRegistry()
529 : {
530 0 : std::lock_guard sentry{mutex_};
531 0 : rootOutputFile_->writeParentageRegistry();
532 0 : }
533 :
534 0 : void RootDAQOut::doWriteFileCatalogMetadata(
535 : FileCatalogMetadata::collection_type const& md,
536 : FileCatalogMetadata::collection_type const& ssmd)
537 : {
538 0 : std::lock_guard sentry{mutex_};
539 0 : rootOutputFile_->writeFileCatalogMetadata(fstats_, md, ssmd);
540 0 : }
541 :
542 0 : void RootDAQOut::writeProductDependencies()
543 : {
544 0 : std::lock_guard sentry{mutex_};
545 0 : rootOutputFile_->writeProductDependencies();
546 0 : }
547 :
548 0 : void RootDAQOut::finishEndFile()
549 : {
550 0 : std::lock_guard sentry{mutex_};
551 0 : string const currentFileName{rootOutputFile_->currentFileName()};
552 0 : rootOutputFile_->writeTTrees();
553 0 : rootOutputFile_.reset();
554 0 : fstats_.recordFileClose();
555 0 : lastClosedFileName_ = fileNameAtClose(currentFileName);
556 0 : ++filesClosedInRun_;
557 0 : TLOG(TLVL_INFO) << __func__ << ": Closed output file \"" << lastClosedFileName_ << "\"";
558 0 : writeSummaryFile(lastClosedFileName_);
559 0 : if (metricMan)
560 : {
561 0 : metricMan->sendMetric("Last Closed Output File", lastClosedFileName_, "", 3, artdaq::MetricMode::LastPoint);
562 0 : metricMan->sendMetric("Output Files Closed", filesClosedInRun_, "files", 3, artdaq::MetricMode::LastPoint | artdaq::MetricMode::Persist);
563 : }
564 0 : subrunStats_.clear();
565 0 : rpm_.invoke(&ResultsProducer::doClear);
566 0 : }
567 :
568 0 : void RootDAQOut::doRegisterProducts(ProductDescriptions& producedProducts,
569 : ModuleDescription const& md)
570 : {
571 0 : std::lock_guard sentry{mutex_};
572 : // Register Results products from ResultsProducers.
573 0 : rpm_.for_each_RPWorker([&producedProducts, &md](RPWorker& w) {
574 0 : auto const& params = w.params();
575 0 : w.setModuleDescription(
576 0 : ModuleDescription{params.rpPSetID,
577 0 : params.rpPluginType,
578 0 : md.moduleLabel() + '#' + params.rpLabel,
579 : ModuleThreadingType::legacy,
580 : md.processConfiguration()});
581 0 : w.rp().registerProducts(producedProducts, w.moduleDescription());
582 0 : });
583 : // Form product table for Results products. We do this here so we
584 : // can appropriately set the product tables for the
585 : // ResultsPrincipal.
586 0 : productsToProduce_ = producedProducts;
587 0 : producedResultsProducts_ = ProductTables{productsToProduce_};
588 0 : }
589 :
590 0 : void RootDAQOut::setFileStatus(OutputFileStatus const ofs)
591 : {
592 0 : std::lock_guard sentry{mutex_};
593 0 : if (isFileOpen())
594 : {
595 0 : rootOutputFile_->setFileStatus(ofs);
596 : }
597 0 : }
598 :
599 0 : bool RootDAQOut::isFileOpen() const
600 : {
601 0 : std::lock_guard sentry{mutex_};
602 0 : return rootOutputFile_.get() != nullptr;
603 0 : }
604 :
605 0 : void RootDAQOut::incrementInputFileNumber()
606 : {
607 0 : std::lock_guard sentry{mutex_};
608 0 : if (isFileOpen())
609 : {
610 0 : rootOutputFile_->incrementInputFileNumber();
611 : }
612 0 : }
613 :
614 0 : bool RootDAQOut::requestsToCloseFile() const
615 : {
616 0 : std::lock_guard sentry{mutex_};
617 0 : return isFileOpen() ? rootOutputFile_->requestsToCloseFile() : false;
618 0 : }
619 :
620 : Granularity
621 0 : RootDAQOut::fileGranularity() const
622 : {
623 0 : std::lock_guard sentry{mutex_};
624 0 : return fileProperties_.granularity();
625 0 : }
626 :
627 0 : void RootDAQOut::doOpenFile()
628 : {
629 0 : std::lock_guard sentry{mutex_};
630 0 : if (inputFileCount_ == 0)
631 : {
632 0 : throw Exception(errors::LogicError) // NOLINT(cert-err60-cpp)
633 : << "Attempt to open output file before input file. "
634 0 : << "Please report this to the core framework developers.\n";
635 : }
636 0 : rootOutputFile_ = make_unique<RootDAQOutFile>(this,
637 0 : fileNameAtOpen(),
638 0 : fileProperties_,
639 0 : compressionLevel_,
640 0 : freePercent_,
641 0 : freeMB_,
642 0 : saveMemoryObjectThreshold_,
643 0 : treeMaxVirtualSize_,
644 0 : splitLevel_,
645 0 : basketSize_,
646 0 : dropMetaData_,
647 0 : dropMetaDataForDroppedData_);
648 0 : fstats_.recordFileOpen();
649 0 : subrunStats_.clear();
650 0 : ++filesOpenedInRun_;
651 0 : if (metricMan)
652 : {
653 0 : metricMan->sendMetric("Output Files Opened", filesOpenedInRun_, "files", 3, artdaq::MetricMode::LastPoint | artdaq::MetricMode::Persist);
654 : }
655 0 : TLOG(TLVL_INFO) << __func__ << ": Opened output file with pattern \"" << filePattern_ << "\"";
656 0 : }
657 :
658 : string
659 0 : RootDAQOut::fileNameAtOpen() const
660 : {
661 0 : return (filePattern_ == dev_null) ? dev_null : unique_filename(tmpDir_ + "/RootDAQOut");
662 : }
663 :
664 : string
665 0 : RootDAQOut::fileNameAtClose(std::string const& currentFileName)
666 : {
667 0 : return (filePattern_ == dev_null) ? dev_null : fRenamer_.maybeRenameFile(currentFileName, filePattern_);
668 : }
669 :
670 : string const&
671 0 : RootDAQOut::lastClosedFileName() const
672 : {
673 0 : std::lock_guard sentry{mutex_};
674 0 : if (lastClosedFileName_.empty())
675 : {
676 0 : throw Exception(errors::LogicError, "RootDAQOut::currentFileName(): ") // NOLINT(cert-err60-cpp)
677 0 : << "called before meaningful.\n";
678 : }
679 0 : return lastClosedFileName_;
680 0 : }
681 :
682 0 : void RootDAQOut::beginJob()
683 : {
684 0 : std::lock_guard sentry{mutex_};
685 0 : rpm_.invoke(&ResultsProducer::doBeginJob);
686 0 : }
687 :
688 0 : void RootDAQOut::endJob()
689 : {
690 0 : std::lock_guard sentry{mutex_};
691 0 : rpm_.invoke(&ResultsProducer::doEndJob);
692 0 : }
693 :
694 0 : void RootDAQOut::event(EventPrincipal const& ep)
695 : {
696 0 : std::lock_guard sentry{mutex_};
697 0 : rpm_.for_each_RPWorker([&ep](RPWorker& w) { w.rp().doEvent(ep); });
698 0 : }
699 :
700 0 : void RootDAQOut::beginSubRun(SubRunPrincipal const& srp)
701 : {
702 0 : std::lock_guard sentry{mutex_};
703 0 : rpm_.for_each_RPWorker([&srp](RPWorker& w) { w.rp().doBeginSubRun(srp); });
704 0 : }
705 :
706 0 : void RootDAQOut::endSubRun(SubRunPrincipal const& srp)
707 : {
708 0 : std::lock_guard sentry{mutex_};
709 0 : rpm_.for_each_RPWorker([&srp](RPWorker& w) { w.rp().doEndSubRun(srp); });
710 0 : }
711 :
712 0 : void RootDAQOut::beginRun(RunPrincipal const& rp)
713 : {
714 0 : std::lock_guard sentry{mutex_};
715 0 : filesOpenedInRun_ = 0;
716 0 : filesClosedInRun_ = 0;
717 0 : rpm_.for_each_RPWorker([&rp](RPWorker& w) { w.rp().doBeginRun(rp); });
718 0 : }
719 :
720 0 : void RootDAQOut::endRun(RunPrincipal const& rp)
721 : {
722 0 : std::lock_guard sentry{mutex_};
723 0 : rpm_.for_each_RPWorker([&rp](RPWorker& w) { w.rp().doEndRun(rp); });
724 0 : }
725 :
726 : std::string
727 0 : RootDAQOut::modifyFilePattern(std::string const& inputPattern, Config const& config)
728 : {
729 : // Make sure that the shared memory is connected
730 0 : art::ServiceHandle<ArtdaqSharedMemoryServiceInterface> shm;
731 :
732 0 : TLOG(TLVL_DEBUG + 32) << __func__ << ": inputPattern=\"" << inputPattern << "\"";
733 :
734 : // fetch the firstLoggerRank and fileNameSubstitutions (if provided) for use in
735 : // substituting keywords in the filename pattern
736 0 : int firstLoggerRank = config.firstLoggerRank();
737 0 : std::vector<Config::FileNameSubstitution> subs;
738 0 : config.fileNameSubstitutions(subs);
739 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ": firstLoggerRank=" << firstLoggerRank
740 0 : << ", numberOfSubstitutionsProvided=" << subs.size();
741 :
742 : // initialization
743 0 : std::string modifiedPattern = inputPattern;
744 0 : std::string searchString;
745 : size_t targetLocation;
746 0 : int zeroBasedRelativeRank = my_rank;
747 0 : int oneBasedRelativeRank = my_rank + 1;
748 0 : if (firstLoggerRank >= 0)
749 : {
750 0 : zeroBasedRelativeRank -= firstLoggerRank;
751 0 : oneBasedRelativeRank -= firstLoggerRank;
752 : }
753 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ": my_rank=" << my_rank << ", zeroBasedRelativeRank=" << zeroBasedRelativeRank
754 0 : << ", oneBasedRelativeRank=" << oneBasedRelativeRank;
755 :
756 : // if the "ZeroBasedRelativeRank" keyword was specified in the filename pattern,
757 : // perform the substitution
758 0 : searchString = "${ZeroBasedRelativeRank}";
759 0 : targetLocation = modifiedPattern.find(searchString);
760 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
761 0 : while (targetLocation != std::string::npos)
762 : {
763 0 : std::ostringstream oss;
764 0 : oss << zeroBasedRelativeRank;
765 0 : modifiedPattern.replace(targetLocation, searchString.length(), oss.str());
766 0 : targetLocation = modifiedPattern.find(searchString);
767 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
768 0 : }
769 :
770 : // if the "OneBasedRelativeRank" keyword was specified in the filename pattern,
771 : // perform the substitution
772 0 : searchString = "${OneBasedRelativeRank}";
773 0 : targetLocation = modifiedPattern.find(searchString);
774 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
775 0 : while (targetLocation != std::string::npos)
776 : {
777 0 : std::ostringstream oss;
778 0 : oss << oneBasedRelativeRank;
779 0 : modifiedPattern.replace(targetLocation, searchString.length(), oss.str());
780 0 : targetLocation = modifiedPattern.find(searchString);
781 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
782 0 : }
783 :
784 : // if the "Rank" keyword was specified in the filename pattern,
785 : // perform the substitution
786 0 : searchString = "${Rank}";
787 0 : targetLocation = modifiedPattern.find(searchString);
788 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
789 0 : while (targetLocation != std::string::npos)
790 : {
791 0 : std::ostringstream oss;
792 0 : oss << my_rank;
793 0 : modifiedPattern.replace(targetLocation, searchString.length(), oss.str());
794 0 : targetLocation = modifiedPattern.find(searchString);
795 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
796 0 : }
797 :
798 : // if the "Rank" keyword was specified in the filename pattern,
799 : // perform the substitution
800 0 : searchString = "${app_name}";
801 0 : targetLocation = modifiedPattern.find(searchString);
802 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
803 0 : while (targetLocation != std::string::npos)
804 : {
805 0 : std::ostringstream oss;
806 0 : oss << artdaq::Globals::app_name_;
807 0 : modifiedPattern.replace(targetLocation, searchString.length(), oss.str());
808 0 : targetLocation = modifiedPattern.find(searchString);
809 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
810 0 : }
811 :
812 : // if one or more free-form substitutions were provided, we'll do them here
813 0 : for (auto& sub : subs)
814 : {
815 : // first look up the replacement string for this process's app_name
816 0 : const std::string BLAH = "none_provided";
817 0 : std::string newString = BLAH;
818 0 : std::vector<Config::NewSubStringForApp> replacementList = sub.replacementList();
819 0 : for (auto& rdx : replacementList)
820 : {
821 0 : if (rdx.appName() == artdaq::Globals::app_name_)
822 : {
823 0 : newString = rdx.newString();
824 0 : break;
825 : }
826 : }
827 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ": app_name=" << artdaq::Globals::app_name_ << ", newString=" << newString;
828 0 : if (newString != BLAH)
829 : {
830 : // first, add the expected surrounding text, and search for that
831 0 : searchString = "${" + sub.targetString() + "}";
832 0 : targetLocation = modifiedPattern.find(searchString);
833 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
834 0 : while (targetLocation != std::string::npos)
835 : {
836 0 : modifiedPattern.replace(targetLocation, searchString.length(), newString);
837 0 : targetLocation = modifiedPattern.find(searchString);
838 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
839 : }
840 :
841 : // then, search for the provided string, verbatim, in case the user specified
842 : // the enclosing text in the configuration document
843 0 : searchString = sub.targetString();
844 0 : targetLocation = modifiedPattern.find(searchString);
845 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
846 0 : while (targetLocation != std::string::npos)
847 : {
848 0 : modifiedPattern.replace(targetLocation, searchString.length(), newString);
849 0 : targetLocation = modifiedPattern.find(searchString);
850 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
851 : }
852 : }
853 0 : }
854 :
855 0 : TLOG(TLVL_DEBUG + 32) << __func__ << ": modifiedPattern = \"" << modifiedPattern << "\"";
856 0 : return modifiedPattern;
857 0 : }
858 :
859 0 : void RootDAQOut::writeSummaryFile(std::string const& closedFileName)
860 : {
861 0 : if (summaryDir_.empty()) { return; }
862 :
863 0 : auto const& seen = fstats_.seenSubRuns();
864 0 : if (seen.empty()) { return; }
865 :
866 0 : std::string const outputFile = std::filesystem::path(closedFileName).filename().string();
867 :
868 : // Group rows by run number so that each run gets its own CSV file,
869 : // even when an output file spans multiple runs.
870 0 : std::map<RunNumber_t, std::ostringstream> runContents;
871 0 : for (auto const& srid : seen)
872 : {
873 0 : auto it = subrunStats_.find(srid);
874 : // Skip subruns that registered in fstats but had no events written to
875 : // this file (e.g. the in-progress subrun at a file-boundary close).
876 0 : if (it == subrunStats_.end()) { continue; }
877 :
878 0 : runContents[srid.run()]
879 0 : << srid.run()
880 0 : << "," << srid.subRun()
881 0 : << "," << it->second.nEvents
882 0 : << "," << it->second.firstEvent
883 0 : << "," << it->second.lastEvent
884 : << "," << outputFile
885 0 : << "\n";
886 : }
887 :
888 : // One CSV file per run, appended — matches CFODataReceiver convention
889 0 : for (auto const& [run, contentStream] : runContents)
890 : {
891 0 : std::ostringstream fname;
892 0 : fname << summaryDir_;
893 0 : if (summaryDir_.back() != '/') { fname << '/'; }
894 0 : fname << "subrun_record_run" << std::setw(6) << std::setfill('0') << run << ".csv";
895 :
896 0 : int fd = open(fname.str().c_str(), O_WRONLY | O_CREAT | O_APPEND, 0666);
897 0 : if (fd < 0)
898 : {
899 0 : TLOG(TLVL_WARNING) << "writeSummaryFile: could not open \"" << fname.str() << "\" for writing: " << strerror(errno);
900 0 : continue;
901 0 : }
902 :
903 0 : flock(fd, LOCK_EX);
904 :
905 : // Write header if file is empty
906 0 : std::string const rows = contentStream.str();
907 0 : std::string buf;
908 0 : if (lseek(fd, 0, SEEK_END) == 0)
909 : {
910 0 : buf = "run,subrun,n_events,first_event,last_event,output_file\n";
911 : }
912 0 : buf += rows;
913 :
914 : // Loop to handle partial writes and EINTR
915 0 : if (buf.size() > static_cast<size_t>(std::numeric_limits<ssize_t>::max()))
916 : {
917 0 : TLOG(TLVL_ERROR) << "writeSummaryFile: buffer too large (" << buf.size() << " bytes) to write to \""
918 0 : << fname.str() << "\"";
919 : }
920 : else
921 : {
922 0 : ssize_t total = 0;
923 0 : auto remaining = static_cast<ssize_t>(buf.size());
924 0 : while (remaining > 0)
925 : {
926 0 : ssize_t written = ::write(fd, buf.c_str() + total, static_cast<size_t>(remaining));
927 0 : if (written < 0)
928 : {
929 0 : if (errno == EINTR) { continue; }
930 0 : TLOG(TLVL_ERROR) << "writeSummaryFile: write error to \"" << fname.str() << "\": " << strerror(errno);
931 0 : break;
932 : }
933 0 : total += written;
934 0 : remaining -= written;
935 : }
936 : }
937 :
938 0 : flock(fd, LOCK_UN);
939 0 : close(fd);
940 :
941 : // Count actual data rows written (each row ends with '\n')
942 0 : size_t const rowsWritten = static_cast<size_t>(std::count(rows.begin(), rows.end(), '\n'));
943 0 : TLOG(TLVL_DEBUG) << "writeSummaryFile: appended " << rowsWritten
944 0 : << " subrun row(s) to \"" << fname.str() << "\"";
945 0 : }
946 0 : }
947 :
948 : } // namespace art
949 :
950 0 : DEFINE_ART_MODULE(art::RootDAQOut) // NOLINT(performance-unnecessary-value-param)
|
