Merge pull request #48859 from smorovic/151x-rawwriter-one-ls-v3

[DAQ] HLT test-stand conversion job patches

Author: cmsbuild

EventFilter/Utilities/interface/EvFDaqDirector.h

@@ -68,7 +68,6 @@ namespace evf {
     static void fillDescriptions(edm::ConfigurationDescriptions& descriptions);
     void preallocate(edm::service::SystemBounds const& bounds);
     void preBeginRun(edm::GlobalContext const& globalContext);
-    void postEndRun(edm::GlobalContext const& globalContext);
     void preGlobalEndLumi(edm::GlobalContext const& globalContext);
     void updateRunParams();
     void overrideRunNumber(unsigned int run) {

EventFilter/Utilities/plugins/RawEventFileWriterForBU.cc

@@ -24,7 +24,9 @@ using namespace edm::streamer;
 
 RawEventFileWriterForBU::RawEventFileWriterForBU(edm::ParameterSet const& ps)
     : microSleep_(ps.getParameter<int>("microSleep")),
-      frdFileVersion_(ps.getParameter<unsigned int>("frdFileVersion")) {
+      frdFileVersion_(ps.getParameter<unsigned int>("frdFileVersion")),
+      writeEoR_(ps.getUntrackedParameter<bool>("writeEoR")),
+      writeToOpen_(ps.getUntrackedParameter<bool>("writeToOpen")) {
   if (edm::Service<evf::FastMonitoringService>().isAvailable())
     fms_ = static_cast<evf::FastMonitoringService*>(edm::Service<evf::FastMonitoringService>().operator->());
 
@@ -185,7 +187,6 @@ void RawEventFileWriterForBU::initialize(std::string const& destinationDir,
 void RawEventFileWriterForBU::writeJsds() {
   std::stringstream ss;
   ss << destinationDir_ << "/jsd";
-  mkdir(ss.str().c_str(), S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
 
   std::string rawJSDName = ss.str() + "/rawData.jsd";
   std::string eolJSDName = ss.str() + "/EoLS.jsd";
@@ -196,6 +197,21 @@ void RawEventFileWriterForBU::writeJsds() {
   runMon_->setDefPath(eorJSDName);
 
   struct stat fstat;
+
+  //only crete JSD definitions from process that created directory
+  std::string outdirpath = ss.str();
+  if (stat(outdirpath.c_str(), &fstat) == 0)
+    return;
+
+  auto retval = mkdir(outdirpath.c_str(), S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
+  if (retval) {
+    if (errno == EEXIST)
+      return;
+    else
+      throw cms::Exception("RawEventFileWriterForBU", "writeJsds")
+          << "Error creating directory " << outdirpath << " : " << strerror(errno);
+  }
+
   if (stat(rawJSDName.c_str(), &fstat) != 0) {
     std::string content;
     JSONSerializer::serialize(&rawJsonDef_, content);
@@ -223,7 +239,8 @@ void RawEventFileWriterForBU::finishFileWrite(unsigned int ls) {
     write(outfd_, (char*)&frdFileHeader, sizeof(FRDFileHeader_v1));
     closefd();
     //move raw file from open to run directory
-    rename(fileName_.c_str(), (destinationDir_ + fileName_.substr(fileName_.rfind('/'))).c_str());
+    if (!writeToOpen_)
+      rename(fileName_.c_str(), (destinationDir_ + fileName_.substr(fileName_.rfind('/'))).c_str());
 
     edm::LogInfo("RawEventFileWriterForBU")
         << "Wrote RAW input file: " << fileName_ << " with perFileEventCount = " << perFileEventCount_.value()
@@ -234,14 +251,16 @@ void RawEventFileWriterForBU::finishFileWrite(unsigned int ls) {
     write(outfd_, (char*)&frdFileHeader, sizeof(FRDFileHeader_v2));
     closefd();
     //move raw file from open to run directory
-    rename(fileName_.c_str(), (destinationDir_ + fileName_.substr(fileName_.rfind('/'))).c_str());
+    if (!writeToOpen_)
+      rename(fileName_.c_str(), (destinationDir_ + fileName_.substr(fileName_.rfind('/'))).c_str());
     edm::LogInfo("RawEventFileWriterForBU")
         << "Wrote RAW input file: " << fileName_ << " with perFileEventCount = " << perFileEventCount_.value()
         << " and size " << perFileSize_.value();
   } else {
     closefd();
     //move raw file from open to run directory
-    rename(fileName_.c_str(), (destinationDir_ + fileName_.substr(fileName_.rfind('/'))).c_str());
+    if (!writeToOpen_)
+      rename(fileName_.c_str(), (destinationDir_ + fileName_.substr(fileName_.rfind('/'))).c_str());
     //create equivalent JSON file
     //TODO:fix this to use DaqDirector convention and better extension replace
     std::filesystem::path source(fileName_);
@@ -252,7 +271,8 @@ void RawEventFileWriterForBU::finishFileWrite(unsigned int ls) {
     fileMon_->discardCollected(ls);
 
     //move the json file from open
-    rename(path.c_str(), (destinationDir_ + path.substr(path.rfind('/'))).c_str());
+    if (!writeToOpen_)
+      rename(path.c_str(), (destinationDir_ + path.substr(path.rfind('/'))).c_str());
 
     edm::LogInfo("RawEventFileWriterForBU")
         << "Wrote JSON input file: " << path << " with perFileEventCount = " << perFileEventCount_.value()
@@ -274,17 +294,23 @@ void RawEventFileWriterForBU::endOfLS(unsigned int ls) {
   }
   lumiMon_->snap(ls);
 
+  std::ostringstream ostrOpen;
   std::ostringstream ostr;
 
+  ostrOpen << destinationDir_ << "/open/" << runPrefix_ << "_ls" << std::setfill('0') << std::setw(4) << ls << "_EoLS"
+           << ".jsn";
   ostr << destinationDir_ << "/" << runPrefix_ << "_ls" << std::setfill('0') << std::setw(4) << ls << "_EoLS"
        << ".jsn";
   //outfd_ = open(ostr.str().c_str(), O_WRONLY | O_CREAT,  S_IWUSR | S_IRUSR | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
   //closefd();
 
-  std::string path = ostr.str();
-  lumiMon_->outputFullJSON(path, ls);
+  std::string pathOpen = ostrOpen.str();
+  lumiMon_->outputFullJSON(pathOpen, ls);
   lumiMon_->discardCollected(ls);
 
+  if (!writeToOpen_)
+    rename(pathOpen.c_str(), ostr.str().c_str());
+
   perRunEventCount_.value() += perLumiEventCount_.value();
   perRunTotalEventCount_.value() = perRunEventCount_.value();
   perRunFileCount_.value() += perLumiFileCount_.value();
@@ -301,16 +327,20 @@ void RawEventFileWriterForBU::endOfLS(unsigned int ls) {
 void RawEventFileWriterForBU::stop() {
   if (lumiOpen_ > lumiClosed_)
     endOfLS(lumiOpen_);
-  edm::LogInfo("RawEventFileWriterForBU") << "Writing EOR file!";
-  if (!destinationDir_.empty()) {
+  if (writeEoR_ && !destinationDir_.empty()) {
+    edm::LogInfo("RawEventFileWriterForBU") << "Writing EOR file!";
     // create EoR file
+    std::string pathOpen = destinationDir_ + "/open/" + runPrefix_ + "_ls0000_EoR.jsn";
     std::string path = destinationDir_ + "/" + runPrefix_ + "_ls0000_EoR.jsn";
     runMon_->snap(0);
-    runMon_->outputFullJSON(path, 0);
+    runMon_->outputFullJSON(pathOpen, 0);
+    if (!writeToOpen_)
+      rename(pathOpen.c_str(), path.c_str());
   }
 }
 
 void RawEventFileWriterForBU::extendDescription(edm::ParameterSetDescription& desc) {
   desc.add<int>("microSleep", 0);
   desc.add<unsigned int>("frdFileVersion", 0);
+  desc.addUntracked<bool>("writeEoR", true);
 }

EventFilter/Utilities/plugins/RawEventFileWriterForBU.h

@@ -88,6 +88,8 @@ class RawEventFileWriterForBU {
 
   int microSleep_;
   unsigned int frdFileVersion_;
+  bool writeEoR_;
+  bool writeToOpen_;
 
   edm::streamer::uint32 adlera_;
   edm::streamer::uint32 adlerb_;

EventFilter/Utilities/plugins/RawEventOutputModuleForBU.h

@@ -255,6 +255,7 @@ void RawEventOutputModuleForBU<Consumer>::fillDescriptions(edm::ConfigurationDes
   desc.addUntracked<std::string>("rawProductName", "FEDRawDataCollection")
       ->setComment("FEDRawDataCollection or RawDataBuffer");
   desc.addUntracked<std::vector<unsigned int>>("sourceIdList", std::vector<unsigned int>());
+  desc.addUntracked<bool>("writeToOpen", false);
   Consumer::extendDescription(desc);
 
   descriptions.addWithDefaultLabel(desc);

EventFilter/Utilities/src/EvFDaqDirector.cc

@@ -57,12 +57,9 @@ namespace evf {
         hltSourceDirectory_(pset.getUntrackedParameter<std::string>("hltSourceDirectory", "")),
         hostname_(""),
         bu_readlock_fd_(-1),
-        bu_writelock_fd_(-1),
         fu_readwritelock_fd_(-1),
         fulocal_rwlock_fd_(-1),
         fulocal_rwlock_fd2_(-1),
-        bu_w_lock_stream(nullptr),
-        bu_r_lock_stream(nullptr),
         fu_rw_lock_stream(nullptr),
         dirManager_(base_dir_),
         previousFileSize_(0),
@@ -74,7 +71,6 @@ namespace evf {
         fu_rw_fulk(make_flock(F_UNLCK, SEEK_SET, 0, 0, getpid())) {
     reg.watchPreallocate(this, &EvFDaqDirector::preallocate);
     reg.watchPreGlobalBeginRun(this, &EvFDaqDirector::preBeginRun);
-    reg.watchPostGlobalEndRun(this, &EvFDaqDirector::postEndRun);
     reg.watchPreGlobalEndLumi(this, &EvFDaqDirector::preGlobalEndLumi);
 
     //save hostname for later
@@ -216,7 +212,6 @@ namespace evf {
     //for BU, it is created at this point
     if (directorBU_) {
       bu_run_dir_ = base_dir_ + "/" + run_string_;
-      std::string bulockfile = bu_run_dir_ + "/bu.lock";
       fulockfile_ = bu_run_dir_ + "/fu.lock";
 
       //make or find bu run dir
@@ -232,30 +227,23 @@ namespace evf {
             << " Error creating bu run open dir -: " << bu_run_open_dir_ << " mkdir error:" << strerror(errno);
       }
 
-      // the BU director does not need to know about the fu lock
-      bu_writelock_fd_ = open(bulockfile.c_str(), O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU);
-      if (bu_writelock_fd_ == -1)
-        edm::LogWarning("EvFDaqDirector") << "problem with creating filedesc for buwritelock -: " << strerror(errno);
-      else
-        edm::LogInfo("EvFDaqDirector") << "creating filedesc for buwritelock -: " << bu_writelock_fd_;
-      bu_w_lock_stream = fdopen(bu_writelock_fd_, "w");
-      if (bu_w_lock_stream == nullptr)
-        edm::LogWarning("EvFDaqDirector") << "Error creating write lock stream -: " << strerror(errno);
-
-      // BU INITIALIZES LOCK FILE
-      // FU LOCK FILE OPEN
-      openFULockfileStream(true);
-      tryInitializeFuLockFile();
-      fflush(fu_rw_lock_stream);
-      close(fu_readwritelock_fd_);
-
       if (!hltSourceDirectory_.empty()) {
         struct stat buf;
         if (stat(hltSourceDirectory_.c_str(), &buf) == 0) {
           std::string hltdir = bu_run_dir_ + "/hlt";
-          std::string tmphltdir = bu_run_open_dir_ + "/hlt";
+          if (!stat(hltdir.c_str(), &buf)) {
+            edm::LogInfo("EvFDaqDirector") << "hlt directory already exists";
+            return;
+          }
+
+          timeval ts_temp;
+          gettimeofday(&ts_temp, nullptr);
+          std::stringstream ss;
+          ss << bu_run_open_dir_ << "/hlt" << getpid() << "_" << ts_temp.tv_sec << "_" << ts_temp.tv_usec;
+          std::string tmphltdir = ss.str();
+          //this directory should be unique
           retval = mkdir(tmphltdir.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
-          if (retval != 0 && errno != EEXIST)
+          if (retval)
             throw cms::Exception("DaqDirector")
                 << " Error creating bu run dir -: " << hltdir << " mkdir error:" << strerror(errno);
 
@@ -274,8 +262,12 @@ namespace evf {
             } catch (...) {
             }
           }
-
-          std::filesystem::rename(tmphltdir, hltdir);
+          try {
+            std::filesystem::rename(tmphltdir, hltdir);
+          } catch (std::filesystem::filesystem_error& e) {
+            if (e.code() != std::errc::file_exists)
+              throw e;
+          }
         } else
           throw cms::Exception("DaqDirector") << " Error looking for HLT configuration -: " << hltSourceDirectory_;
       }
@@ -431,15 +423,6 @@ namespace evf {
     }
   }
 
-  void EvFDaqDirector::postEndRun(edm::GlobalContext const& globalContext) {
-    close(bu_readlock_fd_);
-    close(bu_writelock_fd_);
-    if (directorBU_) {
-      std::string filename = bu_run_dir_ + "/bu.lock";
-      removeFile(filename);
-    }
-  }
-
   void EvFDaqDirector::preGlobalEndLumi(edm::GlobalContext const& globalContext) {
     lsWithFilesMap_.erase(globalContext.luminosityBlockID().luminosityBlock());
   }

EventFilter/Utilities/test/startBU.py

@@ -11,7 +11,7 @@
 options = VarParsing.VarParsing ('analysis')
 
 options.register ('runNumber',
-                  100, # default value
+                  100101, # default value
                   VarParsing.VarParsing.multiplicity.singleton,
                   VarParsing.VarParsing.varType.int,          # string, int, or float
                   "Run Number")
@@ -70,13 +70,19 @@
                   VarParsing.VarParsing.varType.string,          # string, int, or float
                   "List of generated subsystem FEDs. Empty means all.")
 
-
 options.register ('conversionTest',
                   False,
                   VarParsing.VarParsing.multiplicity.singleton,
                   VarParsing.VarParsing.varType.bool,
                   "Test conversion between new and old format")
 
+options.register ('writeToOpen',
+                  0,
+                  VarParsing.VarParsing.multiplicity.singleton,
+                  VarParsing.VarParsing.varType.int,          # string, int, or float
+                  "Write only to open directory")
+
+
 
 options.parseArguments()
 
@@ -155,6 +161,7 @@
         numEventsPerFile = cms.uint32(options.eventsPerFile),
         frdVersion = cms.uint32(6),
         frdFileVersion = cms.uint32(options.frdFileVersion),
+        writeToOpen = cms.untracked.bool(True if options.writeToOpen else False)
         )
 
 elif  options.dataType == "DTH":