evf::FUResourceTable Class Reference

#include <FUResourceTable.h>

Inheritance diagram for evf::FUResourceTable:

Public Member Functions
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 ()
bool	discardDataEvent (MemRef_t *bufRef)
bool	discardDataEventWhileHalting (MemRef_t *bufRef)
bool	discardDqmEvent (MemRef_t *bufRef)
bool	discardDqmEventWhileHalting (MemRef_t *bufRef)
bool	discardWhileHalting (bool sendDiscards)
std::vector< std::string >	dqmCellStates () const
void	dropEvent ()
	FUResourceTable (bool segmentationMode, UInt_t nbRawCells, UInt_t nbRecoCells, UInt_t nbDqmCells, UInt_t rawCellSize, UInt_t recoCellSize, UInt_t dqmCellSize, int freeResReq, BUProxy *bu, SMProxy *sm, log4cplus::Logger logger, unsigned int, EvffedFillerRB *frb, xdaq::Application *) throw (evf::Exception)
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	lastResort ()
UInt_t	nbClients () const
UInt_t	nbResources () const
unsigned int	nbResources ()
void	postEndOfLumiSection (MemRef_t *bufRef)
std::string	printStatus ()
void	resetCounters ()
void	resetIPC ()
	reset the ShmBuffer to the initial state More...
bool	sendData ()
bool	sendDataWhileHalting ()
bool	sendDqm ()
bool	sendDqmWhileHalting ()
void	shutDownClients ()
virtual	~FUResourceTable ()
Public Member Functions inherited from evf::IPCMethod
UInt_t	allocateResource ()
void	dumpEvent (evf::FUShmRawCell *cell)
void	injectCRCError ()
	IPCMethod (bool segmentationMode, UInt_t nbRawCells, UInt_t nbRecoCells, UInt_t nbDqmCells, UInt_t rawCellSize, UInt_t recoCellSize, UInt_t dqmCellSize, int freeResReq, BUProxy *bu, SMProxy *sm, log4cplus::Logger logger, unsigned int timeout, EvffedFillerRB *frb, xdaq::Application *app) throw (evf::Exception)
bool	isActive () const
bool	isLastMessageOfEvent (MemRef_t *bufRef)
bool	isReadyToShutDown () const
void	lock ()
UInt_t	nbAllocated () const
UInt_t	nbAllocSent () 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
int	nbPendingSMDqmDiscards () const
UInt_t	nbSent () const
UInt_t	nbSentDqm () const
UInt_t	nbSentError () const
void	releaseResources ()
	releases all FUResource's More...
void	resetPendingAllocates ()
	resets free resources to the maximum number More...
void	sendAllocate ()
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)
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, UInt_t nExpectedEPs)
void	setActive (bool activeValue)
void	setDoCrcCheck (UInt_t doCrcCheck)
void	setDoDumpEvents (UInt_t doDumpEvents)
void	setReadyToShutDown (bool readyValue)
void	setRunNumber (UInt_t runNumber)
void	setStopFlag (bool status)
UInt_t	shutdownStatus ()
UInt_t	sumOfSizes () const
uint64_t	sumOfSquares () const
void	unlock ()
virtual	~IPCMethod ()

Private Member Functions
void	discardNoReschedule ()
void	rethrowShmBufferException (evf::Exception &e, std::string where) const throw (evf::Exception)
void	shutdownWatchdog (unsigned int timeout)

Private Attributes
bool	sDataActive_
bool	sDqmActive_
FUShmBuffer *	shmBuffer_
std::atomic_bool	watchDogEnd_
std::atomic_bool	watchDogSetFailed_

Additional Inherited Members
Protected Attributes inherited from evf::IPCMethod
bool *	acceptSMDataDiscard_
int *	acceptSMDqmDiscard_
xdaq::Application *	app_
BUProxy *	bu_
UInt_t	doCrcCheck_
UInt_t	doDumpEvents_
EvffedFillerRB *	frb_
std::queue< UInt_t >	freeResourceIds_
unsigned int	freeResRequiredForAllocate_
bool	isActive_
bool	isReadyToShutDown_
sem_t	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_
std::atomic< int >	nbPendingSMDqmDiscards_
UInt_t	nbRawCells_
UInt_t	nbRecoCells_
UInt_t	nbSent_
UInt_t	nbSentDqm_
UInt_t	nbSentError_
FUResourceVec_t	resources_
UInt_t	runNumber_
UInt_t	shutdownStatus_
unsigned int	shutdownTimeout_
SMProxy *	sm_
bool	stopFlag_
UInt_t	sumOfSizes_
uint64_t	sumOfSquares_

Detailed Description

Table of resources linked with the Shared Memory Buffer

Author:

smorovic

Definition at line 40 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,
		int	freeResReq,
		BUProxy *	bu,
		SMProxy *	sm,
		log4cplus::Logger	logger,
		unsigned int	timeout,
		EvffedFillerRB *	frb,
		xdaq::Application *	app
	)
throw	(	evf::Exception
	)

Definition at line 31 of file FUResourceTable.cc.

                                                                       :

    // call super constructor
            IPCMethod(segmentationMode, nbRawCells, nbRecoCells, nbDqmCells,
                    rawCellSize, recoCellSize, dqmCellSize, freeResReq, bu, sm,
                    logger, timeout, frb, app), shmBuffer_(0)

{
    initialize(segmentationMode, nbRawCells, nbRecoCells, nbDqmCells,
            rawCellSize, recoCellSize, dqmCellSize);
}

FUResourceTable::~FUResourceTable ( )

virtual

Definition at line 48 of file FUResourceTable.cc.

References hitfit::clear(), and evf::FUShmBuffer::releaseSharedMemory().

                                  {
    clear();
    //workloop cancels used to be here in the previous version
    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

bool FUResourceTable::buildResource ( MemRef_t *  bufRef )

virtual

Process buffer received via I2O_FU_TAKE message

Implements evf::IPCMethod.

Definition at line 581 of file FUResourceTable.cc.

References evf::FUResource::allocate(), createPayload::block, gather_cfg::cout, evf::FUResource::doCrcCheck(), dumpEvent(), alignCSCRings::e, 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().

                                                    {
    bool eventComplete = false;
    // UPDATED
    bool lastMsg = isLastMessageOfEvent(bufRef);
    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;
    // Check input
    if ((int) block->fuTransactionId < 0 || fuResourceId >= nbRawCells_
            || (int) block->buResourceId < 0) {
        stringstream failureStr;
        failureStr << "Received TAKE message with invalid bu/fu resource id:"
                << " fuResourceId: " << fuResourceId << " buResourceId: "
                << buResourceId;
        LOG4CPLUS_ERROR(log_, failureStr.str());
        XCEPT_RAISE(evf::Exception, failureStr.str());
    }
    FUResource* resource = resources_[fuResourceId];

    // allocate resource
    if (!resource->fatalError() && !resource->isAllocated()) {
        FUShmRawCell* cell = 0;
        try {
            cell = shmBuffer_->rawCellToWrite();
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:buildResource:rawCellToWrite");
        }
        if (cell == 0) {
            bufRef->release();
            return eventComplete;
        }
        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);
    }

#ifdef DEBUG_RES_TAB
    std::cout << "Received frame for resource " << buResourceId << std::endl;
#endif
    // keep building this resource if it is healthy
    if (!resource->fatalError()) {
#ifdef DEBUG_RES_TAB
        std::cout << "No fatal error for  " << buResourceId << ", keep building..."<< std::endl;
#endif
        resource->process(bufRef);
        lock();
        nbErrors_ += resource->nbErrors();
        nbCrcErrors_ += resource->nbCrcErrors();
        unlock();
#ifdef DEBUG_RES_TAB
        std::cout << "Checking if resource is complete " << buResourceId << std::endl;
#endif
        // make resource available for pick-up
        if (resource->isComplete()) {
#ifdef DEBUG_RES_TAB
            std::cout << "@@@@RESOURCE is COMPLETE " << buResourceId << std::endl;
#endif
            lock();
            nbCompleted_++;
            nbPending_--;
            unlock();
            if (doDumpEvents_ > 0 && nbCompleted_ % doDumpEvents_ == 0)
                dumpEvent(resource->shmCell());
            try {
                shmBuffer_->finishWritingRawCell(resource->shmCell());
            } catch (evf::Exception& e) {
                rethrowShmBufferException(e,
                        "FUResourceTable:buildResource:finishWritingRawCell");
            }
            eventComplete = true;
        }

    }
    // bad event, release msg, and the whole resource if this was the last one
    if (resource->fatalError()) {
        if (lastMsg) {
            try {
                shmBuffer_->releaseRawCell(resource->shmCell());
            } catch (evf::Exception& e) {
                rethrowShmBufferException(e,
                        "FUResourceTable:buildResource:releaseRawCell");
            }
            resource->release(true);
            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

virtual

Implements evf::IPCMethod.

Definition at line 1325 of file FUResourceTable.cc.

References alignCSCRings::e, i, n, and query::result.

                                                     {
    vector < UInt_t > result;
    if (0 != shmBuffer_) {
        UInt_t n = nbResources();
        shmBuffer_->lock();
        try {
            for (UInt_t i = 0; i < n; i++)
                result.push_back(shmBuffer_->evtNumber(i));
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:cellEvtNumbers:evtNumber");
        }
        shmBuffer_->unlock();
    }
    return result;
}

vector< pid_t > FUResourceTable::cellPrcIds ( ) const

virtual

Implements evf::IPCMethod.

Definition at line 1343 of file FUResourceTable.cc.

References alignCSCRings::e, i, n, and query::result.

                                                {
    vector < pid_t > result;
    if (0 != shmBuffer_) {
        UInt_t n = nbResources();
        shmBuffer_->lock();
        try {
            for (UInt_t i = 0; i < n; i++)
                result.push_back(shmBuffer_->evtPrcId(i));
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e, "FUResourceTable:cellPrcIds:evtPrcId");
        }
        shmBuffer_->unlock();
    }
    return result;
}

vector< string > FUResourceTable::cellStates ( ) const

virtual

Implements evf::IPCMethod.

Definition at line 1246 of file FUResourceTable.cc.

References evf::evt::DISCARDING, alignCSCRings::e, 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();
        try {
            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");
                // UPDATED
                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");
            }
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e, "FUResourceTable:cellStates:evtState");
        }
        shmBuffer_->unlock();
    }
    return result;
}

vector< time_t > FUResourceTable::cellTimeStamps ( ) const

virtual

Implements evf::IPCMethod.

Definition at line 1360 of file FUResourceTable.cc.

References alignCSCRings::e, i, n, and query::result.

                                                     {
    vector < time_t > result;
    try {
        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();
        }
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:cellTimeStamps:evtTimeStamp");
    }
    return result;
}

void FUResourceTable::clear ( void  )

virtual

Clear contents of resource table. (empty all containers)

Implements evf::IPCMethod.

Definition at line 1142 of file FUResourceTable.cc.

References i.

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

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

vector< pid_t > FUResourceTable::clientPrcIds ( ) const

virtual

Implements evf::IPCMethod.

Definition at line 1211 of file FUResourceTable.cc.

References alignCSCRings::e, i, n, and query::result.

                                                  {
    vector < pid_t > result;
    try {
        if (0 != shmBuffer_) {
            UInt_t n = nbClients();
            for (UInt_t i = 0; i < n; i++)
                result.push_back(shmBuffer_->clientPrcId(i));
        }
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:clientPrcIds:clientPrcIds");
    }
    return result;
}

string FUResourceTable::clientPrcIdsAsString ( ) const

virtual

Implements evf::IPCMethod.

Definition at line 1227 of file FUResourceTable.cc.

References alignCSCRings::e, i, and n.

                                                   {
    stringstream ss;
    try {
        if (0 != shmBuffer_) {
            UInt_t n = nbClients();
            for (UInt_t i = 0; i < n; i++) {
                if (i > 0)
                    ss << ",";
                ss << shmBuffer_->clientPrcId(i);
            }
        }
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:clientPrcIdsAsString:clientPrcId");
    }
    return ss.str();
}

bool FUResourceTable::discard ( )

virtual

Corresponds to the Discard workloop, to be called in normal operation.

Implements evf::IPCMethod.

Definition at line 438 of file FUResourceTable.cc.

References evf::FUShmRawCell::buResourceId(), alignCSCRings::e, evf::evt::EMPTY, evf::FUShmRawCell::fuResourceId(), evf::FUShmRawCell::index(), CommonMethods::lock(), evf::utils::state, evf::evt::STOP, and evf::evt::USEDLS.

                              {
    FUShmRawCell* cell = 0;
    // initialize to a value to avoid warnings
    evt::State_t state = evt::EMPTY;
    try {
        cell = shmBuffer_->rawCellToDiscard();
        state = shmBuffer_->evtState(cell->index());
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:discard:rawCellToRead/evtState");
    }

    bool reschedule = true;
    bool shutDown = (state == evt::STOP);
    bool isLumi = (state == evt::USEDLS);
    UInt_t fuResourceId = cell->fuResourceId();
    UInt_t buResourceId = cell->buResourceId();

    if (state == evt::EMPTY) {
        LOG4CPLUS_ERROR(log_, "WARNING! ATTEMPTING TO DISCARD EMPTY CELL!!!");
        return true;
    }

    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;
        }
    }

    try {
        shmBuffer_->discardRawCell(cell);
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e, "FUResourceTable:discard:discardRawCell");
    }
    // UPDATED
    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(true);
            lock();
            freeResourceIds_.push(fuResourceId);
            assert(freeResourceIds_.size() <= resources_.size());
            unlock();

            sendDiscard(buResourceId);
            sendAllocate();
        }
    }

    if (!reschedule) {
        discardNoReschedule();
    }

    return reschedule;
}

bool FUResourceTable::discardDataEvent ( MemRef_t *  bufRef )

virtual

Process buffer received via I2O_SM_DATA_DISCARD message in normal operation.

Implements evf::IPCMethod.

Definition at line 693 of file FUResourceTable.cc.

References alignCSCRings::e, CommonMethods::lock(), lumiQueryAPI::msg, and _I2O_FU_DATA_DISCARD_MESSAGE_FRAME::rbBufferID.

                                                       {
    I2O_FU_DATA_DISCARD_MESSAGE_FRAME *msg;
    msg = (I2O_FU_DATA_DISCARD_MESSAGE_FRAME*) bufRef->getDataLocation();
    UInt_t recoIndex = msg->rbBufferID;

    // Check input
    if ((int) msg->rbBufferID < 0 || recoIndex >= nbRecoCells_)
        LOG4CPLUS_ERROR(
                log_,
                "Received DISCARD DATA message with invalid recoIndex:"
                        << recoIndex);

    if (acceptSMDataDiscard_[recoIndex]) {
        lock();
        nbPendingSMDiscards_--;
        unlock();
        acceptSMDataDiscard_[recoIndex] = false;

        try {
            shmBuffer_->discardRecoCell(recoIndex);
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:discardDataEvent:discardRecoCell");
        }

    } else {
        LOG4CPLUS_ERROR(log_, "Spurious DATA discard by StorageManager, skip!");
    }

    bufRef->release();
    return true;
}

bool FUResourceTable::discardDataEventWhileHalting ( MemRef_t *  bufRef )

virtual

Process buffer received via I2O_SM_DATA_DISCARD message while halting.

Implements evf::IPCMethod.

Definition at line 727 of file FUResourceTable.cc.

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

                                                                   {
    I2O_FU_DATA_DISCARD_MESSAGE_FRAME *msg;
    msg = (I2O_FU_DATA_DISCARD_MESSAGE_FRAME*) bufRef->getDataLocation();
    UInt_t recoIndex = msg->rbBufferID;

    // Check input
    if ((int) msg->rbBufferID < 0 || recoIndex >= nbRecoCells_)
        LOG4CPLUS_ERROR(
                log_,
                "Received DISCARD DATA message with invalid recoIndex:"
                        << recoIndex);

    if (acceptSMDataDiscard_[recoIndex]) {
        lock();
        nbPendingSMDiscards_--;
        unlock();
        acceptSMDataDiscard_[recoIndex] = false;

    } else {
        LOG4CPLUS_ERROR(log_, "Spurious DATA discard by StorageManager, skip!");
    }

    bufRef->release();
    return false;
}

bool FUResourceTable::discardDqmEvent ( MemRef_t *  bufRef )

virtual

Process buffer received via I2O_SM_DQM_DISCARD message in normal operation.

Implements evf::IPCMethod.

Definition at line 754 of file FUResourceTable.cc.

References alignCSCRings::e, lumiQueryAPI::msg, _I2O_FU_DQM_DISCARD_MESSAGE_FRAME::rbBufferID, and evf::dqm::SENT.

                                                      {
    I2O_FU_DQM_DISCARD_MESSAGE_FRAME *msg;
    msg = (I2O_FU_DQM_DISCARD_MESSAGE_FRAME*) bufRef->getDataLocation();
    UInt_t dqmIndex = msg->rbBufferID;

    // Check input
    if ((int) msg->rbBufferID < 0 || dqmIndex >= nbDqmCells_)
        LOG4CPLUS_ERROR(
                log_,
                "Received DISCARD DQM message with invalid dqmIndex:"
                        << dqmIndex);

    unsigned int ntries = 0;
    try {
        while (shmBuffer_->dqmState(dqmIndex) != dqm::SENT) {
            if (ntries)//tolerate one attempt
            LOG4CPLUS_WARN(
                    log_,
                    "DQM discard for cell " << dqmIndex
                            << " which is not yet 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;
            }
        }
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e, "FUResourceTable:discardDqmEvent:dqmState");
    }
    if (acceptSMDqmDiscard_[dqmIndex] > 0) {
        acceptSMDqmDiscard_[dqmIndex]--;
        if (--nbPendingSMDqmDiscards_ < 0) {
            LOG4CPLUS_WARN(
                    log_,
                    "Spurious??? DQM discard by StorageManager, index "
                            << dqmIndex << " cell state "
                            << shmBuffer_->dqmState(dqmIndex)
                            << " accept flag " << acceptSMDqmDiscard_[dqmIndex]);
        }
        try {
            shmBuffer_->discardDqmCell(dqmIndex);
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:discardDqmEvent:discardDqmCell");
        }

    } else {
        LOG4CPLUS_ERROR(
                log_,
                "Spurious DQM discard for cell " << dqmIndex
                        << " from StorageManager while cell is not accepting discards");
    }

    bufRef->release();
    return true;
}

bool FUResourceTable::discardDqmEventWhileHalting ( MemRef_t *  bufRef )

virtual

Process buffer received via I2O_SM_DQM_DISCARD message while halting.

Implements evf::IPCMethod.

Definition at line 817 of file FUResourceTable.cc.

References alignCSCRings::e, lumiQueryAPI::msg, _I2O_FU_DQM_DISCARD_MESSAGE_FRAME::rbBufferID, and evf::dqm::SENT.

                                                                  {
    I2O_FU_DQM_DISCARD_MESSAGE_FRAME *msg;
    msg = (I2O_FU_DQM_DISCARD_MESSAGE_FRAME*) bufRef->getDataLocation();
    UInt_t dqmIndex = msg->rbBufferID;

    // Check input
    if ((int) msg->rbBufferID < 0 || dqmIndex >= nbDqmCells_)
        LOG4CPLUS_ERROR(
                log_,
                "Received DISCARD DQM message with invalid dqmIndex:"
                        << dqmIndex);

    unsigned int ntries = 0;
    try {
        while (shmBuffer_->dqmState(dqmIndex) != dqm::SENT) {
            if (ntries)//tolerate one attempt
            LOG4CPLUS_WARN(
                    log_,
                    "DQM discard for cell " << dqmIndex
                            << " which is not yet 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;
            }
        }
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:discardDqmEventWhileHalting:dqmState(2)");
    }
    if (acceptSMDqmDiscard_[dqmIndex] > 0) {
        acceptSMDqmDiscard_[dqmIndex]--;
        if (--nbPendingSMDqmDiscards_ < 0) {
            try {
                LOG4CPLUS_WARN(
                        log_,
                        "Spurious??? DQM discard by StorageManager, index "
                                << dqmIndex << " cell state "
                                << shmBuffer_->dqmState(dqmIndex)
                                << " accept flag "
                                << acceptSMDqmDiscard_[dqmIndex]);
            } catch (evf::Exception& e) {
                rethrowShmBufferException(e,
                        "FUResourceTable:discardDqmEventWhileHalting:dqmState");
            }
        }

    } else {
        LOG4CPLUS_ERROR(
                log_,
                "Spurious DQM discard for cell " << dqmIndex
                        << " from StorageManager while cell is not accepting discards");
    }

    bufRef->release();
    return false;
}

void FUResourceTable::discardNoReschedule ( )

private

Called when entering the shutdown cycle. The function sets readyToShutDown to true, allowing the Resource Table to be safely shut down.

Definition at line 354 of file FUResourceTable.cc.

References prof2calltree::count, gather_cfg::cout, alignCSCRings::e, evf::dqm::EMPTY, i, n, evf::FUShmBuffer::shm_nattch(), and evf::utils::state.

                                          {
    std::cout << " entered shutdown cycle " << std::endl;
    shutdownStatus_|=1<<11;
    try {
        shmBuffer_->writeRecoEmptyEvent();
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:discardNoReschedule: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) {
                    shutdownStatus_|=1<<12;
            //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++) {
            // initialize to a value to avoid warnings
            dqm::State_t state = dqm::EMPTY;
            try {
                state = shmBuffer_->dqmState(i);
            } catch (evf::Exception& e) {
                rethrowShmBufferException(e,
                        "FUResourceTable:discardNoReschedule:dqmState");
            }
            if (state != dqm::EMPTY)
                allEmpty = false;
        }
        shmBuffer_->unlock();
    }
    shutdownStatus_|=1<<13;

    std::cout << "Number of  pending discards before declaring ready to shut down: " << nbPendingSMDqmDiscards_ << 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 ");

    }

    try {
        shmBuffer_->writeDqmEmptyEvent();
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:discardNoReschedule:writeDqmEmptyEvent");
    }

    isReadyToShutDown_ = true; // moved here from within the first while loop to make sure the
    // sendDqm loop has been shut down as well
}

bool FUResourceTable::discardWhileHalting ( bool  sendDiscards )

virtual

Function called when FSM is in state stopping / halting, in Discard workloop.

Implements evf::IPCMethod.

Definition at line 507 of file FUResourceTable.cc.

References evf::FUShmRawCell::buResourceId(), alignCSCRings::e, evf::evt::EMPTY, evf::FUShmRawCell::fuResourceId(), evf::FUShmRawCell::index(), CommonMethods::lock(), evf::utils::state, evf::evt::STOP, and evf::evt::USEDLS.

                                                           {
    FUShmRawCell* cell = 0;
    // initialize to a value to avoid warnings
    evt::State_t state = evt::EMPTY;
    try {
        cell = shmBuffer_->rawCellToDiscard();
        state = shmBuffer_->evtState(cell->index());
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:discardWhileHalting:rawCellToRead/evtState");
    }

    bool reschedule = true;
    bool shutDown = (state == evt::STOP);
    bool isLumi = (state == evt::USEDLS);
    UInt_t fuResourceId = cell->fuResourceId();
    UInt_t buResourceId = cell->buResourceId();

    if (state == evt::EMPTY) {
        LOG4CPLUS_ERROR(log_, "WARNING! ATTEMPTING TO DISCARD EMPTY CELL!!!");
        return true;
    }

    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;
        }
    }

    try {
        shmBuffer_->discardRawCell(cell);
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:discardWhileHalting:discardRawCell");
    }
    // UPDATED
    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(true);
            lock();
            freeResourceIds_.push(fuResourceId);
            assert(freeResourceIds_.size() <= resources_.size());
            unlock();

            /*
             sendDiscard(buResourceId);
             sendAllocate();
             */
            if (sendDiscards)
                sendDiscard(buResourceId);
        }
    }

    if (!reschedule) {
        discardNoReschedule();
    }

    return reschedule;
}

vector< string > FUResourceTable::dqmCellStates ( ) const

virtual

Implements evf::IPCMethod.

Definition at line 1294 of file FUResourceTable.cc.

References evf::dqm::DISCARDING, alignCSCRings::e, 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();
        try {
            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");
            }
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:dqmCellStates:dqmState");
        }
        shmBuffer_->unlock();
    }
    return result;
}

void FUResourceTable::dropEvent ( )

virtual

Drop next available event.

Implements evf::IPCMethod.

Definition at line 908 of file FUResourceTable.cc.

References alignCSCRings::e, and evf::FUShmRawCell::fuResourceId().

                                {
    FUShmRawCell* cell = 0;
    try {
        cell = shmBuffer_->rawCellToRead();
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e, "FUResourceTable:dropEvent:rawCellToRead");
    }
    UInt_t fuResourceId = cell->fuResourceId();
    try {
        shmBuffer_->finishReadingRawCell(cell);
        shmBuffer_->scheduleRawCellForDiscard(fuResourceId);
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:dropEvent:finishReadingRawCell/scheduleRawCellForDiscard");
    }
}

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

virtual

Send event belonging to crashed process to error stream. Return false if no event is found

Implements evf::IPCMethod.

Definition at line 926 of file FUResourceTable.cc.

References alignCSCRings::e, i, and summarizeEdmComparisonLogfiles::success.

                                                                 {
    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()) {
        try {
            bool shmret = shmBuffer_->writeErrorEventData(runNumber, pid, iRawCell, true);
            if (!shmret)
                LOG4CPLUS_WARN(log_,"Problem writing to the error stream.");
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:handleCrashedEP:writeErrorEventData");
        }
        retval = true;
    } else
        LOG4CPLUS_WARN(log_,
                "No raw data to send to error stream for process " << pid);
    try {
        bool success = shmBuffer_->removeClientPrcId(pid);
        if (!success)
          LOG4CPLUS_WARN(log_,
                "removeClientPrcId: " << pid << " not in shared memory index, was in raw cell " << iRawCell);
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:handleCrashedEP:removeClientPrcId");
    }
    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
	)

Initialization of the Resource Table with the required resources

Definition at line 65 of file FUResourceTable.cc.

References hitfit::clear(), evf::FUShmBuffer::createShmBuffer(), i, lumiQueryAPI::msg, and evf::FUResource::release().

                                                                      {
    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++) {
        FUResource* newResource = new FUResource(i, log_, frb_, app_);
        newResource->release(true);
        resources_.push_back(newResource);
        freeResourceIds_.push(i);
    }

    acceptSMDataDiscard_ = new bool[nbRecoCells];
    acceptSMDqmDiscard_ = new int[nbDqmCells];

    resetCounters();
    stopFlag_=false;
}

void FUResourceTable::lastResort ( )

virtual

Has to be implemented by subclasses, according to IPC type.

Implements evf::IPCMethod.

Definition at line 1381 of file FUResourceTable.cc.

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

                                 {
    try {
        ostringstream ost;
        ost << "lastResort: " << shmBuffer_->nbRawCellsToRead()
                << " more rawcells to read ";
        LOG4CPLUS_WARN(log_,ost.str());
        std::cout << ost.str() << std::endl;

        while (shmBuffer_->nbRawCellsToRead() != 0) {
            FUShmRawCell* newCell = shmBuffer_->rawCellToRead();
            std::cout << "lastResort: " << shmBuffer_->nbRawCellsToRead()
                    << std::endl;
            // UPDATED
            LOG4CPLUS_WARN(log_,"lastResort: Scheduling raw cell (server side) "<< newCell->index());
            shmBuffer_->scheduleRawCellForDiscardServerSide(newCell->index());

            std::cout << "lastResort: schedule raw cell for discard "
                    << newCell->index() << std::endl;
        }
        //trigger the shutdown (again?)
        LOG4CPLUS_WARN(log_,"lastResort: scheduling empty raw cell (server side) ");
        shmBuffer_->scheduleRawEmptyCellForDiscard();
        LOG4CPLUS_WARN(log_,"lastResort: Finished. cells remaining: "   << shmBuffer_->nbRawCellsToRead());
    } catch (evf::Exception& e) {
        rethrowShmBufferException(
                e,
                "FUResourceTable:lastResort:nbRawCellsToRead/scheduleRawCellForDiscardServerSide");
    }
        LOG4CPLUS_WARN(log_,"Last resort finished ");
}

UInt_t FUResourceTable::nbClients ( ) const

virtual

Implements evf::IPCMethod.

Definition at line 1199 of file FUResourceTable.cc.

References alignCSCRings::e, and query::result.

                                        {
    UInt_t result(0);
    try {
        if (0 != shmBuffer_)
            result = shmBuffer_->nClients();
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e, "FUResourceTable:nbClients:nClients");
    }
    return result;
}

UInt_t evf::FUResourceTable::nbResources ( ) const

inlinevirtual

Implements evf::IPCMethod.

Definition at line 145 of file FUResourceTable.h.

References evf::IPCMethod::resources_.

                               {
        return resources_.size();
    }

unsigned int evf::FUResourceTable::nbResources ( )

inline

Definition at line 162 of file FUResourceTable.h.

References evf::IPCMethod::resources_.

                               {
        return resources_.size();
    }

void FUResourceTable::postEndOfLumiSection ( MemRef_t *  bufRef )

virtual

Post End-of-LumiSection to Shared Memory.

Implements evf::IPCMethod.

Definition at line 881 of file FUResourceTable.cc.

References alignCSCRings::e, i, if(), and lumiQueryAPI::msg.

                                                           {
    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

    // Check input
    int lumiCheck = (int) msg->lumiSection;
    if (lumiCheck < 0)
        LOG4CPLUS_ERROR(log_,
                "Received EOL message with invalid index:" << lumiCheck);

    for (unsigned int i = 0; i < nbRawCells_; i++) {
        // UPDATED
        if (stopFlag_) break;
        nbEolPosted_++;
        try {
            shmBuffer_->writeRawLumiSectionEvent(msg->lumiSection);
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:postEndOfLumiSection:writeRawLumiSectionEvent");
        }
    }
}

std::string FUResourceTable::printStatus ( )

virtual

Print debugging status.

Reimplemented from evf::IPCMethod.

Definition at line 1430 of file FUResourceTable.cc.

                                       {
    if (shmBuffer_) return shmBuffer_->sem_print_s();
    else return std::string("ShmBuffer not initialized");
}

void FUResourceTable::resetCounters ( )

virtual

Reset event & error counters

Implements evf::IPCMethod.

Definition at line 1157 of file FUResourceTable.cc.

References alignCSCRings::e, and i.

                                    {
    if (0 != shmBuffer_) {
        try {
            for (UInt_t i = 0; i < shmBuffer_->nRecoCells(); i++)
                acceptSMDataDiscard_[i] = false;
            for (UInt_t i = 0; i < shmBuffer_->nDqmCells(); i++)
                acceptSMDqmDiscard_[i] = 0;
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:resetCounters:nRecoCells/nDqmCells");
        }
    }

    // UPDATE: reset pending allocate's
    nbAllocated_ = 0;
    nbPending_ = 0;
    nbCompleted_ = 0;
    nbSent_ = 0;
    nbSentError_ = 0;
    nbSentDqm_ = 0;
    nbPendingSMDiscards_ = 0;
    nbPendingSMDqmDiscards_ = 0;
    nbDiscarded_ = 0;
    nbLost_ = 0;
    // UPDATED
    nbEolPosted_ = 0;
    nbEolDiscarded_ = 0;

    nbErrors_ = 0;
    nbCrcErrors_ = 0;
    nbAllocSent_ = 0;

    sumOfSquares_ = 0;
    sumOfSizes_ = 0;

    //"send" workloop states
    sDqmActive_=true;
    sDataActive_=true;

}

void FUResourceTable::resetIPC ( )

virtual

reset the ShmBuffer to the initial state

Implements evf::IPCMethod.

Definition at line 1412 of file FUResourceTable.cc.

References gather_cfg::cout.

                               {
    if (shmBuffer_ != 0) {
        //waiting for sendData and sendDqm workloops to finish
        int countdown_=60;
        while (countdown_-- && (sDataActive_ || sDqmActive_)) ::usleep(50000);
        if (countdown_<=0) {
          std::ostringstream ostr;
          ostr << "Resource broker timed out waiting for workloop shutdowns (3 seconds). Continuing to reset Shm. States - "
               << " sendDqm:"<<sDqmActive_ << " sendData:" << sDataActive_;
          LOG4CPLUS_ERROR(log_,ostr.str());
          std::cout << ostr.str() << std::endl;
        }
        //resetting shm buffer
        shmBuffer_->reset(false);
        LOG4CPLUS_INFO(log_, "ShmBuffer was reset!");
    }
}

void FUResourceTable::rethrowShmBufferException ( evf::Exception &  e, std::string  where  ) const throw ( evf::Exception )

private

Rethrows an exception from the ShmBuffer including details.

Definition at line 1435 of file FUResourceTable.cc.

References alignCSCRings::e, and i.

                             {
    stringstream details;
    vector < string > dataStates = cellStates();
    vector < string > dqmStates = dqmCellStates();
    details << "Exception raised: " << e.what() << " (in module: "
            << e.module() << " in function: " << e.function() << " at line: "
            << e.line() << ")";
    details << "   Dumping cell state...   ";
    details << "data cells --> ";
    for (unsigned int i = 0; i < dataStates.size(); i++)
        details << dataStates[i] << " ";
    details << "dqm cells --> ";
    for (unsigned int i = 0; i < dqmStates.size(); i++)
        details << dqmStates[i] << " ";
    details << " ... originated in: " << where;
    XCEPT_RETHROW(evf::Exception, details.str(), e);
}

bool FUResourceTable::sendData ( )

virtual

Corresponds to the SendData workloop, to be called in normal operation.

Implements evf::IPCMethod.

Definition at line 93 of file FUResourceTable.cc.

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

                               {
    bool reschedule = true;
    FUShmRecoCell* cell = 0;
    try {
        cell = shmBuffer_->recoCellToRead();
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e, "FUResourceTable:sendData:recoCellToRead");
    }

    if (0 == cell->eventSize()) {
        LOG4CPLUS_INFO(log_, "Don't reschedule sendData workloop.");
        UInt_t cellIndex = cell->index();
        try {
            shmBuffer_->finishReadingRecoCell(cell);
            shmBuffer_->discardRecoCell(cellIndex);
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:sendData:finishReadingRecoCell/discardRecoCell");
        }
            shutdownStatus_|=1<<7;
        reschedule = false;
    } 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();
                UInt_t cellExpectedEPs = cell->nExpectedEPs();
                try {
                    shmBuffer_->finishReadingRecoCell(cell);
                } catch (evf::Exception& e) {
                    rethrowShmBufferException(e,
                            "FUResourceTable:sendData:finishReadingRecoCell");
                }

                lock();
                nbPendingSMDiscards_++;
                unlock();

                sendInitMessage(cellIndex, cellOutModId, cellFUProcId,
                        cellFUGuid, cellPayloadAddr, cellEventSize,
                        cellExpectedEPs);
            } 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();
                try {
                    shmBuffer_->finishReadingRecoCell(cell);
                } catch (evf::Exception& e) {
                    rethrowShmBufferException(e,
                            "FUResourceTable:sendData:finishReadingRecoCell");
                }

                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();
                try {
                    shmBuffer_->finishReadingRecoCell(cell);
                } catch (evf::Exception& e) {
                    rethrowShmBufferException(e,
                            "FUResourceTable:sendData:recoCellToRead");
                }

                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;
        }
    }

    sDataActive_=reschedule;
    return reschedule;
}

bool FUResourceTable::sendDataWhileHalting ( )

virtual

Function called when FSM is in state stopping / halting, in SendData workloop.

Implements evf::IPCMethod.

Definition at line 212 of file FUResourceTable.cc.

References alignCSCRings::e, evf::FUShmRecoCell::eventSize(), and evf::FUShmRecoCell::index().

                                           {
    bool reschedule = true;
    FUShmRecoCell* cell = 0;
    try {
        cell = shmBuffer_->recoCellToRead();
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:sendDataWhileHalting:recoCellToRead");
    }

    if (0 == cell->eventSize()) {
        LOG4CPLUS_INFO(log_, "Don't reschedule sendData workloop.");
        UInt_t cellIndex = cell->index();
        try {
            shmBuffer_->finishReadingRecoCell(cell);
            shmBuffer_->discardRecoCell(cellIndex);
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:sendDataWhileHalting:finishReadingRecoCell/discardRecoCell");
        }
            shutdownStatus_|=1<<8;
        reschedule = false;
    } else {
        LOG4CPLUS_INFO(log_, "sendData: isHalting, discard recoCell.");
        UInt_t cellIndex = cell->index();
        try {
            shmBuffer_->finishReadingRecoCell(cell);
            shmBuffer_->discardRecoCell(cellIndex);
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:sendDataWhileHalting:finishReadingRecoCell/discardRecoCell");
        }
    }

    sDataActive_=reschedule;
    return reschedule;
}

bool FUResourceTable::sendDqm ( )

virtual

Corresponds to the SendDqm workloop, to be called in normal operation.

Implements evf::IPCMethod.

Definition at line 251 of file FUResourceTable.cc.

References gather_cfg::cout, alignCSCRings::e, 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.

                              {
    bool reschedule = true;
    FUShmDqmCell* cell = 0;
    // initialize to a value to avoid warnings
    dqm::State_t state = dqm::EMPTY;
    try {
        cell = shmBuffer_->dqmCellToRead();
        state = shmBuffer_->dqmState(cell->index());
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:sendDqm:dqmCellToRead/dqmState");
    }

    if (state == dqm::EMPTY) {
        LOG4CPLUS_INFO(log_, "Don't reschedule sendDqm workloop.");
        std::cout << "shut down dqm workloop " << std::endl;
        UInt_t cellIndex = cell->index();
        try {
            shmBuffer_->finishReadingDqmCell(cell);
            shmBuffer_->discardDqmCell(cellIndex);
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:sendDqm:finishReadingDqmCell/discardDqmCell");
        }
            shutdownStatus_|=1<<9;
        reschedule = false;
    } 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);
            try {
                shmBuffer_->finishReadingDqmCell(cell);
            } catch (evf::Exception& e) {
                rethrowShmBufferException(e,
                        "FUResourceTable:sendDqm:finishReadingDqmCell");
            }
        } catch (xcept::Exception& e) {
            LOG4CPLUS_FATAL(
                    log_,
                    "Failed to send DQM DATA to StorageManager: "
                            << xcept::stdformat_exception_history(e));
            reschedule = false;
        }
    }

    sDqmActive_=reschedule;
    return reschedule;
}

bool FUResourceTable::sendDqmWhileHalting ( )

virtual

Function called when FSM is in state stopping / halting, in SendDqm workloop.

Implements evf::IPCMethod.

Definition at line 310 of file FUResourceTable.cc.

References gather_cfg::cout, alignCSCRings::e, evf::dqm::EMPTY, evf::FUShmDqmCell::index(), and evf::utils::state.

                                          {
    bool reschedule = true;
    FUShmDqmCell* cell = 0;
    // initialize to a value to avoid warnings
    dqm::State_t state = dqm::EMPTY;
    try {
        cell = shmBuffer_->dqmCellToRead();
        state = shmBuffer_->dqmState(cell->index());
    } catch (evf::Exception& e) {
        rethrowShmBufferException(e,
                "FUResourceTable:sendDqmWhileHalting:dqmCellToRead/dqmState");
    }

    if (state == dqm::EMPTY) {
        LOG4CPLUS_INFO(log_, "Don't reschedule sendDqm workloop.");
        std::cout << "shut down dqm workloop " << std::endl;
        UInt_t cellIndex = cell->index();
        try {
            shmBuffer_->finishReadingDqmCell(cell);
            shmBuffer_->discardDqmCell(cellIndex);
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:sendDqmWhileHalting:finishReadingDqmCell/discardDqmCell");
        }
            shutdownStatus_|=1<<10;
        reschedule = false;
    } else {
        UInt_t cellIndex = cell->index();
        try {
            shmBuffer_->finishReadingDqmCell(cell);
            shmBuffer_->discardDqmCell(cellIndex);
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:sendDqmWhileHalting:finishReadingDqmCell/discardDqmCell");
        }
    }

    sDqmActive_=reschedule;
    return reschedule;
}

void FUResourceTable::shutDownClients ( )

virtual

Send empty events to notify clients to shutdown

Implements evf::IPCMethod.

Definition at line 984 of file FUResourceTable.cc.

References alignCSCRings::e, evf::evt::EMPTY, i, lumiQueryAPI::msg, n, evf::utils::pid, sistrip::runNumber_, shutdownWatchdog(), evf::utils::state, and ntuplemaker::status.

                                      {
    nbClientsToShutDown_ = nbClients();
    isReadyToShutDown_ = false;

    shutdownStatus_=1;

    //start watchdog thread
    watchDogEnd_=false;
    watchDogSetFailed_=false;
        #ifdef linux
    std::thread watch(&FUResourceTable::shutdownWatchdog,this,20);
        #endif
    if (nbClientsToShutDown_ == 0) {
            shutdownStatus_|=1<<1;
        LOG4CPLUS_INFO(
                log_,
                "No clients to shut down. Checking if there are raw cells not assigned to any process yet");
        UInt_t n = nbResources();
        try {
            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();
        } catch (evf::Exception& e) {
            rethrowShmBufferException(e,
                    "FUResourceTable:shutDownClients:evtState/scheduleRawEmptyCellForDiscard");
        }
    } else {
        // UPDATED
        int checks = 0;
        try {
            while (shmBuffer_->nbRawCellsToWrite() < nbClients() && nbClients()
                    != 0) {
                    shutdownStatus_|=1<<2;
                checks++;
                {
                                        #ifdef linux
                    auto lk = lockCrashHandlerTimed(10);
                                        #else
                                        bool lk=true;
                                        #endif
                    if (lk) {
                        vector < pid_t > prcids = 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 << " completed with error.");
                                handleCrashedEP(runNumber_, pid);
                            }
                        }
                    }
                    else {
                        XCEPT_RAISE(evf::Exception, 
                            "Timed out access to the Crash Handler in stop. SM discards not arriving?");

                    }
                }

                LOG4CPLUS_WARN(
                        log_,
                        "no cell to write stop "
                                << shmBuffer_->nbRawCellsToWrite()
                                << " nClients " << nbClients());
                if (checks > 15) {
                    string msg = "No Raw Cell to Write STOP messages";
                    XCEPT_RAISE(evf::Exception, msg);
                }
                ::usleep(500000);
            }
                    shutdownStatus_|=1<<3;

        } catch (evf::Exception& e) {
            watchDogEnd_=true;
            #ifdef linux
            watch.join();
            #endif
            rethrowShmBufferException(e,
                    "FUResourceTable:shutDownClients:nbRawCellsToWrite");
        }
        nbClientsToShutDown_ = nbClients();
        if (nbClientsToShutDown_ == 0) {
                    shutdownStatus_|=1<<4;
            UInt_t n = nbResources();
            for (UInt_t i = 0; i < n; i++) {
                // initialize to a value to avoid warnings
                evt::State_t state = evt::EMPTY;
                try {
                    state = shmBuffer_->evtState(i);
                } catch (evf::Exception& e) {
                    watchDogEnd_=true;
                    #ifdef linux
                    watch.join();
                    #endif
                    rethrowShmBufferException(e,
                            "FUResourceTable:shutDownClients:evtState");
                }
                if (state != evt::EMPTY) {
                    LOG4CPLUS_WARN(
                            log_,
                            "Schedule discard at STOP for orphaned event in state "
                                    << state);
                    try {
                        shmBuffer_->setEvtDiscard(i, 1, true);
                        shmBuffer_->scheduleRawCellForDiscardServerSide(i);
                    } catch (evf::Exception& e) {
                        watchDogEnd_=true;
                        #ifdef linux
                        watch.join();
                        #endif
                        rethrowShmBufferException(e,
                                "FUResourceTable:shutDownClients:scheduleRawCellForDiscardServerSide");
                    }
                }
            }
            try {
                shmBuffer_->scheduleRawEmptyCellForDiscard();
            } catch (evf::Exception& e) {
                watchDogEnd_=true;
                #ifdef linux
                watch.join();
                #endif
                rethrowShmBufferException(e,
                        "FUResourceTable:shutDownClients:scheduleRawEmptyCellForDiscard");
            }
        }
        UInt_t n = nbClientsToShutDown_;
            shutdownStatus_|=1<<5;
        try {
            for (UInt_t i = 0; i < n; ++i)
                shmBuffer_->writeRawEmptyEvent();
        } catch (evf::Exception& e) {
            watchDogEnd_=true;
            #ifdef linux
            watch.join();
            #endif
            rethrowShmBufferException(e,
                    "FUResourceTable:shutDownClients:writeRawEmptyEvent");
        }
            shutdownStatus_|=1<<6;
    }
    watchDogEnd_=true;
        #ifdef linux
    watch.join();
    if (watchDogSetFailed_)
      XCEPT_RAISE(evf::Exception, "Failed (timed out) shutdown of clients");
        #endif
}

void FUResourceTable::shutdownWatchdog ( unsigned int  timeout )

private

Spawned as a thread watching for shutdown of clients.

Definition at line 963 of file FUResourceTable.cc.

Referenced by shutDownClients().

{
    unsigned int timeoutUs=timeout*1000000+1;
    bool warned=false;
    while (!watchDogEnd_) {

        usleep(50000);
        timeoutUs-=50000;
        if (timeoutUs<=50000) {
            LOG4CPLUS_ERROR(log_,"Timeout in shutdownClients, status:"<< std::hex << shutdownStatus_);
            watchDogSetFailed_=true;
            break;
        }
        if (timeoutUs<=1000000*timeout/2 && !warned) {
            warned=true;
            LOG4CPLUS_WARN(log_,"Long shutdown of clients, status:" << std::hex << shutdownStatus_);
        }
    }
}

Member Data Documentation

bool evf::FUResourceTable::sDataActive_

private

Definition at line 194 of file FUResourceTable.h.

bool evf::FUResourceTable::sDqmActive_

private

Definition at line 195 of file FUResourceTable.h.

FUShmBuffer* evf::FUResourceTable::shmBuffer_

private

Definition at line 191 of file FUResourceTable.h.

std::atomic_bool evf::FUResourceTable::watchDogEnd_

private

Definition at line 197 of file FUResourceTable.h.

std::atomic_bool evf::FUResourceTable::watchDogSetFailed_

private

Definition at line 198 of file FUResourceTable.h.