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_ */