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 + "_RootDAQOutMF").c_str()
6 :
7 : #include "artdaq/ArtModules/ArtdaqSharedMemoryServiceInterface.h"
8 : #include "artdaq/ArtModules/RootDAQOutFile.h"
9 :
10 : #include "art/Framework/Core/ModuleMacros.h"
11 : #include "art/Framework/Core/OutputModule.h"
12 : #include "art/Framework/Core/RPManager.h"
13 : #include "art/Framework/Core/ResultsProducer.h"
14 : #include "art/Framework/IO/ClosingCriteria.h"
15 : #include "art/Framework/IO/FileStatsCollector.h"
16 : #include "art/Framework/IO/PostCloseFileRenamer.h"
17 : #include "art/Framework/IO/detail/SafeFileNameConfig.h"
18 : #include "art/Framework/IO/detail/logFileAction.h"
19 : #include "art/Framework/IO/detail/validateFileNamePattern.h"
20 : #include "art/Framework/Principal/EventPrincipal.h"
21 : #include "art/Framework/Principal/ResultsPrincipal.h"
22 : #include "art/Framework/Principal/RunPrincipal.h"
23 : #include "art/Framework/Principal/SubRunPrincipal.h"
24 : #include "art/Utilities/Globals.h"
25 : #include "art/Utilities/parent_path.h"
26 : #include "art/Utilities/unique_filename.h"
27 : #include "art_root_io/DropMetaData.h"
28 : #include "art_root_io/FastCloningEnabled.h"
29 : #include "art_root_io/RootFileBlock.h"
30 : #include "art_root_io/detail/rootOutputConfigurationTools.h"
31 : #include "art_root_io/setup.h"
32 : #include "canvas/Persistency/Provenance/ProductTables.h"
33 : #include "canvas/Utilities/Exception.h"
34 : #include "cetlib_except/exception.h"
35 : #include "fhiclcpp/ParameterSet.h"
36 : #include "fhiclcpp/types/Atom.h"
37 : #include "fhiclcpp/types/ConfigurationTable.h"
38 : #include "fhiclcpp/types/OptionalAtom.h"
39 : #include "fhiclcpp/types/OptionalSequence.h"
40 : #include "fhiclcpp/types/Table.h"
41 : #include "fhiclcpp/types/TableFragment.h"
42 : #include "messagefacility/MessageLogger/MessageLogger.h"
43 :
44 : #include <fcntl.h>
45 : #include <sys/file.h>
46 : #include <unistd.h>
47 : #include <algorithm>
48 : #include <cerrno>
49 : #include <cstring>
50 : #include <deque>
51 : #include <filesystem>
52 : #include <iomanip>
53 : #include <limits>
54 : #include <map>
55 : #include <memory>
56 : #include <mutex>
57 : #include <set>
58 : #include <sstream>
59 : #include <string>
60 : #include <tuple>
61 : #include <utility>
62 : #include <vector>
63 :
64 : using namespace std;
65 : using namespace hep::concurrency;
66 :
67 : namespace {
68 : string const dev_null{"/dev/null"};
69 :
70 0 : bool maxCriterionSpecified(art::ClosingCriteria const& cc)
71 : {
72 0 : auto fp = mem_fn(&art::ClosingCriteria::fileProperties);
73 0 : return (fp(cc).nEvents() !=
74 0 : art::ClosingCriteria::Defaults::unsigned_max()) ||
75 0 : (fp(cc).nSubRuns() !=
76 0 : art::ClosingCriteria::Defaults::unsigned_max()) ||
77 0 : (fp(cc).nRuns() != art::ClosingCriteria::Defaults::unsigned_max()) ||
78 0 : (fp(cc).size() != art::ClosingCriteria::Defaults::size_max()) ||
79 0 : (fp(cc).age().count() !=
80 0 : art::ClosingCriteria::Defaults::seconds_max());
81 : }
82 :
83 0 : auto shouldFastClone(bool const fastCloningSet,
84 : bool const fastCloning,
85 : bool const wantAllEvents,
86 : art::ClosingCriteria const& cc)
87 : {
88 0 : art::FastCloningEnabled enabled;
89 0 : if (fastCloningSet and not fastCloning)
90 : {
91 0 : enabled.disable(
92 : "RootDAQOutMF configuration explicitly disables fast cloning.");
93 0 : return enabled;
94 : }
95 :
96 0 : if (not wantAllEvents)
97 : {
98 0 : enabled.disable(
99 : "Event-selection has been specified in the RootDAQOutMF configuration.");
100 : }
101 0 : if (fastCloning && maxCriterionSpecified(cc) &&
102 0 : cc.granularity() < art::Granularity::InputFile)
103 : {
104 0 : enabled.disable(
105 : "File-switching has been requested at event, subrun, or "
106 : "run boundaries.");
107 : }
108 0 : return enabled;
109 0 : }
110 :
111 : struct SubrunStats
112 : {
113 : size_t nEvents{0};
114 : art::EventNumber_t firstEvent{std::numeric_limits<art::EventNumber_t>::max()};
115 : art::EventNumber_t lastEvent{0};
116 : };
117 :
118 0 : static void writeSummaryFile(
119 : std::string const& summaryDir,
120 : std::map<art::SubRunID, SubrunStats> const& subrunStats,
121 : std::string const& closedFileName)
122 : {
123 0 : if (summaryDir.empty() || subrunStats.empty()) { return; }
124 :
125 0 : std::string const outputFile = std::filesystem::path(closedFileName).filename().string();
126 :
127 : // Group rows by run number so that each run gets its own CSV file,
128 : // even when an output file spans multiple runs.
129 0 : std::map<art::RunNumber_t, std::ostringstream> runContents;
130 0 : for (auto const& [srid, stats] : subrunStats)
131 : {
132 0 : runContents[srid.run()]
133 0 : << srid.run()
134 0 : << "," << srid.subRun()
135 0 : << "," << stats.nEvents
136 0 : << "," << stats.firstEvent
137 0 : << "," << stats.lastEvent
138 : << "," << outputFile
139 0 : << "\n";
140 : }
141 :
142 : // One CSV file per run, appended — matches CFODataReceiver convention
143 0 : for (auto const& [run, contentStream] : runContents)
144 : {
145 0 : std::ostringstream fname;
146 0 : fname << summaryDir;
147 0 : if (summaryDir.back() != '/') { fname << '/'; }
148 0 : fname << "subrun_record_run" << std::setw(6) << std::setfill('0') << run << ".csv";
149 :
150 0 : int fd = open(fname.str().c_str(), O_WRONLY | O_CREAT | O_APPEND, 0666);
151 0 : if (fd < 0)
152 : {
153 0 : TLOG(TLVL_WARNING) << "writeSummaryFile: could not open \"" << fname.str() << "\" for writing: " << strerror(errno);
154 0 : continue;
155 0 : }
156 :
157 0 : flock(fd, LOCK_EX);
158 :
159 0 : std::string const rows = contentStream.str();
160 0 : std::string buf;
161 0 : if (lseek(fd, 0, SEEK_END) == 0)
162 : {
163 0 : buf = "run,subrun,n_events,first_event,last_event,output_file\n";
164 : }
165 0 : buf += rows;
166 :
167 : // Loop to handle partial writes and EINTR
168 0 : if (buf.size() > static_cast<size_t>(std::numeric_limits<ssize_t>::max()))
169 : {
170 0 : TLOG(TLVL_ERROR) << "writeSummaryFile: buffer too large (" << buf.size() << " bytes) to write to \""
171 0 : << fname.str() << "\"";
172 : }
173 : else
174 : {
175 0 : ssize_t total = 0;
176 0 : auto remaining = static_cast<ssize_t>(buf.size());
177 0 : while (remaining > 0)
178 : {
179 0 : ssize_t written = ::write(fd, buf.c_str() + total, static_cast<size_t>(remaining));
180 0 : if (written < 0)
181 : {
182 0 : if (errno == EINTR) { continue; }
183 0 : TLOG(TLVL_ERROR) << "writeSummaryFile: write error to \"" << fname.str() << "\": " << strerror(errno);
184 0 : break;
185 : }
186 0 : total += written;
187 0 : remaining -= written;
188 : }
189 : }
190 :
191 0 : flock(fd, LOCK_UN);
192 0 : close(fd);
193 :
194 0 : size_t const rowsWritten = static_cast<size_t>(std::count(rows.begin(), rows.end(), '\n'));
195 0 : TLOG(TLVL_DEBUG) << "writeSummaryFile: appended " << rowsWritten
196 0 : << " subrun row(s) to \"" << fname.str() << "\"";
197 0 : }
198 0 : }
199 : } // namespace
200 :
201 : namespace art {
202 :
203 : // RootDAQOutMF is a variant of RootDAQOut that can keep multiple ROOT files
204 : // open at the same time. When the active file's closing criteria are met (e.g.
205 : // maxEvents, maxSubRuns, maxRuns), it is moved to a "pending-close" queue and a
206 : // new file is opened immediately. Files in the pending-close queue still have
207 : // their TFile open in memory; they are flushed to disk only when the queue
208 : // would exceed maxOpenFiles. This pipelining reduces the gap in data writing
209 : // that occurs during file transitions.
210 : class RootDAQOutMF final : public OutputModule
211 : {
212 : public:
213 : static constexpr char const* default_tmpDir{"<parent-path-of-filename>"};
214 :
215 : struct Config
216 : {
217 : using Name = fhicl::Name;
218 : using Comment = fhicl::Comment;
219 : template<typename T>
220 : using Atom = fhicl::Atom<T>;
221 : template<typename T>
222 : using OptionalAtom = fhicl::OptionalAtom<T>;
223 : fhicl::TableFragment<OutputModule::Config> omConfig;
224 : Atom<string> catalog{Name("catalog"), ""};
225 : OptionalAtom<bool> dropAllEvents{Name("dropAllEvents")};
226 : Atom<bool> dropAllSubRuns{Name("dropAllSubRuns"), false};
227 : OptionalAtom<bool> fastCloning{Name("fastCloning")};
228 : Atom<string> tmpDir{Name("tmpDir"), default_tmpDir};
229 0 : Atom<int> compressionLevel{Name("compressionLevel"), 501};
230 0 : Atom<unsigned> freePercent{Name("freePercent"), 0};
231 0 : Atom<unsigned> freeMB{Name("freeMB"), 0};
232 : Atom<int64_t> saveMemoryObjectThreshold{Name("saveMemoryObjectThreshold"),
233 0 : -1l};
234 0 : Atom<int64_t> treeMaxVirtualSize{Name("treeMaxVirtualSize"), -1};
235 0 : Atom<int> splitLevel{Name("splitLevel"), 0};
236 0 : Atom<int> basketSize{Name("basketSize"), 65536};
237 : Atom<bool> dropMetaDataForDroppedData{Name("dropMetaDataForDroppedData"),
238 : false};
239 : Atom<string> dropMetaData{Name("dropMetaData"), "NONE"};
240 : Atom<bool> writeParameterSets{Name("writeParameterSets"), true};
241 : fhicl::Table<ClosingCriteria::Config> fileProperties{
242 : Name("fileProperties"),
243 : Comment("The 'fileProperties' parameter is specified to enable "
244 : "output-file switching.")};
245 : fhicl::TableFragment<detail::SafeFileNameConfig> safeFileName;
246 0 : Atom<int> firstLoggerRank{Name("firstLoggerRank"), -1};
247 : Atom<unsigned> maxOpenFiles{
248 : Name("maxOpenFiles"),
249 : Comment("Maximum number of ROOT files that can be open simultaneously.\n"
250 : "When this limit is reached, the oldest pending file is flushed\n"
251 : "to disk before a new file is opened. A value of 1 gives the\n"
252 : "same behavior as RootDAQOut (no pipelining). Higher values\n"
253 : "allow TFile::Close() of older files to overlap with writing\n"
254 : "new events to the current file."),
255 0 : 5u};
256 :
257 : struct NewSubStringForApp
258 : {
259 : fhicl::Atom<string> appName{fhicl::Name("appName")};
260 : fhicl::Atom<string> newString{fhicl::Name("newString")};
261 : };
262 : struct FileNameSubstitution
263 : {
264 : fhicl::Atom<string> targetString{fhicl::Name("targetString")};
265 : fhicl::Sequence<fhicl::Table<NewSubStringForApp>> replacementList{fhicl::Name("replacementList")};
266 : };
267 : fhicl::OptionalSequence<fhicl::Table<FileNameSubstitution>> fileNameSubstitutions{Name("fileNameSubstitutions")};
268 : Atom<string> summaryDir{Name("subrunRecordDir"), Comment("Directory for per-file CSV subrun record (subrun/event statistics). Empty = disabled."), ""};
269 :
270 0 : Config()
271 0 : {
272 : // Both RootDAQOutMF module and OutputModule use the "fileName"
273 : // FHiCL parameter. However, whereas in OutputModule the
274 : // parameter has a default, for RootDAQOutMF the parameter should
275 : // not. We therefore have to change the default flag setting
276 : // for 'OutputModule::Config::fileName'.
277 : using namespace fhicl::detail;
278 : ParameterBase* adjustFilename{
279 0 : const_cast<fhicl::Atom<string>*>(&omConfig().fileName)}; // NOLINT(cppcoreguidelines-pro-type-const-cast)
280 0 : adjustFilename->set_par_style(fhicl::par_style::REQUIRED);
281 0 : }
282 :
283 : struct KeysToIgnore
284 : {
285 : set<string>
286 0 : operator()() const
287 : {
288 0 : set<string> keys{OutputModule::Config::KeysToIgnore::get()};
289 0 : keys.insert("results");
290 0 : return keys;
291 0 : }
292 : };
293 : };
294 :
295 : using Parameters = fhicl::WrappedTable<Config, Config::KeysToIgnore>;
296 :
297 : ~RootDAQOutMF() override;
298 : explicit RootDAQOutMF(Parameters const&);
299 : RootDAQOutMF(RootDAQOutMF const&) = delete;
300 : RootDAQOutMF(RootDAQOutMF&&) = delete;
301 : RootDAQOutMF& operator=(RootDAQOutMF const&) = delete;
302 : RootDAQOutMF& operator=(RootDAQOutMF&&) = delete;
303 :
304 : void postSelectProducts() override;
305 : void beginJob() override;
306 : void endJob() override;
307 : void beginRun(RunPrincipal const&) override;
308 : void endRun(RunPrincipal const&) override;
309 : void beginSubRun(SubRunPrincipal const&) override;
310 : void endSubRun(SubRunPrincipal const&) override;
311 : void event(EventPrincipal const&) override;
312 :
313 : private:
314 : // Replace OutputModule Functions.
315 : string fileNameAtOpen() const;
316 : string fileNameAtClose(PostCloseFileRenamer& renamer,
317 : string const& currentFileName);
318 : string const& lastClosedFileName() const override;
319 : Granularity fileGranularity() const override;
320 : void openFile(FileBlock const&) override;
321 : void respondToOpenInputFile(FileBlock const&) override;
322 : void readResults(ResultsPrincipal const& resp) override;
323 : void respondToCloseInputFile(FileBlock const&) override;
324 : void incrementInputFileNumber() override;
325 : void write(EventPrincipal&) override;
326 : void writeSubRun(SubRunPrincipal&) override;
327 : void writeRun(RunPrincipal&) override;
328 : void setSubRunAuxiliaryRangeSetID(RangeSet const&) override;
329 : void setRunAuxiliaryRangeSetID(RangeSet const&) override;
330 : bool isFileOpen() const override;
331 : void setFileStatus(OutputFileStatus) override;
332 : bool requestsToCloseFile() const override;
333 : void startEndFile() override;
334 : void writeFileFormatVersion() override;
335 : void writeFileIndex() override;
336 : void writeProcessConfigurationRegistry() override;
337 : void writeProcessHistoryRegistry() override;
338 : void writeParameterSetRegistry() override;
339 : void writeProductDescriptionRegistry() override;
340 : void writeParentageRegistry() override;
341 : void doWriteFileCatalogMetadata(
342 : FileCatalogMetadata::collection_type const& md,
343 : FileCatalogMetadata::collection_type const& ssmd) override;
344 : void writeProductDependencies() override;
345 : void finishEndFile() override;
346 : void doRegisterProducts(ProductDescriptions& productsToProduce,
347 : ModuleDescription const& md) override;
348 : std::string modifyFilePattern(std::string const& inputPattern,
349 : Config const& config);
350 :
351 : // Per-file state bundle (non-copyable, non-movable -- stored via unique_ptr
352 : // so the PostCloseFileRenamer reference into fstats remains stable).
353 : struct OutputFileBundle
354 : {
355 : FileStatsCollector fstats;
356 : PostCloseFileRenamer fRenamer;
357 : std::unique_ptr<RootDAQOutFile> file{nullptr};
358 : std::string tmpFileName{};
359 : bool metadataNeedsRefresh{false};
360 : std::string closedFileName{};
361 : std::map<art::SubRunID, SubrunStats> subrunStats;
362 : std::set<art::SubRunID> writtenSubRuns;
363 : std::set<art::RunID> writtenRuns;
364 :
365 0 : OutputFileBundle(std::string const& moduleLabel,
366 : std::string const& processName)
367 0 : : fstats(moduleLabel, processName), fRenamer(fstats)
368 0 : {}
369 :
370 : OutputFileBundle(OutputFileBundle const&) = delete;
371 : OutputFileBundle(OutputFileBundle&&) = delete;
372 : OutputFileBundle& operator=(OutputFileBundle const&) = delete;
373 : OutputFileBundle& operator=(OutputFileBundle&&) = delete;
374 : };
375 :
376 : // Implementation Details.
377 : void doOpenFile();
378 : void closePendingFile(std::unique_ptr<OutputFileBundle>& bundle);
379 : void closeOldestPendingFileIfNeeded();
380 : void removeBundleMappings(OutputFileBundle* bundle);
381 : void markLateWrite(OutputFileBundle* bundle);
382 : OutputFileBundle* targetBundleForEvent(EventPrincipal const& ep);
383 : OutputFileBundle* targetBundleForSubRun(SubRunPrincipal const& sr);
384 : OutputFileBundle* targetBundleForRun(RunPrincipal const& rp);
385 :
386 : using RoutingKey = std::tuple<unsigned, unsigned, unsigned>;
387 : using SubRunIDKey = std::pair<unsigned, unsigned>;
388 : unsigned bucketIndex(unsigned idValue, unsigned maxPerFile) const;
389 : RoutingKey makeRoutingKey(EventPrincipal const& ep) const;
390 : RoutingKey makeRoutingKey(SubRunPrincipal const& sr) const;
391 : RoutingKey makeRoutingKey(RunPrincipal const& rp) const;
392 : static SubRunIDKey makeSubRunIDKey(EventPrincipal const& ep);
393 : static SubRunIDKey makeSubRunIDKey(SubRunPrincipal const& sr);
394 : static unsigned makeRunIDKey(SubRunPrincipal const& sr);
395 : static unsigned makeRunIDKey(RunPrincipal const& rp);
396 :
397 : // Data Members.
398 : mutable std::recursive_mutex mutex_;
399 : string const catalog_;
400 : bool dropAllEvents_{false};
401 : bool dropAllSubRuns_;
402 : string const moduleLabel_;
403 : int inputFileCount_{};
404 :
405 : // Multi-file state.
406 : // activeFile_: the file currently receiving events.
407 : // pendingFiles_: files whose writeTTrees() has been called but whose
408 : // TFile has not yet been closed (destructed).
409 : std::unique_ptr<OutputFileBundle> activeFile_{nullptr};
410 : std::deque<std::unique_ptr<OutputFileBundle>> pendingFiles_;
411 : unsigned const maxOpenFiles_;
412 : std::map<RoutingKey, OutputFileBundle*> routeToBundle_;
413 : std::map<SubRunIDKey, std::set<OutputFileBundle*>> subRunToBundles_;
414 : std::map<unsigned, std::set<OutputFileBundle*>> runToBundles_;
415 : FileCatalogMetadata::collection_type lastFileCatalogMetadata_{};
416 : FileCatalogMetadata::collection_type lastSubRunMetadata_{};
417 : bool hasCatalogMetadata_{false};
418 :
419 : string const filePattern_;
420 : string tmpDir_;
421 : string lastClosedFileName_{};
422 : int const compressionLevel_;
423 : unsigned freePercent_;
424 : unsigned freeMB_;
425 : int64_t const saveMemoryObjectThreshold_;
426 : int64_t const treeMaxVirtualSize_;
427 : int const splitLevel_;
428 : int const basketSize_;
429 : DropMetaData dropMetaData_;
430 : bool dropMetaDataForDroppedData_;
431 : FastCloningEnabled fastCloningEnabled_{};
432 : // Set false only for cases where we are guaranteed never to need historical
433 : // ParameterSet information in the downstream file, such as when mixing.
434 : bool writeParameterSets_;
435 : ClosingCriteria fileProperties_;
436 : string summaryDir_;
437 : size_t filesOpenedInRun_{0};
438 : size_t filesClosedInRun_{0};
439 : // Shared %# sequence counter across all OutputFileBundle instances.
440 : // Key = intermediate filename with sentinel in place of the index.
441 : std::map<std::string, size_t> sharedFileIndex_;
442 : ProductDescriptions productsToProduce_{};
443 : ProductTables producedResultsProducts_{ProductTables::invalid()};
444 : RPManager rpm_;
445 : };
446 :
447 0 : RootDAQOutMF::~RootDAQOutMF()
448 : {
449 0 : TLOG(TLVL_INFO) << "RootDAQOutMF DESTRUCTOR this=" << static_cast<void const*>(this)
450 0 : << " label=" << moduleLabel_
451 0 : << " filesOpenedInRun_=" << filesOpenedInRun_
452 0 : << " filesClosedInRun_=" << filesClosedInRun_
453 0 : << " activeFile=" << (activeFile_ ? "non-null" : "null")
454 0 : << " pendingFiles=" << pendingFiles_.size();
455 :
456 0 : if (activeFile_)
457 : {
458 0 : TLOG(TLVL_WARNING) << "RootDAQOutMF DESTRUCTOR: active output file still present at destruction. "
459 0 : << "This indicates shutdown before normal end-file finalization.";
460 : }
461 :
462 : // Best-effort fallback: if endJob is bypassed but destruction proceeds
463 : // normally, close any pending files so metadata/trees are flushed.
464 : // This must never throw from a destructor.
465 : try
466 : {
467 0 : std::lock_guard sentry{mutex_};
468 0 : while (!pendingFiles_.empty())
469 : {
470 0 : closePendingFile(pendingFiles_.front());
471 0 : pendingFiles_.pop_front();
472 : }
473 0 : }
474 0 : catch (cet::exception const& ex)
475 : {
476 0 : TLOG(TLVL_ERROR) << "RootDAQOutMF DESTRUCTOR: exception while closing pending files: "
477 0 : << ex.what();
478 0 : }
479 0 : catch (std::exception const& ex)
480 : {
481 0 : TLOG(TLVL_ERROR) << "RootDAQOutMF DESTRUCTOR: std::exception while closing pending files: "
482 0 : << ex.what();
483 0 : }
484 0 : catch (...)
485 : {
486 0 : TLOG(TLVL_ERROR) << "RootDAQOutMF DESTRUCTOR: unknown exception while closing pending files";
487 0 : }
488 0 : }
489 :
490 0 : RootDAQOutMF::RootDAQOutMF(Parameters const& config)
491 : : OutputModule{
492 0 : config().omConfig}
493 0 : , catalog_{config().catalog()}
494 0 : , dropAllSubRuns_{config().dropAllSubRuns()}
495 0 : , moduleLabel_{config.get_PSet().get<string>("module_label")}
496 0 : , maxOpenFiles_{config().maxOpenFiles()}
497 0 : , filePattern_{modifyFilePattern(config().omConfig().fileName(), config())}
498 0 : , tmpDir_{config().tmpDir() == default_tmpDir ? parent_path(filePattern_) : config().tmpDir()}
499 0 : , compressionLevel_{config().compressionLevel()}
500 0 : , freePercent_{config().freePercent()}
501 0 : , freeMB_{config().freeMB()}
502 0 : , saveMemoryObjectThreshold_{config().saveMemoryObjectThreshold()}
503 0 : , treeMaxVirtualSize_{config().treeMaxVirtualSize()}
504 0 : , splitLevel_{config().splitLevel()}
505 0 : , basketSize_{config().basketSize()}
506 0 : , dropMetaData_{config().dropMetaData()}
507 0 : , dropMetaDataForDroppedData_{config().dropMetaDataForDroppedData()}
508 0 : , writeParameterSets_{config().writeParameterSets()}
509 0 : , fileProperties_{config().fileProperties()}
510 0 : , summaryDir_{config().summaryDir()}
511 0 : , rpm_{config.get_PSet()}
512 : {
513 0 : TLOG(TLVL_INFO) << "RootDAQOutMF_module (s124 version) CONSTRUCTOR Start this=" << static_cast<void const*>(this)
514 0 : << " label=" << moduleLabel_;
515 :
516 0 : if (maxOpenFiles_ == 0)
517 : {
518 0 : throw Exception(errors::Configuration) // NOLINT(cert-err60-cpp)
519 0 : << "RootDAQOutMF: maxOpenFiles must be >= 1.\n";
520 : }
521 :
522 0 : bool const check_filename = config.get_PSet().has_key("fileProperties") and
523 0 : config().safeFileName().checkFileName();
524 0 : detail::validateFileNamePattern(check_filename, filePattern_);
525 :
526 : // Setup the streamers and error handlers.
527 0 : root::setup();
528 :
529 0 : bool const dropAllEventsSet{config().dropAllEvents(dropAllEvents_)};
530 0 : dropAllEvents_ = detail::shouldDropEvents(
531 0 : dropAllEventsSet, dropAllEvents_, dropAllSubRuns_);
532 : // N.B. Any time file switching is enabled at a boundary other than
533 : // InputFile, fastCloningEnabled_ ***MUST*** be deactivated. This is
534 : // to ensure that the Event tree from the InputFile is not
535 : // accidentally cloned to the output file before the output
536 : // module has seen the events that are going to be processed.
537 0 : bool fastCloningEnabled{true};
538 0 : bool const fastCloningSet{config().fastCloning(fastCloningEnabled)};
539 0 : fastCloningEnabled_ = shouldFastClone(
540 0 : fastCloningSet, fastCloningEnabled, wantAllEvents(), fileProperties_);
541 :
542 0 : if (auto const n = Globals::instance()->nschedules(); n > 1)
543 : {
544 0 : std::ostringstream oss;
545 0 : oss << "More than one schedule (" << n << ") is being used.";
546 0 : fastCloningEnabled_.disable(oss.str());
547 0 : }
548 :
549 0 : if (!writeParameterSets_)
550 : {
551 0 : mf::LogWarning("PROVENANCE")
552 0 : << "Output module " << moduleLabel_
553 0 : << " has parameter writeParameterSets set to false.\n"
554 : << "Parameter set provenance will not be available in subsequent "
555 0 : "jobs.\n"
556 : << "Check your experiment's policy on this issue to avoid future "
557 0 : "problems\n"
558 0 : << "with analysis reproducibility.\n";
559 : }
560 0 : }
561 :
562 0 : void RootDAQOutMF::openFile(FileBlock const& fb)
563 : {
564 0 : std::lock_guard sentry{mutex_};
565 0 : TLOG(TLVL_DEBUG) << __func__ << ": entered, isFileOpen=" << isFileOpen()
566 0 : << ", pendingFiles=" << pendingFiles_.size()
567 0 : << ", filesOpenedInRun_=" << filesOpenedInRun_;
568 : // Note: The file block here refers to the currently open
569 : // input file, so we can find out about the available
570 : // products by looping over the branches of the input
571 : // file data trees.
572 0 : if (!isFileOpen())
573 : {
574 : // Close oldest pending file if opening a new one would exceed the limit.
575 0 : closeOldestPendingFileIfNeeded();
576 0 : doOpenFile();
577 0 : respondToOpenInputFile(fb);
578 : }
579 : else
580 : {
581 0 : TLOG(TLVL_DEBUG) << __func__ << ": skipped doOpenFile because file already open";
582 : }
583 0 : }
584 :
585 0 : void RootDAQOutMF::postSelectProducts()
586 : {
587 0 : std::lock_guard sentry{mutex_};
588 0 : if (isFileOpen())
589 : {
590 0 : activeFile_->file->selectProducts();
591 : }
592 0 : }
593 :
594 0 : void RootDAQOutMF::respondToOpenInputFile(FileBlock const& fb)
595 : {
596 0 : std::lock_guard sentry{mutex_};
597 0 : ++inputFileCount_;
598 0 : if (!isFileOpen())
599 : {
600 0 : return;
601 : }
602 0 : auto const* rfb = dynamic_cast<RootFileBlock const*>(&fb);
603 0 : auto fastCloneThisOne = fastCloningEnabled_;
604 0 : if (!rfb)
605 : {
606 0 : fastCloneThisOne.disable("Input source does not read art/ROOT files.");
607 : }
608 : else
609 : {
610 0 : fastCloneThisOne.merge(rfb->fastClonable());
611 : }
612 0 : activeFile_->file->beginInputFile(rfb, std::move(fastCloneThisOne));
613 0 : activeFile_->fstats.recordInputFile(fb.fileName());
614 0 : }
615 :
616 0 : void RootDAQOutMF::readResults(ResultsPrincipal const& resp)
617 : {
618 0 : std::lock_guard sentry{mutex_};
619 0 : rpm_.for_each_RPWorker(
620 0 : [&resp](RPWorker& w) { w.rp().doReadResults(resp); });
621 0 : }
622 :
623 0 : void RootDAQOutMF::respondToCloseInputFile(FileBlock const& fb)
624 : {
625 0 : std::lock_guard sentry{mutex_};
626 0 : if (isFileOpen())
627 : {
628 0 : activeFile_->file->respondToCloseInputFile(fb);
629 : }
630 0 : }
631 :
632 0 : void RootDAQOutMF::write(EventPrincipal& ep)
633 : {
634 0 : std::lock_guard sentry{mutex_};
635 0 : if (dropAllEvents_)
636 : {
637 0 : return;
638 : }
639 0 : if (hasNewlyDroppedBranch()[InEvent])
640 : {
641 0 : ep.addToProcessHistory();
642 0 : ep.refreshProcessHistoryID();
643 : }
644 0 : auto* bundle = targetBundleForEvent(ep);
645 0 : bundle->file->writeOne(ep);
646 0 : bundle->fstats.recordEvent(ep.eventID());
647 0 : auto const& eid = ep.eventID();
648 0 : auto& sr = bundle->subrunStats[eid.subRunID()];
649 0 : ++sr.nEvents;
650 0 : if (eid.event() < sr.firstEvent) { sr.firstEvent = eid.event(); }
651 0 : if (eid.event() > sr.lastEvent) { sr.lastEvent = eid.event(); }
652 0 : routeToBundle_[makeRoutingKey(ep)] = bundle;
653 0 : subRunToBundles_[makeSubRunIDKey(ep)].insert(bundle);
654 0 : }
655 :
656 0 : void RootDAQOutMF::setSubRunAuxiliaryRangeSetID(RangeSet const& rs)
657 : {
658 0 : std::lock_guard sentry{mutex_};
659 0 : if (activeFile_)
660 : {
661 0 : activeFile_->file->setSubRunAuxiliaryRangeSetID(rs);
662 : }
663 0 : }
664 :
665 0 : void RootDAQOutMF::writeSubRun(SubRunPrincipal& sr)
666 : {
667 0 : std::lock_guard sentry{mutex_};
668 0 : if (dropAllSubRuns_)
669 : {
670 0 : return;
671 : }
672 0 : if (hasNewlyDroppedBranch()[InSubRun])
673 : {
674 0 : sr.addToProcessHistory();
675 : }
676 0 : auto* const targetBundle = targetBundleForSubRun(sr);
677 0 : std::set<OutputFileBundle*> bundlesToWrite{targetBundle};
678 0 : if (auto const it = subRunToBundles_.find(makeSubRunIDKey(sr));
679 0 : it != subRunToBundles_.end())
680 : {
681 0 : bundlesToWrite.insert(it->second.begin(), it->second.end());
682 : }
683 0 : for (auto* bundle : bundlesToWrite)
684 : {
685 0 : if (!bundle->writtenSubRuns.insert(sr.subRunID()).second)
686 0 : continue;
687 0 : markLateWrite(bundle);
688 0 : bundle->file->writeSubRun(sr);
689 0 : bundle->fstats.recordSubRun(sr.subRunID());
690 0 : runToBundles_[makeRunIDKey(sr)].insert(bundle);
691 : }
692 0 : routeToBundle_[makeRoutingKey(sr)] = targetBundle;
693 0 : }
694 :
695 0 : void RootDAQOutMF::setRunAuxiliaryRangeSetID(RangeSet const& rs)
696 : {
697 0 : std::lock_guard sentry{mutex_};
698 0 : if (activeFile_)
699 : {
700 0 : activeFile_->file->setRunAuxiliaryRangeSetID(rs);
701 : }
702 0 : }
703 :
704 0 : void RootDAQOutMF::writeRun(RunPrincipal& rp)
705 : {
706 0 : std::lock_guard sentry{mutex_};
707 0 : if (hasNewlyDroppedBranch()[InRun])
708 : {
709 0 : rp.addToProcessHistory();
710 : }
711 0 : auto* const targetBundle = targetBundleForRun(rp);
712 0 : std::set<OutputFileBundle*> bundlesToWrite{targetBundle};
713 0 : if (auto const it = runToBundles_.find(makeRunIDKey(rp));
714 0 : it != runToBundles_.end())
715 : {
716 0 : bundlesToWrite.insert(it->second.begin(), it->second.end());
717 : }
718 0 : for (auto* bundle : bundlesToWrite)
719 : {
720 0 : if (!bundle->writtenRuns.insert(rp.runID()).second)
721 0 : continue;
722 0 : markLateWrite(bundle);
723 0 : bundle->file->writeRun(rp);
724 0 : bundle->fstats.recordRun(rp.runID());
725 : }
726 0 : routeToBundle_[makeRoutingKey(rp)] = targetBundle;
727 0 : }
728 :
729 0 : void RootDAQOutMF::startEndFile()
730 : {
731 0 : std::lock_guard sentry{mutex_};
732 : auto resp = make_unique<ResultsPrincipal>(
733 0 : ResultsAuxiliary{}, moduleDescription().processConfiguration(), nullptr);
734 0 : resp->createGroupsForProducedProducts(producedResultsProducts_);
735 0 : resp->enableLookupOfProducedProducts();
736 0 : if (!producedResultsProducts_.descriptions(InResults).empty() ||
737 0 : hasNewlyDroppedBranch()[InResults])
738 : {
739 0 : resp->addToProcessHistory();
740 : }
741 0 : rpm_.for_each_RPWorker(
742 0 : [&resp](RPWorker& w) { w.rp().doWriteResults(*resp); });
743 0 : activeFile_->file->writeResults(*resp);
744 0 : }
745 :
746 0 : void RootDAQOutMF::writeFileFormatVersion()
747 : {
748 0 : std::lock_guard sentry{mutex_};
749 0 : activeFile_->file->writeFileFormatVersion();
750 0 : }
751 :
752 0 : void RootDAQOutMF::writeFileIndex()
753 : {
754 0 : std::lock_guard sentry{mutex_};
755 : // Intentionally deferred for RootDAQOutMF:
756 : // events may still be routed to a pending file after finishEndFile().
757 : // Writing the index here can therefore miss late events. We write the
758 : // FileIndex immediately before writeTTrees() when the pending file is
759 : // actually being closed in closePendingFile().
760 0 : }
761 :
762 0 : void RootDAQOutMF::writeProcessConfigurationRegistry()
763 : {
764 0 : std::lock_guard sentry{mutex_};
765 0 : activeFile_->file->writeProcessConfigurationRegistry();
766 0 : }
767 :
768 0 : void RootDAQOutMF::writeProcessHistoryRegistry()
769 : {
770 0 : std::lock_guard sentry{mutex_};
771 : // Intentionally deferred for RootDAQOutMF:
772 : // events may still be routed to a pending file after finishEndFile().
773 : // Writing ProcessHistory here can therefore miss entries needed by late
774 : // events. We write ProcessHistory immediately before writeTTrees() when the
775 : // pending file is actually being closed in closePendingFile().
776 0 : }
777 :
778 0 : void RootDAQOutMF::writeParameterSetRegistry()
779 : {
780 0 : std::lock_guard sentry{mutex_};
781 0 : if (writeParameterSets_)
782 : {
783 0 : activeFile_->file->writeParameterSetRegistry();
784 : }
785 0 : }
786 :
787 0 : void RootDAQOutMF::writeProductDescriptionRegistry()
788 : {
789 0 : std::lock_guard sentry{mutex_};
790 0 : activeFile_->file->writeProductDescriptionRegistry();
791 0 : }
792 :
793 0 : void RootDAQOutMF::writeParentageRegistry()
794 : {
795 0 : std::lock_guard sentry{mutex_};
796 0 : activeFile_->file->writeParentageRegistry();
797 0 : }
798 :
799 0 : void RootDAQOutMF::doWriteFileCatalogMetadata(
800 : FileCatalogMetadata::collection_type const& md,
801 : FileCatalogMetadata::collection_type const& ssmd)
802 : {
803 0 : std::lock_guard sentry{mutex_};
804 0 : lastFileCatalogMetadata_ = md;
805 0 : lastSubRunMetadata_ = ssmd;
806 0 : hasCatalogMetadata_ = true;
807 0 : activeFile_->metadataNeedsRefresh = false;
808 0 : activeFile_->file->writeFileCatalogMetadata(activeFile_->fstats, md, ssmd);
809 0 : }
810 :
811 0 : void RootDAQOutMF::writeProductDependencies()
812 : {
813 0 : std::lock_guard sentry{mutex_};
814 0 : activeFile_->file->writeProductDependencies();
815 0 : }
816 :
817 0 : void RootDAQOutMF::finishEndFile()
818 : {
819 0 : std::lock_guard sentry{mutex_};
820 0 : TLOG(TLVL_DEBUG) << __func__ << ": entered, isFileOpen=" << (activeFile_ != nullptr)
821 0 : << ", pendingFiles before move=" << pendingFiles_.size()
822 0 : << ", filesClosedInRun_=" << filesClosedInRun_;
823 0 : string const tmpFileName{activeFile_->file->currentFileName()};
824 :
825 : // Record the close before computing the final name so the %# sequence
826 : // advances the same way it does in RootDAQOut.
827 0 : activeFile_->fstats.recordFileClose();
828 :
829 : // Determine the final output file name now, but defer the actual rename
830 : // until the file has been fully closed.
831 0 : activeFile_->tmpFileName = tmpFileName;
832 0 : lastClosedFileName_ = (filePattern_ == dev_null)
833 0 : ? dev_null
834 0 : : activeFile_->fRenamer.applySubstitutions(
835 0 : filePattern_);
836 :
837 0 : activeFile_->closedFileName = lastClosedFileName_;
838 0 : TLOG(TLVL_INFO) << __func__ << ": Queued output file \"" << lastClosedFileName_
839 0 : << "\" for deferred TFile::Close() (pendingFiles will have "
840 0 : << pendingFiles_.size() + 1 << " entries)";
841 :
842 : // Move the active bundle to the pending queue. The TFile is still open
843 : // in memory; it will be closed (flushed to disk) when closePendingFile()
844 : // is called.
845 0 : pendingFiles_.push_back(std::move(activeFile_));
846 0 : activeFile_.reset(); // Mark as "no active file" from art's perspective.
847 :
848 0 : rpm_.invoke(&ResultsProducer::doClear);
849 0 : }
850 :
851 0 : void RootDAQOutMF::doRegisterProducts(ProductDescriptions& producedProducts,
852 : ModuleDescription const& md)
853 : {
854 0 : std::lock_guard sentry{mutex_};
855 : // Register Results products from ResultsProducers.
856 0 : rpm_.for_each_RPWorker([&producedProducts, &md](RPWorker& w) {
857 0 : auto const& params = w.params();
858 0 : w.setModuleDescription(
859 0 : ModuleDescription{params.rpPSetID,
860 0 : params.rpPluginType,
861 0 : md.moduleLabel() + '#' + params.rpLabel,
862 : ModuleThreadingType::legacy,
863 : md.processConfiguration()});
864 0 : w.rp().registerProducts(producedProducts, w.moduleDescription());
865 0 : });
866 : // Form product table for Results products. We do this here so we
867 : // can appropriately set the product tables for the ResultsPrincipal.
868 0 : productsToProduce_ = producedProducts;
869 0 : producedResultsProducts_ = ProductTables{productsToProduce_};
870 0 : }
871 :
872 0 : void RootDAQOutMF::setFileStatus(OutputFileStatus const ofs)
873 : {
874 0 : std::lock_guard sentry{mutex_};
875 0 : if (isFileOpen())
876 : {
877 0 : activeFile_->file->setFileStatus(ofs);
878 : }
879 0 : }
880 :
881 0 : bool RootDAQOutMF::isFileOpen() const
882 : {
883 0 : std::lock_guard sentry{mutex_};
884 0 : return activeFile_ != nullptr;
885 0 : }
886 :
887 0 : void RootDAQOutMF::incrementInputFileNumber()
888 : {
889 0 : std::lock_guard sentry{mutex_};
890 0 : if (isFileOpen())
891 : {
892 0 : activeFile_->file->incrementInputFileNumber();
893 : }
894 0 : }
895 :
896 0 : bool RootDAQOutMF::requestsToCloseFile() const
897 : {
898 0 : std::lock_guard sentry{mutex_};
899 0 : return isFileOpen() ? activeFile_->file->requestsToCloseFile() : false;
900 0 : }
901 :
902 : Granularity
903 0 : RootDAQOutMF::fileGranularity() const
904 : {
905 0 : std::lock_guard sentry{mutex_};
906 0 : return fileProperties_.granularity();
907 0 : }
908 :
909 0 : void RootDAQOutMF::doOpenFile()
910 : {
911 0 : std::lock_guard sentry{mutex_};
912 0 : if (inputFileCount_ == 0)
913 : {
914 0 : throw Exception(errors::LogicError) // NOLINT(cert-err60-cpp)
915 : << "Attempt to open output file before input file. "
916 0 : << "Please report this to the core framework developers.\n";
917 : }
918 0 : activeFile_ = std::make_unique<OutputFileBundle>(moduleLabel_, processName());
919 0 : activeFile_->file = make_unique<RootDAQOutFile>(this,
920 0 : fileNameAtOpen(),
921 0 : fileProperties_,
922 0 : compressionLevel_,
923 0 : freePercent_,
924 0 : freeMB_,
925 0 : saveMemoryObjectThreshold_,
926 0 : treeMaxVirtualSize_,
927 0 : splitLevel_,
928 0 : basketSize_,
929 0 : dropMetaData_,
930 0 : dropMetaDataForDroppedData_);
931 0 : activeFile_->fstats.recordFileOpen();
932 0 : ++filesOpenedInRun_;
933 0 : TLOG(TLVL_DEBUG) << __func__ << ": filesOpenedInRun_ now " << filesOpenedInRun_
934 0 : << ", metricMan=" << (metricMan ? "non-null" : "NULL");
935 0 : if (metricMan)
936 : {
937 0 : metricMan->sendMetric("Output Files Opened", filesOpenedInRun_, "files", 3, artdaq::MetricMode::LastPoint | artdaq::MetricMode::Persist);
938 : }
939 0 : TLOG(TLVL_INFO) << __func__ << ": Opened output file with pattern \"" << filePattern_ << "\"";
940 0 : }
941 :
942 0 : void RootDAQOutMF::closePendingFile(std::unique_ptr<OutputFileBundle>& bundle)
943 : {
944 0 : removeBundleMappings(bundle.get());
945 0 : if (bundle->metadataNeedsRefresh && hasCatalogMetadata_)
946 : {
947 0 : bundle->file->writeFileCatalogMetadata(
948 0 : bundle->fstats, lastFileCatalogMetadata_, lastSubRunMetadata_);
949 : }
950 : // ProcessHistory must be written at true close time so it captures all
951 : // history information needed by late-routed events.
952 0 : bundle->file->writeProcessHistoryRegistry();
953 : // FileIndex must be written at true close time so it captures all events
954 : // that may have been routed to this file while it was pending.
955 0 : bundle->file->writeFileIndex();
956 0 : bundle->file->writeTTrees();
957 : // Destroying the RootDAQOutFile calls TFile::Close(), which flushes the
958 : // ROOT key directory and closes the file descriptor.
959 0 : TLOG(TLVL_INFO) << __func__ << ": Closing pending file (TFile::Close)";
960 0 : bundle->file.reset();
961 0 : bundle->closedFileName =
962 0 : fileNameAtClose(bundle->fRenamer, bundle->tmpFileName);
963 0 : writeSummaryFile(summaryDir_, bundle->subrunStats, bundle->closedFileName);
964 0 : ++filesClosedInRun_;
965 0 : TLOG(TLVL_DEBUG) << __func__ << ": filesClosedInRun_ now " << filesClosedInRun_
966 0 : << ", metricMan=" << (metricMan ? "non-null" : "NULL");
967 0 : if (metricMan)
968 : {
969 0 : metricMan->sendMetric("Last Closed Output File", bundle->closedFileName, "", 3, artdaq::MetricMode::LastPoint);
970 0 : metricMan->sendMetric("Output Files Closed", filesClosedInRun_, "files", 3, artdaq::MetricMode::LastPoint | artdaq::MetricMode::Persist);
971 : }
972 0 : bundle.reset();
973 0 : }
974 :
975 0 : void RootDAQOutMF::closeOldestPendingFileIfNeeded()
976 : {
977 0 : std::lock_guard sentry{mutex_};
978 : // Total open files after opening a new active = pendingFiles_.size() + 1.
979 : // Close oldest pending entries until we are within the limit.
980 0 : while (!pendingFiles_.empty() &&
981 0 : (pendingFiles_.size() + 1 > maxOpenFiles_))
982 : {
983 0 : TLOG(TLVL_INFO) << __func__
984 0 : << ": Closing oldest pending file to respect maxOpenFiles="
985 0 : << maxOpenFiles_
986 0 : << " (currently pending=" << pendingFiles_.size() << ")";
987 0 : closePendingFile(pendingFiles_.front());
988 0 : pendingFiles_.pop_front();
989 : }
990 0 : }
991 :
992 0 : void RootDAQOutMF::removeBundleMappings(OutputFileBundle* bundle)
993 : {
994 0 : for (auto it = routeToBundle_.begin(); it != routeToBundle_.end();)
995 : {
996 0 : if (it->second == bundle)
997 : {
998 0 : it = routeToBundle_.erase(it);
999 : }
1000 : else
1001 : {
1002 0 : ++it;
1003 : }
1004 : }
1005 0 : for (auto it = subRunToBundles_.begin(); it != subRunToBundles_.end();)
1006 : {
1007 0 : it->second.erase(bundle);
1008 0 : if (it->second.empty())
1009 : {
1010 0 : it = subRunToBundles_.erase(it);
1011 : }
1012 : else
1013 : {
1014 0 : ++it;
1015 : }
1016 : }
1017 0 : for (auto it = runToBundles_.begin(); it != runToBundles_.end();)
1018 : {
1019 0 : it->second.erase(bundle);
1020 0 : if (it->second.empty())
1021 : {
1022 0 : it = runToBundles_.erase(it);
1023 : }
1024 : else
1025 : {
1026 0 : ++it;
1027 : }
1028 : }
1029 0 : }
1030 :
1031 0 : void RootDAQOutMF::markLateWrite(OutputFileBundle* bundle)
1032 : {
1033 0 : if (bundle != activeFile_.get())
1034 : {
1035 0 : bundle->metadataNeedsRefresh = true;
1036 : }
1037 0 : }
1038 :
1039 : RootDAQOutMF::OutputFileBundle*
1040 0 : RootDAQOutMF::targetBundleForEvent(EventPrincipal const& ep)
1041 : {
1042 0 : auto* bundle = activeFile_.get();
1043 0 : if (auto const it = routeToBundle_.find(makeRoutingKey(ep));
1044 0 : it != routeToBundle_.end())
1045 : {
1046 0 : bundle = it->second;
1047 : }
1048 0 : markLateWrite(bundle);
1049 0 : return bundle;
1050 : }
1051 :
1052 : RootDAQOutMF::OutputFileBundle*
1053 0 : RootDAQOutMF::targetBundleForSubRun(SubRunPrincipal const& sr)
1054 : {
1055 0 : auto* bundle = activeFile_.get();
1056 0 : if (auto const it = routeToBundle_.find(makeRoutingKey(sr));
1057 0 : it != routeToBundle_.end())
1058 : {
1059 0 : bundle = it->second;
1060 : }
1061 0 : markLateWrite(bundle);
1062 0 : return bundle;
1063 : }
1064 :
1065 : RootDAQOutMF::OutputFileBundle*
1066 0 : RootDAQOutMF::targetBundleForRun(RunPrincipal const& rp)
1067 : {
1068 0 : auto* bundle = activeFile_.get();
1069 0 : if (auto const it = routeToBundle_.find(makeRoutingKey(rp));
1070 0 : it != routeToBundle_.end())
1071 : {
1072 0 : bundle = it->second;
1073 : }
1074 0 : markLateWrite(bundle);
1075 0 : return bundle;
1076 : }
1077 :
1078 : unsigned
1079 0 : RootDAQOutMF::bucketIndex(unsigned const idValue, unsigned const maxPerFile) const
1080 : {
1081 0 : return (maxPerFile == ClosingCriteria::Defaults::unsigned_max() ||
1082 0 : maxPerFile == 0u || idValue == 0u)
1083 0 : ? 0u
1084 0 : : (idValue - 1u) / maxPerFile;
1085 : }
1086 :
1087 : RootDAQOutMF::RoutingKey
1088 0 : RootDAQOutMF::makeRoutingKey(EventPrincipal const& ep) const
1089 : {
1090 0 : auto const limits = fileProperties_.fileProperties();
1091 0 : auto const id = ep.eventID();
1092 0 : return {bucketIndex(static_cast<unsigned>(id.run()), limits.nRuns()),
1093 0 : bucketIndex(static_cast<unsigned>(id.subRun()), limits.nSubRuns()),
1094 0 : bucketIndex(static_cast<unsigned>(id.event()), limits.nEvents())};
1095 0 : }
1096 :
1097 : RootDAQOutMF::RoutingKey
1098 0 : RootDAQOutMF::makeRoutingKey(SubRunPrincipal const& sr) const
1099 : {
1100 0 : auto const limits = fileProperties_.fileProperties();
1101 0 : auto const id = sr.subRunID();
1102 0 : return {bucketIndex(static_cast<unsigned>(id.run()), limits.nRuns()),
1103 0 : bucketIndex(static_cast<unsigned>(id.subRun()), limits.nSubRuns()),
1104 0 : 0u};
1105 0 : }
1106 :
1107 : RootDAQOutMF::RoutingKey
1108 0 : RootDAQOutMF::makeRoutingKey(RunPrincipal const& rp) const
1109 : {
1110 0 : auto const limits = fileProperties_.fileProperties();
1111 0 : return {bucketIndex(static_cast<unsigned>(rp.runID().run()),
1112 : limits.nRuns()),
1113 0 : 0u,
1114 0 : 0u};
1115 0 : }
1116 :
1117 : RootDAQOutMF::SubRunIDKey
1118 0 : RootDAQOutMF::makeSubRunIDKey(EventPrincipal const& ep)
1119 : {
1120 0 : auto const id = ep.eventID();
1121 0 : return {static_cast<unsigned>(id.run()), static_cast<unsigned>(id.subRun())};
1122 : }
1123 :
1124 : RootDAQOutMF::SubRunIDKey
1125 0 : RootDAQOutMF::makeSubRunIDKey(SubRunPrincipal const& sr)
1126 : {
1127 0 : auto const id = sr.subRunID();
1128 0 : return {static_cast<unsigned>(id.run()), static_cast<unsigned>(id.subRun())};
1129 : }
1130 :
1131 : unsigned
1132 0 : RootDAQOutMF::makeRunIDKey(SubRunPrincipal const& sr)
1133 : {
1134 0 : return static_cast<unsigned>(sr.subRunID().run());
1135 : }
1136 :
1137 : unsigned
1138 0 : RootDAQOutMF::makeRunIDKey(RunPrincipal const& rp)
1139 : {
1140 0 : return static_cast<unsigned>(rp.runID().run());
1141 : }
1142 :
1143 : string
1144 0 : RootDAQOutMF::fileNameAtOpen() const
1145 : {
1146 0 : return (filePattern_ == dev_null) ? dev_null : unique_filename(tmpDir_ + "/RootDAQOutMF");
1147 : }
1148 :
1149 : string
1150 0 : RootDAQOutMF::fileNameAtClose(PostCloseFileRenamer& renamer,
1151 : string const& currentFileName)
1152 : {
1153 0 : if (filePattern_ == dev_null) return dev_null;
1154 :
1155 : // We need a shared %# index counter across all OutputFileBundle instances,
1156 : // because multiple bundles may write files with the same %r/%s values
1157 : // (same subrun), causing independent per-bundle counters to produce the
1158 : // same destination filename.
1159 : //
1160 : // Strategy:
1161 : // 1. Replace "%N#" in the pattern with a sentinel literal that the renamer
1162 : // will pass through unchanged (it only substitutes %X tokens).
1163 : // 2. Let the bundle's renamer resolve %r, %s, timestamps, etc., producing
1164 : // an intermediate filename that contains the sentinel.
1165 : // 3. Use that intermediate name as the key into the module-level shared
1166 : // counter, increment it, then rename to the final destination.
1167 0 : static const boost::regex kSeqRe{R"(%([0-9]*)#)"};
1168 0 : static const std::string kSentinel{"__SEQIDX__"};
1169 :
1170 0 : bool const hasSeq = boost::regex_search(filePattern_, kSeqRe);
1171 0 : if (!hasSeq)
1172 : {
1173 : // No %# in pattern — delegate directly to the bundle's renamer.
1174 0 : return renamer.maybeRenameFile(currentFileName, filePattern_);
1175 : }
1176 :
1177 : // Extract optional width from the first %N# occurrence (default 3).
1178 0 : unsigned width = 3;
1179 : {
1180 0 : boost::smatch m;
1181 0 : if (boost::regex_search(filePattern_, m, boost::regex{R"(%([0-9]+)#)"}))
1182 : {
1183 0 : width = static_cast<unsigned>(std::stoul(m[1].str()));
1184 : }
1185 0 : }
1186 :
1187 : // Build a modified pattern with %N# replaced by the sentinel literal.
1188 : std::string const sentinelPattern =
1189 0 : boost::regex_replace(filePattern_, kSeqRe, kSentinel);
1190 :
1191 : // Rename tmpFile → intermediate path (sentinel still present in name).
1192 : // This resolves %r, %s, timestamps, etc. via the bundle's renamer.
1193 : std::string const intermediateName =
1194 0 : renamer.maybeRenameFile(currentFileName, sentinelPattern);
1195 :
1196 : // Assign the next unique index for this base name across all bundles.
1197 0 : size_t const idx = ++sharedFileIndex_[intermediateName];
1198 0 : if (idx > 1)
1199 : {
1200 0 : TLOG(TLVL_INFO) << __func__ << ": shared %# index=" << idx
1201 0 : << " for base pattern \"" << intermediateName
1202 0 : << "\" (multiple bundles had the same run/subrun)";
1203 : }
1204 :
1205 : // Build final name by replacing the sentinel with the formatted index.
1206 0 : std::ostringstream oss;
1207 0 : oss << std::setfill('0') << std::setw(width) << idx;
1208 : std::string const finalName =
1209 0 : boost::regex_replace(intermediateName, boost::regex{kSentinel}, oss.str());
1210 :
1211 : // Rename intermediate → final (single atomic rename on same filesystem).
1212 0 : std::error_code ec;
1213 0 : std::filesystem::rename(intermediateName, finalName, ec);
1214 0 : if (ec)
1215 : {
1216 : // Cross-filesystem fallback: copy then remove.
1217 0 : std::filesystem::copy_file(intermediateName, finalName,
1218 : std::filesystem::copy_options::overwrite_existing);
1219 0 : std::filesystem::remove(intermediateName);
1220 : }
1221 0 : return finalName;
1222 0 : }
1223 :
1224 : string const&
1225 0 : RootDAQOutMF::lastClosedFileName() const
1226 : {
1227 0 : std::lock_guard sentry{mutex_};
1228 0 : if (lastClosedFileName_.empty())
1229 : {
1230 0 : throw Exception(errors::LogicError, "RootDAQOutMF::lastClosedFileName(): ") // NOLINT(cert-err60-cpp)
1231 0 : << "called before meaningful.\n";
1232 : }
1233 0 : return lastClosedFileName_;
1234 0 : }
1235 :
1236 0 : void RootDAQOutMF::beginJob()
1237 : {
1238 0 : std::lock_guard sentry{mutex_};
1239 0 : rpm_.invoke(&ResultsProducer::doBeginJob);
1240 0 : }
1241 :
1242 0 : void RootDAQOutMF::endJob()
1243 : {
1244 0 : std::lock_guard sentry{mutex_};
1245 : // Close any files still in the pending queue (TFile::Close not yet called).
1246 0 : TLOG(TLVL_INFO) << __func__ << ": Closing " << pendingFiles_.size()
1247 0 : << " pending file(s) at end of job";
1248 0 : while (!pendingFiles_.empty())
1249 : {
1250 0 : closePendingFile(pendingFiles_.front());
1251 0 : pendingFiles_.pop_front();
1252 : }
1253 0 : rpm_.invoke(&ResultsProducer::doEndJob);
1254 0 : }
1255 :
1256 0 : void RootDAQOutMF::event(EventPrincipal const& ep)
1257 : {
1258 0 : std::lock_guard sentry{mutex_};
1259 0 : rpm_.for_each_RPWorker([&ep](RPWorker& w) { w.rp().doEvent(ep); });
1260 0 : }
1261 :
1262 0 : void RootDAQOutMF::beginSubRun(SubRunPrincipal const& srp)
1263 : {
1264 0 : std::lock_guard sentry{mutex_};
1265 0 : rpm_.for_each_RPWorker([&srp](RPWorker& w) { w.rp().doBeginSubRun(srp); });
1266 0 : }
1267 :
1268 0 : void RootDAQOutMF::endSubRun(SubRunPrincipal const& srp)
1269 : {
1270 0 : std::lock_guard sentry{mutex_};
1271 0 : rpm_.for_each_RPWorker([&srp](RPWorker& w) { w.rp().doEndSubRun(srp); });
1272 0 : }
1273 :
1274 0 : void RootDAQOutMF::beginRun(RunPrincipal const& rp)
1275 : {
1276 0 : std::lock_guard sentry{mutex_};
1277 0 : rpm_.for_each_RPWorker([&rp](RPWorker& w) { w.rp().doBeginRun(rp); });
1278 0 : }
1279 :
1280 0 : void RootDAQOutMF::endRun(RunPrincipal const& rp)
1281 : {
1282 0 : std::lock_guard sentry{mutex_};
1283 0 : rpm_.for_each_RPWorker([&rp](RPWorker& w) { w.rp().doEndRun(rp); });
1284 0 : }
1285 :
1286 : std::string
1287 0 : RootDAQOutMF::modifyFilePattern(std::string const& inputPattern, Config const& config)
1288 : {
1289 : // Make sure that the shared memory is connected
1290 0 : art::ServiceHandle<ArtdaqSharedMemoryServiceInterface> shm;
1291 :
1292 0 : TLOG(TLVL_DEBUG + 32) << __func__ << ": inputPattern=\"" << inputPattern << "\"";
1293 :
1294 : // fetch the firstLoggerRank and fileNameSubstitutions (if provided) for use in
1295 : // substituting keywords in the filename pattern
1296 0 : int firstLoggerRank = config.firstLoggerRank();
1297 0 : std::vector<Config::FileNameSubstitution> subs;
1298 0 : config.fileNameSubstitutions(subs);
1299 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ": firstLoggerRank=" << firstLoggerRank
1300 0 : << ", numberOfSubstitutionsProvided=" << subs.size();
1301 :
1302 : // initialization
1303 0 : std::string modifiedPattern = inputPattern;
1304 0 : std::string searchString;
1305 : size_t targetLocation;
1306 0 : int zeroBasedRelativeRank = my_rank;
1307 0 : int oneBasedRelativeRank = my_rank + 1;
1308 0 : if (firstLoggerRank >= 0)
1309 : {
1310 0 : zeroBasedRelativeRank -= firstLoggerRank;
1311 0 : oneBasedRelativeRank -= firstLoggerRank;
1312 : }
1313 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ": my_rank=" << my_rank << ", zeroBasedRelativeRank=" << zeroBasedRelativeRank
1314 0 : << ", oneBasedRelativeRank=" << oneBasedRelativeRank;
1315 :
1316 : // if the "ZeroBasedRelativeRank" keyword was specified in the filename pattern,
1317 : // perform the substitution
1318 0 : searchString = "${ZeroBasedRelativeRank}";
1319 0 : targetLocation = modifiedPattern.find(searchString);
1320 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
1321 0 : while (targetLocation != std::string::npos)
1322 : {
1323 0 : std::ostringstream oss;
1324 0 : oss << zeroBasedRelativeRank;
1325 0 : modifiedPattern.replace(targetLocation, searchString.length(), oss.str());
1326 0 : targetLocation = modifiedPattern.find(searchString);
1327 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
1328 0 : }
1329 :
1330 : // if the "OneBasedRelativeRank" keyword was specified in the filename pattern,
1331 : // perform the substitution
1332 0 : searchString = "${OneBasedRelativeRank}";
1333 0 : targetLocation = modifiedPattern.find(searchString);
1334 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
1335 0 : while (targetLocation != std::string::npos)
1336 : {
1337 0 : std::ostringstream oss;
1338 0 : oss << oneBasedRelativeRank;
1339 0 : modifiedPattern.replace(targetLocation, searchString.length(), oss.str());
1340 0 : targetLocation = modifiedPattern.find(searchString);
1341 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
1342 0 : }
1343 :
1344 : // if the "Rank" keyword was specified in the filename pattern,
1345 : // perform the substitution
1346 0 : searchString = "${Rank}";
1347 0 : targetLocation = modifiedPattern.find(searchString);
1348 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
1349 0 : while (targetLocation != std::string::npos)
1350 : {
1351 0 : std::ostringstream oss;
1352 0 : oss << my_rank;
1353 0 : modifiedPattern.replace(targetLocation, searchString.length(), oss.str());
1354 0 : targetLocation = modifiedPattern.find(searchString);
1355 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
1356 0 : }
1357 :
1358 : // if the "app_name" keyword was specified in the filename pattern,
1359 : // perform the substitution
1360 0 : searchString = "${app_name}";
1361 0 : targetLocation = modifiedPattern.find(searchString);
1362 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
1363 0 : while (targetLocation != std::string::npos)
1364 : {
1365 0 : std::ostringstream oss;
1366 0 : oss << artdaq::Globals::app_name_;
1367 0 : modifiedPattern.replace(targetLocation, searchString.length(), oss.str());
1368 0 : targetLocation = modifiedPattern.find(searchString);
1369 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
1370 0 : }
1371 :
1372 : // if one or more free-form substitutions were provided, we'll do them here
1373 0 : for (auto& sub : subs)
1374 : {
1375 : // first look up the replacement string for this process's app_name
1376 0 : const std::string BLAH = "none_provided";
1377 0 : std::string newString = BLAH;
1378 0 : std::vector<Config::NewSubStringForApp> replacementList = sub.replacementList();
1379 0 : for (auto& rdx : replacementList)
1380 : {
1381 0 : if (rdx.appName() == artdaq::Globals::app_name_)
1382 : {
1383 0 : newString = rdx.newString();
1384 0 : break;
1385 : }
1386 : }
1387 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ": app_name=" << artdaq::Globals::app_name_ << ", newString=" << newString;
1388 0 : if (newString != BLAH)
1389 : {
1390 : // first, add the expected surrounding text, and search for that
1391 0 : searchString = "${" + sub.targetString() + "}";
1392 0 : targetLocation = modifiedPattern.find(searchString);
1393 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
1394 0 : while (targetLocation != std::string::npos)
1395 : {
1396 0 : modifiedPattern.replace(targetLocation, searchString.length(), newString);
1397 0 : targetLocation = modifiedPattern.find(searchString);
1398 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
1399 : }
1400 :
1401 : // then, search for the provided string, verbatim, in case the user specified
1402 : // the enclosing text in the configuration document
1403 0 : searchString = sub.targetString();
1404 0 : targetLocation = modifiedPattern.find(searchString);
1405 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
1406 0 : while (targetLocation != std::string::npos)
1407 : {
1408 0 : modifiedPattern.replace(targetLocation, searchString.length(), newString);
1409 0 : targetLocation = modifiedPattern.find(searchString);
1410 0 : TLOG(TLVL_DEBUG + 33) << __func__ << ":" << __LINE__ << " searchString=" << searchString << ", targetLocation=" << targetLocation;
1411 : }
1412 : }
1413 0 : }
1414 :
1415 0 : TLOG(TLVL_DEBUG + 32) << __func__ << ": modifiedPattern = \"" << modifiedPattern << "\"";
1416 0 : return modifiedPattern;
1417 0 : }
1418 :
1419 : } // namespace art
1420 :
1421 0 : DEFINE_ART_MODULE(art::RootDAQOutMF) // NOLINT(performance-unnecessary-value-param)
|
