CMS 3D CMS Logo

All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros Groups Pages
EvFOutputModule.cc
Go to the documentation of this file.
2 
4 
8 
12 
16 
19 
20 #include <sys/stat.h>
21 #include <filesystem>
22 #include <boost/algorithm/string.hpp>
23 
24 namespace evf {
25 
28  std::string const& streamLabel)
29  : streamerCommon_(ps, selections),
30  processed_(0),
31  accepted_(0),
32  errorEvents_(0),
33  retCodeMask_(0),
34  filelist_(),
35  filesize_(0),
36  inputFiles_(),
37  fileAdler32_(1),
38  hltErrorEvents_(0) {
39  transferDestination_ = edm::Service<evf::EvFDaqDirector>()->getStreamDestinations(streamLabel);
40  mergeType_ = edm::Service<evf::EvFDaqDirector>()->getStreamMergeType(streamLabel, evf::MergeTypeDAT);
41 
42  std::string baseRunDir = edm::Service<evf::EvFDaqDirector>()->baseRunDir();
43  LogDebug("EvFOutputModule") << "writing .dat files to -: " << baseRunDir;
44 
45  edm::Service<evf::EvFDaqDirector>()->createRunOpendirMaybe();
46 
47  processed_.setName("Processed");
48  accepted_.setName("Accepted");
49  errorEvents_.setName("ErrorEvents");
50  retCodeMask_.setName("ReturnCodeMask");
51  filelist_.setName("Filelist");
52  filesize_.setName("Filesize");
53  inputFiles_.setName("InputFiles");
54  fileAdler32_.setName("FileAdler32");
55  transferDestination_.setName("TransferDestination");
56  mergeType_.setName("MergeType");
57  hltErrorEvents_.setName("HLTErrorEvents");
58 
68  outJsonDef_.addLegendItem("TransferDestination", "string", jsoncollector::DataPointDefinition::SAME);
71 
72  std::stringstream tmpss, ss;
73  tmpss << baseRunDir << "/open/"
74  << "output_" << getpid() << ".jsd";
75  ss << baseRunDir << "/"
76  << "output_" << getpid() << ".jsd";
77  std::string outTmpJsonDefName = tmpss.str();
78  std::string outJsonDefName = ss.str();
79 
80  edm::Service<evf::EvFDaqDirector>()->lockInitLock();
81  struct stat fstat;
82  if (stat(outJsonDefName.c_str(), &fstat) != 0) { //file does not exist
83  LogDebug("EvFOutputModule") << "writing output definition file -: " << outJsonDefName;
86  jsoncollector::FileIO::writeStringToFile(outTmpJsonDefName, content);
87  std::filesystem::rename(outTmpJsonDefName, outJsonDefName);
88  }
89  edm::Service<evf::EvFDaqDirector>()->unlockInitLock();
90 
92  jsonMonitor_->setDefPath(outJsonDefName);
93  jsonMonitor_->registerGlobalMonitorable(&processed_, false);
94  jsonMonitor_->registerGlobalMonitorable(&accepted_, false);
95  jsonMonitor_->registerGlobalMonitorable(&errorEvents_, false);
96  jsonMonitor_->registerGlobalMonitorable(&retCodeMask_, false);
97  jsonMonitor_->registerGlobalMonitorable(&filelist_, false);
98  jsonMonitor_->registerGlobalMonitorable(&filesize_, false);
99  jsonMonitor_->registerGlobalMonitorable(&inputFiles_, false);
100  jsonMonitor_->registerGlobalMonitorable(&fileAdler32_, false);
101  jsonMonitor_->registerGlobalMonitorable(&transferDestination_, false);
102  jsonMonitor_->registerGlobalMonitorable(&mergeType_, false);
103  jsonMonitor_->registerGlobalMonitorable(&hltErrorEvents_, false);
104  jsonMonitor_->commit(nullptr);
105  }
106 
108  : edm::one::OutputModuleBase(ps),
110  ps_(ps),
111  streamLabel_(ps.getParameter<std::string>("@module_label")),
112  trToken_(consumes<edm::TriggerResults>(edm::InputTag("TriggerResults"))),
113  psetToken_(consumes<edm::SendJobHeader::ParameterSetMap, edm::InRun>(
114  ps.getUntrackedParameter<edm::InputTag>("psetMap"))) {
115  //replace hltOutoputA with stream if the HLT menu uses this convention
116  std::string testPrefix = "hltOutput";
117  if (streamLabel_.find(testPrefix) == 0)
118  streamLabel_ = std::string("stream") + streamLabel_.substr(testPrefix.size());
119 
120  if (streamLabel_.find('_') != std::string::npos) {
121  throw cms::Exception("EvFOutputModule") << "Underscore character is reserved can not be used for stream names in "
122  "FFF, but was detected in stream name -: "
123  << streamLabel_;
124  }
125 
126  std::string streamLabelLow = streamLabel_;
127  boost::algorithm::to_lower(streamLabelLow);
128  auto streampos = streamLabelLow.rfind("stream");
129  if (streampos != 0 && streampos != std::string::npos)
130  throw cms::Exception("EvFOutputModule")
131  << "stream (case-insensitive) sequence was found in stream suffix. This is reserved and can not be used for "
132  "names in FFF based HLT, but was detected in stream name";
133 
135  }
136 
138 
143  desc.addUntracked<edm::InputTag>("psetMap", {"hltPSetMap"})
144  ->setComment("Optionally allow the map of ParameterSets to be calculated externally.");
145  descriptions.add("evfOutputModule", desc);
146  }
147 
149  //create run Cache holding JSON file writer and variables
150  jsonWriter_ = std::make_unique<EvFOutputJSONWriter>(ps_, &keptProducts()[edm::InEvent], streamLabel_);
151 
152  //output INI file (non-const). This doesn't require globalBeginRun to be finished
153  const std::string openIniFileName = edm::Service<evf::EvFDaqDirector>()->getOpenInitFilePath(streamLabel_);
154  edm::LogInfo("EvFOutputModule") << "beginRun init stream -: " << openIniFileName;
155 
156  StreamerOutputFile stream_writer_preamble(openIniFileName);
157  uint32 preamble_adler32 = 1;
158  edm::BranchIDLists const* bidlPtr = branchIDLists();
159 
160  auto psetMapHandle = run.getHandle(psetToken_);
161 
162  std::unique_ptr<InitMsgBuilder> init_message =
163  jsonWriter_->streamerCommon_.serializeRegistry(*jsonWriter_->streamerCommon_.getSerializerBuffer(),
164  *bidlPtr,
169  psetMapHandle.isValid() ? psetMapHandle.product() : nullptr);
170 
171  //Let us turn it into a View
172  InitMsgView view(init_message->startAddress());
173 
174  //output header
175  stream_writer_preamble.write(view);
176  preamble_adler32 = stream_writer_preamble.adler32();
177  stream_writer_preamble.close();
178 
179  struct stat istat;
180  stat(openIniFileName.c_str(), &istat);
181  //read back file to check integrity of what was written
182  off_t readInput = 0;
183  uint32_t adlera = 1, adlerb = 0;
184  FILE* src = fopen(openIniFileName.c_str(), "r");
185 
186  //allocate buffer to write INI file
187  unsigned char* outBuf = new unsigned char[1024 * 1024];
188  while (readInput < istat.st_size) {
189  size_t toRead = readInput + 1024 * 1024 < istat.st_size ? 1024 * 1024 : istat.st_size - readInput;
190  fread(outBuf, toRead, 1, src);
191  cms::Adler32((const char*)outBuf, toRead, adlera, adlerb);
192  readInput += toRead;
193  }
194  fclose(src);
195 
196  //clear serialization buffers
197  jsonWriter_->streamerCommon_.getSerializerBuffer()->clearHeaderBuffer();
198 
199  //free output buffer needed only for the file write
200  delete[] outBuf;
201  outBuf = nullptr;
202 
203  uint32_t adler32c = (adlerb << 16) | adlera;
204  if (adler32c != preamble_adler32) {
205  throw cms::Exception("EvFOutputModule") << "Checksum mismatch of ini file -: " << openIniFileName
206  << " expected:" << preamble_adler32 << " obtained:" << adler32c;
207  } else {
208  LogDebug("EvFOutputModule") << "Ini file checksum -: " << streamLabel_ << " " << adler32c;
209  std::filesystem::rename(openIniFileName, edm::Service<evf::EvFDaqDirector>()->getInitFilePath(streamLabel_));
210  }
211  }
212 
214  edm::EventForOutput const& e) const {
215  Trig result;
217  return result;
218  }
219 
220  std::shared_ptr<EvFOutputEventWriter> EvFOutputModule::globalBeginLuminosityBlock(
221  edm::LuminosityBlockForOutput const& iLB) const {
222  auto openDatFilePath = edm::Service<evf::EvFDaqDirector>()->getOpenDatFilePath(iLB.luminosityBlock(), streamLabel_);
223 
224  return std::make_shared<EvFOutputEventWriter>(openDatFilePath);
225  }
226 
229 
230  //auto lumiWriter = const_cast<EvFOutputEventWriter*>(luminosityBlockCache(e.getLuminosityBlock().index() ));
231  auto lumiWriter = luminosityBlockCache(e.getLuminosityBlock().index());
232  std::unique_ptr<EventMsgBuilder> msg = jsonWriter_->streamerCommon_.serializeEvent(
233  *jsonWriter_->streamerCommon_.getSerializerBuffer(), e, triggerResults, selectorConfig());
234  lumiWriter->incAccepted();
235  lumiWriter->doOutputEvent(*msg); //msg is written and discarded at this point
236  }
237 
239  auto lumiWriter = luminosityBlockCache(iLB.index());
240  //close dat file
241  lumiWriter->close();
242 
243  jsonWriter_->fileAdler32_.value() = lumiWriter->get_adler32();
244  jsonWriter_->accepted_.value() = lumiWriter->getAccepted();
245 
246  bool abortFlag = false;
247  jsonWriter_->processed_.value() = fms_->getEventsProcessedForLumi(iLB.luminosityBlock(), &abortFlag);
248  if (abortFlag) {
249  edm::LogInfo("EvFOutputModule") << "Abort flag has been set. Output is suppressed";
250  return;
251  }
252 
253  if (jsonWriter_->processed_.value() != 0) {
254  struct stat istat;
255  std::filesystem::path openDatFilePath = lumiWriter->getFilePath();
256  stat(openDatFilePath.string().c_str(), &istat);
257  jsonWriter_->filesize_ = istat.st_size;
258  std::filesystem::rename(openDatFilePath.string().c_str(),
260  jsonWriter_->filelist_ = openDatFilePath.filename().string();
261  } else {
262  //remove empty file when no event processing has occurred
263  remove(lumiWriter->getFilePath().c_str());
264  jsonWriter_->filesize_ = 0;
265  jsonWriter_->filelist_ = "";
266  jsonWriter_->fileAdler32_.value() = -1; //no files in signed long
267  }
268 
269  //produce JSON file
270  jsonWriter_->jsonMonitor_->snap(iLB.luminosityBlock());
271  const std::string outputJsonNameStream =
272  edm::Service<evf::EvFDaqDirector>()->getOutputJsonFilePath(iLB.luminosityBlock(), streamLabel_);
273  jsonWriter_->jsonMonitor_->outputFullJSON(outputJsonNameStream, iLB.luminosityBlock());
274  }
275 
276 } // namespace evf
void addLegendItem(std::string const &name, std::string const &type, std::string const &operation)
ParameterSetID const & mainParameterSetID() const
EvFOutputModule(edm::ParameterSet const &ps)
std::unique_ptr< evf::EvFOutputJSONWriter > jsonWriter_
std::vector< BranchIDList > BranchIDLists
Definition: BranchIDList.h:19
edm::ParameterSet const & ps_
jsoncollector::IntJ fileAdler32_
ParameterDescriptionBase * addUntracked(U const &iLabel, T const &value)
std::shared_ptr< EvFOutputEventWriter > globalBeginLuminosityBlock(edm::LuminosityBlockForOutput const &iLB) const override
static bool serialize(JsonSerializable *pObj, std::string &output)
void write(const InitMsgBuilder &)
EvFOutputJSONWriter(edm::ParameterSet const &ps, edm::SelectedProducts const *selections, std::string const &streamLabel)
std::string to_lower(const std::string &s)
jsoncollector::DataPointDefinition outJsonDef_
jsoncollector::IntJ hltErrorEvents_
std::string const & moduleLabel() const
jsoncollector::StringJ inputFiles_
void beginRun(edm::RunForOutput const &run) override
Handle< PROD > getHandle(EDGetTokenT< PROD > token) const
tuple result
Definition: mps_fire.py:311
LuminosityBlockIndex index() const
std::vector< std::pair< BranchDescription const *, EDGetToken > > SelectedProducts
LuminosityBlockForOutput const & getLuminosityBlock() const
LuminosityBlockNumber_t luminosityBlock() const
jsoncollector::IntJ retCodeMask_
std::string streamLabel_
static void fillDescription(ParameterSetDescription &desc)
virtual void setName(std::string name)
ModuleDescription const & description() const
void globalEndLuminosityBlock(edm::LuminosityBlockForOutput const &iLB) override
static std::string const triggerResults
Definition: EdmProvDump.cc:44
ThinnedAssociationsHelper const * thinnedAssociationsHelper() const
static void writeStringToFile(std::string const &filename, std::string &content)
Definition: FileIO.cc:21
unsigned int uint32
Definition: MsgTools.h:13
jsoncollector::IntJ processed_
static void fillDescriptions(edm::ConfigurationDescriptions &descriptions)
BranchIDLists const * branchIDLists() const
Log< level::Info, false > LogInfo
void Adler32(char const *data, size_t len, uint32_t &a, uint32_t &b)
void write(edm::EventForOutput const &e) override
uint32 adler32() const
ParameterSetID selectorConfig() const
edm::EDGetTokenT< edm::TriggerResults > trToken_
void add(std::string const &label, ParameterSetDescription const &psetDescription)
tuple msg
Definition: mps_check.py:285
BasicHandle getByToken(EDGetToken token, TypeID const &typeID) const
std::map< ParameterSetID, ParameterSetBlob > ParameterSetMap
jsoncollector::IntJ filesize_
jsoncollector::StringJ filelist_
jsoncollector::StringJ transferDestination_
bool isValid() const
Definition: Hash.h:141
edm::EDGetTokenT< edm::SendJobHeader::ParameterSetMap > psetToken_
std::shared_ptr< jsoncollector::FastMonitor > jsonMonitor_
SelectedProductsForBranchType const & keptProducts() const
const ModuleDescription & moduleDescription() const
unsigned int getEventsProcessedForLumi(unsigned int lumi, bool *abortFlag=nullptr)
void setDefaultGroup(std::string const &group)
jsoncollector::IntJ accepted_
jsoncollector::StringJ mergeType_
static void fillDescription(ParameterSetDescription &desc, std::vector< std::string > const &iDefaultOutputCommands=ProductSelectorRules::defaultSelectionStrings())
jsoncollector::IntJ errorEvents_
Trig getTriggerResults(edm::EDGetTokenT< edm::TriggerResults > const &token, edm::EventForOutput const &e) const
evf::FastMonitoringService * fms_
#define LogDebug(id)