diff --git a/EventFilter/Utilities/doc/README-DTH.md b/EventFilter/Utilities/doc/README-DTH.md
new file mode 100644
index 0000000000000..3a70bc0957531
--- /dev/null
+++ b/EventFilter/Utilities/doc/README-DTH.md
@@ -0,0 +1,55 @@
+
+# DTH orbit/event unpacker for DAQSource
+
+https://github.com/smorovic/cmssw/tree/15_0_0_pre1-source-improvements
+
+This patch implements unpacking of the the DTH data format by `DAQSource` into `FedRawDataCollection`.
+
+It is rebased over CMSSW master (compatible with 15_0_0_pre1 at the time this file is commited), but it builds and runs in 14_2_0 as well. All changes are contained in `EventFilter/Utilities`.
+
+## Fetching the code
+
+```
+scram project CMSSW_15_0_0_pre1 #or CMSSW_14_2_0 (currently it compiles and runs also in 14_X releases)
+git cms-addpkg EventFilter/Utilities
+git remote add smorovic https://github.com/smorovic/cmssw.git
+git fetch smorovic 15_0_0_pre1-source-improvements:15_0_0_pre1-source-improvements
+git checkout 15_0_0_pre1-source-improvements
+scram b
+```
+
+Run the unit test (generates and consumes files with DTH format):
+```
+cmsenv
+cd src/EventFilter/Utilities/test
+./RunBUFU.sh
+```
+
+## Important code and scripts in `EventFilter/Utilities`:
+
+Definition of DTH orbit header, fragment trailer and SLinkRocket header/trailer (could potentially be moved to DataFormats or another package in the future):
+
+[interface/DTHHeaders.h](../interface/DTHHeaders.h)
+
+Plugin for DAQSource (input source) which parses the DTH format:
+
+[src/DAQSourceModelsDTH.cc](../src/DAQSourceModelsDTH.cc)
+
+Generator of dummy DTH payload for the fake "BU" process used in unit tests:
+
+[plugins/DTHFakeReader.cc](../plugins/DTHFakeReader.cc)
+
+Script which runs the unit test with "fakeBU" process generating payload from multiple DTH sources (per orbit) and "FU" CMSSW job consuming it:
+
+[test/testDTH.sh](../test/testDTH.sh)
+
+FU cmsRun configuration used in above tests:
+
+[test/unittest_FU_daqsource.py](../test/unittest_FU_daqsource.py)
+
+## Running on custom input files
+`unittest_FU_daqsource.py` script can be used as a starting point to create a custom runner with inputs such as DTH dumps (not generated as in the unit test). DAQSource should be set to `dataMode = cms.untracked.string("DTH")` to process DTH format. Change `fileListMode` to `True` and fill in `fileList` parameter with file paths to run with custom files, however they should be named similarly and could also be placed in similar directory structure, `ramdisk/runXX`, to provide initial run and lumisection to the source. Run number is also passed to the source via the command line as well as the working directory (see `testDTH.sh` script).
+
+Note on the file format: apart of parsing single DTH orbit dump, input source plugin is capable also of building events from multiple DTH orbit blocks, but for the same orbit they must come sequentially in the file. Source scans the file and will find all blocks with orbit headers from the same orbit number, until a different orbit number is found or EOF, then it proceeds to build events from them by starting from last DTH event fragment trailer in each of the orbits found. This is then iterated for the next set of orbit blocks with the same orbit number in the file until file is processed.
+
+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 RAW data), to provide additional metadata to improve integrity and completeness checks after aggregation of data in DAQ. At present, only RAW DTH is supported by "DTH" format.
diff --git a/EventFilter/Utilities/interface/DAQSource.h b/EventFilter/Utilities/interface/DAQSource.h
index 6c21262ce57d6..10f9ae2f7264e 100644
--- a/EventFilter/Utilities/interface/DAQSource.h
+++ b/EventFilter/Utilities/interface/DAQSource.h
@@ -7,6 +7,7 @@
#include
#include
#include
+#include
#include "oneapi/tbb/concurrent_queue.h"
#include "oneapi/tbb/concurrent_vector.h"
@@ -20,7 +21,8 @@
#include "EventFilter/Utilities/interface/EvFDaqDirector.h"
//import InputChunk
-#include "EventFilter/Utilities/interface/FedRawDataInputSource.h"
+#include "DataFormats/FEDRawData/interface/FEDRawDataCollection.h"
+#include "EventFilter/Utilities/interface/SourceRawFile.h"
class FEDRawDataCollection;
class InputSourceDescription;
@@ -72,6 +74,7 @@ class DAQSource : public edm::RawInputSource {
void maybeOpenNewLumiSection(const uint32_t lumiSection);
void readSupervisor();
+ void fileDeleter();
void dataArranger();
void readWorker(unsigned int tid);
void threadError();
@@ -92,10 +95,11 @@ class DAQSource : public edm::RawInputSource {
uint64_t maxChunkSize_; // for buffered read-ahead
uint64_t eventChunkBlock_; // how much read(2) asks at the time
unsigned int readBlocks_;
+ int numConcurrentReads_;
unsigned int numBuffers_;
unsigned int maxBufferedFiles_;
- unsigned int numConcurrentReads_;
std::atomic readingFilesCount_;
+ std::atomic heldFilesCount_;
// get LS from filename instead of event header
const bool alwaysStartFromFirstLS_;
@@ -136,6 +140,7 @@ class DAQSource : public edm::RawInputSource {
bool startedSupervisorThread_ = false;
std::unique_ptr readSupervisorThread_;
+ std::unique_ptr fileDeleterThread_;
std::unique_ptr dataArrangerThread_;
std::vector workerThreads_;
@@ -164,6 +169,7 @@ class DAQSource : public edm::RawInputSource {
//supervisor thread wakeup
std::mutex mWakeup_;
std::condition_variable cvWakeup_;
+ std::condition_variable cvWakeupAll_;
//variables for the single buffered mode
int fileDescriptor_ = -1;
@@ -176,28 +182,4 @@ class DAQSource : public edm::RawInputSource {
std::shared_ptr dataMode_;
};
-class RawInputFile : public InputFile {
-public:
- RawInputFile(evf::EvFDaqDirector::FileStatus status,
- unsigned int lumi = 0,
- std::string const& name = std::string(),
- bool deleteFile = true,
- int rawFd = -1,
- uint64_t fileSize = 0,
- uint16_t rawHeaderSize = 0,
- uint32_t nChunks = 0,
- int nEvents = 0,
- DAQSource* parent = nullptr)
- : InputFile(status, lumi, name, deleteFile, rawFd, fileSize, rawHeaderSize, nChunks, nEvents, nullptr),
- sourceParent_(parent) {}
- bool advance(unsigned char*& dataPosition, const size_t size);
- void advance(const size_t size) {
- chunkPosition_ += size;
- bufferPosition_ += size;
- }
-
-private:
- DAQSource* sourceParent_;
-};
-
#endif // EventFilter_Utilities_DAQSource_h
diff --git a/EventFilter/Utilities/interface/DAQSourceModels.h b/EventFilter/Utilities/interface/DAQSourceModels.h
index 5727dc7aa2164..c17509b3ae1c1 100644
--- a/EventFilter/Utilities/interface/DAQSourceModels.h
+++ b/EventFilter/Utilities/interface/DAQSourceModels.h
@@ -24,8 +24,10 @@
#include "DataFormats/Provenance/interface/LuminosityBlockAuxiliary.h"
//import InputChunk
-#include "EventFilter/Utilities/interface/FedRawDataInputSource.h"
+#include "EventFilter/Utilities/interface/SourceRawFile.h"
+class RawInputFile;
+class UnpackedRawEventWrapper;
class DAQSource;
//evf?
@@ -40,18 +42,16 @@ class DataMode {
virtual uint32_t headerSize() const = 0;
virtual bool versionCheck() const = 0;
virtual uint64_t dataBlockSize() const = 0;
- virtual void makeDataBlockView(unsigned char* addr,
- size_t maxSize,
- std::vector const& fileSizes,
- size_t fileHeaderSize) = 0;
- virtual bool nextEventView() = 0;
+ virtual void makeDataBlockView(unsigned char* addr, RawInputFile* rawFile) = 0;
+ virtual bool nextEventView(RawInputFile*) = 0;
+ virtual bool blockChecksumValid() = 0;
virtual bool checksumValid() = 0;
virtual std::string getChecksumError() const = 0;
- virtual bool isRealData() const = 0;
virtual uint32_t run() const = 0;
virtual bool dataBlockCompleted() const = 0;
virtual bool requireHeader() const = 0;
virtual bool fitToBuffer() const = 0;
+ virtual void unpackFile(RawInputFile* file) = 0;
virtual bool dataBlockInitialized() const = 0;
virtual void setDataBlockInitialized(bool) = 0;
@@ -66,9 +66,12 @@ class DataMode {
std::string const& runDir) = 0;
void setTesting(bool testing) { testing_ = testing; }
+ bool errorDetected() { return errorDetected_; }
+
protected:
DAQSource* daqSource_;
bool testing_ = false;
+ bool errorDetected_ = false;
};
#endif // EventFilter_Utilities_DAQSourceModels_h
diff --git a/EventFilter/Utilities/interface/DAQSourceModelsDTH.h b/EventFilter/Utilities/interface/DAQSourceModelsDTH.h
new file mode 100644
index 0000000000000..0e6ced0150146
--- /dev/null
+++ b/EventFilter/Utilities/interface/DAQSourceModelsDTH.h
@@ -0,0 +1,89 @@
+#ifndef EventFilter_Utilities_DAQSourceModelsDTH_h
+#define EventFilter_Utilities_DAQSourceModelsDTH_h
+
+#include
+#include
+
+#include "EventFilter/Utilities/interface/DAQSourceModels.h"
+#include "DataFormats/FEDRawData/interface/FEDRawDataCollection.h"
+#include "EventFilter/Utilities/interface/DTHHeaders.h"
+
+class FEDRawDataCollection;
+
+class DataModeDTH : public DataMode {
+public:
+ DataModeDTH(DAQSource* daqSource, bool verifyChecksum) : DataMode(daqSource), verifyChecksum_(verifyChecksum) {}
+ ~DataModeDTH() override {}
+ std::vector>& makeDaqProvenanceHelpers() override;
+ void readEvent(edm::EventPrincipal& eventPrincipal) override;
+
+ //non-virtual
+ edm::Timestamp fillFEDRawDataCollection(FEDRawDataCollection& rawData);
+
+ int dataVersion() const override { return detectedDTHversion_; }
+ void detectVersion(unsigned char* fileBuf, uint32_t fileHeaderOffset) override {
+ detectedDTHversion_ = 1; //TODO: read version
+ }
+
+ uint32_t headerSize() const override { return sizeof(evf::DTHOrbitHeader_v1); }
+
+ bool versionCheck() const override { return detectedDTHversion_ == 1; }
+
+ uint64_t dataBlockSize() const override { return dataBlockSize_; }
+
+ void makeDataBlockView(unsigned char* addr, RawInputFile* rawFile) override;
+
+ bool nextEventView(RawInputFile*) override;
+ bool blockChecksumValid() override { return checksumValid_; }
+ bool checksumValid() override { return checksumValid_; }
+ std::string getChecksumError() const override { return checksumError_; }
+
+ bool isRealData() const { return true; } //this flag could be added to RU/BU-generated index
+
+ uint32_t run() const override { return firstOrbitHeader_->runNumber(); }
+
+ bool dataBlockCompleted() const override { return blockCompleted_; }
+
+ bool requireHeader() const override { return false; }
+
+ bool fitToBuffer() const override { return true; }
+
+ void unpackFile(RawInputFile*) override {}
+
+ bool dataBlockInitialized() const override { return dataBlockInitialized_; }
+
+ void setDataBlockInitialized(bool val) override { dataBlockInitialized_ = val; }
+
+ void setTCDSSearchRange(uint16_t MINTCDSuTCAFEDID, uint16_t MAXTCDSuTCAFEDID) override {}
+
+ void makeDirectoryEntries(std::vector const& baseDirs,
+ std::vector const& numSources,
+ std::string const& runDir) override {}
+
+ std::pair> defineAdditionalFiles(std::string const& primaryName, bool) const override {
+ return std::make_pair(true, std::vector());
+ }
+
+private:
+ bool verifyChecksum_;
+ std::vector> daqProvenanceHelpers_;
+ uint16_t detectedDTHversion_ = 0;
+ evf::DTHOrbitHeader_v1* firstOrbitHeader_ = nullptr;
+ uint64_t nextEventID_ = 0;
+ std::vector eventFragments_; //events in block (DTH trailer)
+ bool dataBlockInitialized_ = false;
+ bool blockCompleted_ = true;
+
+ std::vector addrsStart_; //start of orbit payloads per source
+ std::vector addrsEnd_; //dth trailers per source (go through events from the end)
+
+ bool checksumValid_ = false;
+ std::string checksumError_;
+ //total
+ size_t dataBlockSize_ = 0;
+ //uint16_t MINTCDSuTCAFEDID_ = FEDNumbering::MINTCDSuTCAFEDID;
+ //uint16_t MAXTCDSuTCAFEDID_ = FEDNumbering::MAXTCDSuTCAFEDID;
+ bool eventCached_ = false;
+};
+
+#endif // EventFilter_Utilities_DAQSourceModelsDTH_h
diff --git a/EventFilter/Utilities/interface/DAQSourceModelsFRD.h b/EventFilter/Utilities/interface/DAQSourceModelsFRD.h
index c49fd8280b5b5..9d7fdbedd4daa 100644
--- a/EventFilter/Utilities/interface/DAQSourceModelsFRD.h
+++ b/EventFilter/Utilities/interface/DAQSourceModelsFRD.h
@@ -2,8 +2,10 @@
#define EventFilter_Utilities_DAQSourceModelsFRD_h
#include
+#include
#include "EventFilter/Utilities/interface/DAQSourceModels.h"
+#include "DataFormats/FEDRawData/interface/FEDRawDataCollection.h"
class FEDRawDataCollection;
@@ -30,22 +32,13 @@ class DataModeFRD : public DataMode {
uint64_t dataBlockSize() const override { return event_->size(); }
- void makeDataBlockView(unsigned char* addr,
- size_t maxSize,
- std::vector const& fileSizes,
- size_t fileHeaderSize) override {
- dataBlockAddr_ = addr;
- dataBlockMax_ = maxSize;
- eventCached_ = false;
- nextEventView();
- eventCached_ = true;
- }
-
- bool nextEventView() override;
+ void makeDataBlockView(unsigned char* addr, RawInputFile* rawFile) override;
+ bool nextEventView(RawInputFile*) override;
+ bool blockChecksumValid() override { return true; }
bool checksumValid() override;
std::string getChecksumError() const override;
- bool isRealData() const override { return event_->isRealData(); }
+ //bool isRealData() const override { return event_->isRealData(); }
uint32_t run() const override { return event_->run(); }
@@ -55,6 +48,9 @@ class DataModeFRD : public DataMode {
bool requireHeader() const override { return true; }
bool fitToBuffer() const override { return false; }
+
+ void unpackFile(RawInputFile*) override {}
+
bool dataBlockInitialized() const override { return true; }
void setDataBlockInitialized(bool) override {}
@@ -86,8 +82,88 @@ class DataModeFRD : public DataMode {
bool eventCached_ = false;
};
+/*
+ * FRD source prebuffering in reader thread
+ */
+
+class DataModeFRDPreUnpack : public DataMode {
+public:
+ DataModeFRDPreUnpack(DAQSource* daqSource) : DataMode(daqSource) {}
+ ~DataModeFRDPreUnpack() override {};
+ std::vector>& makeDaqProvenanceHelpers() override;
+ void readEvent(edm::EventPrincipal& eventPrincipal) override;
+
+ //non-virtual
+ void unpackEvent(edm::streamer::FRDEventMsgView* eview, UnpackedRawEventWrapper* ec);
+ void unpackFile(RawInputFile*) override;
+ edm::Timestamp fillFEDRawDataCollection(edm::streamer::FRDEventMsgView* eview,
+ FEDRawDataCollection& rawData,
+ bool& tcdsInRange,
+ unsigned char*& tcds_pointer,
+ bool& err,
+ std::string& errmsg);
+
+ int dataVersion() const override { return detectedFRDversion_; }
+ void detectVersion(unsigned char* fileBuf, uint32_t fileHeaderOffset) override {
+ detectedFRDversion_ = *((uint16_t*)(fileBuf + fileHeaderOffset));
+ }
+
+ uint32_t headerSize() const override { return edm::streamer::FRDHeaderVersionSize[detectedFRDversion_]; }
+
+ bool versionCheck() const override { return detectedFRDversion_ <= edm::streamer::FRDHeaderMaxVersion; }
+
+ //used
+ uint64_t dataBlockSize() const override { return event_->size(); }
+
+ void makeDataBlockView(unsigned char* addr, RawInputFile* rawFile) override;
+ bool nextEventView(RawInputFile*) override;
+ bool blockChecksumValid() override { return true; }
+ bool checksumValid() override;
+ std::string getChecksumError() const override;
+
+ uint32_t run() const override { return ec_->run(); }
+
+ //true for DAQ3 FRD
+ bool dataBlockCompleted() const override { return true; }
+
+ bool requireHeader() const override { return true; }
+
+ bool fitToBuffer() const override { return true; }
+
+ bool dataBlockInitialized() const override { return true; }
+
+ void setDataBlockInitialized(bool) override {};
+
+ void setTCDSSearchRange(uint16_t MINTCDSuTCAFEDID, uint16_t MAXTCDSuTCAFEDID) override {
+ MINTCDSuTCAFEDID_ = MINTCDSuTCAFEDID;
+ MAXTCDSuTCAFEDID_ = MAXTCDSuTCAFEDID;
+ }
+
+ void makeDirectoryEntries(std::vector const& baseDirs,
+ std::vector const& numSources,
+ std::string const& runDir) override {}
+
+ std::pair> defineAdditionalFiles(std::string const& primaryName, bool) const override {
+ return std::make_pair(true, std::vector());
+ }
+
+private:
+ std::vector> daqProvenanceHelpers_;
+ uint16_t detectedFRDversion_ = 0;
+ size_t headerSize_ = 0;
+ std::unique_ptr event_;
+ std::unique_ptr ec_;
+ uint32_t crc_ = 0;
+ unsigned char* dataBlockAddr_ = nullptr;
+ size_t dataBlockMax_ = 0;
+ size_t fileHeaderSize_ = 0;
+ uint16_t MINTCDSuTCAFEDID_ = FEDNumbering::MINTCDSuTCAFEDID;
+ uint16_t MAXTCDSuTCAFEDID_ = FEDNumbering::MAXTCDSuTCAFEDID;
+ bool eventCached_ = false;
+};
+
/*
- * Demo for FRD source reading files from multiple striped destinations
+ * FRD source reading files from multiple striped destinations
*
* */
@@ -117,41 +193,16 @@ class DataModeFRDStriped : public DataMode {
return events_[0]->size();
}
- void makeDataBlockView(unsigned char* addr,
- size_t maxSize,
- std::vector const& fileSizes,
- size_t fileHeaderSize) override {
- fileHeaderSize_ = fileHeaderSize;
- numFiles_ = fileSizes.size();
- //add offset address for each file payload
- dataBlockAddrs_.clear();
- dataBlockAddrs_.push_back(addr);
- dataBlockMaxAddrs_.clear();
- dataBlockMaxAddrs_.push_back(addr + fileSizes[0] - fileHeaderSize);
- auto fileAddr = addr;
- for (unsigned int i = 1; i < fileSizes.size(); i++) {
- fileAddr += fileSizes[i - 1];
- dataBlockAddrs_.push_back(fileAddr);
- dataBlockMaxAddrs_.push_back(fileAddr + fileSizes[i] - fileHeaderSize);
- }
-
- dataBlockMax_ = maxSize;
- blockCompleted_ = false;
- //set event cached as we set initial address here
- bool result = makeEvents();
- assert(result);
- eventCached_ = true;
- setDataBlockInitialized(true);
- }
-
- bool nextEventView() override;
+ void makeDataBlockView(unsigned char* addr, RawInputFile* rawFile) override;
+ bool nextEventView(RawInputFile*) override;
+ bool blockChecksumValid() override { return true; }
bool checksumValid() override;
std::string getChecksumError() const override;
- bool isRealData() const override {
- assert(!events_.empty());
- return events_[0]->isRealData();
- }
+ //bool isRealData() const override {
+ // assert(!events_.empty());
+ // return events_[0]->isRealData();
+ //}
uint32_t run() const override {
assert(!events_.empty());
@@ -164,6 +215,8 @@ class DataModeFRDStriped : public DataMode {
bool fitToBuffer() const override { return true; }
+ void unpackFile(RawInputFile*) override {}
+
bool dataBlockInitialized() const override { return dataBlockInitialized_; }
void setDataBlockInitialized(bool val) override { dataBlockInitialized_ = val; }
diff --git a/EventFilter/Utilities/interface/DAQSourceModelsScoutingRun3.h b/EventFilter/Utilities/interface/DAQSourceModelsScoutingRun3.h
index 82da922ec6a86..5d34ecb7da888 100644
--- a/EventFilter/Utilities/interface/DAQSourceModelsScoutingRun3.h
+++ b/EventFilter/Utilities/interface/DAQSourceModelsScoutingRun3.h
@@ -39,11 +39,9 @@ class DataModeScoutingRun3 : public DataMode {
return events_[0]->size();
}
- void makeDataBlockView(unsigned char* addr,
- size_t maxSize,
- std::vector const& fileSizes,
- size_t fileHeaderSize) override {
- fileHeaderSize_ = fileHeaderSize;
+ void makeDataBlockView(unsigned char* addr, RawInputFile* rawFile) override {
+ std::vector const& fileSizes = rawFile->fileSizes_;
+ fileHeaderSize_ = rawFile->rawHeaderSize_;
numFiles_ = fileSizes.size();
// initalize vectors keeping tracks of valid orbits and completed blocks
@@ -57,15 +55,15 @@ class DataModeScoutingRun3 : public DataMode {
dataBlockAddrs_.clear();
dataBlockAddrs_.push_back(addr);
dataBlockMaxAddrs_.clear();
- dataBlockMaxAddrs_.push_back(addr + fileSizes[0] - fileHeaderSize);
+ dataBlockMaxAddrs_.push_back(addr + fileSizes[0] - fileHeaderSize_);
auto fileAddr = addr;
for (unsigned int i = 1; i < fileSizes.size(); i++) {
fileAddr += fileSizes[i - 1];
dataBlockAddrs_.push_back(fileAddr);
- dataBlockMaxAddrs_.push_back(fileAddr + fileSizes[i] - fileHeaderSize);
+ dataBlockMaxAddrs_.push_back(fileAddr + fileSizes[i] - fileHeaderSize_);
}
- dataBlockMax_ = maxSize;
+ dataBlockMax_ = rawFile->currentChunkSize();
blockCompleted_ = false;
//set event cached as we set initial address here
bool result = makeEvents();
@@ -74,15 +72,11 @@ class DataModeScoutingRun3 : public DataMode {
setDataBlockInitialized(true);
}
- bool nextEventView() override;
+ bool nextEventView(RawInputFile*) override;
+ bool blockChecksumValid() override { return true; }
bool checksumValid() override;
std::string getChecksumError() const override;
- bool isRealData() const override {
- assert(!events_.empty());
- return events_[0]->isRealData();
- }
-
uint32_t run() const override {
assert(!events_.empty());
return events_[0]->run();
@@ -93,6 +87,7 @@ class DataModeScoutingRun3 : public DataMode {
bool requireHeader() const override { return true; }
bool fitToBuffer() const override { return true; }
+ void unpackFile(RawInputFile* file) override {}
bool dataBlockInitialized() const override { return dataBlockInitialized_; }
diff --git a/EventFilter/Utilities/interface/DTHHeaders.h b/EventFilter/Utilities/interface/DTHHeaders.h
new file mode 100644
index 0000000000000..20bda17b8a217
--- /dev/null
+++ b/EventFilter/Utilities/interface/DTHHeaders.h
@@ -0,0 +1,274 @@
+#ifndef EventFilter_Utilities_DTHHeaders_h
+#define EventFilter_Utilities_DTHHeaders_h
+
+#include
+#include
+#include
+
+//#include "IOPool/Streamer/interface/MsgTools.h"
+/*
+ * DTH Orbit header and event fragment trailer accompanying slink payload.
+ * In this version, big-endian number format is assumed to be written
+ * by DTH and requires byte swapping on low-endian platforms when converting
+ * to numerical representation
+ *
+ * Version 1 Format defined
+ * */
+
+namespace evf {
+ constexpr std::array DTHOrbitMarker{{0x4f, 0x48}};
+ constexpr std::array DTHFragmentTrailerMarker{{0x46, 0x54}};
+ constexpr uint32_t DTH_WORD_NUM_BYTES = 16;
+ constexpr uint32_t DTH_WORD_NUM_BYTES_SHIFT = 4;
+ constexpr uint32_t SLR_WORD_NUM_BYTES = 16;
+ constexpr uint32_t SLR_WORD_NUM_BYTES_SHIFT = 4;
+ constexpr uint32_t SLR_MAX_EVENT_LEN = (1 << 20);
+
+ constexpr uint64_t convert(std::array v) {
+ //LSB first
+ uint64_t a = v[0], b = v[1], c = v[2], d = v[3], e = v[4], f = v[5];
+ return a | (b << 8) | (c << 16) | (d << 24) | (e << 32) | (f << 40);
+ }
+
+ constexpr uint32_t convert(std::array v) {
+ //LSB first
+ uint32_t a = v[0], b = v[1], c = v[2], d = v[3];
+ return a | (b << 8) | (c << 16) | (d << 24);
+ }
+
+ constexpr uint16_t convert(std::array v) {
+ //LSB first
+ uint16_t a = v[0], b = v[1];
+ return a | (b << 8);
+ }
+
+ constexpr std::array convert48(uint64_t i) {
+ return std::array{{uint8_t(i & 0xff),
+ uint8_t((i >> 8) & 0xff),
+ uint8_t((i >> 16) & 0xff),
+ uint8_t((i >> 24) & 0xff),
+ uint8_t((i >> 32) & 0xff),
+ uint8_t((i >> 40) & 0xff)}};
+ }
+
+ constexpr std::array convert(uint32_t i) {
+ return std::array{
+ {uint8_t(i & 0xff), uint8_t((i >> 8) & 0xff), uint8_t((i >> 16) & 0xff), uint8_t((i >> 24) & 0xff)}};
+ }
+
+ constexpr std::array convert(uint16_t i) {
+ return std::array{{uint8_t(i & 0xff), uint8_t((i >> 8) & 0xff)}};
+ }
+
+ class DTHOrbitHeader_v1 {
+ public:
+ DTHOrbitHeader_v1(uint32_t source_id,
+ uint32_t orbit_number,
+ uint32_t run_number,
+ uint32_t packed_word_count,
+ uint16_t event_count,
+ uint32_t crc,
+ uint32_t flags)
+ : //convert numbers into binary representation
+ source_id_(convert(source_id)),
+ orbit_number_(convert(orbit_number)),
+ run_number_(convert(run_number)),
+ packed_word_count_(convert(packed_word_count)),
+ event_count_(convert(event_count)),
+ crc32c_(convert(crc)),
+ flags_(convert(flags)) {}
+
+ uint32_t sourceID() const { return convert(source_id_); }
+ //this should be 1 but can be used for autodetection or consistency check
+ uint16_t version() const { return convert(version_); }
+ uint32_t orbitNumber() const { return convert(orbit_number_); }
+ uint32_t runNumber() const { return convert(run_number_); }
+ uint32_t packed_word_count() const { return convert(packed_word_count_); }
+ uint64_t totalSize() const { return (DTH_WORD_NUM_BYTES * uint64_t(packed_word_count())); }
+ uint64_t payloadSizeBytes() const { return totalSize() - sizeof(DTHOrbitHeader_v1); }
+ uint64_t headerSize() const { return sizeof(DTHOrbitHeader_v1); }
+ uint16_t eventCount() const { return convert(event_count_); }
+ uint32_t crc() const { return convert(crc32c_); }
+ uint32_t flags() const { return convert(flags_); }
+ const void* payload() const { return (uint8_t*)this + sizeof(DTHOrbitHeader_v1); }
+ bool verifyMarker() const {
+ for (size_t i = 0; i < DTHOrbitMarker.size(); i++) {
+ if (marker_[i] != DTHOrbitMarker[i])
+ return false;
+ }
+ return true;
+ }
+
+ bool verifyChecksum() const;
+
+ private:
+ std::array source_id_;
+ std::array version_ = {{0, 1}};
+ std::array marker_ = DTHOrbitMarker;
+ std::array orbit_number_;
+ std::array run_number_;
+ std::array packed_word_count_; //128-bit-words
+ std::array reserved_ = {{0, 0}};
+ std::array event_count_;
+ std::array crc32c_;
+ std::array flags_;
+ };
+
+ //TODO: change init to use packed word count
+ class DTHFragmentTrailer_v1 {
+ public:
+ DTHFragmentTrailer_v1(uint32_t payload_word_count, uint16_t flags, uint16_t crc, uint64_t event_id)
+ : payload_word_count_(convert(payload_word_count)),
+ flags_(convert(flags)),
+ crc_(convert(crc)),
+ res_and_eid_({{uint8_t((event_id & 0x0f0000000000) >> 40),
+ uint8_t((event_id & 0xff00000000) >> 32),
+ uint8_t((event_id & 0xff000000) >> 24),
+ uint8_t((event_id & 0xff0000) >> 16),
+ uint8_t((event_id & 0xff00) >> 8),
+ uint8_t(event_id & 0xff)}}) {}
+
+ uint64_t eventID() const {
+ return (uint64_t(res_and_eid_[0] & 0xf) << 40) + (uint64_t(res_and_eid_[1]) << 32) +
+ (uint32_t(res_and_eid_[2]) << 24) + (uint32_t(res_and_eid_[3]) << 16) + (uint16_t(res_and_eid_[4]) << 8) +
+ res_and_eid_[5];
+ }
+ uint32_t payloadWordCount() const { return convert(payload_word_count_); }
+ uint64_t payloadSizeBytes() const { return uint64_t(convert(payload_word_count_)) * DTH_WORD_NUM_BYTES; }
+ uint16_t flags() const { return convert(flags_); }
+ uint16_t crc() const { return convert(crc_); }
+ const void* payload() const { return (uint8_t*)this - payloadSizeBytes(); }
+ bool verifyMarker() const {
+ for (size_t i = 0; i < DTHFragmentTrailerMarker.size(); i++) {
+ if (marker_[i] != DTHFragmentTrailerMarker[i])
+ return false;
+ }
+ return true;
+ }
+
+ private:
+ std::array payload_word_count_;
+ std::array flags_;
+ std::array marker_ = DTHFragmentTrailerMarker;
+ std::array crc_;
+ std::array res_and_eid_;
+ };
+
+ class DTHFragmentTrailerView {
+ public:
+ DTHFragmentTrailerView(void* buf)
+
+ : trailer_((DTHFragmentTrailer_v1*)buf),
+ payload_size_(trailer_->payloadSizeBytes()),
+ flags_(trailer_->flags()),
+ crc_(trailer_->crc()),
+ eventID_(trailer_->eventID()) {}
+
+ uint8_t* startAddress() const { return (uint8_t*)trailer_; }
+ const void* payload() const { return trailer_->payload(); }
+ uint64_t payloadSizeBytes() const { return payload_size_; }
+ uint16_t flags() const { return flags_; }
+ uint16_t crc() const { return crc_; }
+ uint64_t eventID() const { return eventID_; }
+ bool verifyMarker() const { return trailer_ ? trailer_->verifyMarker() : false; }
+
+ private:
+ DTHFragmentTrailer_v1* trailer_;
+ uint64_t payload_size_;
+ uint16_t flags_;
+ uint16_t crc_;
+ uint64_t eventID_;
+ };
+
+ //SLinkExpress classes
+
+ //begin and end event
+ constexpr uint8_t SLR_BOE = 0x55;
+ constexpr uint8_t SLR_EOE = 0xaa;
+
+ //minimal SLinkRocket format version version overlay
+ class SLinkRocketHeader_version {
+ public:
+ SLinkRocketHeader_version(uint8_t version, uint8_t trail = 0) : v_and_r_(version << 4 | (trail & 0xf)) {}
+ uint8_t version() const { return v_and_r_ >> 4; }
+ bool verifyMarker() const { return boe_ == SLR_BOE; }
+
+ private:
+ uint8_t boe_ = SLR_BOE;
+ uint8_t v_and_r_;
+ };
+
+ class SLinkRocketHeader_v3 {
+ public:
+ SLinkRocketHeader_v3(uint64_t glob_event_id, uint32_t content_id, uint32_t source_id)
+ : r_and_eid_(convert48(glob_event_id & 0x0fffffffffff)), //44 used, 4 reserved
+ r_and_e_(uint8_t((content_id >> 24) & 0x03)), //2 used, 6 reserved
+ l1a_subtype_(uint8_t((content_id >> 16) & 0xff)),
+ l1a_t_fc_(convert(uint16_t(content_id & 0xffff))),
+ source_id_(convert(source_id)) {}
+
+ SLinkRocketHeader_v3(uint64_t glob_event_id,
+ uint8_t emu_status,
+ uint8_t l1a_subtype,
+ uint16_t l1a_types_fragcont,
+ uint32_t source_id)
+ : r_and_eid_(convert48(glob_event_id & 0x0fffffffffff)),
+ r_and_e_(emu_status & 0x03),
+ l1a_subtype_(l1a_subtype),
+ l1a_t_fc_(convert(l1a_types_fragcont)),
+ source_id_(convert(source_id)) {}
+
+ uint8_t version() const { return version_and_r_ >> 4; }
+ uint64_t globalEventID() const { return convert(r_and_eid_) & 0x0fffffffffff; }
+ uint32_t contentID() const {
+ return (uint32_t(convert(l1a_t_fc_)) << 16) | (uint32_t(l1a_subtype_) << 8) | (r_and_e_ & 0x3);
+ }
+ uint8_t emuStatus() const { return r_and_e_ & 0x03; }
+ uint8_t l1aSubtype() const { return l1a_subtype_; }
+ uint16_t l1aTypeAndFragmentContent() const { return convert(l1a_t_fc_); }
+ uint32_t sourceID() const { return convert(source_id_); }
+ bool verifyMarker() const { return boe_ == SLR_BOE; }
+
+ private:
+ uint8_t boe_ = SLR_BOE;
+ uint8_t version_and_r_ = 3 << 4;
+ std::array r_and_eid_;
+ uint8_t r_and_e_;
+ uint8_t l1a_subtype_;
+ std::array l1a_t_fc_;
+ std::array source_id_;
+ };
+
+ class SLinkRocketTrailer_v3 {
+ public:
+ SLinkRocketTrailer_v3(
+ uint16_t daq_crc, uint32_t evtlen_word_count, uint16_t bxid, uint32_t orbit_id, uint16_t crc, uint16_t status)
+ : daq_crc_(convert(daq_crc)),
+ evtlen_w_count_and_bxid_(convert((evtlen_word_count << 12) | uint32_t(bxid & 0x0fff))),
+ orbit_id_(convert(orbit_id)),
+ crc_(convert(crc)),
+ status_(convert(status)) {}
+
+ uint16_t daqCRC() const { return convert(daq_crc_); }
+ uint32_t eventLenBytes() const {
+ return ((convert(evtlen_w_count_and_bxid_) >> 12) & 0x0fffff) * SLR_WORD_NUM_BYTES;
+ }
+ uint16_t bxID() const { return convert(evtlen_w_count_and_bxid_) & 0x0fff; }
+ uint32_t orbitID() const { return convert(orbit_id_); }
+ uint16_t crc() const { return convert(crc_); }
+ uint16_t status() const { return convert(status_); }
+ bool verifyMarker() const { return eoe_ == SLR_EOE; }
+
+ private:
+ uint8_t eoe_ = SLR_EOE;
+ std::array daq_crc_;
+ uint8_t reserved_ = 0;
+ std::array evtlen_w_count_and_bxid_; //event 128-bit word length includes header and trailer
+ std::array orbit_id_;
+ std::array crc_;
+ std::array status_;
+ };
+
+} // namespace evf
+
+#endif
diff --git a/EventFilter/Utilities/interface/EvFDaqDirector.h b/EventFilter/Utilities/interface/EvFDaqDirector.h
index 6b9a7ebc6616c..4631fdacda585 100644
--- a/EventFilter/Utilities/interface/EvFDaqDirector.h
+++ b/EventFilter/Utilities/interface/EvFDaqDirector.h
@@ -27,14 +27,12 @@
#include
#include
+#include
class SystemBounds;
class GlobalContext;
class StreamID;
-class InputFile;
-struct InputChunk;
-
namespace edm {
class PathsAndConsumesOfModulesBase;
class ProcessContext;
@@ -70,7 +68,11 @@ namespace evf {
void preBeginRun(edm::GlobalContext const& globalContext);
void postEndRun(edm::GlobalContext const& globalContext);
void preGlobalEndLumi(edm::GlobalContext const& globalContext);
- void overrideRunNumber(unsigned int run) { run_ = run; }
+ void updateRunParams();
+ void overrideRunNumber(unsigned int run) {
+ run_ = run;
+ updateRunParams();
+ }
std::string const& runString() const { return run_string_; }
std::string& baseRunDir() { return run_dir_; }
std::string& buBaseRunDir() { return bu_run_dir_; }
@@ -178,11 +180,6 @@ namespace evf {
int readLastLSEntry(std::string const& file);
unsigned int getLumisectionToStart() const;
unsigned int getStartLumisectionFromEnv() const { return startFromLS_; }
- void setDeleteTracking(std::mutex* fileDeleteLock,
- std::list>>* filesToDelete) {
- fileDeleteLockPtr_ = fileDeleteLock;
- filesToDeletePtr_ = filesToDelete;
- }
std::string getStreamDestinations(std::string const&) const { return std::string(""); }
std::string getStreamMergeType(std::string const&, MergeType defaultType) const {
@@ -193,6 +190,9 @@ namespace evf {
bool lumisectionDiscarded(unsigned int ls);
std::vector const& getBUBaseDirs() const { return bu_base_dirs_all_; }
std::vector const& getBUBaseDirsNSources() const { return bu_base_dirs_nSources_; }
+ void setFileListMode() { fileListMode_ = true; }
+ bool fileListMode() const { return fileListMode_; }
+ unsigned int lsWithFilesOpen(unsigned int ls) const;
private:
bool bumpFile(unsigned int& ls,
@@ -229,6 +229,7 @@ namespace evf {
std::string hltSourceDirectory_;
unsigned int startFromLS_ = 1;
+ oneapi::tbb::concurrent_hash_map lsWithFilesMap_;
std::string hostname_;
std::string run_string_;
@@ -264,9 +265,6 @@ namespace evf {
evf::FastMonitoringService* fms_ = nullptr;
- std::mutex* fileDeleteLockPtr_ = nullptr;
- std::list>>* filesToDeletePtr_ = nullptr;
-
pthread_mutex_t init_lock_ = PTHREAD_MUTEX_INITIALIZER;
unsigned int nStreams_ = 0;
@@ -293,6 +291,7 @@ namespace evf {
std::string input_throttled_file_;
std::string discard_ls_filestem_;
+ bool fileListMode_ = false;
};
} // namespace evf
diff --git a/EventFilter/Utilities/interface/FastMonitoringService.h b/EventFilter/Utilities/interface/FastMonitoringService.h
index b1526a8b01c42..e8ae97b80b87f 100644
--- a/EventFilter/Utilities/interface/FastMonitoringService.h
+++ b/EventFilter/Utilities/interface/FastMonitoringService.h
@@ -153,6 +153,10 @@ namespace evf {
inWaitChunk_newFileWaitChunk,
inSupThrottled,
inThrottled,
+ //additions (appended to keep the color scheme)
+ inSupFileHeldLimit,
+ inWaitInput_fileHeldLimit,
+ inWaitChunk_fileHeldLimit,
inCOUNT
};
} // namespace FastMonState
diff --git a/EventFilter/Utilities/interface/FedRawDataInputSource.h b/EventFilter/Utilities/interface/FedRawDataInputSource.h
index 687ce0e50148c..295bc4647a87f 100644
--- a/EventFilter/Utilities/interface/FedRawDataInputSource.h
+++ b/EventFilter/Utilities/interface/FedRawDataInputSource.h
@@ -65,6 +65,7 @@ class FedRawDataInputSource : public edm::RawInputSource {
edm::Timestamp fillFEDRawDataCollection(FEDRawDataCollection& rawData, bool& tcdsInRange);
void readSupervisor();
+ void fileDeleter();
void readWorker(unsigned int tid);
void threadError();
bool exceptionState() { return setExceptionState_; }
@@ -90,10 +91,11 @@ class FedRawDataInputSource : public edm::RawInputSource {
unsigned int eventChunkSize_; // for buffered read-ahead
unsigned int eventChunkBlock_; // how much read(2) asks at the time
unsigned int readBlocks_;
+ int numConcurrentReads_;
unsigned int numBuffers_;
unsigned int maxBufferedFiles_;
- unsigned int numConcurrentReads_;
std::atomic readingFilesCount_;
+ std::atomic heldFilesCount_;
// get LS from filename instead of event header
const bool getLSFromFilename_;
@@ -145,6 +147,7 @@ class FedRawDataInputSource : public edm::RawInputSource {
bool startedSupervisorThread_ = false;
std::unique_ptr readSupervisorThread_;
+ std::unique_ptr fileDeleterThread_;
std::vector workerThreads_;
tbb::concurrent_queue workerPool_;
@@ -165,7 +168,6 @@ class FedRawDataInputSource : public edm::RawInputSource {
int currentFileIndex_ = -1;
std::list>> filesToDelete_;
- std::list> fileNamesToDelete_;
std::mutex fileDeleteLock_;
std::vector streamFileTracker_;
unsigned int checkEvery_ = 10;
@@ -173,9 +175,8 @@ class FedRawDataInputSource : public edm::RawInputSource {
//supervisor thread wakeup
std::mutex mWakeup_;
std::condition_variable cvWakeup_;
+ std::condition_variable cvWakeupAll_;
- //variables for the single buffered mode
- bool singleBufferMode_;
int fileDescriptor_ = -1;
uint32_t bufferInputRead_ = 0;
@@ -185,135 +186,6 @@ class FedRawDataInputSource : public edm::RawInputSource {
std::mutex monlock_;
};
-struct InputChunk {
- unsigned char* buf_;
- InputChunk* next_ = nullptr;
- uint64_t size_;
- uint64_t usedSize_ = 0;
- //unsigned int index_;
- uint64_t offset_;
- unsigned int fileIndex_;
- std::atomic readComplete_;
-
- InputChunk(uint64_t size) : size_(size) {
- buf_ = new unsigned char[size_];
- reset(0, 0, 0);
- }
- void reset(uint64_t newOffset, uint64_t toRead, unsigned int fileIndex) {
- offset_ = newOffset;
- usedSize_ = toRead;
- fileIndex_ = fileIndex;
- readComplete_ = false;
- }
-
- bool resize(uint64_t wantedSize, uint64_t maxSize) {
- if (wantedSize > maxSize)
- return false;
- if (size_ < wantedSize) {
- size_ = uint64_t(wantedSize * 1.05);
- delete[] buf_;
- buf_ = new unsigned char[size_];
- }
- return true;
- }
-
- ~InputChunk() { delete[] buf_; }
-};
-
-class InputFile {
-public:
- FedRawDataInputSource* parent_;
- evf::EvFDaqDirector::FileStatus status_;
- unsigned int lumi_;
- std::string fileName_;
- //used by DAQSource
- std::vector fileNames_;
- std::vector diskFileSizes_;
- std::vector bufferOffsets_;
- std::vector fileSizes_;
- std::vector fileOrder_;
- bool deleteFile_;
- int rawFd_;
- uint64_t fileSize_;
- uint16_t rawHeaderSize_;
- uint16_t nChunks_;
- uint16_t numFiles_;
- int nEvents_;
- unsigned int nProcessed_;
-
- tbb::concurrent_vector chunks_;
-
- uint32_t bufferPosition_ = 0;
- uint32_t chunkPosition_ = 0;
- unsigned int currentChunk_ = 0;
-
- InputFile(evf::EvFDaqDirector::FileStatus status,
- unsigned int lumi = 0,
- std::string const& name = std::string(),
- bool deleteFile = true,
- int rawFd = -1,
- uint64_t fileSize = 0,
- uint16_t rawHeaderSize = 0,
- uint16_t nChunks = 0,
- int nEvents = 0,
- FedRawDataInputSource* parent = nullptr)
- : parent_(parent),
- status_(status),
- lumi_(lumi),
- fileName_(name),
- deleteFile_(deleteFile),
- rawFd_(rawFd),
- fileSize_(fileSize),
- rawHeaderSize_(rawHeaderSize),
- nChunks_(nChunks),
- numFiles_(1),
- nEvents_(nEvents),
- nProcessed_(0) {
- fileNames_.push_back(name);
- fileOrder_.push_back(fileOrder_.size());
- diskFileSizes_.push_back(fileSize);
- fileSizes_.push_back(0);
- bufferOffsets_.push_back(0);
- chunks_.reserve(nChunks_);
- for (unsigned int i = 0; i < nChunks; i++)
- chunks_.push_back(nullptr);
- }
- virtual ~InputFile();
-
- void setChunks(uint16_t nChunks) {
- nChunks_ = nChunks;
- chunks_.clear();
- chunks_.reserve(nChunks_);
- for (unsigned int i = 0; i < nChunks_; i++)
- chunks_.push_back(nullptr);
- }
-
- void appendFile(std::string const& name, uint64_t size) {
- size_t prevOffset = bufferOffsets_.back();
- size_t prevSize = diskFileSizes_.back();
- numFiles_++;
- fileNames_.push_back(name);
- fileOrder_.push_back(fileOrder_.size());
- diskFileSizes_.push_back(size);
- fileSizes_.push_back(0);
- bufferOffsets_.push_back(prevOffset + prevSize);
- }
-
- bool waitForChunk(unsigned int chunkid) {
- //some atomics to make sure everything is cache synchronized for the main thread
- return chunks_[chunkid] != nullptr && chunks_[chunkid]->readComplete_;
- }
- bool advance(unsigned char*& dataPosition, const size_t size);
- void moveToPreviousChunk(const size_t size, const size_t offset);
- void rewindChunk(const size_t size);
- void unsetDeleteFile() { deleteFile_ = false; }
- void randomizeOrder(std::default_random_engine& rng) {
- std::shuffle(std::begin(fileOrder_), std::end(fileOrder_), rng);
- }
- uint64_t currentChunkSize() const { return chunks_[currentChunk_]->size_; }
- int64_t fileSizeLeft() const { return (int64_t)fileSize_ - (int64_t)bufferPosition_; }
-};
-
#endif // EventFilter_Utilities_FedRawDataInputSource_h
/// emacs configuration
diff --git a/EventFilter/Utilities/interface/SourceRawFile.h b/EventFilter/Utilities/interface/SourceRawFile.h
new file mode 100644
index 0000000000000..df87ef79f0fbf
--- /dev/null
+++ b/EventFilter/Utilities/interface/SourceRawFile.h
@@ -0,0 +1,229 @@
+#ifndef EventFilter_Utilities_SourceRawFile_h
+#define EventFilter_Utilities_SourceRawFile_h
+
+//#include
+//#include
+//#include
+//#include
+//#include
+//#include
+#include
+
+#include "FWCore/Framework/interface/EventPrincipal.h"
+#include "DataFormats/FEDRawData/interface/FEDRawDataCollection.h"
+#include "EventFilter/Utilities/interface/FedRawDataInputSource.h"
+
+//used by some models that use FEDRawDataCollection
+class UnpackedRawEventWrapper {
+public:
+ UnpackedRawEventWrapper() {}
+ ~UnpackedRawEventWrapper() {}
+ void setError(std::string msg) {
+ errmsg_ = msg;
+ error_ = true;
+ }
+ void setChecksumError(std::string msg) {
+ errmsg_ = msg;
+ checksumError_ = true;
+ }
+ void setRawData(FEDRawDataCollection* rawData) { rawData_.reset(rawData); }
+ void setAux(edm::EventAuxiliary* aux) { aux_.reset(aux); }
+ void setRun(uint32_t run) { run_ = run; }
+ FEDRawDataCollection* rawData() { return rawData_.get(); }
+ std::unique_ptr& rawDataRef() { return rawData_; }
+ edm::EventAuxiliary* aux() { return aux_.get(); }
+ uint32_t run() const { return run_; }
+ bool checksumError() const { return checksumError_; }
+ bool error() const { return error_; }
+ std::string const& errmsg() { return errmsg_; }
+
+private:
+ std::unique_ptr rawData_;
+ std::unique_ptr aux_;
+ uint32_t run_;
+ bool checksumError_ = false;
+ bool error_ = false;
+ std::string errmsg_;
+};
+
+struct InputChunk {
+ unsigned char* buf_;
+ InputChunk* next_ = nullptr;
+ uint64_t size_;
+ uint64_t usedSize_ = 0;
+ //unsigned int index_;
+ uint64_t offset_;
+ unsigned int fileIndex_;
+ std::atomic readComplete_;
+
+ InputChunk(uint64_t size) : size_(size) {
+ buf_ = new unsigned char[size_];
+ reset(0, 0, 0);
+ }
+ void reset(uint64_t newOffset, uint64_t toRead, unsigned int fileIndex) {
+ offset_ = newOffset;
+ usedSize_ = toRead;
+ fileIndex_ = fileIndex;
+ readComplete_ = false;
+ }
+
+ bool resize(uint64_t wantedSize, uint64_t maxSize) {
+ if (wantedSize > maxSize)
+ return false;
+ if (size_ < wantedSize) {
+ size_ = uint64_t(wantedSize * 1.05);
+ delete[] buf_;
+ buf_ = new unsigned char[size_];
+ }
+ return true;
+ }
+
+ ~InputChunk() { delete[] buf_; }
+};
+
+class InputFile {
+public:
+ FedRawDataInputSource* parent_;
+ evf::EvFDaqDirector::FileStatus status_;
+ unsigned int lumi_;
+ std::string fileName_;
+ //used by DAQSource
+ std::vector fileNames_;
+ std::vector diskFileSizes_;
+ std::vector bufferOffsets_;
+ std::vector fileSizes_;
+ std::vector fileOrder_;
+ bool deleteFile_;
+ int rawFd_;
+ uint64_t fileSize_;
+ uint16_t rawHeaderSize_;
+ uint16_t nChunks_;
+ uint16_t numFiles_;
+ int nEvents_;
+ unsigned int nProcessed_;
+
+ tbb::concurrent_vector chunks_;
+
+ uint32_t bufferPosition_ = 0;
+ uint32_t chunkPosition_ = 0;
+ unsigned int currentChunk_ = 0;
+
+ InputFile(evf::EvFDaqDirector::FileStatus status,
+ unsigned int lumi = 0,
+ std::string const& name = std::string(),
+ bool deleteFile = true,
+ int rawFd = -1,
+ uint64_t fileSize = 0,
+ uint16_t rawHeaderSize = 0,
+ uint16_t nChunks = 0,
+ int nEvents = 0,
+ FedRawDataInputSource* parent = nullptr)
+ : parent_(parent),
+ status_(status),
+ lumi_(lumi),
+ fileName_(name),
+ deleteFile_(deleteFile),
+ rawFd_(rawFd),
+ fileSize_(fileSize),
+ rawHeaderSize_(rawHeaderSize),
+ nChunks_(nChunks),
+ numFiles_(1),
+ nEvents_(nEvents),
+ nProcessed_(0) {
+ fileNames_.push_back(name);
+ fileOrder_.push_back(fileOrder_.size());
+ diskFileSizes_.push_back(fileSize);
+ fileSizes_.push_back(0);
+ bufferOffsets_.push_back(0);
+ chunks_.reserve(nChunks_);
+ for (unsigned int i = 0; i < nChunks; i++)
+ chunks_.push_back(nullptr);
+ }
+ virtual ~InputFile();
+
+ void setChunks(uint16_t nChunks) {
+ nChunks_ = nChunks;
+ chunks_.clear();
+ chunks_.reserve(nChunks_);
+ for (unsigned int i = 0; i < nChunks_; i++)
+ chunks_.push_back(nullptr);
+ }
+
+ void appendFile(std::string const& name, uint64_t size) {
+ size_t prevOffset = bufferOffsets_.back();
+ size_t prevSize = diskFileSizes_.back();
+ numFiles_++;
+ fileNames_.push_back(name);
+ fileOrder_.push_back(fileOrder_.size());
+ diskFileSizes_.push_back(size);
+ fileSizes_.push_back(0);
+ bufferOffsets_.push_back(prevOffset + prevSize);
+ }
+
+ bool waitForChunk(unsigned int chunkid) {
+ //some atomics to make sure everything is cache synchronized for the main thread
+ return chunks_[chunkid] != nullptr && chunks_[chunkid]->readComplete_;
+ }
+ bool advance(std::mutex& m, std::condition_variable& cv, unsigned char*& dataPosition, const size_t size);
+ bool advanceSimple(unsigned char*& dataPosition, const size_t size) {
+ size_t currentLeft = chunks_[currentChunk_]->size_ - chunkPosition_;
+ if (currentLeft < size)
+ return true;
+ dataPosition = chunks_[currentChunk_]->buf_ + chunkPosition_;
+ chunkPosition_ += size;
+ bufferPosition_ += size;
+ return false;
+ }
+ void resetPos() {
+ chunkPosition_ = 0;
+ bufferPosition_ = 0;
+ }
+ void moveToPreviousChunk(const size_t size, const size_t offset);
+ void rewindChunk(const size_t size);
+ void unsetDeleteFile() { deleteFile_ = false; }
+ void randomizeOrder(std::default_random_engine& rng) {
+ std::shuffle(std::begin(fileOrder_), std::end(fileOrder_), rng);
+ }
+ uint64_t currentChunkSize() const { return chunks_[currentChunk_]->size_; }
+ int64_t fileSizeLeft() const { return (int64_t)fileSize_ - (int64_t)bufferPosition_; }
+};
+
+class DAQSource;
+
+class RawInputFile : public InputFile {
+public:
+ RawInputFile(evf::EvFDaqDirector::FileStatus status,
+ unsigned int lumi = 0,
+ std::string const& name = std::string(),
+ bool deleteFile = true,
+ int rawFd = -1,
+ uint64_t fileSize = 0,
+ uint16_t rawHeaderSize = 0,
+ uint32_t nChunks = 0,
+ int nEvents = 0,
+ DAQSource* parent = nullptr)
+ : InputFile(status, lumi, name, deleteFile, rawFd, fileSize, rawHeaderSize, nChunks, nEvents, nullptr),
+ sourceParent_(parent) {}
+ bool advance(std::mutex& m, std::condition_variable& cv, unsigned char*& dataPosition, const size_t size);
+ void advance(const size_t size) {
+ chunkPosition_ += size;
+ bufferPosition_ += size;
+ }
+ void queue(UnpackedRawEventWrapper* ec) {
+ if (!frdcQueue_.get())
+ frdcQueue_.reset(new std::queue>());
+ std::unique_ptr uptr(ec);
+ frdcQueue_->push(std::move(uptr));
+ }
+ void popQueue(std::unique_ptr& uptr) {
+ uptr = std::move(frdcQueue_->front());
+ frdcQueue_->pop();
+ }
+
+private:
+ DAQSource* sourceParent_;
+ //optional unpacked raw data queue (currently here because DAQSource controls lifetime of the RawInputfile)
+ std::unique_ptr>> frdcQueue_;
+};
+
+#endif // EventFilter_Utilities_SourceRawFile_h
diff --git a/EventFilter/Utilities/plugins/DTHFakeReader.cc b/EventFilter/Utilities/plugins/DTHFakeReader.cc
new file mode 100644
index 0000000000000..a112c8123422b
--- /dev/null
+++ b/EventFilter/Utilities/plugins/DTHFakeReader.cc
@@ -0,0 +1,233 @@
+#include "DTHFakeReader.h"
+#include "DataFormats/FEDRawData/interface/FEDHeader.h"
+#include "DataFormats/FEDRawData/interface/FEDTrailer.h"
+#include "DataFormats/FEDRawData/interface/FEDNumbering.h"
+#include "DataFormats/TCDS/interface/TCDSRaw.h"
+
+//#include "EventFilter/Utilities/interface/GlobalEventNumber.h"
+#include "EventFilter/Utilities/interface/crc32c.h"
+
+#include "DataFormats/FEDRawData/interface/FEDRawData.h"
+
+#include "FWCore/ParameterSet/interface/ParameterSet.h"
+
+#include "CLHEP/Random/RandGauss.h"
+
+#include
+#include
+#include
+#include
+#include
+
+//using namespace edm;
+namespace evf {
+
+ constexpr unsigned minOrbitBx = 1;
+ constexpr unsigned maxOrbitBx = 2464;
+ constexpr unsigned avgEventsPerOrbit = 70;
+
+ //constexpr unsigned h_size_ = 8;//for SLink FEDs
+ //constexpr unsigned t_size_ = 8;
+
+ constexpr unsigned h_size_ = sizeof(SLinkRocketHeader_v3);
+ constexpr unsigned t_size_ = sizeof(SLinkRocketTrailer_v3);
+
+ constexpr double rndFactor = (maxOrbitBx - minOrbitBx + 1) / (double(avgEventsPerOrbit) * RAND_MAX);
+
+ DTHFakeReader::DTHFakeReader(const edm::ParameterSet& pset)
+ : fillRandom_(pset.getUntrackedParameter("fillRandom", false)),
+ meansize_(pset.getUntrackedParameter("meanSize", 1024)),
+ width_(pset.getUntrackedParameter("width", 1024)),
+ injected_errors_per_million_events_(pset.getUntrackedParameter("injectErrPpm", 0)),
+ sourceIdList_(
+ pset.getUntrackedParameter>("sourceIdList", std::vector())),
+ modulo_error_events_(injected_errors_per_million_events_ ? 1000000 / injected_errors_per_million_events_
+ : 0xffffffff) {
+ if (fillRandom_) {
+ //intialize random seed
+ auto time_count =
+ static_cast(std::chrono::high_resolution_clock::now().time_since_epoch().count());
+ std::srand(time_count & 0xffffffff);
+ }
+ produces();
+ }
+
+ void DTHFakeReader::fillRawData(edm::Event& e, FEDRawDataCollection*& data) {
+ // a null pointer is passed, need to allocate the fed collection (reusing it as container)
+ data = new FEDRawDataCollection();
+ //auto ls = e.luminosityBlock();
+ //this will be used as orbit counter
+ edm::EventNumber_t orbitId = e.id().event();
+
+ //generate eventID. Orbits start from 0 or 1?
+ std::vector eventIdList_;
+ std::map> randFedSizes;
+ for (auto sourceId : sourceIdList_) {
+ randFedSizes[sourceId] = std::map();
+ }
+
+ //randomize which orbit was accepted
+ for (unsigned i = minOrbitBx; i <= maxOrbitBx; i++) {
+ if ((std::rand() * rndFactor) < 1) {
+ uint64_t eventId = orbitId * maxOrbitBx + i;
+ eventIdList_.push_back(eventId);
+ for (auto sourceId : sourceIdList_) {
+ float logsiz = CLHEP::RandGauss::shoot(std::log(meansize_), std::log(meansize_) - std::log(width_ / 2.));
+ size_t size = int(std::exp(logsiz));
+ size -= size % 16; // all blocks aligned to 128 bit words (with header+trailer being 16, this remains valid)
+ if (!size)
+ size = 16;
+ randFedSizes[sourceId][eventId] = size;
+ }
+ }
+ }
+
+ for (auto sourceId : sourceIdList_) {
+ FEDRawData& feddata = data->FEDData(sourceId);
+
+ auto size = sizeof(DTHOrbitHeader_v1);
+ for (auto eventId : eventIdList_)
+ size += randFedSizes[sourceId][eventId] + h_size_ + t_size_ + sizeof(DTHFragmentTrailer_v1);
+ feddata.resize(size);
+
+ uint64_t fragments_size_bytes = sizeof(DTHOrbitHeader_v1);
+ //uint32_t runningChecksum = 0xffffffffU;
+ uint32_t runningChecksum = 0;
+ for (auto eventId : eventIdList_) {
+ unsigned char* fedaddr = feddata.data() + fragments_size_bytes;
+ //fragments_size_bytes += fillFED(fedaddr, sourceId, eventId, randFedSizes[sourceId][eventId], runningChecksum);
+ fragments_size_bytes +=
+ fillSLRFED(fedaddr, sourceId, eventId, orbitId, randFedSizes[sourceId][eventId], runningChecksum);
+ }
+ //in place construction
+ new (feddata.data()) DTHOrbitHeader_v1(sourceId,
+ orbitId,
+ e.id().run(),
+ fragments_size_bytes >> evf::DTH_WORD_NUM_BYTES_SHIFT,
+ eventIdList_.size(),
+ runningChecksum,
+ 0);
+ }
+ }
+
+ void DTHFakeReader::produce(edm::Event& e, edm::EventSetup const& es) {
+ edm::Handle rawdata;
+ FEDRawDataCollection* fedcoll = nullptr;
+ fillRawData(e, fedcoll);
+ std::unique_ptr bare_product(fedcoll);
+ e.put(std::move(bare_product));
+ }
+
+ uint32_t DTHFakeReader::fillSLRFED(unsigned char* buf,
+ const uint32_t sourceId,
+ edm::EventNumber_t eventId,
+ const uint32_t orbitId,
+ uint32_t size,
+ uint32_t& accum_crc32c) {
+ // Generate size...
+ const unsigned h_size_ = sizeof(SLinkRocketHeader_v3);
+ const unsigned t_size_ = sizeof(SLinkRocketTrailer_v3);
+
+ uint32_t totsize = size + h_size_ + t_size_ + sizeof(DTHFragmentTrailer_v1);
+ const unsigned fragsize = size + h_size_ + t_size_;
+
+ //Fill SLinkRocket header
+ uint8_t emu_status = 2; //set 2 indicating fragment generated by DTH
+ uint16_t l1a_types = 1; //set provisionally to 1, to be revised later
+ uint8_t l1a_subtype = 0;
+ new ((void*)buf) SLinkRocketHeader_v3(eventId, emu_status, l1a_subtype, l1a_types, sourceId);
+
+ // Payload = all 0s or random
+ if (fillRandom_) {
+ //fill FED with random values
+ size_t size_ui = size - size % sizeof(unsigned int);
+ for (size_t i = 0; i < size_ui; i += sizeof(unsigned int)) {
+ *((unsigned int*)(buf + h_size_ + i)) = (unsigned int)std::rand();
+ }
+ //remainder
+ for (size_t i = size_ui; i < size; i++) {
+ *(buf + h_size_ + i) = std::rand() & 0xff;
+ }
+ }
+
+ //Fill SLinkRocket trailer
+ uint16_t crc = 0; // FIXME : get CRC16
+ uint16_t bxid = 0;
+ uint8_t status = 0;
+ //size is in bytes, it will be converted by constructor
+ new ((void*)(buf + h_size_ + size))
+ SLinkRocketTrailer_v3(crc, fragsize >> evf::SLR_WORD_NUM_BYTES_SHIFT, bxid, orbitId, crc, status);
+
+ //fill DTH fragment trailer
+ void* dthTrailerAddr = buf + fragsize;
+ new (dthTrailerAddr) DTHFragmentTrailer_v1(fragsize >> evf::DTH_WORD_NUM_BYTES_SHIFT, 0, crc, eventId);
+
+ //accumulate crc32 checksum
+ accum_crc32c = crc32c(accum_crc32c, (const uint8_t*)buf, totsize);
+
+ return totsize;
+ }
+
+ uint32_t DTHFakeReader::fillFED(
+ unsigned char* buf, const int sourceId, edm::EventNumber_t eventId, uint32_t size, uint32_t& accum_crc32c) {
+ // Generate size...
+ const unsigned h_size = 8;
+ const unsigned t_size = 8;
+
+ //header+trailer+payload
+ uint32_t totsize = size + h_size + t_size + sizeof(DTHFragmentTrailer_v1);
+
+ // Generate header
+ //FEDHeader::set(feddata.data(),
+ FEDHeader::set(buf,
+ 1, // Trigger type
+ eventId, // LV1_id (24 bits)
+ 0, // BX_id
+ sourceId); // source_id
+
+ // Payload = all 0s or random
+ if (fillRandom_) {
+ //fill FED with random values
+ size_t size_ui = size - size % sizeof(unsigned int);
+ for (size_t i = 0; i < size_ui; i += sizeof(unsigned int)) {
+ *((unsigned int*)(buf + h_size + i)) = (unsigned int)std::rand();
+ }
+ //remainder
+ for (size_t i = size_ui; i < size; i++) {
+ *(buf + h_size + i) = std::rand() & 0xff;
+ }
+ }
+
+ // Generate trailer
+ int crc = 0; // FIXME : get CRC16
+ FEDTrailer::set(buf + h_size + size,
+ size / 8 + 2, // in 64 bit words
+ crc,
+ 0, // Evt_stat
+ 0); // TTS bits
+
+ //FIXME: accumulate crc32 checksum
+ //crc32c = 0;
+
+ void* dthTrailerAddr = buf + h_size + t_size + size;
+ new (dthTrailerAddr)
+ DTHFragmentTrailer_v1((h_size + t_size + size) >> evf::DTH_WORD_NUM_BYTES_SHIFT, 0, crc, eventId);
+ return totsize;
+ }
+
+ void DTHFakeReader::beginLuminosityBlock(edm::LuminosityBlock const& iL, edm::EventSetup const& iE) {
+ std::cout << "DTHFakeReader begin Lumi " << iL.luminosityBlock() << std::endl;
+ fakeLs_ = iL.luminosityBlock();
+ }
+
+ void DTHFakeReader::fillDescriptions(edm::ConfigurationDescriptions& descriptions) {
+ edm::ParameterSetDescription desc;
+ desc.setComment("Injector of generated DTH raw orbit fragments for DSAQ testing");
+ desc.addUntracked("fillRandom", false);
+ desc.addUntracked("meanSize", 1024);
+ desc.addUntracked("width", 1024);
+ desc.addUntracked("injectErrPpm", 1024);
+ desc.addUntracked>("sourceIdList", std::vector());
+ descriptions.add("DTHFakeReader", desc);
+ }
+} //namespace evf
diff --git a/EventFilter/Utilities/plugins/DTHFakeReader.h b/EventFilter/Utilities/plugins/DTHFakeReader.h
new file mode 100644
index 0000000000000..025ada2d33a0d
--- /dev/null
+++ b/EventFilter/Utilities/plugins/DTHFakeReader.h
@@ -0,0 +1,57 @@
+#ifndef DaqSource_DTHFakeReader_h
+#define DaqSource_DTHFakeReader_h
+
+/** \class DTHFakeReader
+ * Fills FedRawData with DTH orbits for writeout to emulate EVB file writing
+ * Proper Phase-2 headers and trailers are included;
+ */
+
+#include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
+#include "FWCore/Framework/interface/one/EDProducer.h"
+#include "FWCore/Framework/interface/Event.h"
+#include "FWCore/Framework/interface/EventSetup.h"
+#include "FWCore/Framework/interface/LuminosityBlock.h"
+#include "FWCore/Framework/interface/EventSetup.h"
+#include "DataFormats/Provenance/interface/EventID.h"
+#include "DataFormats/FEDRawData/interface/FEDRawDataCollection.h"
+#include "EventFilter/Utilities/interface/DTHHeaders.h"
+#include
+
+namespace evf {
+
+ class DTHFakeReader : public edm::one::EDProducer<> {
+ public:
+ DTHFakeReader(const edm::ParameterSet& pset);
+ ~DTHFakeReader() override {}
+
+ void produce(edm::Event&, edm::EventSetup const&) override;
+
+ static void fillDescriptions(edm::ConfigurationDescriptions& descriptions);
+
+ private:
+ // Generate and fill FED raw data for a full event
+ void fillRawData(edm::Event& e, FEDRawDataCollection*& data);
+
+ uint32_t fillSLRFED(unsigned char* buf,
+ const uint32_t sourceId,
+ edm::EventNumber_t eventId,
+ const uint32_t orbitId,
+ uint32_t size,
+ uint32_t& accum_crc32c);
+ uint32_t fillFED(
+ unsigned char* buf, const int sourceId, edm::EventNumber_t eventId, uint32_t size, uint32_t& accum_crc32c);
+ //void fillTCDSFED(edm::EventID& eID, FEDRawDataCollection& data, uint32_t ls, timeval* now);
+ virtual void beginLuminosityBlock(edm::LuminosityBlock const& iL, edm::EventSetup const& iE);
+
+ private:
+ bool fillRandom_;
+ unsigned int meansize_; // in bytes
+ unsigned int width_;
+ unsigned int injected_errors_per_million_events_;
+ std::vector sourceIdList_;
+ unsigned int modulo_error_events_;
+ unsigned int fakeLs_ = 0;
+ };
+} //namespace evf
+
+#endif
diff --git a/EventFilter/Utilities/plugins/GlobalEvFOutputModule.cc b/EventFilter/Utilities/plugins/GlobalEvFOutputModule.cc
index b545b079df607..8b77441ad6caa 100644
--- a/EventFilter/Utilities/plugins/GlobalEvFOutputModule.cc
+++ b/EventFilter/Utilities/plugins/GlobalEvFOutputModule.cc
@@ -359,10 +359,14 @@ namespace evf {
if (!edm::Service().isAvailable())
throw cms::Exception("GlobalEvFOutputModule") << "EvFDaqDirector is not available";
+ auto const& baseRunDir = edm::Service()->baseRunDir();
+ if (edm::Service()->fileListMode() && !std::filesystem::is_directory(baseRunDir))
+ std::filesystem::create_directory(baseRunDir);
+
const std::string iniFileName = edm::Service()->getInitTempFilePath(streamLabel_);
std::ofstream file(iniFileName);
if (!file)
- throw cms::Exception("GlobalEvFOutputModule") << "can not create " << iniFileName << "error: " << strerror(errno);
+ throw cms::Exception("GlobalEvFOutputModule") << "can not create " << iniFileName << "\n" << strerror(errno);
file.close();
edm::LogInfo("GlobalEvFOutputModule") << "Constructor created initemp file -: " << iniFileName;
diff --git a/EventFilter/Utilities/plugins/RawEventFileWriterForBU.cc b/EventFilter/Utilities/plugins/RawEventFileWriterForBU.cc
index f5745a159efd3..02aa3a783e6dd 100644
--- a/EventFilter/Utilities/plugins/RawEventFileWriterForBU.cc
+++ b/EventFilter/Utilities/plugins/RawEventFileWriterForBU.cc
@@ -5,8 +5,9 @@
#include
#include
-// CMSSW headers
+#include "FWCore/ServiceRegistry/interface/Service.h"
#include "EventFilter/Utilities/interface/EvFDaqDirector.h"
+#include "EventFilter/Utilities/interface/FastMonitoringService.h"
#include "EventFilter/Utilities/interface/FileIO.h"
#include "EventFilter/Utilities/interface/JSONSerializer.h"
#include "EventFilter/Utilities/plugins/RawEventFileWriterForBU.h"
@@ -24,6 +25,9 @@ using namespace edm::streamer;
RawEventFileWriterForBU::RawEventFileWriterForBU(edm::ParameterSet const& ps)
: microSleep_(ps.getParameter("microSleep")),
frdFileVersion_(ps.getParameter("frdFileVersion")) {
+ if (edm::Service().isAvailable())
+ fms_ = static_cast(edm::Service().operator->());
+
//per-file JSD and FastMonitor
rawJsonDef_.setDefaultGroup("legend");
rawJsonDef_.addLegendItem("NEvents", "integer", DataPointDefinition::SUM);
@@ -42,6 +46,7 @@ RawEventFileWriterForBU::RawEventFileWriterForBU(edm::ParameterSet const& ps)
eolJsonDef_.addLegendItem("NFiles", "integer", DataPointDefinition::SUM);
eolJsonDef_.addLegendItem("TotalEvents", "integer", DataPointDefinition::SUM);
eolJsonDef_.addLegendItem("NLostEvents", "integer", DataPointDefinition::SUM);
+ eolJsonDef_.addLegendItem("NBytes", "integer", DataPointDefinition::SUM);
perLumiEventCount_.setName("NEvents");
perLumiFileCount_.setName("NFiles");
@@ -63,17 +68,24 @@ RawEventFileWriterForBU::RawEventFileWriterForBU(edm::ParameterSet const& ps)
eorJsonDef_.addLegendItem("NFiles", "integer", DataPointDefinition::SUM);
eorJsonDef_.addLegendItem("NLumis", "integer", DataPointDefinition::SUM);
eorJsonDef_.addLegendItem("LastLumi", "integer", DataPointDefinition::SUM);
+ eorJsonDef_.addLegendItem("TotalEvents", "integer", DataPointDefinition::SUM);
+ eorJsonDef_.addLegendItem("NLostEvents", "integer", DataPointDefinition::SUM);
perRunEventCount_.setName("NEvents");
perRunFileCount_.setName("NFiles");
perRunLumiCount_.setName("NLumis");
perRunLastLumi_.setName("LastLumi");
+ perRunTotalEventCount_.setName("TotalEvents");
+ perRunLostEventCount_.setName("NLostEvents");
runMon_ = new FastMonitor(&eorJsonDef_, false);
runMon_->registerGlobalMonitorable(&perRunEventCount_, false, nullptr);
runMon_->registerGlobalMonitorable(&perRunFileCount_, false, nullptr);
runMon_->registerGlobalMonitorable(&perRunLumiCount_, false, nullptr);
runMon_->registerGlobalMonitorable(&perRunLastLumi_, false, nullptr);
+ runMon_->registerGlobalMonitorable(&perRunTotalEventCount_, false, nullptr);
+ runMon_->registerGlobalMonitorable(&perRunLostEventCount_, false, nullptr);
+
runMon_->commit(nullptr);
}
@@ -85,6 +97,23 @@ RawEventFileWriterForBU::~RawEventFileWriterForBU() {
delete runMon_;
}
+void RawEventFileWriterForBU::doOutputEvent(void* startAddress, size_t size) {
+ ssize_t retval = write(outfd_, startAddress, size);
+
+ if ((unsigned)retval != size) {
+ throw cms::Exception("RawEventFileWriterForBU", "doOutputEvent")
+ << "Error writing FED Raw Data event data to " << fileName_ << ". Possibly the output disk "
+ << "is full?" << std::endl;
+ }
+
+ // throttle event output
+ usleep(microSleep_);
+ perFileEventCount_.value()++;
+ perFileSize_.value() += size;
+
+ // cms::Adler32((const char*) msg.startAddress(), msg.size(), adlera_, adlerb_);
+}
+
void RawEventFileWriterForBU::doOutputEvent(FRDEventMsgView const& msg) {
ssize_t retval = write(outfd_, (void*)msg.startAddress(), msg.size());
@@ -102,11 +131,20 @@ void RawEventFileWriterForBU::doOutputEvent(FRDEventMsgView const& msg) {
// cms::Adler32((const char*) msg.startAddress(), msg.size(), adlera_, adlerb_);
}
-void RawEventFileWriterForBU::initialize(std::string const& destinationDir, std::string const& name, int ls) {
+void RawEventFileWriterForBU::initialize(std::string const& destinationDir,
+ std::string const& name,
+ int run,
+ unsigned int ls) {
destinationDir_ = destinationDir;
+ run_ = run;
+
+ std::stringstream ss;
+ ss << "run" << std::setfill('0') << std::setw(6) << run_;
+ runPrefix_ = ss.str();
if (outfd_ != -1) {
- finishFileWrite(ls);
+ if (!fms_ || !fms_->exceptionDetected() || !fms_->getAbortFlagForLumi(ls))
+ finishFileWrite(ls);
closefd();
}
@@ -114,37 +152,6 @@ void RawEventFileWriterForBU::initialize(std::string const& destinationDir, std:
if (!writtenJSDs_) {
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";
- std::string eorJSDName = ss.str()+"/EoR.jsd";
-
- fileMon_->setDefPath(rawJSDName);
- lumiMon_->setDefPath(eolJSDName);
- runMon_->setDefPath(eorJSDName);
-
- struct stat fstat;
- if (stat (rawJSDName.c_str(), &fstat) != 0) {
- std::string content;
- JSONSerializer::serialize(&rawJsonDef_,content);
- FileIO::writeStringToFile(rawJSDName, content);
- }
-
- if (stat (eolJSDName.c_str(), &fstat) != 0) {
- std::string content;
- JSONSerializer::serialize(&eolJsonDef_,content);
- FileIO::writeStringToFile(eolJSDName, content);
- }
-
- if (stat (eorJSDName.c_str(), &fstat) != 0) {
- std::string content;
- JSONSerializer::serialize(&eorJsonDef_,content);
- FileIO::writeStringToFile(eorJSDName, content);
- }
-*/
writtenJSDs_ = true;
}
@@ -208,7 +215,7 @@ void RawEventFileWriterForBU::writeJsds() {
}
}
-void RawEventFileWriterForBU::finishFileWrite(int ls) {
+void RawEventFileWriterForBU::finishFileWrite(unsigned int ls) {
if (frdFileVersion_ == 1) {
//rewind
lseek(outfd_, 0, SEEK_SET);
@@ -260,7 +267,7 @@ void RawEventFileWriterForBU::finishFileWrite(int ls) {
lumiOpen_ = ls;
}
-void RawEventFileWriterForBU::endOfLS(int ls) {
+void RawEventFileWriterForBU::endOfLS(unsigned int ls) {
if (outfd_ != -1) {
finishFileWrite(ls);
closefd();
@@ -269,9 +276,6 @@ void RawEventFileWriterForBU::endOfLS(int ls) {
std::ostringstream ostr;
- if (run_ == -1)
- makeRunPrefix(destinationDir_);
-
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);
@@ -282,6 +286,7 @@ void RawEventFileWriterForBU::endOfLS(int ls) {
lumiMon_->discardCollected(ls);
perRunEventCount_.value() += perLumiEventCount_.value();
+ perRunTotalEventCount_.value() = perRunEventCount_.value();
perRunFileCount_.value() += perLumiFileCount_.value();
perRunLumiCount_.value() += 1;
perRunLastLumi_.value() = ls;
@@ -299,25 +304,12 @@ void RawEventFileWriterForBU::stop() {
edm::LogInfo("RawEventFileWriterForBU") << "Writing EOR file!";
if (!destinationDir_.empty()) {
// create EoR file
- if (run_ == -1)
- makeRunPrefix(destinationDir_);
std::string path = destinationDir_ + "/" + runPrefix_ + "_ls0000_EoR.jsn";
runMon_->snap(0);
runMon_->outputFullJSON(path, 0);
}
}
-//TODO:get from DaqDirector !
-void RawEventFileWriterForBU::makeRunPrefix(std::string const& destinationDir) {
- //dirty hack: extract run number from destination directory
- std::string::size_type pos = destinationDir.rfind("/run");
- std::string run = destinationDir.substr(pos + 4);
- run_ = atoi(run.c_str());
- std::stringstream ss;
- ss << "run" << std::setfill('0') << std::setw(6) << run_;
- runPrefix_ = ss.str();
-}
-
void RawEventFileWriterForBU::extendDescription(edm::ParameterSetDescription& desc) {
desc.add("microSleep", 0);
desc.add("frdFileVersion", 0);
diff --git a/EventFilter/Utilities/plugins/RawEventFileWriterForBU.h b/EventFilter/Utilities/plugins/RawEventFileWriterForBU.h
index 6bdcefa7d5dd8..b93b5e1d51e03 100644
--- a/EventFilter/Utilities/plugins/RawEventFileWriterForBU.h
+++ b/EventFilter/Utilities/plugins/RawEventFileWriterForBU.h
@@ -18,6 +18,10 @@
#include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
#include "IOPool/Streamer/interface/FRDEventMessage.h"
+namespace evf {
+ class FastMonitoringService;
+}
+
class RawEventFileWriterForBU {
public:
explicit RawEventFileWriterForBU(edm::ParameterSet const& ps);
@@ -25,14 +29,14 @@ class RawEventFileWriterForBU {
~RawEventFileWriterForBU();
void doOutputEvent(edm::streamer::FRDEventMsgView const& msg);
+ void doOutputEvent(void* startAddress, size_t size);
edm::streamer::uint32 adler32() const { return (adlerb_ << 16) | adlera_; }
void start() {}
void stop();
- void initialize(std::string const& destinationDir, std::string const& name, int ls);
- void endOfLS(int ls);
- void makeRunPrefix(std::string const& destinationDir);
+ void initialize(std::string const& destinationDir, std::string const& name, int run, unsigned int ls);
+ void endOfLS(unsigned int ls);
static void extendDescription(edm::ParameterSetDescription& desc);
@@ -45,17 +49,20 @@ class RawEventFileWriterForBU {
} else
return false;
}
- void finishFileWrite(int ls);
+ void finishFileWrite(unsigned int ls);
void writeJsds();
int outfd_ = -1;
int run_ = -1;
std::string runPrefix_;
+ evf::FastMonitoringService* fms_ = nullptr;
jsoncollector::IntJ perRunEventCount_;
jsoncollector::IntJ perRunFileCount_;
jsoncollector::IntJ perRunLumiCount_;
jsoncollector::IntJ perRunLastLumi_;
+ jsoncollector::IntJ perRunTotalEventCount_;
+ jsoncollector::IntJ perRunLostEventCount_;
jsoncollector::IntJ perLumiEventCount_;
jsoncollector::IntJ perLumiFileCount_;
diff --git a/EventFilter/Utilities/plugins/RawEventOutputModuleForBU.h b/EventFilter/Utilities/plugins/RawEventOutputModuleForBU.h
index d504d922fe28c..0a5cf93801b3a 100644
--- a/EventFilter/Utilities/plugins/RawEventOutputModuleForBU.h
+++ b/EventFilter/Utilities/plugins/RawEventOutputModuleForBU.h
@@ -24,7 +24,6 @@
template
class RawEventOutputModuleForBU : public edm::one::OutputModule {
- typedef unsigned int uint32;
/**
* Consumers are suppose to provide:
* void doOutputEvent(const FRDEventMsgView& msg)
@@ -52,12 +51,9 @@ class RawEventOutputModuleForBU : public edm::one::OutputModule token_;
const unsigned int numEventsPerFile_;
const unsigned int frdVersion_;
- unsigned long long totsize = 0LL;
- unsigned long long writtensize = 0LL;
- unsigned long long writtenSizeLast = 0LL;
- unsigned int totevents = 0;
+ std::vector sourceIdList_;
+ unsigned int totevents_ = 0;
unsigned int index_ = 0;
- timeval startOfLastLumi;
bool firstLumi_ = true;
};
@@ -68,100 +64,106 @@ RawEventOutputModuleForBU::RawEventOutputModuleForBU(edm::ParameterSet
templateConsumer_(new Consumer(ps)),
token_(consumes(ps.getParameter("source"))),
numEventsPerFile_(ps.getParameter("numEventsPerFile")),
- frdVersion_(ps.getParameter("frdVersion")) {}
+ frdVersion_(ps.getParameter("frdVersion")),
+ sourceIdList_(ps.getUntrackedParameter>("sourceIdList", std::vector())) {
+ if (frdVersion_ > 0 && frdVersion_ < 5)
+ throw cms::Exception("RawEventOutputModuleForBU")
+ << "Generating data with FRD version " << frdVersion_ << " is no longer supported";
+ else if (frdVersion_ > edm::streamer::FRDHeaderMaxVersion)
+ throw cms::Exception("RawEventOutputModuleForBU") << "Unknown FRD version " << frdVersion_;
+}
template
RawEventOutputModuleForBU::~RawEventOutputModuleForBU() {}
template
void RawEventOutputModuleForBU::write(edm::EventForOutput const& e) {
- using namespace edm::streamer;
+ //using namespace edm::streamer;
- unsigned int ls = e.luminosityBlock();
- if (totevents > 0 && totevents % numEventsPerFile_ == 0) {
+ if (totevents_ > 0 && totevents_ % numEventsPerFile_ == 0) {
index_++;
+ unsigned int ls = e.luminosityBlock();
std::string filename = edm::Service()->getOpenRawFilePath(ls, index_);
std::string destinationDir = edm::Service()->buBaseRunDir();
- templateConsumer_->initialize(destinationDir, filename, ls);
+ int run = edm::Service()->getRunNumber();
+ templateConsumer_->initialize(destinationDir, filename, run, ls);
}
- totevents++;
+ totevents_++;
// serialize the FEDRawDataCollection into the format that we expect for
// FRDEventMsgView objects (may be better ways to do this)
edm::Handle fedBuffers;
e.getByToken(token_, fedBuffers);
- // determine the expected size of the FRDEvent IN BYTES !!!!!
- assert(frdVersion_ <= edm::streamer::FRDHeaderMaxVersion);
+ // determine the expected size of the FRDEvent IN bytes
int headerSize = edm::streamer::FRDHeaderVersionSize[frdVersion_];
int expectedSize = headerSize;
- int nFeds = frdVersion_ < 3 ? 1024 : FEDNumbering::lastFEDId() + 1;
+ int nFeds = FEDNumbering::lastFEDId() + 1;
- for (int idx = 0; idx < nFeds; ++idx) {
- FEDRawData singleFED = fedBuffers->FEDData(idx);
- expectedSize += singleFED.size();
+ if (sourceIdList_.size()) {
+ for (int idx : sourceIdList_) {
+ FEDRawData singleFED = fedBuffers->FEDData(idx);
+ expectedSize += singleFED.size();
+ }
+ } else {
+ for (int idx = 0; idx < nFeds; ++idx) {
+ FEDRawData singleFED = fedBuffers->FEDData(idx);
+ expectedSize += singleFED.size();
+ }
}
- totsize += expectedSize;
// build the FRDEvent into a temporary buffer
std::unique_ptr> workBuffer(
std::make_unique>(expectedSize + 256));
- uint32* bufPtr = (uint32*)(workBuffer.get()->data());
- if (frdVersion_ <= 5) {
- *bufPtr++ = (uint32)frdVersion_; // version number only
- } else {
- uint16 flags = 0;
- if (!e.eventAuxiliary().isRealData())
- flags |= FRDEVENT_MASK_ISGENDATA;
- *(uint16*)bufPtr = (uint16)(frdVersion_ & 0xffff);
- *((uint16*)bufPtr + 1) = flags;
- bufPtr++;
- }
- *bufPtr++ = (uint32)e.id().run();
- *bufPtr++ = (uint32)e.luminosityBlock();
- *bufPtr++ = (uint32)e.id().event();
- if (frdVersion_ == 4)
- *bufPtr++ = 0; //64-bit event id high part
-
- if (frdVersion_ < 3) {
- uint32 fedsize[1024];
- for (int idx = 0; idx < 1024; ++idx) {
- FEDRawData singleFED = fedBuffers->FEDData(idx);
- fedsize[idx] = singleFED.size();
- //std::cout << "fed size " << singleFED.size()<< std::endl;
+ uint32_t* bufPtr = (uint32_t*)(workBuffer.get()->data());
+
+ if (frdVersion_) {
+ if (frdVersion_ <= 5) {
+ //32-bits version field
+ *bufPtr++ = (uint32_t)frdVersion_;
+ } else {
+ //16 bits version and 16 bits flags
+ uint16_t flags = 0;
+ if (!e.eventAuxiliary().isRealData())
+ flags |= edm::streamer::FRDEVENT_MASK_ISGENDATA;
+ *(uint16_t*)bufPtr = (uint16_t)(frdVersion_ & 0xffff);
+ *((uint16_t*)bufPtr + 1) = flags;
+ bufPtr++;
}
- memcpy(bufPtr, fedsize, 1024 * sizeof(uint32));
- bufPtr += 1024;
- } else {
+ *bufPtr++ = (uint32_t)e.id().run();
+ *bufPtr++ = (uint32_t)e.luminosityBlock();
+ *bufPtr++ = (uint32_t)e.id().event();
*bufPtr++ = expectedSize - headerSize;
*bufPtr++ = 0;
- if (frdVersion_ <= 4)
- *bufPtr++ = 0;
}
- uint32* payloadPtr = bufPtr;
- for (int idx = 0; idx < nFeds; ++idx) {
- FEDRawData singleFED = fedBuffers->FEDData(idx);
- if (singleFED.size() > 0) {
- memcpy(bufPtr, singleFED.data(), singleFED.size());
- bufPtr += singleFED.size() / 4;
+ uint32_t* payloadPtr = bufPtr;
+ if (sourceIdList_.size())
+ for (int idx : sourceIdList_) {
+ FEDRawData singleFED = fedBuffers->FEDData(idx);
+ if (singleFED.size() > 0) {
+ memcpy(bufPtr, singleFED.data(), singleFED.size());
+ bufPtr += singleFED.size() / 4;
+ }
}
- }
- if (frdVersion_ > 4) {
+ else
+ for (int idx = 0; idx < nFeds; ++idx) {
+ FEDRawData singleFED = fedBuffers->FEDData(idx);
+ if (singleFED.size() > 0) {
+ memcpy(bufPtr, singleFED.data(), singleFED.size());
+ bufPtr += singleFED.size() / 4;
+ }
+ }
+ if (frdVersion_) {
//crc32c checksum
uint32_t crc = 0;
*(payloadPtr - 1) = crc32c(crc, (const unsigned char*)payloadPtr, expectedSize - headerSize);
- } else if (frdVersion_ >= 3) {
- //adler32 checksum
- uint32 adlera = 1;
- uint32 adlerb = 0;
- cms::Adler32((const char*)payloadPtr, expectedSize - headerSize, adlera, adlerb);
- *(payloadPtr - 1) = (adlerb << 16) | adlera;
- }
- // create the FRDEventMsgView and use the template consumer to write it out
- edm::streamer::FRDEventMsgView msg(workBuffer.get()->data());
- writtensize += msg.size();
-
- templateConsumer_->doOutputEvent(msg);
+ // create the FRDEventMsgView and use the template consumer to write it out
+ edm::streamer::FRDEventMsgView msg(workBuffer.get()->data());
+ templateConsumer_->doOutputEvent(msg);
+ } else {
+ //write only raw FEDs
+ templateConsumer_->doOutputEvent((void*)workBuffer.get()->data(), expectedSize);
+ }
}
template
@@ -180,25 +182,13 @@ void RawEventOutputModuleForBU::beginLuminosityBlock(edm::LuminosityBl
index_ = 0;
std::string filename = edm::Service()->getOpenRawFilePath(ls.id().luminosityBlock(), index_);
std::string destinationDir = edm::Service()->buBaseRunDir();
+ int run = edm::Service()->getRunNumber();
std::cout << " writing to destination dir " << destinationDir << " name: " << filename << std::endl;
- templateConsumer_->initialize(destinationDir, filename, ls.id().luminosityBlock());
- //edm::Service()->updateBuLock(ls.id().luminosityBlock()+1);
+ templateConsumer_->initialize(destinationDir, filename, run, ls.id().luminosityBlock());
if (!firstLumi_) {
- timeval now;
- ::gettimeofday(&now, nullptr);
- //long long elapsedusec = (now.tv_sec - startOfLastLumi.tv_sec)*1000000+now.tv_usec-startOfLastLumi.tv_usec;
- /* std::cout << "(now.tv_sec - startOfLastLumi.tv_sec) " << now.tv_sec <<"-" << startOfLastLumi.tv_sec */
- /* <<" (now.tv_usec-startOfLastLumi.tv_usec) " << now.tv_usec << "-" << startOfLastLumi.tv_usec << std::endl; */
- /* std::cout << "elapsedusec " << elapsedusec << " totevents " << totevents << " size (GB)" << writtensize */
- /* << " rate " << (writtensize-writtenSizeLast)/elapsedusec << " MB/s" <()->writeLsStatisticsBU(ls.id().luminosityBlock(), totevents, totsize, elapsedusec);
- } else
- ::gettimeofday(&startOfLastLumi, nullptr);
- totevents = 0;
- totsize = 0LL;
- firstLumi_ = false;
+ totevents_ = 0;
+ firstLumi_ = false;
+ }
}
template
@@ -213,6 +203,7 @@ void RawEventOutputModuleForBU::fillDescriptions(edm::ConfigurationDes
desc.add("source", edm::InputTag("rawDataCollector"));
desc.add("numEventsPerFile", 100);
desc.add("frdVersion", 6);
+ desc.addUntracked>("sourceIdList", std::vector());
Consumer::extendDescription(desc);
descriptions.addWithDefaultLabel(desc);
diff --git a/EventFilter/Utilities/plugins/modules.cc b/EventFilter/Utilities/plugins/modules.cc
index 32f6fe5c7ad54..a648bbd920aed 100644
--- a/EventFilter/Utilities/plugins/modules.cc
+++ b/EventFilter/Utilities/plugins/modules.cc
@@ -3,6 +3,7 @@
#include "EventFilter/Utilities/interface/FedRawDataInputSource.h"
#include "EventFilter/Utilities/interface/DAQSource.h"
#include "EventFilter/Utilities/plugins/DaqFakeReader.h"
+#include "EventFilter/Utilities/plugins/DTHFakeReader.h"
#include "EventFilter/Utilities/plugins/EvFBuildingThrottle.h"
#include "EventFilter/Utilities/plugins/EvFFEDSelector.h"
#include "EventFilter/Utilities/plugins/EvFFEDExcluder.h"
@@ -27,5 +28,6 @@ DEFINE_FWK_MODULE(ExceptionGenerator);
DEFINE_FWK_MODULE(EvFFEDSelector);
DEFINE_FWK_MODULE(EvFFEDExcluder);
DEFINE_FWK_MODULE(DaqFakeReader);
+DEFINE_FWK_MODULE(DTHFakeReader);
DEFINE_FWK_INPUT_SOURCE(FedRawDataInputSource);
DEFINE_FWK_INPUT_SOURCE(DAQSource);
diff --git a/EventFilter/Utilities/src/DAQSource.cc b/EventFilter/Utilities/src/DAQSource.cc
index bcc63f7e70bd9..07c062d7b3375 100644
--- a/EventFilter/Utilities/src/DAQSource.cc
+++ b/EventFilter/Utilities/src/DAQSource.cc
@@ -5,10 +5,12 @@
#include
#include
+#include "EventFilter/Utilities/interface/SourceRawFile.h"
#include "EventFilter/Utilities/interface/DAQSource.h"
#include "EventFilter/Utilities/interface/DAQSourceModels.h"
#include "EventFilter/Utilities/interface/DAQSourceModelsFRD.h"
#include "EventFilter/Utilities/interface/DAQSourceModelsScoutingRun3.h"
+#include "EventFilter/Utilities/interface/DAQSourceModelsDTH.h"
#include "FWCore/Framework/interface/Event.h"
#include "FWCore/Framework/interface/InputSourceDescription.h"
@@ -36,6 +38,7 @@ DAQSource::DAQSource(edm::ParameterSet const& pset, edm::InputSourceDescription
eventChunkSize_(uint64_t(pset.getUntrackedParameter("eventChunkSize")) << 20),
maxChunkSize_(uint64_t(pset.getUntrackedParameter("maxChunkSize")) << 20),
eventChunkBlock_(uint64_t(pset.getUntrackedParameter("eventChunkBlock")) << 20),
+ numConcurrentReads_(pset.getUntrackedParameter("numConcurrentReads", -1)),
numBuffers_(pset.getUntrackedParameter("numBuffers")),
maxBufferedFiles_(pset.getUntrackedParameter("maxBufferedFiles")),
alwaysStartFromFirstLS_(pset.getUntrackedParameter("alwaysStartFromFirstLS", false)),
@@ -80,10 +83,14 @@ DAQSource::DAQSource(edm::ParameterSet const& pset, edm::InputSourceDescription
//load mode class based on parameter
if (dataModeConfig_ == "FRD") {
dataMode_ = std::make_shared(this);
+ } else if (dataModeConfig_ == "FRDPreUnpack") {
+ dataMode_ = std::make_shared(this);
} else if (dataModeConfig_ == "FRDStriped") {
dataMode_ = std::make_shared(this);
} else if (dataModeConfig_ == "ScoutingRun3") {
dataMode_ = std::make_shared(this);
+ } else if (dataModeConfig_ == "DTH") {
+ dataMode_ = std::make_shared(this, verifyChecksum_);
} else
throw cms::Exception("DAQSource::DAQSource") << "Unknown data mode " << dataModeConfig_;
@@ -95,6 +102,7 @@ DAQSource::DAQSource(edm::ParameterSet const& pset, edm::InputSourceDescription
long autoRunNumber = -1;
if (fileListMode_) {
autoRunNumber = initFileList();
+ daqDirector_->setFileListMode();
if (!fileListLoopMode_) {
if (autoRunNumber < 0)
throw cms::Exception("DAQSource::DAQSource") << "Run number not found from filename";
@@ -123,9 +131,11 @@ DAQSource::DAQSource(edm::ParameterSet const& pset, edm::InputSourceDescription
if (!numBuffers_)
throw cms::Exception("DAQSource::DAQSource") << "no reading enabled with numBuffers parameter 0";
- numConcurrentReads_ = numBuffers_ - 1;
+ if (numConcurrentReads_ <= 0)
+ numConcurrentReads_ = numBuffers_ - 1;
assert(numBuffers_ > 1);
readingFilesCount_ = 0;
+ heldFilesCount_ = 0;
if (!crc32c_hw_test())
edm::LogError("DAQSource::DAQSource") << "Intel crc32c checksum computation unavailable";
@@ -149,8 +159,6 @@ DAQSource::DAQSource(edm::ParameterSet const& pset, edm::InputSourceDescription
cms::Exception("DAQSource") << "EvFDaqDirector not found";
edm::LogInfo("DAQSource") << "EvFDaqDirector/Source configured to use file service";
- //set DaqDirector to delete files in preGlobalEndLumi callback
- daqDirector_->setDeleteTracking(&fileDeleteLock_, &filesToDelete_);
if (fms_) {
daqDirector_->setFMS(fms_);
fms_->setInputSource(this);
@@ -165,7 +173,7 @@ DAQSource::DAQSource(edm::ParameterSet const& pset, edm::InputSourceDescription
quit_threads_ = false;
//prepare data shared by threads
- for (unsigned int i = 0; i < numConcurrentReads_; i++) {
+ for (unsigned int i = 0; i < (unsigned)numConcurrentReads_; i++) {
thread_quit_signal.push_back(false);
workerJob_.push_back(ReaderInfo(nullptr, nullptr));
cvReader_.push_back(std::make_unique());
@@ -173,7 +181,7 @@ DAQSource::DAQSource(edm::ParameterSet const& pset, edm::InputSourceDescription
}
//start threads
- for (unsigned int i = 0; i < numConcurrentReads_; i++) {
+ for (unsigned int i = 0; i < (unsigned)numConcurrentReads_; i++) {
//wait for each thread to complete initialization
std::unique_lock lk(startupLock_);
workerThreads_.push_back(new std::thread(&DAQSource::readWorker, this, i));
@@ -186,6 +194,9 @@ DAQSource::DAQSource(edm::ParameterSet const& pset, edm::InputSourceDescription
DAQSource::~DAQSource() {
quit_threads_ = true;
+ if (startedSupervisorThread_)
+ fileDeleterThread_->join();
+
//delete any remaining open files
if (!fms_ || !fms_->exceptionDetected()) {
std::unique_lock lkw(fileDeleteLock_);
@@ -233,6 +244,10 @@ void DAQSource::fillDescriptions(edm::ConfigurationDescriptions& descriptions) {
"Block size used in a single file read call (must be smaller or equal to the initial chunk buffer size). If "
"0 is specified, use chunk size.");
+ desc.addUntracked("numConcurrentReads", -1)
+ ->setComment(
+ "Max number of concurrent reads. If not positive, it will "
+ "be set to numBuffers - 1");
desc.addUntracked("numBuffers", 2)->setComment("Number of buffers used for reading input");
desc.addUntracked("maxBufferedFiles", 2)
->setComment("Maximum number of simultaneously buffered raw files");
@@ -258,6 +273,8 @@ edm::RawInputSource::Next DAQSource::checkNext() {
//this thread opens new files and dispatches reading to worker readers
readSupervisorThread_ = std::make_unique(&DAQSource::readSupervisor, this);
+ //this thread deletes files out of the main loop
+ fileDeleterThread_ = std::make_unique(&DAQSource::fileDeleter, this);
startedSupervisorThread_ = true;
startupCv_.wait(lk);
@@ -348,7 +365,7 @@ void DAQSource::maybeOpenNewLumiSection(const uint32_t lumiSection) {
evf::EvFDaqDirector::FileStatus DAQSource::getNextEventFromDataBlock() {
setMonState(inChecksumEvent);
- bool found = dataMode_->nextEventView();
+ bool found = dataMode_->nextEventView(currentFile_.get());
//file(s) completely parsed
if (!found) {
if (dataMode_->dataBlockInitialized()) {
@@ -362,7 +379,8 @@ evf::EvFDaqDirector::FileStatus DAQSource::getNextEventFromDataBlock() {
if (verifyChecksum_ && !dataMode_->checksumValid()) {
if (fms_)
fms_->setExceptionDetected(currentLumiSection_);
- throw cms::Exception("DAQSource::getNextEventFromDataBlock") << dataMode_->getChecksumError();
+ throw cms::Exception("DAQSource::getNextEventFromDataBlock")
+ << "InvalidChecksum - " << dataMode_->getChecksumError();
}
setMonState(inCachedEvent);
@@ -412,6 +430,7 @@ evf::EvFDaqDirector::FileStatus DAQSource::getNextDataBlock() {
//file is empty
if (!currentFile_->fileSize_) {
readingFilesCount_--;
+ heldFilesCount_--;
//try to open new lumi
assert(currentFile_->nChunks_ == 0);
if (currentFile_->lumi_ > currentLumiSection_) {
@@ -427,13 +446,35 @@ evf::EvFDaqDirector::FileStatus DAQSource::getNextDataBlock() {
//file is finished
if (currentFile_->bufferPosition_ == currentFile_->fileSize_) {
readingFilesCount_--;
+ if (fileListMode_)
+ heldFilesCount_--;
//release last chunk (it is never released elsewhere)
freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_]);
if (currentFile_->nEvents_ >= 0 && currentFile_->nEvents_ != int(currentFile_->nProcessed_)) {
+ std::stringstream str;
+ for (auto& s : currentFile_->fileNames_) {
+ struct stat bufs;
+ if (stat(s.c_str(), &bufs) != 0)
+ throw cms::Exception("DAQSource::getNextDataBlock") << "Could not stat file " << s;
+ str << s << " (size:" << (bufs.st_size / 1000) << " kB)" << std::endl;
+ }
+ throw cms::Exception("DAQSource::getNextDataBlock")
+ << "Fully processed " << currentFile_->nProcessed_ << " from:" << std::endl
+ << str.str() << "but according to RAW header there should be " << currentFile_->nEvents_
+ << " events. Check previous error log for details.";
+ } else if (dataMode_->errorDetected()) {
+ std::stringstream str;
+ for (auto& s : currentFile_->fileNames_) {
+ struct stat bufs;
+ if (stat(s.c_str(), &bufs) != 0)
+ throw cms::Exception("DAQSource::getNextDataBlock") << "Could not stat file " << s;
+ str << s << " (size:" << (bufs.st_size / 1000) << " kB)" << std::endl;
+ }
throw cms::Exception("DAQSource::getNextDataBlock")
- << "Fully processed " << currentFile_->nProcessed_ << " from the file " << currentFile_->fileName_
- << " but according to BU JSON there should be " << currentFile_->nEvents_ << " events";
+ << "Processed " << currentFile_->nProcessed_ << " from:" << std::endl
+ << str.str() << "but there was a mismatch detected by the data model. Check previous error log for details.";
}
+ setMonState(inReadCleanup);
if (!daqDirector_->isSingleStreamThread() && !fileListMode_) {
//put the file in pending delete list;
std::unique_lock lkw(fileDeleteLock_);
@@ -443,6 +484,7 @@ evf::EvFDaqDirector::FileStatus DAQSource::getNextDataBlock() {
//in single-thread and stream jobs, events are already processed
currentFile_.reset();
}
+ setMonState(inProcessingFile);
return evf::EvFDaqDirector::noFile;
}
@@ -466,7 +508,7 @@ evf::EvFDaqDirector::FileStatus DAQSource::getNextDataBlock() {
currentFile_->advance(currentFile_->rawHeaderSize_);
}
- //file is too short to fit event header
+ //file is too short to fit event (or event block, orbit...) header
if (currentFile_->fileSizeLeft() < dataMode_->headerSize())
throw cms::Exception("DAQSource::getNextDataBlock")
<< "Premature end of input file while reading event header. Missing: "
@@ -478,7 +520,8 @@ evf::EvFDaqDirector::FileStatus DAQSource::getNextDataBlock() {
{
IdleSourceSentry ids(fms_);
while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
- usleep(10000);
+ std::unique_lock lkw(mWakeup_);
+ cvWakeupAll_.wait_for(lkw, std::chrono::milliseconds(100));
if (setExceptionState_)
threadError();
}
@@ -490,19 +533,25 @@ evf::EvFDaqDirector::FileStatus DAQSource::getNextDataBlock() {
unsigned char* dataPosition;
//read event header, copy it to a single chunk if necessary
- chunkEnd = currentFile_->advance(dataPosition, dataMode_->headerSize());
+ chunkEnd = currentFile_->advance(mWakeup_, cvWakeupAll_, dataPosition, dataMode_->headerSize());
//get buffer size of current chunk (can be resized)
uint64_t currentChunkSize = currentFile_->currentChunkSize();
- //prepare view based on header that was read
- dataMode_->makeDataBlockView(dataPosition, currentChunkSize, currentFile_->fileSizes_, currentFile_->rawHeaderSize_);
+ //prepare view based on header that was read. It could parse through the whole buffer for fitToBuffer models
+ dataMode_->makeDataBlockView(dataPosition, currentFile_.get());
- //check that payload size is within the file
+ if (verifyChecksum_ && !dataMode_->blockChecksumValid()) {
+ if (fms_)
+ fms_->setExceptionDetected(currentLumiSection_);
+ throw cms::Exception("DAQSource::getNextDataBlock") << dataMode_->getChecksumError();
+ }
+
+ //check that the (remaining) payload size is within the file
const size_t msgSize = dataMode_->dataBlockSize() - dataMode_->headerSize();
if (currentFile_->fileSizeLeft() < (int64_t)msgSize)
- throw cms::Exception("DAQSource::getNextEventDataBlock")
+ throw cms::Exception("DAQSource::getNextDataBlock")
<< "Premature end of input file (missing:" << (msgSize - currentFile_->fileSizeLeft())
<< ") while parsing block";
@@ -521,18 +570,17 @@ evf::EvFDaqDirector::FileStatus DAQSource::getNextDataBlock() {
{
IdleSourceSentry ids(fms_);
//do the copy to the beginning of the starting chunk. move pointers for next event in the next chunk
- chunkEnd = currentFile_->advance(dataPosition, dataMode_->headerSize() + msgSize);
+ chunkEnd = currentFile_->advance(mWakeup_, cvWakeupAll_, dataPosition, dataMode_->headerSize() + msgSize);
assert(chunkEnd);
//mark to release old chunk
chunkIsFree_ = true;
}
setMonState(inChunkReceived);
//header and payload is moved, update view
- dataMode_->makeDataBlockView(
- dataPosition, currentFile_->currentChunkSize(), currentFile_->fileSizes_, currentFile_->rawHeaderSize_);
+ dataMode_->makeDataBlockView(dataPosition, currentFile_.get());
} else {
- //everything is in a single chunk, only move pointers forward
- chunkEnd = currentFile_->advance(dataPosition, msgSize);
+ //everything is in a single chunk, only move pointers forward. Also used for fitToBuffer models
+ chunkEnd = currentFile_->advance(mWakeup_, cvWakeupAll_, dataPosition, msgSize);
assert(!chunkEnd);
chunkIsFree_ = false;
}
@@ -554,41 +602,76 @@ void DAQSource::read(edm::EventPrincipal& eventPrincipal) {
eventsThisLumi_++;
setMonState(inReadCleanup);
- //resize vector if needed
- while (streamFileTracker_.size() <= eventPrincipal.streamID())
+ //resize vector if needed (lock if resizing needed)
+ while (streamFileTracker_.size() <= eventPrincipal.streamID()) {
+ std::unique_lock lkw(fileDeleteLock_);
streamFileTracker_.push_back(-1);
+ }
streamFileTracker_[eventPrincipal.streamID()] = currentFileIndex_;
- //this old file check runs no more often than every 10 events
- if (!((currentFile_->nProcessed_ - 1) % (checkEvery_))) {
- //delete files that are not in processing
- std::unique_lock lkw(fileDeleteLock_);
- auto it = filesToDelete_.begin();
- while (it != filesToDelete_.end()) {
- bool fileIsBeingProcessed = false;
- for (unsigned int i = 0; i < streamFileTracker_.size(); i++) {
- if (it->first == streamFileTracker_.at(i)) {
- fileIsBeingProcessed = true;
- break;
- }
- }
- if (!fileIsBeingProcessed && !(fms_ && fms_->isExceptionOnData(it->second->lumi_))) {
- it = filesToDelete_.erase(it);
- } else
- it++;
- }
- }
+ setMonState(inNoRequest);
if (dataMode_->dataBlockCompleted() && chunkIsFree_) {
freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_ - 1]);
chunkIsFree_ = false;
}
- setMonState(inNoRequest);
return;
}
void DAQSource::rewind_() {}
+void DAQSource::fileDeleter() {
+ bool stop = false;
+
+ while (!stop) {
+ std::vector deleteVec;
+ {
+ unsigned int lastFileLS = 0;
+ bool fileLSOpen = false;
+ std::unique_lock lkw(fileDeleteLock_);
+ auto it = filesToDelete_.begin();
+ while (it != filesToDelete_.end()) {
+ bool fileIsBeingProcessed = false;
+ //check if file LS has already reached global EoL, reuse cached check
+ if (!(lastFileLS && lastFileLS == it->second->lumi_)) {
+ lastFileLS = it->second->lumi_;
+ fileLSOpen = daqDirector_->lsWithFilesOpen(lastFileLS);
+ }
+ for (unsigned int i = 0; i < streamFileTracker_.size(); i++) {
+ if (it->first == streamFileTracker_.at(i)) {
+ //only skip if LS is open
+ if (fileLSOpen) {
+ fileIsBeingProcessed = true;
+ break;
+ }
+ }
+ }
+ if (!fileIsBeingProcessed && (!fms_ || !fms_->isExceptionOnData(it->second->lumi_))) {
+ std::string fileToDelete = it->second->fileName_;
+ //do not actuallt delete, but do it later
+ deleteVec.push_back(it->second.get());
+ //deletion will happen later
+ it->second.release();
+ it = filesToDelete_.erase(it);
+ } else
+ it++;
+ }
+ }
+ //do this after lock is released to avoid contention
+ for (auto v : deleteVec) {
+ //deletion happens here
+ delete v;
+ heldFilesCount_--;
+ }
+ deleteVec.clear();
+
+ if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed))
+ stop = true;
+
+ usleep(500000);
+ }
+}
+
void DAQSource::dataArranger() {}
void DAQSource::readSupervisor() {
@@ -611,7 +694,11 @@ void DAQSource::readSupervisor() {
//wait for at least one free thread and chunk
int counter = 0;
- while (workerPool_.empty() || freeChunks_.empty() || readingFilesCount_ >= maxBufferedFiles_) {
+ //held files include files queued in the deleting thread.
+ //We require no more than maxBufferedFiles + 2 of total held files until deletion
+
+ while (workerPool_.empty() || freeChunks_.empty() || readingFilesCount_ >= maxBufferedFiles_ ||
+ heldFilesCount_ >= maxBufferedFiles_ + 2) {
//report state to monitoring
if (fms_) {
bool copy_active = false;
@@ -620,6 +707,8 @@ void DAQSource::readSupervisor() {
copy_active = true;
if (readingFilesCount_ >= maxBufferedFiles_)
setMonStateSup(inSupFileLimit);
+ if (heldFilesCount_ >= maxBufferedFiles_ + 2)
+ setMonStateSup(inSupFileHeldLimit);
else if (freeChunks_.empty()) {
if (copy_active)
setMonStateSup(inSupWaitFreeChunkCopying);
@@ -637,14 +726,13 @@ void DAQSource::readSupervisor() {
if (cvWakeup_.wait_for(lkw, std::chrono::milliseconds(100)) == std::cv_status::timeout) {
counter++;
if (!(counter % 6000)) {
- edm::LogWarning("FedRawDataInputSource")
- << "No free chunks or threads. Worker pool empty:" << workerPool_.empty()
- << ", free chunks empty:" << freeChunks_.empty() << ", number of files buffered:" << readingFilesCount_
- << " / " << maxBufferedFiles_;
+ edm::LogWarning("DAQSource") << "No free chunks or threads. Worker pool empty:" << workerPool_.empty()
+ << ", free chunks empty:" << freeChunks_.empty()
+ << ", number of files buffered (held):" << readingFilesCount_ << "("
+ << heldFilesCount_ << ")"
+ << " / " << maxBufferedFiles_;
}
LogDebug("DAQSource") << "No free chunks or threads...";
- } else {
- assert(!workerPool_.empty() || freeChunks_.empty());
}
if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed)) {
stop = true;
@@ -810,6 +898,10 @@ void DAQSource::readSupervisor() {
}
}
currentLumiSection = ls;
+
+ //wakeup main thread for the new non-data file obj
+ std::unique_lock lkw(mWakeup_);
+ cvWakeupAll_.notify_all();
}
//else
if (currentLumiSection > 0 && ls < currentLumiSection) {
@@ -851,7 +943,7 @@ void DAQSource::readSupervisor() {
struct stat st;
int stat_res = stat(rawFile.c_str(), &st);
if (stat_res == -1) {
- edm::LogError("DAQSource") << "Can not stat file (" << errno << "):-" << rawFile << std::endl;
+ edm::LogError("DAQSource") << "Can not stat file (" << errno << ") :- " << rawFile << std::endl;
setExceptionState_ = true;
break;
}
@@ -974,6 +1066,7 @@ void DAQSource::readSupervisor() {
newInputFile->randomizeOrder(rng_);
readingFilesCount_++;
+ heldFilesCount_++;
auto newInputFilePtr = newInputFile.get();
fileQueue_.push(std::move(newInputFile));
@@ -1070,7 +1163,7 @@ void DAQSource::readWorker(unsigned int tid) {
tid_active_[tid] = false;
std::unique_lock lk(mReader_);
workerJob_[tid].first = nullptr;
- workerJob_[tid].first = nullptr;
+ workerJob_[tid].second = nullptr;
assert(!thread_quit_signal[tid]); //should never get it here
workerPool_.push(tid);
@@ -1081,6 +1174,7 @@ void DAQSource::readWorker(unsigned int tid) {
startupCv_.notify_one();
}
cvWakeup_.notify_all();
+
cvReader_[tid]->wait(lk);
lk.unlock();
@@ -1192,8 +1286,8 @@ void DAQSource::readWorker(unsigned int tid) {
bufferLeft += last;
}
if ((uint64_t)last < eventChunkBlock_) { //last read
- edm::LogInfo("DAQSource") << "chunkUsedSize" << chunk->usedSize_ << " u-s:" << (chunk->usedSize_ - skipped)
- << " ix:" << i * eventChunkBlock_ << " " << (size_t)last;
+ LogDebug("DAQSource") << "chunkUsedSize:" << chunk->usedSize_ << " u-s:" << (chunk->usedSize_ - skipped)
+ << " ix:" << i * eventChunkBlock_ << " " << (size_t)last;
//check if this is last block if single file, then total read size must match file size
if (file->numFiles_ == 1 && !(chunk->usedSize_ - skipped == i * eventChunkBlock_ + (size_t)last)) {
edm::LogError("DAQSource") << "readWorker failed to read file -: " << file->fileName_
@@ -1317,9 +1411,23 @@ void DAQSource::readWorker(unsigned int tid) {
}
assert(dataMode_->versionCheck());
- chunk->readComplete_ =
- true; //this is atomic to secure the sequential buffer fill before becoming available for processing)
- file->chunks_[chunk->fileIndex_] = chunk; //put the completed chunk in the file chunk vector at predetermined index
+ //for models that unpack RAW data in reader thread (must fit to buffer!)
+
+ //put the completed chunk in the file chunk vector at predetermined index
+ file->chunks_[chunk->fileIndex_] = chunk;
+
+ if (dataMode_->fitToBuffer())
+ dataMode_->unpackFile(file);
+
+ //possibly not needed (except for better timing on cvWakeupAll_)
+ std::unique_lock lkw(mWakeup_);
+
+ //this is atomic to ensure all previous writes are consistent
+ chunk->readComplete_ = true;
+
+ //wakeup for chunk
+ cvWakeupAll_.notify_all();
+ //lkw.unlock();
}
}
@@ -1338,16 +1446,16 @@ void DAQSource::setMonStateSup(evf::FastMonState::InputState state) {
fms_->setInStateSup(state);
}
-bool RawInputFile::advance(unsigned char*& dataPosition, const size_t size) {
+bool RawInputFile::advance(std::mutex& m, std::condition_variable& cv, unsigned char*& dataPosition, const size_t size) {
+ sourceParent_->setMonState(inWaitChunk);
//wait for chunk
-
while (!waitForChunk(currentChunk_)) {
- sourceParent_->setMonState(inWaitChunk);
- usleep(100000);
- sourceParent_->setMonState(inChunkReceived);
+ std::unique_lock lk(m);
+ cv.wait_for(lk, std::chrono::milliseconds(100));
if (sourceParent_->exceptionState())
sourceParent_->threadError();
}
+ sourceParent_->setMonState(inChunkReceived);
dataPosition = chunks_[currentChunk_]->buf_ + chunkPosition_;
size_t currentLeft = chunks_[currentChunk_]->size_ - chunkPosition_;
@@ -1355,13 +1463,15 @@ bool RawInputFile::advance(unsigned char*& dataPosition, const size_t size) {
if (currentLeft < size) {
//we need next chunk
assert(chunks_.size() > currentChunk_ + 1);
+
+ sourceParent_->setMonState(inWaitChunk);
while (!waitForChunk(currentChunk_ + 1)) {
- sourceParent_->setMonState(inWaitChunk);
- usleep(100000);
- sourceParent_->setMonState(inChunkReceived);
+ std::unique_lock lk(m);
+ cv.wait_for(lk, std::chrono::milliseconds(100));
if (sourceParent_->exceptionState())
sourceParent_->threadError();
}
+ sourceParent_->setMonState(inChunkReceived);
//copy everything to beginning of the first chunk
dataPosition -= chunkPosition_;
assert(dataPosition == chunks_[currentChunk_]->buf_);
diff --git a/EventFilter/Utilities/src/DAQSourceModelsDTH.cc b/EventFilter/Utilities/src/DAQSourceModelsDTH.cc
new file mode 100644
index 0000000000000..6c492d573f827
--- /dev/null
+++ b/EventFilter/Utilities/src/DAQSourceModelsDTH.cc
@@ -0,0 +1,240 @@
+#include "EventFilter/Utilities/interface/DAQSource.h"
+#include "EventFilter/Utilities/interface/DAQSourceModelsDTH.h"
+
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+
+#include "FWCore/MessageLogger/interface/MessageLogger.h"
+
+#include "DataFormats/FEDRawData/interface/FEDHeader.h"
+#include "DataFormats/FEDRawData/interface/FEDTrailer.h"
+#include "DataFormats/FEDRawData/interface/FEDRawDataCollection.h"
+
+#include "DataFormats/TCDS/interface/TCDSRaw.h"
+
+#include "FWCore/Framework/interface/Event.h"
+#include "EventFilter/Utilities/interface/GlobalEventNumber.h"
+#include "EventFilter/Utilities/interface/DAQSourceModels.h"
+#include "EventFilter/Utilities/interface/DAQSource.h"
+
+#include "EventFilter/Utilities/interface/AuxiliaryMakers.h"
+
+#include "DataFormats/Provenance/interface/EventAuxiliary.h"
+#include "DataFormats/Provenance/interface/EventID.h"
+#include "DataFormats/Provenance/interface/Timestamp.h"
+#include "EventFilter/Utilities/interface/crc32c.h"
+
+using namespace evf;
+
+void DataModeDTH::readEvent(edm::EventPrincipal& eventPrincipal) {
+ std::unique_ptr rawData(new FEDRawDataCollection);
+ edm::Timestamp tstamp = fillFEDRawDataCollection(*rawData);
+
+ edm::EventID eventID = edm::EventID(daqSource_->eventRunNumber(), daqSource_->currentLumiSection(), nextEventID_);
+ edm::EventAuxiliary aux(
+ eventID, daqSource_->processGUID(), tstamp, isRealData(), edm::EventAuxiliary::PhysicsTrigger);
+ aux.setProcessHistoryID(daqSource_->processHistoryID());
+ daqSource_->makeEventWrapper(eventPrincipal, aux);
+
+ std::unique_ptr edp(new edm::Wrapper(std::move(rawData)));
+ eventPrincipal.put(
+ daqProvenanceHelpers_[0]->branchDescription(), std::move(edp), daqProvenanceHelpers_[0]->dummyProvenance());
+ eventCached_ = false;
+}
+
+edm::Timestamp DataModeDTH::fillFEDRawDataCollection(FEDRawDataCollection& rawData) {
+ //generate timestamp for this event until parsing of TCDS2 data is available
+ edm::TimeValue_t time;
+ timeval stv;
+ gettimeofday(&stv, nullptr);
+ time = stv.tv_sec;
+ time = (time << 32) + stv.tv_usec;
+ edm::Timestamp tstamp(time);
+
+ for (size_t i = 0; i < eventFragments_.size(); i++) {
+ auto fragTrailer = eventFragments_[i];
+ uint8_t* payload = (uint8_t*)fragTrailer->payload();
+ auto fragSize = fragTrailer->payloadSizeBytes();
+ /*
+ //Slink header and trailer
+ assert(fragSize >= (FEDTrailer::length + FEDHeader::length));
+ const FEDHeader fedHeader(payload);
+ const FEDTrailer fedTrailer((uint8_t*)fragTrailer - FEDTrailer::length);
+ const uint32_t fedSize = fedTrailer.fragmentLength() << 3; //trailer length counts in 8 bytes
+ const uint16_t fedId = fedHeader.sourceID();
+*/
+
+ //SLinkRocket header and trailer
+ if (fragSize < sizeof(SLinkRocketTrailer_v3) + sizeof(SLinkRocketHeader_v3))
+ throw cms::Exception("DAQSource::DAQSourceModelsDTH") << "Invalid fragment size: " << fragSize;
+
+ const SLinkRocketHeader_v3* fedHeader = (const SLinkRocketHeader_v3*)payload;
+ const SLinkRocketTrailer_v3* fedTrailer =
+ (const SLinkRocketTrailer_v3*)((uint8_t*)fragTrailer - sizeof(SLinkRocketTrailer_v3));
+
+ //check SLR trailer first as it comes just before fragmen trailer
+ if (!fedTrailer->verifyMarker())
+ throw cms::Exception("DAQSource::DAQSourceModelsDTH") << "Invalid SLinkRocket trailer";
+ if (!fedHeader->verifyMarker())
+ throw cms::Exception("DAQSource::DAQSourceModelsDTH") << "Invalid SLinkRocket header";
+
+ const uint32_t fedSize = fedTrailer->eventLenBytes();
+ const uint16_t fedId = fedHeader->sourceID();
+
+ /*
+ * @SM: CRC16 in trailer was not checked up to Run3, no need to do production check.
+ * if we already check orbit CRC32.If CRC16 check is to be added,
+ * in phase1 crc16 was calculated on sequential 64-byte little-endian words
+ * (see FWCore/Utilities/interface/CRC16.h).
+ * See also optimized pclmulqdq implementation in XDAQ.
+ * Note: check if for phase-2 crc16 is still based on 8-byte words
+ */
+ //const uint32_t crc16 = fedTrailer->crc();
+
+ if (fedSize != fragSize)
+ throw cms::Exception("DAQSource::DAQSourceModelsDTH")
+ << "Fragment size mismatch. From DTHTrailer: " << fragSize << " and from SLinkRocket trailer: " << fedSize;
+ FEDRawData& fedData = rawData.FEDData(fedId);
+ fedData.resize(fedSize);
+ memcpy(fedData.data(), payload, fedSize); //copy with header and trailer
+ }
+ return tstamp;
+}
+
+std::vector>& DataModeDTH::makeDaqProvenanceHelpers() {
+ //set FRD data collection
+ daqProvenanceHelpers_.clear();
+ daqProvenanceHelpers_.emplace_back(std::make_shared(
+ edm::TypeID(typeid(FEDRawDataCollection)), "FEDRawDataCollection", "FEDRawDataCollection", "DAQSource"));
+ return daqProvenanceHelpers_;
+}
+
+void DataModeDTH::makeDataBlockView(unsigned char* addr, RawInputFile* rawFile) {
+ //TODO: optimize by merging into a pair or tuple and reserve size
+ addrsEnd_.clear();
+ addrsStart_.clear();
+ constexpr size_t hsize = sizeof(evf::DTHOrbitHeader_v1);
+
+ LogDebug("DataModeDTH::makeDataBlockView") << "blockAddr: 0x" << std::hex << (uint64_t)addr << " chunkOffset: 0x"
+ << std::hex << (uint64_t)(addr - rawFile->chunks_[0]->buf_);
+
+ //intial orbit header was advanced over by source
+ size_t maxAllowedSize = rawFile->fileSizeLeft() + headerSize();
+ auto nextAddr = addr;
+ checksumValid_ = true;
+ if (!checksumError_.empty())
+ checksumError_ = std::string();
+
+ firstOrbitHeader_ = nullptr;
+ while (nextAddr < addr + maxAllowedSize) {
+ //ensure header fits
+ assert(nextAddr + hsize < addr + maxAllowedSize);
+
+ auto orbitHeader = (evf::DTHOrbitHeader_v1*)(nextAddr);
+ if (!orbitHeader->verifyMarker())
+ throw cms::Exception("DAQSource::DAQSourceModelsDTH") << "Invalid DTH orbit marker";
+ if (!firstOrbitHeader_) {
+ firstOrbitHeader_ = orbitHeader;
+ } else {
+ assert(orbitHeader->runNumber() == firstOrbitHeader_->runNumber());
+ if (orbitHeader->orbitNumber() != firstOrbitHeader_->orbitNumber()) {
+ firstOrbitHeader_ = orbitHeader;
+ //next orbit ID reached, do not include this orbit in this block
+ break;
+ }
+ }
+
+ auto srcOrbitSize = orbitHeader->totalSize();
+ auto nextEnd = nextAddr + srcOrbitSize;
+ assert(nextEnd <= addr + maxAllowedSize); //boundary check
+
+ if (verifyChecksum_) {
+ auto crc = crc32c(0U, (const uint8_t*)orbitHeader->payload(), orbitHeader->payloadSizeBytes());
+ if (crc != orbitHeader->crc()) {
+ checksumValid_ = false;
+ if (!checksumError_.empty())
+ checksumError_ += "\n";
+ checksumError_ +=
+ fmt::format("Found a wrong crc32c checksum in orbit: {} sourceID: {}. Expected {:x} but calculated {:x}",
+ orbitHeader->orbitNumber(),
+ orbitHeader->sourceID(),
+ orbitHeader->crc(),
+ crc);
+ }
+ }
+
+ addrsStart_.push_back(nextAddr + hsize);
+ addrsEnd_.push_back(nextAddr + srcOrbitSize);
+ nextAddr += srcOrbitSize;
+ }
+ dataBlockSize_ = nextAddr - addr;
+
+ eventCached_ = false;
+ nextEventView(rawFile);
+ eventCached_ = true;
+}
+
+bool DataModeDTH::nextEventView(RawInputFile*) {
+ blockCompleted_ = false;
+ if (eventCached_)
+ return true;
+
+ bool blockCompletedAll = !addrsEnd_.empty() ? true : false;
+ bool blockCompletedAny = false;
+ eventFragments_.clear();
+ size_t last_eID = 0;
+
+ for (size_t i = 0; i < addrsEnd_.size(); i++) {
+ evf::DTHFragmentTrailer_v1* trailer =
+ (evf::DTHFragmentTrailer_v1*)(addrsEnd_[i] - sizeof(evf::DTHFragmentTrailer_v1));
+
+ if (!trailer->verifyMarker())
+ throw cms::Exception("DAQSource::DAQSourceModelsDTH") << "Invalid DTH trailer marker";
+
+ assert((uint8_t*)trailer >= addrsStart_[i]);
+
+ uint64_t eID = trailer->eventID();
+ eventFragments_.push_back(trailer);
+ auto payload_size = trailer->payloadSizeBytes();
+ if (payload_size > evf::SLR_MAX_EVENT_LEN) //max possible by by SlinkRocket (1 MB)
+ throw cms::Exception("DAQSource::DAQSourceModelsDTH")
+ << "DTHFragment size " << payload_size << " larger than the SLinkRocket limit of " << evf::SLR_MAX_EVENT_LEN;
+
+ if (i == 0) {
+ nextEventID_ = eID;
+ last_eID = eID;
+ } else if (last_eID != nextEventID_)
+ throw cms::Exception("DAQSource::DAQSourceModelsDTH") << "Inconsistent event number between fragments";
+
+ //update address array
+ addrsEnd_[i] -= sizeof(evf::DTHFragmentTrailer_v1) + payload_size;
+
+ if (trailer->flags())
+ throw cms::Exception("DAQSource::DAQSourceModelsDTH")
+ << "Detected error condition in DTH trailer of event " << trailer->eventID()
+ << " flags: " << std::bitset<16>(trailer->flags());
+
+ if (addrsEnd_[i] == addrsStart_[i]) {
+ blockCompletedAny = true;
+ } else {
+ assert(addrsEnd_[i] > addrsStart_[i]);
+ blockCompletedAll = false;
+ }
+ }
+ if (blockCompletedAny != blockCompletedAll)
+ throw cms::Exception("DAQSource::DAQSourceModelsDTH")
+ << "Some orbit sources have inconsistent number of event fragments.";
+
+ if (blockCompletedAll) {
+ blockCompleted_ = blockCompletedAll;
+ firstOrbitHeader_ = nullptr;
+ return false;
+ }
+ return true;
+}
diff --git a/EventFilter/Utilities/src/DAQSourceModelsFRD.cc b/EventFilter/Utilities/src/DAQSourceModelsFRD.cc
index 22afae18caa4c..47508811220bb 100644
--- a/EventFilter/Utilities/src/DAQSourceModelsFRD.cc
+++ b/EventFilter/Utilities/src/DAQSourceModelsFRD.cc
@@ -123,7 +123,15 @@ std::vector>& DataModeFRD::makeD
return daqProvenanceHelpers_;
}
-bool DataModeFRD::nextEventView() {
+void DataModeFRD::makeDataBlockView(unsigned char* addr, RawInputFile* rawFile) {
+ dataBlockAddr_ = addr;
+ dataBlockMax_ = rawFile->currentChunkSize();
+ eventCached_ = false;
+ nextEventView(rawFile);
+ eventCached_ = true;
+}
+
+bool DataModeFRD::nextEventView(RawInputFile*) {
if (eventCached_)
return true;
event_ = std::make_unique(dataBlockAddr_);
@@ -132,17 +140,19 @@ bool DataModeFRD::nextEventView() {
<< " event id:" << event_->event() << " lumi:" << event_->lumi() << " run:" << event_->run()
<< " of size:" << event_->size() << " bytes does not fit into a chunk of size:" << dataBlockMax_ << " bytes";
}
+ if (event_->version() < 5)
+ throw cms::Exception("DAQSource::getNextEvent")
+ << "Unsupported FRD version " << event_->version() << ". Minimum supported is v5.";
return true;
}
bool DataModeFRD::checksumValid() {
crc_ = 0;
- if (event_->version() >= 5) {
- crc_ = crc32c(crc_, (const unsigned char*)event_->payload(), event_->eventSize());
- if (crc_ != event_->crc32c())
- return false;
- }
- return true;
+ crc_ = crc32c(crc_, (const unsigned char*)event_->payload(), event_->eventSize());
+ if (crc_ != event_->crc32c())
+ return false;
+ else
+ return true;
}
std::string DataModeFRD::getChecksumError() const {
@@ -152,7 +162,202 @@ std::string DataModeFRD::getChecksumError() const {
}
/*
- * FRD Multi Test
+ * FRD preRead
+ */
+
+void DataModeFRDPreUnpack::unpackEvent(edm::streamer::FRDEventMsgView* eview, UnpackedRawEventWrapper* ec) {
+ //TODO: also walk the file and build checksum
+ FEDRawDataCollection* rawData = new FEDRawDataCollection;
+ bool tcdsInRange;
+ unsigned char* tcds_pointer = nullptr;
+ std::string errmsg;
+ bool err = false;
+ edm::Timestamp tstamp = fillFEDRawDataCollection(eview, *rawData, tcdsInRange, tcds_pointer, err, errmsg);
+ ec->setRawData(rawData);
+
+ uint32_t L1EventID = eview->event();
+ if (err) {
+ ec->setError(errmsg);
+ } else if (daqSource_->useL1EventID()) {
+ edm::EventID eventID = edm::EventID(daqSource_->eventRunNumber(), daqSource_->currentLumiSection(), L1EventID);
+ ec->setAux(new edm::EventAuxiliary(
+ eventID, daqSource_->processGUID(), tstamp, eview->isRealData(), edm::EventAuxiliary::PhysicsTrigger));
+ ec->aux()->setProcessHistoryID(daqSource_->processHistoryID());
+ } else if (tcds_pointer == nullptr) {
+ std::stringstream ss;
+ ss << "No TCDS FED in event with FEDHeader EID -: " << L1EventID;
+ ec->setError(ss.str());
+ } else {
+ const FEDHeader fedHeader(tcds_pointer);
+ tcds::Raw_v1 const* tcds = reinterpret_cast(tcds_pointer + FEDHeader::length);
+ edm::EventAuxiliary* aux = new edm::EventAuxiliary(); //allocate empty aux
+ *aux = evf::evtn::makeEventAuxiliary(tcds,
+ daqSource_->eventRunNumber(),
+ daqSource_->currentLumiSection(),
+ eview->isRealData(),
+ static_cast(fedHeader.triggerType()),
+ daqSource_->processGUID(),
+ !daqSource_->fileListLoopMode(),
+ !tcdsInRange);
+ ec->setAux(aux);
+ ec->aux()->setProcessHistoryID(daqSource_->processHistoryID());
+ ec->setRun(eview->run());
+ }
+}
+
+void DataModeFRDPreUnpack::readEvent(edm::EventPrincipal& eventPrincipal) {
+ if (ec_->error())
+ throw cms::Exception("DAQSource::read") << ec_->errmsg();
+
+ daqSource_->makeEventWrapper(eventPrincipal, *ec_->aux());
+
+ std::unique_ptr edp(new edm::Wrapper(std::move(ec_->rawDataRef())));
+ eventPrincipal.put(
+ daqProvenanceHelpers_[0]->branchDescription(), std::move(edp), daqProvenanceHelpers_[0]->dummyProvenance());
+}
+
+void DataModeFRDPreUnpack::unpackFile(RawInputFile* currentFile) {
+ const uint64_t fileSize = currentFile->fileSize_;
+ const unsigned rawHeaderSize = currentFile->rawHeaderSize_;
+
+ //TODO: set threadError for issues in this function
+ if (rawHeaderSize > 0) {
+ assert(fileSize >= rawHeaderSize);
+ }
+ assert(fileSize >= headerSize());
+
+ uint64_t bufpos = rawHeaderSize;
+
+ while (bufpos < fileSize) { //loop while there is file/events to read
+
+ assert(bufpos + headerSize() <= fileSize);
+
+ //fit to buffer model
+ auto dataBlockAddr = (unsigned char*)currentFile->chunks_[0]->buf_ + bufpos;
+
+ //first view for header only, check if it fits
+ auto eview = std::make_unique(dataBlockAddr);
+
+ assert(bufpos + eview->size() <= fileSize);
+ bufpos += eview->size();
+
+ //create event wrapper
+ //we will store this per each event queued to fwk
+ UnpackedRawEventWrapper* ec = new UnpackedRawEventWrapper();
+
+ assert(eview->version() >= 5);
+
+ //crc check
+ uint32_t crc = crc32c(0, (const unsigned char*)eview->payload(), eview->eventSize());
+ if (crc != eview->crc32c()) {
+ std::stringstream ss;
+ ss << "Found a wrong crc32c checksum: expected 0x" << std::hex << eview->crc32c() << " but calculated 0x" << crc;
+ ec->setChecksumError(ss.str());
+ //unpackEvent(eview.get(), ec);
+ } else
+ unpackEvent(eview.get(), ec);
+ currentFile->queue(ec);
+ }
+}
+
+edm::Timestamp DataModeFRDPreUnpack::fillFEDRawDataCollection(edm::streamer::FRDEventMsgView* eview,
+ FEDRawDataCollection& rawData,
+ bool& tcdsInRange,
+ unsigned char*& tcds_pointer,
+ bool& err,
+ std::string& errmsg) {
+ edm::TimeValue_t time;
+ timeval stv;
+ gettimeofday(&stv, nullptr);
+ time = stv.tv_sec;
+ time = (time << 32) + stv.tv_usec;
+ edm::Timestamp tstamp(time);
+
+ uint32_t eventSize = eview->eventSize();
+ unsigned char* event = (unsigned char*)eview->payload();
+ tcds_pointer = nullptr;
+ tcdsInRange = false;
+ uint16_t selectedTCDSFed = 0;
+ while (eventSize > 0) {
+ assert(eventSize >= FEDTrailer::length);
+ eventSize -= FEDTrailer::length;
+ const FEDTrailer fedTrailer(event + eventSize);
+ const uint32_t fedSize = fedTrailer.fragmentLength() << 3; //trailer length counts in 8 bytes
+ assert(eventSize >= fedSize - FEDHeader::length);
+ eventSize -= (fedSize - FEDHeader::length);
+ const FEDHeader fedHeader(event + eventSize);
+ const uint16_t fedId = fedHeader.sourceID();
+ if (fedId > FEDNumbering::MAXFEDID) {
+ err = true;
+ std::stringstream str;
+ str << "Out of range FED ID : " << fedId;
+ errmsg = str.str();
+ return tstamp;
+ } else if (fedId >= MINTCDSuTCAFEDID_ && fedId <= MAXTCDSuTCAFEDID_) {
+ if (!selectedTCDSFed) {
+ selectedTCDSFed = fedId;
+ tcds_pointer = event + eventSize;
+ if (fedId >= FEDNumbering::MINTCDSuTCAFEDID && fedId <= FEDNumbering::MAXTCDSuTCAFEDID) {
+ tcdsInRange = true;
+ }
+ } else {
+ err = true;
+ std::stringstream str;
+ str << "Second TCDS FED ID " << fedId << " found. First ID: " << selectedTCDSFed;
+ errmsg = str.str();
+ return tstamp;
+ }
+ }
+ //take event ID from GTPE FED
+ FEDRawData& fedData = rawData.FEDData(fedId);
+ fedData.resize(fedSize);
+ memcpy(fedData.data(), event + eventSize, fedSize);
+ }
+ assert(eventSize == 0);
+
+ return tstamp;
+}
+
+std::vector>& DataModeFRDPreUnpack::makeDaqProvenanceHelpers() {
+ //set FRD data collection
+ daqProvenanceHelpers_.clear();
+ daqProvenanceHelpers_.emplace_back(std::make_shared(
+ edm::TypeID(typeid(FEDRawDataCollection)), "FEDRawDataCollection", "FEDRawDataCollection", "DAQSource"));
+ return daqProvenanceHelpers_;
+}
+
+void DataModeFRDPreUnpack::makeDataBlockView(unsigned char* addr, RawInputFile* rawFile) {
+ dataBlockAddr_ = addr;
+ dataBlockMax_ = rawFile->currentChunkSize();
+ eventCached_ = false;
+ nextEventView(rawFile);
+ eventCached_ = true;
+}
+
+bool DataModeFRDPreUnpack::nextEventView(RawInputFile* currentFile) {
+ if (eventCached_)
+ return true;
+ event_ = std::make_unique(dataBlockAddr_);
+ if (event_->size() > dataBlockMax_) {
+ throw cms::Exception("DAQSource::getNextEvent")
+ << " event id:" << event_->event() << " lumi:" << event_->lumi() << " run:" << event_->run()
+ << " of size:" << event_->size() << " bytes does not fit into a chunk of size:" << dataBlockMax_ << " bytes";
+ }
+
+ if (event_->version() < 5)
+ throw cms::Exception("DAQSource::getNextEvent")
+ << "Unsupported FRD version " << event_->version() << ". Minimum supported is v5.";
+
+ currentFile->popQueue(ec_);
+ return true;
+}
+
+bool DataModeFRDPreUnpack::checksumValid() { return !ec_->checksumError(); }
+
+std::string DataModeFRDPreUnpack::getChecksumError() const { return ec_->errmsg(); }
+
+/*
+ * FRD Multi Source
*/
void DataModeFRDStriped::makeDirectoryEntries(std::vector const& baseDirs,
@@ -282,15 +487,13 @@ bool DataModeFRDStriped::checksumValid() {
for (size_t i = 0; i < events_.size(); i++) {
uint32_t crc = 0;
auto const& event = events_[i];
- if (event->version() >= 5) {
- crc = crc32c(crc, (const unsigned char*)event->payload(), event->eventSize());
- if (crc != event->crc32c()) {
- std::ostringstream ss;
- ss << "Found a wrong crc32c checksum at readout index " << i << ": expected 0x" << std::hex << event->crc32c()
- << " but calculated 0x" << crc << ". ";
- crcMsg_ += ss.str();
- status = false;
- }
+ crc = crc32c(crc, (const unsigned char*)event->payload(), event->eventSize());
+ if (crc != event->crc32c()) {
+ std::ostringstream ss;
+ ss << "Found a wrong crc32c checksum at readout index " << i << ": expected 0x" << std::hex << event->crc32c()
+ << " but calculated 0x" << crc << ". ";
+ crcMsg_ += ss.str();
+ status = false;
}
}
return status;
@@ -322,7 +525,32 @@ std::pair> DataModeFRDStriped::defineAdditionalFi
return std::make_pair(true, additionalFiles);
}
-bool DataModeFRDStriped::nextEventView() {
+void DataModeFRDStriped::makeDataBlockView(unsigned char* addr, RawInputFile* rawFile) {
+ fileHeaderSize_ = rawFile->rawHeaderSize_;
+ std::vector const& fileSizes = rawFile->fileSizes_;
+ numFiles_ = fileSizes.size();
+ //add offset address for each file payload
+ dataBlockAddrs_.clear();
+ dataBlockAddrs_.push_back(addr);
+ dataBlockMaxAddrs_.clear();
+ dataBlockMaxAddrs_.push_back(addr + fileSizes[0] - fileHeaderSize_);
+ auto fileAddr = addr;
+ for (unsigned int i = 1; i < fileSizes.size(); i++) {
+ fileAddr += fileSizes[i - 1];
+ dataBlockAddrs_.push_back(fileAddr);
+ dataBlockMaxAddrs_.push_back(fileAddr + fileSizes[i] - fileHeaderSize_);
+ }
+
+ dataBlockMax_ = rawFile->currentChunkSize();
+ blockCompleted_ = false;
+ //set event cached as we set initial address here
+ bool result = makeEvents();
+ assert(result);
+ eventCached_ = true;
+ setDataBlockInitialized(true);
+}
+
+bool DataModeFRDStriped::nextEventView(RawInputFile*) {
blockCompleted_ = false;
if (eventCached_)
return true;
@@ -336,16 +564,15 @@ bool DataModeFRDStriped::nextEventView() {
bool DataModeFRDStriped::makeEvents() {
events_.clear();
assert(!blockCompleted_);
+ int completed = 0;
+
for (int i = 0; i < numFiles_; i++) {
if (dataBlockAddrs_[i] >= dataBlockMaxAddrs_[i]) {
//must be exact
assert(dataBlockAddrs_[i] == dataBlockMaxAddrs_[i]);
blockCompleted_ = true;
- return false;
- } else {
- if (blockCompleted_)
- throw cms::Exception("DataModeFRDStriped::makeEvents")
- << "not all striped blocks were completed at the same time";
+ completed++;
+ continue;
}
if (blockCompleted_)
continue;
@@ -354,6 +581,18 @@ bool DataModeFRDStriped::makeEvents() {
throw cms::Exception("DAQSource::getNextEvent")
<< " event id:" << events_[i]->event() << " lumi:" << events_[i]->lumi() << " run:" << events_[i]->run()
<< " of size:" << events_[i]->size() << " bytes does not fit into the buffer or has corrupted header";
+
+ if (events_[i]->version() < 5)
+ throw cms::Exception("DAQSource::getNextEvent")
+ << "Unsupported FRD version " << events_[i]->version() << ". Minimum supported is v5.";
+ }
+ if (completed < numFiles_) {
+ for (int i = 0; i < numFiles_; i++) {
+ if (dataBlockAddrs_[i] == dataBlockMaxAddrs_[i]) {
+ edm::LogError("dataModeFRDStriped::makeEvents") << "incomplete file block read from directory " << buPaths_[i];
+ errorDetected_ = true;
+ }
+ }
}
return !blockCompleted_;
}
diff --git a/EventFilter/Utilities/src/DAQSourceModelsScoutingRun3.cc b/EventFilter/Utilities/src/DAQSourceModelsScoutingRun3.cc
index fff4240566716..fec04eaa08e92 100644
--- a/EventFilter/Utilities/src/DAQSourceModelsScoutingRun3.cc
+++ b/EventFilter/Utilities/src/DAQSourceModelsScoutingRun3.cc
@@ -109,7 +109,7 @@ std::vector