Utilities/interface/FedRawDataInputSource.h

0001 #ifndef EventFilter_Utilities_FedRawDataInputSource_h
0002 #define EventFilter_Utilities_FedRawDataInputSource_h
0003
0004 #include <condition_variable>
0005 #include <cstdio>
0006 #include <filesystem>
0007 #include <memory>
0008 #include <mutex>
0009 #include <thread>
0010 #include <random>
0011 #include <algorithm>
0012
0013 #include "oneapi/tbb/concurrent_queue.h"
0014 #include "oneapi/tbb/concurrent_vector.h"
0015
0016 #include "DataFormats/Provenance/interface/ProcessHistoryID.h"
0017 #include "DataFormats/Provenance/interface/Timestamp.h"
0018 #include "EventFilter/Utilities/interface/EvFDaqDirector.h"
0019 #include "FWCore/Sources/interface/RawInputSource.h"
0020 #include "FWCore/Framework/interface/EventPrincipal.h"
0021 #include "FWCore/Sources/interface/DaqProvenanceHelper.h"
0022 #include "FWCore/ServiceRegistry/interface/Service.h"
0023 #include "IOPool/Streamer/interface/FRDEventMessage.h"
0024
0025 #include "DataFormats/FEDRawData/interface/FEDNumbering.h"
0026 #include "DataFormats/Provenance/interface/LuminosityBlockAuxiliary.h"
0027
0028 class FEDRawDataCollection;
0029 class InputSourceDescription;
0030 class ParameterSet;
0031
0032 class InputFile;
0033 struct InputChunk;
0034
0035 namespace evf {
0036   class FastMonitoringService;
0037   namespace FastMonState {
0038     enum InputState : short;
0039   }
0040 }  // namespace evf
0041
0042 class FedRawDataInputSource : public edm::RawInputSource {
0043   friend class InputFile;
0044   friend struct InputChunk;
0045
0046 public:
0047   explicit FedRawDataInputSource(edm::ParameterSet const&, edm::InputSourceDescription const&);
0048   ~FedRawDataInputSource() override;
0049   static void fillDescriptions(edm::ConfigurationDescriptions& descriptions);
0050
0051   std::pair<bool, unsigned int> getEventReport(unsigned int lumi, bool erase);
0052
0053 protected:
0054   Next checkNext() override;
0055   void read(edm::EventPrincipal& eventPrincipal) override;
0056   void setMonState(evf::FastMonState::InputState state);
0057   void setMonStateSup(evf::FastMonState::InputState state);
0058
0059 private:
0060   void rewind_() override;
0061
0062   void maybeOpenNewLumiSection(const uint32_t lumiSection);
0063   evf::EvFDaqDirector::FileStatus nextEvent();
0064   evf::EvFDaqDirector::FileStatus getNextEvent();
0065   edm::Timestamp fillFEDRawDataCollection(FEDRawDataCollection& rawData, bool& tcdsInRange);
0066
0067   void readSupervisor();
0068   void readWorker(unsigned int tid);
0069   void threadError();
0070   bool exceptionState() { return setExceptionState_; }
0071
0072   //functions for single buffered reader
0073   void readNextChunkIntoBuffer(InputFile* file);
0074
0075   //monitoring
0076   void reportEventsThisLumiInSource(unsigned int lumi, unsigned int events);
0077
0078   long initFileList();
0079   evf::EvFDaqDirector::FileStatus getFile(unsigned int& ls,
0080                                           std::string& nextFile,
0081                                           uint32_t& fsize,
0082                                           uint64_t& lockWaitTime);
0083
0084   //variables
0085   evf::FastMonitoringService* fms_ = nullptr;
0086   evf::EvFDaqDirector* daqDirector_ = nullptr;
0087
0088   std::string defPath_;
0089
0090   unsigned int eventChunkSize_;   // for buffered read-ahead
0091   unsigned int eventChunkBlock_;  // how much read(2) asks at the time
0092   unsigned int readBlocks_;
0093   unsigned int numBuffers_;
0094   unsigned int maxBufferedFiles_;
0095   unsigned int numConcurrentReads_;
0096   std::atomic<unsigned int> readingFilesCount_;
0097
0098   // get LS from filename instead of event header
0099   const bool getLSFromFilename_;
0100   const bool alwaysStartFromFirstLS_;
0101   const bool verifyChecksum_;
0102   const bool useL1EventID_;
0103   const std::vector<unsigned int> testTCDSFEDRange_;
0104   std::vector<std::string> fileNames_;
0105   bool useFileBroker_;
0106   //std::vector<std::string> fileNamesSorted_;
0107
0108   const bool fileListMode_;
0109   unsigned int fileListIndex_ = 0;
0110   const bool fileListLoopMode_;
0111   unsigned int loopModeIterationInc_ = 0;
0112
0113   edm::RunNumber_t runNumber_;
0114   std::string fuOutputDir_;
0115
0116   const edm::DaqProvenanceHelper daqProvenanceHelper_;
0117
0118   std::unique_ptr<edm::streamer::FRDEventMsgView> event_;
0119
0120   edm::EventID eventID_;
0121   edm::ProcessHistoryID processHistoryID_;
0122
0123   unsigned int currentLumiSection_;
0124   uint32_t eventRunNumber_ = 0;
0125   uint32_t GTPEventID_ = 0;
0126   uint32_t L1EventID_ = 0;
0127   unsigned char* tcds_pointer_;
0128   unsigned int eventsThisLumi_;
0129   unsigned long eventsThisRun_ = 0;
0130
0131   uint16_t MINTCDSuTCAFEDID_ = FEDNumbering::MINTCDSuTCAFEDID;
0132   uint16_t MAXTCDSuTCAFEDID_ = FEDNumbering::MAXTCDSuTCAFEDID;
0133
0134   /*
0135    *
0136    * Multithreaded file reader
0137    *
0138    **/
0139
0140   typedef std::pair<InputFile*, InputChunk*> ReaderInfo;
0141
0142   uint16_t detectedFRDversion_ = 0;
0143   std::unique_ptr<InputFile> currentFile_;
0144   bool chunkIsFree_ = false;
0145
0146   bool startedSupervisorThread_ = false;
0147   std::unique_ptr<std::thread> readSupervisorThread_;
0148   std::vector<std::thread*> workerThreads_;
0149
0150   tbb::concurrent_queue<unsigned int> workerPool_;
0151   std::vector<ReaderInfo> workerJob_;
0152
0153   tbb::concurrent_queue<InputChunk*> freeChunks_;
0154   tbb::concurrent_queue<std::unique_ptr<InputFile>> fileQueue_;
0155
0156   std::mutex mReader_;
0157   std::vector<std::unique_ptr<std::condition_variable>> cvReader_;
0158   std::vector<unsigned int> tid_active_;
0159
0160   std::atomic<bool> quit_threads_;
0161   std::vector<unsigned int> thread_quit_signal;
0162   bool setExceptionState_ = false;
0163   std::mutex startupLock_;
0164   std::condition_variable startupCv_;
0165
0166   int currentFileIndex_ = -1;
0167   std::list<std::pair<int, std::unique_ptr<InputFile>>> filesToDelete_;
0168   std::list<std::pair<int, std::string>> fileNamesToDelete_;
0169   std::mutex fileDeleteLock_;
0170   std::vector<int> streamFileTracker_;
0171   unsigned int checkEvery_ = 10;
0172
0173   //supervisor thread wakeup
0174   std::mutex mWakeup_;
0175   std::condition_variable cvWakeup_;
0176
0177   //variables for the single buffered mode
0178   bool singleBufferMode_;
0179   int fileDescriptor_ = -1;
0180   uint32_t bufferInputRead_ = 0;
0181
0182   std::atomic<bool> threadInit_;
0183
0184   std::map<unsigned int, unsigned int> sourceEventsReport_;
0185   std::mutex monlock_;
0186 };
0187
0188 struct InputChunk {
0189   unsigned char* buf_;
0190   InputChunk* next_ = nullptr;
0191   uint64_t size_;
0192   uint64_t usedSize_ = 0;
0193   //unsigned int index_;
0194   uint64_t offset_;
0195   unsigned int fileIndex_;
0196   std::atomic<bool> readComplete_;
0197
0198   InputChunk(uint64_t size) : size_(size) {
0199     buf_ = new unsigned char[size_];
0200     reset(0, 0, 0);
0201   }
0202   void reset(uint64_t newOffset, uint64_t toRead, unsigned int fileIndex) {
0203     offset_ = newOffset;
0204     usedSize_ = toRead;
0205     fileIndex_ = fileIndex;
0206     readComplete_ = false;
0207   }
0208
0209   bool resize(uint64_t wantedSize, uint64_t maxSize) {
0210     if (wantedSize > maxSize)
0211       return false;
0212     if (size_ < wantedSize) {
0213       size_ = uint64_t(wantedSize * 1.05);
0214       delete[] buf_;
0215       buf_ = new unsigned char[size_];
0216     }
0217     return true;
0218   }
0219
0220   ~InputChunk() { delete[] buf_; }
0221 };
0222
0223 class InputFile {
0224 public:
0225   FedRawDataInputSource* parent_;
0226   evf::EvFDaqDirector::FileStatus status_;
0227   unsigned int lumi_;
0228   std::string fileName_;
0229   //used by DAQSource
0230   std::vector<std::string> fileNames_;
0231   std::vector<uint64_t> diskFileSizes_;
0232   std::vector<uint64_t> bufferOffsets_;
0233   std::vector<uint64_t> fileSizes_;
0234   std::vector<unsigned int> fileOrder_;
0235   bool deleteFile_;
0236   int rawFd_;
0237   uint64_t fileSize_;
0238   uint16_t rawHeaderSize_;
0239   uint16_t nChunks_;
0240   uint16_t numFiles_;
0241   int nEvents_;
0242   unsigned int nProcessed_;
0243
0244   tbb::concurrent_vector<InputChunk*> chunks_;
0245
0246   uint32_t bufferPosition_ = 0;
0247   uint32_t chunkPosition_ = 0;
0248   unsigned int currentChunk_ = 0;
0249
0250   InputFile(evf::EvFDaqDirector::FileStatus status,
0251             unsigned int lumi = 0,
0252             std::string const& name = std::string(),
0253             bool deleteFile = true,
0254             int rawFd = -1,
0255             uint64_t fileSize = 0,
0256             uint16_t rawHeaderSize = 0,
0257             uint16_t nChunks = 0,
0258             int nEvents = 0,
0259             FedRawDataInputSource* parent = nullptr)
0260       : parent_(parent),
0261         status_(status),
0262         lumi_(lumi),
0263         fileName_(name),
0264         deleteFile_(deleteFile),
0265         rawFd_(rawFd),
0266         fileSize_(fileSize),
0267         rawHeaderSize_(rawHeaderSize),
0268         nChunks_(nChunks),
0269         numFiles_(1),
0270         nEvents_(nEvents),
0271         nProcessed_(0) {
0272     fileNames_.push_back(name);
0273     fileOrder_.push_back(fileOrder_.size());
0274     diskFileSizes_.push_back(fileSize);
0275     fileSizes_.push_back(0);
0276     bufferOffsets_.push_back(0);
0277     chunks_.reserve(nChunks_);
0278     for (unsigned int i = 0; i < nChunks; i++)
0279       chunks_.push_back(nullptr);
0280   }
0281   virtual ~InputFile();
0282
0283   void setChunks(uint16_t nChunks) {
0284     nChunks_ = nChunks;
0285     chunks_.clear();
0286     chunks_.reserve(nChunks_);
0287     for (unsigned int i = 0; i < nChunks_; i++)
0288       chunks_.push_back(nullptr);
0289   }
0290
0291   void appendFile(std::string const& name, uint64_t size) {
0292     size_t prevOffset = bufferOffsets_.back();
0293     size_t prevSize = diskFileSizes_.back();
0294     numFiles_++;
0295     fileNames_.push_back(name);
0296     fileOrder_.push_back(fileOrder_.size());
0297     diskFileSizes_.push_back(size);
0298     fileSizes_.push_back(0);
0299     bufferOffsets_.push_back(prevOffset + prevSize);
0300   }
0301
0302   bool waitForChunk(unsigned int chunkid) {
0303     //some atomics to make sure everything is cache synchronized for the main thread
0304     return chunks_[chunkid] != nullptr && chunks_[chunkid]->readComplete_;
0305   }
0306   bool advance(unsigned char*& dataPosition, const size_t size);
0307   void moveToPreviousChunk(const size_t size, const size_t offset);
0308   void rewindChunk(const size_t size);
0309   void unsetDeleteFile() { deleteFile_ = false; }
0310   void randomizeOrder(std::default_random_engine& rng) {
0311     std::shuffle(std::begin(fileOrder_), std::end(fileOrder_), rng);
0312   }
0313   uint64_t currentChunkSize() const { return chunks_[currentChunk_]->size_; }
0314   int64_t fileSizeLeft() const { return (int64_t)fileSize_ - (int64_t)bufferPosition_; }
0315 };
0316
0317 #endif  // EventFilter_Utilities_FedRawDataInputSource_h
0318
0319 /// emacs configuration
0320 /// Local Variables: -
0321 /// mode: c++ -
0322 /// c-basic-offset: 2 -
0323 /// indent-tabs-mode: nil -
0324 /// End: -