#include <FUResourceTable.h>

Inheritance diagram for evf::FUResourceTable:

Public Member Functions
UInt_t	allocateResource ()

bool	buildResource (MemRef_t *bufRef)

std::vector< UInt_t >	cellEvtNumbers () const

std::vector< pid_t >	cellPrcIds () const

std::vector< std::string >	cellStates () const

std::vector< time_t >	cellTimeStamps () const

void	clear ()

std::vector< pid_t >	clientPrcIds () const

std::string	clientPrcIdsAsString () const

bool	discard (toolbox::task::WorkLoop *workLoop)

bool	discardDataEvent (MemRef_t *bufRef)

bool	discardDqmEvent (MemRef_t *bufRef)

std::vector< std::string >	dqmCellStates () const

void	dropEvent ()

void	dumpEvent (evf::FUShmRawCell *cell)

	FUResourceTable (bool segmentationMode, UInt_t nbRawCells, UInt_t nbRecoCells, UInt_t nbDqmCells, UInt_t rawCellSize, UInt_t recoCellSize, UInt_t dqmCellSize, BUProxy bu, SMProxy sm, log4cplus::Logger logger, unsigned int, EvffedFillerRB frb, xdaq::Application ) throw (evf::Exception)

void	halt ()

bool	handleCrashedEP (UInt_t runNumber, pid_t pid)

void	initialize (bool segmentationMode, UInt_t nbRawCells, UInt_t nbRecoCells, UInt_t nbDqmCells, UInt_t rawCellSize, UInt_t recoCellSize, UInt_t dqmCellSize) throw (evf::Exception)

void	injectCRCError ()

bool	isActive () const

bool	isLastMessageOfEvent (MemRef_t *bufRef)

bool	isReadyToShutDown () const

void	lastResort ()

void	lock ()

UInt_t	nbAllocated () const

UInt_t	nbAllocSent () const

UInt_t	nbClients () const

UInt_t	nbCompleted () const

UInt_t	nbCrcErrors () const

UInt_t	nbDiscarded () const

UInt_t	nbEolDiscarded () const

UInt_t	nbEolPosted () const

UInt_t	nbErrors () const

UInt_t	nbFreeSlots () const

UInt_t	nbLost () const

UInt_t	nbPending () const

UInt_t	nbPendingSMDiscards () const

UInt_t	nbPendingSMDqmDiscards () const

UInt_t	nbResources () const

UInt_t	nbSent () const

UInt_t	nbSentDqm () const

UInt_t	nbSentError () const

void	postEndOfLumiSection (MemRef_t *bufRef)

void	printWorkLoopStatus ()

void	resetCounters ()

void	sendAllocate ()

bool	sendData (toolbox::task::WorkLoop *workLoop)

void	sendDataEvent (UInt_t fuResourceId, UInt_t runNumber, UInt_t evtNumber, UInt_t outModId, UInt_t fuProcessId, UInt_t fuGuid, UChar_t *data, UInt_t dataSize)

void	sendDiscard (UInt_t buResourceId)

bool	sendDqm (toolbox::task::WorkLoop *workLoop)

void	sendDqmEvent (UInt_t fuDqmId, UInt_t runNumber, UInt_t evtAtUpdate, UInt_t folderId, UInt_t fuProcessId, UInt_t fuGuid, UChar_t *data, UInt_t dataSize)

void	sendErrorEvent (UInt_t fuResourceId, UInt_t runNumber, UInt_t evtNumber, UInt_t fuProcessId, UInt_t fuGuid, UChar_t *data, UInt_t dataSize)

void	sendInitMessage (UInt_t fuResourceId, UInt_t outModId, UInt_t fuProcessId, UInt_t fuGuid, UChar_t *data, UInt_t dataSize)

void	setDoCrcCheck (UInt_t doCrcCheck)

void	setDoDumpEvents (UInt_t doDumpEvents)

void	setRunNumber (UInt_t runNumber)

void	shutDownClients ()

void	startDiscardWorkLoop () throw (evf::Exception)

void	startSendDataWorkLoop () throw (evf::Exception)

void	startSendDqmWorkLoop () throw (evf::Exception)

void	stop ()

UInt_t	sumOfSizes () const

uint64_t	sumOfSquares () const

void	unlock ()

virtual	~FUResourceTable ()

Private Types
typedef toolbox::task::ActionSignature	ActionSignature_t

typedef toolbox::task::WorkLoop	WorkLoop_t

Private Attributes
bool *	acceptSMDataDiscard_

int *	acceptSMDqmDiscard_

xdaq::Application *	app_

ActionSignature_t *	asDiscard_

ActionSignature_t *	asSendData_

ActionSignature_t *	asSendDqm_

BUProxy *	bu_

UInt_t	doCrcCheck_

UInt_t	doDumpEvents_

EvffedFillerRB *	frb_

std::queue< UInt_t >	freeResourceIds_

bool	isActive_

bool	isHalting_

bool	isReadyToShutDown_

bool	isStopping_

toolbox::BSem	lock_

log4cplus::Logger	log_

UInt_t	nbAllocated_

UInt_t	nbAllocSent_

UInt_t	nbClientsToShutDown_

UInt_t	nbCompleted_

UInt_t	nbCrcErrors_

UInt_t	nbDiscarded_

UInt_t	nbDqmCells_

UInt_t	nbEolDiscarded_

UInt_t	nbEolPosted_

UInt_t	nbErrors_

UInt_t	nbLost_

UInt_t	nbPending_

UInt_t	nbPendingSMDiscards_

UInt_t	nbPendingSMDqmDiscards_

UInt_t	nbRawCells_

UInt_t	nbRecoCells_

UInt_t	nbSent_

UInt_t	nbSentDqm_

UInt_t	nbSentError_

FUResourceVec_t	resources_

UInt_t	runNumber_

FUShmBuffer *	shmBuffer_

unsigned int	shutdownTimeout_

SMProxy *	sm_

UInt_t	sumOfSizes_

uint64_t	sumOfSquares_

WorkLoop_t *	wlDiscard_

WorkLoop_t *	wlSendData_

WorkLoop_t *	wlSendDqm_

Detailed Description

Definition at line 28 of file FUResourceTable.h.

Member Typedef Documentation

typedef toolbox::task::ActionSignature evf::FUResourceTable::ActionSignature_t

private

Definition at line 213 of file FUResourceTable.h.

typedef toolbox::task::WorkLoop evf::FUResourceTable::WorkLoop_t

private

Definition at line 212 of file FUResourceTable.h.

Constructor & Destructor Documentation

FUResourceTable::FUResourceTable	(	bool	segmentationMode,
		UInt_t	nbRawCells,
		UInt_t	nbRecoCells,
		UInt_t	nbDqmCells,
		UInt_t	rawCellSize,
		UInt_t	recoCellSize,
		UInt_t	dqmCellSize,
		BUProxy *	bu,
		SMProxy *	sm,
		log4cplus::Logger	logger,
		unsigned int	timeout,
		EvffedFillerRB *	frb,
		xdaq::Application *	app
	)
throw	(	evf::Exception
	)

Definition at line 34 of file FUResourceTable.cc.

   : bu_(bu)
   , sm_(sm)
   , log_(logger)
   , wlSendData_(0)
   , asSendData_(0)
   , wlSendDqm_(0)
   , asSendDqm_(0)
   , wlDiscard_(0)
   , asDiscard_(0)
   , shmBuffer_(0)
   , nbDqmCells_(nbDqmCells)
   , nbRawCells_(nbRawCells)
   , nbRecoCells_(nbRecoCells)
   , acceptSMDataDiscard_(0)
   , acceptSMDqmDiscard_(0)
   , doCrcCheck_(1)
   , shutdownTimeout_(timeout)
   , nbPending_(0)
   , nbClientsToShutDown_(0)
   , isReadyToShutDown_(true)
   , isActive_(false)
   , isHalting_(false)
   , isStopping_(false)
   , runNumber_(0xffffffff)
   , lock_(toolbox::BSem::FULL)
   , frb_(frb)
   , app_(app)
 {
   initialize(segmentationMode,
          nbRawCells,nbRecoCells,nbDqmCells,
          rawCellSize,recoCellSize,dqmCellSize);
 }

FUResourceTable::~FUResourceTable ( )

virtual

Definition at line 83 of file FUResourceTable.cc.

References evf::FUShmBuffer::releaseSharedMemory().

 {
   clear();
   if(wlSendData_){
     wlSendData_->cancel();
     toolbox::task::getWorkLoopFactory()->removeWorkLoop("SendData","waiting");
   }
   if(wlSendDqm_){
     wlSendDqm_->cancel();
     toolbox::task::getWorkLoopFactory()->removeWorkLoop("SendDqm","waiting");
   }
   if(wlDiscard_){
     wlDiscard_->cancel();
     toolbox::task::getWorkLoopFactory()->removeWorkLoop("Discard","waiting");
   }
   shmdt(shmBuffer_);
   if (FUShmBuffer::releaseSharedMemory())
     LOG4CPLUS_INFO(log_,"SHARED MEMORY SUCCESSFULLY RELEASED.");
   if (0!=acceptSMDataDiscard_) delete [] acceptSMDataDiscard_;
   if (0!= acceptSMDqmDiscard_) delete [] acceptSMDqmDiscard_;
 }

Member Function Documentation

UInt_t FUResourceTable::allocateResource ( )

Definition at line 442 of file FUResourceTable.cc.

References CommonMethods::lock().

 {
   assert(!freeResourceIds_.empty());
 
   lock();
   UInt_t fuResourceId=freeResourceIds_.front();
   freeResourceIds_.pop();
   nbPending_++;
   nbAllocated_++;
   unlock();
   
   return fuResourceId;
 }

bool FUResourceTable::buildResource ( MemRef_t * bufRef )

Definition at line 458 of file FUResourceTable.cc.

References evf::FUResource::allocate(), createPayload::block, evf::FUResource::doCrcCheck(), dumpEvent(), evf::FUResource::fatalError(), evf::FUResource::isAllocated(), evf::FUResource::isComplete(), CommonMethods::lock(), evf::FUResource::nbCrcErrors(), evf::FUResource::nbErrors(), cmsPerfSuiteHarvest::now, evf::FUResource::process(), evf::FUResource::release(), and evf::FUResource::shmCell().

Referenced by evf::FUResourceBroker::I2O_FU_TAKE_Callback().

 {
   bool eventComplete=false;
   
   I2O_EVENT_DATA_BLOCK_MESSAGE_FRAME *block=
     (I2O_EVENT_DATA_BLOCK_MESSAGE_FRAME*)bufRef->getDataLocation();
   
   UInt_t      fuResourceId=(UInt_t)block->fuTransactionId;
   UInt_t      buResourceId=(UInt_t)block->buResourceId;
   FUResource* resource    =resources_[fuResourceId];
   
   // allocate resource
   if (!resource->fatalError()&&!resource->isAllocated()) {
     FUShmRawCell* cell=shmBuffer_->rawCellToWrite();
     resource->allocate(cell);
     timeval now;
     gettimeofday(&now,0);
 
     frb_->setRBTimeStamp(((uint64_t)(now.tv_sec) << 32) + (uint64_t)(now.tv_usec));
 
     frb_->setRBEventCount(nbCompleted_);
 
     if (doCrcCheck_>0&&0==nbAllocated_%doCrcCheck_)  resource->doCrcCheck(true);
     else                                             resource->doCrcCheck(false);
   }
   
   
   // keep building this resource if it is healthy
   if (!resource->fatalError()) {
     resource->process(bufRef);
     lock();
     nbErrors_   +=resource->nbErrors();
     nbCrcErrors_+=resource->nbCrcErrors();
     unlock();
     
     // make resource available for pick-up
     if (resource->isComplete()) {
       lock();
       nbCompleted_++;
       nbPending_--;
       unlock();
       if (doDumpEvents_>0&&nbCompleted_%doDumpEvents_==0)
     dumpEvent(resource->shmCell());
       shmBuffer_->finishWritingRawCell(resource->shmCell());
       eventComplete=true;
     }
     
   }
   // bad event, release msg, and the whole resource if this was the last one
   //else {
   if (resource->fatalError()) {
     bool lastMsg=isLastMessageOfEvent(bufRef);
     if (lastMsg) {
       shmBuffer_->releaseRawCell(resource->shmCell());
       resource->release();
       lock();
       freeResourceIds_.push(fuResourceId);
       nbDiscarded_++;
       nbLost_++;
       nbPending_--;
       unlock();
       bu_->sendDiscard(buResourceId);
       sendAllocate();
     }
     bufRef->release(); // this should now be safe re: appendToSuperFrag as corrupted blocks will be removed... 
   }
   
   return eventComplete;
 }

vector< UInt_t > FUResourceTable::cellEvtNumbers ( ) const

Definition at line 847 of file FUResourceTable.cc.

References i, n, and query::result.

Referenced by evf::FUResourceBroker::watching().

 {
   vector<UInt_t> result;
   if (0!=shmBuffer_) {
     UInt_t n=nbResources();
     shmBuffer_->lock();
     for (UInt_t i=0;i<n;i++) result.push_back(shmBuffer_->evtNumber(i));
     shmBuffer_->unlock();
   }
   return result;
 }

vector< pid_t > FUResourceTable::cellPrcIds ( ) const

Definition at line 861 of file FUResourceTable.cc.

References i, n, and query::result.

Referenced by evf::FUResourceBroker::watching().

 {
   vector<pid_t> result;
   if (0!=shmBuffer_) {
     UInt_t n=nbResources();
     shmBuffer_->lock();
     for (UInt_t i=0;i<n;i++) result.push_back(shmBuffer_->evtPrcId(i));
     shmBuffer_->unlock();
   }
   return result;
 }

vector< string > FUResourceTable::cellStates ( ) const

Definition at line 796 of file FUResourceTable.cc.

References evf::evt::DISCARDING, evf::evt::EMPTY, i, evf::evt::LUMISECTION, n, evf::evt::PROCESSED, evf::evt::PROCESSING, evf::evt::RAWREAD, evf::evt::RAWREADING, evf::evt::RAWWRITING, evf::evt::RAWWRITTEN, evf::evt::RECOWRITING, evf::evt::RECOWRITTEN, query::result, evf::evt::SENDING, evf::evt::SENT, evf::utils::state, evf::evt::STOP, and evf::evt::USEDLS.

 {
   vector<string> result;
   if (0!=shmBuffer_) {
     UInt_t n=nbResources();
     shmBuffer_->lock();
     for (UInt_t i=0;i<n;i++) {
       evt::State_t state=shmBuffer_->evtState(i);
       if      (state==evt::EMPTY)      result.push_back("EMPTY");
       else if (state==evt::STOP)       result.push_back("STOP");
       else if (state==evt::LUMISECTION)result.push_back("LUMISECTION");
       else if (state==evt::USEDLS)     result.push_back("USEDLS");
       else if (state==evt::RAWWRITING) result.push_back("RAWWRITING");
       else if (state==evt::RAWWRITTEN) result.push_back("RAWWRITTEN");
       else if (state==evt::RAWREADING) result.push_back("RAWREADING");
       else if (state==evt::RAWREAD)    result.push_back("RAWREAD");
       else if (state==evt::PROCESSING) result.push_back("PROCESSING");
       else if (state==evt::PROCESSED)  result.push_back("PROCESSED");
       else if (state==evt::RECOWRITING)result.push_back("RECOWRITING");
       else if (state==evt::RECOWRITTEN)result.push_back("RECOWRITTEN");
       else if (state==evt::SENDING)    result.push_back("SENDING");
       else if (state==evt::SENT)       result.push_back("SENT");
       else if (state==evt::DISCARDING) result.push_back("DISCARDING");
     }
     shmBuffer_->unlock();
   }
   return result;
 }

vector< time_t > FUResourceTable::cellTimeStamps ( ) const

Definition at line 875 of file FUResourceTable.cc.

References i, n, and query::result.

Referenced by evf::FUResourceBroker::watching().

 {
   vector<time_t> result;
   if (0!=shmBuffer_) {
     UInt_t n=nbResources();
     shmBuffer_->lock();
     for (UInt_t i=0;i<n;i++) result.push_back(shmBuffer_->evtTimeStamp(i));
     shmBuffer_->unlock();
   }
   return result;
 }

void FUResourceTable::clear ( void )

Definition at line 718 of file FUResourceTable.cc.

References i.

Referenced by python.Vispa.Views.WidgetView.WidgetView::closeEvent(), python.Vispa.Views.BoxDecayView.BoxDecayView::closeEvent(), python.Vispa.Share.FindAlgorithm.FindAlgorithm::findUsingFindDialog(), python.Vispa.Views.LineDecayView.LineDecayView::setDataObjects(), python.Vispa.Views.WidgetView.WidgetView::setDataObjects(), python.Vispa.Views.TreeView.TreeView::updateContent(), python.Vispa.Views.TableView.TableView::updateContent(), python.Vispa.Views.BoxDecayView.BoxDecayView::updateContent(), and python.Vispa.Views.PropertyView.PropertyView::updateContent().

 {
   for (UInt_t i=0;i<resources_.size();i++) {
     resources_[i]->release();
     delete resources_[i];
   }
   resources_.clear();
   while (!freeResourceIds_.empty()) freeResourceIds_.pop();
 }

vector< pid_t > FUResourceTable::clientPrcIds ( ) const

Definition at line 769 of file FUResourceTable.cc.

References i, n, and query::result.

Referenced by evf::FUResourceBroker::emergencyStop(), and evf::FUResourceBroker::watching().

 {
   vector<pid_t> result;
   if (0!=shmBuffer_) {
     UInt_t n=nbClients();
     for (UInt_t i=0;i<n;i++) result.push_back(shmBuffer_->clientPrcId(i));
   }
   return result;
 }

string FUResourceTable::clientPrcIdsAsString ( ) const

Definition at line 781 of file FUResourceTable.cc.

References i, and n.

Referenced by evf::FUResourceBroker::actionPerformed().

 {
   stringstream ss;
   if (0!=shmBuffer_) {
     UInt_t n=nbClients();
     for (UInt_t i=0;i<n;i++) {
       if (i>0) ss<<",";
       ss<<shmBuffer_->clientPrcId(i);
     }
   }
   return ss.str();
 }

bool FUResourceTable::discard ( toolbox::task::WorkLoop * workLoop )

Definition at line 337 of file FUResourceTable.cc.

References evf::FUShmRawCell::buResourceId(), prof2calltree::count, gather_cfg::cout, evf::dqm::EMPTY, evf::FUShmRawCell::fuResourceId(), i, evf::FUShmRawCell::index(), CommonMethods::lock(), n, evf::FUShmBuffer::shm_nattch(), evf::utils::state, evf::evt::STOP, and evf::evt::USEDLS.

Referenced by startDiscardWorkLoop().

 {
   FUShmRawCell* cell =shmBuffer_->rawCellToDiscard();
   evt::State_t  state=shmBuffer_->evtState(cell->index());
 
   bool   reschedule  =true;
   bool   shutDown    =(state==evt::STOP);
   bool   isLumi      =(state==evt::USEDLS);
   UInt_t fuResourceId=cell->fuResourceId();
   UInt_t buResourceId=cell->buResourceId();
 
   //  std::cout << "discard loop, state, shutDown, isLumi " << state << " " 
   //        << shutDown << " " << isLumi << std::endl;
   //  std::cout << "resource ids " << fuResourceId << " " << buResourceId << std::endl;
 
   if (shutDown) {
     LOG4CPLUS_INFO(log_,"nbClientsToShutDown = "<<nbClientsToShutDown_);
     if (nbClientsToShutDown_>0) --nbClientsToShutDown_;
     if (nbClientsToShutDown_==0) {
       LOG4CPLUS_INFO(log_,"Don't reschedule discard-workloop.");
       isActive_ =false;
       reschedule=false;
     }
   }
   
   shmBuffer_->discardRawCell(cell);
   if(isLumi) nbEolDiscarded_++;
 
   if (!shutDown && !isLumi) {
     if(fuResourceId >= nbResources()){
       LOG4CPLUS_WARN(log_,"cell " << cell->index() << " in state " << state 
              << " scheduled for discard has no associated FU resource ");
     }
     else{
       resources_[fuResourceId]->release();
       lock();
       freeResourceIds_.push(fuResourceId);
       assert(freeResourceIds_.size()<=resources_.size());
       unlock();
     
       if (!isHalting_) {
     sendDiscard(buResourceId);
     if(!isStopping_)sendAllocate();
       }
     }
   }
   
   if (!reschedule) {
     std::cout << " entered shutdown cycle " << std::endl;
     shmBuffer_->writeRecoEmptyEvent();
     UInt_t count=0;
     while (count<100) {
       std::cout << " shutdown cycle " <<shmBuffer_->nClients() << " "
         <<  FUShmBuffer::shm_nattch(shmBuffer_->shmid()) << std::endl;
       if (shmBuffer_->nClients()==0&&
       FUShmBuffer::shm_nattch(shmBuffer_->shmid())==1) {
     //  isReadyToShutDown_ = true;
     break;
       }
       else {
     count++;
     std::cout << " shutdown cycle attempt " << count << std::endl;
     LOG4CPLUS_DEBUG(log_,"FUResourceTable: Wait for all clients to detach,"
             <<" nClients="<<shmBuffer_->nClients()
             <<" nattch="<<FUShmBuffer::shm_nattch(shmBuffer_->shmid())
             <<" ("<<count<<")");
     ::usleep(shutdownTimeout_);
     if(count*shutdownTimeout_ > 10000000)
       LOG4CPLUS_WARN(log_,"FUResourceTable:LONG Wait (>10s) for all clients to detach,"
               <<" nClients="<<shmBuffer_->nClients()
               <<" nattch="<<FUShmBuffer::shm_nattch(shmBuffer_->shmid())
               <<" ("<<count<<")");
 
       }
     }
     bool allEmpty = false;
     std::cout << "Checking if all dqm cells are empty " << std::endl;
     while(!allEmpty){
       UInt_t n=nbDqmCells_;
       allEmpty = true;
       shmBuffer_->lock();
       for (UInt_t i=0;i<n;i++) {
     dqm::State_t state=shmBuffer_->dqmState(i);
     if(state!=dqm::EMPTY) allEmpty = false; 
       } 
       shmBuffer_->unlock();
     }
     std::cout << "Making sure there are no dqm pending discards " << std::endl;
     if(nbPendingSMDqmDiscards_ != 0)
       {
     LOG4CPLUS_WARN(log_,"FUResourceTable: pending DQM discards not zero: ="
                << nbPendingSMDqmDiscards_ << " while cells are all empty. This may cause problems at next start ");
 
       }
     shmBuffer_->writeDqmEmptyEvent();
     isReadyToShutDown_ = true; // moved here from within the first while loop to make sure the 
                                // sendDqm loop has been shut down as well
   }
   
   return reschedule;
 }

bool FUResourceTable::discardDataEvent ( MemRef_t * bufRef )

Definition at line 530 of file FUResourceTable.cc.

References CommonMethods::lock(), runTheMatrix::msg, and _I2O_FU_DATA_DISCARD_MESSAGE_FRAME::rbBufferID.

Referenced by evf::FUResourceBroker::I2O_FU_DATA_DISCARD_Callback().

 {
   I2O_FU_DATA_DISCARD_MESSAGE_FRAME *msg;
   msg=(I2O_FU_DATA_DISCARD_MESSAGE_FRAME*)bufRef->getDataLocation();
   UInt_t recoIndex=msg->rbBufferID;
   
   if (acceptSMDataDiscard_[recoIndex]) {
     lock();
     nbPendingSMDiscards_--;
     unlock();
     acceptSMDataDiscard_[recoIndex] = false;
     
     if (!isHalting_) {
       shmBuffer_->discardRecoCell(recoIndex);
       bufRef->release();
     }
   }
   else {
     LOG4CPLUS_ERROR(log_,"Spurious DATA discard by StorageManager, skip!");
   }
   
   if (isHalting_) {
     bufRef->release();
     return false;
   }
   
   return true;
 }

bool FUResourceTable::discardDqmEvent ( MemRef_t * bufRef )

Definition at line 561 of file FUResourceTable.cc.

References runTheMatrix::msg, _I2O_FU_DQM_DISCARD_MESSAGE_FRAME::rbBufferID, and evf::dqm::SENT.

Referenced by evf::FUResourceBroker::I2O_FU_DQM_DISCARD_Callback().

 {
   I2O_FU_DQM_DISCARD_MESSAGE_FRAME *msg;
   msg=(I2O_FU_DQM_DISCARD_MESSAGE_FRAME*)bufRef->getDataLocation();
   UInt_t dqmIndex=msg->rbBufferID;
   unsigned int ntries = 0;
   while(shmBuffer_->dqmState(dqmIndex)!=dqm::SENT){
     LOG4CPLUS_WARN(log_,"DQM discard for cell "<< dqmIndex << " which is not yer in SENT state - waiting");
     ::usleep(10000);
     if(ntries++>10){
       LOG4CPLUS_ERROR(log_,"DQM cell " << dqmIndex 
               << " discard timed out while cell still in state " << shmBuffer_->dqmState(dqmIndex) );
       bufRef->release();
       return true;
     }
   }
   if (acceptSMDqmDiscard_[dqmIndex]>0) {
     acceptSMDqmDiscard_[dqmIndex]--;
     if(nbPendingSMDqmDiscards_>0){
       nbPendingSMDqmDiscards_--;
     }
     else {
       LOG4CPLUS_WARN(log_,"Spurious??? DQM discard by StorageManager, index " << dqmIndex 
              << " cell state " << shmBuffer_->dqmState(dqmIndex) << " accept flag " << acceptSMDqmDiscard_[dqmIndex];);
     }
     
     if (!isHalting_) {
       shmBuffer_->discardDqmCell(dqmIndex);
       bufRef->release();
     }
 
   }
   else {
     LOG4CPLUS_ERROR(log_,"Spurious DQM discard for cell " << dqmIndex 
             << " from StorageManager while cell is not accepting discards");
   }
   
   if (isHalting_) {
     bufRef->release();
     return false;
   }
   
   return true;
 }

vector< string > FUResourceTable::dqmCellStates ( ) const

Definition at line 825 of file FUResourceTable.cc.

References evf::dqm::DISCARDING, evf::dqm::EMPTY, i, n, query::result, evf::dqm::SENDING, evf::dqm::SENT, evf::utils::state, evf::dqm::WRITING, and evf::dqm::WRITTEN.

 {
   vector<string> result;
   if (0!=shmBuffer_) {
     UInt_t n=nbDqmCells_;
     shmBuffer_->lock();
     for (UInt_t i=0;i<n;i++) {
       dqm::State_t state=shmBuffer_->dqmState(i);
       if      (state==dqm::EMPTY)      result.push_back("EMPTY");
       else if (state==dqm::WRITING) result.push_back("WRITING");
       else if (state==dqm::WRITTEN) result.push_back("WRITTEN");
       else if (state==dqm::SENDING)    result.push_back("SENDING");
       else if (state==dqm::SENT)       result.push_back("SENT");
       else if (state==dqm::DISCARDING) result.push_back("DISCARDING");
     }
     shmBuffer_->unlock();
   }
   return result;
 }

void FUResourceTable::dropEvent ( )

Definition at line 624 of file FUResourceTable.cc.

References evf::FUShmRawCell::fuResourceId().

Referenced by evf::FUResourceBroker::I2O_FU_TAKE_Callback().

 {
   FUShmRawCell* cell=shmBuffer_->rawCellToRead();
   UInt_t fuResourceId=cell->fuResourceId();
   shmBuffer_->finishReadingRawCell(cell);
   shmBuffer_->scheduleRawCellForDiscard(fuResourceId);
 }

void FUResourceTable::dumpEvent ( evf::FUShmRawCell * cell )

Definition at line 653 of file FUResourceTable.cc.

References evf::FUShmRawCell::evtNumber(), evf::FUShmRawCell::fedAddr(), evf::FUShmRawCell::fedSize(), makeHLTPrescaleTable::fout, i, j, and evf::FUShmRawCell::nFed().

 {
   ostringstream oss; oss<<"/tmp/evt"<<cell->evtNumber()<<".dump";
   ofstream fout(oss.str().c_str());
   fout.fill('0');
 
   fout<<"#\n# evt "<<cell->evtNumber()<<"\n#\n"<<endl;
   for (unsigned int i=0;i<cell->nFed();i++) {
     if (cell->fedSize(i)==0) continue;
     fout<<"# fedid "<<i<<endl;
     unsigned char* addr=cell->fedAddr(i);
     for (unsigned int j=0;j<cell->fedSize(i);j++) {
       fout<<setiosflags(ios::right)<<setw(2)<<hex<<(int)(*addr)<<dec;
       if ((j+1)%8) fout<<" "; else fout<<endl;
       ++addr;
     }
     fout<<endl;
   }
   fout.close();
 }

void FUResourceTable::halt ( )

Definition at line 684 of file FUResourceTable.cc.

Referenced by evf::FUResourceBroker::halting().

 {
   isHalting_=true;
   shutDownClients();
 }

bool FUResourceTable::handleCrashedEP	(	UInt_t	runNumber,
		pid_t	pid
	)

Definition at line 634 of file FUResourceTable.cc.

References i.

Referenced by evf::FUResourceBroker::emergencyStop(), and evf::FUResourceBroker::watching().

 {
   bool retval = false;
   vector<pid_t> pids=cellPrcIds();
   UInt_t iRawCell=pids.size();
   for (UInt_t i=0;i<pids.size();i++) { if (pid==pids[i]) { iRawCell=i; break; } }
   
   if (iRawCell<pids.size()){
     shmBuffer_->writeErrorEventData(runNumber,pid,iRawCell);
     retval = true;
   }
   else
     LOG4CPLUS_WARN(log_,"No raw data to send to error stream for process " << pid);
   shmBuffer_->removeClientPrcId(pid);
   return retval;
 }

void FUResourceTable::initialize	(	bool	segmentationMode,
		UInt_t	nbRawCells,
		UInt_t	nbRecoCells,
		UInt_t	nbDqmCells,
		UInt_t	rawCellSize,
		UInt_t	recoCellSize,
		UInt_t	dqmCellSize
	)
throw	(	evf::Exception
	)

Definition at line 111 of file FUResourceTable.cc.

References evf::FUShmBuffer::createShmBuffer(), i, and runTheMatrix::msg.

 {
   clear();
   
   shmBuffer_=FUShmBuffer::createShmBuffer(segmentationMode,
                       nbRawCells,nbRecoCells,nbDqmCells,
                       rawCellSize,recoCellSize,dqmCellSize);
   if (0==shmBuffer_) {
     string msg = "CREATION OF SHARED MEMORY SEGMENT FAILED!";
     LOG4CPLUS_FATAL(log_,msg);
     XCEPT_RAISE(evf::Exception,msg);
   }
   
   for (UInt_t i=0;i<nbRawCells_;i++) {
     resources_.push_back(new FUResource(i,log_,frb_,app_));
     freeResourceIds_.push(i);
   }
 
   acceptSMDataDiscard_ = new bool[nbRecoCells];
   acceptSMDqmDiscard_  = new int[nbDqmCells];
   
   resetCounters();
 }

void FUResourceTable::injectCRCError ( )

Definition at line 1050 of file FUResourceTable.cc.

References i.

 {
   for (UInt_t i=0;i<resources_.size();i++) {
     resources_[i]->scheduleCRCError();
   }
 }

bool evf::FUResourceTable::isActive ( ) const

inline

Definition at line 116 of file FUResourceTable.h.

References isActive_.

Referenced by evf::FUResourceBroker::halting().

116 { return isActive_; }

evf::FUResourceTable::isActive_

bool isActive_

Definition: FUResourceTable.h:258

bool FUResourceTable::isLastMessageOfEvent ( MemRef_t * bufRef )

Definition at line 1034 of file FUResourceTable.cc.

References createPayload::block.

 {
   while (0!=bufRef->getNextReference()) bufRef=bufRef->getNextReference();
   
   I2O_EVENT_DATA_BLOCK_MESSAGE_FRAME *block=
     (I2O_EVENT_DATA_BLOCK_MESSAGE_FRAME*)bufRef->getDataLocation();
   
   UInt_t iBlock    =block->blockNb;
   UInt_t nBlock    =block->nbBlocksInSuperFragment;
   UInt_t iSuperFrag=block->superFragmentNb;
   UInt_t nSuperFrag=block->nbSuperFragmentsInEvent;
 
   return ((iSuperFrag==nSuperFrag-1)&&(iBlock==nBlock-1));
 }

bool evf::FUResourceTable::isReadyToShutDown ( ) const

inline

Definition at line 119 of file FUResourceTable.h.

References isReadyToShutDown_.

Referenced by evf::FUResourceBroker::emergencyStop(), evf::FUResourceBroker::halting(), and evf::FUResourceBroker::stopping().

119 { return isReadyToShutDown_; }

evf::FUResourceTable::isReadyToShutDown_

bool isReadyToShutDown_

Definition: FUResourceTable.h:257

void FUResourceTable::lastResort ( )

Definition at line 1071 of file FUResourceTable.cc.

References gather_cfg::cout, and evf::FUShmRawCell::index().

Referenced by evf::FUResourceBroker::emergencyStop().

 {
   std::cout << "lastResort: " << shmBuffer_->nbRawCellsToRead() 
         << " more rawcells to read " << std::endl;
   while(shmBuffer_->nbRawCellsToRead()!=0){
     FUShmRawCell* newCell=shmBuffer_->rawCellToRead();
     std::cout << "lastResort: " << shmBuffer_->nbRawCellsToRead() << std::endl;
     shmBuffer_->scheduleRawCellForDiscardServerSide(newCell->index());
     std::cout << "lastResort: schedule raw cell for discard" << std::endl;
   }
 }

void evf::FUResourceTable::lock ( void )

inline

Definition at line 200 of file FUResourceTable.h.

References lock_.

200 { lock_.take(); }

evf::FUResourceTable::lock_

toolbox::BSem lock_

Definition: FUResourceTable.h:271

UInt_t evf::FUResourceTable::nbAllocated ( ) const

inline

Definition at line 124 of file FUResourceTable.h.

References nbAllocated_.

Referenced by evf::FUResourceBroker::actionPerformed().

124 { return nbAllocated_; }

evf::FUResourceTable::nbAllocated_

UInt_t nbAllocated_

Definition: FUResourceTable.h:243

UInt_t evf::FUResourceTable::nbAllocSent ( ) const

inline

Definition at line 139 of file FUResourceTable.h.

References nbAllocSent_.

Referenced by evf::FUResourceBroker::actionPerformed().

139 { return nbAllocSent_; }

evf::FUResourceTable::nbAllocSent_

UInt_t nbAllocSent_

Definition: FUResourceTable.h:264

UInt_t FUResourceTable::nbClients ( ) const

Definition at line 760 of file FUResourceTable.cc.

References query::result.

Referenced by evf::FUResourceBroker::actionPerformed().

 {
   UInt_t result(0);
   if (0!=shmBuffer_) result=shmBuffer_->nClients();
   return result;
 }

UInt_t evf::FUResourceTable::nbCompleted ( ) const

inline

Definition at line 126 of file FUResourceTable.h.

References nbCompleted_.

Referenced by evf::FUResourceBroker::actionPerformed().

126 { return nbCompleted_; }

evf::FUResourceTable::nbCompleted_

UInt_t nbCompleted_

Definition: FUResourceTable.h:245

UInt_t evf::FUResourceTable::nbCrcErrors ( ) const

inline

Definition at line 138 of file FUResourceTable.h.

References nbCrcErrors_.

Referenced by evf::FUResourceBroker::actionPerformed().

138 { return nbCrcErrors_; }

evf::FUResourceTable::nbCrcErrors_

UInt_t nbCrcErrors_

Definition: FUResourceTable.h:263

UInt_t evf::FUResourceTable::nbDiscarded ( ) const

inline

Definition at line 132 of file FUResourceTable.h.

References nbDiscarded_.

Referenced by evf::FUResourceBroker::actionPerformed().

132 { return nbDiscarded_; }

evf::FUResourceTable::nbDiscarded_

UInt_t nbDiscarded_

Definition: FUResourceTable.h:251

UInt_t evf::FUResourceTable::nbEolDiscarded ( ) const

inline

Definition at line 135 of file FUResourceTable.h.

References nbEolDiscarded_.

Referenced by evf::FUResourceBroker::actionPerformed().

135 { return nbEolDiscarded_; }

evf::FUResourceTable::nbEolDiscarded_

UInt_t nbEolDiscarded_

Definition: FUResourceTable.h:254

UInt_t evf::FUResourceTable::nbEolPosted ( ) const

inline

Definition at line 134 of file FUResourceTable.h.

References nbEolPosted_.

Referenced by evf::FUResourceBroker::actionPerformed().

134 { return nbEolPosted_; }

evf::FUResourceTable::nbEolPosted_

UInt_t nbEolPosted_

Definition: FUResourceTable.h:253

UInt_t evf::FUResourceTable::nbErrors ( ) const

inline

Definition at line 137 of file FUResourceTable.h.

References nbErrors_.

Referenced by evf::FUResourceBroker::actionPerformed().

137 { return nbErrors_; }

evf::FUResourceTable::nbErrors_

UInt_t nbErrors_

Definition: FUResourceTable.h:262

UInt_t evf::FUResourceTable::nbFreeSlots ( ) const

inline

Definition at line 123 of file FUResourceTable.h.

References freeResourceIds_.

Referenced by evf::FUResourceBroker::configureResources(), evf::FUResourceBroker::configuring(), and evf::FUResourceBroker::watching().

123 { return freeResourceIds_.size(); }

evf::FUResourceTable::freeResourceIds_

std::queue< UInt_t > freeResourceIds_

Definition: FUResourceTable.h:234

UInt_t evf::FUResourceTable::nbLost ( ) const

inline

Definition at line 133 of file FUResourceTable.h.

References nbLost_.

Referenced by evf::FUResourceBroker::actionPerformed().

133 { return nbLost_; }

evf::FUResourceTable::nbLost_

UInt_t nbLost_

Definition: FUResourceTable.h:252

UInt_t evf::FUResourceTable::nbPending ( ) const

inline

Definition at line 125 of file FUResourceTable.h.

References nbPending_.

Referenced by evf::FUResourceBroker::actionPerformed().

125 { return nbPending_; }

evf::FUResourceTable::nbPending_

UInt_t nbPending_

Definition: FUResourceTable.h:244

UInt_t evf::FUResourceTable::nbPendingSMDiscards ( ) const

inline

Definition at line 130 of file FUResourceTable.h.

References nbPendingSMDiscards_.

Referenced by evf::FUResourceBroker::actionPerformed().

130 { return nbPendingSMDiscards_; }

evf::FUResourceTable::nbPendingSMDiscards_

UInt_t nbPendingSMDiscards_

Definition: FUResourceTable.h:249

UInt_t evf::FUResourceTable::nbPendingSMDqmDiscards ( ) const

inline

Definition at line 131 of file FUResourceTable.h.

References nbPendingSMDqmDiscards_.

Referenced by evf::FUResourceBroker::actionPerformed().

131 { return nbPendingSMDqmDiscards_; }

evf::FUResourceTable::nbPendingSMDqmDiscards_

UInt_t nbPendingSMDqmDiscards_

Definition: FUResourceTable.h:250

UInt_t evf::FUResourceTable::nbResources ( ) const

inline

Definition at line 122 of file FUResourceTable.h.

References resources_.

Referenced by evf::FUResourceBroker::configureResources(), evf::FUResourceBroker::configuring(), and evf::FUResourceBroker::watching().

122 { return resources_.size(); }

evf::FUResourceTable::resources_

FUResourceVec_t resources_

Definition: FUResourceTable.h:230

UInt_t evf::FUResourceTable::nbSent ( ) const

inline

Definition at line 127 of file FUResourceTable.h.

References nbSent_.

Referenced by evf::FUResourceBroker::actionPerformed(), and evf::FUResourceBroker::monitoring().

127 { return nbSent_; }

evf::FUResourceTable::nbSent_

UInt_t nbSent_

Definition: FUResourceTable.h:246

UInt_t evf::FUResourceTable::nbSentDqm ( ) const

inline

Definition at line 129 of file FUResourceTable.h.

References nbSentDqm_.

Referenced by evf::FUResourceBroker::actionPerformed().

129 { return nbSentDqm_; }

evf::FUResourceTable::nbSentDqm_

UInt_t nbSentDqm_

Definition: FUResourceTable.h:248

UInt_t evf::FUResourceTable::nbSentError ( ) const

inline

Definition at line 128 of file FUResourceTable.h.

References nbSentError_.

Referenced by evf::FUResourceBroker::actionPerformed().

128 { return nbSentError_; }

evf::FUResourceTable::nbSentError_

UInt_t nbSentError_

Definition: FUResourceTable.h:247

void FUResourceTable::postEndOfLumiSection ( MemRef_t * bufRef )

Definition at line 608 of file FUResourceTable.cc.

References i, and runTheMatrix::msg.

Referenced by evf::FUResourceBroker::I2O_EVM_LUMISECTION_Callback().

 {
   I2O_EVM_END_OF_LUMISECTION_MESSAGE_FRAME *msg = 
     (I2O_EVM_END_OF_LUMISECTION_MESSAGE_FRAME *)bufRef->getDataLocation();
   //make sure to fill up the shmem so no process will miss it
   // but processes will have to handle duplicates
 
   for(unsigned int i = 0; i < nbRawCells_; i++) 
     {
       nbEolPosted_++;
       shmBuffer_->writeRawLumiSectionEvent(msg->lumiSection);
     }
 }

void FUResourceTable::printWorkLoopStatus ( )

Definition at line 1056 of file FUResourceTable.cc.

References gather_cfg::cout.

Referenced by evf::FUResourceBroker::emergencyStop().

 {
   std::cout << "Workloop status===============" << std::endl;
   std::cout << "==============================" << std::endl;
   if(wlSendData_!=0)
     std::cout << "SendData -> " << wlSendData_->isActive() << std::endl;
   if(wlSendDqm_!=0)
     std::cout << "SendDqm  -> " << wlSendDqm_->isActive()  << std::endl;
   if(wlDiscard_!=0)
     std::cout << "Discard  -> " << wlDiscard_->isActive()  << std::endl;
   std::cout << "Workloops Active  -> " << isActive_  << std::endl;
 
 }

void FUResourceTable::resetCounters ( )

Definition at line 730 of file FUResourceTable.cc.

References i.

Referenced by evf::FUResourceBroker::enabling().

 {
   if (0!=shmBuffer_) {
     for (UInt_t i=0;i<shmBuffer_->nRecoCells();i++) acceptSMDataDiscard_[i]= false;
     for (UInt_t i=0;i<shmBuffer_->nDqmCells();i++)  acceptSMDqmDiscard_[i] = 0;
   }
   
   nbAllocated_           =nbPending_;
   nbCompleted_           =0;
   nbSent_                =0;
   nbSentError_           =0;
   nbSentDqm_             =0;
   nbPendingSMDiscards_   =0;
   nbPendingSMDqmDiscards_=0;
   nbDiscarded_           =0;
   nbLost_                =0;
   nbEolPosted_           =0;
   nbEolDiscarded_        =0;
 
   nbErrors_              =0;
   nbCrcErrors_           =0;
   nbAllocSent_           =0;
 
   sumOfSquares_          =0;
   sumOfSizes_            =0;
   isStopping_            =false;
 }

void FUResourceTable::sendAllocate ( )

Definition at line 893 of file FUResourceTable.cc.

References i.

Referenced by evf::FUResourceBroker::enabling().

 {
   UInt_t nbFreeSlots    = this->nbFreeSlots();
   UInt_t nbFreeSlotsMax = resources_.size()/2;
   if (nbFreeSlots>nbFreeSlotsMax) {
     UIntVec_t fuResourceIds;
     for (UInt_t i=0;i<nbFreeSlots;i++)
       fuResourceIds.push_back(allocateResource());
     bu_->sendAllocate(fuResourceIds);
     nbAllocSent_++;
   }
 }

bool FUResourceTable::sendData ( toolbox::task::WorkLoop * workLoop )

Definition at line 161 of file FUResourceTable.cc.

References evf::FUShmRecoCell::eventSize(), evf::FUShmRecoCell::evtNumber(), edm::hlt::Exception, evf::FUShmRecoCell::fuGuid(), evf::FUShmRecoCell::fuProcessId(), evf::FUShmRecoCell::index(), CommonMethods::lock(), evf::FUShmRecoCell::outModId(), evf::FUShmRecoCell::payloadAddr(), evf::FUShmRecoCell::rawCellIndex(), evf::FUShmRecoCell::runNumber(), sistrip::runNumber_, and evf::FUShmRecoCell::type().

Referenced by startSendDataWorkLoop().

 {
   bool reschedule=true;
 
   FUShmRecoCell* cell=shmBuffer_->recoCellToRead();
   
   if (0==cell->eventSize()) {
     LOG4CPLUS_INFO(log_,"Don't reschedule sendData workloop.");
     UInt_t cellIndex=cell->index();
     shmBuffer_->finishReadingRecoCell(cell);
     shmBuffer_->discardRecoCell(cellIndex);
     reschedule=false;
   }
   else if (isHalting_) {
     LOG4CPLUS_INFO(log_,"sendData: isHalting, discard recoCell.");
     UInt_t cellIndex=cell->index();
     shmBuffer_->finishReadingRecoCell(cell);
     shmBuffer_->discardRecoCell(cellIndex);
   }
   else {
     try {
       if (cell->type()==0) {
     UInt_t   cellIndex       = cell->index();
     UInt_t   cellOutModId    = cell->outModId();
     UInt_t   cellFUProcId    = cell->fuProcessId();
     UInt_t   cellFUGuid      = cell->fuGuid();
     UChar_t* cellPayloadAddr = cell->payloadAddr();
     UInt_t   cellEventSize   = cell->eventSize();
     shmBuffer_->finishReadingRecoCell(cell);
 
     lock();
     nbPendingSMDiscards_++;
     unlock();
 
     sendInitMessage(cellIndex,cellOutModId,cellFUProcId,cellFUGuid,
             cellPayloadAddr,cellEventSize);
       }
       else if (cell->type()==1) {
     UInt_t   cellIndex       = cell->index();
     UInt_t   cellRawIndex    = cell->rawCellIndex();
     UInt_t   cellRunNumber   = cell->runNumber();
     UInt_t   cellEvtNumber   = cell->evtNumber();
     UInt_t   cellOutModId    = cell->outModId();
     UInt_t   cellFUProcId    = cell->fuProcessId();
     UInt_t   cellFUGuid      = cell->fuGuid();
     UChar_t *cellPayloadAddr = cell->payloadAddr();
     UInt_t   cellEventSize   = cell->eventSize();
     shmBuffer_->finishReadingRecoCell(cell);    
 
     lock();
     nbPendingSMDiscards_++;
     resources_[cellRawIndex]->incNbSent();
     if (resources_[cellRawIndex]->nbSent()==1) nbSent_++;
     unlock();
 
     sendDataEvent(cellIndex,cellRunNumber,cellEvtNumber,cellOutModId,
               cellFUProcId,cellFUGuid,cellPayloadAddr,cellEventSize);
       }
       else if (cell->type()==2) {
     UInt_t   cellIndex       = cell->index();
     UInt_t   cellRawIndex    = cell->rawCellIndex();
     //UInt_t   cellRunNumber   = cell->runNumber();
     UInt_t   cellEvtNumber   = cell->evtNumber();
     UInt_t   cellFUProcId    = cell->fuProcessId();
     UInt_t   cellFUGuid      = cell->fuGuid();
     UChar_t *cellPayloadAddr = cell->payloadAddr();
     UInt_t   cellEventSize   = cell->eventSize();
     shmBuffer_->finishReadingRecoCell(cell);    
 
     lock();
     nbPendingSMDiscards_++;
     resources_[cellRawIndex]->incNbSent();
     if (resources_[cellRawIndex]->nbSent()==1) { nbSent_++; nbSentError_++; }
     unlock();
     
     sendErrorEvent(cellIndex,runNumber_,cellEvtNumber,
                cellFUProcId,cellFUGuid,cellPayloadAddr,cellEventSize);
       }
       else {
     string errmsg="Unknown RecoCell type (neither INIT/DATA/ERROR).";
     XCEPT_RAISE(evf::Exception,errmsg);
       }
     }
     catch (xcept::Exception& e) {
       LOG4CPLUS_FATAL(log_,"Failed to send EVENT DATA to StorageManager: "
               <<xcept::stdformat_exception_history(e));
       reschedule=false;
     }
   }
   
   return reschedule;
 }

void FUResourceTable::sendDataEvent	(	UInt_t	fuResourceId,
		UInt_t	runNumber,
		UInt_t	evtNumber,
		UInt_t	outModId,
		UInt_t	fuProcessId,
		UInt_t	fuGuid,
		UChar_t *	data,
		UInt_t	dataSize
	)

Definition at line 937 of file FUResourceTable.cc.

 {
   if (0==sm_) {
     LOG4CPLUS_ERROR(log_,"No StorageManager, DROP DATA EVENT!");
   }
   else {
     acceptSMDataDiscard_[fuResourceId] = true;
     UInt_t nbBytes=sm_->sendDataEvent(fuResourceId,runNumber,evtNumber,
                       outModId,fuProcessId,fuGuid,
                       data,dataSize);
     sumOfSquares_+=(uint64_t)nbBytes*(uint64_t)nbBytes;
     sumOfSizes_  +=nbBytes;
   }
 }

void FUResourceTable::sendDiscard ( UInt_t buResourceId )

Definition at line 908 of file FUResourceTable.cc.

 {
   bu_->sendDiscard(buResourceId);
   nbDiscarded_++;
 }

bool FUResourceTable::sendDqm ( toolbox::task::WorkLoop * workLoop )

Definition at line 272 of file FUResourceTable.cc.

References gather_cfg::cout, evf::dqm::EMPTY, evf::FUShmDqmCell::eventSize(), evf::FUShmDqmCell::evtAtUpdate(), edm::hlt::Exception, evf::FUShmDqmCell::folderId(), evf::FUShmDqmCell::fuGuid(), evf::FUShmDqmCell::fuProcessId(), evf::FUShmDqmCell::index(), evf::FUShmDqmCell::payloadAddr(), evf::FUShmDqmCell::runNumber(), and evf::utils::state.

Referenced by startSendDqmWorkLoop().

 {
   bool reschedule=true;
   
   FUShmDqmCell* cell=shmBuffer_->dqmCellToRead();
   dqm::State_t  state=shmBuffer_->dqmState(cell->index());
   
   if (state==dqm::EMPTY) {
     LOG4CPLUS_WARN(log_,"Don't reschedule sendDqm workloop.");
     std::cout << "shut down dqm workloop " << std::endl;
     UInt_t cellIndex=cell->index();
     shmBuffer_->finishReadingDqmCell(cell);
     shmBuffer_->discardDqmCell(cellIndex);
     reschedule=false;
   }
   else if (isHalting_) {
     UInt_t cellIndex=cell->index();
     shmBuffer_->finishReadingDqmCell(cell);
     shmBuffer_->discardDqmCell(cellIndex);
   }
   else {
     try {
       UInt_t   cellIndex       = cell->index();
       UInt_t   cellRunNumber   = cell->runNumber();
       UInt_t   cellEvtAtUpdate = cell->evtAtUpdate();
       UInt_t   cellFolderId    = cell->folderId();
       UInt_t   cellFUProcId    = cell->fuProcessId();
       UInt_t   cellFUGuid      = cell->fuGuid();
       UChar_t *cellPayloadAddr = cell->payloadAddr();
       UInt_t   cellEventSize   = cell->eventSize();
       sendDqmEvent(cellIndex,cellRunNumber,cellEvtAtUpdate,cellFolderId,
            cellFUProcId,cellFUGuid,cellPayloadAddr,cellEventSize);
       shmBuffer_->finishReadingDqmCell(cell);
     }
     catch (xcept::Exception& e) {
       LOG4CPLUS_FATAL(log_,"Failed to send DQM DATA to StorageManager: "
               <<xcept::stdformat_exception_history(e));
       reschedule=false;
     }
   }
   
   return reschedule;
 }

void FUResourceTable::sendDqmEvent	(	UInt_t	fuDqmId,
		UInt_t	runNumber,
		UInt_t	evtAtUpdate,
		UInt_t	folderId,
		UInt_t	fuProcessId,
		UInt_t	fuGuid,
		UChar_t *	data,
		UInt_t	dataSize
	)

Definition at line 1006 of file FUResourceTable.cc.

 {
   if (0==sm_) {
     LOG4CPLUS_WARN(log_,"No StorageManager, DROP DQM EVENT.");
   }
   else {
     sm_->sendDqmEvent(fuDqmId,runNumber,evtAtUpdate,folderId,
               fuProcessId,fuGuid,data,dataSize);
 
     nbPendingSMDqmDiscards_++;
 
     acceptSMDqmDiscard_[fuDqmId]++;
     if(acceptSMDqmDiscard_[fuDqmId]>1)
       LOG4CPLUS_WARN(log_,"DQM Cell " << fuDqmId << " being sent more than once for folder " 
              << folderId << " process " << fuProcessId << " guid " << fuGuid);
     nbSentDqm_++;
   }
 }

void FUResourceTable::sendErrorEvent	(	UInt_t	fuResourceId,
		UInt_t	runNumber,
		UInt_t	evtNumber,
		UInt_t	fuProcessId,
		UInt_t	fuGuid,
		UChar_t *	data,
		UInt_t	dataSize
	)

Definition at line 961 of file FUResourceTable.cc.

 {
   if (0==sm_) {
     LOG4CPLUS_ERROR(log_,"No StorageManager, DROP ERROR EVENT!");
   }
   else {
     acceptSMDataDiscard_[fuResourceId] = true;
     UInt_t nbBytes=sm_->sendErrorEvent(fuResourceId,runNumber,evtNumber,
                        fuProcessId,fuGuid,data,dataSize);
     sumOfSquares_+=(uint64_t)nbBytes*(uint64_t)nbBytes;
     sumOfSizes_  +=nbBytes;
   }
 
 //   if (0!=shmBuffer_) {
 //     UInt_t n=nbDqmCells_;
 
 //     for (UInt_t i=0;i<n;i++) {
 //       if(shmBuffer_->dqmCell(i)->fuProcessId()==fuProcessId)
 //  {
 //    if(shmBuffer_->dqmState(i)!=dqm::SENT){
 //      shmBuffer_->setDqmState(i,dqm::SENT);   
 //      shmBuffer_->discardDqmCell(i);  
 //      acceptSMDqmDiscard_[i] = false;
 //    }  
 //  }
 //     }
 //     n=nbRecoCells_;
 //     for (UInt_t i=0;i<n;i++) {
 //       if(shmBuffer_->recoCell(i)->fuProcessId()==fuProcessId)
 //  {
 //    shmBuffer_->discardOrphanedRecoCell(i);  
 //  }
 //     }
 
 //   }
 }

void FUResourceTable::sendInitMessage	(	UInt_t	fuResourceId,
		UInt_t	outModId,
		UInt_t	fuProcessId,
		UInt_t	fuGuid,
		UChar_t *	data,
		UInt_t	dataSize
	)

Definition at line 916 of file FUResourceTable.cc.

 {
   if (0==sm_) {
     LOG4CPLUS_ERROR(log_,"No StorageManager, DROP INIT MESSAGE!");
   }
   else {
     acceptSMDataDiscard_[fuResourceId] = true;
     UInt_t nbBytes=sm_->sendInitMessage(fuResourceId,outModId,fuProcessId,
                     fuGuid,data,dataSize);
     sumOfSquares_+=(uint64_t)nbBytes*(uint64_t)nbBytes;
     sumOfSizes_  +=nbBytes;
   }
 }

void evf::FUResourceTable::setDoCrcCheck ( UInt_t doCrcCheck )

inline

Definition at line 110 of file FUResourceTable.h.

References doCrcCheck_.

Referenced by evf::FUResourceBroker::actionPerformed(), and evf::FUResourceBroker::configureResources().

110 { doCrcCheck_=doCrcCheck; }

evf::FUResourceTable::doCrcCheck_

UInt_t doCrcCheck_

Definition: FUResourceTable.h:239

void evf::FUResourceTable::setDoDumpEvents ( UInt_t doDumpEvents )

inline

Definition at line 113 of file FUResourceTable.h.

References doDumpEvents_.

Referenced by evf::FUResourceBroker::actionPerformed(), and evf::FUResourceBroker::configureResources().

113 { doDumpEvents_=doDumpEvents; }

evf::FUResourceTable::doDumpEvents_

UInt_t doDumpEvents_

Definition: FUResourceTable.h:240

void evf::FUResourceTable::setRunNumber ( UInt_t runNumber )

inline

Definition at line 53 of file FUResourceTable.h.

References inputsource_file_cfi::runNumber, and runNumber_.

Referenced by evf::FUResourceBroker::enabling().

53 { runNumber_ = runNumber; }

evf::FUResourceTable::runNumber_

UInt_t runNumber_

Definition: FUResourceTable.h:269

inputsource_file_cfi.runNumber

int runNumber

Definition: inputsource_file_cfi.py:32

void FUResourceTable::shutDownClients ( )

Definition at line 692 of file FUResourceTable.cc.

References evf::evt::EMPTY, i, n, and evf::utils::state.

 {
   nbClientsToShutDown_ = nbClients();
   isReadyToShutDown_   = false;
   
   if (nbClientsToShutDown_==0) {
     LOG4CPLUS_INFO(log_,"No clients to shut down. Checking if there are raw cells not assigned to any process yet");
     UInt_t n=nbResources();
     for (UInt_t i=0;i<n;i++) {
       evt::State_t state=shmBuffer_->evtState(i);
       if (state!=evt::EMPTY){
     LOG4CPLUS_WARN(log_,"Schedule discard at STOP for orphaned event in state " 
                << state);
     shmBuffer_->scheduleRawCellForDiscardServerSide(i);
       }
     }
     shmBuffer_->scheduleRawEmptyCellForDiscard();
   }
   else {
     UInt_t n=nbClientsToShutDown_;
     for (UInt_t i=0;i<n;++i) shmBuffer_->writeRawEmptyEvent();
   }
 }

void FUResourceTable::startDiscardWorkLoop	(		)
throw	(	evf::Exception
	)

Definition at line 318 of file FUResourceTable.cc.

References discard(), edm::hlt::Exception, and runTheMatrix::msg.

Referenced by evf::FUResourceBroker::enabling().

 {
   try {
     LOG4CPLUS_INFO(log_,"Start 'discard' workloop.");
     wlDiscard_=toolbox::task::getWorkLoopFactory()->getWorkLoop("Discard","waiting");
     if (!wlDiscard_->isActive()) wlDiscard_->activate();
     asDiscard_=toolbox::task::bind(this,&FUResourceTable::discard,"Discard");
     wlDiscard_->submit(asDiscard_);
     isActive_=true;
   }
   catch (xcept::Exception& e) {
     string msg = "Failed to start workloop 'Discard'.";
     XCEPT_RETHROW(evf::Exception,msg,e);
   }
   isReadyToShutDown_=false;
 }

void FUResourceTable::startSendDataWorkLoop	(		)
throw	(	evf::Exception
	)

Definition at line 144 of file FUResourceTable.cc.

References edm::hlt::Exception, runTheMatrix::msg, and sendData().

Referenced by evf::FUResourceBroker::enabling().

 {
   try {
     wlSendData_=
       toolbox::task::getWorkLoopFactory()->getWorkLoop("SendData","waiting");
     if (!wlSendData_->isActive()) wlSendData_->activate();
     asSendData_=toolbox::task::bind(this,&FUResourceTable::sendData,"SendData");
     wlSendData_->submit(asSendData_);
   }
   catch (xcept::Exception& e) {
     string msg = "Failed to start workloop 'SendData'.";
     XCEPT_RETHROW(evf::Exception,msg,e);
   }
 }

void FUResourceTable::startSendDqmWorkLoop	(		)
throw	(	evf::Exception
	)

Definition at line 256 of file FUResourceTable.cc.

References edm::hlt::Exception, runTheMatrix::msg, and sendDqm().

Referenced by evf::FUResourceBroker::enabling().

 {
   try {
     wlSendDqm_=toolbox::task::getWorkLoopFactory()->getWorkLoop("SendDqm","waiting");
     if (!wlSendDqm_->isActive()) wlSendDqm_->activate();
     asSendDqm_=toolbox::task::bind(this,&FUResourceTable::sendDqm,"SendDqm");
     wlSendDqm_->submit(asSendDqm_);
   }
   catch (xcept::Exception& e) {
     string msg = "Failed to start workloop 'SendDqm'.";
     XCEPT_RETHROW(evf::Exception,msg,e);
   }
 }

void FUResourceTable::stop ( )

Definition at line 676 of file FUResourceTable.cc.

Referenced by evf::FUResourceBroker::stopping().

 {
   isStopping_ = true;
   shutDownClients();
 }

UInt_t evf::FUResourceTable::sumOfSizes ( ) const

inline

Definition at line 142 of file FUResourceTable.h.

References sumOfSizes_.

Referenced by evf::FUResourceBroker::monitoring().

142 { return sumOfSizes_; }

evf::FUResourceTable::sumOfSizes_

UInt_t sumOfSizes_

Definition: FUResourceTable.h:267

uint64_t evf::FUResourceTable::sumOfSquares ( ) const

inline

Definition at line 141 of file FUResourceTable.h.

References sumOfSquares_.

Referenced by evf::FUResourceBroker::monitoring().

141 { return sumOfSquares_; }

evf::FUResourceTable::sumOfSquares_

uint64_t sumOfSquares_

Definition: FUResourceTable.h:266

void evf::FUResourceTable::unlock ( void )

inline

Definition at line 201 of file FUResourceTable.h.

References lock_.

201 { lock_.give(); }

evf::FUResourceTable::lock_

toolbox::BSem lock_

Definition: FUResourceTable.h:271

Member Data Documentation

bool* evf::FUResourceTable::acceptSMDataDiscard_

private

Definition at line 236 of file FUResourceTable.h.

int* evf::FUResourceTable::acceptSMDqmDiscard_

private

Definition at line 237 of file FUResourceTable.h.

xdaq::Application* evf::FUResourceTable::app_

private

Definition at line 273 of file FUResourceTable.h.

ActionSignature_t* evf::FUResourceTable::asDiscard_

private

Definition at line 227 of file FUResourceTable.h.

ActionSignature_t* evf::FUResourceTable::asSendData_

private

Definition at line 221 of file FUResourceTable.h.

ActionSignature_t* evf::FUResourceTable::asSendDqm_

private

Definition at line 224 of file FUResourceTable.h.

BUProxy* evf::FUResourceTable::bu_

private

Definition at line 215 of file FUResourceTable.h.

UInt_t evf::FUResourceTable::doCrcCheck_

private

Definition at line 239 of file FUResourceTable.h.

Referenced by setDoCrcCheck().

UInt_t evf::FUResourceTable::doDumpEvents_

private

Definition at line 240 of file FUResourceTable.h.

Referenced by setDoDumpEvents().

EvffedFillerRB* evf::FUResourceTable::frb_

private

Definition at line 272 of file FUResourceTable.h.

std::queue<UInt_t> evf::FUResourceTable::freeResourceIds_

private

Definition at line 234 of file FUResourceTable.h.

Referenced by nbFreeSlots().

bool evf::FUResourceTable::isActive_

private

Definition at line 258 of file FUResourceTable.h.

Referenced by isActive().

bool evf::FUResourceTable::isHalting_

private

Definition at line 259 of file FUResourceTable.h.

bool evf::FUResourceTable::isReadyToShutDown_

private

Definition at line 257 of file FUResourceTable.h.

Referenced by isReadyToShutDown().

bool evf::FUResourceTable::isStopping_

private

Definition at line 260 of file FUResourceTable.h.

toolbox::BSem evf::FUResourceTable::lock_

private

Definition at line 271 of file FUResourceTable.h.

Referenced by lock(), and unlock().

log4cplus::Logger evf::FUResourceTable::log_

private

Definition at line 218 of file FUResourceTable.h.

UInt_t evf::FUResourceTable::nbAllocated_

private

Definition at line 243 of file FUResourceTable.h.

Referenced by nbAllocated().

UInt_t evf::FUResourceTable::nbAllocSent_

private

Definition at line 264 of file FUResourceTable.h.

Referenced by nbAllocSent().

UInt_t evf::FUResourceTable::nbClientsToShutDown_

private

Definition at line 256 of file FUResourceTable.h.

UInt_t evf::FUResourceTable::nbCompleted_

private

Definition at line 245 of file FUResourceTable.h.

Referenced by nbCompleted().

UInt_t evf::FUResourceTable::nbCrcErrors_

private

Definition at line 263 of file FUResourceTable.h.

Referenced by nbCrcErrors().

UInt_t evf::FUResourceTable::nbDiscarded_

private

Definition at line 251 of file FUResourceTable.h.

Referenced by nbDiscarded().

UInt_t evf::FUResourceTable::nbDqmCells_

private

Definition at line 231 of file FUResourceTable.h.

UInt_t evf::FUResourceTable::nbEolDiscarded_

private

Definition at line 254 of file FUResourceTable.h.

Referenced by nbEolDiscarded().

UInt_t evf::FUResourceTable::nbEolPosted_

private

Definition at line 253 of file FUResourceTable.h.

Referenced by nbEolPosted().

UInt_t evf::FUResourceTable::nbErrors_

private

Definition at line 262 of file FUResourceTable.h.

Referenced by nbErrors().

UInt_t evf::FUResourceTable::nbLost_

private

Definition at line 252 of file FUResourceTable.h.

Referenced by nbLost().

UInt_t evf::FUResourceTable::nbPending_

private

Definition at line 244 of file FUResourceTable.h.

Referenced by nbPending().

UInt_t evf::FUResourceTable::nbPendingSMDiscards_

private

Definition at line 249 of file FUResourceTable.h.

Referenced by nbPendingSMDiscards().

UInt_t evf::FUResourceTable::nbPendingSMDqmDiscards_

private

Definition at line 250 of file FUResourceTable.h.

Referenced by nbPendingSMDqmDiscards().

UInt_t evf::FUResourceTable::nbRawCells_

private

Definition at line 232 of file FUResourceTable.h.

UInt_t evf::FUResourceTable::nbRecoCells_

private

Definition at line 233 of file FUResourceTable.h.

UInt_t evf::FUResourceTable::nbSent_

private

Definition at line 246 of file FUResourceTable.h.

Referenced by nbSent().

UInt_t evf::FUResourceTable::nbSentDqm_

private

Definition at line 248 of file FUResourceTable.h.

Referenced by nbSentDqm().

UInt_t evf::FUResourceTable::nbSentError_

private

Definition at line 247 of file FUResourceTable.h.

Referenced by nbSentError().

FUResourceVec_t evf::FUResourceTable::resources_

private

Definition at line 230 of file FUResourceTable.h.

Referenced by nbResources().

UInt_t evf::FUResourceTable::runNumber_

private

Definition at line 269 of file FUResourceTable.h.

Referenced by setRunNumber().

FUShmBuffer* evf::FUResourceTable::shmBuffer_

private

Definition at line 229 of file FUResourceTable.h.

unsigned int evf::FUResourceTable::shutdownTimeout_

private

Definition at line 241 of file FUResourceTable.h.

SMProxy* evf::FUResourceTable::sm_

private

Definition at line 216 of file FUResourceTable.h.

UInt_t evf::FUResourceTable::sumOfSizes_

private

Definition at line 267 of file FUResourceTable.h.

Referenced by sumOfSizes().

uint64_t evf::FUResourceTable::sumOfSquares_

private

Definition at line 266 of file FUResourceTable.h.

Referenced by sumOfSquares().

WorkLoop_t* evf::FUResourceTable::wlDiscard_

private

Definition at line 226 of file FUResourceTable.h.

WorkLoop_t* evf::FUResourceTable::wlSendData_

private

Definition at line 220 of file FUResourceTable.h.

WorkLoop_t* evf::FUResourceTable::wlSendDqm_

private

Definition at line 223 of file FUResourceTable.h.

Public Member Functions

Private Types

Private Attributes

Detailed Description

Member Typedef Documentation

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation