Merge pull request #48704 from smorovic/151x-phase2-req

cmsbuild · web-flow · commit 05ed86df4dff · 2025-08-09T13:46:59.000+02:00
[DAQ] DAQSource read-only mode and processing specific lumis in file discovery mode (Phase-2)
diff --git a/EventFilter/Utilities/doc/README-DTH.md b/EventFilter/Utilities/doc/README-DTH.md
@@ -44,6 +44,8 @@ Example file name for this mode is `run123456_ls000_index000000_source01234.raw`
 
 It is possible that another DAQ-specific header will be added to both file and per-orbit to better encapsulate data (similar is done for Run2/3 FRD files), to provide additional metadata to improve integrity and completeness checks after aggregation of data in DAQ. At present, only RAW DTH is supported by the "DTH" module.
 
+Additional parameters were added **for file discovery mode only** to allow only specific lumisection range to be processed, `overrideRangeLS = cms.untracked.vuint32($N,$M)`, and to keep input read only, `keepRawFiles = cms.untracked.bool(True)`
+
 # DAQ file formats
 Documentation:
 https://twiki.cern.ch/twiki/bin/view/CMS/FFFMetafileFormats
diff --git a/EventFilter/Utilities/interface/DAQSource.h b/EventFilter/Utilities/interface/DAQSource.h
@@ -126,6 +126,9 @@ class DAQSource : public edm::RawInputSource {
   const bool fileListLoopMode_;
   unsigned int loopModeIterationInc_ = 0;
 
+  std::vector<unsigned int> overrideRangeLS_;
+  bool keepRawFiles_;
+
   edm::RunNumber_t runNumber_;
   std::string fuOutputDir_;
 
diff --git a/EventFilter/Utilities/interface/EvFDaqDirector.h b/EventFilter/Utilities/interface/EvFDaqDirector.h
@@ -187,6 +187,11 @@ namespace evf {
     bool fileListMode() const { return fileListMode_; }
     unsigned int lsWithFilesOpen(unsigned int ls) const;
 
+    void setDiscoveryRange(int minDiscoveryLS, bool readOnly) {
+      minDiscoveryLS_ = minDiscoveryLS;
+      discoveryReadOnly_ = readOnly;
+    }
+
   private:
     void createLumiSectionFiles(const uint32_t lumiSection,
                                 const uint32_t currentLumiSection,
@@ -314,6 +319,9 @@ namespace evf {
     std::string discard_ls_filestem_;
     bool fileListMode_ = false;
     std::pair<unsigned, int> lastFileIdx_ = std::make_pair<unsigned, int>(0, -1);
+
+    int minDiscoveryLS_ = -1;
+    bool discoveryReadOnly_ = false;
   };
 }  // namespace evf
 
diff --git a/EventFilter/Utilities/src/DAQSource.cc b/EventFilter/Utilities/src/DAQSource.cc
@@ -50,6 +50,9 @@ DAQSource::DAQSource(edm::ParameterSet const& pset, edm::InputSourceDescription
       fileListMode_(pset.getUntrackedParameter<bool>("fileListMode")),
       fileDiscoveryMode_(pset.getUntrackedParameter<bool>("fileDiscoveryMode", false)),
       fileListLoopMode_(pset.getUntrackedParameter<bool>("fileListLoopMode", false)),
+      overrideRangeLS_(
+          pset.getUntrackedParameter<std::vector<unsigned int>>("overrideRangeLS", std::vector<unsigned int>())),
+      keepRawFiles_(pset.getUntrackedParameter<bool>("keepRawFiles", false)),
       runNumber_(edm::Service<evf::EvFDaqDirector>()->getRunNumber()),
       processHistoryID_(),
       currentLumiSection_(0),
@@ -719,6 +722,8 @@ void DAQSource::readSupervisor() {
   uint64_t sumLockWaitTimeUs = 0.;
 
   bool requireHeader = dataMode_->requireHeader();
+  int minDiscoveryLS = !overrideRangeLS_.empty() ? overrideRangeLS_.at(0) : -1;
+  daqDirector_->setDiscoveryRange(minDiscoveryLS, keepRawFiles_);
 
   while (!stop) {
     //wait for at least one free thread and chunk
@@ -855,6 +860,7 @@ void DAQSource::readSupervisor() {
             [&](std::string const& name, int& fd, int64_t& fsize, uint32_t sLS, bool& found) -> unsigned int {
           return dataMode_->eventCounterCallback(name, fd, fsize, sLS, found);
         };
+        nextFile = std::string();
 
         status = daqDirector_->getNextFromFileBroker(currentLumiSection,
                                                      ls,
@@ -904,16 +910,39 @@ void DAQSource::readSupervisor() {
       if (ls > currentLumiSection) {
         //new file service
         if (currentLumiSection == 0 && !alwaysStartFromFirstLS_) {
-          if (daqDirector_->getStartLumisectionFromEnv() > 1) {
+          auto lsFromEnv = daqDirector_->getStartLumisectionFromEnv();
+          unsigned int lsFromEnvMax = 0;
+          //override if parameter is specified
+          if (!overrideRangeLS_.empty()) {
+            lsFromEnv = overrideRangeLS_.at(0);
+            if (overrideRangeLS_.size() > 1) {
+              lsFromEnvMax = overrideRangeLS_.at(1);
+            }
+            if (overrideRangeLS_.size() > 2)
+              edm::LogError("DAQSource") << "More than two parameters in overrideRangeLS, they will be ignored.";
+          }
+          if (lsFromEnv > 1) {
             //start transitions from LS specified by env, continue if not reached
-            if (ls < daqDirector_->getStartLumisectionFromEnv()) {
+            //not currently compatible with fileListMode
+            //note that this should not be higher LS than daqDirector has already handled (if daqDirector is used)
+            if (ls < lsFromEnv) {
               //skip file if from earlier LS than specified by env
               if (rawFd != -1) {
                 close(rawFd);
                 rawFd = -1;
               }
               status = evf::EvFDaqDirector::noFile;
               continue;
+            } else if (lsFromEnvMax && lsFromEnvMax < ls) {
+              //finished specified LS window
+              if (rawFd != -1) {
+                close(rawFd);
+                rawFd = -1;
+              }
+              status = evf::EvFDaqDirector::runEnded;
+              fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::runEnded));
+              stop = true;
+              break;
             } else {
               fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::newLumi, ls));
             }
@@ -930,6 +959,16 @@ void DAQSource::readSupervisor() {
           }
         } else {
           //queue all lumisections after last one seen to avoid gaps
+          unsigned lsFromEnvMax = overrideRangeLS_.size() > 1 ? overrideRangeLS_[1] : 0;
+          if (lsFromEnvMax && lsFromEnvMax < ls) {
+            for (unsigned int nextLS = currentLumiSection + 1; nextLS <= lsFromEnvMax; nextLS++)
+              fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::newLumi, nextLS));
+            status = evf::EvFDaqDirector::runEnded;
+            fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::runEnded));
+            stop = true;
+            break;
+          }
+
           for (unsigned int nextLS = currentLumiSection + 1; nextLS <= ls; nextLS++) {
             fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::newLumi, nextLS));
           }
@@ -1024,6 +1063,10 @@ void DAQSource::readSupervisor() {
                                                                   0,
                                                                   eventsInNewFile,
                                                                   this));
+      if (keepRawFiles_) {
+        edm::LogWarning("DAQSource") << "Disabling raw file deletion for " << rawFile;
+        newInputFile->unsetDeleteFile();
+      }
 
       uint64_t neededSize = fileSize;
       for (const auto& addFile : additionalFiles.second) {
diff --git a/EventFilter/Utilities/src/DAQSourceModelsDTH.cc b/EventFilter/Utilities/src/DAQSourceModelsDTH.cc
@@ -476,8 +476,7 @@ int DataModeDTH::eventCounterCallback(
   if ((rawFd = ::open(name.c_str(), O_RDONLY)) < 0) {
     assert(rawFd == -1);
     found = false;
-    edm::LogError("EvFDaqDirector") << "parseFRDFileHeader - failed to open input file -: " << name << " : "
-                                    << strerror(errno);
+    edm::LogError("EvFDaqDirector") << "rawCounter - failed to open input file -: " << name << " : " << strerror(errno);
     return -1;
   }
   found = true;
diff --git a/EventFilter/Utilities/src/EvFDaqDirector.cc b/EventFilter/Utilities/src/EvFDaqDirector.cc
@@ -1138,7 +1138,7 @@ namespace evf {
     if (rawFd == -1) {
       if ((rawFd = ::open(rawPath.c_str(), O_RDONLY)) < 0) {
         edm::LogWarning("EvFDaqDirector")
-            << "parseFRDFileHeader - failed to open input file -: " << rawPath << " : " << strerror(errno);
+            << "hasFRDFileHeader - failed to open input file -: " << rawPath << " : " << strerror(errno);
         return retErr();
       }
     }
@@ -1930,6 +1930,8 @@ namespace evf {
               continue;
             }
             auto lumi = extractLumiSectionNumber(fname);
+            if (lumi > 0 && lumi < minDiscoveryLS_)
+              continue;
             if (fname.find("_EoLS.jsn") != std::string::npos) {
               if (lumi > (int)maxClosedLS)
                 maxClosedLS = lumi;
@@ -1957,7 +1959,7 @@ namespace evf {
             if (fname.size() < 4 || fname.substr(fname.size() - 4) != std::string(".raw"))
               continue;
             if (lumi >= (int)lastFileIdx_.first) {
-              if (extractIndexNumber(fname) >= lastFileIdx_.second) {
+              if (extractIndexNumber(fname) > lastFileIdx_.second) {
                 filenames.push_back(entry.path().filename().string());
               }
             }
@@ -2037,9 +2039,14 @@ namespace evf {
             fmt::format("{}{}{}{}", bu_run_dir_, fileprefix, p.stem().string(), p.extension().string());
         try {
           //grab file if possible
-          std::filesystem::rename(rawpath, nextFileRawTmp);
-          //apply changes
-          nextFileRaw = nextFileRawTmp;
+          if (!discoveryReadOnly_) {
+            //read-only mode allows only one client in file discovery option
+            std::filesystem::rename(rawpath, nextFileRawTmp);
+            //apply changes
+            nextFileRaw = nextFileRawTmp;
+          } else
+            nextFileRaw = rawpath;
+
           serverLS = nextLS;  //if changed
           closedServerLS = nextLS - 1;
 
