/data/refman/pasoursint/CMSSW_6_1_1/src/EventFilter/ResourceBroker/src/IPCMethod.cc

Go to the documentation of this file.

//
// IPCMethod.cc
// -------
//
// Contains common functionality for FUResourceTable and FUResourceQueue.
//
//  Created on: Oct 26, 2011
//                                                                              Andrei Spataru : aspataru@cern.ch

#include "EventFilter/ResourceBroker/interface/FUResourceTable.h"
#include "EventFilter/ResourceBroker/interface/IPCMethod.h"

#include "interface/evb/i2oEVBMsgs.h"

#include <iomanip>

using std::ofstream;
using std::endl;
using namespace evf;

// construction/destruction

//______________________________________________________________________________
IPCMethod::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) :
        bu_(bu), sm_(sm), log_(logger), nbDqmCells_(nbDqmCells),
                        nbRawCells_(nbRawCells), nbRecoCells_(nbRecoCells),
                        acceptSMDataDiscard_(0), acceptSMDqmDiscard_(0), doCrcCheck_(1),
                        shutdownTimeout_(timeout), nbPending_(0), nbClientsToShutDown_(0),
                        isReadyToShutDown_(true), isActive_(false), runNumber_(0xffffffff),
                        frb_(frb), app_(app) {

        // if the freeResRequiredForAllocate_ threshold is set in configuration use that
        // otherwise use nbRawCells / 2
        if (freeResReq < 0)
                freeResRequiredForAllocate_ = nbRawCells_ / 2;
        else
                freeResRequiredForAllocate_ = freeResReq;

        sem_init(&lock_, 0, 1);
        //pthread_mutex_init(&crashHandlerLock_, NULL);
}

IPCMethod::~IPCMethod() {

}

// implementation of member functions

//______________________________________________________________________________
UInt_t IPCMethod::allocateResource() {
        assert(!freeResourceIds_.empty());

        lock();
        UInt_t fuResourceId = freeResourceIds_.front();
        freeResourceIds_.pop();
        nbPending_++;
        nbAllocated_++;
        unlock();

        return fuResourceId;
}

//______________________________________________________________________________
void IPCMethod::dumpEvent(FUShmRawCell* cell) {
        std::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 << std::setiosflags(std::ios::right) << std::setw(2)
                                        << std::hex << (int) (*addr) << std::dec;
                        if ((j + 1) % 8)
                                fout << " ";
                        else
                                fout << endl;
                        ++addr;
                }
                fout << endl;
        }
        fout.close();
}

//______________________________________________________________________________
std::string IPCMethod::printStatus() {
        std::string s = "Status not implemented";
        return s;
}

// implementation of private member functions

//______________________________________________________________________________
void IPCMethod::sendAllocate() {
        UInt_t nbFreeSlots = this->nbFreeSlots();
        /*UInt_t nbFreeSlotsMax = 0*///reverting to larger chunk requests for BU
        //UInt_t nbFreeSlotsMax = nbResources() / 2;
        UInt_t nbFreeSlotsMax = freeResRequiredForAllocate_;

        if (nbFreeSlots > nbFreeSlotsMax) {
                UIntVec_t fuResourceIds;
                for (UInt_t i = 0; i < nbFreeSlots; i++)
                        fuResourceIds.push_back(allocateResource());

                bu_->sendAllocate(fuResourceIds);

                nbAllocSent_++;
        }
}

//______________________________________________________________________________
void IPCMethod::resetPendingAllocates() {
        if (freeResourceIds_.size() < nbRawCells_) {
                LOG4CPLUS_INFO(
                                log_,
                                "There are " << nbRawCells_ - freeResourceIds_.size()
                                                << " pending ALLOCATE messages! Forgetting...");
                while (!freeResourceIds_.empty())
                        freeResourceIds_.pop();
                for (UInt_t i = 0; i < nbRawCells_; i++)
                        freeResourceIds_.push(i);
        }
}

//______________________________________________________________________________
void IPCMethod::releaseResources() {
        for (UInt_t i = 0; i < resources_.size(); i++)
                resources_[i]->release(true);

}

//______________________________________________________________________________
void IPCMethod::sendDiscard(UInt_t buResourceId) {
        bu_->sendDiscard(buResourceId);
        nbDiscarded_++;
}

//______________________________________________________________________________
void IPCMethod::sendInitMessage(UInt_t fuResourceId, UInt_t outModId,
                UInt_t fuProcessId, UInt_t fuGuid, UChar_t *data, UInt_t dataSize,
                UInt_t nExpectedEPs) {
        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, nExpectedEPs);
                sumOfSquares_ += (uint64_t) nbBytes * (uint64_t) nbBytes;
                sumOfSizes_ += nbBytes;
        }
}

//______________________________________________________________________________
void IPCMethod::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) {
        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 IPCMethod::sendErrorEvent(UInt_t fuResourceId, UInt_t runNumber,
                UInt_t evtNumber, UInt_t fuProcessId, UInt_t fuGuid, UChar_t *data,
                UInt_t dataSize) {
        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 IPCMethod::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) {
        if (0 == sm_) {
                LOG4CPLUS_WARN(log_, "No StorageManager, DROP DQM EVENT.");
        } else {

                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_++;
                sm_->sendDqmEvent(fuDqmId, runNumber, evtAtUpdate, folderId,
                                fuProcessId, fuGuid, data, dataSize);
        }
}

//______________________________________________________________________________
bool IPCMethod::isLastMessageOfEvent(MemRef_t* bufRef) {
        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));
}

//______________________________________________________________________________
void IPCMethod::injectCRCError() {
        for (UInt_t i = 0; i < resources_.size(); i++) {
                resources_[i]->scheduleCRCError();
        }
}