#include <FUResourceBroker.h>

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_
xdata::UnsignedInteger32	highestEolReceived_
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	nbEolDiscarded_
xdata::UnsignedInteger32	nbEolPosted_
xdata::UnsignedInteger32	nbLostEvents_
xdata::UnsignedInteger32	nbPendingRequests_
xdata::UnsignedInteger32	nbPendingSMDiscards_
xdata::UnsignedInteger32	nbPendingSMDqmDiscards_
xdata::UnsignedInteger32	nbProcessedEvents_
xdata::UnsignedInteger32	nbRawCells_
xdata::UnsignedInteger32	nbReceivedEol_
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(), 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_, mergeVDriftHistosByStation::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)
  , nbReceivedEol_(0)
  , highestEolReceived_(0)
  , nbEolPosted_(0)
  , nbEolDiscarded_(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 197 of file FUResourceBroker.cc.

{

}

Member Function Documentation

void FUResourceBroker::actionPerformed ( xdata::Event & e )

Definition at line 483 of file FUResourceBroker.cc.

{
  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();
      nbEolPosted_        =resourceTable_->nbEolPosted();
      nbEolDiscarded_     =resourceTable_->nbEolDiscarded();
      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 1051 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 209 of file FUResourceBroker.cc.

References configureResources(), connectToBUandSM(), 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 434 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 866 of file FUResourceBroker.cc.

References gather_cfg::cout, asciidump::els, i, recoMuon::in, CommonMethods::lock(), dbtoconf::out, evf::utils::pid, 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>"<<i<<"</td>"
          <<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 844 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 1017 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 238 of file FUResourceBroker.cc.

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

{
  try {
    LOG4CPLUS_INFO(log_, "Start enabling ...");
    reset();
    startMonitoringWorkLoop();
    startWatchingWorkLoop();
    resourceTable_->setRunNumber(runNumber_);
    lock();
    resourceTable_->resetCounters();
    unlock();
    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 737 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_, highestEolReceived_, instance_, monSleepSec_, nbAllocatedEvents_, nbAllocateSent_, nbClients_, nbCrcErrors_, nbDataDiscardReceived_, nbDataErrors_, nbDiscardedEvents_, nbDqmCells_, nbDqmDiscardReceived_, nbEolDiscarded_, nbEolPosted_, nbLostEvents_, nbPendingRequests_, nbPendingSMDiscards_, nbPendingSMDqmDiscards_, nbRawCells_, nbReceivedEol_, 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("nbReceivedEol",          &nbReceivedEol_);
  gui_->addMonitorCounter("highestEolReceived",     &highestEolReceived_);
  gui_->addMonitorCounter("nbEolPosted",            &nbEolPosted_);
  gui_->addMonitorCounter("nbEolDiscarded",         &nbEolDiscarded_);

  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 359 of file FUResourceBroker.cc.

References runTheMatrix::msg.

{
  return fsm_.commandCallback(msg);
}

bool FUResourceBroker::halting ( toolbox::task::WorkLoop * wl )

Definition at line 308 of file FUResourceBroker.cc.

References prof2calltree::count, 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 382 of file FUResourceBroker.cc.

References evf::StateMachine::fireFailed(), fsm_, highestEolReceived_, log_, runTheMatrix::msg, nbReceivedEol_, 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);
  }
  nbReceivedEol_++;
  if(highestEolReceived_.value_+100 < msg->lumiSection) 
    {
      LOG4CPLUS_ERROR(log_,"EOL message not in sequence, expected " 
                      << highestEolReceived_.value_+1
                      << " received " << msg->lumiSection);
      fsm_.fireFailed("EOL message with corrupted LS ",this);
    }
  if(highestEolReceived_.value_+1 != msg->lumiSection) 
    LOG4CPLUS_WARN(log_,"EOL message not in sequence, expected " 
                    << highestEolReceived_.value_+1
                    << " received " << msg->lumiSection);

  if(highestEolReceived_.value_ < msg->lumiSection) 
    highestEolReceived_.value_ = msg->lumiSection;
  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 418 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 426 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 367 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 ) [inline, private]

Definition at line 117 of file FUResourceBroker.h.

References lock_.

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

{ lock_.take(); }

bool FUResourceBroker::monitoring ( toolbox::task::WorkLoop * wl )

Definition at line 573 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 823 of file FUResourceBroker.cc.

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

Referenced by configureResources(), and enabling().

{
  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 551 of file FUResourceBroker.cc.

References asMonitoring_, 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 656 of file FUResourceBroker.cc.

References asWatching_, 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 272 of file FUResourceBroker.cc.

References gather_cfg::cout, emergencyStop(), 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 ) [inline, private]

Definition at line 118 of file FUResourceBroker.h.

References lock_.

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

{ lock_.give(); }

bool FUResourceBroker::watching ( toolbox::task::WorkLoop * wl )

Definition at line 675 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 473 of file FUResourceBroker.cc.

References recoMuon::in, mergeVDriftHistosByStation::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);
}