Utilities/src/DAQSourceModelsScoutingRun3.cc

0001 #include "EventFilter//Utilities/interface/DAQSourceModelsScoutingRun3.h"
0002
0003 using namespace edm::streamer;
0004
0005 void DataModeScoutingRun3::makeDirectoryEntries(std::vector<std::string> const& baseDirs,
0006                                                 std::vector<int> const& numSources,
0007                                                 std::string const& runDir) {
0008   std::filesystem::path runDirP(runDir);
0009   for (auto& baseDir : baseDirs) {
0010     std::filesystem::path baseDirP(baseDir);
0011     buPaths_.emplace_back(baseDirP / runDirP);
0012   }
0013
0014   // store the number of sources in each BU
0015   buNumSources_ = numSources;
0016 }
0017
0018 std::pair<bool, std::vector<std::string>> DataModeScoutingRun3::defineAdditionalFiles(std::string const& primaryName,
0019                                                                                       bool fileListMode) const {
0020   std::vector<std::string> additionalFiles;
0021
0022   if (fileListMode) {
0023     // Expected file naming when working in file list mode
0024     for (int j = 1; j < buNumSources_[0]; j++) {
0025       additionalFiles.push_back(primaryName + "_" + std::to_string(j));
0026     }
0027     return std::make_pair(true, additionalFiles);
0028   }
0029
0030   auto fullpath = std::filesystem::path(primaryName);
0031   auto fullname = fullpath.filename();
0032
0033   for (size_t i = 0; i < buPaths_.size(); i++) {
0034     std::filesystem::path newPath = buPaths_[i] / fullname;
0035
0036     if (i != 0) {
0037       // secondary files from other ramdisks
0038       additionalFiles.push_back(newPath.generic_string());
0039     }
0040
0041     // add extra sources from the same ramdisk
0042     for (int j = 1; j < buNumSources_[i]; j++) {
0043       additionalFiles.push_back(newPath.generic_string() + "_" + std::to_string(j));
0044     }
0045   }
0046   return std::make_pair(true, additionalFiles);
0047 }
0048
0049 void DataModeScoutingRun3::readEvent(edm::EventPrincipal& eventPrincipal) {
0050   assert(!events_.empty());
0051
0052   edm::TimeValue_t time;
0053   timeval stv;
0054   gettimeofday(&stv, nullptr);
0055   time = stv.tv_sec;
0056   time = (time << 32) + stv.tv_usec;
0057   edm::Timestamp tstamp(time);
0058
0059   // set provenance helpers
0060   uint32_t hdrEventID = currOrbit_;
0061   edm::EventID eventID = edm::EventID(daqSource_->eventRunNumber(), daqSource_->currentLumiSection(), hdrEventID);
0062   edm::EventAuxiliary aux(
0063       eventID, daqSource_->processGUID(), tstamp, events_[0]->isRealData(), edm::EventAuxiliary::PhysicsTrigger);
0064
0065   aux.setProcessHistoryID(daqSource_->processHistoryID());
0066   daqSource_->makeEventWrapper(eventPrincipal, aux);
0067
0068   // create scouting raw data collection
0069   std::unique_ptr<SDSRawDataCollection> rawData(new SDSRawDataCollection);
0070
0071   // Fill the ScoutingRawDataCollection with valid orbit data from the multiple sources
0072   for (const auto& pair : sourceValidOrbitPair_) {
0073     fillSDSRawDataCollection(*rawData, (char*)events_[pair.second]->payload(), events_[pair.second]->eventSize());
0074   }
0075
0076   std::unique_ptr<edm::WrapperBase> edp(new edm::Wrapper<SDSRawDataCollection>(std::move(rawData)));
0077   eventPrincipal.put(
0078       daqProvenanceHelpers_[0]->branchDescription(), std::move(edp), daqProvenanceHelpers_[0]->dummyProvenance());
0079
0080   eventCached_ = false;
0081 }
0082
0083 void DataModeScoutingRun3::fillSDSRawDataCollection(SDSRawDataCollection& rawData, char* buff, size_t len) {
0084   size_t pos = 0;
0085
0086   // get the source ID
0087   int sourceId = *((uint32_t*)(buff + pos));
0088   pos += 4;
0089
0090   // size of the orbit paylod
0091   size_t orbitSize = len - pos;
0092
0093   // set the size (=orbit size) in the SRDColletion of the current source.
0094   // FRD size is expecting 8 bytes words, while scouting is using 4 bytes
0095   // words. This could be different for some future sources.
0096   FEDRawData& fedData = rawData.FEDData(sourceId);
0097   fedData.resize(orbitSize, 4);
0098
0099   memcpy(fedData.data(), buff + pos, orbitSize);
0100
0101   return;
0102 }
0103
0104 std::vector<std::shared_ptr<const edm::DaqProvenanceHelper>>& DataModeScoutingRun3::makeDaqProvenanceHelpers() {
0105   //set SRD data collection
0106   daqProvenanceHelpers_.clear();
0107   daqProvenanceHelpers_.emplace_back(std::make_shared<const edm::DaqProvenanceHelper>(
0108       edm::TypeID(typeid(SDSRawDataCollection)), "SDSRawDataCollection", "SDSRawDataCollection", "DAQSource"));
0109   return daqProvenanceHelpers_;
0110 }
0111
0112 bool DataModeScoutingRun3::nextEventView() {
0113   blockCompleted_ = false;
0114   if (eventCached_)
0115     return true;
0116
0117   // move the data block address only for the sources processed
0118   // un the previous event by adding the last event size
0119   for (const auto& pair : sourceValidOrbitPair_) {
0120     dataBlockAddrs_[pair.first] += events_[pair.second]->size();
0121   }
0122
0123   return makeEvents();
0124 }
0125
0126 bool DataModeScoutingRun3::makeEvents() {
0127   // clear events and reset current orbit
0128   events_.clear();
0129   sourceValidOrbitPair_.clear();
0130   currOrbit_ = 0xFFFFFFFF;  // max uint
0131   assert(!blockCompleted_);
0132
0133   // create current "events" (= orbits) list from each data source,
0134   // check if one dataBlock terminated earlier than others.
0135   for (int i = 0; i < numFiles_; i++) {
0136     if (dataBlockAddrs_[i] >= dataBlockMaxAddrs_[i]) {
0137       completedBlocks_[i] = true;
0138       continue;
0139     }
0140
0141     // event contains data, add it to the events list
0142     events_.emplace_back(std::make_unique<FRDEventMsgView>(dataBlockAddrs_[i]));
0143     if (dataBlockAddrs_[i] + events_.back()->size() > dataBlockMaxAddrs_[i])
0144       throw cms::Exception("DAQSource::getNextEvent")
0145           << " event id:" << events_.back()->event() << " lumi:" << events_.back()->lumi()
0146           << " run:" << events_.back()->run() << " of size:" << events_.back()->size()
0147           << " bytes does not fit into the buffer or has corrupted header";
0148
0149     // find the minimum orbit for the current event between all files
0150     if ((events_.back()->event() < currOrbit_) && (!completedBlocks_[i])) {
0151       currOrbit_ = events_.back()->event();
0152     }
0153   }
0154
0155   // mark valid orbits from each data source
0156   // e.g. find when orbit is missing from one source
0157   bool allBlocksCompleted = true;
0158   int evt_idx = 0;
0159   for (int i = 0; i < numFiles_; i++) {
0160     if (completedBlocks_[i]) {
0161       continue;
0162     }
0163
0164     if (events_[evt_idx]->event() != currOrbit_) {
0165       // current source (=i-th source) doesn't contain the expected orbit.
0166       // skip it, and move to the next orbit
0167     } else {
0168       // add a pair <current surce index, event index>
0169       // evt_idx can be different from variable i, as some data blocks can be
0170       // completed before others
0171       sourceValidOrbitPair_.emplace_back(std::make_pair(i, evt_idx));
0172       allBlocksCompleted = false;
0173     }
0174
0175     evt_idx++;
0176   }
0177
0178   if (allBlocksCompleted) {
0179     blockCompleted_ = true;
0180   }
0181   return !allBlocksCompleted;
0182 }
0183
0184 bool DataModeScoutingRun3::checksumValid() { return true; }
0185
0186 std::string DataModeScoutingRun3::getChecksumError() const { return std::string(); }