CMS 3D CMS Logo

 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros Pages
List of all members | Classes | Public Member Functions | Static Public Member Functions | Static Public Attributes | Private Member Functions | Private Attributes
evf::FastMonitoringService Class Reference

#include <FastMonitoringService.h>

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

Classes

struct  Encoding
 

Public Member Functions

void accumulateFileSize (unsigned int lumi, unsigned long fileSize)
 
 FastMonitoringService (const edm::ParameterSet &, edm::ActivityRegistry &)
 
bool getAbortFlagForLumi (unsigned int lumi)
 
unsigned int getEventsProcessedForLumi (unsigned int lumi, bool *abortFlag=0)
 
std::string getRunDirName () const
 
void jobFailure ()
 
std::string makeModuleLegendaJson ()
 
std::string makePathLegendaJson ()
 
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 postSourceEvent (edm::StreamID)
 
void postStreamBeginLumi (edm::StreamContext const &)
 
void postStreamEndLumi (edm::StreamContext const &)
 
void preallocate (edm::service::SystemBounds const &)
 
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 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 setMicroState (MicroStateService::Microstate)
 
void setMicroState (edm::StreamID, MicroStateService::Microstate)
 
bool shouldWriteFiles (unsigned int lumi, unsigned int *proc=0)
 
void startedLookingForFile ()
 
void stoppedLookingForFile (unsigned int lumi)
 
 ~FastMonitoringService ()
 
- Public Member Functions inherited from evf::MicroStateService
virtual std::string getMicroState1 ()
 
virtual std::string const & getMicroState2 ()
 
 MicroStateService (const edm::ParameterSet &, edm::ActivityRegistry &)
 
virtual ~MicroStateService ()
 

Static Public Member Functions

static void fillDescriptions (edm::ConfigurationDescriptions &descriptions)
 

Static Public Attributes

static const std::string macroStateNames [FastMonitoringThread::MCOUNT]
 
static const std::string nopath_ = "NoPath"
 
- Static Public Attributes inherited from evf::MicroStateService
static const edm::ModuleDescription reservedMicroStateNames [mCOUNT]
 

Private Member Functions

void doSnapshot (const unsigned int ls, const bool isGlobalEOL)
 
void doStreamEOLSnapshot (const unsigned int ls, const unsigned int streamID)
 
void dowork ()
 

Private Attributes

std::map< unsigned int,
unsigned long > 
accuSize_
 
std::map< unsigned int, double > avgLeadTime_
 
std::vector< std::atomic< bool > * > collectedPathList_
 
bool emptyLumisectionMode_ = false
 
Encoding encModule_
 
std::vector< EncodingencPath_
 
std::vector< unsigned int > eventCountForPathInit_
 
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::vector< unsigned long > firstEventId_
 
FastMonitoringThread fmt_
 
FedRawDataInputSourceinputSource_ = 0
 
bool isGlobalLumiTransition_
 
unsigned int lastGlobalLumi_
 
std::queue< unsigned int > lastGlobalLumisClosed_
 
std::vector< double > leadTimes_
 
std::map< unsigned int,
std::pair< double, unsigned
int > > 
lockStatsDuringLumi_
 
unsigned int lumiFromSource_
 
std::map< unsigned int, timeval > lumiStartTime_
 
FastMonitoringThread::Macrostate macrostate_
 
std::vector< const void * > microstate_
 
std::string microstateDefPath_
 
std::vector< const void * > ministate_
 
std::string moduleLegendFile_
 
std::string moduleLegendFileJson_
 
std::atomic< bool > monInit_
 
unsigned int nOutputModules_ =0
 
unsigned int nStreams_
 
unsigned int nThreads_
 
std::string pathLegendFile_
 
std::string pathLegendFileJson_
 
bool pathLegendWritten_ = false
 
std::vector< bool > pathNamesReady_
 
std::map< unsigned int,
std::pair< unsigned int, bool > > 
processedEventsPerLumi_
 
boost::filesystem::path runDirectory_
 
int sleepTime_
 
std::string slowName_
 
unsigned int snapCounter_ = 0
 
std::vector< std::atomic< bool > * > streamCounterUpdating_
 
bool threadIDAvailable_ = false
 
std::vector< const void * > threadMicrostate_
 
std::atomic< unsigned long > totalEventsProcessed_
 
boost::filesystem::path workingDirectory_
 

Additional Inherited Members

- Public Types inherited from evf::MicroStateService
enum  Microstate {
  mInvalid = 0, mFwkOvh, mIdle, mInput,
  mInputDone, mDqm, mEoL, mCOUNT
}
 
- Static Protected Attributes inherited from evf::MicroStateService
static const std::string default_return_ ="NotImplemented"
 

Detailed Description

Definition at line 56 of file FastMonitoringService.h.

Constructor & Destructor Documentation

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

Definition at line 41 of file FastMonitoringService.cc.

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

42  :
43  MicroStateService(iPS,reg)
45  ,nStreams_(0)//until initialized
46  ,sleepTime_(iPS.getUntrackedParameter<int>("sleepTime", 1))
47  ,fastMonIntervals_(iPS.getUntrackedParameter<unsigned int>("fastMonIntervals", 2))
48  ,fastName_("fastmoni")
49  ,slowName_("slowmoni")
51  {
52  reg.watchPreallocate(this, &FastMonitoringService::preallocate);//receiving information on number of threads
54 
58 
62 
67 
69 
72 
73  reg.watchPreSourceEvent(this,&FastMonitoringService::preSourceEvent);//source (with streamID of requestor)
75 
78 
82 
83  //find microstate definition path (required by the module)
84  struct stat statbuf;
85  std::string microstateBaseSuffix = "src/EventFilter/Utilities/plugins/microstatedef.jsd";
86  std::string microstatePath = std::string(getenv("CMSSW_BASE")) + "/" + microstateBaseSuffix;
87  if (stat(microstatePath.c_str(), &statbuf)) {
88  microstatePath = std::string(getenv("CMSSW_RELEASE_BASE")) + "/" + microstateBaseSuffix;
89  if (stat(microstatePath.c_str(), &statbuf)) {
90  microstatePath = microstateBaseSuffix;
91  if (stat(microstatePath.c_str(), &statbuf))
92  throw cms::Exception("FastMonitoringService") << "microstate definition file not found";
93  }
94  }
95  fastMicrostateDefPath_ = microstateDefPath_ = microstatePath;
96  }
void prePathEvent(edm::StreamContext const &, edm::PathContext const &)
void watchPreStreamEarlyTermination(PreStreamEarlyTermination::slot_type const &iSlot)
T getUntrackedParameter(std::string const &, T const &) const
void watchPreEvent(PreEvent::slot_type const &iSlot)
void watchPrePathEvent(PrePathEvent::slot_type const &iSlot)
void watchPreallocate(Preallocate::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)
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
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)
static const int nReservedModules
void watchPostGlobalEndLumi(PostGlobalEndLumi::slot_type const &iSlot)
void watchPreModuleBeginJob(PreModuleBeginJob::slot_type const &iSlot)
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 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
evf::FastMonitoringService::~FastMonitoringService ( )

Definition at line 99 of file FastMonitoringService.cc.

100  {
101  }

Member Function Documentation

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

Definition at line 592 of file FastMonitoringService.cc.

References accuSize_, filesProcessedDuringLumi_, fmt_, CommonMethods::lock(), fjr2json::lumi, and evf::FastMonitoringThread::monlock_.

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

592  {
593  std::lock_guard<std::mutex> lock(fmt_.monlock_);
594 
595  if (accuSize_.find(lumi)==accuSize_.end()) accuSize_[lumi] = fileSize;
596  else accuSize_[lumi] += fileSize;
597 
600  else
602  }
std::map< unsigned int, unsigned long > accuSize_
tuple lumi
Definition: fjr2json.py:35
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
void evf::FastMonitoringService::doSnapshot ( const unsigned int  ls,
const bool  isGlobalEOL 
)
private

Definition at line 676 of file FastMonitoringService.cc.

References avgLeadTime_, encModule_, evf::FastMonitoringService::Encoding::encode(), encPath_, evf::FastMonitoringThread::MonitorData::fastAvgLeadTimeJ_, evf::FastMonitoringThread::MonitorData::fastFilesProcessedJ_, evf::FastMonitoringThread::MonitorData::fastLockCountJ_, evf::FastMonitoringThread::MonitorData::fastLockWaitJ_, evf::FastMonitoringThread::MonitorData::fastMacrostateJ_, filesProcessedDuringLumi_, fmt_, i, isGlobalLumiTransition_, evf::FastMonitoringThread::jsonMonitor_, lockStatsDuringLumi_, evf::FastMonitoringThread::m_data, macrostate_, microstate_, evf::FastMonitoringThread::MonitorData::microstateEncoded_, ministate_, evf::FastMonitoringThread::MonitorData::ministateEncoded_, and nStreams_.

Referenced by dowork(), and preGlobalEndLumi().

676  {
677  // update macrostate
679 
680  //update these unless in the midst of a global transition
682 
683  auto itd = avgLeadTime_.find(ls);
684  if (itd != avgLeadTime_.end())
685  fmt_.m_data.fastAvgLeadTimeJ_ = itd->second;
686  else fmt_.m_data.fastAvgLeadTimeJ_=0.;
687 
688  auto iti = filesProcessedDuringLumi_.find(ls);
689  if (iti != filesProcessedDuringLumi_.end())
690  fmt_.m_data.fastFilesProcessedJ_ = iti->second;
692 
693  auto itrd = lockStatsDuringLumi_.find(ls);
694  if (itrd != lockStatsDuringLumi_.end()) {
695  fmt_.m_data.fastLockWaitJ_ = itrd->second.first;
696  fmt_.m_data.fastLockCountJ_ = itrd->second.second;
697  }
698  else {
701  }
702 
703  }
704  else return;
705 
706  //capture latest mini/microstate of streams
707  for (unsigned int i=0;i<nStreams_;i++) {
710  }
711  //for (unsigned int i=0;i<nThreads_;i++)
712  // fmt_.m_data.threadMicrostateEncoded_[i] = encModule_.encode(threadMicrostate_[i]);
713 
714  if (isGlobalEOL)
715  {//only update global variables
716  fmt_.jsonMonitor_->snapGlobal(ls);
717  }
718  else
719  fmt_.jsonMonitor_->snap(ls);
720  }
int i
Definition: DBlmapReader.cc:9
def ls
Definition: eostools.py:348
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
std::map< unsigned int, double > avgLeadTime_
std::unique_ptr< jsoncollector::FastMonitor > jsonMonitor_
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_
std::vector< const void * > microstate_
std::vector< unsigned int > microstateEncoded_
FastMonitoringThread::Macrostate macrostate_
std::vector< const void * > ministate_
std::vector< Encoding > encPath_
std::vector< unsigned int > ministateEncoded_
void evf::FastMonitoringService::doStreamEOLSnapshot ( const unsigned int  ls,
const unsigned int  streamID 
)
inlineprivate

Definition at line 174 of file FastMonitoringService.h.

References fmt_, and evf::FastMonitoringThread::jsonMonitor_.

Referenced by preStreamEndLumi().

174  {
175  //pick up only event count here
176  fmt_.jsonMonitor_->snapStreamAtomic(ls,streamID);
177  }
def ls
Definition: eostools.py:348
std::unique_ptr< jsoncollector::FastMonitor > jsonMonitor_
void evf::FastMonitoringService::dowork ( )
inlineprivate

Definition at line 179 of file FastMonitoringService.h.

References doSnapshot(), encModule_, evf::FastMonitoringService::Encoding::encode(), encPath_, evf::FastMonitoringThread::MonitorData::fastMacrostateJ_, fastMonIntervals_, fastPath_, fmt_, evf::FastMonitoringThread::jsonMonitor_, lastGlobalLumi_, CommonMethods::lock(), evf::FastMonitoringThread::m_data, evf::FastMonitoringThread::m_stoprequest, microstate_, ministate_, monInit_, evf::FastMonitoringThread::monlock_, sleepTime_, snapCounter_, AlCaHLTBitMon_QueryRunRegistry::string, and jsoncollector::IntJ::value().

Referenced by preallocate().

179  { // the function to be called in the thread. Thread completes when function returns.
180  monInit_.exchange(true,std::memory_order_acquire);
181  while (!fmt_.m_stoprequest) {
182  edm::LogInfo("FastMonitoringService") << "Current states: Ms=" << fmt_.m_data.fastMacrostateJ_.value()
183  << " ms=" << encPath_[0].encode(ministate_[0])
184  << " us=" << encModule_.encode(microstate_[0]) << std::endl;
185 
186  {
187  std::lock_guard<std::mutex> lock(fmt_.monlock_);
188 
190 
192  std::string CSV = fmt_.jsonMonitor_->getCSVString();
193  //release mutex before writing out fast path file
194  fmt_.monlock_.unlock();
195  if (CSV.size())
196  fmt_.jsonMonitor_->outputCSV(fastPath_,CSV);
197  }
198 
199  snapCounter_++;
200 
201  }
202  ::sleep(sleepTime_);
203  }
204  }
void doSnapshot(const unsigned int ls, const bool isGlobalEOL)
std::unique_ptr< jsoncollector::FastMonitor > jsonMonitor_
std::vector< const void * > microstate_
std::atomic< bool > m_stoprequest
std::vector< const void * > ministate_
std::vector< Encoding > encPath_
void evf::FastMonitoringService::fillDescriptions ( edm::ConfigurationDescriptions descriptions)
static

Definition at line 103 of file FastMonitoringService.cc.

References edm::ConfigurationDescriptions::add(), edm::ParameterSetDescription::addUntracked(), edm::ParameterSetDescription::setAllowAnything(), and edm::ParameterSetDescription::setComment().

104  {
106  desc.setComment("Service for File-based DAQ monitoring and event accounting");
107  desc.addUntracked<int> ("sleepTime",1)->setComment("Sleep time of the monitoring thread");
108  desc.addUntracked<unsigned int> ("fastMonIntervals",2)->setComment("Modulo of sleepTime intervals on which fastmon file is written out");
109  desc.setAllowAnything();
110  descriptions.add("FastMonitoringService", desc);
111  }
ParameterDescriptionBase * addUntracked(U const &iLabel, T const &value)
void setAllowAnything()
allow any parameter label/value pairs
void setComment(std::string const &value)
void add(std::string const &label, ParameterSetDescription const &psetDescription)
bool evf::FastMonitoringService::getAbortFlagForLumi ( unsigned int  lumi)

Definition at line 662 of file FastMonitoringService.cc.

References Exception, fmt_, CommonMethods::lock(), fjr2json::lumi, evf::FastMonitoringThread::monlock_, and processedEventsPerLumi_.

Referenced by shouldWriteFiles().

662  {
663  std::lock_guard<std::mutex> lock(fmt_.monlock_);
664 
665  auto it = processedEventsPerLumi_.find(lumi);
666  if (it!=processedEventsPerLumi_.end()) {
667  unsigned int abortFlag = it->second.second;
668  return abortFlag;
669  }
670  else {
671  throw cms::Exception("FastMonitoringService") << "output module wants already deleted (or never reported by SOURCE) lumisection status for LUMI -: "<<lumi;
672  return 0;
673  }
674  }
tuple lumi
Definition: fjr2json.py:35
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
unsigned int evf::FastMonitoringService::getEventsProcessedForLumi ( unsigned int  lumi,
bool *  abortFlag = 0 
)

Definition at line 646 of file FastMonitoringService.cc.

References Exception, fmt_, CommonMethods::lock(), fjr2json::lumi, evf::FastMonitoringThread::monlock_, proc, and processedEventsPerLumi_.

Referenced by dqm::DQMFileSaverPB::fillJson(), DQMFileSaver::fillJson(), DQMFileSaver::saveForFilterUnit(), dqm::DQMFileSaverPB::saveLumi(), and shouldWriteFiles().

646  {
647  std::lock_guard<std::mutex> lock(fmt_.monlock_);
648 
649  auto it = processedEventsPerLumi_.find(lumi);
650  if (it!=processedEventsPerLumi_.end()) {
651  unsigned int proc = it->second.first;
652  if (abortFlag) *abortFlag=it->second.second;
653  return proc;
654  }
655  else {
656  throw cms::Exception("FastMonitoringService") << "output module wants already deleted (or never reported by SOURCE) lumisection event count for LUMI -: "<<lumi;
657  return 0;
658  }
659  }
TrainProcessor *const proc
Definition: MVATrainer.cc:101
tuple lumi
Definition: fjr2json.py:35
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
std::string evf::FastMonitoringService::getRunDirName ( ) const
inline

Definition at line 167 of file FastMonitoringService.h.

References runDirectory_.

167 { return runDirectory_.stem().string(); }
boost::filesystem::path runDirectory_
void evf::FastMonitoringService::jobFailure ( )
std::string evf::FastMonitoringService::makeModuleLegendaJson ( )

Definition at line 126 of file FastMonitoringService.cc.

References Json::Value::append(), Json::arrayValue, evf::FastMonitoringService::Encoding::current_, evf::FastMonitoringService::Encoding::decode(), encModule_, i, nOutputModules_, nReservedModules, nSpecialModules, and Json::StyledWriter::write().

Referenced by postBeginJob().

126  {
127  Json::Value legendaVector(Json::arrayValue);
128  for(int i = 0; i < encModule_.current_; i++)
129  legendaVector.append(Json::Value((static_cast<const edm::ModuleDescription *>(encModule_.decode(i)))->moduleLabel()));
130  Json::Value valReserved(nReservedModules);
131  Json::Value valSpecial(nSpecialModules);
132  Json::Value valOutputModules(nOutputModules_);
133  Json::Value moduleLegend;
134  moduleLegend["names"]=legendaVector;
135  moduleLegend["reserved"]=valReserved;
136  moduleLegend["special"]=valSpecial;
137  moduleLegend["output"]=valOutputModules;
138  Json::StyledWriter writer;
139  return writer.write(moduleLegend);
140  }
int i
Definition: DBlmapReader.cc:9
Represents a JSON value.
Definition: value.h:111
static const int nReservedModules
virtual std::string write(const Value &root)
Serialize a Value in JSON format.
static const int nSpecialModules
Writes a Value in JSON format in a human friendly way.
Definition: writer.h:65
const void * decode(unsigned int index)
array value (ordered list)
Definition: value.h:31
std::string evf::FastMonitoringService::makePathLegendaJson ( )

Definition at line 114 of file FastMonitoringService.cc.

References Json::Value::append(), Json::arrayValue, encPath_, i, nReservedPaths, and Json::StyledWriter::write().

Referenced by prePathEvent().

114  {
115  Json::Value legendaVector(Json::arrayValue);
116  for(int i = 0; i < encPath_[0].current_; i++)
117  legendaVector.append(Json::Value(*(static_cast<const std::string *>(encPath_[0].decode(i)))));
118  Json::Value valReserved(nReservedPaths);
119  Json::Value pathLegend;
120  pathLegend["names"]=legendaVector;
121  pathLegend["reserved"]=valReserved;
122  Json::StyledWriter writer;
123  return writer.write(pathLegend);
124  }
int i
Definition: DBlmapReader.cc:9
static const int nReservedPaths
Represents a JSON value.
Definition: value.h:111
virtual std::string write(const Value &root)
Serialize a Value in JSON format.
Writes a Value in JSON format in a human friendly way.
Definition: writer.h:65
std::vector< Encoding > encPath_
array value (ordered list)
Definition: value.h:31
void evf::FastMonitoringService::postBeginJob ( )

Definition at line 320 of file FastMonitoringService.cc.

References encModule_, fmt_, CommonMethods::lock(), evf::FastMonitoringThread::m_data, macrostate_, makeModuleLegendaJson(), evf::FastMonitoringThread::MonitorData::microstateBins_, moduleLegendFileJson_, evf::FastMonitoringThread::monlock_, evf::FastMonitoringThread::sJobReady, AlCaHLTBitMon_QueryRunRegistry::string, and evf::FastMonitoringService::Encoding::vecsize().

Referenced by FastMonitoringService().

321  {
322  std::string && moduleLegStrJson = makeModuleLegendaJson();
323  FileIO::writeStringToFile(moduleLegendFileJson_, moduleLegStrJson);
324 
326 
327  //update number of entries in module histogram
328  std::lock_guard<std::mutex> lock(fmt_.monlock_);
330  }
FastMonitoringThread::Macrostate macrostate_
void evf::FastMonitoringService::postEndJob ( )
void evf::FastMonitoringService::postEvent ( edm::StreamContext const &  sc)

Definition at line 527 of file FastMonitoringService.cc.

References eventCountForPathInit_, evf::FastMonitoringThread::MonitorData::fastPathProcessedJ_, fmt_, evf::FastMonitoringThread::m_data, microstate_, evf::MicroStateService::mIdle, ministate_, nopath_, evf::FastMonitoringThread::MonitorData::processed_, evf::MicroStateService::reservedMicroStateNames, streamCounterUpdating_, edm::StreamContext::streamID(), and totalEventsProcessed_.

Referenced by FastMonitoringService().

528  {
529  microstate_[sc.streamID()] = &reservedMicroStateNames[mIdle];
530 
531  ministate_[sc.streamID()] = &nopath_;
532 
533  #if ATOMIC_LEVEL>=2
534  //use atomic flag to make sure end of lumi sees this
535  streamCounterUpdating_[sc.streamID()]->store(true,std::memory_order_release);
536  fmt_.m_data.processed_[sc.streamID()]->fetch_add(1,std::memory_order_release);
537  streamCounterUpdating_[sc.streamID()]->store(false,std::memory_order_release);
538 
539  #elif ATOMIC_LEVEL==1
540  //writes are atomic, we assume writes propagate to memory before stream EOL snap
541  fmt_.m_data.processed_[sc.streamID()]->fetch_add(1,std::memory_order_relaxed);
542 
543  #elif ATOMIC_LEVEL==0 //default
544  (*(fmt_.m_data.processed_[sc.streamID()]))++;
545  #endif
546  eventCountForPathInit_[sc.streamID()]++;
547 
548  //fast path counter (events accumulated in a run)
549  unsigned long res = totalEventsProcessed_.fetch_add(1,std::memory_order_relaxed);
551  //fmt_.m_data.fastPathProcessedJ_ = totalEventsProcessed_.load(std::memory_order_relaxed);
552  }
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< unsigned int > eventCountForPathInit_
std::vector< std::atomic< bool > * > streamCounterUpdating_
static const std::string nopath_
std::vector< const void * > microstate_
std::atomic< unsigned long > totalEventsProcessed_
std::vector< jsoncollector::AtomicMonUInt * > processed_
std::vector< const void * > ministate_
void evf::FastMonitoringService::postGlobalBeginRun ( edm::GlobalContext const &  gc)
void evf::FastMonitoringService::postGlobalEndLumi ( edm::GlobalContext const &  gc)

Definition at line 444 of file FastMonitoringService.cc.

References lastGlobalLumisClosed_, edm::LuminosityBlockID::luminosityBlock(), and edm::GlobalContext::luminosityBlockID().

Referenced by FastMonitoringService().

445  {
446  //mark closed lumis (still keep map entries until next one)
447  lastGlobalLumisClosed_.push(gc.luminosityBlockID().luminosityBlock());
448  }
std::queue< unsigned int > lastGlobalLumisClosed_
void evf::FastMonitoringService::postModuleEvent ( edm::StreamContext const &  sc,
edm::ModuleCallingContext const &  mcc 
)

Definition at line 569 of file FastMonitoringService.cc.

References evf::MicroStateService::mFwkOvh, microstate_, evf::MicroStateService::reservedMicroStateNames, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

570  {
571  //microstate_[sc.streamID().value()] = (void*)(mcc.moduleDescription());
572  microstate_[sc.streamID().value()] = &reservedMicroStateNames[mFwkOvh];
573  }
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< const void * > microstate_
void evf::FastMonitoringService::postSourceEvent ( edm::StreamID  sid)

Definition at line 559 of file FastMonitoringService.cc.

References evf::MicroStateService::mFwkOvh, microstate_, evf::MicroStateService::reservedMicroStateNames, and edm::StreamID::value().

Referenced by FastMonitoringService().

560  {
562  }
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< const void * > microstate_
unsigned int value() const
Definition: StreamID.h:46
void evf::FastMonitoringService::postStreamBeginLumi ( edm::StreamContext const &  sc)
void evf::FastMonitoringService::postStreamEndLumi ( edm::StreamContext const &  sc)
void evf::FastMonitoringService::preallocate ( edm::service::SystemBounds const &  bounds)

Definition at line 142 of file FastMonitoringService.cc.

References collectedPathList_, evf::FastMonitoringService::Encoding::completeReservedWithDummies(), dowork(), emptyLumisectionMode_, encModule_, encPath_, eventCountForPathInit_, Exception, fastMicrostateDefPath_, fastName_, fastPath_, firstEventId_, fmt_, i, isGlobalLumiTransition_, evf::FastMonitoringThread::jsonMonitor_, lastGlobalLumi_, LogDebug, lumiFromSource_, evf::FastMonitoringThread::m_data, macrostate_, evf::FastMonitoringThread::MonitorData::macrostateBins_, edm::service::SystemBounds::maxNumberOfStreams(), edm::service::SystemBounds::maxNumberOfThreads(), evf::FastMonitoringThread::MCOUNT, evf::MicroStateService::mCOUNT, microstate_, evf::FastMonitoringThread::MonitorData::microstateBins_, microstateDefPath_, ministate_, evf::FastMonitoringThread::MonitorData::ministateBins_, evf::MicroStateService::mInvalid, moduleLegendFile_, moduleLegendFileJson_, monInit_, nopath_, nStreams_, nThreads_, cppFunctionSkipper::operator, fed_dqm_sourceclient-live_cfg::path, pathLegendFile_, pathLegendFileJson_, evf::FastMonitoringThread::MonitorData::registerVariables(), evf::MicroStateService::reservedMicroStateNames, evf::FastMonitoringThread::resetFastMonitor(), runDirectory_, evf::FastMonitoringThread::sInit, evf::FastMonitoringThread::start(), streamCounterUpdating_, threadIDAvailable_, evf::FastMonitoringService::Encoding::updateReserved(), and workingDirectory_.

Referenced by FastMonitoringService().

143  {
144 
145  // FIND RUN DIRECTORY
146  // The run dir should be set via the configuration of EvFDaqDirector
147 
148  if (edm::Service<evf::EvFDaqDirector>().operator->()==nullptr)
149  {
150  throw cms::Exception("FastMonitoringService") << "EvFDaqDirector is not present";
151 
152  }
153  emptyLumisectionMode_ = edm::Service<evf::EvFDaqDirector>()->emptyLumisectionMode();
154  boost::filesystem::path runDirectory(edm::Service<evf::EvFDaqDirector>()->baseRunDir());
155  workingDirectory_ = runDirectory_ = runDirectory;
156  workingDirectory_ /= "mon";
157 
158  if ( !boost::filesystem::is_directory(workingDirectory_)) {
159  LogDebug("FastMonitoringService") << "<MON> DIR NOT FOUND! Trying to create -: " << workingDirectory_.string() ;
160  boost::filesystem::create_directories(workingDirectory_);
161  if ( !boost::filesystem::is_directory(workingDirectory_))
162  edm::LogWarning("FastMonitoringService") << "Unable to create <MON> DIR -: " << workingDirectory_.string()
163  << ". No monitoring data will be written.";
164  }
165 
166  std::ostringstream fastFileName;
167 
168  fastFileName << fastName_ << "_pid" << std::setfill('0') << std::setw(5) << getpid() << ".fast";
170  fast /= fastFileName.str();
171  fastPath_ = fast.string();
172 
173  std::ostringstream moduleLegFile;
174  std::ostringstream moduleLegFileJson;
175  moduleLegFile << "microstatelegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".leg";
176  moduleLegFileJson << "microstatelegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
177  moduleLegendFile_ = (workingDirectory_/moduleLegFile.str()).string();
178  moduleLegendFileJson_ = (workingDirectory_/moduleLegFileJson.str()).string();
179 
180  std::ostringstream pathLegFile;
181  std::ostringstream pathLegFileJson;
182  pathLegFile << "pathlegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".leg";
183  pathLegendFile_ = (workingDirectory_/pathLegFile.str()).string();
184  pathLegFileJson << "pathlegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
185  pathLegendFileJson_ = (workingDirectory_/pathLegFileJson.str()).string();
186 
187  LogDebug("FastMonitoringService") << "Initializing FastMonitor with microstate def path -: "
189  //<< encPath_.current_ + 1 << " " << encModule_.current_ + 1
190 
191  nStreams_=bounds.maxNumberOfStreams();
192  nThreads_=bounds.maxNumberOfThreads();
193 
194  //this should already be >=1
195  if (nStreams_==0) nStreams_=1;
196  if (nThreads_==0) nThreads_=1;
197 
198  /*
199  * initialize the fast monitor with:
200  * vector of pointers to monitorable parameters
201  * path to definition
202  *
203  */
204 
206 
207  for(unsigned int i = 0; i < (mCOUNT); i++)
208  encModule_.updateReserved(static_cast<const void*>(reservedMicroStateNames+i));
210 
211  for (unsigned int i=0;i<nStreams_;i++) {
212  ministate_.push_back(&nopath_);
214 
215  //for synchronization
216  streamCounterUpdating_.push_back(new std::atomic<bool>(0));
217 
218  //path (mini) state
219  encPath_.emplace_back(0);
220  encPath_[i].update(static_cast<const void*>(&nopath_));
221  eventCountForPathInit_.push_back(0);
222  firstEventId_.push_back(0);
223  collectedPathList_.push_back(new std::atomic<bool>(0));
224 
225  }
226  //for (unsigned int i=0;i<nThreads_;i++)
227  // threadMicrostate_.push_back(&reservedMicroStateNames[mInvalid]);
228 
229  //initial size until we detect number of bins
233 
234  lastGlobalLumi_=0;
236  lumiFromSource_=0;
237 
238  //startup monitoring
240  fmt_.jsonMonitor_->setNStreams(nStreams_);
242  monInit_.store(false,std::memory_order_release);
244 
245  //this definition needs: #include "tbb/compat/thread"
246  //however this would results in TBB imeplementation replacing std::thread
247  //(both supposedly call pthread_self())
248  //number of threads created in process could be obtained from /proc,
249  //assuming that all posix threads are true kernel threads capable of running in parallel
250 
251  //#if TBB_IMPLEMENT_CPP0X
253  //threadIDAvailable_=true;
254  //#endif
255 
256  }
#define LogDebug(id)
int i
Definition: DBlmapReader.cc:9
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
boost::filesystem::path runDirectory_
void start(void(FastMonitoringService::*fp)(), FastMonitoringService *cp)
std::vector< unsigned int > eventCountForPathInit_
std::vector< std::atomic< bool > * > streamCounterUpdating_
void registerVariables(jsoncollector::FastMonitor *fm, unsigned int nStreams, unsigned int nThreads)
std::vector< std::atomic< bool > * > collectedPathList_
std::unique_ptr< jsoncollector::FastMonitor > jsonMonitor_
static const std::string nopath_
void resetFastMonitor(std::string const &microStateDefPath, std::string const &fastMicroStateDefPath)
std::vector< unsigned long > firstEventId_
std::vector< const void * > microstate_
FastMonitoringThread::Macrostate macrostate_
boost::filesystem::path workingDirectory_
std::vector< const void * > ministate_
std::vector< Encoding > encPath_
void evf::FastMonitoringService::preEvent ( edm::StreamContext const &  sc)

Definition at line 523 of file FastMonitoringService.cc.

Referenced by FastMonitoringService().

524  {
525  }
void evf::FastMonitoringService::preGlobalBeginLumi ( edm::GlobalContext const &  gc)

Definition at line 343 of file FastMonitoringService.cc.

References accuSize_, avgLeadTime_, filesProcessedDuringLumi_, fmt_, isGlobalLumiTransition_, lastGlobalLumi_, lastGlobalLumisClosed_, CommonMethods::lock(), lockStatsDuringLumi_, edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), lumiStartTime_, evf::FastMonitoringThread::monlock_, and processedEventsPerLumi_.

Referenced by FastMonitoringService().

344  {
345 
346  timeval lumiStartTime;
347  gettimeofday(&lumiStartTime, 0);
348  unsigned int newLumi = gc.luminosityBlockID().luminosityBlock();
349 
350  std::lock_guard<std::mutex> lock(fmt_.monlock_);
351 
352  lumiStartTime_[newLumi]=lumiStartTime;
353  while (!lastGlobalLumisClosed_.empty()) {
354  //wipe out old map entries as they aren't needed and slow down access
355  unsigned int oldLumi = lastGlobalLumisClosed_.back();
357  lumiStartTime_.erase(oldLumi);
358  avgLeadTime_.erase(oldLumi);
359  filesProcessedDuringLumi_.erase(oldLumi);
360  accuSize_.erase(oldLumi);
361  lockStatsDuringLumi_.erase(oldLumi);
362  processedEventsPerLumi_.erase(oldLumi);
363  }
364  lastGlobalLumi_= newLumi;
366  }
std::map< unsigned int, timeval > lumiStartTime_
std::map< unsigned int, unsigned long > accuSize_
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
std::queue< unsigned int > lastGlobalLumisClosed_
std::map< unsigned int, double > avgLeadTime_
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
void evf::FastMonitoringService::preGlobalEarlyTermination ( edm::GlobalContext const &  gc,
edm::TerminationOrigin  to 
)

Definition at line 270 of file FastMonitoringService.cc.

References edm::ExceptionFromAnotherContext, edm::ExceptionFromThisContext, exceptionInLS_, edm::ExternalSignal, fmt_, CommonMethods::lock(), edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), evf::FastMonitoringThread::monlock_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FastMonitoringService().

271  {
272  std::string context;
273  if (to==edm::TerminationOrigin::ExceptionFromThisContext) context = " FromThisContext ";
274  if (to==edm::TerminationOrigin::ExceptionFromAnotherContext) context = " FromAnotherContext";
275  if (to==edm::TerminationOrigin::ExternalSignal) context = " FromExternalSignal";
276  edm::LogWarning("FastMonitoringService") << " GLOBAL " << "earlyTermination -: LS:"
277  << gc.luminosityBlockID().luminosityBlock() << " " << context;
278  std::lock_guard<std::mutex> lock(fmt_.monlock_);
279  exceptionInLS_.push_back(gc.luminosityBlockID().luminosityBlock());
280  }
std::vector< unsigned int > exceptionInLS_
void evf::FastMonitoringService::preGlobalEndLumi ( edm::GlobalContext const &  gc)

Definition at line 368 of file FastMonitoringService.cc.

References accuSize_, doSnapshot(), Exception, exception_detected_, exceptionInLS_, evf::FastMonitoringThread::MonitorData::fastThroughputJ_, fmt_, FedRawDataInputSource::getEventReport(), inputSource_, isGlobalLumiTransition_, evf::FastMonitoringThread::jsonMonitor_, CommonMethods::lock(), LogDebug, fjr2json::lumi, edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), lumiStartTime_, evf::FastMonitoringThread::m_data, evf::FastMonitoringThread::monlock_, fed_dqm_sourceclient-live_cfg::path, processedEventsPerLumi_, edm::shutdown_flag, slowName_, throughputFactor(), jsoncollector::IntJ::value(), jsoncollector::DoubleJ::value(), and workingDirectory_.

Referenced by FastMonitoringService().

369  {
370  unsigned int lumi = gc.luminosityBlockID().luminosityBlock();
371  LogDebug("FastMonitoringService") << "Lumi ended. Writing JSON information. LUMI -: "
372  << lumi;
373  timeval lumiStopTime;
374  gettimeofday(&lumiStopTime, 0);
375 
376  std::lock_guard<std::mutex> lock(fmt_.monlock_);
377 
378  // Compute throughput
379  timeval stt = lumiStartTime_[lumi];
380  unsigned long usecondsForLumi = (lumiStopTime.tv_sec - stt.tv_sec)*1000000
381  + (lumiStopTime.tv_usec - stt.tv_usec);
382  unsigned long accuSize = accuSize_.find(lumi)==accuSize_.end() ? 0 : accuSize_[lumi];
383  double throughput = throughputFactor()* double(accuSize) / double(usecondsForLumi);
384  //store to registered variable
385  fmt_.m_data.fastThroughputJ_.value() = throughput;
386 
387  //update
388  doSnapshot(lumi,true);
389 
390  // create file name for slow monitoring file
391  std::stringstream slowFileName;
392  slowFileName << slowName_ << "_ls" << std::setfill('0') << std::setw(4)
393  << lumi << "_pid" << std::setfill('0')
394  << std::setw(5) << getpid() << ".jsn";
396  slow /= slowFileName.str();
397 
398  //retrieve one result we need (todo: sanity check if it's found)
399  IntJ *lumiProcessedJptr = dynamic_cast<IntJ*>(fmt_.jsonMonitor_->getMergedIntJForLumi("Processed",lumi));
400  if (!lumiProcessedJptr)
401  throw cms::Exception("FastMonitoringService") << "Internal error: got null pointer from FastMonitor";
402  processedEventsPerLumi_[lumi] = std::pair<unsigned int,bool>(lumiProcessedJptr->value(),false);
403 
404  //checking if exception has been thrown (in case of Global/Stream early termination, for this LS)
405  bool exception_detected = exception_detected_;
406  for (auto ex : exceptionInLS_)
407  if (lumi == ex) exception_detected=true;
408 
409  if (edm::shutdown_flag || exception_detected) {
410  edm::LogInfo("FastMonitoringService") << "Run interrupted. Skip writing EoL information -: "
411  << processedEventsPerLumi_[lumi].first << " events were processed in LUMI " << lumi;
412  //this will prevent output modules from producing json file for possibly incomplete lumi
413  processedEventsPerLumi_[lumi].first=0;
414  processedEventsPerLumi_[lumi].second=true;
415  //disable this exception, so service can be used standalone (will be thrown if output module asks for this information)
416  //throw cms::Exception("FastMonitoringService") << "SOURCE did not send update for lumi block. LUMI -:" << lumi;
417  return;
418 
419  }
420 
421  if (inputSource_) {
422  auto sourceReport = inputSource_->getEventReport(lumi, true);
423  if (sourceReport.first) {
424  if (sourceReport.second!=processedEventsPerLumi_[lumi].first) {
425  throw cms::Exception("FastMonitoringService") << "MISMATCH with SOURCE update. LUMI -: "
426  << lumi
427  << ", events(processed):" << processedEventsPerLumi_[lumi].first
428  << " events(source):" << sourceReport.second;
429  }
430  }
431  }
432  edm::LogInfo("FastMonitoringService") << "Statistics for lumisection -: lumi = " << lumi << " events = "
433  << lumiProcessedJptr->value() << " time = " << usecondsForLumi/1000000
434  << " size = " << accuSize << " thr = " << throughput;
435  delete lumiProcessedJptr;
436 
437  //full global and stream merge&output for this lumi
438  fmt_.jsonMonitor_->outputFullJSON(slow.string(),lumi);//full global and stream merge and JSON write for this lumi
439  fmt_.jsonMonitor_->discardCollected(lumi);//we don't do further updates for this lumi
440 
442  }
#define LogDebug(id)
std::map< unsigned int, timeval > lumiStartTime_
std::pair< bool, unsigned int > getEventReport(unsigned int lumi, bool erase)
double throughputFactor()
std::map< unsigned int, unsigned long > accuSize_
tuple lumi
Definition: fjr2json.py:35
volatile std::atomic< bool > shutdown_flag
void doSnapshot(const unsigned int ls, const bool isGlobalEOL)
std::unique_ptr< jsoncollector::FastMonitor > jsonMonitor_
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
FedRawDataInputSource * inputSource_
boost::filesystem::path workingDirectory_
std::vector< unsigned int > exceptionInLS_
void evf::FastMonitoringService::preModuleBeginJob ( edm::ModuleDescription const &  desc)

Definition at line 304 of file FastMonitoringService.cc.

References encModule_, fmt_, CommonMethods::lock(), edm::ModuleDescription::moduleName(), evf::FastMonitoringThread::monlock_, nOutputModules_, evf::FastMonitoringService::Encoding::update(), and evf::FastMonitoringService::Encoding::updateReserved().

Referenced by FastMonitoringService().

305  {
306  std::lock_guard<std::mutex> lock(fmt_.monlock_);
307  //std::cout << " Pre module Begin Job module: " << desc.moduleName() << std::endl;
308 
309  //build a map of modules keyed by their module description address
310  //here we need to treat output modules in a special way so they can be easily singled out
311  if(desc.moduleName() == "Stream" || desc.moduleName() == "ShmStreamConsumer" || desc.moduleName() == "EvFOutputModule" ||
312  desc.moduleName() == "EventStreamFileWriter" || desc.moduleName() == "PoolOutputModule") {
313  encModule_.updateReserved((void*)&desc);
314  nOutputModules_++;
315  }
316  else
317  encModule_.update((void*)&desc);
318  }
void evf::FastMonitoringService::preModuleEvent ( edm::StreamContext const &  sc,
edm::ModuleCallingContext const &  mcc 
)

Definition at line 564 of file FastMonitoringService.cc.

References microstate_, edm::ModuleCallingContext::moduleDescription(), edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

565  {
566  microstate_[sc.streamID().value()] = (void*)(mcc.moduleDescription());
567  }
std::vector< const void * > microstate_
void evf::FastMonitoringService::prePathEvent ( edm::StreamContext const &  sc,
edm::PathContext const &  pc 
)

Definition at line 490 of file FastMonitoringService.cc.

References collectedPathList_, encPath_, edm::EventID::event(), eventCountForPathInit_, edm::StreamContext::eventID(), firstEventId_, fmt_, CommonMethods::lock(), evf::FastMonitoringThread::m_data, makePathLegendaJson(), ministate_, evf::FastMonitoringThread::MonitorData::ministateBins_, evf::FastMonitoringThread::monlock_, pathLegendFileJson_, pathLegendWritten_, edm::PathContext::pathName(), edm::StreamContext::streamID(), AlCaHLTBitMon_QueryRunRegistry::string, and unlikely.

Referenced by FastMonitoringService().

491  {
492  //make sure that all path names are retrieved before allowing ministate to change
493  //hack: assume memory is synchronized after ~50 events seen by each stream
494  if (unlikely(eventCountForPathInit_[sc.streamID()]<50) && false==collectedPathList_[sc.streamID()]->load(std::memory_order_acquire))
495  {
496  //protection between stream threads, as well as the service monitoring thread
497  std::lock_guard<std::mutex> lock(fmt_.monlock_);
498 
499  if (firstEventId_[sc.streamID()]==0)
500  firstEventId_[sc.streamID()]=sc.eventID().event();
501  if (sc.eventID().event()==firstEventId_[sc.streamID()])
502  {
503  encPath_[sc.streamID()].update((void*)&pc.pathName());
504  return;
505  }
506  else {
507  //finished collecting path names
508  collectedPathList_[sc.streamID()]->store(true,std::memory_order_seq_cst);
509  fmt_.m_data.ministateBins_=encPath_[sc.streamID()].vecsize();
510  if (!pathLegendWritten_) {
511  std::string pathLegendStrJson = makePathLegendaJson();
512  FileIO::writeStringToFile(pathLegendFileJson_, pathLegendStrJson);
513  pathLegendWritten_=true;
514  }
515  }
516  }
517  else {
518  ministate_[sc.streamID()] = &(pc.pathName());
519  }
520  }
std::vector< unsigned int > eventCountForPathInit_
#define unlikely(x)
std::vector< std::atomic< bool > * > collectedPathList_
std::vector< unsigned long > firstEventId_
std::vector< const void * > ministate_
std::vector< Encoding > encPath_
void evf::FastMonitoringService::preSourceEarlyTermination ( edm::TerminationOrigin  to)
void evf::FastMonitoringService::preSourceEvent ( edm::StreamID  sid)

Definition at line 554 of file FastMonitoringService.cc.

References microstate_, evf::MicroStateService::mInput, evf::MicroStateService::reservedMicroStateNames, and edm::StreamID::value().

Referenced by FastMonitoringService().

555  {
557  }
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< const void * > microstate_
unsigned int value() const
Definition: StreamID.h:46
void evf::FastMonitoringService::preStreamBeginLumi ( edm::StreamContext const &  sc)

Definition at line 450 of file FastMonitoringService.cc.

References edm::StreamContext::eventID(), fmt_, CommonMethods::lock(), edm::EventID::luminosityBlock(), evf::FastMonitoringThread::m_data, evf::MicroStateService::mFwkOvh, microstate_, ministate_, evf::FastMonitoringThread::monlock_, nopath_, evf::FastMonitoringThread::MonitorData::processed_, evf::MicroStateService::reservedMicroStateNames, edm::StreamContext::streamID(), evf::FastMonitoringThread::MonitorData::streamLumi_, and edm::StreamID::value().

Referenced by FastMonitoringService().

451  {
452  unsigned int sid = sc.streamID().value();
453  std::lock_guard<std::mutex> lock(fmt_.monlock_);
454  fmt_.m_data.streamLumi_[sid] = sc.eventID().luminosityBlock();
455 
456  //reset collected values for this stream
457  *(fmt_.m_data.processed_[sid])=0;
458 
459  ministate_[sid]=&nopath_;
461  }
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
static const std::string nopath_
std::vector< const void * > microstate_
std::vector< jsoncollector::AtomicMonUInt * > processed_
std::vector< const void * > ministate_
std::vector< unsigned int > streamLumi_
void evf::FastMonitoringService::preStreamEarlyTermination ( edm::StreamContext const &  sc,
edm::TerminationOrigin  to 
)

Definition at line 258 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

259  {
260  std::string context;
261  if (to==edm::TerminationOrigin::ExceptionFromThisContext) context = " FromThisContext ";
262  if (to==edm::TerminationOrigin::ExceptionFromAnotherContext) context = " FromAnotherContext";
263  if (to==edm::TerminationOrigin::ExternalSignal) context = " FromExternalSignal";
264  edm::LogWarning("FastMonitoringService") << " STREAM " << sc.streamID().value() << " earlyTermination -: ID:"<< sc.eventID()
265  << " LS:" << sc.eventID().luminosityBlock() << " " << context;
266  std::lock_guard<std::mutex> lock(fmt_.monlock_);
267  exceptionInLS_.push_back(sc.eventID().luminosityBlock());
268  }
std::vector< unsigned int > exceptionInLS_
void evf::FastMonitoringService::preStreamEndLumi ( edm::StreamContext const &  sc)

Definition at line 468 of file FastMonitoringService.cc.

References doStreamEOLSnapshot(), edm::StreamContext::eventID(), fmt_, load(), CommonMethods::lock(), edm::EventID::luminosityBlock(), evf::MicroStateService::mEoL, microstate_, ministate_, evf::FastMonitoringThread::monlock_, nopath_, evf::MicroStateService::reservedMicroStateNames, streamCounterUpdating_, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

469  {
470  unsigned int sid = sc.streamID().value();
471  std::lock_guard<std::mutex> lock(fmt_.monlock_);
472 
473  #if ATOMIC_LEVEL>=2
474  //spinlock to make sure we are not still updating event counter somewhere
475  while (streamCounterUpdating_[sid]->load(std::memory_order_acquire)) {}
476  #endif
477 
478  //update processed count to be complete at this time
479  doStreamEOLSnapshot(sc.eventID().luminosityBlock(),sid);
480  //reset this in case stream does not get notified of next lumi (we keep processed events only)
481  ministate_[sid]=&nopath_;
483  }
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< std::atomic< bool > * > streamCounterUpdating_
void doStreamEOLSnapshot(const unsigned int ls, const unsigned int streamID)
static const std::string nopath_
std::vector< const void * > microstate_
void load(int perCUT=90)
Definition: getMaxPt.h:59
std::vector< const void * > ministate_
void evf::FastMonitoringService::reportLockWait ( unsigned int  ls,
double  waitTime,
unsigned int  lockCount 
)

Definition at line 638 of file FastMonitoringService.cc.

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

Referenced by FedRawDataInputSource::readSupervisor().

639  {
640  std::lock_guard<std::mutex> lock(fmt_.monlock_);
641  lockStatsDuringLumi_[ls]=std::pair<double,unsigned int>(waitTime,lockCount);
642 
643  }
def ls
Definition: eostools.py:348
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_
void evf::FastMonitoringService::setExceptionDetected ( unsigned int  ls)

Definition at line 293 of file FastMonitoringService.cc.

References exception_detected_, and exceptionInLS_.

Referenced by FedRawDataInputSource::getNextEvent().

293  {
294  if (!ls) exception_detected_=true;
295  else exceptionInLS_.push_back(ls);
296  }
def ls
Definition: eostools.py:348
std::vector< unsigned int > exceptionInLS_
void evf::FastMonitoringService::setInputSource ( FedRawDataInputSource inputSource)
inline

Definition at line 168 of file FastMonitoringService.h.

References inputSource_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

168 {inputSource_=inputSource;}
FedRawDataInputSource * inputSource_
void evf::FastMonitoringService::setMicroState ( MicroStateService::Microstate  m)
virtual

Implements evf::MicroStateService.

Definition at line 579 of file FastMonitoringService.cc.

References i, microstate_, nStreams_, and evf::MicroStateService::reservedMicroStateNames.

580  {
581  for (unsigned int i=0;i<nStreams_;i++)
583  }
int i
Definition: DBlmapReader.cc:9
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< const void * > microstate_
void evf::FastMonitoringService::setMicroState ( edm::StreamID  sid,
MicroStateService::Microstate  m 
)
virtual
bool evf::FastMonitoringService::shouldWriteFiles ( unsigned int  lumi,
unsigned int *  proc = 0 
)
inline

Definition at line 161 of file FastMonitoringService.h.

References emptyLumisectionMode_, getAbortFlagForLumi(), getEventsProcessedForLumi(), and proc.

Referenced by DQMFileSaver::globalEndLuminosityBlock(), and TriggerJSONMonitoring::globalEndLuminosityBlockSummary().

162  {
163  unsigned int processed = getEventsProcessedForLumi(lumi);
164  if (proc) *proc = processed;
165  return !getAbortFlagForLumi(lumi) && (processed || emptyLumisectionMode_);
166  }
unsigned int getEventsProcessedForLumi(unsigned int lumi, bool *abortFlag=0)
TrainProcessor *const proc
Definition: MVATrainer.cc:101
tuple lumi
Definition: fjr2json.py:35
bool getAbortFlagForLumi(unsigned int lumi)
void evf::FastMonitoringService::startedLookingForFile ( )

Definition at line 604 of file FastMonitoringService.cc.

References fileLookStart_.

Referenced by FedRawDataInputSource::readSupervisor().

604  {
605  gettimeofday(&fileLookStart_, 0);
606  /*
607  std::cout << "Started looking for .raw file at: s=" << fileLookStart_.tv_sec << ": ms = "
608  << fileLookStart_.tv_usec / 1000.0 << std::endl;
609  */
610  }
void evf::FastMonitoringService::stoppedLookingForFile ( unsigned int  lumi)

Definition at line 612 of file FastMonitoringService.cc.

References avgLeadTime_, fileLookStart_, fileLookStop_, fmt_, i, leadTimes_, CommonMethods::lock(), fjr2json::lumi, lumiFromSource_, and evf::FastMonitoringThread::monlock_.

Referenced by FedRawDataInputSource::readSupervisor().

612  {
613  gettimeofday(&fileLookStop_, 0);
614  /*
615  std::cout << "Stopped looking for .raw file at: s=" << fileLookStop_.tv_sec << ": ms = "
616  << fileLookStop_.tv_usec / 1000.0 << std::endl;
617  */
618  std::lock_guard<std::mutex> lock(fmt_.monlock_);
619 
620  if (lumi>lumiFromSource_) {
622  leadTimes_.clear();
623  }
624  unsigned long elapsedTime = (fileLookStop_.tv_sec - fileLookStart_.tv_sec) * 1000000 // sec to us
625  + (fileLookStop_.tv_usec - fileLookStart_.tv_usec); // us
626  // add this to lead times for this lumi
627  leadTimes_.push_back((double)elapsedTime);
628 
629  // recompute average lead time for this lumi
630  if (leadTimes_.size() == 1) avgLeadTime_[lumi] = leadTimes_[0];
631  else {
632  double totTime = 0;
633  for (unsigned int i = 0; i < leadTimes_.size(); i++) totTime += leadTimes_[i];
634  avgLeadTime_[lumi] = 0.001*(totTime / leadTimes_.size());
635  }
636  }
int i
Definition: DBlmapReader.cc:9
tuple lumi
Definition: fjr2json.py:35
std::map< unsigned int, double > avgLeadTime_
std::vector< double > leadTimes_

Member Data Documentation

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

Definition at line 240 of file FastMonitoringService.h.

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

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

Definition at line 255 of file FastMonitoringService.h.

Referenced by preallocate(), and prePathEvent().

bool evf::FastMonitoringService::emptyLumisectionMode_ = false
private

Definition at line 275 of file FastMonitoringService.h.

Referenced by preallocate(), and shouldWriteFiles().

Encoding evf::FastMonitoringService::encModule_
private
std::vector<Encoding> evf::FastMonitoringService::encPath_
private
std::vector<unsigned int> evf::FastMonitoringService::eventCountForPathInit_
private

Definition at line 256 of file FastMonitoringService.h.

Referenced by postEvent(), preallocate(), and prePathEvent().

bool evf::FastMonitoringService::exception_detected_ = false
private
std::vector<unsigned int> evf::FastMonitoringService::exceptionInLS_
private
std::string evf::FastMonitoringService::fastMicrostateDefPath_
private

Definition at line 217 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preallocate().

unsigned int evf::FastMonitoringService::fastMonIntervals_
private

Definition at line 215 of file FastMonitoringService.h.

Referenced by dowork().

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

Definition at line 218 of file FastMonitoringService.h.

Referenced by preallocate().

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

Definition at line 218 of file FastMonitoringService.h.

Referenced by dowork(), and preallocate().

timeval evf::FastMonitoringService::fileLookStart_
private

Definition at line 223 of file FastMonitoringService.h.

Referenced by startedLookingForFile(), and stoppedLookingForFile().

timeval evf::FastMonitoringService::fileLookStop_
private

Definition at line 223 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

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

Definition at line 241 of file FastMonitoringService.h.

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

std::vector<unsigned long> evf::FastMonitoringService::firstEventId_
private

Definition at line 254 of file FastMonitoringService.h.

Referenced by preallocate(), and prePathEvent().

FastMonitoringThread evf::FastMonitoringService::fmt_
private
FedRawDataInputSource* evf::FastMonitoringService::inputSource_ = 0
private

Definition at line 210 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi(), and setInputSource().

bool evf::FastMonitoringService::isGlobalLumiTransition_
private
unsigned int evf::FastMonitoringService::lastGlobalLumi_
private

Definition at line 225 of file FastMonitoringService.h.

Referenced by dowork(), preallocate(), and preGlobalBeginLumi().

std::queue<unsigned int> evf::FastMonitoringService::lastGlobalLumisClosed_
private

Definition at line 226 of file FastMonitoringService.h.

Referenced by postGlobalEndLumi(), and preGlobalBeginLumi().

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

Definition at line 244 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

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

Definition at line 245 of file FastMonitoringService.h.

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

unsigned int evf::FastMonitoringService::lumiFromSource_
private

Definition at line 228 of file FastMonitoringService.h.

Referenced by preallocate(), and stoppedLookingForFile().

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

Definition at line 222 of file FastMonitoringService.h.

Referenced by preGlobalBeginLumi(), and preGlobalEndLumi().

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

Definition at line 113 of file FastMonitoringService.h.

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

Definition at line 217 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preallocate().

std::vector<const void*> evf::FastMonitoringService::ministate_
private
std::string evf::FastMonitoringService::moduleLegendFile_
private

Definition at line 265 of file FastMonitoringService.h.

Referenced by preallocate().

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

Definition at line 266 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preallocate().

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

Definition at line 272 of file FastMonitoringService.h.

Referenced by dowork(), and preallocate().

const std::string evf::FastMonitoringService::nopath_ = "NoPath"
static
unsigned int evf::FastMonitoringService::nOutputModules_ =0
private

Definition at line 270 of file FastMonitoringService.h.

Referenced by makeModuleLegendaJson(), and preModuleBeginJob().

unsigned int evf::FastMonitoringService::nStreams_
private

Definition at line 212 of file FastMonitoringService.h.

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

unsigned int evf::FastMonitoringService::nThreads_
private

Definition at line 213 of file FastMonitoringService.h.

Referenced by preallocate().

std::string evf::FastMonitoringService::pathLegendFile_
private

Definition at line 267 of file FastMonitoringService.h.

Referenced by preallocate().

std::string evf::FastMonitoringService::pathLegendFileJson_
private

Definition at line 268 of file FastMonitoringService.h.

Referenced by preallocate(), and prePathEvent().

bool evf::FastMonitoringService::pathLegendWritten_ = false
private

Definition at line 269 of file FastMonitoringService.h.

Referenced by prePathEvent().

std::vector<bool> evf::FastMonitoringService::pathNamesReady_
private

Definition at line 257 of file FastMonitoringService.h.

std::map<unsigned int, std::pair<unsigned int,bool> > evf::FastMonitoringService::processedEventsPerLumi_
private
boost::filesystem::path evf::FastMonitoringService::runDirectory_
private

Definition at line 259 of file FastMonitoringService.h.

Referenced by getRunDirName(), and preallocate().

int evf::FastMonitoringService::sleepTime_
private

Definition at line 214 of file FastMonitoringService.h.

Referenced by dowork().

std::string evf::FastMonitoringService::slowName_
private

Definition at line 218 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi().

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

Definition at line 216 of file FastMonitoringService.h.

Referenced by dowork().

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

Definition at line 252 of file FastMonitoringService.h.

Referenced by postEvent(), preallocate(), and preStreamEndLumi().

bool evf::FastMonitoringService::threadIDAvailable_ = false
private

Definition at line 261 of file FastMonitoringService.h.

Referenced by preallocate().

std::vector<const void*> evf::FastMonitoringService::threadMicrostate_
private

Definition at line 236 of file FastMonitoringService.h.

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

Definition at line 263 of file FastMonitoringService.h.

Referenced by postEvent().

boost::filesystem::path evf::FastMonitoringService::workingDirectory_
private

Definition at line 259 of file FastMonitoringService.h.

Referenced by preallocate(), and preGlobalEndLumi().