CMS 3D CMS Logo

List of all members | Public Member Functions | Static Public Member Functions | Static Public Attributes | Private Member Functions | Static Private Member Functions | Private Attributes
evf::FastMonitoringService Class Reference

#include <FastMonitoringService.h>

Inheritance diagram for evf::FastMonitoringService:
evf::MicroStateService

Public Member Functions

void accumulateFileSize (unsigned int lumi, unsigned long fileSize)
 
bool exceptionDetected () const
 
 FastMonitoringService (const edm::ParameterSet &, edm::ActivityRegistry &)
 
bool getAbortFlagForLumi (unsigned int lumi)
 
unsigned int getEventsProcessedForLumi (unsigned int lumi, bool *abortFlag=nullptr)
 
std::string getRunDirName () const
 
bool isExceptionOnData (unsigned int ls)
 
void jobFailure ()
 
std::string makeInputLegendaJson ()
 
std::string makeModuleLegendaJson ()
 
void postBeginJob ()
 
void postEndJob ()
 
void postEvent (edm::StreamContext const &)
 
void postGlobalBeginRun (edm::GlobalContext const &)
 
void postGlobalEndLumi (edm::GlobalContext const &)
 
void postModuleEvent (edm::StreamContext const &, edm::ModuleCallingContext const &)
 
void postModuleEventAcquire (edm::StreamContext const &, edm::ModuleCallingContext const &)
 
void postSourceEvent (edm::StreamID)
 
void postStreamBeginLumi (edm::StreamContext const &)
 
void postStreamEndLumi (edm::StreamContext const &)
 
void preallocate (edm::service::SystemBounds const &)
 
void preBeginJob (edm::PathsAndConsumesOfModulesBase const &, edm::ProcessContext const &pc)
 
void preEvent (edm::StreamContext const &)
 
void preGlobalBeginLumi (edm::GlobalContext const &)
 
void preGlobalEarlyTermination (edm::GlobalContext const &, edm::TerminationOrigin)
 
void preGlobalEndLumi (edm::GlobalContext const &)
 
void preModuleBeginJob (edm::ModuleDescription const &)
 
void preModuleEvent (edm::StreamContext const &, edm::ModuleCallingContext const &)
 
void preModuleEventAcquire (edm::StreamContext const &, edm::ModuleCallingContext const &)
 
void prePathEvent (edm::StreamContext const &, edm::PathContext const &)
 
void preSourceEarlyTermination (edm::TerminationOrigin)
 
void preSourceEvent (edm::StreamID)
 
void preStreamBeginLumi (edm::StreamContext const &)
 
void preStreamEarlyTermination (edm::StreamContext const &, edm::TerminationOrigin)
 
void preStreamEndLumi (edm::StreamContext const &)
 
void reportLockWait (unsigned int ls, double waitTime, unsigned int lockCount)
 
void setExceptionDetected (unsigned int ls)
 
void setInputSource (FedRawDataInputSource *inputSource)
 
void setInputSource (DAQSource *inputSource)
 
void setInState (FastMonState::InputState inputState)
 
void setInStateSup (FastMonState::InputState inputState)
 
void setTMicrostate (FastMonState::Microstate m)
 
bool shouldWriteFiles (unsigned int lumi, unsigned int *proc=nullptr)
 
void startedLookingForFile ()
 
void stoppedLookingForFile (unsigned int lumi)
 
 ~FastMonitoringService () override
 
- Public Member Functions inherited from evf::MicroStateService
 MicroStateService (const edm::ParameterSet &, edm::ActivityRegistry &)
 
virtual ~MicroStateService ()
 

Static Public Member Functions

static void fillDescriptions (edm::ConfigurationDescriptions &descriptions)
 
static unsigned int getTID ()
 

Static Public Attributes

static const std::string inputStateNames [FastMonState::inCOUNT]
 
static const std::string macroStateNames [FastMonState::MCOUNT]
 
static const edm::ModuleDescription specialMicroStateNames [FastMonState::mCOUNT]
 

Private Member Functions

void doSnapshot (const unsigned int ls, const bool isGlobalEOL)
 
void snapshotRunner ()
 

Static Private Member Functions

static unsigned int getSID (edm::StreamContext const &sc)
 
static unsigned int getSID (edm::StreamID const &sid)
 

Private Attributes

std::map< unsigned int, unsigned long > accuSize_
 
std::map< unsigned int, double > avgLeadTime_
 
std::unique_ptr< ConcurrencyTrackerct_
 
DAQSourcedaqInputSource_ = nullptr
 
bool exception_detected_ = false
 
std::vector< unsigned int > exceptionInLS_
 
std::string fastMicrostateDefPath_
 
unsigned int fastMonIntervals_
 
std::string fastName_
 
std::string fastPath_
 
timeval fileLookStart_
 
timeval fileLookStop_
 
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
 
std::unique_ptr< FastMonitoringThreadfmt_
 
std::atomic< bool > has_data_exception_ = false
 
std::atomic< bool > has_source_exception_ = false
 
std::string inputLegendFileJson_
 
FedRawDataInputSourceinputSource_ = nullptr
 
std::atomic< FastMonState::InputStateinputState_ {FastMonState::InputState::inInit}
 
std::atomic< FastMonState::InputStateinputSupervisorState_ {FastMonState::InputState::inInit}
 
std::atomic< bool > isInitTransition_
 
unsigned int lastGlobalLumi_
 
std::vector< double > leadTimes_
 
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_
 
unsigned int lumiFromSource_
 
std::map< unsigned int, timeval > lumiStartTime_
 
std::vector< ContainableAtomic< const void * > > microstate_
 
std::vector< ContainableAtomic< unsigned char > > microstateAcqFlag_
 
std::string microstateDefPath_
 
std::string moduleLegendFile_
 
std::string moduleLegendFileJson_
 
std::atomic< bool > monInit_
 
unsigned int nMonThreads_ = 0
 
unsigned int nOutputModules_ = 0
 
unsigned int nStreams_ = 0
 
unsigned int nThreads_ = 0
 
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
 
std::filesystem::path runDirectory_
 
int sleepTime_
 
unsigned int snapCounter_ = 0
 
std::vector< std::atomic< bool > * > streamCounterUpdating_
 
bool tbbConcurrencyTracker_
 
bool tbbMonitoringMode_
 
bool threadIDAvailable_ = false
 
std::vector< ContainableAtomic< const void * > > tmicrostate_
 
std::vector< ContainableAtomic< unsigned char > > tmicrostateAcqFlag_
 
std::atomic< unsigned long > totalEventsProcessed_
 
bool verbose_ = false
 
std::filesystem::path workingDirectory_
 

Detailed Description

Definition at line 167 of file FastMonitoringService.h.

Constructor & Destructor Documentation

◆ FastMonitoringService()

evf::FastMonitoringService::FastMonitoringService ( const edm::ParameterSet iPS,
edm::ActivityRegistry reg 
)

Definition at line 197 of file FastMonitoringService.cc.

References Exception, fastMicrostateDefPath_, jobFailure(), microstateDefPath_, postBeginJob(), postEndJob(), postEvent(), postGlobalEndLumi(), postModuleEvent(), postModuleEventAcquire(), postSourceEvent(), postStreamBeginLumi(), postStreamEndLumi(), preallocate(), preBeginJob(), preEvent(), preGlobalBeginLumi(), preGlobalEarlyTermination(), preGlobalEndLumi(), preModuleBeginJob(), preModuleEvent(), preModuleEventAcquire(), preSourceEarlyTermination(), preSourceEvent(), preStreamBeginLumi(), preStreamEarlyTermination(), preStreamEndLumi(), edm_modernize_messagelogger::stat, AlCaHLTBitMon_QueryRunRegistry::string, edm::ActivityRegistry::watchJobFailure(), edm::ActivityRegistry::watchPostBeginJob(), edm::ActivityRegistry::watchPostEndJob(), edm::ActivityRegistry::watchPostEvent(), edm::ActivityRegistry::watchPostGlobalEndLumi(), edm::ActivityRegistry::watchPostModuleEvent(), edm::ActivityRegistry::watchPostModuleEventAcquire(), edm::ActivityRegistry::watchPostSourceEvent(), edm::ActivityRegistry::watchPostStreamBeginLumi(), edm::ActivityRegistry::watchPostStreamEndLumi(), edm::ActivityRegistry::watchPreallocate(), edm::ActivityRegistry::watchPreBeginJob(), edm::ActivityRegistry::watchPreEvent(), edm::ActivityRegistry::watchPreGlobalBeginLumi(), edm::ActivityRegistry::watchPreGlobalEarlyTermination(), edm::ActivityRegistry::watchPreGlobalEndLumi(), edm::ActivityRegistry::watchPreModuleBeginJob(), edm::ActivityRegistry::watchPreModuleEvent(), edm::ActivityRegistry::watchPreModuleEventAcquire(), edm::ActivityRegistry::watchPreSourceEarlyTermination(), edm::ActivityRegistry::watchPreSourceEvent(), edm::ActivityRegistry::watchPreStreamBeginLumi(), edm::ActivityRegistry::watchPreStreamEarlyTermination(), and edm::ActivityRegistry::watchPreStreamEndLumi().

198  : MicroStateService(iPS, reg),
199  fmt_(new FastMonitoringThread()),
200  tbbMonitoringMode_(iPS.getUntrackedParameter<bool>("tbbMonitoringMode", true)),
201  tbbConcurrencyTracker_(iPS.getUntrackedParameter<bool>("tbbConcurrencyTracker", true) && tbbMonitoringMode_),
202  sleepTime_(iPS.getUntrackedParameter<int>("sleepTime", 1)),
203  fastMonIntervals_(iPS.getUntrackedParameter<unsigned int>("fastMonIntervals", 2)),
204  fastName_("fastmoni"),
206  verbose_(iPS.getUntrackedParameter<bool>("verbose")) {
207  reg.watchPreallocate(this, &FastMonitoringService::preallocate); //receiving information on number of threads
209 
214 
218 
223 
224  reg.watchPreEvent(this, &FastMonitoringService::preEvent); //stream
226 
227  //readEvent (not getNextItemType)
228  reg.watchPreSourceEvent(this, &FastMonitoringService::preSourceEvent); //source (with streamID of requestor)
230 
233 
236 
240 
241  //find microstate definition path (required by the module)
242  struct stat statbuf;
243  std::string microstateBaseSuffix = "src/EventFilter/Utilities/plugins/microstatedef.jsd";
244  std::string microstatePath = std::string(std::getenv("CMSSW_BASE")) + "/" + microstateBaseSuffix;
245  if (stat(microstatePath.c_str(), &statbuf)) {
246  microstatePath = std::string(std::getenv("CMSSW_RELEASE_BASE")) + "/" + microstateBaseSuffix;
247  if (stat(microstatePath.c_str(), &statbuf)) {
248  microstatePath = microstateBaseSuffix;
249  if (stat(microstatePath.c_str(), &statbuf))
250  throw cms::Exception("FastMonitoringService") << "microstate definition file not found";
251  }
252  }
253  fastMicrostateDefPath_ = microstateDefPath_ = microstatePath;
254  }
void watchPreStreamEarlyTermination(PreStreamEarlyTermination::slot_type const &iSlot)
void watchPreEvent(PreEvent::slot_type const &iSlot)
void postModuleEventAcquire(edm::StreamContext const &, edm::ModuleCallingContext const &)
void watchPreallocate(Preallocate::slot_type const &iSlot)
void watchPreModuleEventAcquire(PreModuleEventAcquire::slot_type const &iSlot)
void watchPostEndJob(PostEndJob::slot_type const &iSlot)
void preallocate(edm::service::SystemBounds const &)
void preGlobalBeginLumi(edm::GlobalContext const &)
void watchPreModuleEvent(PreModuleEvent::slot_type const &iSlot)
void postGlobalEndLumi(edm::GlobalContext const &)
void postEvent(edm::StreamContext const &)
void watchPreGlobalEndLumi(PreGlobalEndLumi::slot_type const &iSlot)
void watchPostEvent(PostEvent::slot_type const &iSlot)
void watchPostStreamEndLumi(PostStreamEndLumi::slot_type const &iSlot)
void watchPreGlobalBeginLumi(PreGlobalBeginLumi::slot_type const &iSlot)
void watchPostModuleEvent(PostModuleEvent::slot_type const &iSlot)
void watchPostSourceEvent(PostSourceEvent::slot_type const &iSlot)
void preGlobalEndLumi(edm::GlobalContext const &)
void watchPreStreamEndLumi(PreStreamEndLumi::slot_type const &iSlot)
std::unique_ptr< FastMonitoringThread > fmt_
void preGlobalEarlyTermination(edm::GlobalContext const &, edm::TerminationOrigin)
void watchPreSourceEarlyTermination(PreSourceEarlyTermination::slot_type const &iSlot)
void watchJobFailure(JobFailure::slot_type const &iSlot)
convenience function for attaching to signal
T getUntrackedParameter(std::string const &, T const &) const
void preModuleBeginJob(edm::ModuleDescription const &)
MicroStateService(const edm::ParameterSet &, edm::ActivityRegistry &)
void preStreamEndLumi(edm::StreamContext const &)
void watchPostStreamBeginLumi(PostStreamBeginLumi::slot_type const &iSlot)
void preStreamEarlyTermination(edm::StreamContext const &, edm::TerminationOrigin)
void watchPreGlobalEarlyTermination(PreGlobalEarlyTermination::slot_type const &iSlot)
void watchPostModuleEventAcquire(PostModuleEventAcquire::slot_type const &iSlot)
void watchPostGlobalEndLumi(PostGlobalEndLumi::slot_type const &iSlot)
void watchPreModuleBeginJob(PreModuleBeginJob::slot_type const &iSlot)
void preModuleEventAcquire(edm::StreamContext const &, edm::ModuleCallingContext const &)
void postModuleEvent(edm::StreamContext const &, edm::ModuleCallingContext const &)
void postStreamBeginLumi(edm::StreamContext const &)
void postStreamEndLumi(edm::StreamContext const &)
void preStreamBeginLumi(edm::StreamContext const &)
std::atomic< unsigned long > totalEventsProcessed_
void watchPreStreamBeginLumi(PreStreamBeginLumi::slot_type const &iSlot)
void preBeginJob(edm::PathsAndConsumesOfModulesBase const &, edm::ProcessContext const &pc)
void watchPreBeginJob(PreBeginJob::slot_type const &iSlot)
convenience function for attaching to signal
void preEvent(edm::StreamContext const &)
void preSourceEarlyTermination(edm::TerminationOrigin)
void watchPreSourceEvent(PreSourceEvent::slot_type const &iSlot)
void preModuleEvent(edm::StreamContext const &, edm::ModuleCallingContext const &)
void watchPostBeginJob(PostBeginJob::slot_type const &iSlot)
convenience function for attaching to signal

◆ ~FastMonitoringService()

evf::FastMonitoringService::~FastMonitoringService ( )
override

Definition at line 256 of file FastMonitoringService.cc.

256 {}

Member Function Documentation

◆ accumulateFileSize()

void evf::FastMonitoringService::accumulateFileSize ( unsigned int  lumi,
unsigned long  fileSize 
)

Definition at line 700 of file FastMonitoringService.cc.

References accuSize_, filesProcessedDuringLumi_, fmt_, CommonMethods::lock(), and BXlumiParameters_cfi::lumi.

Referenced by evf::EvFDaqDirector::bumpFile().

700  {
701  std::lock_guard<std::mutex> lock(fmt_->monlock_);
702 
703  if (accuSize_.find(lumi) == accuSize_.end())
704  accuSize_[lumi] = fileSize;
705  else
706  accuSize_[lumi] += fileSize;
707 
710  else
712  }
std::map< unsigned int, unsigned long > accuSize_
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
std::unique_ptr< FastMonitoringThread > fmt_

◆ doSnapshot()

void evf::FastMonitoringService::doSnapshot ( const unsigned int  ls,
const bool  isGlobalEOL 
)
private

Definition at line 842 of file FastMonitoringService.cc.

References avgLeadTime_, ct_, filesProcessedDuringLumi_, fmt_, evf::getmEoL(), evf::getmFwk(), evf::getmFwkEoL(), evf::getmIdle(), mps_fire::i, caHitNtupletGeneratorKernels::if(), evf::FastMonState::inIgnore, evf::FastMonState::inNewLumi, evf::FastMonState::inNoRequest, evf::FastMonState::inNoRequestWithEoLThreads, evf::FastMonState::inNoRequestWithIdleThreads, inputState_, inputSupervisorState_, evf::FastMonState::inRunEnd, evf::FastMonState::inSupBusy, evf::FastMonState::inSupFileLimit, evf::FastMonState::inSupLockPolling, evf::FastMonState::inSupLockPollingCopying, evf::FastMonState::inSupNewFile, evf::FastMonState::inSupNewFileWaitChunk, evf::FastMonState::inSupNewFileWaitChunkCopying, evf::FastMonState::inSupNewFileWaitThread, evf::FastMonState::inSupNewFileWaitThreadCopying, evf::FastMonState::inSupNoFile, evf::FastMonState::inSupThrottled, evf::FastMonState::inSupWaitFreeChunk, evf::FastMonState::inSupWaitFreeChunkCopying, evf::FastMonState::inSupWaitFreeThread, evf::FastMonState::inSupWaitFreeThreadCopying, evf::FastMonState::inWaitChunk, evf::FastMonState::inWaitChunk_busy, evf::FastMonState::inWaitChunk_fileLimit, evf::FastMonState::inWaitChunk_lockPolling, evf::FastMonState::inWaitChunk_lockPollingCopying, evf::FastMonState::inWaitChunk_newFile, evf::FastMonState::inWaitChunk_newFileWaitChunk, evf::FastMonState::inWaitChunk_newFileWaitChunkCopying, evf::FastMonState::inWaitChunk_newFileWaitThread, evf::FastMonState::inWaitChunk_newFileWaitThreadCopying, evf::FastMonState::inWaitChunk_noFile, evf::FastMonState::inWaitChunk_runEnd, evf::FastMonState::inWaitChunk_waitFreeChunk, evf::FastMonState::inWaitChunk_waitFreeChunkCopying, evf::FastMonState::inWaitChunk_waitFreeThread, evf::FastMonState::inWaitChunk_waitFreeThreadCopying, evf::FastMonState::inWaitInput, evf::FastMonState::inWaitInput_busy, evf::FastMonState::inWaitInput_fileLimit, evf::FastMonState::inWaitInput_lockPolling, evf::FastMonState::inWaitInput_lockPollingCopying, evf::FastMonState::inWaitInput_newFile, evf::FastMonState::inWaitInput_newFileWaitChunk, evf::FastMonState::inWaitInput_newFileWaitChunkCopying, evf::FastMonState::inWaitInput_newFileWaitThread, evf::FastMonState::inWaitInput_newFileWaitThreadCopying, evf::FastMonState::inWaitInput_noFile, evf::FastMonState::inWaitInput_runEnd, evf::FastMonState::inWaitInput_waitFreeChunk, evf::FastMonState::inWaitInput_waitFreeChunkCopying, evf::FastMonState::inWaitInput_waitFreeThread, evf::FastMonState::inWaitInput_waitFreeThreadCopying, isInitTransition_, lockStatsDuringLumi_, eostools::ls(), microstate_, microstateAcqFlag_, nMonThreads_, nStreams_, nThreads_, tmicrostate_, and tmicrostateAcqFlag_.

Referenced by preGlobalEndLumi(), and snapshotRunner().

842  {
843  // update macrostate
844  fmt_->m_data.fastMacrostateJ_ = fmt_->m_data.macrostate_;
845 
846  std::vector<const void*> microstateCopy(microstate_.begin(), microstate_.end());
847  std::vector<const void*> tmicrostateCopy(tmicrostate_.begin(), tmicrostate_.end());
848  std::vector<unsigned char> microstateAcqCopy(microstateAcqFlag_.begin(), microstateAcqFlag_.end());
849  std::vector<unsigned char> tmicrostateAcqCopy(tmicrostateAcqFlag_.begin(), tmicrostateAcqFlag_.end());
850 
851  if (!isInitTransition_) {
852  auto itd = avgLeadTime_.find(ls);
853  if (itd != avgLeadTime_.end())
854  fmt_->m_data.fastAvgLeadTimeJ_ = itd->second;
855  else
856  fmt_->m_data.fastAvgLeadTimeJ_ = 0.;
857 
858  auto iti = filesProcessedDuringLumi_.find(ls);
859  if (iti != filesProcessedDuringLumi_.end())
860  fmt_->m_data.fastFilesProcessedJ_ = iti->second;
861  else
862  fmt_->m_data.fastFilesProcessedJ_ = 0;
863 
864  auto itrd = lockStatsDuringLumi_.find(ls);
865  if (itrd != lockStatsDuringLumi_.end()) {
866  fmt_->m_data.fastLockWaitJ_ = itrd->second.first;
867  fmt_->m_data.fastLockCountJ_ = itrd->second.second;
868  } else {
869  fmt_->m_data.fastLockWaitJ_ = 0.;
870  fmt_->m_data.fastLockCountJ_ = 0.;
871  }
872  }
873 
874  for (unsigned int i = 0; i < nThreads_; i++) {
875  if (tmicrostateCopy[i] == getmIdle() && ct_->isThreadActive(i)) {
876  //overhead if thread is running
877  tmicrostateCopy[i] = getmFwk();
878  }
879  if (tmicrostateAcqCopy[i])
880  fmt_->m_data.tmicrostateEncoded_[i] =
881  fmt_->m_data.microstateBins_ + fmt_->m_data.encModule_.encode(tmicrostateCopy[i]);
882  else
883  fmt_->m_data.tmicrostateEncoded_[i] = fmt_->m_data.encModule_.encode(tmicrostateCopy[i]);
884  }
885 
886  for (unsigned int i = 0; i < nStreams_; i++) {
887  if (microstateAcqCopy[i])
888  fmt_->m_data.microstateEncoded_[i] =
889  fmt_->m_data.microstateBins_ + fmt_->m_data.encModule_.encode(microstateCopy[i]);
890  else
891  fmt_->m_data.microstateEncoded_[i] = fmt_->m_data.encModule_.encode(microstateCopy[i]);
892  }
893 
894  bool inputStatePerThread = false;
895 
897  switch (inputSupervisorState_) {
899  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_fileLimit;
900  break;
902  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeChunk;
903  break;
905  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeChunkCopying;
906  break;
908  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeThread;
909  break;
911  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeThreadCopying;
912  break;
914  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_busy;
915  break;
917  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_lockPolling;
918  break;
920  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_lockPollingCopying;
921  break;
923  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_runEnd;
924  break;
926  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_noFile;
927  break;
929  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFile;
930  break;
933  break;
935  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFileWaitThread;
936  break;
939  break;
941  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFileWaitChunk;
942  break;
943  default:
944  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput;
945  }
946  } else if (inputState_ == FastMonState::inWaitChunk) {
947  switch (inputSupervisorState_) {
949  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_fileLimit;
950  break;
952  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeChunk;
953  break;
955  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeChunkCopying;
956  break;
958  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeThread;
959  break;
961  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeThreadCopying;
962  break;
964  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_busy;
965  break;
967  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_lockPolling;
968  break;
970  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_lockPollingCopying;
971  break;
973  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_runEnd;
974  break;
976  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_noFile;
977  break;
979  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFile;
980  break;
983  break;
985  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFileWaitThread;
986  break;
989  break;
991  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFileWaitChunk;
992  break;
993  default:
994  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk;
995  }
996  } else if (inputState_ == FastMonState::inNoRequest) {
997  inputStatePerThread = true;
998  for (unsigned int i = 0; i < nMonThreads_; i++) {
999  if (i >= nStreams_)
1000  fmt_->m_data.inputState_[i] = FastMonState::inIgnore;
1001  else if (microstateCopy[i] == getmIdle())
1002  fmt_->m_data.inputState_[i] = FastMonState::inNoRequestWithIdleThreads;
1003  else if (microstateCopy[i] == getmEoL() || microstateCopy[i] == getmFwkEoL())
1004  fmt_->m_data.inputState_[i] = FastMonState::inNoRequestWithEoLThreads;
1005  else
1006  fmt_->m_data.inputState_[i] = FastMonState::inNoRequest;
1007  }
1008  } else if (inputState_ == FastMonState::inNewLumi) {
1009  inputStatePerThread = true;
1010  for (unsigned int i = 0; i < nMonThreads_; i++) {
1011  if (i >= nStreams_)
1012  fmt_->m_data.inputState_[i] = FastMonState::inIgnore;
1013  else if (microstateCopy[i] == getmEoL() || microstateCopy[i] == getmFwkEoL())
1014  fmt_->m_data.inputState_[i] = FastMonState::inNewLumi;
1015  }
1017  //apply directly throttled state from supervisor
1018  fmt_->m_data.inputState_[0] = inputSupervisorState_;
1019  } else
1020  fmt_->m_data.inputState_[0] = inputState_;
1021 
1022  //this is same for all streams
1023  if (!inputStatePerThread)
1024  for (unsigned int i = 1; i < nMonThreads_; i++)
1025  fmt_->m_data.inputState_[i] = fmt_->m_data.inputState_[0];
1026 
1027  if (isGlobalEOL) { //only update global variables
1028  fmt_->jsonMonitor_->snapGlobal(ls);
1029  } else
1030  fmt_->jsonMonitor_->snap(ls);
1031  }
constexpr edm::ModuleDescription const * getmFwkEoL()
std::atomic< FastMonState::InputState > inputState_
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmIdle()
std::atomic< bool > isInitTransition_
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
std::unique_ptr< FastMonitoringThread > fmt_
constexpr edm::ModuleDescription const * getmFwk()
constexpr edm::ModuleDescription const * getmEoL()
std::vector< ContainableAtomic< unsigned char > > tmicrostateAcqFlag_
std::map< unsigned int, double > avgLeadTime_
std::atomic< FastMonState::InputState > inputSupervisorState_
std::vector< ContainableAtomic< unsigned char > > microstateAcqFlag_
std::unique_ptr< ConcurrencyTracker > ct_
def ls(path, rec=False)
Definition: eostools.py:349
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ exceptionDetected()

bool evf::FastMonitoringService::exceptionDetected ( ) const

Definition at line 457 of file FastMonitoringService.cc.

References has_data_exception_, and has_source_exception_.

Referenced by DAQSource::~DAQSource(), and FedRawDataInputSource::~FedRawDataInputSource().

457  {
458  return has_source_exception_.load() || has_data_exception_.load();
459  }
std::atomic< bool > has_data_exception_
std::atomic< bool > has_source_exception_

◆ fillDescriptions()

void evf::FastMonitoringService::fillDescriptions ( edm::ConfigurationDescriptions descriptions)
static

Definition at line 258 of file FastMonitoringService.cc.

References edm::ConfigurationDescriptions::add(), and submitPVResolutionJobs::desc.

258  {
260  desc.setComment("Service for File-based DAQ monitoring and event accounting");
261  desc.addUntracked<bool>("tbbMonitoringMode", true)
262  ->setComment("Monitor individual module processing per TBB thread instead of stream");
263  desc.addUntracked<bool>("tbbConcurrencyTracker", true)
264  ->setComment("Monitor TBB thread activity to flag microstate as real idle or overhead/other");
265  desc.addUntracked<int>("sleepTime", 1)->setComment("Sleep time of the monitoring thread");
266  desc.addUntracked<unsigned int>("fastMonIntervals", 2)
267  ->setComment("Modulo of sleepTime intervals on which fastmon file is written out");
268  desc.addUntracked<bool>("filePerFwkStream", true) //obsolete
269  ->setComment("Switches on monitoring output per framework stream");
270  desc.addUntracked<bool>("verbose", false)->setComment("Set to use LogInfo messages from the monitoring thread");
271  desc.setAllowAnything();
272  descriptions.add("FastMonitoringService", desc);
273  }
void add(std::string const &label, ParameterSetDescription const &psetDescription)

◆ getAbortFlagForLumi()

bool evf::FastMonitoringService::getAbortFlagForLumi ( unsigned int  lumi)

Definition at line 778 of file FastMonitoringService.cc.

References Exception, fmt_, CommonMethods::lock(), BXlumiParameters_cfi::lumi, and processedEventsPerLumi_.

Referenced by shouldWriteFiles().

778  {
779  std::lock_guard<std::mutex> lock(fmt_->monlock_);
780 
781  auto it = processedEventsPerLumi_.find(lumi);
782  if (it != processedEventsPerLumi_.end()) {
783  unsigned int abortFlag = it->second.second;
784  return abortFlag;
785  } else {
786  throw cms::Exception("FastMonitoringService")
787  << "output module wants already deleted (or never reported by SOURCE) lumisection status for LUMI -: "
788  << lumi;
789  return false;
790  }
791  }
std::unique_ptr< FastMonitoringThread > fmt_
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_

◆ getEventsProcessedForLumi()

unsigned int evf::FastMonitoringService::getEventsProcessedForLumi ( unsigned int  lumi,
bool *  abortFlag = nullptr 
)

Definition at line 760 of file FastMonitoringService.cc.

References Exception, fmt_, CommonMethods::lock(), BXlumiParameters_cfi::lumi, ValidateTausOnZEEFastSim_cff::proc, and processedEventsPerLumi_.

Referenced by dqm::DQMFileSaverPB::fillJson(), dqmfilesaver::fillJson(), evf::EvFOutputModule::globalEndLuminosityBlock(), evf::GlobalEvFOutputModule::globalEndLuminosityBlock(), dqm::DQMFileSaverPB::saveLumi(), and shouldWriteFiles().

760  {
761  std::lock_guard<std::mutex> lock(fmt_->monlock_);
762 
763  auto it = processedEventsPerLumi_.find(lumi);
764  if (it != processedEventsPerLumi_.end()) {
765  unsigned int proc = it->second.first;
766  if (abortFlag)
767  *abortFlag = it->second.second;
768  return proc;
769  } else {
770  throw cms::Exception("FastMonitoringService")
771  << "output module wants already deleted (or never reported by SOURCE) lumisection event count for LUMI -: "
772  << lumi;
773  return 0;
774  }
775  }
std::unique_ptr< FastMonitoringThread > fmt_
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_

◆ getRunDirName()

std::string evf::FastMonitoringService::getRunDirName ( ) const
inline

Definition at line 226 of file FastMonitoringService.h.

References runDirectory_.

226 { return runDirectory_.stem().string(); }
std::filesystem::path runDirectory_

◆ getSID() [1/2]

static unsigned int evf::FastMonitoringService::getSID ( edm::StreamContext const &  sc)
inlinestaticprivate

◆ getSID() [2/2]

static unsigned int evf::FastMonitoringService::getSID ( edm::StreamID const &  sid)
inlinestaticprivate

Definition at line 243 of file FastMonitoringService.h.

References edm::StreamID::value().

243 { return sid.value(); }

◆ getTID()

static unsigned int evf::FastMonitoringService::getTID ( )
inlinestatic

Definition at line 234 of file FastMonitoringService.h.

Referenced by postModuleEvent(), postModuleEventAcquire(), postSourceEvent(), preModuleEvent(), preModuleEventAcquire(), and preSourceEvent().

234 { return tbb::this_task_arena::current_thread_index(); }

◆ isExceptionOnData()

bool evf::FastMonitoringService::isExceptionOnData ( unsigned int  ls)

Definition at line 461 of file FastMonitoringService.cc.

References exceptionInLS_, fmt_, has_data_exception_, has_source_exception_, CommonMethods::lock(), and eostools::ls().

Referenced by evf::EvFDaqDirector::preGlobalEndLumi(), FedRawDataInputSource::read(), DAQSource::read(), DAQSource::~DAQSource(), and FedRawDataInputSource::~FedRawDataInputSource().

461  {
462  if (!has_data_exception_.load())
463  return false;
464  if (has_source_exception_.load())
465  return true;
466  std::lock_guard<std::mutex> lock(fmt_->monlock_);
467  for (auto ex : exceptionInLS_) {
468  if (ls == ex)
469  return true;
470  }
471  return false;
472  }
std::atomic< bool > has_data_exception_
std::unique_ptr< FastMonitoringThread > fmt_
def ls(path, rec=False)
Definition: eostools.py:349
std::atomic< bool > has_source_exception_
std::vector< unsigned int > exceptionInLS_

◆ jobFailure()

void evf::FastMonitoringService::jobFailure ( )

Definition at line 474 of file FastMonitoringService.cc.

References fmt_, and evf::FastMonState::sError.

Referenced by FastMonitoringService().

474 { fmt_->m_data.macrostate_ = FastMonState::sError; }
std::unique_ptr< FastMonitoringThread > fmt_

◆ makeInputLegendaJson()

std::string evf::FastMonitoringService::makeInputLegendaJson ( )

Definition at line 296 of file FastMonitoringService.cc.

References Json::Value::append(), Json::arrayValue, mps_fire::i, evf::FastMonState::inCOUNT, inputStateNames, and convertToRaw::writer.

Referenced by postBeginJob().

296  {
297  Json::Value legendaVector(Json::arrayValue);
298  for (int i = 0; i < FastMonState::inCOUNT; i++)
299  legendaVector.append(Json::Value(inputStateNames[i]));
300  Json::Value moduleLegend;
301  moduleLegend["names"] = legendaVector;
303  return writer.write(moduleLegend);
304  }
static const std::string inputStateNames[FastMonState::inCOUNT]
Represents a JSON value.
Definition: value.h:99
Writes a Value in JSON format in a human friendly way.
Definition: writer.h:63
array value (ordered list)
Definition: value.h:30

◆ makeModuleLegendaJson()

std::string evf::FastMonitoringService::makeModuleLegendaJson ( )

Definition at line 275 of file FastMonitoringService.cc.

References Json::Value::append(), Json::arrayValue, fmt_, mps_fire::i, nOutputModules_, evf::nReservedModules, evf::nSpecialModules, and convertToRaw::writer.

Referenced by postBeginJob().

275  {
276  Json::Value legendaVector(Json::arrayValue);
277  for (int i = 0; i < fmt_->m_data.encModule_.current_; i++)
278  legendaVector.append(
279  Json::Value((static_cast<const edm::ModuleDescription*>(fmt_->m_data.encModule_.decode(i)))->moduleLabel()));
280  //duplicate modules adding a list for acquire states (not all modules actually have it)
281  for (int i = 0; i < fmt_->m_data.encModule_.current_; i++)
282  legendaVector.append(Json::Value(
283  (static_cast<const edm::ModuleDescription*>(fmt_->m_data.encModule_.decode(i)))->moduleLabel() + "__ACQ"));
284  Json::Value valReserved(nReservedModules);
285  Json::Value valSpecial(nSpecialModules);
286  Json::Value valOutputModules(nOutputModules_);
287  Json::Value moduleLegend;
288  moduleLegend["names"] = legendaVector;
289  moduleLegend["reserved"] = valReserved;
290  moduleLegend["special"] = valSpecial;
291  moduleLegend["output"] = valOutputModules;
293  return writer.write(moduleLegend);
294  }
Represents a JSON value.
Definition: value.h:99
std::unique_ptr< FastMonitoringThread > fmt_
constexpr int nSpecialModules
constexpr int nReservedModules
Writes a Value in JSON format in a human friendly way.
Definition: writer.h:63
array value (ordered list)
Definition: value.h:30

◆ postBeginJob()

void evf::FastMonitoringService::postBeginJob ( )

Definition at line 492 of file FastMonitoringService.cc.

References fmt_, inputLegendFileJson_, CommonMethods::lock(), makeInputLegendaJson(), makeModuleLegendaJson(), moduleLegendFileJson_, evf::FastMonState::sJobReady, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FastMonitoringService().

492  {
493  std::string&& moduleLegStrJson = makeModuleLegendaJson();
494  FileIO::writeStringToFile(moduleLegendFileJson_, moduleLegStrJson);
495 
496  std::string inputLegendStrJson = makeInputLegendaJson();
497  FileIO::writeStringToFile(inputLegendFileJson_, inputLegendStrJson);
498 
499  fmt_->m_data.macrostate_ = FastMonState::sJobReady;
500 
501  //update number of entries in module histogram
502  std::lock_guard<std::mutex> lock(fmt_->monlock_);
503  //double the size to add post-acquire states
504  fmt_->m_data.microstateBins_ = fmt_->m_data.encModule_.vecsize() * 2;
505  }
std::unique_ptr< FastMonitoringThread > fmt_

◆ postEndJob()

void evf::FastMonitoringService::postEndJob ( )

Definition at line 507 of file FastMonitoringService.cc.

References fmt_, and evf::FastMonState::sJobEnded.

Referenced by FastMonitoringService().

507  {
508  fmt_->m_data.macrostate_ = FastMonState::sJobEnded;
509  fmt_->stop();
510  }
std::unique_ptr< FastMonitoringThread > fmt_

◆ postEvent()

void evf::FastMonitoringService::postEvent ( edm::StreamContext const &  sc)

Definition at line 642 of file FastMonitoringService.cc.

References fmt_, evf::getmIdle(), microstate_, edm::StreamContext::streamID(), totalEventsProcessed_, and edm::StreamID::value().

Referenced by FastMonitoringService().

642  {
643  (*(fmt_->m_data.processed_[sc.streamID().value()]))++;
644  //fast path counter (events accumulated in a run)
645  unsigned long res = totalEventsProcessed_.fetch_add(1, std::memory_order_relaxed);
646  fmt_->m_data.fastPathProcessedJ_ = res + 1;
647 
648  microstate_[sc.streamID().value()] = getmIdle();
649  }
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmIdle()
std::unique_ptr< FastMonitoringThread > fmt_
Definition: Electron.h:6
std::atomic< unsigned long > totalEventsProcessed_

◆ postGlobalBeginRun()

void evf::FastMonitoringService::postGlobalBeginRun ( edm::GlobalContext const &  gc)

Definition at line 512 of file FastMonitoringService.cc.

References fmt_, isInitTransition_, and evf::FastMonState::sRunning.

512  {
513  fmt_->m_data.macrostate_ = FastMonState::sRunning;
514  isInitTransition_ = false;
515  }
std::atomic< bool > isInitTransition_
std::unique_ptr< FastMonitoringThread > fmt_

◆ postGlobalEndLumi()

void evf::FastMonitoringService::postGlobalEndLumi ( edm::GlobalContext const &  gc)

Definition at line 599 of file FastMonitoringService.cc.

References avgLeadTime_, filesProcessedDuringLumi_, fmt_, CommonMethods::lock(), lockStatsDuringLumi_, edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), and processedEventsPerLumi_.

Referenced by FastMonitoringService().

599  {
600  std::lock_guard<std::mutex> lock(fmt_->monlock_);
601  unsigned int lumi = gc.luminosityBlockID().luminosityBlock();
602  //LS monitoring snapshot with input source data has been taken in previous callback
603  avgLeadTime_.erase(lumi);
605  lockStatsDuringLumi_.erase(lumi);
606 
607  //output module already used this in end lumi (this could be migrated to EvFDaqDirector as it is essential for FFF bookkeeping)
609  }
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
std::unique_ptr< FastMonitoringThread > fmt_
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
std::map< unsigned int, double > avgLeadTime_
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_

◆ postModuleEvent()

void evf::FastMonitoringService::postModuleEvent ( edm::StreamContext const &  sc,
edm::ModuleCallingContext const &  mcc 
)

Definition at line 692 of file FastMonitoringService.cc.

References evf::getmFwkOvhMod(), evf::getmIdle(), getSID(), getTID(), microstate_, tbbMonitoringMode_, and tmicrostate_.

Referenced by FastMonitoringService().

692  {
694  if (!tbbMonitoringMode_)
695  return;
697  }
constexpr edm::ModuleDescription const * getmFwkOvhMod()
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmIdle()
static unsigned int getSID(edm::StreamContext const &sc)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ postModuleEventAcquire()

void evf::FastMonitoringService::postModuleEventAcquire ( edm::StreamContext const &  sc,
edm::ModuleCallingContext const &  mcc 
)

Definition at line 675 of file FastMonitoringService.cc.

References evf::getmFwkOvhMod(), evf::getmIdle(), getSID(), getTID(), microstate_, microstateAcqFlag_, tbbMonitoringMode_, tmicrostate_, and tmicrostateAcqFlag_.

Referenced by FastMonitoringService().

676  {
678  microstateAcqFlag_[getSID(sc)] = 0;
679  if (!tbbMonitoringMode_)
680  return;
683  }
constexpr edm::ModuleDescription const * getmFwkOvhMod()
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmIdle()
std::vector< ContainableAtomic< unsigned char > > tmicrostateAcqFlag_
std::vector< ContainableAtomic< unsigned char > > microstateAcqFlag_
static unsigned int getSID(edm::StreamContext const &sc)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ postSourceEvent()

void evf::FastMonitoringService::postSourceEvent ( edm::StreamID  sid)

Definition at line 658 of file FastMonitoringService.cc.

References evf::getmFwkOvhSrc(), evf::getmIdle(), getSID(), getTID(), microstate_, tbbMonitoringMode_, and tmicrostate_.

Referenced by FastMonitoringService().

658  {
659  microstate_[getSID(sid)] = getmFwkOvhSrc();
660  if (!tbbMonitoringMode_)
661  return;
663  }
constexpr edm::ModuleDescription const * getmFwkOvhSrc()
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmIdle()
static unsigned int getSID(edm::StreamContext const &sc)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ postStreamBeginLumi()

void evf::FastMonitoringService::postStreamBeginLumi ( edm::StreamContext const &  sc)

Definition at line 621 of file FastMonitoringService.cc.

References evf::getmIdle(), microstate_, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

621  {
622  microstate_[sc.streamID().value()] = getmIdle();
623  }
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmIdle()

◆ postStreamEndLumi()

void evf::FastMonitoringService::postStreamEndLumi ( edm::StreamContext const &  sc)

Definition at line 634 of file FastMonitoringService.cc.

References evf::getmFwkEoL(), microstate_, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

634  {
635  microstate_[sc.streamID().value()] = getmFwkEoL();
636  }
constexpr edm::ModuleDescription const * getmFwkEoL()
std::vector< ContainableAtomic< const void * > > microstate_

◆ preallocate()

void evf::FastMonitoringService::preallocate ( edm::service::SystemBounds const &  bounds)

Definition at line 306 of file FastMonitoringService.cc.

References ct_, SiStripPI::max, nMonThreads_, nStreams_, and nThreads_.

Referenced by FastMonitoringService().

306  {
307  nStreams_ = bounds.maxNumberOfStreams();
308  nThreads_ = bounds.maxNumberOfThreads();
309  //this should already be >=1
310  if (nStreams_ == 0)
311  nStreams_ = 1;
312  if (nThreads_ == 0)
313  nThreads_ = 1;
315  ct_ = std::make_unique<ConcurrencyTracker>(nThreads_);
316  //start concurrency tracking
317  }
std::unique_ptr< ConcurrencyTracker > ct_

◆ preBeginJob()

void evf::FastMonitoringService::preBeginJob ( edm::PathsAndConsumesOfModulesBase const &  ,
edm::ProcessContext const &  pc 
)

Definition at line 319 of file FastMonitoringService.cc.

References ct_, Exception, fastMicrostateDefPath_, fastName_, fastPath_, fmt_, evf::getmInvalid(), mps_fire::i, evf::FastMonState::inCOUNT, inputLegendFileJson_, isInitTransition_, lastGlobalLumi_, LogDebug, lumiFromSource_, evf::FastMonState::mCOUNT, evf::FastMonState::MCOUNT, microstate_, microstateAcqFlag_, microstateDefPath_, moduleLegendFile_, moduleLegendFileJson_, monInit_, nMonThreads_, nStreams_, nThreads_, Utilities::operator, castor_dqm_sourceclient_file_cfg::path, runDirectory_, evf::FastMonState::sInit, sleepTime_, snapshotRunner(), specialMicroStateNames, streamCounterUpdating_, tbbConcurrencyTracker_, tmicrostate_, tmicrostateAcqFlag_, and workingDirectory_.

Referenced by FastMonitoringService().

319  {
320  // FIND RUN DIRECTORY
321  // The run dir should be set via the configuration of EvFDaqDirector
323  ct_->activate();
324 
325  if (edm::Service<evf::EvFDaqDirector>().operator->() == nullptr) {
326  throw cms::Exception("FastMonitoringService") << "EvFDaqDirector is not present";
327  }
328  std::filesystem::path runDirectory{edm::Service<evf::EvFDaqDirector>()->baseRunDir()};
329  workingDirectory_ = runDirectory_ = runDirectory;
330  workingDirectory_ /= "mon";
331 
332  if (!std::filesystem::is_directory(workingDirectory_)) {
333  LogDebug("FastMonitoringService") << "<MON> DIR NOT FOUND! Trying to create -: " << workingDirectory_.string();
334  std::filesystem::create_directories(workingDirectory_);
335  if (!std::filesystem::is_directory(workingDirectory_))
336  edm::LogWarning("FastMonitoringService") << "Unable to create <MON> DIR -: " << workingDirectory_.string()
337  << ". No monitoring data will be written.";
338  }
339 
340  std::ostringstream fastFileName;
341 
342  fastFileName << fastName_ << "_pid" << std::setfill('0') << std::setw(5) << getpid() << ".fast";
344  fast /= fastFileName.str();
345  fastPath_ = fast.string();
346 
347  std::ostringstream moduleLegFile;
348  std::ostringstream moduleLegFileJson;
349  moduleLegFile << "microstatelegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".leg";
350  moduleLegFileJson << "microstatelegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
351  moduleLegendFile_ = (workingDirectory_ / moduleLegFile.str()).string();
352  moduleLegendFileJson_ = (workingDirectory_ / moduleLegFileJson.str()).string();
353 
354  std::ostringstream inputLegFileJson;
355  inputLegFileJson << "inputlegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
356  inputLegendFileJson_ = (workingDirectory_ / inputLegFileJson.str()).string();
357 
358  LogDebug("FastMonitoringService") << "Initializing FastMonitor with microstate def path -: " << microstateDefPath_;
359 
360  /*
361  * initialize the fast monitor with:
362  * vector of pointers to monitorable parameters
363  * path to definition
364  *
365  */
366 
367  fmt_->m_data.macrostate_ = FastMonState::sInit;
368 
369  for (unsigned int i = 0; i < (FastMonState::mCOUNT); i++)
370  fmt_->m_data.encModule_.updateReserved(static_cast<const void*>(specialMicroStateNames + i));
371  fmt_->m_data.encModule_.completeReservedWithDummies();
372 
373  for (unsigned int i = 0; i < nMonThreads_; i++) {
374  microstate_.emplace_back(getmInvalid());
375  microstateAcqFlag_.push_back(0);
376  tmicrostate_.emplace_back(getmInvalid());
377  tmicrostateAcqFlag_.push_back(0);
378 
379  //for synchronization
380  streamCounterUpdating_.push_back(new std::atomic<bool>(false));
381  }
382 
383  //initial size until we detect number of bins
384  fmt_->m_data.macrostateBins_ = FastMonState::MCOUNT;
385  fmt_->m_data.microstateBins_ = 0;
386  fmt_->m_data.inputstateBins_ = FastMonState::inCOUNT;
387 
388  lastGlobalLumi_ = 0;
389  isInitTransition_ = true;
390  lumiFromSource_ = 0;
391 
392  //startup monitoring
393  fmt_->resetFastMonitor(microstateDefPath_, fastMicrostateDefPath_);
394  fmt_->jsonMonitor_->setNStreams(nMonThreads_);
395  fmt_->m_data.registerVariables(fmt_->jsonMonitor_.get(), nMonThreads_, nStreams_, nThreads_);
396  monInit_.store(false, std::memory_order_release);
397  if (sleepTime_ > 0)
399  }
constexpr edm::ModuleDescription const * getmInvalid()
std::vector< ContainableAtomic< const void * > > microstate_
std::atomic< bool > isInitTransition_
std::filesystem::path workingDirectory_
std::vector< std::atomic< bool > * > streamCounterUpdating_
std::unique_ptr< FastMonitoringThread > fmt_
std::vector< ContainableAtomic< unsigned char > > tmicrostateAcqFlag_
std::filesystem::path runDirectory_
std::vector< ContainableAtomic< unsigned char > > microstateAcqFlag_
std::unique_ptr< ConcurrencyTracker > ct_
Log< level::Warning, false > LogWarning
static const edm::ModuleDescription specialMicroStateNames[FastMonState::mCOUNT]
#define LogDebug(id)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ preEvent()

void evf::FastMonitoringService::preEvent ( edm::StreamContext const &  sc)

Definition at line 638 of file FastMonitoringService.cc.

References evf::getmEvent(), microstate_, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

638  {
639  microstate_[sc.streamID().value()] = getmEvent();
640  }
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmEvent()

◆ preGlobalBeginLumi()

void evf::FastMonitoringService::preGlobalBeginLumi ( edm::GlobalContext const &  gc)

Definition at line 517 of file FastMonitoringService.cc.

References fmt_, evf::getmIdle(), evf::getmInvalid(), mps_fire::i, lastGlobalLumi_, CommonMethods::lock(), edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), lumiStartTime_, nThreads_, tbbMonitoringMode_, and tmicrostate_.

Referenced by FastMonitoringService().

517  {
518  timeval lumiStartTime;
519  gettimeofday(&lumiStartTime, nullptr);
520  unsigned int newLumi = gc.luminosityBlockID().luminosityBlock();
521  lastGlobalLumi_ = newLumi;
522 
523  std::lock_guard<std::mutex> lock(fmt_->monlock_);
524  lumiStartTime_[newLumi] = lumiStartTime;
525  //reset all states to idle
526  if (tbbMonitoringMode_)
527  for (unsigned i = 0; i < nThreads_; i++)
528  if (tmicrostate_[i] == getmInvalid())
529  tmicrostate_[i] = getmIdle();
530  }
constexpr edm::ModuleDescription const * getmInvalid()
constexpr edm::ModuleDescription const * getmIdle()
std::map< unsigned int, timeval > lumiStartTime_
std::unique_ptr< FastMonitoringThread > fmt_
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ preGlobalEarlyTermination()

void evf::FastMonitoringService::preGlobalEarlyTermination ( edm::GlobalContext const &  gc,
edm::TerminationOrigin  to 
)

Definition at line 417 of file FastMonitoringService.cc.

References visDQMUpload::context, edm::ExceptionFromAnotherContext, edm::ExceptionFromThisContext, exceptionInLS_, edm::ExternalSignal, fmt_, has_data_exception_, CommonMethods::lock(), edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FastMonitoringService().

417  {
420  context = " FromThisContext ";
422  context = " FromAnotherContext";
424  context = " FromExternalSignal";
425  edm::LogWarning("FastMonitoringService")
426  << " GLOBAL "
427  << "earlyTermination -: LS:" << gc.luminosityBlockID().luminosityBlock() << " " << context;
428  std::lock_guard<std::mutex> lock(fmt_->monlock_);
429  exceptionInLS_.push_back(gc.luminosityBlockID().luminosityBlock());
430  has_data_exception_.store(true);
431  }
std::atomic< bool > has_data_exception_
std::unique_ptr< FastMonitoringThread > fmt_
Log< level::Warning, false > LogWarning
std::vector< unsigned int > exceptionInLS_

◆ preGlobalEndLumi()

void evf::FastMonitoringService::preGlobalEndLumi ( edm::GlobalContext const &  gc)

Definition at line 532 of file FastMonitoringService.cc.

References accuSize_, daqInputSource_, doSnapshot(), Exception, exception_detected_, exceptionInLS_, fmt_, DAQSource::getEventReport(), FedRawDataInputSource::getEventReport(), inputSource_, CommonMethods::lock(), LogDebug, BXlumiParameters_cfi::lumi, edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), lumiStartTime_, processedEventsPerLumi_, edm::shutdown_flag, throughputFactor(), and jsoncollector::IntJ::value().

Referenced by FastMonitoringService().

532  {
533  unsigned int lumi = gc.luminosityBlockID().luminosityBlock();
534  LogDebug("FastMonitoringService") << "Lumi ended. Writing JSON information. LUMI -: " << lumi;
535  timeval lumiStopTime;
536  gettimeofday(&lumiStopTime, nullptr);
537 
538  std::lock_guard<std::mutex> lock(fmt_->monlock_);
539 
540  // Compute throughput
541  timeval stt = lumiStartTime_[lumi];
542  lumiStartTime_.erase(lumi);
543  unsigned long usecondsForLumi = (lumiStopTime.tv_sec - stt.tv_sec) * 1000000 + (lumiStopTime.tv_usec - stt.tv_usec);
544  unsigned long accuSize = accuSize_.find(lumi) == accuSize_.end() ? 0 : accuSize_[lumi];
545  accuSize_.erase(lumi);
546  double throughput = throughputFactor() * double(accuSize) / double(usecondsForLumi);
547  //store to registered variable
548  fmt_->m_data.fastThroughputJ_.value() = throughput;
549 
550  //update
551  doSnapshot(lumi, true);
552 
553  //retrieve one result we need (todo: sanity check if it's found)
554  IntJ* lumiProcessedJptr = dynamic_cast<IntJ*>(fmt_->jsonMonitor_->getMergedIntJForLumi("Processed", lumi));
555  if (!lumiProcessedJptr)
556  throw cms::Exception("FastMonitoringService") << "Internal error: got null pointer from FastMonitor";
557  processedEventsPerLumi_[lumi] = std::pair<unsigned int, bool>(lumiProcessedJptr->value(), false);
558 
559  //checking if exception has been thrown (in case of Global/Stream early termination, for this LS)
560  bool exception_detected = exception_detected_;
561  for (auto ex : exceptionInLS_)
562  if (lumi == ex)
563  exception_detected = true;
564 
565  if (edm::shutdown_flag || exception_detected) {
566  edm::LogInfo("FastMonitoringService")
567  << "Run interrupted. Skip writing EoL information -: " << processedEventsPerLumi_[lumi].first
568  << " events were processed in LUMI " << lumi;
569  //this will prevent output modules from producing json file for possibly incomplete lumi
570  processedEventsPerLumi_[lumi].first = 0;
571  processedEventsPerLumi_[lumi].second = true;
572  //disable this exception, so service can be used standalone (will be thrown if output module asks for this information)
573  //throw cms::Exception("FastMonitoringService") << "SOURCE did not send update for lumi block. LUMI -:" << lumi;
574  return;
575  }
576 
577  if (inputSource_ || daqInputSource_) {
578  auto sourceReport =
580  if (sourceReport.first) {
581  if (sourceReport.second != processedEventsPerLumi_[lumi].first) {
582  throw cms::Exception("FastMonitoringService") << "MISMATCH with SOURCE update. LUMI -: " << lumi
583  << ", events(processed):" << processedEventsPerLumi_[lumi].first
584  << " events(source):" << sourceReport.second;
585  }
586  }
587  }
588 
589  edm::LogInfo("FastMonitoringService")
590  << "Statistics for lumisection -: lumi = " << lumi << " events = " << lumiProcessedJptr->value()
591  << " time = " << usecondsForLumi / 1000000 << " size = " << accuSize << " thr = " << throughput;
592  delete lumiProcessedJptr;
593 
594  //full global and stream merge (will be used by output modules), output from this service is deprecated
595  fmt_->jsonMonitor_->outputFullJSON("dummy", lumi, false);
596  fmt_->jsonMonitor_->discardCollected(lumi); //we don't do further updates for this lumi
597  }
std::map< unsigned int, timeval > lumiStartTime_
std::pair< bool, unsigned int > getEventReport(unsigned int lumi, bool erase)
std::map< unsigned int, unsigned long > accuSize_
volatile std::atomic< bool > shutdown_flag
std::pair< bool, unsigned int > getEventReport(unsigned int lumi, bool erase)
Definition: DAQSource.cc:1388
std::unique_ptr< FastMonitoringThread > fmt_
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
void doSnapshot(const unsigned int ls, const bool isGlobalEOL)
constexpr double throughputFactor()
Log< level::Info, false > LogInfo
FedRawDataInputSource * inputSource_
std::vector< unsigned int > exceptionInLS_
#define LogDebug(id)

◆ preModuleBeginJob()

void evf::FastMonitoringService::preModuleBeginJob ( edm::ModuleDescription const &  desc)

Definition at line 477 of file FastMonitoringService.cc.

References submitPVResolutionJobs::desc, fmt_, CommonMethods::lock(), and nOutputModules_.

Referenced by FastMonitoringService().

477  {
478  std::lock_guard<std::mutex> lock(fmt_->monlock_);
479  //std::cout << " Pre module Begin Job module: " << desc.moduleName() << std::endl;
480 
481  //build a map of modules keyed by their module description address
482  //here we need to treat output modules in a special way so they can be easily singled out
483  if (desc.moduleName() == "Stream" || desc.moduleName() == "GlobalEvFOutputModule" ||
484  desc.moduleName() == "EvFOutputModule" || desc.moduleName() == "EventStreamFileWriter" ||
485  desc.moduleName() == "PoolOutputModule") {
486  fmt_->m_data.encModule_.updateReserved((void*)&desc);
487  nOutputModules_++;
488  } else
489  fmt_->m_data.encModule_.update((void*)&desc);
490  }
std::unique_ptr< FastMonitoringThread > fmt_

◆ preModuleEvent()

void evf::FastMonitoringService::preModuleEvent ( edm::StreamContext const &  sc,
edm::ModuleCallingContext const &  mcc 
)

Definition at line 685 of file FastMonitoringService.cc.

References getSID(), getTID(), microstate_, edm::ModuleCallingContext::moduleDescription(), tbbMonitoringMode_, and tmicrostate_.

Referenced by FastMonitoringService().

685  {
686  microstate_[getSID(sc)] = (void*)(mcc.moduleDescription());
687  if (!tbbMonitoringMode_)
688  return;
689  tmicrostate_[getTID()] = (void*)(mcc.moduleDescription());
690  }
std::vector< ContainableAtomic< const void * > > microstate_
static unsigned int getSID(edm::StreamContext const &sc)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ preModuleEventAcquire()

void evf::FastMonitoringService::preModuleEventAcquire ( edm::StreamContext const &  sc,
edm::ModuleCallingContext const &  mcc 
)

Definition at line 665 of file FastMonitoringService.cc.

References getSID(), getTID(), microstate_, microstateAcqFlag_, edm::ModuleCallingContext::moduleDescription(), tbbMonitoringMode_, tmicrostate_, and tmicrostateAcqFlag_.

Referenced by FastMonitoringService().

666  {
667  microstate_[getSID(sc)] = (void*)(mcc.moduleDescription());
668  microstateAcqFlag_[getSID(sc)] = 1;
669  if (!tbbMonitoringMode_)
670  return;
671  tmicrostate_[getTID()] = (void*)(mcc.moduleDescription());
673  }
std::vector< ContainableAtomic< const void * > > microstate_
std::vector< ContainableAtomic< unsigned char > > tmicrostateAcqFlag_
std::vector< ContainableAtomic< unsigned char > > microstateAcqFlag_
static unsigned int getSID(edm::StreamContext const &sc)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ prePathEvent()

void evf::FastMonitoringService::prePathEvent ( edm::StreamContext const &  ,
edm::PathContext const &   
)

◆ preSourceEarlyTermination()

void evf::FastMonitoringService::preSourceEarlyTermination ( edm::TerminationOrigin  to)

Definition at line 433 of file FastMonitoringService.cc.

References visDQMUpload::context, exception_detected_, edm::ExceptionFromAnotherContext, edm::ExceptionFromThisContext, edm::ExternalSignal, fmt_, has_data_exception_, has_source_exception_, CommonMethods::lock(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FastMonitoringService().

433  {
436  context = " FromThisContext ";
438  context = " FromAnotherContext";
440  context = " FromExternalSignal";
441  edm::LogWarning("FastMonitoringService") << " SOURCE "
442  << "earlyTermination -: " << context;
443  std::lock_guard<std::mutex> lock(fmt_->monlock_);
444  exception_detected_ = true;
445  has_source_exception_.store(true);
446  has_data_exception_.store(true);
447  }
std::atomic< bool > has_data_exception_
std::unique_ptr< FastMonitoringThread > fmt_
std::atomic< bool > has_source_exception_
Log< level::Warning, false > LogWarning

◆ preSourceEvent()

void evf::FastMonitoringService::preSourceEvent ( edm::StreamID  sid)

Definition at line 651 of file FastMonitoringService.cc.

References evf::getmInput(), getSID(), getTID(), microstate_, tbbMonitoringMode_, and tmicrostate_.

Referenced by FastMonitoringService().

651  {
652  microstate_[getSID(sid)] = getmInput();
653  if (!tbbMonitoringMode_)
654  return;
656  }
std::vector< ContainableAtomic< const void * > > microstate_
static unsigned int getSID(edm::StreamContext const &sc)
constexpr edm::ModuleDescription const * getmInput()
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ preStreamBeginLumi()

void evf::FastMonitoringService::preStreamBeginLumi ( edm::StreamContext const &  sc)

Definition at line 611 of file FastMonitoringService.cc.

References edm::StreamContext::eventID(), fmt_, evf::getmBoL(), CommonMethods::lock(), edm::EventID::luminosityBlock(), microstate_, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

611  {
612  std::lock_guard<std::mutex> lock(fmt_->monlock_);
613  fmt_->m_data.streamLumi_[sc.streamID().value()] = sc.eventID().luminosityBlock();
614 
615  //reset collected values for this stream
616  *(fmt_->m_data.processed_[sc.streamID().value()]) = 0;
617 
618  microstate_[sc.streamID().value()] = getmBoL();
619  }
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmBoL()
std::unique_ptr< FastMonitoringThread > fmt_

◆ preStreamEarlyTermination()

void evf::FastMonitoringService::preStreamEarlyTermination ( edm::StreamContext const &  sc,
edm::TerminationOrigin  to 
)

Definition at line 401 of file FastMonitoringService.cc.

References visDQMUpload::context, edm::StreamContext::eventID(), edm::ExceptionFromAnotherContext, edm::ExceptionFromThisContext, exceptionInLS_, edm::ExternalSignal, fmt_, has_data_exception_, CommonMethods::lock(), edm::EventID::luminosityBlock(), edm::StreamContext::streamID(), AlCaHLTBitMon_QueryRunRegistry::string, and edm::StreamID::value().

Referenced by FastMonitoringService().

401  {
404  context = " FromThisContext ";
406  context = " FromAnotherContext";
408  context = " FromExternalSignal";
409  edm::LogWarning("FastMonitoringService")
410  << " STREAM " << sc.streamID().value() << " earlyTermination -: ID:" << sc.eventID()
411  << " LS:" << sc.eventID().luminosityBlock() << " " << context;
412  std::lock_guard<std::mutex> lock(fmt_->monlock_);
413  exceptionInLS_.push_back(sc.eventID().luminosityBlock());
414  has_data_exception_.store(true);
415  }
std::atomic< bool > has_data_exception_
std::unique_ptr< FastMonitoringThread > fmt_
Log< level::Warning, false > LogWarning
std::vector< unsigned int > exceptionInLS_

◆ preStreamEndLumi()

void evf::FastMonitoringService::preStreamEndLumi ( edm::StreamContext const &  sc)

Definition at line 625 of file FastMonitoringService.cc.

References edm::StreamContext::eventID(), fmt_, evf::getmEoL(), CommonMethods::lock(), edm::EventID::luminosityBlock(), microstate_, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

625  {
626  std::lock_guard<std::mutex> lock(fmt_->monlock_);
627  //update processed count to be complete at this time
628  //doStreamEOLSnapshot(sc.eventID().luminosityBlock(), sid);
629  fmt_->jsonMonitor_->snapStreamAtomic(sc.eventID().luminosityBlock(), sc.streamID().value());
630  //reset this in case stream does not get notified of next lumi (we keep processed events only)
631  microstate_[sc.streamID().value()] = getmEoL();
632  }
std::vector< ContainableAtomic< const void * > > microstate_
std::unique_ptr< FastMonitoringThread > fmt_
constexpr edm::ModuleDescription const * getmEoL()

◆ reportLockWait()

void evf::FastMonitoringService::reportLockWait ( unsigned int  ls,
double  waitTime,
unsigned int  lockCount 
)

Definition at line 750 of file FastMonitoringService.cc.

References fmt_, CommonMethods::lock(), lockStatsDuringLumi_, and eostools::ls().

Referenced by FedRawDataInputSource::readSupervisor(), and DAQSource::readSupervisor().

750  {
751  std::lock_guard<std::mutex> lock(fmt_->monlock_);
752  lockStatsDuringLumi_[ls] = std::pair<double, unsigned int>(waitTime, lockCount);
753  }
std::unique_ptr< FastMonitoringThread > fmt_
def ls(path, rec=False)
Definition: eostools.py:349
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_

◆ setExceptionDetected()

void evf::FastMonitoringService::setExceptionDetected ( unsigned int  ls)

Definition at line 449 of file FastMonitoringService.cc.

References exception_detected_, exceptionInLS_, fmt_, CommonMethods::lock(), and eostools::ls().

Referenced by FedRawDataInputSource::getNextEvent(), and DAQSource::getNextEventFromDataBlock().

449  {
450  std::lock_guard<std::mutex> lock(fmt_->monlock_);
451  if (!ls)
452  exception_detected_ = true;
453  else
454  exceptionInLS_.push_back(ls);
455  }
std::unique_ptr< FastMonitoringThread > fmt_
def ls(path, rec=False)
Definition: eostools.py:349
std::vector< unsigned int > exceptionInLS_

◆ setInputSource() [1/2]

void evf::FastMonitoringService::setInputSource ( FedRawDataInputSource inputSource)
inline

Definition at line 227 of file FastMonitoringService.h.

References inputSource_.

Referenced by DAQSource::DAQSource(), and FedRawDataInputSource::FedRawDataInputSource().

227 { inputSource_ = inputSource; }
FedRawDataInputSource * inputSource_

◆ setInputSource() [2/2]

void evf::FastMonitoringService::setInputSource ( DAQSource inputSource)
inline

Definition at line 228 of file FastMonitoringService.h.

References daqInputSource_.

228 { daqInputSource_ = inputSource; }

◆ setInState()

void evf::FastMonitoringService::setInState ( FastMonState::InputState  inputState)
inline

◆ setInStateSup()

void evf::FastMonitoringService::setInStateSup ( FastMonState::InputState  inputState)
inline

◆ setTMicrostate()

void evf::FastMonitoringService::setTMicrostate ( FastMonState::Microstate  m)

Definition at line 755 of file FastMonitoringService.cc.

References visualization-live-secondInstance_cfg::m, specialMicroStateNames, and tmicrostate_.

Referenced by IdleSourceSentry::IdleSourceSentry(), and IdleSourceSentry::~IdleSourceSentry().

755  {
756  tmicrostate_[tbb::this_task_arena::current_thread_index()] = &specialMicroStateNames[m];
757  }
static const edm::ModuleDescription specialMicroStateNames[FastMonState::mCOUNT]
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ shouldWriteFiles()

bool evf::FastMonitoringService::shouldWriteFiles ( unsigned int  lumi,
unsigned int *  proc = nullptr 
)
inline

Definition at line 220 of file FastMonitoringService.h.

References getAbortFlagForLumi(), getEventsProcessedForLumi(), and ValidateTausOnZEEFastSim_cff::proc.

Referenced by L1TriggerJSONMonitoring::globalEndLuminosityBlockSummary(), and HLTriggerJSONMonitoring::globalEndLuminosityBlockSummary().

220  {
221  unsigned int processed = getEventsProcessedForLumi(lumi);
222  if (proc)
223  *proc = processed;
224  return !getAbortFlagForLumi(lumi);
225  }
unsigned int getEventsProcessedForLumi(unsigned int lumi, bool *abortFlag=nullptr)
bool getAbortFlagForLumi(unsigned int lumi)

◆ snapshotRunner()

void evf::FastMonitoringService::snapshotRunner ( )
private

Definition at line 794 of file FastMonitoringService.cc.

References doSnapshot(), f, fastMonIntervals_, fastPath_, fmt_, mps_fire::i, inputState_, inputStateNames, inputSupervisorState_, lastGlobalLumi_, CommonMethods::lock(), monInit_, mps_check::msg, nMonThreads_, AlCaHLTBitMon_ParallelJobs::p, sleepTime_, snapCounter_, and verbose_.

Referenced by preBeginJob().

794  {
795  monInit_.exchange(true, std::memory_order_acquire);
796  while (!fmt_->m_stoprequest) {
797  std::vector<std::vector<unsigned int>> lastEnc;
798  {
799  std::unique_lock<std::mutex> lock(fmt_->monlock_);
800 
801  doSnapshot(lastGlobalLumi_, false);
802 
803  lastEnc.emplace_back(fmt_->m_data.tmicrostateEncoded_);
804  lastEnc.emplace_back(fmt_->m_data.microstateEncoded_);
805 
807  std::vector<std::string> CSVv;
808  for (unsigned int i = 0; i < nMonThreads_; i++) {
809  CSVv.push_back(fmt_->jsonMonitor_->getCSVString((int)i));
810  }
811  // release mutex before writing out fast path file
812  lock.release()->unlock();
813  fmt_->jsonMonitor_->outputCSV(fastPath_, CSVv);
814  }
815  snapCounter_++;
816  }
817 
818  if (verbose_) {
819  edm::LogInfo msg("FastMonitoringService");
820  auto f = [&](std::vector<unsigned int> const& p) {
821  for (unsigned int i = 0; i < nMonThreads_; i++) {
822  if (i == 0)
823  msg << "[" << p[i] << ",";
824  else if (i <= nMonThreads_ - 1)
825  msg << p[i] << ",";
826  else
827  msg << p[i] << "]";
828  }
829  };
830 
831  msg << "Current states: Ms=" << fmt_->m_data.fastMacrostateJ_.value() << " ms=";
832  f(lastEnc[0]);
833  msg << " us=";
834  f(lastEnc[1]);
836  }
837 
838  ::sleep(sleepTime_);
839  }
840  }
std::atomic< FastMonState::InputState > inputState_
static const std::string inputStateNames[FastMonState::inCOUNT]
std::unique_ptr< FastMonitoringThread > fmt_
void doSnapshot(const unsigned int ls, const bool isGlobalEOL)
std::atomic< FastMonState::InputState > inputSupervisorState_
double f[11][100]
tuple msg
Definition: mps_check.py:286

◆ startedLookingForFile()

void evf::FastMonitoringService::startedLookingForFile ( )

Definition at line 714 of file FastMonitoringService.cc.

References fileLookStart_.

Referenced by FedRawDataInputSource::readSupervisor(), and DAQSource::readSupervisor().

714  {
715  gettimeofday(&fileLookStart_, nullptr);
716  /*
717  std::cout << "Started looking for .raw file at: s=" << fileLookStart_.tv_sec << ": ms = "
718  << fileLookStart_.tv_usec / 1000.0 << std::endl;
719  */
720  }

◆ stoppedLookingForFile()

void evf::FastMonitoringService::stoppedLookingForFile ( unsigned int  lumi)

Definition at line 722 of file FastMonitoringService.cc.

References avgLeadTime_, fileLookStart_, fileLookStop_, fmt_, mps_fire::i, leadTimes_, CommonMethods::lock(), BXlumiParameters_cfi::lumi, and lumiFromSource_.

Referenced by FedRawDataInputSource::readSupervisor(), and DAQSource::readSupervisor().

722  {
723  gettimeofday(&fileLookStop_, nullptr);
724  /*
725  std::cout << "Stopped looking for .raw file at: s=" << fileLookStop_.tv_sec << ": ms = "
726  << fileLookStop_.tv_usec / 1000.0 << std::endl;
727  */
728  std::lock_guard<std::mutex> lock(fmt_->monlock_);
729 
730  if (lumi > lumiFromSource_) {
732  leadTimes_.clear();
733  }
734  unsigned long elapsedTime = (fileLookStop_.tv_sec - fileLookStart_.tv_sec) * 1000000 // sec to us
735  + (fileLookStop_.tv_usec - fileLookStart_.tv_usec); // us
736  // add this to lead times for this lumi
737  leadTimes_.push_back((double)elapsedTime);
738 
739  // recompute average lead time for this lumi
740  if (leadTimes_.size() == 1)
742  else {
743  double totTime = 0;
744  for (unsigned int i = 0; i < leadTimes_.size(); i++)
745  totTime += leadTimes_[i];
746  avgLeadTime_[lumi] = 0.001 * (totTime / leadTimes_.size());
747  }
748  }
std::unique_ptr< FastMonitoringThread > fmt_
std::map< unsigned int, double > avgLeadTime_
std::vector< double > leadTimes_

Member Data Documentation

◆ accuSize_

std::map<unsigned int, unsigned long> evf::FastMonitoringService::accuSize_
private

Definition at line 280 of file FastMonitoringService.h.

Referenced by accumulateFileSize(), and preGlobalEndLumi().

◆ avgLeadTime_

std::map<unsigned int, double> evf::FastMonitoringService::avgLeadTime_
private

Definition at line 277 of file FastMonitoringService.h.

Referenced by doSnapshot(), postGlobalEndLumi(), and stoppedLookingForFile().

◆ ct_

std::unique_ptr<ConcurrencyTracker> evf::FastMonitoringService::ct_
private

Definition at line 247 of file FastMonitoringService.h.

Referenced by doSnapshot(), preallocate(), and preBeginJob().

◆ daqInputSource_

DAQSource* evf::FastMonitoringService::daqInputSource_ = nullptr
private

Definition at line 251 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi(), and setInputSource().

◆ exception_detected_

bool evf::FastMonitoringService::exception_detected_ = false
private

◆ exceptionInLS_

std::vector<unsigned int> evf::FastMonitoringService::exceptionInLS_
private

◆ fastMicrostateDefPath_

std::string evf::FastMonitoringService::fastMicrostateDefPath_
private

Definition at line 263 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preBeginJob().

◆ fastMonIntervals_

unsigned int evf::FastMonitoringService::fastMonIntervals_
private

Definition at line 261 of file FastMonitoringService.h.

Referenced by snapshotRunner().

◆ fastName_

std::string evf::FastMonitoringService::fastName_
private

Definition at line 264 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ fastPath_

std::string evf::FastMonitoringService::fastPath_
private

Definition at line 264 of file FastMonitoringService.h.

Referenced by preBeginJob(), and snapshotRunner().

◆ fileLookStart_

timeval evf::FastMonitoringService::fileLookStart_
private

Definition at line 269 of file FastMonitoringService.h.

Referenced by startedLookingForFile(), and stoppedLookingForFile().

◆ fileLookStop_

timeval evf::FastMonitoringService::fileLookStop_
private

Definition at line 269 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

◆ filesProcessedDuringLumi_

std::map<unsigned int, unsigned int> evf::FastMonitoringService::filesProcessedDuringLumi_
private

Definition at line 278 of file FastMonitoringService.h.

Referenced by accumulateFileSize(), doSnapshot(), and postGlobalEndLumi().

◆ fmt_

std::unique_ptr<FastMonitoringThread> evf::FastMonitoringService::fmt_
private

◆ has_data_exception_

std::atomic<bool> evf::FastMonitoringService::has_data_exception_ = false
private

◆ has_source_exception_

std::atomic<bool> evf::FastMonitoringService::has_source_exception_ = false
private

◆ inputLegendFileJson_

std::string evf::FastMonitoringService::inputLegendFileJson_
private

Definition at line 299 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preBeginJob().

◆ inputSource_

FedRawDataInputSource* evf::FastMonitoringService::inputSource_ = nullptr
private

Definition at line 250 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi(), and setInputSource().

◆ inputState_

std::atomic<FastMonState::InputState> evf::FastMonitoringService::inputState_ {FastMonState::InputState::inInit}
private

Definition at line 252 of file FastMonitoringService.h.

Referenced by doSnapshot(), setInState(), and snapshotRunner().

◆ inputStateNames

const std::string evf::FastMonitoringService::inputStateNames
static

Definition at line 172 of file FastMonitoringService.h.

Referenced by makeInputLegendaJson(), and snapshotRunner().

◆ inputSupervisorState_

std::atomic<FastMonState::InputState> evf::FastMonitoringService::inputSupervisorState_ {FastMonState::InputState::inInit}
private

Definition at line 253 of file FastMonitoringService.h.

Referenced by doSnapshot(), setInStateSup(), and snapshotRunner().

◆ isInitTransition_

std::atomic<bool> evf::FastMonitoringService::isInitTransition_
private

Definition at line 272 of file FastMonitoringService.h.

Referenced by doSnapshot(), postGlobalBeginRun(), and preBeginJob().

◆ lastGlobalLumi_

unsigned int evf::FastMonitoringService::lastGlobalLumi_
private

Definition at line 271 of file FastMonitoringService.h.

Referenced by preBeginJob(), preGlobalBeginLumi(), and snapshotRunner().

◆ leadTimes_

std::vector<double> evf::FastMonitoringService::leadTimes_
private

Definition at line 281 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

◆ lockStatsDuringLumi_

std::map<unsigned int, std::pair<double, unsigned int> > evf::FastMonitoringService::lockStatsDuringLumi_
private

Definition at line 282 of file FastMonitoringService.h.

Referenced by doSnapshot(), postGlobalEndLumi(), and reportLockWait().

◆ lumiFromSource_

unsigned int evf::FastMonitoringService::lumiFromSource_
private

Definition at line 273 of file FastMonitoringService.h.

Referenced by preBeginJob(), and stoppedLookingForFile().

◆ lumiStartTime_

std::map<unsigned int, timeval> evf::FastMonitoringService::lumiStartTime_
private

Definition at line 268 of file FastMonitoringService.h.

Referenced by preGlobalBeginLumi(), and preGlobalEndLumi().

◆ macroStateNames

const std::string evf::FastMonitoringService::macroStateNames
static
Initial value:
= {"Init",
"JobReady",
"RunGiven",
"Running",
"Stopping",
"Done",
"JobEnded",
"Error",
"ErrorEnded",
"End",
"Invalid"}

Definition at line 171 of file FastMonitoringService.h.

◆ microstate_

std::vector<ContainableAtomic<const void*> > evf::FastMonitoringService::microstate_
private

◆ microstateAcqFlag_

std::vector<ContainableAtomic<unsigned char> > evf::FastMonitoringService::microstateAcqFlag_
private

◆ microstateDefPath_

std::string evf::FastMonitoringService::microstateDefPath_
private

Definition at line 263 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preBeginJob().

◆ moduleLegendFile_

std::string evf::FastMonitoringService::moduleLegendFile_
private

Definition at line 297 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ moduleLegendFileJson_

std::string evf::FastMonitoringService::moduleLegendFileJson_
private

Definition at line 298 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preBeginJob().

◆ monInit_

std::atomic<bool> evf::FastMonitoringService::monInit_
private

Definition at line 302 of file FastMonitoringService.h.

Referenced by preBeginJob(), and snapshotRunner().

◆ nMonThreads_

unsigned int evf::FastMonitoringService::nMonThreads_ = 0
private

Definition at line 256 of file FastMonitoringService.h.

Referenced by doSnapshot(), preallocate(), preBeginJob(), and snapshotRunner().

◆ nOutputModules_

unsigned int evf::FastMonitoringService::nOutputModules_ = 0
private

Definition at line 300 of file FastMonitoringService.h.

Referenced by makeModuleLegendaJson(), and preModuleBeginJob().

◆ nStreams_

unsigned int evf::FastMonitoringService::nStreams_ = 0
private

Definition at line 255 of file FastMonitoringService.h.

Referenced by doSnapshot(), preallocate(), and preBeginJob().

◆ nThreads_

unsigned int evf::FastMonitoringService::nThreads_ = 0
private

Definition at line 257 of file FastMonitoringService.h.

Referenced by doSnapshot(), preallocate(), preBeginJob(), and preGlobalBeginLumi().

◆ processedEventsPerLumi_

std::map<unsigned int, std::pair<unsigned int, bool> > evf::FastMonitoringService::processedEventsPerLumi_
private

◆ runDirectory_

std::filesystem::path evf::FastMonitoringService::runDirectory_
private

Definition at line 291 of file FastMonitoringService.h.

Referenced by getRunDirName(), and preBeginJob().

◆ sleepTime_

int evf::FastMonitoringService::sleepTime_
private

Definition at line 260 of file FastMonitoringService.h.

Referenced by preBeginJob(), and snapshotRunner().

◆ snapCounter_

unsigned int evf::FastMonitoringService::snapCounter_ = 0
private

Definition at line 262 of file FastMonitoringService.h.

Referenced by snapshotRunner().

◆ specialMicroStateNames

const edm::ModuleDescription evf::FastMonitoringService::specialMicroStateNames
static
Initial value:
= {
edm::ModuleDescription("Dummy", "Invalid"),
edm::ModuleDescription("Dummy", "Idle"),
edm::ModuleDescription("Dummy", "FwkOvhSrc"),
edm::ModuleDescription("Dummy", "FwkOvhMod"),
edm::ModuleDescription("Dummy", "FwkEoL"),
edm::ModuleDescription("Dummy", "Input"),
edm::ModuleDescription("Dummy", "DQM"),
edm::ModuleDescription("Dummy", "BoL"),
edm::ModuleDescription("Dummy", "EoL"),
edm::ModuleDescription("Dummy", "GlobalEoL"),
edm::ModuleDescription("Dummy", "Fwk"),
edm::ModuleDescription("Dummy", "IdleSource"),
edm::ModuleDescription("Dummy", "Event"),
edm::ModuleDescription("Dummy", "Ignore")}

Definition at line 170 of file FastMonitoringService.h.

Referenced by preBeginJob(), and setTMicrostate().

◆ streamCounterUpdating_

std::vector<std::atomic<bool>*> evf::FastMonitoringService::streamCounterUpdating_
private

Definition at line 289 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ tbbConcurrencyTracker_

bool evf::FastMonitoringService::tbbConcurrencyTracker_
private

Definition at line 259 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ tbbMonitoringMode_

bool evf::FastMonitoringService::tbbMonitoringMode_
private

◆ threadIDAvailable_

bool evf::FastMonitoringService::threadIDAvailable_ = false
private

Definition at line 293 of file FastMonitoringService.h.

◆ tmicrostate_

std::vector<ContainableAtomic<const void*> > evf::FastMonitoringService::tmicrostate_
private

◆ tmicrostateAcqFlag_

std::vector<ContainableAtomic<unsigned char> > evf::FastMonitoringService::tmicrostateAcqFlag_
private

◆ totalEventsProcessed_

std::atomic<unsigned long> evf::FastMonitoringService::totalEventsProcessed_
private

Definition at line 295 of file FastMonitoringService.h.

Referenced by postEvent().

◆ verbose_

bool evf::FastMonitoringService::verbose_ = false
private

Definition at line 315 of file FastMonitoringService.h.

Referenced by snapshotRunner().

◆ workingDirectory_

std::filesystem::path evf::FastMonitoringService::workingDirectory_
private

Definition at line 291 of file FastMonitoringService.h.

Referenced by preBeginJob().