evf::FUResourceBroker Class Reference

#include <FUResourceBroker.h>

Inheritance diagram for evf::FUResourceBroker:

Public Member Functions
void	actionPerformed (xdata::Event &e)
bool	configuring (toolbox::task::WorkLoop *wl)
void	connectToBUandSM () throw (evf::Exception)
void	customWebPage (xgi::Input *in, xgi::Output *out) throw (xgi::exception::Exception)
bool	enabling (toolbox::task::WorkLoop *wl)
xoap::MessageReference	fsmCallback (xoap::MessageReference msg) throw (xoap::exception::Exception)
	FUResourceBroker (xdaq::ApplicationStub *s)
bool	halting (toolbox::task::WorkLoop *wl)
void	I2O_EVM_LUMISECTION_Callback (toolbox::mem::Reference *bufRef)
void	I2O_FU_DATA_DISCARD_Callback (toolbox::mem::Reference *bufRef)
void	I2O_FU_DQM_DISCARD_Callback (toolbox::mem::Reference *bufRef)
void	I2O_FU_TAKE_Callback (toolbox::mem::Reference *bufRef)
unsigned int	instanceNumber () const
bool	monitoring (toolbox::task::WorkLoop *wl)
void	startMonitoringWorkLoop () throw (evf::Exception)
void	startWatchingWorkLoop () throw (evf::Exception)
bool	stopping (toolbox::task::WorkLoop *wl)
bool	watching (toolbox::task::WorkLoop *wl)
void	webPageRequest (xgi::Input *in, xgi::Output *out) throw (xgi::exception::Exception)
	XDAQ_INSTANTIATOR ()
virtual	~FUResourceBroker ()

Static Public Attributes
static const int	CRC_ERROR_SHIFT = 0x0
static const int	DATA_ERROR_SHIFT = 0x1
static const int	LOST_ERROR_SHIFT = 0x2
static const int	SENT_ERREVENT_ERROR_SHIFT = 0x5
static const int	TIMEOUT_EVENT_ERROR_SHIFT = 0x4
static const int	TIMEOUT_NOEVENT_ERROR_SHIFT = 0x3

Private Member Functions
void	configureResources ()
double	deltaT (const struct timeval *start, const struct timeval *end)
void	emergencyStop ()
void	exportParameters ()
void	lock ()
void	reset ()
void	unlock ()

Private Attributes
toolbox::task::ActionSignature *	asMonitoring_
toolbox::task::ActionSignature *	asWatching_
xdata::Double	average_
BUProxy *	bu_
xdata::String	buClassName_
xdata::UnsignedInteger32	buInstance_
xdata::String	class_
xdata::String	clientPrcIds_
xdata::UnsignedInteger32	dataErrorFlag_
xdata::UnsignedInteger32	deltaN_
xdata::UnsignedInteger32	deltaSumOfSizes_
xdata::Double	deltaSumOfSquares_
xdata::Double	deltaT_
xdata::UnsignedInteger32	doCrcCheck_
xdata::Boolean	doDropEvents_
xdata::UnsignedInteger32	doDumpEvents_
xdata::Boolean	doFedIdCheck_
xdata::UnsignedInteger32	dqmCellSize_
EvffedFillerRB *	frb_
evf::StateMachine	fsm_
WebGUI *	gui_
toolbox::mem::Pool *	i2oPool_
xdata::UnsignedInteger32	instance_
toolbox::BSem	lock_
Logger	log_
xdata::UnsignedInteger32	monSleepSec_
struct timeval	monStartTime_
xdata::UnsignedInteger32	nbAllocatedEvents_
xdata::UnsignedInteger32	nbAllocateSent_
xdata::UnsignedInteger32	nbClients_
xdata::UnsignedInteger32	nbCrcErrors_
xdata::UnsignedInteger32	nbDataDiscardReceived_
xdata::UnsignedInteger32	nbDataErrors_
xdata::UnsignedInteger32	nbDiscardedEvents_
xdata::UnsignedInteger32	nbDqmCells_
xdata::UnsignedInteger32	nbDqmDiscardReceived_
xdata::UnsignedInteger32	nbLostEvents_
xdata::UnsignedInteger32	nbPendingRequests_
xdata::UnsignedInteger32	nbPendingSMDiscards_
xdata::UnsignedInteger32	nbPendingSMDqmDiscards_
xdata::UnsignedInteger32	nbProcessedEvents_
xdata::UnsignedInteger32	nbRawCells_
xdata::UnsignedInteger32	nbReceivedEvents_
xdata::UnsignedInteger32	nbRecoCells_
xdata::UnsignedInteger32	nbSentDqmEvents_
xdata::UnsignedInteger32	nbSentErrorEvents_
xdata::UnsignedInteger32	nbSentEvents_
UInt_t	nbSentLast_
xdata::UnsignedInteger32	nbTakeReceived_
xdata::UnsignedInteger32	nbTimeoutsWithEvent_
xdata::UnsignedInteger32	nbTimeoutsWithoutEvent_
xdata::Boolean	processKillerEnabled_
xdata::Double	rate_
xdata::UnsignedInteger32	rawCellSize_
xdata::String	reasonForFailed_
xdata::UnsignedInteger32	recoCellSize_
FUResourceTable *	resourceTable_
xdata::Double	rms_
xdata::UnsignedInteger32	runNumber_
xdata::Boolean	segmentationMode_
bool	shmInconsistent_
xdata::UnsignedInteger32	shmResourceTableTimeout_
SMProxy *	sm_
xdata::String	smClassName_
xdata::UnsignedInteger32	smInstance_
std::string	sourceId_
UInt_t	sumOfSizesLast_
uint64_t	sumOfSquaresLast_
xdata::Double	throughput_
xdata::UnsignedInteger32	timeOutSec_
xdata::String	url_
xdata::Boolean	useEvmBoard_
xdata::UnsignedInteger32	watchSleepSec_
toolbox::task::WorkLoop *	wlMonitoring_
toolbox::task::WorkLoop *	wlWatching_

Friends
class	evf::EvffedFillerRB

Detailed Description

Definition at line 40 of file FUResourceBroker.h.

Constructor & Destructor Documentation

FUResourceBroker::FUResourceBroker

(

xdaq::ApplicationStub *

s

)

Definition at line 56 of file FUResourceBroker.cc.

References class_, customWebPage(), ExpressReco_HICollisions_FallBack::e, edm::hlt::Exception, exportParameters(), evf::StateMachine::findRcmsStateListener(), fsm_, gui_, I2O_EVM_LUMISECTION_Callback(), I2O_FU_DATA_DISCARD, I2O_FU_DATA_DISCARD_Callback(), I2O_FU_DQM_DISCARD, I2O_FU_DQM_DISCARD_Callback(), I2O_FU_TAKE_Callback(), i2oPool_, evf::StateMachine::initialize(), instance_, log_, AlCaRecoCosmics_cfg::name, sourceId_, url_, and webPageRequest().

  : xdaq::Application(s)
  , fsm_(this)
  , gui_(0)
  , log_(getApplicationLogger())
  , bu_(0)
  , sm_(0)
  , i2oPool_(0)
  , resourceTable_(0)
  , wlMonitoring_(0)
  , asMonitoring_(0)
  , wlWatching_(0)
  , asWatching_(0)
  , instance_(0)
  , runNumber_(0)
  , deltaT_(0.0)
  , deltaN_(0)
  , deltaSumOfSquares_(0)
  , deltaSumOfSizes_(0)
  , throughput_(0.0)
  , rate_(0.0)
  , average_(0.0)
  , rms_(0.0)
  , nbAllocatedEvents_(0)
  , nbPendingRequests_(0)
  , nbReceivedEvents_(0)
  , nbSentEvents_(0)
  , nbSentDqmEvents_(0)
  , nbSentErrorEvents_(0)
  , nbPendingSMDiscards_(0)
  , nbPendingSMDqmDiscards_(0)
  , nbDiscardedEvents_(0)
  , nbLostEvents_(0)
  , nbDataErrors_(0)
  , nbCrcErrors_(0)
  , nbTimeoutsWithEvent_(0)
  , nbTimeoutsWithoutEvent_(0)
  , dataErrorFlag_(0)
  , segmentationMode_(false)
  , nbClients_(0)
  , clientPrcIds_("")
  , nbRawCells_(16)
  , nbRecoCells_(8)
  , nbDqmCells_(8)
  , rawCellSize_(0x400000)  // 4MB
  , recoCellSize_(0x800000) // 8MB
  , dqmCellSize_(0x800000)  // 8MB
  , doDropEvents_(false)
  , doFedIdCheck_(true)
  , doCrcCheck_(1)
  , doDumpEvents_(0)
  , buClassName_("BU")
  , buInstance_(0)
  , smClassName_("StorageManager")
  , smInstance_(0)
  , shmResourceTableTimeout_(200000)
  , monSleepSec_(2)
  , watchSleepSec_(10)
  , timeOutSec_(30)
  , processKillerEnabled_(true)
  , useEvmBoard_(true)
  , reasonForFailed_("")
  , nbAllocateSent_(0)
  , nbTakeReceived_(0)
  , nbDataDiscardReceived_(0)
  , nbDqmDiscardReceived_(0)
  , nbSentLast_(0)
  , sumOfSquaresLast_(0)
  , sumOfSizesLast_(0)
  , lock_(toolbox::BSem::FULL)
  , frb_(0)
  , shmInconsistent_(false)
{
  // setup finite state machine (binding relevant callbacks)
  fsm_.initialize<evf::FUResourceBroker>(this);
  
  // set url, class, instance, and sourceId (=class_instance)
  url_     =
    getApplicationDescriptor()->getContextDescriptor()->getURL()+"/"+
    getApplicationDescriptor()->getURN();
  class_   =getApplicationDescriptor()->getClassName();
  instance_=getApplicationDescriptor()->getInstance();
  sourceId_=class_.toString()+"_"+instance_.toString();
  
  // bind i2o callbacks
  i2o::bind(this,&FUResourceBroker::I2O_FU_TAKE_Callback,
        I2O_FU_TAKE,XDAQ_ORGANIZATION_ID);
  i2o::bind(this,&FUResourceBroker::I2O_FU_DATA_DISCARD_Callback,
        I2O_FU_DATA_DISCARD,XDAQ_ORGANIZATION_ID);
  i2o::bind(this,&FUResourceBroker::I2O_FU_DQM_DISCARD_Callback,
        I2O_FU_DQM_DISCARD,XDAQ_ORGANIZATION_ID);
  i2o::bind(this,&FUResourceBroker::I2O_EVM_LUMISECTION_Callback,
        I2O_EVM_LUMISECTION,XDAQ_ORGANIZATION_ID);
  
  
  // bind HyperDAQ web pages
  xgi::bind(this,&evf::FUResourceBroker::webPageRequest,"Default");
  gui_=new WebGUI(this,&fsm_);
  vector<toolbox::lang::Method*> methods=gui_->getMethods();
  vector<toolbox::lang::Method*>::iterator it;
  for (it=methods.begin();it!=methods.end();++it) {
    if ((*it)->type()=="cgi") {
      string name=static_cast<xgi::MethodSignature*>(*it)->name();
      xgi::bind(this,&evf::FUResourceBroker::webPageRequest,name);
    }
  }
  xgi::bind(this,&evf::FUResourceBroker::customWebPage,"customWebPage");
  

  // allocate i2o memery pool
  string i2oPoolName=sourceId_+"_i2oPool";
  try {
    toolbox::mem::HeapAllocator *allocator=new toolbox::mem::HeapAllocator();
    toolbox::net::URN urn("toolbox-mem-pool",i2oPoolName);
    toolbox::mem::MemoryPoolFactory* poolFactory=
      toolbox::mem::getMemoryPoolFactory();
    i2oPool_=poolFactory->createPool(urn,allocator);
  }
  catch (toolbox::mem::exception::Exception& e) {
    string s="Failed to create pool: "+i2oPoolName;
    LOG4CPLUS_FATAL(log_,s);
    XCEPT_RETHROW(xcept::Exception,s,e);
  }
  
  // publish all parameters to app info space
  exportParameters();
  
  // findRcmsStateListener
  fsm_.findRcmsStateListener();
  
  // set application icon for hyperdaq
  getApplicationDescriptor()->setAttribute("icon", "/evf/images/rbicon.jpg");
  //FUResource::useEvmBoard_ = useEvmBoard_;
}

FUResourceBroker::~FUResourceBroker ( )

virtual

Definition at line 193 of file FUResourceBroker.cc.

{

}

Member Function Documentation

void FUResourceBroker::actionPerformed

(

xdata::Event &

e

)

Definition at line 461 of file FUResourceBroker.cc.

References clientPrcIds_, evf::FUResourceTable::clientPrcIdsAsString(), dataErrorFlag_, doCrcCheck_, doDumpEvents_, evf::FUResource::doFedIdCheck(), doFedIdCheck_, ExpressReco_HICollisions_FallBack::e, lock(), evf::FUResourceTable::nbAllocated(), nbAllocatedEvents_, nbAllocateSent_, evf::FUResourceTable::nbAllocSent(), evf::FUResourceTable::nbClients(), nbClients_, evf::FUResourceTable::nbCompleted(), evf::FUResourceTable::nbCrcErrors(), nbCrcErrors_, nbDataErrors_, evf::FUResourceTable::nbDiscarded(), nbDiscardedEvents_, evf::FUResourceTable::nbErrors(), evf::FUResourceTable::nbLost(), nbLostEvents_, evf::FUResourceTable::nbPending(), nbPendingRequests_, evf::FUResourceTable::nbPendingSMDiscards(), nbPendingSMDiscards_, evf::FUResourceTable::nbPendingSMDqmDiscards(), nbPendingSMDqmDiscards_, nbReceivedEvents_, evf::FUResourceTable::nbSent(), evf::FUResourceTable::nbSentDqm(), nbSentDqmEvents_, evf::FUResourceTable::nbSentError(), nbSentErrorEvents_, nbSentEvents_, nbTimeoutsWithEvent_, nbTimeoutsWithoutEvent_, resourceTable_, evf::FUResourceTable::setDoCrcCheck(), evf::FUResourceTable::setDoDumpEvents(), unlock(), evf::FUResource::useEvmBoard(), and useEvmBoard_.

{
  lock();
  
  if (0!=resourceTable_) {
    
    //gui_->monInfoSpace()->lock();
    
    if (e.type()=="urn:xdata-event:ItemGroupRetrieveEvent") {
      nbClients_          =resourceTable_->nbClients();
      clientPrcIds_       =resourceTable_->clientPrcIdsAsString();
      nbAllocatedEvents_  =resourceTable_->nbAllocated();
      nbPendingRequests_  =resourceTable_->nbPending();
      nbReceivedEvents_   =resourceTable_->nbCompleted();
      nbSentEvents_       =resourceTable_->nbSent();
      nbSentDqmEvents_    =resourceTable_->nbSentDqm();
      nbSentErrorEvents_  =resourceTable_->nbSentError();
      nbPendingSMDiscards_=resourceTable_->nbPendingSMDiscards();
      nbPendingSMDqmDiscards_=resourceTable_->nbPendingSMDqmDiscards();
      nbDiscardedEvents_  =resourceTable_->nbDiscarded();
      nbLostEvents_       =resourceTable_->nbLost();
      nbDataErrors_       =resourceTable_->nbErrors();
      nbCrcErrors_        =resourceTable_->nbCrcErrors();
      nbAllocateSent_     =resourceTable_->nbAllocSent();
      dataErrorFlag_.value_ = (nbCrcErrors_.value_ != 0u + 
                   ((nbDataErrors_.value_ != 0u) << 1) +
                   ((nbLostEvents_.value_ != 0u) << 2) +
                   ((nbTimeoutsWithEvent_.value_ != 0u) << 3) +
                   ((nbTimeoutsWithoutEvent_.value_ != 0u) << 4) +
                   ((nbSentErrorEvents_.value_ != 0u) << 5)
                   );
    }
    else if (e.type()=="ItemChangedEvent") {
      
      string item=dynamic_cast<xdata::ItemChangedEvent&>(e).itemName();
      
      if (item=="doFedIdCheck") FUResource::doFedIdCheck(doFedIdCheck_);
      if (item=="useEvmBoard")  FUResource::useEvmBoard(useEvmBoard_);
      if (item=="doCrcCheck")   resourceTable_->setDoCrcCheck(doCrcCheck_);
      if (item=="doDumpEvents") resourceTable_->setDoDumpEvents(doDumpEvents_);
    }
    
    //gui_->monInfoSpace()->unlock();
  }
  else {
    nbClients_          =0;
    clientPrcIds_       ="";
    nbAllocatedEvents_  =0;
    nbPendingRequests_  =0;
    nbReceivedEvents_   =0;
    nbSentEvents_       =0;
    nbSentDqmEvents_    =0;
    nbSentErrorEvents_  =0;
    nbPendingSMDiscards_=0;
    nbPendingSMDqmDiscards_=0;
    nbDiscardedEvents_  =0;
    nbLostEvents_       =0;
    nbDataErrors_       =0;
    nbCrcErrors_        =0;
    nbAllocateSent_     =0;
  }
  unlock();
}

void FUResourceBroker::configureResources ( )

private

Definition at line 1022 of file FUResourceBroker.cc.

References bu_, doCrcCheck_, doDumpEvents_, evf::FUResource::doFedIdCheck(), doFedIdCheck_, dqmCellSize_, frb_, log_, nbDqmCells_, evf::FUResourceTable::nbFreeSlots(), nbRawCells_, nbRecoCells_, evf::FUResourceTable::nbResources(), rawCellSize_, recoCellSize_, reset(), resourceTable_, segmentationMode_, evf::FUResourceTable::setDoCrcCheck(), evf::FUResourceTable::setDoDumpEvents(), shmInconsistent_, shmResourceTableTimeout_, sm_, evf::FUResource::useEvmBoard(), and useEvmBoard_.

Referenced by configuring(), and emergencyStop().

{
  resourceTable_=new FUResourceTable(segmentationMode_.value_,
                     nbRawCells_.value_,
                     nbRecoCells_.value_,
                     nbDqmCells_.value_,
                     rawCellSize_.value_,
                     recoCellSize_.value_,
                     dqmCellSize_.value_,
                     bu_,sm_,
                     log_, 
                     shmResourceTableTimeout_.value_, 
                     frb_,
                     this);
  FUResource::doFedIdCheck(doFedIdCheck_);
  FUResource::useEvmBoard(useEvmBoard_);
  resourceTable_->setDoCrcCheck(doCrcCheck_);
  resourceTable_->setDoDumpEvents(doDumpEvents_);
  reset();
  shmInconsistent_ = false;
  if(resourceTable_->nbResources() != nbRawCells_.value_ || 
     resourceTable_->nbFreeSlots() != nbRawCells_.value_)
    shmInconsistent_ = true;
}

bool FUResourceBroker::configuring

(

toolbox::task::WorkLoop *

wl

)

Definition at line 205 of file FUResourceBroker.cc.

References configureResources(), connectToBUandSM(), ExpressReco_HICollisions_FallBack::e, edm::hlt::Exception, evf::StateMachine::fireEvent(), evf::StateMachine::fireFailed(), frb_, fsm_, log_, runTheMatrix::msg, evf::FUResourceTable::nbFreeSlots(), nbRawCells_, evf::FUResourceTable::nbResources(), reasonForFailed_, resourceTable_, and shmInconsistent_.

{
  try {
    LOG4CPLUS_INFO(log_, "Start configuring ...");
    connectToBUandSM();
    frb_ = new EvffedFillerRB(this);
    configureResources();
    if(shmInconsistent_){
      std::ostringstream ost;
      ost << "configuring FAILED: Inconsistency in ResourceTable - nbRaw=" 
      << nbRawCells_.value_ << " but nbResources=" << resourceTable_->nbResources()
      << " and nbFreeSlots=" << resourceTable_->nbFreeSlots();
      XCEPT_RAISE(evf::Exception,ost.str());
    } 
    LOG4CPLUS_INFO(log_, "Finished configuring!");
    fsm_.fireEvent("ConfigureDone",this);

  }
  catch (xcept::Exception &e) {
    std::string msg  = "configuring FAILED: " + (string)e.what();
    reasonForFailed_ = e.what();
    fsm_.fireFailed(msg,this);
  }

  return false;
}

void FUResourceBroker::connectToBUandSM	(		)
throw	(	evf::Exception
	)

Definition at line 412 of file FUResourceBroker.cc.

References bu_, buClassName_, buInstance_, i2oPool_, log_, runTheMatrix::msg, sm_, smClassName_, smInstance_, and sourceId_.

Referenced by configuring().

{
  typedef set<xdaq::ApplicationDescriptor*> AppDescSet_t;
  typedef AppDescSet_t::iterator            AppDescIter_t;
  
  // locate input BU
  AppDescSet_t setOfBUs=
    getApplicationContext()->getDefaultZone()->
    getApplicationDescriptors(buClassName_.toString());
  
  if (0!=bu_) { delete bu_; bu_=0; }
  
  for (AppDescIter_t it=setOfBUs.begin();it!=setOfBUs.end();++it)
    if ((*it)->getInstance()==buInstance_)
      bu_=new BUProxy(getApplicationDescriptor(),*it,
              getApplicationContext(),i2oPool_);
  
  if (0==bu_) {
    string msg=sourceId_+" failed to locate input BU!";
    XCEPT_RAISE(evf::Exception,msg);
  }
  
  // locate output SM
  AppDescSet_t setOfSMs=
    getApplicationContext()->getDefaultZone()->
    getApplicationDescriptors(smClassName_.toString());
  
  if (0!=sm_) { delete sm_; sm_=0; }
  
  for (AppDescIter_t it=setOfSMs.begin();it!=setOfSMs.end();++it)
    if ((*it)->getInstance()==smInstance_)
      sm_=new SMProxy(getApplicationDescriptor(),*it,
              getApplicationContext(),i2oPool_);
  
  if (0==sm_) LOG4CPLUS_WARN(log_,sourceId_<<" failed to locate output SM!");
}

void FUResourceBroker::customWebPage	(	xgi::Input *	in,
		xgi::Output *	out
	)
throw	(	xgi::exception::Exception
	)

Definition at line 837 of file FUResourceBroker.cc.

References gather_cfg::cout, asciidump::els, i, recoMuon::in, CommonMethods::lock(), dbtoconf::out, evf::utils::pid, evf::utils::state, ntuplemaker::status, asciidump::table, and cond::rpcobgas::time.

Referenced by FUResourceBroker().

{
  using namespace cgicc;
  Cgicc cgi(in);
  std::vector<FormEntry> els = cgi.getElements() ;
  for(std::vector<FormEntry>::iterator it = els.begin(); it != els.end(); it++)
    std::cout << "form entry " << (*it).getValue() << std::endl;

  std::vector<FormEntry> el1;
  cgi.getElement("crcError",el1);
  *out<<"<html>"<<endl;
  gui_->htmlHead(in,out,sourceId_);
  *out<<"<body>"<<endl;
  gui_->htmlHeadline(in,out);

  lock();
  
  if (0!=resourceTable_) {
    if(el1.size()!=0) {
      resourceTable_->injectCRCError();
    }
    *out << "<form method=\"GET\" action=\"customWebPage\" >";
    *out << "<button name=\"crcError\" type=\"submit\" value=\"injCRC\">Inject CRC</button>" << endl;
    *out << "</form>" << endl;
    *out << "<hr/>" << std::endl;
    vector<pid_t> client_prc_ids = resourceTable_->clientPrcIds();
    *out<<table().set("frame","void").set("rules","rows")
                 .set("class","modules").set("width","250")<<endl
    <<tr()<<th("Client Processes").set("colspan","3")<<tr()<<endl
    <<tr()
    <<th("client").set("align","left")
    <<th("process id").set("align","center")
    <<th("status").set("align","center")
    <<tr()
    <<endl;
    for (UInt_t i=0;i<client_prc_ids.size();i++) {

      pid_t pid   =client_prc_ids[i];
      int   status=kill(pid,0);

      stringstream ssi;      ssi<<i+1;
      stringstream sspid;    sspid<<pid;
      stringstream ssstatus; ssstatus<<status;
      
      string bg_status = (status==0) ? "#00ff00" : "ff0000";
      *out<<tr()
      <<td(ssi.str()).set("align","left")
      <<td(sspid.str()).set("align","center")
      <<td(ssstatus.str()).set("align","center").set("bgcolor",bg_status)
      <<tr()<<endl;
    }
    *out<<table()<<endl;
    *out<<"<br><br>"<<endl;

    vector<string> states      = resourceTable_->cellStates();
    vector<UInt_t> evt_numbers = resourceTable_->cellEvtNumbers();
    vector<pid_t>  prc_ids     = resourceTable_->cellPrcIds();
    vector<time_t> time_stamps = resourceTable_->cellTimeStamps();

    *out<<table().set("frame","void").set("rules","rows")
                 .set("class","modules").set("width","500")<<endl
    <<tr()<<th("Shared Memory Cells").set("colspan","6")<<tr()<<endl
    <<tr()
    <<th("cell").set("align","left")
    <<th("state").set("align","center")
    <<th("event").set("align","center")
    <<th("process id").set("align","center")
    <<th("timestamp").set("align","center")
    <<th("time").set("align","center")
    <<tr()
    <<endl;
    for (UInt_t i=0;i<states.size();i++) {
      string state=states[i];
      UInt_t evt   = evt_numbers[i];
      pid_t  pid   = prc_ids[i];
      time_t tstamp= time_stamps[i];
      double tdiff = difftime(time(0),tstamp);
      
      stringstream ssi;      ssi<<i;
      stringstream ssevt;    if (evt!=0xffffffff) ssevt<<evt; else ssevt<<" - ";
      stringstream sspid;    if (pid!=0) sspid<<pid; else sspid<<" - ";
      stringstream sststamp; if (tstamp!=0) sststamp<<tstamp; else sststamp<<" - ";
      stringstream sstdiff;  if (tstamp!=0) sstdiff<<tdiff; else sstdiff<<" - ";
      
      string bg_state = "#ffffff";
      if (state=="RAWWRITING"||state=="RAWWRITTEN"||
      state=="RAWREADING"||state=="RAWREAD")
    bg_state="#99CCff";
      else if (state=="PROCESSING")
    bg_state="#ff0000";
      else if (state=="PROCESSED"||state=="RECOWRITING"||state=="RECOWRITTEN")
    bg_state="#CCff99";
      else if (state=="SENDING")
    bg_state="#00FF33";
      else if (state=="SENT")
    bg_state="#006633";
      else if (state=="DISCARDING")
    bg_state="#FFFF00";
      else if (state=="LUMISECTION")
    bg_state="#0000FF";
      
      *out<<tr()
      <<td(ssi.str()).set("align","left")
      <<td(state).set("align","center").set("bgcolor",bg_state)
      <<td(ssevt.str()).set("align","center")
      <<td(sspid.str()).set("align","center")
      <<td(sststamp.str()).set("align","center")
      <<td(sstdiff.str()).set("align","center")
      <<tr()<<endl;
    }
    *out<<table()<<endl;
    *out<<"<br><br>"<<endl;

    vector<string> dqmstates      = resourceTable_->dqmCellStates();

    *out<<table().set("frame","void").set("rules","rows")
                 .set("class","modules").set("width","500")<<endl
    <<tr()<<th("Shared Memory DQM Cells").set("colspan","6")<<tr()<<endl
    <<tr()
    <<th("cell").set("align","left")
    <<th("state").set("align","center")
    <<tr()
    <<endl;
    for (UInt_t i=0;i<dqmstates.size();i++) {
      string state=dqmstates[i];
      
      string bg_state = "#ffffff";
      if (state=="WRITING"||state=="WRITTEN")
    bg_state="#99CCff";
      else if (state=="SENDING")
    bg_state="#00FF33";
      else if (state=="SENT")
    bg_state="#006633";
      else if (state=="DISCARDING")
    bg_state="#FFFF00";
      
      *out<<tr()
      <<td(state).set("align","center").set("bgcolor",bg_state)
      <<tr()<<endl;
    }
    *out<<table()<<endl;


    
  }
  *out<<"</body>"<<endl<<"</html>"<<endl;
  
  unlock();
}

double FUResourceBroker::deltaT ( const struct timeval *  start, const struct timeval *  end  )

private

Definition at line 815 of file FUResourceBroker.cc.

Referenced by monitoring().

{
  unsigned int  sec;
  unsigned int  usec;
  
  sec = end->tv_sec - start->tv_sec;
  
  if(end->tv_usec > start->tv_usec) {
    usec = end->tv_usec - start->tv_usec;
  }
  else {
    sec--;
    usec = 1000000 - ((unsigned int )(start->tv_usec - end->tv_usec));
  }
  
  return ((double)sec) + ((double)usec) / 1000000.0;
}

void FUResourceBroker::emergencyStop ( )

private

Definition at line 988 of file FUResourceBroker.cc.

References evf::FUResourceTable::clientPrcIds(), configureResources(), gather_cfg::cout, evf::FUResourceTable::handleCrashedEP(), i, evf::FUResourceTable::isReadyToShutDown(), evf::FUResourceTable::lastResort(), lock(), log_, nbTimeoutsWithEvent_, nbTimeoutsWithoutEvent_, evf::utils::pid, evf::FUResourceTable::printWorkLoopStatus(), reasonForFailed_, resourceTable_, runNumber_, shmInconsistent_, stor::utils::sleep(), and unlock().

Referenced by stopping().

{
  LOG4CPLUS_WARN(log_, "in Emergency stop - handle non-clean stops");
  vector<pid_t> client_prc_ids = resourceTable_->clientPrcIds();
  for (UInt_t i=0;i<client_prc_ids.size();i++) {
    pid_t  pid   =client_prc_ids[i];
    std::cout << "B: killing process " << i << "pid=" << pid << std::endl;
    if(pid!=0){
      //assume processes are dead by now
      if(!resourceTable_->handleCrashedEP(runNumber_,pid))
    nbTimeoutsWithoutEvent_++;
      else
    nbTimeoutsWithEvent_++;
    }
  }
  resourceTable_->lastResort();
  ::sleep(1);
  if(!resourceTable_->isReadyToShutDown())
    {
      reasonForFailed_ = "EmergencyStop: failed to shut down ResourceTable";
      XCEPT_RAISE(evf::Exception,reasonForFailed_);
    }
  resourceTable_->printWorkLoopStatus();
  lock();
  std::cout << "delete resourcetable" <<std::endl;
  delete resourceTable_;
  resourceTable_=0;
  std::cout << "cycle through resourcetable config " << std::endl;
  configureResources();
  unlock();
  if(shmInconsistent_) XCEPT_RAISE(evf::Exception,"Inconsistent shm state");
  std::cout << "done with emergency stop"  << std::endl;
}

bool FUResourceBroker::enabling

(

toolbox::task::WorkLoop *

wl

)

Definition at line 234 of file FUResourceBroker.cc.

References dataErrorFlag_, ExpressReco_HICollisions_FallBack::e, edm::hlt::Exception, evf::StateMachine::fireEvent(), evf::StateMachine::fireFailed(), fsm_, log_, runTheMatrix::msg, nbTimeoutsWithEvent_, nbTimeoutsWithoutEvent_, reasonForFailed_, evf::FUResourceTable::resetCounters(), resourceTable_, runNumber_, evf::FUResourceTable::sendAllocate(), evf::FUResourceTable::setRunNumber(), evf::FUResourceTable::startDiscardWorkLoop(), startMonitoringWorkLoop(), evf::FUResourceTable::startSendDataWorkLoop(), evf::FUResourceTable::startSendDqmWorkLoop(), and startWatchingWorkLoop().

{
  try {
    LOG4CPLUS_INFO(log_, "Start enabling ...");
    startMonitoringWorkLoop();
    startWatchingWorkLoop();
    resourceTable_->setRunNumber(runNumber_);
    resourceTable_->resetCounters();
    resourceTable_->startDiscardWorkLoop();
    resourceTable_->startSendDataWorkLoop();
    resourceTable_->startSendDqmWorkLoop();
    resourceTable_->sendAllocate();
    
    nbTimeoutsWithEvent_         = 0;
    nbTimeoutsWithoutEvent_      = 0;
    dataErrorFlag_               = 0;

    LOG4CPLUS_INFO(log_, "Finished enabling!");
    fsm_.fireEvent("EnableDone",this);
  }
  catch (xcept::Exception &e) {
    std::string msg  = "enabling FAILED: "+xcept::stdformat_exception_history(e);
    reasonForFailed_ = e.what();
    fsm_.fireFailed(msg,this);
  }
  
  return false;
}

void FUResourceBroker::exportParameters ( )

private

Definition at line 713 of file FUResourceBroker.cc.

References evf::WebGUI::addDebugCounter(), evf::WebGUI::addItemChangedListener(), evf::WebGUI::addMonitorCounter(), evf::WebGUI::addMonitorParam(), evf::WebGUI::addStandardParam(), average_, buClassName_, buInstance_, class_, clientPrcIds_, dataErrorFlag_, deltaN_, deltaSumOfSizes_, deltaSumOfSquares_, deltaT_, doCrcCheck_, doDropEvents_, doDumpEvents_, doFedIdCheck_, dqmCellSize_, evf::WebGUI::exportParameters(), evf::StateMachine::foundRcmsStateListener(), fsm_, gui_, instance_, monSleepSec_, nbAllocatedEvents_, nbAllocateSent_, nbClients_, nbCrcErrors_, nbDataDiscardReceived_, nbDataErrors_, nbDiscardedEvents_, nbDqmCells_, nbDqmDiscardReceived_, nbLostEvents_, nbPendingRequests_, nbPendingSMDiscards_, nbPendingSMDqmDiscards_, nbRawCells_, nbReceivedEvents_, nbRecoCells_, nbSentDqmEvents_, nbSentErrorEvents_, nbSentEvents_, nbTakeReceived_, nbTimeoutsWithEvent_, nbTimeoutsWithoutEvent_, processKillerEnabled_, rate_, rawCellSize_, evf::StateMachine::rcmsStateListener(), reasonForFailed_, recoCellSize_, rms_, runNumber_, segmentationMode_, shmResourceTableTimeout_, smClassName_, smInstance_, evf::StateMachine::stateName(), throughput_, timeOutSec_, url_, useEvmBoard_, and watchSleepSec_.

Referenced by FUResourceBroker().

{
  assert(0!=gui_);
  
  gui_->addMonitorParam("url",                      &url_);
  gui_->addMonitorParam("class",                    &class_);
  gui_->addMonitorParam("instance",                 &instance_);
  gui_->addMonitorParam("runNumber",                &runNumber_);
  gui_->addMonitorParam("stateName",                 fsm_.stateName());

  gui_->addMonitorParam("deltaT",                   &deltaT_);
  gui_->addMonitorParam("deltaN",                   &deltaN_);
  gui_->addMonitorParam("deltaSumOfSquares",        &deltaSumOfSquares_);
  gui_->addMonitorParam("deltaSumOfSizes",          &deltaSumOfSizes_);
    
  gui_->addMonitorParam("throughput",               &throughput_);
  gui_->addMonitorParam("rate",                     &rate_);
  gui_->addMonitorParam("average",                  &average_);
  gui_->addMonitorParam("rms",                      &rms_);
  gui_->addMonitorParam("dataErrorFlag",            &dataErrorFlag_);
  
  gui_->addMonitorCounter("nbAllocatedEvents",      &nbAllocatedEvents_);
  gui_->addMonitorCounter("nbPendingRequests",      &nbPendingRequests_);
  gui_->addMonitorCounter("nbReceivedEvents",       &nbReceivedEvents_);
  gui_->addMonitorCounter("nbSentEvents",           &nbSentEvents_);
  gui_->addMonitorCounter("nbSentErrorEvents",      &nbSentErrorEvents_);
  gui_->addMonitorCounter("nbDiscardedEvents",      &nbDiscardedEvents_);
  gui_->addMonitorCounter("nbPendingSMDiscards",    &nbPendingSMDiscards_);

  gui_->addMonitorCounter("nbSentDqmEvents",        &nbSentDqmEvents_);
  gui_->addMonitorCounter("nbDqmDiscardReceived",   &nbDqmDiscardReceived_);
  gui_->addMonitorCounter("nbPendingSMDqmDiscards", &nbPendingSMDqmDiscards_);

  gui_->addMonitorCounter("nbLostEvents",           &nbLostEvents_);
  gui_->addMonitorCounter("nbDataErrors",           &nbDataErrors_);
  gui_->addMonitorCounter("nbCrcErrors",            &nbCrcErrors_);
  gui_->addMonitorCounter("nbTimeoutsWithEvent",    &nbTimeoutsWithEvent_);
  gui_->addMonitorCounter("nbTimeoutsWithoutEvent", &nbTimeoutsWithoutEvent_);

  gui_->addStandardParam("segmentationMode",        &segmentationMode_);
  gui_->addStandardParam("nbClients",               &nbClients_);
  gui_->addStandardParam("clientPrcIds",            &clientPrcIds_);
  gui_->addStandardParam("nbRawCells",              &nbRawCells_);
  gui_->addStandardParam("nbRecoCells",             &nbRecoCells_);
  gui_->addStandardParam("nbDqmCells",              &nbDqmCells_);
  gui_->addStandardParam("rawCellSize",             &rawCellSize_);
  gui_->addStandardParam("recoCellSize",            &recoCellSize_);
  gui_->addStandardParam("dqmCellSize",             &dqmCellSize_);

  gui_->addStandardParam("doDropEvents",            &doDropEvents_);
  gui_->addStandardParam("doFedIdCheck",            &doFedIdCheck_);
  gui_->addStandardParam("doCrcCheck",              &doCrcCheck_);
  gui_->addStandardParam("doDumpEvents",            &doDumpEvents_);
  gui_->addStandardParam("buClassName",             &buClassName_);
  gui_->addStandardParam("buInstance",              &buInstance_);
  gui_->addStandardParam("smClassName",             &smClassName_);
  gui_->addStandardParam("smInstance",              &smInstance_);
  gui_->addStandardParam("shmResourceTableTimeout", &shmResourceTableTimeout_);
  gui_->addStandardParam("monSleepSec",             &monSleepSec_);
  gui_->addStandardParam("watchSleepSec",           &watchSleepSec_);
  gui_->addStandardParam("timeOutSec",              &timeOutSec_);
  gui_->addStandardParam("processKillerEnabled",    &processKillerEnabled_);
  gui_->addStandardParam("useEvmBoard",             &useEvmBoard_);
  gui_->addStandardParam("rcmsStateListener",        fsm_.rcmsStateListener());
  gui_->addStandardParam("foundRcmsStateListener",   fsm_.foundRcmsStateListener());
  gui_->addStandardParam("reasonForFailed",         &reasonForFailed_);
  
  gui_->addDebugCounter("nbAllocateSent",           &nbAllocateSent_);
  gui_->addDebugCounter("nbTakeReceived",           &nbTakeReceived_);
  gui_->addDebugCounter("nbDataDiscardReceived",    &nbDataDiscardReceived_);

  gui_->exportParameters();

  gui_->addItemChangedListener("doFedIdCheck",      this);
  gui_->addItemChangedListener("useEvmBoard",       this);
  gui_->addItemChangedListener("doCrcCheck",        this);
  gui_->addItemChangedListener("doDumpEvents",      this);
}

xoap::MessageReference FUResourceBroker::fsmCallback	(	xoap::MessageReference	msg	)
throw	(	xoap::exception::Exception
	)

Definition at line 352 of file FUResourceBroker.cc.

References runTheMatrix::msg.

{
  return fsm_.commandCallback(msg);
}

bool FUResourceBroker::halting

(

toolbox::task::WorkLoop *

wl

)

Definition at line 301 of file FUResourceBroker.cc.

References prof2calltree::count, ExpressReco_HICollisions_FallBack::e, edm::hlt::Exception, evf::StateMachine::fireEvent(), evf::StateMachine::fireFailed(), frb_, fsm_, evf::FUResourceTable::halt(), evf::FUResourceTable::isActive(), evf::FUResourceTable::isReadyToShutDown(), lock(), log_, runTheMatrix::msg, reasonForFailed_, resourceTable_, stor::utils::sleep(), and unlock().

{
  try {
    LOG4CPLUS_INFO(log_, "Start halting ...");
    if (resourceTable_->isActive()) {
      resourceTable_->halt();
      UInt_t count = 0;
      while (count<10) {
    if (resourceTable_->isReadyToShutDown()) {
      lock();
      delete resourceTable_;
      resourceTable_=0;
      unlock();
      LOG4CPLUS_INFO(log_,count+1<<". try to destroy resource table succeeded!");
      break;
    }
    else {
      count++;
      LOG4CPLUS_DEBUG(log_,count<<". try to destroy resource table failed ...");
      ::sleep(1);
    }
      }
    }
    else {
      lock();
      delete resourceTable_;
      resourceTable_=0;
      unlock();
    }
    
    if (0==resourceTable_) {
      LOG4CPLUS_INFO(log_,"Finished halting!");
      fsm_.fireEvent("HaltDone",this);
    }
    else {
      std::string msg  = "halting FAILED: ResourceTable shutdown timed out.";
      reasonForFailed_ = "RESOURCETABLE SHUTDOWN TIMED OUT";
      fsm_.fireFailed(msg,this);
    }
  }
  catch (xcept::Exception &e) {
    std::string msg  = "halting FAILED: "+xcept::stdformat_exception_history(e);
    reasonForFailed_ = e.what();
    fsm_.fireFailed(msg,this);
  }
  if(frb_) delete frb_;
  return false;
}

void FUResourceBroker::I2O_EVM_LUMISECTION_Callback

(

toolbox::mem::Reference *

bufRef

)

Definition at line 375 of file FUResourceBroker.cc.

References evf::StateMachine::fireFailed(), fsm_, log_, runTheMatrix::msg, evf::FUResourceTable::postEndOfLumiSection(), and resourceTable_.

Referenced by FUResourceBroker().

{

  I2O_EVM_END_OF_LUMISECTION_MESSAGE_FRAME *msg = 
    (I2O_EVM_END_OF_LUMISECTION_MESSAGE_FRAME *)bufRef->getDataLocation();
  if(msg->lumiSection==0){
    LOG4CPLUS_ERROR(log_,"EOL message received for ls=0!!! ");
    fsm_.fireFailed("EOL message received for ls=0!!! ",this);
  }
  resourceTable_->postEndOfLumiSection(bufRef); // this method dummy for now
//   I2O_EVM_END_OF_LUMISECTION_MESSAGE_FRAME *msg =
//     (I2O_EVM_END_OF_LUMISECTION_MESSAGE_FRAME *)bufRef->getDataLocation();
  
//   LOG4CPLUS_WARN(log_, "Received END-OF-LS from EVM for LS " << msg->lumiSection);
  
}

void FUResourceBroker::I2O_FU_DATA_DISCARD_Callback

(

toolbox::mem::Reference *

bufRef

)

Definition at line 396 of file FUResourceBroker.cc.

References evf::FUResourceTable::discardDataEvent(), nbDataDiscardReceived_, and resourceTable_.

Referenced by FUResourceBroker().

{
  nbDataDiscardReceived_.value_++;
  resourceTable_->discardDataEvent(bufRef);
}

void FUResourceBroker::I2O_FU_DQM_DISCARD_Callback

(

toolbox::mem::Reference *

bufRef

)

Definition at line 404 of file FUResourceBroker.cc.

References evf::FUResourceTable::discardDqmEvent(), nbDqmDiscardReceived_, and resourceTable_.

Referenced by FUResourceBroker().

{
  nbDqmDiscardReceived_.value_++;
  resourceTable_->discardDqmEvent(bufRef);
}

void FUResourceBroker::I2O_FU_TAKE_Callback

(

toolbox::mem::Reference *

bufRef

)

Definition at line 360 of file FUResourceBroker.cc.

References evf::FUResourceTable::buildResource(), evf::StateMachine::checkIfEnabled(), doDropEvents_, evf::FUResourceTable::dropEvent(), fsm_, log_, nbTakeReceived_, resourceTable_, and evf::StateMachine::stateName().

Referenced by FUResourceBroker().

{
  nbTakeReceived_.value_++;
  if(fsm_.checkIfEnabled()){
    bool eventComplete=resourceTable_->buildResource(bufRef);
    if (eventComplete&&doDropEvents_) resourceTable_->dropEvent();
  }
  else{
    LOG4CPLUS_ERROR(log_,"TAKE i2o frame received in state " 
            << fsm_.stateName() << " is being lost");
    bufRef->release();
  }
}

unsigned int evf::FUResourceBroker::instanceNumber ( ) const

inline

Definition at line 97 of file FUResourceBroker.h.

References instance_.

Referenced by evf::EvffedFillerRB::EvffedFillerRB().

{return instance_.value_;}

void evf::FUResourceBroker::lock ( void  )

inlineprivate

Definition at line 117 of file FUResourceBroker.h.

References lock_.

Referenced by actionPerformed(), emergencyStop(), halting(), monitoring(), and watching().

{ lock_.take(); }

bool FUResourceBroker::monitoring

(

toolbox::task::WorkLoop *

wl

)

Definition at line 549 of file FUResourceBroker.cc.

References average_, deltaN_, deltaSumOfSizes_, deltaSumOfSquares_, deltaT(), deltaT_, gui_, lock(), plotscripts::mean(), evf::WebGUI::monInfoSpace(), monSleepSec_, monStartTime_, evf::FUResourceTable::nbSent(), nbSentLast_, rate_, resourceTable_, rms_, stor::utils::sleep(), mathSSE::sqrt(), evf::FUResourceTable::sumOfSizes(), sumOfSizesLast_, evf::FUResourceTable::sumOfSquares(), sumOfSquaresLast_, throughput_, and unlock().

Referenced by startMonitoringWorkLoop().

{
  unsigned int nbSent;
  uint64_t     sumOfSquares;
  unsigned int sumOfSizes;
  uint64_t     deltaSumOfSquares;

  lock();
  if (0==resourceTable_) {
    deltaT_.value_           =0.0;
    deltaN_.value_           =  0;
    deltaSumOfSquares_.value_=0.0;
    deltaSumOfSizes_.value_  =  0;
    throughput_              =0.0;
    rate_                    =0.0;
    average_                 =0.0;
    rms_                     =0.0;
    unlock();    
    return false;
  }
  else {
    nbSent      =resourceTable_->nbSent();
    sumOfSquares=resourceTable_->sumOfSquares();
    sumOfSizes  =resourceTable_->sumOfSizes();
  }
  unlock();
  
  struct timeval  monEndTime;
  struct timezone timezone;
  
  gettimeofday(&monEndTime,&timezone);
  
  xdata::getInfoSpaceFactory()->lock();
  gui_->monInfoSpace()->lock();
  
  deltaT_.value_=deltaT(&monStartTime_,&monEndTime);
  monStartTime_=monEndTime;
  
  deltaN_.value_=nbSent-nbSentLast_;
  nbSentLast_=nbSent;
  
  deltaSumOfSquares=sumOfSquares-sumOfSquaresLast_;
  deltaSumOfSquares_.value_=(double)deltaSumOfSquares;
  sumOfSquaresLast_=sumOfSquares;
  
  deltaSumOfSizes_.value_=sumOfSizes-sumOfSizesLast_;
  sumOfSizesLast_=sumOfSizes;
  
  if (deltaT_.value_!=0) {
    throughput_=deltaSumOfSizes_.value_/deltaT_.value_;
    rate_      =deltaN_.value_/deltaT_.value_;
  }
  else {
    throughput_=0.0;
    rate_      =0.0;
  }
  
  double meanOfSquares,mean,squareOfMean,variance;
  
  if(deltaN_.value_!=0) {
    meanOfSquares=deltaSumOfSquares_.value_/((double)(deltaN_.value_));
    mean=((double)(deltaSumOfSizes_.value_))/((double)(deltaN_.value_));
    squareOfMean=mean*mean;
    variance=meanOfSquares-squareOfMean; if(variance<0.0) variance=0.0;
    
    average_=deltaSumOfSizes_.value_/deltaN_.value_;
    rms_    =std::sqrt(variance);
  }
  else {
    average_=0.0;
    rms_    =0.0;
  }
  
  gui_->monInfoSpace()->unlock();  
  xdata::getInfoSpaceFactory()->unlock();
    
  ::sleep(monSleepSec_.value_);
  
  return true;
}

void FUResourceBroker::reset ( void  )

private

Definition at line 794 of file FUResourceBroker.cc.

References average_, deltaN_, deltaSumOfSizes_, deltaSumOfSquares_, deltaT_, gui_, nbSentLast_, rate_, evf::WebGUI::resetCounters(), rms_, sumOfSizesLast_, sumOfSquaresLast_, and throughput_.

Referenced by configureResources().

{
  gui_->resetCounters();
  
  deltaT_           =0.0;
  deltaN_           =  0;
  deltaSumOfSquares_=0.0;
  deltaSumOfSizes_  =  0;
  
  throughput_       =0.0;
  rate_             =0.0;
  average_          =0.0;
  rms_              =0.0;
  
  nbSentLast_       =  0;
  sumOfSquaresLast_ =  0;
  sumOfSizesLast_   =  0;
}

void FUResourceBroker::startMonitoringWorkLoop	(		)
throw	(	evf::Exception
	)

Definition at line 527 of file FUResourceBroker.cc.

References asMonitoring_, ExpressReco_HICollisions_FallBack::e, edm::hlt::Exception, monitoring(), monStartTime_, runTheMatrix::msg, sourceId_, and wlMonitoring_.

Referenced by enabling().

{
  struct timezone timezone;
  gettimeofday(&monStartTime_,&timezone);
  
  try {
    wlMonitoring_=
      toolbox::task::getWorkLoopFactory()->getWorkLoop(sourceId_+"Monitoring",
                               "waiting");
    if (!wlMonitoring_->isActive()) wlMonitoring_->activate();
    asMonitoring_=toolbox::task::bind(this,&FUResourceBroker::monitoring,
                      sourceId_+"Monitoring");
    wlMonitoring_->submit(asMonitoring_);
  }
  catch (xcept::Exception& e) {
    string msg = "Failed to start workloop 'Monitoring'.";
    XCEPT_RETHROW(evf::Exception,msg,e);
  }
}

void FUResourceBroker::startWatchingWorkLoop	(		)
throw	(	evf::Exception
	)

Definition at line 632 of file FUResourceBroker.cc.

References asWatching_, ExpressReco_HICollisions_FallBack::e, edm::hlt::Exception, runTheMatrix::msg, sourceId_, watching(), and wlWatching_.

Referenced by enabling().

{
  try {
    wlWatching_=
      toolbox::task::getWorkLoopFactory()->getWorkLoop(sourceId_+"Watching",
                               "waiting");
    if (!wlWatching_->isActive()) wlWatching_->activate();
    asWatching_=toolbox::task::bind(this,&FUResourceBroker::watching,
                    sourceId_+"Watching");
    wlWatching_->submit(asWatching_);
  }
  catch (xcept::Exception& e) {
    string msg = "Failed to start workloop 'Watching'.";
    XCEPT_RETHROW(evf::Exception,msg,e);
  }
}

bool FUResourceBroker::stopping

(

toolbox::task::WorkLoop *

wl

)

Definition at line 265 of file FUResourceBroker.cc.

References gather_cfg::cout, ExpressReco_HICollisions_FallBack::e, emergencyStop(), edm::hlt::Exception, evf::StateMachine::fireEvent(), evf::StateMachine::fireFailed(), fsm_, evf::FUResourceTable::isReadyToShutDown(), log_, runTheMatrix::msg, cmsPerfSuiteHarvest::now, reasonForFailed_, resourceTable_, shmResourceTableTimeout_, and evf::FUResourceTable::stop().

{
  try {
    LOG4CPLUS_INFO(log_, "Start stopping :) ...");
    resourceTable_->stop();
    timeval now;
    timeval then;
    gettimeofday(&then,0);
    while (!resourceTable_->isReadyToShutDown()) {
      ::usleep(shmResourceTableTimeout_.value_*10);
      gettimeofday(&now,0);
      if ((unsigned int)(now.tv_sec-then.tv_sec) > shmResourceTableTimeout_.value_/10000) {
    std::cout << "times: " << now.tv_sec << " " << then.tv_sec << " " 
          <<  shmResourceTableTimeout_.value_/10000 << std::endl;
    LOG4CPLUS_WARN(log_, "Some Process did not detach - going to Emergency stop!");
    emergencyStop();
    break;
      }
    }

    if (resourceTable_->isReadyToShutDown()) {
      LOG4CPLUS_INFO(log_, "Finished stopping!");
      fsm_.fireEvent("StopDone",this);
    }
  }
  catch (xcept::Exception &e) {
    std::string msg  = "stopping FAILED: "+xcept::stdformat_exception_history(e);
    reasonForFailed_ = e.what();
    fsm_.fireFailed(msg,this);
  }
  
  return false;
}

void evf::FUResourceBroker::unlock ( void  )

inlineprivate

Definition at line 118 of file FUResourceBroker.h.

References lock_.

Referenced by actionPerformed(), emergencyStop(), halting(), monitoring(), and watching().

{ lock_.give(); }

bool FUResourceBroker::watching

(

toolbox::task::WorkLoop *

wl

)

Definition at line 651 of file FUResourceBroker.cc.

References evf::FUResourceTable::cellEvtNumbers(), evf::FUResourceTable::cellPrcIds(), evf::FUResourceTable::cellTimeStamps(), evf::FUResourceTable::clientPrcIds(), evf::FUResourceTable::handleCrashedEP(), i, lock(), log_, evf::FUResourceTable::nbFreeSlots(), nbRawCells_, evf::FUResourceTable::nbResources(), nbTimeoutsWithEvent_, nbTimeoutsWithoutEvent_, evf::utils::pid, processKillerEnabled_, resourceTable_, runNumber_, shmInconsistent_, stor::utils::sleep(), ntuplemaker::status, cond::rpcobgas::time, timeOutSec_, unlock(), and watchSleepSec_.

Referenced by startWatchingWorkLoop().

{
  lock();
  
  if (0==resourceTable_) {
    unlock();
    return false;
  }

  vector<pid_t>  evt_prcids =resourceTable_->cellPrcIds();
  vector<UInt_t> evt_numbers=resourceTable_->cellEvtNumbers();
  vector<time_t> evt_tstamps=resourceTable_->cellTimeStamps(); 
  
  time_t tcurr=time(0);  
  for (UInt_t i=0;i<evt_tstamps.size();i++) {
    pid_t  pid   =evt_prcids[i];
    UInt_t evt   =evt_numbers[i];
    time_t tstamp=evt_tstamps[i]; if (tstamp==0) continue;
    double tdiff =difftime(tcurr,tstamp);
    if (tdiff>timeOutSec_) {
      if(processKillerEnabled_) {
    LOG4CPLUS_ERROR(log_,"evt "<<evt<<" timed out, "<<"kill prc "<<pid);
    kill(pid,9);
    nbTimeoutsWithEvent_++;
      }
      else {
    LOG4CPLUS_INFO(log_,"evt "<<evt<<" under processing for more than "
               <<timeOutSec_<<"sec for process "<<pid);
      }
    }
  }
  
  vector<pid_t> prcids=resourceTable_->clientPrcIds();
  for (UInt_t i=0;i<prcids.size();i++) {
    pid_t pid   =prcids[i];
    int   status=kill(pid,0);
    if (status!=0) {
      LOG4CPLUS_ERROR(log_,"EP prc "<<pid<<" died, send to error stream if processing.");
      if(!resourceTable_->handleCrashedEP(runNumber_,pid))
    nbTimeoutsWithoutEvent_++;
    }
  }
  
  if((resourceTable_->nbResources() != nbRawCells_.value_) && !shmInconsistent_){
    std::ostringstream ost;
    ost << "Watchdog spotted inconsistency in ResourceTable - nbRaw=" 
    << nbRawCells_.value_ << " but nbResources=" << resourceTable_->nbResources()
    << " and nbFreeSlots=" << resourceTable_->nbFreeSlots();
    XCEPT_DECLARE(evf::Exception,
          sentinelException, ost.str());
    notifyQualified("error",sentinelException);
    shmInconsistent_ = true;
  }

  unlock();
  
  ::sleep(watchSleepSec_.value_);
  return true;
}

void FUResourceBroker::webPageRequest	(	xgi::Input *	in,
		xgi::Output *	out
	)
throw	(	xgi::exception::Exception
	)

Definition at line 451 of file FUResourceBroker.cc.

References recoMuon::in, AlCaRecoCosmics_cfg::name, and dbtoconf::out.

Referenced by FUResourceBroker().

{
  string name=in->getenv("PATH_INFO");
  if (name.empty()) name="defaultWebPage";
  static_cast<xgi::MethodSignature*>(gui_->getMethod(name))->invoke(in,out);
}

evf::FUResourceBroker::XDAQ_INSTANTIATOR

(

)

Friends And Related Function Documentation

friend class evf::EvffedFillerRB

friend

Definition at line 243 of file FUResourceBroker.h.

Member Data Documentation

toolbox::task::ActionSignature* evf::FUResourceBroker::asMonitoring_

private

Definition at line 149 of file FUResourceBroker.h.

Referenced by startMonitoringWorkLoop().

toolbox::task::ActionSignature* evf::FUResourceBroker::asWatching_

private

Definition at line 153 of file FUResourceBroker.h.

Referenced by startWatchingWorkLoop().

xdata::Double evf::FUResourceBroker::average_

private

Definition at line 172 of file FUResourceBroker.h.

Referenced by exportParameters(), monitoring(), and reset().

BUProxy* evf::FUResourceBroker::bu_

private

Definition at line 136 of file FUResourceBroker.h.

Referenced by configureResources(), and connectToBUandSM().

xdata::String evf::FUResourceBroker::buClassName_

private

Definition at line 210 of file FUResourceBroker.h.

Referenced by connectToBUandSM(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::buInstance_

private

Definition at line 211 of file FUResourceBroker.h.

Referenced by connectToBUandSM(), and exportParameters().

xdata::String evf::FUResourceBroker::class_

private

Definition at line 160 of file FUResourceBroker.h.

Referenced by exportParameters(), and FUResourceBroker().

xdata::String evf::FUResourceBroker::clientPrcIds_

private

Definition at line 197 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

const int evf::FUResourceBroker::CRC_ERROR_SHIFT = 0x0

static

Definition at line 100 of file FUResourceBroker.h.

const int evf::FUResourceBroker::DATA_ERROR_SHIFT = 0x1

static

Definition at line 101 of file FUResourceBroker.h.

xdata::UnsignedInteger32 evf::FUResourceBroker::dataErrorFlag_

private

Definition at line 192 of file FUResourceBroker.h.

Referenced by actionPerformed(), enabling(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::deltaN_

private

Definition at line 166 of file FUResourceBroker.h.

Referenced by exportParameters(), monitoring(), and reset().

xdata::UnsignedInteger32 evf::FUResourceBroker::deltaSumOfSizes_

private

Definition at line 168 of file FUResourceBroker.h.

Referenced by exportParameters(), monitoring(), and reset().

xdata::Double evf::FUResourceBroker::deltaSumOfSquares_

private

Definition at line 167 of file FUResourceBroker.h.

Referenced by exportParameters(), monitoring(), and reset().

xdata::Double evf::FUResourceBroker::deltaT_

private

Definition at line 165 of file FUResourceBroker.h.

Referenced by exportParameters(), monitoring(), and reset().

xdata::UnsignedInteger32 evf::FUResourceBroker::doCrcCheck_

private

Definition at line 207 of file FUResourceBroker.h.

Referenced by actionPerformed(), configureResources(), and exportParameters().

xdata::Boolean evf::FUResourceBroker::doDropEvents_

private

Definition at line 205 of file FUResourceBroker.h.

Referenced by exportParameters(), and I2O_FU_TAKE_Callback().

xdata::UnsignedInteger32 evf::FUResourceBroker::doDumpEvents_

private

Definition at line 208 of file FUResourceBroker.h.

Referenced by actionPerformed(), configureResources(), and exportParameters().

xdata::Boolean evf::FUResourceBroker::doFedIdCheck_

private

Definition at line 206 of file FUResourceBroker.h.

Referenced by actionPerformed(), configureResources(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::dqmCellSize_

private

Definition at line 203 of file FUResourceBroker.h.

Referenced by configureResources(), and exportParameters().

EvffedFillerRB* evf::FUResourceBroker::frb_

private

Definition at line 240 of file FUResourceBroker.h.

Referenced by configureResources(), configuring(), and halting().

evf::StateMachine evf::FUResourceBroker::fsm_

private

Definition at line 127 of file FUResourceBroker.h.

Referenced by configuring(), enabling(), exportParameters(), FUResourceBroker(), halting(), I2O_EVM_LUMISECTION_Callback(), I2O_FU_TAKE_Callback(), and stopping().

WebGUI* evf::FUResourceBroker::gui_

private

Definition at line 130 of file FUResourceBroker.h.

Referenced by exportParameters(), FUResourceBroker(), monitoring(), and reset().

toolbox::mem::Pool* evf::FUResourceBroker::i2oPool_

private

Definition at line 142 of file FUResourceBroker.h.

Referenced by connectToBUandSM(), and FUResourceBroker().

xdata::UnsignedInteger32 evf::FUResourceBroker::instance_

private

Definition at line 161 of file FUResourceBroker.h.

Referenced by exportParameters(), FUResourceBroker(), and instanceNumber().

toolbox::BSem evf::FUResourceBroker::lock_

private

Definition at line 239 of file FUResourceBroker.h.

Referenced by lock(), and unlock().

Logger evf::FUResourceBroker::log_

private

Definition at line 133 of file FUResourceBroker.h.

Referenced by configureResources(), configuring(), connectToBUandSM(), emergencyStop(), enabling(), FUResourceBroker(), halting(), I2O_EVM_LUMISECTION_Callback(), I2O_FU_TAKE_Callback(), stopping(), and watching().

const int evf::FUResourceBroker::LOST_ERROR_SHIFT = 0x2

static

Definition at line 102 of file FUResourceBroker.h.

xdata::UnsignedInteger32 evf::FUResourceBroker::monSleepSec_

private

Definition at line 216 of file FUResourceBroker.h.

Referenced by exportParameters(), and monitoring().

struct timeval evf::FUResourceBroker::monStartTime_

private

Definition at line 233 of file FUResourceBroker.h.

Referenced by monitoring(), and startMonitoringWorkLoop().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbAllocatedEvents_

private

Definition at line 176 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbAllocateSent_

private

Definition at line 226 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbClients_

private

Definition at line 196 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbCrcErrors_

private

Definition at line 189 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbDataDiscardReceived_

private

Definition at line 228 of file FUResourceBroker.h.

Referenced by exportParameters(), and I2O_FU_DATA_DISCARD_Callback().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbDataErrors_

private

Definition at line 188 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbDiscardedEvents_

private

Definition at line 185 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbDqmCells_

private

Definition at line 200 of file FUResourceBroker.h.

Referenced by configureResources(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbDqmDiscardReceived_

private

Definition at line 229 of file FUResourceBroker.h.

Referenced by exportParameters(), and I2O_FU_DQM_DISCARD_Callback().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbLostEvents_

private

Definition at line 187 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbPendingRequests_

private

Definition at line 177 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbPendingSMDiscards_

private

Definition at line 183 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbPendingSMDqmDiscards_

private

Definition at line 184 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbProcessedEvents_

private

Definition at line 179 of file FUResourceBroker.h.

xdata::UnsignedInteger32 evf::FUResourceBroker::nbRawCells_

private

Definition at line 198 of file FUResourceBroker.h.

Referenced by configureResources(), configuring(), exportParameters(), and watching().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbReceivedEvents_

private

Definition at line 178 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbRecoCells_

private

Definition at line 199 of file FUResourceBroker.h.

Referenced by configureResources(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbSentDqmEvents_

private

Definition at line 181 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbSentErrorEvents_

private

Definition at line 182 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbSentEvents_

private

Definition at line 180 of file FUResourceBroker.h.

Referenced by actionPerformed(), and exportParameters().

UInt_t evf::FUResourceBroker::nbSentLast_

private

Definition at line 234 of file FUResourceBroker.h.

Referenced by monitoring(), and reset().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbTakeReceived_

private

Definition at line 227 of file FUResourceBroker.h.

Referenced by exportParameters(), and I2O_FU_TAKE_Callback().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbTimeoutsWithEvent_

private

Definition at line 190 of file FUResourceBroker.h.

Referenced by actionPerformed(), emergencyStop(), enabling(), exportParameters(), and watching().

xdata::UnsignedInteger32 evf::FUResourceBroker::nbTimeoutsWithoutEvent_

private

Definition at line 191 of file FUResourceBroker.h.

Referenced by actionPerformed(), emergencyStop(), enabling(), exportParameters(), and watching().

xdata::Boolean evf::FUResourceBroker::processKillerEnabled_

private

Definition at line 219 of file FUResourceBroker.h.

Referenced by exportParameters(), and watching().

xdata::Double evf::FUResourceBroker::rate_

private

Definition at line 171 of file FUResourceBroker.h.

Referenced by exportParameters(), monitoring(), and reset().

xdata::UnsignedInteger32 evf::FUResourceBroker::rawCellSize_

private

Definition at line 201 of file FUResourceBroker.h.

Referenced by configureResources(), and exportParameters().

xdata::String evf::FUResourceBroker::reasonForFailed_

private

Definition at line 222 of file FUResourceBroker.h.

Referenced by configuring(), emergencyStop(), enabling(), exportParameters(), halting(), and stopping().

xdata::UnsignedInteger32 evf::FUResourceBroker::recoCellSize_

private

Definition at line 202 of file FUResourceBroker.h.

Referenced by configureResources(), and exportParameters().

FUResourceTable* evf::FUResourceBroker::resourceTable_

private

Definition at line 145 of file FUResourceBroker.h.

Referenced by actionPerformed(), configureResources(), configuring(), emergencyStop(), enabling(), halting(), I2O_EVM_LUMISECTION_Callback(), I2O_FU_DATA_DISCARD_Callback(), I2O_FU_DQM_DISCARD_Callback(), I2O_FU_TAKE_Callback(), monitoring(), stopping(), and watching().

xdata::Double evf::FUResourceBroker::rms_

private

Definition at line 173 of file FUResourceBroker.h.

Referenced by exportParameters(), monitoring(), and reset().

xdata::UnsignedInteger32 evf::FUResourceBroker::runNumber_

private

Definition at line 162 of file FUResourceBroker.h.

Referenced by emergencyStop(), enabling(), exportParameters(), and watching().

xdata::Boolean evf::FUResourceBroker::segmentationMode_

private

Definition at line 195 of file FUResourceBroker.h.

Referenced by configureResources(), and exportParameters().

const int evf::FUResourceBroker::SENT_ERREVENT_ERROR_SHIFT = 0x5

static

Definition at line 105 of file FUResourceBroker.h.

bool evf::FUResourceBroker::shmInconsistent_

private

Definition at line 241 of file FUResourceBroker.h.

Referenced by configureResources(), configuring(), emergencyStop(), and watching().

xdata::UnsignedInteger32 evf::FUResourceBroker::shmResourceTableTimeout_

private

Definition at line 215 of file FUResourceBroker.h.

Referenced by configureResources(), exportParameters(), and stopping().

SMProxy* evf::FUResourceBroker::sm_

private

Definition at line 139 of file FUResourceBroker.h.

Referenced by configureResources(), and connectToBUandSM().

xdata::String evf::FUResourceBroker::smClassName_

private

Definition at line 212 of file FUResourceBroker.h.

Referenced by connectToBUandSM(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::smInstance_

private

Definition at line 213 of file FUResourceBroker.h.

Referenced by connectToBUandSM(), and exportParameters().

std::string evf::FUResourceBroker::sourceId_

private

Definition at line 156 of file FUResourceBroker.h.

Referenced by connectToBUandSM(), FUResourceBroker(), startMonitoringWorkLoop(), and startWatchingWorkLoop().

UInt_t evf::FUResourceBroker::sumOfSizesLast_

private

Definition at line 236 of file FUResourceBroker.h.

Referenced by monitoring(), and reset().

uint64_t evf::FUResourceBroker::sumOfSquaresLast_

private

Definition at line 235 of file FUResourceBroker.h.

Referenced by monitoring(), and reset().

xdata::Double evf::FUResourceBroker::throughput_

private

Definition at line 170 of file FUResourceBroker.h.

Referenced by exportParameters(), monitoring(), and reset().

const int evf::FUResourceBroker::TIMEOUT_EVENT_ERROR_SHIFT = 0x4

static

Definition at line 104 of file FUResourceBroker.h.

const int evf::FUResourceBroker::TIMEOUT_NOEVENT_ERROR_SHIFT = 0x3

static

Definition at line 103 of file FUResourceBroker.h.

xdata::UnsignedInteger32 evf::FUResourceBroker::timeOutSec_

private

Definition at line 218 of file FUResourceBroker.h.

Referenced by exportParameters(), and watching().

xdata::String evf::FUResourceBroker::url_

private

Definition at line 159 of file FUResourceBroker.h.

Referenced by exportParameters(), and FUResourceBroker().

xdata::Boolean evf::FUResourceBroker::useEvmBoard_

private

Definition at line 220 of file FUResourceBroker.h.

Referenced by actionPerformed(), configureResources(), and exportParameters().

xdata::UnsignedInteger32 evf::FUResourceBroker::watchSleepSec_

private

Definition at line 217 of file FUResourceBroker.h.

Referenced by exportParameters(), and watching().

toolbox::task::WorkLoop* evf::FUResourceBroker::wlMonitoring_

private

Definition at line 148 of file FUResourceBroker.h.

Referenced by startMonitoringWorkLoop().

toolbox::task::WorkLoop* evf::FUResourceBroker::wlWatching_

private

Definition at line 152 of file FUResourceBroker.h.

Referenced by startWatchingWorkLoop().