---

edm::ServiceManager Class Reference

#include <EventFilter/StorageManager/interface/ServiceManager.h>

List of all members.

Public Member Functions
void	closeFilesIfNeeded ()
std::list< std::string > &	get_currfiles ()
std::list< std::string > &	get_filelist ()
boost::shared_ptr < stor::SMOnlyStats >	get_stats ()
std::vector< uint32 > &	get_storedEvents ()
std::vector< std::string > &	get_storedNames ()
std::map< std::string, Strings >	getStreamSelectionTable ()
	Returns a map of the trigger selection strings for each output stream.
void	manageErrorEventMsg (std::string catalog, uint32 disks, std::string sourceId, FRDEventMsgView &msg)
void	manageEventMsg (EventMsgView &msg)
void	manageInitMsg (std::string catalog, uint32 disks, std::string sourceId, InitMsgView &init_message, stor::InitMsgCollection &initMsgCollection)
	ServiceManager (const std::string &config)
void	start ()
void	stop ()
	~ServiceManager ()
Private Member Functions
void	collectStreamerPSets (const std::string &config)
Private Attributes
int	currentlumi_
std::list< std::string >	currfiles_
bool	errorStreamCreated_
int	errorStreamPSetIndex_
std::list< std::string >	filelist_
double	lasttimechecked_
Streams	managedOutputs_
std::vector< ParameterSet >	outModPSets_
std::vector< uint32 >	outputModuleIds_
unsigned long	period4samples_
stor::SMPerformanceMeter *	pmeter_
Strings	psetHLTOutputLabels_
unsigned long	samples_
std::vector< uint32 >	storedEvents_
std::vector< std::string >	storedNames_
double	timeouttime_

---

Detailed Description

Definition at line 29 of file ServiceManager.h.

---

Constructor & Destructor Documentation

ServiceManager::ServiceManager

(

const std::string &

config

)

[explicit]

Definition at line 17 of file ServiceManager.cc.

References collectStreamerPSets(), stor::SMPerformanceMeter::init(), period4samples_, pmeter_, samples_, and storedNames_.

                                                     :
  outModPSets_(0),
  managedOutputs_(0),
  psetHLTOutputLabels_(0),
  outputModuleIds_(0),
  storedEvents_(0),
  currentlumi_(0),
  timeouttime_(0),
  lasttimechecked_(0),
  errorStreamPSetIndex_(-1),
  errorStreamCreated_(false),
  samples_(1000),
  period4samples_(10)
{
  storedNames_.clear();
  collectStreamerPSets(config);
  pmeter_ = new stor::SMPerformanceMeter();
  pmeter_->init(samples_, period4samples_);
} 

ServiceManager::~ServiceManager

(

)

Definition at line 38 of file ServiceManager.cc.

References managedOutputs_, outputModuleIds_, pmeter_, storedEvents_, and storedNames_.

{ 
  managedOutputs_.clear();
  outputModuleIds_.clear();
  storedEvents_.clear();
  storedNames_.clear();
  delete pmeter_;
}

---

Member Function Documentation

void ServiceManager::closeFilesIfNeeded

(

)

Definition at line 257 of file ServiceManager.cc.

References it, and managedOutputs_.

{
  StreamsIterator itBeg = managedOutputs_.begin();
  StreamsIterator itEnd = managedOutputs_.end();
  for(StreamsIterator it = itBeg; it != itEnd; ++it) {
    (*it)->closeTimedOutFiles();
  }
}

void ServiceManager::collectStreamerPSets

(

const std::string &

config

)

[private]

Definition at line 332 of file ServiceManager.cc.

References TestMuL1L2Filter_cff::cerr, e, edm::ParameterSet::empty(), lat::endl(), errorStreamPSetIndex_, Exception, cms::Exception::explainSelf(), edm::ParameterSet::getParameter(), i, it, outModPSets_, psetHLTOutputLabels_, and cms::Exception::what().

Referenced by ServiceManager().

{

     try{
       
       PythonProcessDesc py_pdesc(config.c_str());
       boost::shared_ptr<ProcessDesc> pdesc = py_pdesc.processDesc();

       boost::shared_ptr<ParameterSet> procPset = pdesc->getProcessPSet();
       
        ParameterSet allTrigPaths = procPset->
         getUntrackedParameter<ParameterSet>("@trigger_paths");
       
       if (allTrigPaths.empty())
         throw cms::Exception("collectStreamerPSets","ServiceManager")
           << "No Trigger or End Path Found in the Config File" <<endl;
       
       std::vector<std::string> allEndPaths = 
         procPset->getParameter<std::vector<std::string> >("@end_paths");
       
       if (allEndPaths.empty())
         throw cms::Exception("collectStreamerPSets","ServiceManager")
           << "No End Path Found in the Config File" <<endl;
       
       for(std::vector<std::string>::iterator it = allEndPaths.begin(), itEnd = allEndPaths.end();
           it != itEnd;
           ++it) {
           std::vector<std::string> anEndPath = procPset->getParameter<std::vector<std::string> >((*it));
           for(std::vector<std::string>::iterator i = anEndPath.begin(), iEnd = anEndPath.end();
               i != iEnd; ++i) {
               ParameterSet aModInEndPathPset = 
                 procPset->getParameter<ParameterSet>((*i));
               if (aModInEndPathPset.empty())
                 throw cms::Exception("collectStreamerPSets","ServiceManager")
                   << "Empty End Path Found in the Config File" <<endl;
              
               std::string mod_type = aModInEndPathPset.getParameter<std::string> ("@module_type");
               if (mod_type == "EventStreamFileWriter") {
                 outModPSets_.push_back(aModInEndPathPset);
                 psetHLTOutputLabels_.push_back(std::string());  // empty string
               }
               else if (mod_type == "ErrorStreamFileWriter" ||
                        mod_type == "FRDStreamFileWriter") {
                 errorStreamPSetIndex_ = outModPSets_.size();
                 outModPSets_.push_back(aModInEndPathPset);
                 psetHLTOutputLabels_.push_back(std::string());  // empty string
               }
           }
       }
     } catch (cms::Exception & e) {
       std::cerr << "cms::Exception: " << e.explainSelf() << std::endl;
       std::cerr << "std::Exception: " << e.what() << std::endl;
       throw cms::Exception("collectStreamerPSets") << e.explainSelf() << std::endl;
     }
}

std::list< std::string > & ServiceManager::get_currfiles

(

)

Definition at line 285 of file ServiceManager.cc.

References currfiles_, it, and managedOutputs_.

{ 
  currfiles_.clear();
  for(StreamsIterator it = managedOutputs_.begin(), itEnd = managedOutputs_.end();
      it != itEnd; ++it) {
      std::list<std::string> sub_list = (*it)->getCurrentFileList();
      if(sub_list.size() > 0)
        currfiles_.insert(currfiles_.end(), sub_list.begin(), sub_list.end());
  }
  return currfiles_;  
}

std::list< std::string > & ServiceManager::get_filelist

(

)

Definition at line 269 of file ServiceManager.cc.

References filelist_, it, and managedOutputs_.

{ 
  filelist_.clear();
  for(StreamsIterator it = managedOutputs_.begin(), itEnd = managedOutputs_.end();
      it != itEnd; ++it) {
      std::list<std::string> sub_list = (*it)->getFileList();
      if(sub_list.size() > 0)
        filelist_.insert(filelist_.end(), sub_list.begin(), sub_list.end());
  } 
  return filelist_; 
}

boost::shared_ptr< stor::SMOnlyStats > ServiceManager::get_stats

(

)

Definition at line 388 of file ServiceManager.cc.

References stor::BaseCounter::getCurrentTime(), stor::SMPerformanceMeter::getStats(), stor::SMPerfStats::longTermCounter_, stor::SMPerfStats::maxBandwidth2_, stor::SMPerfStats::maxBandwidth_, stor::SMPerfStats::minBandwidth2_, stor::SMPerfStats::minBandwidth_, period4samples_, pmeter_, samples_, stor::SMPerfStats::shortPeriodCounter_, stor::SMPerfStats::shortTermCounter_, and stor::SMPerformanceMeter::totalvolumemb().

{ 
// Copy measurements for a different thread potentially
// TODO create each time or use a data member?
    boost::shared_ptr<stor::SMOnlyStats> outstats(new stor::SMOnlyStats() );

    if ( pmeter_->getStats().shortTermCounter_->hasValidResult() )
    {
      stor::SMPerfStats stats = pmeter_->getStats();

      outstats->instantBandwidth_= stats.shortTermCounter_->getValueRate();
      outstats->instantRate_     = stats.shortTermCounter_->getSampleRate();
      outstats->instantLatency_  = 1000000.0 / outstats->instantRate_;

      double now = stor::ForeverCounter::getCurrentTime();
      outstats->totalSamples_    = stats.longTermCounter_->getSampleCount();
      outstats->duration_        = stats.longTermCounter_->getDuration(now);
      outstats->meanBandwidth_   = stats.longTermCounter_->getValueRate(now);
      outstats->meanRate_        = stats.longTermCounter_->getSampleRate(now);
      outstats->meanLatency_     = 1000000.0 / outstats->meanRate_;

      outstats->maxBandwidth_    = stats.maxBandwidth_;
      outstats->minBandwidth_    = stats.minBandwidth_;
    }

    // for time period bandwidth performance measurements
    if ( pmeter_->getStats().shortPeriodCounter_->hasValidResult() )
    {
      stor::SMPerfStats stats = pmeter_->getStats();

      outstats->instantBandwidth2_= stats.shortPeriodCounter_->getValueRate();
      outstats->instantRate2_     = stats.shortPeriodCounter_->getSampleRate();
      outstats->instantLatency2_  = 1000000.0 / outstats->instantRate2_;

      double now = stor::ForeverCounter::getCurrentTime();
      outstats->totalSamples2_    = stats.longTermCounter_->getSampleCount();
      outstats->duration2_        = stats.longTermCounter_->getDuration(now);
      outstats->meanBandwidth2_   = stats.longTermCounter_->getValueRate(now);
      outstats->meanRate2_        = stats.longTermCounter_->getSampleRate(now);
      outstats->meanLatency2_     = 1000000.0 / outstats->meanRate2_;

      outstats->maxBandwidth2_    = stats.maxBandwidth2_;
      outstats->minBandwidth2_    = stats.minBandwidth2_;
    }
    outstats->receivedVolume_ = pmeter_->totalvolumemb();
    outstats->samples_ = samples_;
    outstats->period4samples_ = period4samples_;
    return outstats;
}

std::vector< uint32 > & ServiceManager::get_storedEvents

(

)

Definition at line 298 of file ServiceManager.cc.

References storedEvents_.

{ 
  return storedEvents_;  
}

std::vector< std::string > & ServiceManager::get_storedNames

(

)

Definition at line 302 of file ServiceManager.cc.

References storedNames_.

{ 
  return storedNames_;  
}

std::map< std::string, Strings > ServiceManager::getStreamSelectionTable

(

)

Returns a map of the trigger selection strings for each output stream.

Definition at line 310 of file ServiceManager.cc.

References errorStreamPSetIndex_, edm::EventSelector::getEventSelectionVString(), it, outModPSets_, and python::StorageManager_cfg::streamLabel.

{
  std::map<std::string, Strings> selTable;
  int psetIdx = -1;
  for(std::vector<ParameterSet>::iterator it = outModPSets_.begin();
      it != outModPSets_.end(); ++it) {
    ++psetIdx;
    if (psetIdx == errorStreamPSetIndex_) continue;

    std::string streamLabel = it->getParameter<string> ("streamLabel");
    if (streamLabel.size() > 0) {
      selTable[streamLabel] = EventSelector::getEventSelectionVString(*it);
    }
  }
  return selTable;
}

void ServiceManager::manageErrorEventMsg	(	std::string	catalog,
		uint32	disks,
		std::string	sourceId,
		FRDEventMsgView &	msg
	)

Definition at line 205 of file ServiceManager.cc.

References GenMuonPlsPt100GeV_cfg::cout, errorStreamCreated_, errorStreamPSetIndex_, aod_PYTHIA_cfg::fileName, stor::Configurator::instance(), managedOutputs_, name, outModPSets_, outputModuleIds_, psetHLTOutputLabels_, FRDEventMsgView::startAddress(), storedEvents_, and storedNames_.

{
  // if no error stream was configured, we can exit early
  if (errorStreamPSetIndex_ < 0) return;

  // create the error stream, if needed
  if (! errorStreamCreated_) {
    ParameterSet errorStreamPSet = outModPSets_.at(errorStreamPSetIndex_);
    boost::shared_ptr<stor::Parameter> smParameter_ = stor::Configurator::instance()->getParameter();

    shared_ptr<StreamService> stream =
      shared_ptr<StreamService>(new FRDStreamService(errorStreamPSet));
    stream->setCatalog(catalog);
    stream->setNumberOfFileSystems(disks);
    stream->setSourceId(sourceId);
    stream->setFileName(smParameter_ -> fileName());
    stream->setFilePath(smParameter_ -> filePath());
    stream->setMaxFileSize(smParameter_ -> maxFileSize());
    stream->setSetupLabel(smParameter_ -> setupLabel());
    stream->setHighWaterMark(smParameter_ -> highWaterMark());
    stream->setLumiSectionTimeOut(smParameter_ -> lumiSectionTimeOut());
    managedOutputs_.push_back(stream);
    outputModuleIds_.push_back(0xffffffff);
    storedEvents_.push_back(0);
    storedNames_.push_back(stream->getStreamLabel());
    stream->report(cout,3);

    psetHLTOutputLabels_[errorStreamPSetIndex_] = "ResourceBroker Error Output";

    errorStreamCreated_ = true;
  }

  // process the event
  int outputIdx = -1;
  for(StreamsIterator strIter = managedOutputs_.begin(), strIterEnd = managedOutputs_.end(); strIter != strIterEnd; ++strIter) {
    ++outputIdx;
    std::string streamClassName = typeid(*(*strIter)).name();
    if (streamClassName.find("FRDStreamService", 0) == string::npos)
      continue;

    bool thisEventAccepted = (*strIter)->nextEvent(msg.startAddress());
    if (!thisEventAccepted)
      continue;

    ++storedEvents_[outputIdx];

    // for now, we don't have any lumi section information in the
    // FRDEvent messages, so we don't try to do any lumi-boundary processing
  }
}

void ServiceManager::manageEventMsg

(

EventMsgView &

msg

)

Definition at line 184 of file ServiceManager.cc.

References stor::SMPerformanceMeter::addSample(), currentlumi_, it, managedOutputs_, EventMsgView::outModId(), outputModuleIds_, pmeter_, EventMsgView::size(), EventMsgView::startAddress(), storedEvents_, and timeouttime_.

{
  int outputIdx = -1;
  for(StreamsIterator it = managedOutputs_.begin(), itEnd = managedOutputs_.end(); it != itEnd; ++it) {
    ++outputIdx;
    if (msg.outModId() != outputModuleIds_[outputIdx])
      continue;

    bool thisEventAccepted = (*it)->nextEvent(msg.startAddress());
    if (!thisEventAccepted)
      continue;

    ++storedEvents_[outputIdx];
    pmeter_->addSample(msg.size());
    if ((*it)->lumiSection() > currentlumi_) {
      currentlumi_ = (*it)->lumiSection();
      timeouttime_ = (*it)->getCurrentTime();
    }
  }
}

void ServiceManager::manageInitMsg	(	std::string	catalog,
		uint32	disks,
		std::string	sourceId,
		InitMsgView &	init_message,
		stor::InitMsgCollection &	initMsgCollection
	)

Definition at line 77 of file ServiceManager.cc.

References GenMuonPlsPt100GeV_cfg::cout, empty, lat::endl(), errorStreamPSetIndex_, Exception, aod_PYTHIA_cfg::fileName, stor::InitMsgCollection::getSelectionHelpString(), stor::Configurator::instance(), it, managedOutputs_, outModPSets_, InitMsgView::outputModuleId(), outputModuleIds_, InitMsgView::outputModuleLabel(), psetHLTOutputLabels_, storedEvents_, and storedNames_.

{
  boost::shared_ptr<stor::Parameter> smParameter_ = stor::Configurator::instance()->getParameter();
  std::string inputOMLabel = view.outputModuleLabel();
  int psetIdx = -1;
  for(std::vector<ParameterSet>::iterator it = outModPSets_.begin(), itEnd = outModPSets_.end();
      it != itEnd; ++it) {
    ++psetIdx;
    bool createStreamNow = false;

    // if this ParameterSet corresponds to the error stream, skip over it
    // since the error stream doesn't use the INIT messages.
    if (psetIdx == errorStreamPSetIndex_) continue;

    // test if this INIT message is the right one for this output module
    // (that is, whether the HLT output module specified in the 
    // SM output module SelectHLTOutput parameter matches the HLT output
    // module in the INIT message)

    // fetch the SelectHLTOutput parameter from the SM output PSet
    std::string requestedOMLabel =
      it->getUntrackedParameter<std::string>("SelectHLTOutput", std::string());

    // if the SM output PSet didn't specify an HLT output module...
    //
    // (Note that the SelectHLTOutput parameter is optional.  If it is not
    // specified, we create the stream using the first INIT message that
    // we receive.  However, if we get multiple, different, INIT messages,
    // we complain loudly.
    // By allowing it to be optional, though, we provide some level
    // of backward compatibility - setups that only have one HLT output
    // module aren't forced to add this parameter.)
    if (requestedOMLabel.empty()) {

      // if we haven't yet created the stream object, go ahead and do
      // that using this INIT message
      if (psetHLTOutputLabels_[psetIdx].empty()) {
        createStreamNow = true;
      }

      // if we already created the stream object and this is a different
      // INIT message than what we used to create it, we need to complain
      // because SM output PSets are required to have a SelectHLTOutput
      // parameter in the presence of multiple INIT messages (multiple
      // HLT output modules)
      else if (inputOMLabel != psetHLTOutputLabels_[psetIdx]) {
        std::string errorString;
        errorString.append("ERROR: The configuration for Stream ");
        errorString.append((*it).getParameter<string> ("streamLabel"));
        errorString.append(" does not specify an HLT output module.\n");
        errorString.append("Please specify one of the HLT output modules ");
        errorString.append("listed below as the SelectHLTOutput parameter ");
        errorString.append("in the EventStreamFileWriter configuration ");
        errorString.append("for Stream ");
        errorString.append((*it).getParameter<string> ("streamLabel"));
        errorString.append(".\n");
        errorString.append(initMsgCollection.getSelectionHelpString());
        errorString.append("\n");
        throw cms::Exception("ServiceManager","manageInitMsg")
          << errorString << std::endl;
      }
    }

    // if the SM output PSet did specify an HLT output module...
    else {

      // if the HLT output module labels match...
      if (inputOMLabel == requestedOMLabel) {

        // if we haven't yet created the stream object, go ahead and do
        // that using this INIT message
        if (psetHLTOutputLabels_[psetIdx].empty()) {
          createStreamNow = true;
        }

        // if we already created the stream object, we could complain,
        // but won't (for now) so that this method can support multiple
        // calls with the same INIT message
        else {}
      }

      // if the HLT output module labels do not match, do nothing
      else {}
    }

    if (createStreamNow) {
      shared_ptr<StreamService> stream = shared_ptr<StreamService>(new EventStreamService((*it),view));
      stream->setCatalog(catalog);
      stream->setNumberOfFileSystems(disks);
      stream->setSourceId(sourceId);
      stream->setFileName(smParameter_ -> fileName());
      stream->setFilePath(smParameter_ -> filePath());
      stream->setMaxFileSize(smParameter_ -> maxFileSize());
      stream->setSetupLabel(smParameter_ -> setupLabel());
      stream->setHighWaterMark(smParameter_ -> highWaterMark());
      stream->setLumiSectionTimeOut(smParameter_ -> lumiSectionTimeOut());
      managedOutputs_.push_back(stream);
      outputModuleIds_.push_back(view.outputModuleId());
      storedEvents_.push_back(0);
      storedNames_.push_back(stream->getStreamLabel());
      stream->report(cout,3);

      psetHLTOutputLabels_[psetIdx] = inputOMLabel;
    }
  }
}

void ServiceManager::start

(

void

)

Definition at line 48 of file ServiceManager.cc.

References currentlumi_, errorStreamCreated_, stor::SMPerformanceMeter::init(), lasttimechecked_, managedOutputs_, outModPSets_, outputModuleIds_, period4samples_, pmeter_, psetHLTOutputLabels_, samples_, storedEvents_, storedNames_, and timeouttime_.

{
  psetHLTOutputLabels_.clear();
  for (unsigned int idx = 0; idx < outModPSets_.size(); idx++) {
    psetHLTOutputLabels_.push_back(std::string());  // empty string
  }

  managedOutputs_.clear();
  outputModuleIds_.clear();
  storedEvents_.clear();
  storedNames_.clear();

  currentlumi_ = 0;
  timeouttime_ = 0;
  lasttimechecked_ = 0;
  errorStreamCreated_ = false;
  pmeter_->init(samples_, period4samples_);
}

void ServiceManager::stop

(

)

Definition at line 68 of file ServiceManager.cc.

References it, and managedOutputs_.

{
  for(StreamsIterator  it = managedOutputs_.begin(), itEnd = managedOutputs_.end();
      it != itEnd; ++it) {
      (*it)->stop();
  }
}

---

Member Data Documentation

int edm::ServiceManager::currentlumi_ [private]

Definition at line 67 of file ServiceManager.h.

Referenced by manageEventMsg(), and start().

std::list<std::string> edm::ServiceManager::currfiles_ [private]

Definition at line 62 of file ServiceManager.h.

Referenced by get_currfiles().

bool edm::ServiceManager::errorStreamCreated_ [private]

Definition at line 71 of file ServiceManager.h.

Referenced by manageErrorEventMsg(), and start().

int edm::ServiceManager::errorStreamPSetIndex_ [private]

Definition at line 70 of file ServiceManager.h.

Referenced by collectStreamerPSets(), getStreamSelectionTable(), manageErrorEventMsg(), and manageInitMsg().

std::list<std::string> edm::ServiceManager::filelist_ [private]

Definition at line 61 of file ServiceManager.h.

Referenced by get_filelist().

double edm::ServiceManager::lasttimechecked_ [private]

Definition at line 69 of file ServiceManager.h.

Referenced by start().

Streams edm::ServiceManager::managedOutputs_ [private]

Definition at line 60 of file ServiceManager.h.

Referenced by closeFilesIfNeeded(), get_currfiles(), get_filelist(), manageErrorEventMsg(), manageEventMsg(), manageInitMsg(), start(), stop(), and ~ServiceManager().

std::vector<ParameterSet> edm::ServiceManager::outModPSets_ [private]

Definition at line 59 of file ServiceManager.h.

Referenced by collectStreamerPSets(), getStreamSelectionTable(), manageErrorEventMsg(), manageInitMsg(), and start().

std::vector<uint32> edm::ServiceManager::outputModuleIds_ [private]

Definition at line 64 of file ServiceManager.h.

Referenced by manageErrorEventMsg(), manageEventMsg(), manageInitMsg(), start(), and ~ServiceManager().

unsigned long edm::ServiceManager::period4samples_ [private]

Definition at line 73 of file ServiceManager.h.

Referenced by get_stats(), ServiceManager(), and start().

stor::SMPerformanceMeter* edm::ServiceManager::pmeter_ [private]

Definition at line 74 of file ServiceManager.h.

Referenced by get_stats(), manageEventMsg(), ServiceManager(), start(), and ~ServiceManager().

Strings edm::ServiceManager::psetHLTOutputLabels_ [private]

Definition at line 63 of file ServiceManager.h.

Referenced by collectStreamerPSets(), manageErrorEventMsg(), manageInitMsg(), and start().

unsigned long edm::ServiceManager::samples_ [private]

Definition at line 72 of file ServiceManager.h.

Referenced by get_stats(), ServiceManager(), and start().

std::vector<uint32> edm::ServiceManager::storedEvents_ [private]

Definition at line 65 of file ServiceManager.h.

Referenced by get_storedEvents(), manageErrorEventMsg(), manageEventMsg(), manageInitMsg(), start(), and ~ServiceManager().

std::vector<std::string> edm::ServiceManager::storedNames_ [private]

Definition at line 66 of file ServiceManager.h.

Referenced by get_storedNames(), manageErrorEventMsg(), manageInitMsg(), ServiceManager(), start(), and ~ServiceManager().

double edm::ServiceManager::timeouttime_ [private]

Definition at line 68 of file ServiceManager.h.

Referenced by manageEventMsg(), and start().

---

The documentation for this class was generated from the following files:

• EventFilter/StorageManager/interface/ServiceManager.h
• EventFilter/StorageManager/src/ServiceManager.cc

---