Merge pull request cms-sw#43368 from smorovic/master-per-thread-microstate

DAQ - add per-tbb-thread microstate (14_0_X)

Author: cmsbuild

EventFilter/Utilities/interface/FastMonitor.h

@@ -77,13 +77,12 @@ namespace jsoncollector {
     std::string getCSVString(int sid = -1);
 
     //fastpath file output
-    void outputCSV(std::string const& path, std::string const& csvString);
+    void outputCSV(std::string const& path, std::vector<std::string> const& csvString);
 
     //provide merged variable back to user
     JsonMonitorable* getMergedIntJForLumi(std::string const& name, unsigned int forLumi);
 
     // merges and outputs everything collected for the given stream to JSON file
-    bool outputFullJSONs(std::string const& pathstem, std::string const& ext, unsigned int lumi, bool output = true);
     bool outputFullJSON(std::string const& path, unsigned int lumi, bool output = true);
 
     //discard what was collected for a lumisection

EventFilter/Utilities/interface/FastMonitoringService.h

@@ -3,6 +3,7 @@
 
 #include "FWCore/ParameterSet/interface/ParameterSet.h"
 #include "FWCore/ServiceRegistry/interface/ActivityRegistry.h"
+#include "FWCore/ServiceRegistry/interface/StreamContext.h"
 #include "DataFormats/Provenance/interface/EventID.h"
 #include "DataFormats/Provenance/interface/LuminosityBlockID.h"
 #include "DataFormats/Provenance/interface/Timestamp.h"
@@ -20,6 +21,8 @@
 #include <queue>
 #include <sstream>
 #include <unordered_map>
+#include "oneapi/tbb/task_arena.h"
+#include "oneapi/tbb/task_scheduler_observer.h"
 
 /*Description
   this is an evolution of the MicroStateService intended to be run in standalone multi-threaded cmsRun jobs
@@ -28,9 +31,6 @@
   moduledesc pointer to key into the map instead and no string or string pointers are used for the microstates.
   Only a pointer value is stored using relaxed ordering at the time of module execution  which is fast.
   At snapshot time only (every few seconds) we do the map lookup to produce snapshot.
-  Path names use a similar logic. However path names are not accessible in the same way as later so they need to be
-  when starting to run associated to the memory location of path name strings as accessible when path is executed.
-  Path intermediate info will be called "ministate" :D
   The general counters and status variables (event number, number of processed events, number of passed and stored 
   events, luminosity section etc.) are also monitored here.
 
@@ -47,7 +47,10 @@ namespace edm {
 
 namespace evf {
 
+  template <typename T>
+  struct ContainableAtomic;
   class FastMonitoringThread;
+  class ConcurrencyTracker;
 
   namespace FastMonState {
 
@@ -62,7 +65,11 @@ namespace evf {
       mBoL,
       mEoL,
       mGlobEoL,
-      mCOUNT
+      mFwk,
+      mIdleSource,
+      mEvent,
+      mIgnore,
+      mCOUNT,
     };
 
     enum Macrostate {
@@ -153,19 +160,21 @@ namespace evf {
     };
   }  // namespace FastMonState
 
+  constexpr int nSpecialModules = FastMonState::mCOUNT;
+  //reserve output module space
+  constexpr int nReservedModules = 128;
+
   class FastMonitoringService : public MicroStateService {
   public:
     // the names of the states - some of them are never reached in an online app
-    static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT];
+    static const edm::ModuleDescription specialMicroStateNames[FastMonState::mCOUNT];
     static const std::string macroStateNames[FastMonState::MCOUNT];
     static const std::string inputStateNames[FastMonState::inCOUNT];
     // Reserved names for microstates
-    static const std::string nopath_;
     FastMonitoringService(const edm::ParameterSet&, edm::ActivityRegistry&);
     ~FastMonitoringService() override;
     static void fillDescriptions(edm::ConfigurationDescriptions& descriptions);
 
-    std::string makePathLegendaJson();
     std::string makeModuleLegendaJson();
     std::string makeInputLegendaJson();
 
@@ -200,10 +209,6 @@ namespace evf {
     void preSourceEarlyTermination(edm::TerminationOrigin);
     void setExceptionDetected(unsigned int ls);
 
-    //this is still needed for use in special functions like DQM which are in turn framework services
-    void setMicroState(FastMonState::Microstate);
-    void setMicroState(edm::StreamID, FastMonState::Microstate);
-
     void accumulateFileSize(unsigned int lumi, unsigned long fileSize);
     void startedLookingForFile();
     void stoppedLookingForFile(unsigned int lumi);
@@ -223,29 +228,40 @@ namespace evf {
     void setInputSource(DAQSource* inputSource) { daqInputSource_ = inputSource; }
     void setInState(FastMonState::InputState inputState) { inputState_ = inputState; }
     void setInStateSup(FastMonState::InputState inputState) { inputSupervisorState_ = inputState; }
+    //available for other modules
+    void setTMicrostate(FastMonState::Microstate m);
+
+    static unsigned int getTID() { return tbb::this_task_arena::current_thread_index(); }
 
   private:
     void doSnapshot(const unsigned int ls, const bool isGlobalEOL);
 
     void snapshotRunner();
 
+    static unsigned int getSID(edm::StreamContext const& sc) { return sc.streamID().value(); }
+
+    static unsigned int getSID(edm::StreamID const& sid) { return sid.value(); }
+
     //the actual monitoring thread is held by a separate class object for ease of maintenance
-    std::shared_ptr<FastMonitoringThread> fmt_;
+    std::unique_ptr<FastMonitoringThread> fmt_;
+    std::unique_ptr<ConcurrencyTracker> ct_;
     //Encoding encModule_;
     //std::vector<Encoding> encPath_;
     FedRawDataInputSource* inputSource_ = nullptr;
     DAQSource* daqInputSource_ = nullptr;
     std::atomic<FastMonState::InputState> inputState_{FastMonState::InputState::inInit};
     std::atomic<FastMonState::InputState> inputSupervisorState_{FastMonState::InputState::inInit};
 
-    unsigned int nStreams_;
-    unsigned int nThreads_;
+    unsigned int nStreams_ = 0;
+    unsigned int nMonThreads_ = 0;
+    unsigned int nThreads_ = 0;
+    bool tbbMonitoringMode_;
+    bool tbbConcurrencyTracker_;
     int sleepTime_;
     unsigned int fastMonIntervals_;
     unsigned int snapCounter_ = 0;
     std::string microstateDefPath_, fastMicrostateDefPath_;
-    std::string fastName_, fastPath_, slowName_;
-    bool filePerFwkStream_;
+    std::string fastName_, fastPath_;
 
     //variables that are used by/monitored by FastMonitoringThread / FastMonitor
 
@@ -272,9 +288,6 @@ namespace evf {
     //to disable this behavior, set #ATOMIC_LEVEL 0 or 1 in DataPoint.h
     std::vector<std::atomic<bool>*> streamCounterUpdating_;
 
-    std::vector<std::atomic<bool>*> collectedPathList_;
-    std::vector<bool> pathNamesReady_;
-
     std::filesystem::path workingDirectory_, runDirectory_;
 
     bool threadIDAvailable_ = false;
@@ -283,8 +296,6 @@ namespace evf {
 
     std::string moduleLegendFile_;
     std::string moduleLegendFileJson_;
-    std::string pathLegendFile_;
-    std::string pathLegendFileJson_;
     std::string inputLegendFileJson_;
     unsigned int nOutputModules_ = 0;
 
@@ -293,7 +304,13 @@ namespace evf {
     std::atomic<bool> has_source_exception_ = false;
     std::atomic<bool> has_data_exception_ = false;
     std::vector<unsigned int> exceptionInLS_;
-    std::vector<std::string> fastPathList_;
+
+    //per stream
+    std::vector<ContainableAtomic<const void*>> microstate_;
+    std::vector<ContainableAtomic<unsigned char>> microstateAcqFlag_;
+    //per thread
+    std::vector<ContainableAtomic<const void*>> tmicrostate_;
+    std::vector<ContainableAtomic<unsigned char>> tmicrostateAcqFlag_;
 
     bool verbose_ = false;
   };

EventFilter/Utilities/interface/FastMonitoringThread.h

@@ -13,13 +13,9 @@
 
 namespace evf {
 
-  constexpr int nReservedModules = 128;
-  constexpr int nSpecialModules = 10;
-  constexpr int nReservedPaths = 1;
-
-  namespace FastMonState {
-    enum Macrostate;
-  }
+  //namespace FastMonState {
+  //  enum Macrostate;
+  //}
 
   class FastMonitoringService;
 
@@ -131,31 +127,23 @@ namespace evf {
       unsigned int varIndexThrougput_;
 
       //per stream
+      std::vector<unsigned int> tmicrostateEncoded_;
       std::vector<unsigned int> microstateEncoded_;
-      std::vector<unsigned int> ministateEncoded_;
       std::vector<jsoncollector::AtomicMonUInt*> processed_;
       jsoncollector::IntJ fastPathProcessedJ_;
-      std::vector<unsigned int> threadMicrostateEncoded_;
       std::vector<unsigned int> inputState_;
 
       //tracking luminosity of a stream
       std::vector<unsigned int> streamLumi_;
 
       //N bins for histograms
       unsigned int macrostateBins_;
-      unsigned int ministateBins_;
       unsigned int microstateBins_;
       unsigned int inputstateBins_;
 
       //global state
       std::atomic<FastMonState::Macrostate> macrostate_;
 
-      //per stream
-      std::vector<ContainableAtomic<const std::string*>> ministate_;
-      std::vector<ContainableAtomic<const void*>> microstate_;
-      std::vector<ContainableAtomic<unsigned char>> microstateAcqFlag_;
-      std::vector<ContainableAtomic<const void*>> threadMicrostate_;
-
       FastMonEncoding encModule_;
       std::vector<FastMonEncoding> encPath_;
 
@@ -180,7 +168,10 @@ namespace evf {
       }
 
       //to be called after fast monitor is constructed
-      void registerVariables(jsoncollector::FastMonitor* fm, unsigned int nStreams, unsigned int nThreads) {
+      void registerVariables(jsoncollector::FastMonitor* fm,
+                             unsigned nMaxSlices,
+                             unsigned nMaxStreams,
+                             unsigned nMaxThreads) {
         //tell FM to track these global variables(for fast and slow monitoring)
         fm->registerGlobalMonitorable(&fastMacrostateJ_, true, &macrostateBins_);
         fm->registerGlobalMonitorable(&fastThroughputJ_, false);
@@ -189,27 +180,30 @@ namespace evf {
         fm->registerGlobalMonitorable(&fastLockWaitJ_, false);
         fm->registerGlobalMonitorable(&fastLockCountJ_, false);
 
-        for (unsigned int i = 0; i < nStreams; i++) {
+        for (unsigned int i = 0; i < nMaxSlices; i++) {
           jsoncollector::AtomicMonUInt* p = new jsoncollector::AtomicMonUInt;
           *p = 0;
           processed_.push_back(p);
           streamLumi_.push_back(0);
         }
 
-        microstateEncoded_.resize(nStreams);
-        ministateEncoded_.resize(nStreams);
-        threadMicrostateEncoded_.resize(nThreads);
-        inputState_.resize(nStreams);
-        for (unsigned int j = 0; j < inputState_.size(); j++)
-          inputState_[j] = 0;
+        tmicrostateEncoded_.resize(nMaxSlices, FastMonState::mInvalid);
+        for (unsigned int i = nMaxThreads; i < nMaxSlices; i++) {
+          tmicrostateEncoded_[i] = FastMonState::mIgnore;
+        }
+        microstateEncoded_.resize(nMaxSlices, FastMonState::mInvalid);
+        inputState_.resize(nMaxSlices, FastMonState::inInit);
+        for (unsigned int i = nMaxStreams; i < nMaxSlices; i++) {
+          microstateEncoded_[i] = FastMonState::mIgnore;
+          inputState_[i] = FastMonState::inIgnore;
+        }
+        //for (unsigned int j = 0; j < nMaxStreams; j++)
+        //  inputState_[j] = 0;
 
         //tell FM to track these int vectors
-        fm->registerStreamMonitorableUIntVec("Ministate", &ministateEncoded_, true, &ministateBins_);
+        fm->registerStreamMonitorableUIntVec("tMicrostate", &tmicrostateEncoded_, true, &microstateBins_);
 
-        if (nThreads <= nStreams)  //no overlapping in module execution per stream
-          fm->registerStreamMonitorableUIntVec("Microstate", &microstateEncoded_, true, &microstateBins_);
-        else
-          fm->registerStreamMonitorableUIntVec("Microstate", &threadMicrostateEncoded_, true, &microstateBins_);
+        fm->registerStreamMonitorableUIntVec("Microstate", &microstateEncoded_, true, &microstateBins_);
 
         fm->registerStreamMonitorableUIntVecAtomic("Processed", &processed_, false, nullptr);
 

EventFilter/Utilities/interface/SourceCommon.h

@@ -0,0 +1,25 @@
+#ifndef EventFilter_Utilities_SourceCommon_h
+#define EventFilter_Utilities_SourceCommon_h
+
+/*
+ * This header will host common definitions used by FedRawDataInputSource and DAQSource
+ * */
+
+#include "EventFilter/Utilities/interface/FastMonitoringService.h"
+
+class IdleSourceSentry {
+public:
+  IdleSourceSentry(evf::FastMonitoringService* fms) : fms_(fms) {
+    if (fms_)
+      fms_->setTMicrostate(evf::FastMonState::mIdleSource);
+  }
+  ~IdleSourceSentry() {
+    if (fms_)
+      fms_->setTMicrostate(evf::FastMonState::mIdle);
+  }
+
+private:
+  evf::FastMonitoringService* fms_;
+};
+
+#endif

EventFilter/Utilities/plugins/microstatedef.jsd

@@ -5,7 +5,7 @@
          "operation" : "histo"
       },
       {
-         "name" : "Ministate",
+         "name" : "tMicrostate",
          "operation" : "histo"
       },
       {

EventFilter/Utilities/plugins/microstatedeffast.jsd

@@ -5,7 +5,7 @@
          "operation" : "histo"
       },
       {
-         "name" : "Ministate",
+         "name" : "tMicrostate",
          "operation" : "histo"
       },
       {

EventFilter/Utilities/src/DAQSource.cc

@@ -19,7 +19,7 @@
 #include "DataFormats/Provenance/interface/EventID.h"
 #include "DataFormats/Provenance/interface/Timestamp.h"
 
-#include "EventFilter/Utilities/interface/FastMonitoringService.h"
+#include "EventFilter/Utilities/interface/SourceCommon.h"
 #include "EventFilter/Utilities/interface/DataPointDefinition.h"
 #include "EventFilter/Utilities/interface/FFFNamingSchema.h"
 #include "EventFilter/Utilities/interface/crc32c.h"
@@ -373,11 +373,14 @@ evf::EvFDaqDirector::FileStatus DAQSource::getNextDataBlock() {
   if (!currentFile_.get()) {
     evf::EvFDaqDirector::FileStatus status = evf::EvFDaqDirector::noFile;
     setMonState(inWaitInput);
-    if (!fileQueue_.try_pop(currentFile_)) {
-      //sleep until wakeup (only in single-buffer mode) or timeout
-      std::unique_lock<std::mutex> lkw(mWakeup_);
-      if (cvWakeup_.wait_for(lkw, std::chrono::milliseconds(100)) == std::cv_status::timeout || !currentFile_.get())
-        return evf::EvFDaqDirector::noFile;
+    {
+      IdleSourceSentry ids(fms_);
+      if (!fileQueue_.try_pop(currentFile_)) {
+        //sleep until wakeup (only in single-buffer mode) or timeout
+        std::unique_lock<std::mutex> lkw(mWakeup_);
+        if (cvWakeup_.wait_for(lkw, std::chrono::milliseconds(100)) == std::cv_status::timeout || !currentFile_.get())
+          return evf::EvFDaqDirector::noFile;
+      }
     }
     status = currentFile_->status_;
     if (status == evf::EvFDaqDirector::runEnded) {
@@ -468,10 +471,13 @@ evf::EvFDaqDirector::FileStatus DAQSource::getNextDataBlock() {
   //multibuffer mode
   //wait for the current chunk to become added to the vector
   setMonState(inWaitChunk);
-  while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
-    usleep(10000);
-    if (setExceptionState_)
-      threadError();
+  {
+    IdleSourceSentry ids(fms_);
+    while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
+      usleep(10000);
+      if (setExceptionState_)
+        threadError();
+    }
   }
   setMonState(inChunkReceived);
 
@@ -508,13 +514,14 @@ evf::EvFDaqDirector::FileStatus DAQSource::getNextDataBlock() {
     currentFile_->rewindChunk(dataMode_->headerSize());
 
     setMonState(inWaitChunk);
-
-    //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);
-    assert(chunkEnd);
-    //mark to release old chunk
-    chunkIsFree_ = true;
-
+    {
+      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);
+      assert(chunkEnd);
+      //mark to release old chunk
+      chunkIsFree_ = true;
+    }
     setMonState(inChunkReceived);
     //header and payload is moved, update view
     dataMode_->makeDataBlockView(