/afs/cern.ch/work/a/aaltunda/public/www/CMSSW_6_2_5/src/EventFilter/ResourceBroker/interface/IPCMethod.h

Go to the documentation of this file.

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


#ifndef IPCMETHOD_H_
#define IPCMETHOD_H_

#include "EventFilter/ResourceBroker/interface/FUResource.h"
#include "EventFilter/ResourceBroker/interface/BUProxy.h"
#include "EventFilter/ResourceBroker/interface/SMProxy.h"
#include "EventFilter/ResourceBroker/interface/FUTypes.h"
#include "EventFilter/Utilities/interface/Exception.h"

#include "log4cplus/logger.h"

#include <sys/types.h>
#include <string>
#include <vector>
#include <queue>
#include <semaphore.h>
//#include <pthread.h>
#include <mutex>
#include <atomic>

namespace evf {

class IPCMethod: public toolbox::lang::Class {

public:
        //
        // construction/destruction
        //
        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);

        virtual ~IPCMethod();

        //
        // member functions
        //

        // set the run number
        void setRunNumber(UInt_t runNumber) {
                runNumber_ = runNumber;
        }

        virtual bool sendData() = 0;
        virtual bool sendDataWhileHalting() = 0;
        virtual bool sendDqm() = 0;
        virtual bool sendDqmWhileHalting() = 0;
        virtual bool discard() = 0;
        virtual bool discardWhileHalting(bool sendDiscards) = 0;

        UInt_t allocateResource();

        virtual bool buildResource(MemRef_t* bufRef) = 0;

        virtual bool discardDataEvent(MemRef_t* bufRef) = 0;
        virtual bool discardDataEventWhileHalting(MemRef_t* bufRef) = 0;

        virtual bool discardDqmEvent(MemRef_t* bufRef) = 0;
        virtual bool discardDqmEventWhileHalting(MemRef_t* bufRef) = 0;

        virtual void postEndOfLumiSection(MemRef_t* bufRef) = 0;

        virtual void dropEvent() = 0;

        virtual bool handleCrashedEP(UInt_t runNumber, pid_t pid) = 0;

        void dumpEvent(evf::FUShmRawCell* cell);

        virtual std::string printStatus();

        virtual void shutDownClients() = 0;

        virtual void clear() = 0;

        virtual void resetCounters() = 0;

        void setDoCrcCheck(UInt_t doCrcCheck) {
                doCrcCheck_ = doCrcCheck;
        }

        void setDoDumpEvents(UInt_t doDumpEvents) {
                doDumpEvents_ = doDumpEvents;
        }

        bool isActive() const {
                return isActive_;
        }

        void setActive(bool activeValue) {
                isActive_ = activeValue;
        }

        bool isReadyToShutDown() const {
                return isReadyToShutDown_;
        }

        void setReadyToShutDown(bool readyValue) {
                isReadyToShutDown_ = readyValue;
        }

        UInt_t shutdownStatus() {
                return shutdownStatus_;
        }

        void setStopFlag(bool status) {
                stopFlag_=status;
        }

        // various counters
        virtual UInt_t nbResources() const = 0; /* Implemented in subclass */
        UInt_t nbFreeSlots() const {
                return freeResourceIds_.size();
        }
        UInt_t nbAllocated() const {
                return nbAllocated_;
        }
        UInt_t nbPending() const {
                return nbPending_;
        }
        UInt_t nbCompleted() const {
                return nbCompleted_;
        }
        UInt_t nbSent() const {
                return nbSent_;
        }
        UInt_t nbSentError() const {
                return nbSentError_;
        }
        UInt_t nbSentDqm() const {
                return nbSentDqm_;
        }
        UInt_t nbPendingSMDiscards() const {
                return nbPendingSMDiscards_;
        }
        int nbPendingSMDqmDiscards() const {
                return nbPendingSMDqmDiscards_;
        }
        UInt_t nbDiscarded() const {
                return nbDiscarded_;
        }
        UInt_t nbLost() const {
                return nbLost_;
        }

        // UPDATED
        UInt_t nbEolPosted() const {
                return nbEolPosted_;
        }
        UInt_t nbEolDiscarded() const {
                return nbEolDiscarded_;
        }

        UInt_t nbErrors() const {
                return nbErrors_;
        }
        UInt_t nbCrcErrors() const {
                return nbCrcErrors_;
        }
        UInt_t nbAllocSent() const {
                return nbAllocSent_;
        }

        uint64_t sumOfSquares() const {
                return sumOfSquares_;
        }
        UInt_t sumOfSizes() const {
                return sumOfSizes_;
        }

        // information about (raw) shared memory cells
        virtual UInt_t nbClients() const = 0;
        virtual std::vector<pid_t> clientPrcIds() const = 0;
        virtual std::string clientPrcIdsAsString() const = 0;
        virtual std::vector<std::string> cellStates() const = 0;
        virtual std::vector<std::string> dqmCellStates() const = 0;
        virtual std::vector<UInt_t> cellEvtNumbers() const = 0;
        virtual std::vector<pid_t> cellPrcIds() const = 0;
        virtual std::vector<time_t> cellTimeStamps() const = 0;

        //
        // helpers
        //
        void sendAllocate();
        void resetPendingAllocates();
        void releaseResources();
        virtual void resetIPC() = 0;

        void sendDiscard(UInt_t buResourceId);

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

        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 sendErrorEvent(UInt_t fuResourceId, UInt_t runNumber,
                        UInt_t evtNumber, UInt_t fuProcessId, UInt_t fuGuid, UChar_t*data,
                        UInt_t dataSize);

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

        bool isLastMessageOfEvent(MemRef_t* bufRef);

        void injectCRCError();

        void lock() {
                //lock_.take();
                while (0 != sem_wait(&lock_)) {
                        if (errno != EINTR) {
                                LOG4CPLUS_ERROR(log_, "Cannot obtain lock on sem LOCK!");
                        }
                }
        }
        void unlock() {
                //lock_.give();
                sem_post(&lock_);
        }
        #ifdef linux
        std::unique_lock<std::timed_mutex> lockCrashHandler() {
                std::unique_lock<std::timed_mutex> lk(crashHandlerLock_);
                return lk;
        }

        std::unique_lock<std::timed_mutex> lockCrashHandlerTimed(unsigned int seconds) {
                std::unique_lock<std::timed_mutex> lk(crashHandlerLock_,std::chrono::seconds(seconds));
                return lk;
        }
        #endif

        virtual void lastResort() = 0;

protected:
        //
        // member data
        //

        BUProxy *bu_;
        SMProxy *sm_;

        log4cplus::Logger log_;

        UInt_t nbDqmCells_;
        UInt_t nbRawCells_;
        UInt_t nbRecoCells_;

        std::queue<UInt_t> freeResourceIds_;
        // number of free resources required to ask BU for more events
        unsigned int freeResRequiredForAllocate_;

        bool *acceptSMDataDiscard_;
        int *acceptSMDqmDiscard_;

        UInt_t doCrcCheck_;
        UInt_t doDumpEvents_;
        unsigned int shutdownTimeout_;

        UInt_t nbAllocated_;
        UInt_t nbPending_;
        UInt_t nbCompleted_;
        UInt_t nbSent_;
        UInt_t nbSentError_;
        UInt_t nbSentDqm_;
        UInt_t nbPendingSMDiscards_;
        std::atomic<int> nbPendingSMDqmDiscards_;
        UInt_t nbDiscarded_;
        UInt_t nbLost_;
        // UPDATED
        UInt_t nbEolPosted_;
        UInt_t nbEolDiscarded_;

        UInt_t nbClientsToShutDown_;
        bool isReadyToShutDown_;
        bool isActive_;

        UInt_t nbErrors_;
        UInt_t nbCrcErrors_;
        UInt_t nbAllocSent_;

        uint64_t sumOfSquares_;
        UInt_t sumOfSizes_;

        UInt_t runNumber_;

        sem_t lock_;
        #ifdef linux
        std::timed_mutex crashHandlerLock_;
        #endif
        EvffedFillerRB *frb_;
        xdaq::Application *app_;

        FUResourceVec_t resources_;

        UInt_t shutdownStatus_;
        bool stopFlag_;

};

}

#endif /* IPCMETHOD_H_ */