Utilities/plugins/RawEventFileWriterForBU.cc

0001 #include <cerrno>
0002 #include <cstdio>
0003 #include <cstdlib>
0004 #include <cstring>
0005 #include <iostream>
0006 #include <sstream>
0007
0008 // CMSSW headers
0009 #include "EventFilter/Utilities/interface/EvFDaqDirector.h"
0010 #include "EventFilter/Utilities/interface/FileIO.h"
0011 #include "EventFilter/Utilities/interface/JSONSerializer.h"
0012 #include "EventFilter/Utilities/plugins/RawEventFileWriterForBU.h"
0013 #include "FWCore/MessageLogger/interface/MessageLogger.h"
0014 #include "FWCore/Utilities/interface/Adler32Calculator.h"
0015 #include "FWCore/Utilities/interface/Exception.h"
0016 #include "IOPool/Streamer/interface/FRDEventMessage.h"
0017 #include "IOPool/Streamer/interface/FRDFileHeader.h"
0018
0019 using namespace jsoncollector;
0020
0021 //TODO:get run directory information from DaqDirector
0022
0023 RawEventFileWriterForBU::RawEventFileWriterForBU(edm::ParameterSet const& ps)
0024     : microSleep_(ps.getParameter<int>("microSleep")),
0025       frdFileVersion_(ps.getParameter<unsigned int>("frdFileVersion")) {
0026   //per-file JSD and FastMonitor
0027   rawJsonDef_.setDefaultGroup("legend");
0028   rawJsonDef_.addLegendItem("NEvents", "integer", DataPointDefinition::SUM);
0029
0030   perFileEventCount_.setName("NEvents");
0031   perFileSize_.setName("NBytes");
0032
0033   fileMon_ = new FastMonitor(&rawJsonDef_, false);
0034   fileMon_->registerGlobalMonitorable(&perFileEventCount_, false, nullptr);
0035   fileMon_->registerGlobalMonitorable(&perFileSize_, false, nullptr);
0036   fileMon_->commit(nullptr);
0037
0038   //per-lumi JSD and FastMonitor
0039   eolJsonDef_.setDefaultGroup("legend");
0040   eolJsonDef_.addLegendItem("NEvents", "integer", DataPointDefinition::SUM);
0041   eolJsonDef_.addLegendItem("NFiles", "integer", DataPointDefinition::SUM);
0042   eolJsonDef_.addLegendItem("TotalEvents", "integer", DataPointDefinition::SUM);
0043   eolJsonDef_.addLegendItem("NLostEvents", "integer", DataPointDefinition::SUM);
0044
0045   perLumiEventCount_.setName("NEvents");
0046   perLumiFileCount_.setName("NFiles");
0047   perLumiTotalEventCount_.setName("TotalEvents");
0048   perLumiLostEventCount_.setName("NLostEvents");
0049   perLumiSize_.setName("NBytes");
0050
0051   lumiMon_ = new FastMonitor(&eolJsonDef_, false);
0052   lumiMon_->registerGlobalMonitorable(&perLumiEventCount_, false, nullptr);
0053   lumiMon_->registerGlobalMonitorable(&perLumiFileCount_, false, nullptr);
0054   lumiMon_->registerGlobalMonitorable(&perLumiTotalEventCount_, false, nullptr);
0055   lumiMon_->registerGlobalMonitorable(&perLumiLostEventCount_, false, nullptr);
0056   lumiMon_->registerGlobalMonitorable(&perLumiSize_, false, nullptr);
0057   lumiMon_->commit(nullptr);
0058
0059   //per-run JSD and FastMonitor
0060   eorJsonDef_.setDefaultGroup("legend");
0061   eorJsonDef_.addLegendItem("NEvents", "integer", DataPointDefinition::SUM);
0062   eorJsonDef_.addLegendItem("NFiles", "integer", DataPointDefinition::SUM);
0063   eorJsonDef_.addLegendItem("NLumis", "integer", DataPointDefinition::SUM);
0064   eorJsonDef_.addLegendItem("LastLumi", "integer", DataPointDefinition::SUM);
0065
0066   perRunEventCount_.setName("NEvents");
0067   perRunFileCount_.setName("NFiles");
0068   perRunLumiCount_.setName("NLumis");
0069   perRunLastLumi_.setName("LastLumi");
0070
0071   runMon_ = new FastMonitor(&eorJsonDef_, false);
0072   runMon_->registerGlobalMonitorable(&perRunEventCount_, false, nullptr);
0073   runMon_->registerGlobalMonitorable(&perRunFileCount_, false, nullptr);
0074   runMon_->registerGlobalMonitorable(&perRunLumiCount_, false, nullptr);
0075   runMon_->registerGlobalMonitorable(&perRunLastLumi_, false, nullptr);
0076   runMon_->commit(nullptr);
0077 }
0078
0079 RawEventFileWriterForBU::RawEventFileWriterForBU(std::string const& fileName) {}
0080
0081 RawEventFileWriterForBU::~RawEventFileWriterForBU() {
0082   delete fileMon_;
0083   delete lumiMon_;
0084   delete runMon_;
0085 }
0086
0087 void RawEventFileWriterForBU::doOutputEvent(FRDEventMsgView const& msg) {
0088   ssize_t retval = write(outfd_, (void*)msg.startAddress(), msg.size());
0089
0090   if ((unsigned)retval != msg.size()) {
0091     throw cms::Exception("RawEventFileWriterForBU", "doOutputEvent")
0092         << "Error writing FED Raw Data event data to " << fileName_ << ".  Possibly the output disk "
0093         << "is full?" << std::endl;
0094   }
0095
0096   // throttle event output
0097   usleep(microSleep_);
0098   perFileEventCount_.value()++;
0099   perFileSize_.value() += msg.size();
0100
0101   //  cms::Adler32((const char*) msg.startAddress(), msg.size(), adlera_, adlerb_);
0102 }
0103
0104 void RawEventFileWriterForBU::initialize(std::string const& destinationDir, std::string const& name, int ls) {
0105   destinationDir_ = destinationDir;
0106
0107   if (outfd_ != -1) {
0108     finishFileWrite(ls);
0109     closefd();
0110   }
0111
0112   fileName_ = name;
0113
0114   if (!writtenJSDs_) {
0115     writeJsds();
0116     /*    std::stringstream ss;
0117     ss << destinationDir_ << "/jsd";
0118     mkdir(ss.str().c_str(), S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
0119
0120     std::string rawJSDName = ss.str()+"/rawData.jsd";
0121     std::string eolJSDName = ss.str()+"/EoLS.jsd";
0122     std::string eorJSDName = ss.str()+"/EoR.jsd";
0123
0124     fileMon_->setDefPath(rawJSDName);
0125     lumiMon_->setDefPath(eolJSDName);
0126     runMon_->setDefPath(eorJSDName);
0127
0128     struct stat   fstat;
0129     if (stat (rawJSDName.c_str(), &fstat) != 0) {
0130       std::string content;
0131       JSONSerializer::serialize(&rawJsonDef_,content);
0132       FileIO::writeStringToFile(rawJSDName, content);
0133     }
0134
0135     if (stat (eolJSDName.c_str(), &fstat) != 0) {
0136       std::string content;
0137       JSONSerializer::serialize(&eolJsonDef_,content);
0138       FileIO::writeStringToFile(eolJSDName, content);
0139     }
0140
0141     if (stat (eorJSDName.c_str(), &fstat) != 0) {
0142       std::string content;
0143       JSONSerializer::serialize(&eorJsonDef_,content);
0144       FileIO::writeStringToFile(eorJSDName, content);
0145     }
0146 */
0147     writtenJSDs_ = true;
0148   }
0149
0150   outfd_ = open(fileName_.c_str(), O_WRONLY | O_CREAT, S_IRWXU | S_IRWXG | S_IRWXO);
0151   edm::LogInfo("RawEventFileWriterForBU") << " opened " << fileName_;
0152
0153   if (outfd_ < 0) {  //attention here... it may happen that outfd_ is *not* set (e.g. missing initialize call...)
0154     throw cms::Exception("RawEventFileWriterForBU", "initialize")
0155         << "Error opening FED Raw Data event output file: " << name << ": " << strerror(errno) << "\n";
0156   }
0157
0158   perFileEventCount_.value() = 0;
0159   perFileSize_.value() = 0;
0160
0161   adlera_ = 1;
0162   adlerb_ = 0;
0163
0164   if (frdFileVersion_ > 0) {
0165     assert(frdFileVersion_ == 1);
0166     //reserve space for file header
0167     ftruncate(outfd_, sizeof(FRDFileHeader_v1));
0168     lseek(outfd_, sizeof(FRDFileHeader_v1), SEEK_SET);
0169     perFileSize_.value() = sizeof(FRDFileHeader_v1);
0170   }
0171 }
0172
0173 void RawEventFileWriterForBU::writeJsds() {
0174   std::stringstream ss;
0175   ss << destinationDir_ << "/jsd";
0176   mkdir(ss.str().c_str(), S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
0177
0178   std::string rawJSDName = ss.str() + "/rawData.jsd";
0179   std::string eolJSDName = ss.str() + "/EoLS.jsd";
0180   std::string eorJSDName = ss.str() + "/EoR.jsd";
0181
0182   fileMon_->setDefPath(rawJSDName);
0183   lumiMon_->setDefPath(eolJSDName);
0184   runMon_->setDefPath(eorJSDName);
0185
0186   struct stat fstat;
0187   if (stat(rawJSDName.c_str(), &fstat) != 0) {
0188     std::string content;
0189     JSONSerializer::serialize(&rawJsonDef_, content);
0190     FileIO::writeStringToFile(rawJSDName, content);
0191   }
0192
0193   if (stat(eolJSDName.c_str(), &fstat) != 0) {
0194     std::string content;
0195     JSONSerializer::serialize(&eolJsonDef_, content);
0196     FileIO::writeStringToFile(eolJSDName, content);
0197   }
0198
0199   if (stat(eorJSDName.c_str(), &fstat) != 0) {
0200     std::string content;
0201     JSONSerializer::serialize(&eorJsonDef_, content);
0202     FileIO::writeStringToFile(eorJSDName, content);
0203   }
0204 }
0205
0206 void RawEventFileWriterForBU::finishFileWrite(int ls) {
0207   if (frdFileVersion_ > 0) {
0208     //rewind
0209     lseek(outfd_, 0, SEEK_SET);
0210     FRDFileHeader_v1 frdFileHeader(perFileEventCount_.value(), (uint32_t)ls, perFileSize_.value());
0211     write(outfd_, (char*)&frdFileHeader, sizeof(FRDFileHeader_v1));
0212     closefd();
0213     //move raw file from open to run directory
0214     rename(fileName_.c_str(), (destinationDir_ + fileName_.substr(fileName_.rfind('/'))).c_str());
0215
0216     edm::LogInfo("RawEventFileWriterForBU")
0217         << "Wrote RAW input file: " << fileName_ << " with perFileEventCount = " << perFileEventCount_.value()
0218         << " and size " << perFileSize_.value();
0219   } else {
0220     closefd();
0221     //move raw file from open to run directory
0222     rename(fileName_.c_str(), (destinationDir_ + fileName_.substr(fileName_.rfind('/'))).c_str());
0223     //create equivalent JSON file
0224     //TODO:fix this to use DaqDirector convention and better extension replace
0225     std::filesystem::path source(fileName_);
0226     std::string path = source.replace_extension(".jsn").string();
0227
0228     fileMon_->snap(ls);
0229     fileMon_->outputFullJSON(path, ls);
0230     fileMon_->discardCollected(ls);
0231
0232     //move the json file from open
0233     rename(path.c_str(), (destinationDir_ + path.substr(path.rfind('/'))).c_str());
0234
0235     edm::LogInfo("RawEventFileWriterForBU")
0236         << "Wrote JSON input file: " << path << " with perFileEventCount = " << perFileEventCount_.value()
0237         << " and size " << perFileSize_.value();
0238   }
0239   //there is a small chance that script gets interrupted while this isn't consistent (non-atomic)
0240   perLumiFileCount_.value()++;
0241   perLumiEventCount_.value() += perFileEventCount_.value();
0242   perLumiSize_.value() += perFileSize_.value();
0243   perLumiTotalEventCount_.value() += perFileEventCount_.value();
0244   //update open lumi value when first file is completed
0245   lumiOpen_ = ls;
0246 }
0247
0248 void RawEventFileWriterForBU::endOfLS(int ls) {
0249   if (outfd_ != -1) {
0250     finishFileWrite(ls);
0251     closefd();
0252   }
0253   lumiMon_->snap(ls);
0254
0255   std::ostringstream ostr;
0256
0257   if (run_ == -1)
0258     makeRunPrefix(destinationDir_);
0259
0260   ostr << destinationDir_ << "/" << runPrefix_ << "_ls" << std::setfill('0') << std::setw(4) << ls << "_EoLS"
0261        << ".jsn";
0262   //outfd_ = open(ostr.str().c_str(), O_WRONLY | O_CREAT,  S_IWUSR | S_IRUSR | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
0263   //closefd();
0264
0265   std::string path = ostr.str();
0266   lumiMon_->outputFullJSON(path, ls);
0267   lumiMon_->discardCollected(ls);
0268
0269   perRunEventCount_.value() += perLumiEventCount_.value();
0270   perRunFileCount_.value() += perLumiFileCount_.value();
0271   perRunLumiCount_.value() += 1;
0272   perRunLastLumi_.value() = ls;
0273
0274   perLumiEventCount_ = 0;
0275   perLumiFileCount_ = 0;
0276   perLumiTotalEventCount_ = 0;
0277   perLumiSize_ = 0;
0278   lumiClosed_ = ls;
0279 }
0280
0281 void RawEventFileWriterForBU::stop() {
0282   if (lumiOpen_ > lumiClosed_)
0283     endOfLS(lumiOpen_);
0284   edm::LogInfo("RawEventFileWriterForBU") << "Writing EOR file!";
0285   if (!destinationDir_.empty()) {
0286     // create EoR file
0287     if (run_ == -1)
0288       makeRunPrefix(destinationDir_);
0289     std::string path = destinationDir_ + "/" + runPrefix_ + "_ls0000_EoR.jsn";
0290     runMon_->snap(0);
0291     runMon_->outputFullJSON(path, 0);
0292   }
0293 }
0294
0295 //TODO:get from DaqDirector !
0296 void RawEventFileWriterForBU::makeRunPrefix(std::string const& destinationDir) {
0297   //dirty hack: extract run number from destination directory
0298   std::string::size_type pos = destinationDir.find("run");
0299   std::string run = destinationDir.substr(pos + 3);
0300   run_ = atoi(run.c_str());
0301   std::stringstream ss;
0302   ss << "run" << std::setfill('0') << std::setw(6) << run_;
0303   runPrefix_ = ss.str();
0304 }
0305
0306 void RawEventFileWriterForBU::extendDescription(edm::ParameterSetDescription& desc) {
0307   desc.add<int>("microSleep", 0);
0308   desc.add<unsigned int>("frdFileVersion", 0);
0309 }